LuigiCombinator Tasks Cannot Yield Regular Luigi Tasks

[DanielScholtyssek]

Bug: LuigiCombinator Tasks Cannot Yield Regular Luigi Tasks

Description

When a LuigiCombinator task attempts to yield a regular Luigi task in its run() method, the task execution fails. This prevents proper task composition between LuigiCombinator tasks and regular Luigi tasks.

Minimal Reproduction

from cls_luigi.inhabitation_task import RepoMeta, LuigiCombinator, ClsParameter
from cls.fcl import FiniteCombinatoryLogic
from cls.subtypes import Subtypes
import luigi
import os
import shutil


class FirstTask(luigi.Task, LuigiCombinator):
    abstract = False
    
    def requires(self):
        return []
    
    def run(self):
        with self.output().open('w') as f:
            f.write("FirstTask completed")
    
    def output(self):
        return luigi.LocalTarget('outputs/first_task.txt')

class TestLuigiTask(luigi.Task):

    def requires(self):
        return []
    
    def run(self):
        with self.output().open('w') as f:
            f.write("PureLuigiTask completed")
    
    def output(self):
        return luigi.LocalTarget('outputs/pure_luigi_task.txt')

class AbstractTask(luigi.Task, LuigiCombinator):
    abstract = True
    first_task = ClsParameter(tpe=FirstTask.return_type())


class ConcreteTask1(AbstractTask):
    abstract = False
    
    def requires(self):
        return [self.first_task()]
    
    def run(self):
        yield TestLuigiTask()
        # Then do our own work
        with self.output().open('w') as f:
            f.write("ConcreteTask1 completed")
    
    def output(self):
        return luigi.LocalTarget('outputs/concrete_task1.txt')


if __name__ == '__main__':
    max_tasks_when_infinite = 100

    # Clean up and recreate outputs directory
    if os.path.exists('outputs'):
        shutil.rmtree('outputs')
    os.makedirs('outputs')

    print("=" * 50)

    target = AbstractTask.return_type()
    repository = RepoMeta.repository

    fcl = FiniteCombinatoryLogic(repository, Subtypes(RepoMeta.subtypes), processes=1)

    inhabitation_result = fcl.inhabit(target)
    actual = inhabitation_result.size()
    max_results = max_tasks_when_infinite
    if actual > 0:
        max_results = actual

    results = [t() for t in inhabitation_result.evaluated[0:max_results]]
    print("\nTasks to be executed:")
    for r in results:
        print(r)

    print("\nExecuting tasks...")
    luigi.build(results, local_scheduler=True)

    # Print output file contents
    print("\nTask Outputs:")
    for f in os.listdir('outputs'):
        with open(os.path.join('outputs', f), 'r') as file:
            print(f"{f}: {file.read()}")

    print("\n" + "=" * 50)

Expected Behavior

• ConcreteTask1 should be able to yield TestLuigiTask
• TestLuigiTask should execute first
• After TestLuigiTask completes, ConcreteTask1 should continue execution
• Both output files should be created

Actual Behavior

The task execution fails with the following error:

Traceback (most recent call last):
  File "bug.py", line 82, in <module>
    luigi.build(results, local_scheduler=True)
  File "luigi/interface.py", line 243, in build
    luigi_run_result = _schedule_and_run(tasks, worker_scheduler_factory, override_defaults=env_params)
  File "luigi/interface.py", line 177, in _schedule_and_run
    success &= worker.run()
  File "luigi/worker.py", line 1236, in run
    self._handle_next_task()
  File "luigi/worker.py", line 1131, in _handle_next_task
    new_req = [load_task(module, name, params)
  File "luigi/task_register.py", line 252, in load_task
    task_cls = Register.get_task_cls(task_name)
  File "luigi/task_register.py", line 177, in get_task_cls
    task_cls = cls._get_reg().get(name)
  File "luigi/task_register.py", line 136, in _get_reg
    if not task_cls._visible_in_registry:
AttributeError: type object 'LuigiCombinator' has no attribute '_visible_in_registry'

The error indicates that the LuigiCombinator class is missing the _visible_in_registry attribute that Luigi uses internally to track task registration.

Additional Context

• Regular Luigi tasks can yield other regular Luigi tasks successfully
• The issue only occurs when a LuigiCombinator task attempts to yield a Luigi task

Possible Solution

The LuigiCombinator framework needs to be modified to:

1. Properly implement the _visible_in_registry attribute required by Luigi's task registration system
2. Handle task execution delegation between LuigiCombinator and regular Luigi tasks
3. Ensure proper task dependency resolution for mixed task types

Environment

• Python 3.13.0
• Luigi latest version
• cls_luigi framework

[DanielScholtyssek]

This issue appears to be fixed in commit 6d12374 where the Luigi task registration system was properly implemented for LuigiCombinator tasks.

However, @JanBessai could you please review this change to ensure that the CLS inhabitation mechanism is still functioning correctly? The changes to task registration might have subtle implications for how combinatory tasks are handled during inhabitation.

The commit adds the missing _visible_in_registry attribute, but we should verify that:

1. Task registration doesn't interfere with the inhabitation process
2. Generated tasks are still properly handled by the combinatory logic
3. Task dependencies are correctly resolved when mixing LuigiCombinator and regular Luigi tasks

Once confirmed, we can close this issue.