Add possibility that a output of multiple tasks become input for a task

[alifbe]

One example use case, we can have multiple tasks (could be different type) that produce RegularSurface object. We also have a task that take List[Surface] as input.

graph LR
    A[Load surface] -- "Surface A" --> J[ ]
    B[Generate surface] -- "Surface B" --> J
    C[Generate surface] -- "Surface C" --> J
    D[Load surface] -- "Surface D" --> J
    J -- "List[Surface]" --> Main[Create Grid]

    style J width:0px,height:0px,stroke:none 

Loading

[kriben]

Currently taskmaestro maps each dependency to one statically declared input field. It cannot collect repeated task outputs into a list or dictionary. A workflow with two surface loaders therefore cannot naturally populate dict[str, RegularSurface] or list[RegularSurface].

A collection syntax for dict could preserve aliases and type information:

depends_on:
  surfaces:
    collect:
      top: top_load_result
      base: bottom_load_result

The runner would validate every collected output against the dictionary value type and construct.

Useful variants would include:

• collection into list[T], tuple[T, ...], and dict[str, T];
• output-field routing for each member;
• deterministic ordering;
• duplicate-key validation; and
• optional empty collections.

[kriben]

Design and implementation plan for issue #4

Summary

Add an explicit collection dependency that gathers multiple task outputs into one field of a downstream Pydantic input model.

The first version will support:

• list[T]
• dict[str, T]
• Whole task outputs
• Output-field routing
• Subtypes of the declared element type
• Deterministic declaration ordering
• Explicit empty collections
• Python and YAML workflow definitions
• Mermaid collection nodes

Deferred:

• tuple[T, ...]
• Sets and arbitrary collection protocols
• Heterogeneous fixed-length tuples
• Collection into a task's entire root input
• Union element types, unless they fall out naturally from shared type-compatibility work

1. Public Python API

Add a public collect() helper exported from taskmaestro.

Lists

from taskmaestro import collect

workflow = (
    Workflow.builder("create_grid")
    .add_task(LoadTop, name="top")
    .add_task(GenerateMiddle, name="middle")
    .add_task(LoadBase, name="base")
    .add_task(
        CreateGrid,
        depends_on={
            "surfaces": collect("top", "middle", "base"),
        },
    )
    .build()
)

The destination input model declares the resulting type:

class CreateGridInput(BaseModel):
    surfaces: list[Surface]

Declaration order controls list order regardless of task execution order.

Dictionaries

.add_task(
    CreateGrid,
    depends_on={
        "surfaces": collect(
            {
                "top": "top_loader",
                "middle": "middle_generator",
                "base": "base_loader",
            }
        ),
    },
)

With:

class CreateGridInput(BaseModel):
    surfaces: dict[str, Surface]

Dictionary keys must be explicitly provided and are not inferred from task names.

Output-field routing

Existing (task, field) references remain valid inside collections:

depends_on={
    "surfaces": collect(
        ("top_loader", "surface"),
        ("middle_generator", "generated_surface"),
        ("base_loader", "surface"),
    ),
}

Class references are supported when unambiguous:

depends_on={
    "surfaces": collect(
        (LoadTop, "surface"),
        GenerateMiddle,
        (LoadBase, "surface"),
    ),
}

Dictionary collections support the same routing:

depends_on={
    "surfaces": collect(
        {
            "top": (LoadTop, "surface"),
            "middle": (GenerateMiddle, "generated_surface"),
            "base": (LoadBase, "surface"),
        }
    ),
}

Empty collections

Empty collections are explicit:

# list[T]
depends_on={"surfaces": collect()}

# dict[str, T]
depends_on={"surfaces_by_name": collect({})}

The destination annotation determines and validates the expected collection kind.

Invalid forms

These fail while building the workflow:

# Positional collection assigned to dict[str, Surface]
"surfaces": collect(Top, Base)

# Keyed collection assigned to list[Surface]
"surfaces": collect({"top": Top, "base": Base})

# Collection assigned to a non-collection field
"surface": collect(Top, Base)

# Output incompatible with collection element type
"surfaces": collect(TaskReturningString)

# Same field supplied by a dependency and JobConfiguration
depends_on={"surfaces": collect(Top, Base)},
config_fields=["surfaces"]

2. YAML syntax

List collection

depends_on:
  surfaces:
    collect:
      - top_loader
      - middle_generator
      - base_loader

List with output-field routing

depends_on:
  surfaces:
    collect:
      - [top_loader, surface]
      - [middle_generator, generated_surface]
      - [base_loader, surface]

Dictionary collection

depends_on:
  surfaces:
    collect:
      top: [top_loader, surface]
      middle: [middle_generator, generated_surface]
      base: [base_loader, surface]

Empty collections

# Empty list
depends_on:
  surfaces:
    collect: []

# Empty dictionary
depends_on:
  surfaces_by_name:
    collect: {}

The explicit collect discriminator prevents ambiguity with existing fan-in mappings and [task, field] output references. Duplicate YAML mapping keys should be rejected rather than silently allowing the last value to win.

3. Internal dependency representation

The current representation is a growing union of strings, tuples, and dictionaries. Collection support would make conditionals throughout workflow validation, execution, and visualization increasingly fragile.

Introduce normalized immutable dependency objects, preferably in a new taskmaestro/dependencies.py:

@dataclass(frozen=True)
class OutputRef:
    task_name: str
    output_field: str | None = None


@dataclass(frozen=True)
class CollectionRef:
    kind: Literal["positional", "keyed"]
    positional_members: tuple[OutputRef, ...] = ()
    keyed_members: tuple[tuple[str, OutputRef], ...] = ()

The public collect() helper initially contains unresolved task references:

type TaskReference = type[Task[Any, Any]] | str
type OutputReference = TaskReference | tuple[TaskReference, str]

WorkflowBuilder.add_task() resolves task classes and names, then stores normalized OutputRef instances.

It would also be valuable to normalize existing single and fan-in dependencies at this boundary. This gives downstream code a single traversal mechanism instead of repeated isinstance(str/tuple/dict) branches.

Existing workflow syntax remains unchanged:

depends_on=Upstream
depends_on="upstream"
depends_on=(Upstream, "field")
depends_on={"a": Upstream, "b": (Other, "field")}

If get_dependencies() is considered public API, it should retain its existing observable results for old dependency forms. A separate internal accessor can expose normalized bindings to the runner and visualization code.

4. Validation rules

Collection validation belongs in Workflow._validate_types() alongside current fan-in validation.

For each collection binding:

1. Confirm the destination field exists.
2. Inspect the destination field annotation with typing.get_origin() and typing.get_args().
3. Require:
   • Positional collection -> list[T]
   • Keyed collection -> dict[str, T]
4. Resolve every member:
   • Whole output -> get_output_type(upstream_task)
   • Field reference -> annotation of the selected output field
5. Validate every resolved type against T.
6. Reject collection/config overlap.
7. Treat the collection as covering the destination field for completeness validation.
8. Permit a collection with zero members.
9. Include every member's task in cycle detection, topological ordering, and sink detection.

Type compatibility

A shared helper should replace direct issubclass() calls where appropriate:

_is_type_compatible(produced_type, expected_type) -> bool

For this feature, compatibility includes exact types and subclasses, for example RegularSurface -> Surface. It must avoid calling issubclass() blindly on parameterized typing constructs.

Runtime Pydantic validation remains the final safeguard when constructing the downstream input model.

Errors

Continue using existing exception types:

• WorkflowDefinitionError
  • Wrong collection shape
  • Invalid output field
  • Type mismatch
  • Unknown task
  • Duplicate or conflicting binding
• IncompleteInputError
  • Required downstream fields not covered

Messages should identify the downstream task and field, collection member key or index, upstream reference, produced type, and expected type.

Example:

Collection type mismatch for 'create_grid.surfaces[1]':
'generate_middle.surface' produces StringOutput, but list element type is Surface

5. Runner behavior

Refactor input assembly into small resolver methods rather than extending the current conditional block:

_resolve_output_ref(ref, outputs)
_resolve_collection(ref, outputs)
_assemble_task_input(...)

A positional collection resolves members in declaration order. A keyed collection preserves explicit aliases and insertion order. Collection values, ordinary fan-in values, and configuration values are combined before calling:

input_type.model_validate(field_values)

Collection members do not need separate runtime validation if final Pydantic model validation is reliable, but build-time validation should still provide earlier and clearer failures.

Empty assembled roots

A task with only collect() has zero graph predecessors but does not consume Job.config. The workflow must distinguish between:

• A task requiring external root input
• A task whose input is assembled from an explicitly empty collection

The YAML loader should use EmptyConfig when no task requires external input, similar to a fully configured workflow.

6. Graph traversal

Create a common recursive dependency-reference iterator:

_iter_output_refs(dependency_spec) -> Iterator[OutputRef]

Use it for:

• Reference validation
• In-degree calculation
• Adjacency construction
• Cycle detection
• Sink detection
• Nested workflow visualization

Repeated references to the same upstream task contribute one graph edge for in-degree purposes, even if its output appears multiple times in a collection. The resulting list may contain that output more than once.

7. Mermaid visualization

Render collections as explicit junction nodes, matching the issue's conceptual diagram.

For:

depends_on={
    "surfaces": collect("top", "middle", "base"),
}

Generate something conceptually similar to:

graph TD
    top -->|Surface| collect_create_grid_surfaces
    middle -->|Surface| collect_create_grid_surfaces
    base -->|Surface| collect_create_grid_surfaces
    collect_create_grid_surfaces -->|surfaces: list[Surface]| create_grid

Loading

The collection node ID should be deterministic and Mermaid-safe, for example _collect_create_grid_surfaces_.

Suggested labels:

• List incoming edges: [0]: Surface, [1]: Surface
• Dictionary incoming edges: top: Surface, base: Surface
• Outgoing edge: surfaces: list[Surface] or surfaces: dict[str, Surface]

Collection nodes should work when an upstream or downstream task is an expanded nested workflow. For an empty collection, render the node and outgoing edge even though there are no incoming task edges.

8. Compatibility

Existing workflows should behave identically:

• Linear workflows
• Single dependencies
• Fan-in dictionaries
• Output-field routing
• Named task instances
• Per-task configuration
• Nested workflows
• Existing YAML syntax
• Existing Mermaid output where collections are not used

collect is additive and should be exported from taskmaestro/__init__.py.

No existing list syntax should be repurposed in Python. In YAML, lists continue to mean [task, output_field] except inside an explicit collect: block.

9. Implementation plan

Phase 1: Dependency model

Files:

• taskmaestro/dependencies.py (new, preferred)
• taskmaestro/workflow.py
• taskmaestro/__init__.py

Work:

1. Add public collect().
2. Add unresolved and resolved collection representations.
3. Add helpers for resolving task and output references.
4. Add recursive dependency traversal.
5. Preserve existing builder syntax.

Phase 2: Workflow validation

File:

• taskmaestro/workflow.py

Work:

1. Validate destination collection annotations.
2. Resolve list element and dictionary value types.
3. Validate whole outputs and routed fields.
4. Support subtype compatibility.
5. Validate empty collections.
6. Reject dependency/config field collisions.
7. Update reference, cycle, topology, and sink logic.

Phase 3: Runtime assembly

File:

• taskmaestro/runner.py

Work:

1. Extract output-reference resolution.
2. Assemble positional and keyed collections.
3. Preserve declaration order.
4. Merge ordinary dependencies and config safely.
5. Handle workflows with no external-input root.

Phase 4: YAML

File:

• taskmaestro/yaml_config.py

Work:

1. Add schema/parser support for collect.
2. Parse positional and keyed forms.
3. Resolve names and routed output fields.
4. Produce the same collection objects as the Python builder.
5. Reject malformed collection members.
6. Reject duplicate YAML mapping keys.
7. Support empty-list and empty-dictionary workflows.

Phase 5: Visualization

File:

• taskmaestro/visualization.py

Work:

1. Add deterministic collection nodes.
2. Add indexed or keyed incoming labels.
3. Add typed outgoing labels.
4. Handle empty collections.
5. Verify nested-workflow redirects.

Phase 6: Documentation and examples

Files:

• README.md
• Potentially examples/resinsight/

Add:

1. Python list example.
2. Python dictionary example.
3. Output-field routing example.
4. YAML equivalents.
5. Ordering and type-validation rules.
6. Empty collection behavior.
7. Mermaid example.

10. Test plan

Workflow validation tests

Add to tests/test_workflow.py:

• Build list[Surface] from multiple identical output types.
• Build it from multiple subclasses of Surface.
• Build dict[str, Surface].
• Whole-output and output-field members.
• Named task references.
• Class references.
• Unambiguous and ambiguous class references.
• Unknown task.
• Unknown output field.
• Incompatible member type.
• Positional collection into dictionary field.
• Keyed collection into list field.
• Collection into scalar field.
• Empty list.
• Empty dictionary.
• Missing required unrelated field.
• Collection field conflicting with config_fields.
• Cycle involving a collection member.
• Sink detection involving collection members.
• Same upstream included more than once.

Runner tests

Add to tests/test_runner.py:

• List values arrive in declaration order.
• Execution or topological ordering does not change list order.
• Dictionary aliases and values are preserved.
• Mixed whole outputs and routed fields.
• Empty list and dictionary.
• Pydantic receives concrete collection types.
• Repeated upstream output appears repeatedly in the result.
• Collection plus other fan-in fields.
• Collection plus unrelated configured fields.

YAML tests

Add to tests/test_yaml_config.py:

• Positional collection end-to-end.
• Keyed collection end-to-end.
• Routed fields inside both forms.
• Empty forms.
• Named instances.
• Unknown dependencies.
• Invalid routed-field syntax.
• Wrong collection shape.
• Duplicate keys.
• Nested workflow output as a collection member.

Visualization tests

Add to tests/test_visualization.py:

• List collection node and indexed edges.
• Dictionary collection node and keyed edges.
• Typed outgoing edge.
• Empty collection node.
• Collection involving routed output fields.
• Collection involving an expanded nested workflow.
• Stable output ordering.

Quality gates

pytest -v
ruff check taskmaestro/ tests/ examples/
ruff format --check taskmaestro/ tests/ examples/
mypy taskmaestro

Acceptance criteria

Issue #4 is complete when the following works through both Python and YAML:

class CreateGridInput(BaseModel):
    surfaces: list[Surface]


workflow = (
    Workflow.builder("grid")
    .add_task(LoadSurface, name="surface_a")
    .add_task(GenerateSurface, name="surface_b")
    .add_task(GenerateSurface, name="surface_c")
    .add_task(LoadSurface, name="surface_d")
    .add_task(
        CreateGrid,
        depends_on={
            "surfaces": collect(
                "surface_a",
                "surface_b",
                "surface_c",
                "surface_d",
            )
        },
    )
    .build()
)

CreateGrid must receive the four surfaces in exactly that declaration order, with all element types checked when the workflow is built.