feat(grammar): apply outputs schema per branch on multi-model tasks

Allow a typed `outputs` schema on multi-model tasks. Each branch's result is
validated and coerced against the schema and stored as the `result` of its
fan-in record. A branch whose result violates the schema is treated as a failed
branch, so the task's `completion` policy (all/any) reduces it like any other
branch failure. Removes the prior load-time restriction that rejected `outputs`
on multi-model tasks.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: anticomputer

doc/GRAMMAR.md

@@ -177,8 +177,11 @@ Notes and semantics:
   To consume multi-model results downstream, give the task an `id`: each
   branch's final result is aggregated into `outputs.<id>` as a list of records
   `{"model": ..., "item": ..., "result": ...}` (see "Typed named outputs").
-- The inline `outputs` schema is not yet supported on multi-model tasks (the
-  aggregate is a list of records); use `id` for fan-in.
+- The inline `outputs` schema is applied per branch on multi-model tasks: each
+  branch's result is validated/coerced against the schema and stored as the
+  `result` of its fan-in record. A branch whose result violates the schema is
+  treated as a failed branch, which the `completion` policy (`all`/`any`) then
+  reduces like any other branch failure (see "Typed named outputs").
 
 ### Completion Requirement
 
@@ -456,8 +459,10 @@ Notes:
 - Without `id`/`outputs`/`over`, the implicit last-tool-result `repeat_prompt`
   behaviour is unchanged.
 - On multi-model tasks, `id` fans in per-branch results as a list of
-  `{model, item, result}` records (see "Multiple Models"); the inline `outputs`
-  schema is not yet supported there.
+  `{model, item, result}` records (see "Multiple Models"). When an `outputs`
+  schema is present it is applied per branch: each `result` is the validated
+  value, and a branch that violates the schema is treated as a failed branch
+  under the task's `completion` policy.
 
 ### Toolboxes / MCP Servers
 

src/seclab_taskflow_agent/models.py

@@ -207,11 +207,6 @@ def _validate_outputs(self) -> TaskDefinition:
                 build_output_model(f"{self.id or self.name or 'task'}_output", self.outputs)
             except OutputSchemaError as exc:
                 raise ValueError(f"invalid 'outputs' schema: {exc}") from exc
-            if len(self.models) > 1:
-                raise ValueError(
-                    "'outputs' schema is not yet supported on multi-model tasks; "
-                    "use 'id' to fan-in per-model results as a list"
-                )
         if self.over and not self.repeat_prompt:
             raise ValueError("'over' only applies to repeat_prompt tasks")
         return self

src/seclab_taskflow_agent/runner.py

@@ -28,6 +28,7 @@
 from typing import TYPE_CHECKING, Any
 
 import jinja2
+from pydantic import ValidationError
 
 from ._stream import drive_backend_stream
 from ._watchdog import start_watchdog, watchdog_ping
@@ -233,6 +234,13 @@ class _Branch:
     item_index: int
     label: str
     sink: list[ToolResult] = field(default_factory=list)
+    # Typed multi-model output: when the task declares an ``outputs`` schema,
+    # each branch's result is validated and the coerced value stored here
+    # (``None`` when the branch failed or violated the schema). ``output_set``
+    # marks that validation ran so fan-in uses this value instead of decoding
+    # the raw sink.
+    output: Any = None
+    output_set: bool = False
 
 
 def _aggregate_fanin(branches: list[_Branch]) -> list[dict[str, Any]]:
@@ -241,21 +249,67 @@ def _aggregate_fanin(branches: list[_Branch]) -> list[dict[str, Any]]:
     Produces one record per branch: ``{"model": <label>, "item": <index>,
     "result": <value>}``. ``result`` is the decoded final tool result of the
     branch (its raw text when not JSON, or ``None`` when the branch produced no
-    tool result). Used to expose multi-model / cross-product outputs to later
-    tasks as ``outputs.<id>``.
+    tool result). When the task declares an ``outputs`` schema, ``result`` is
+    the per-branch validated/coerced value instead (``None`` for a branch that
+    failed or violated the schema). Used to expose multi-model / cross-product
+    outputs to later tasks as ``outputs.<id>``.
     """
     records: list[dict[str, Any]] = []
     for branch in branches:
-        value: Any = None
-        if branch.sink:
-            try:
-                value = decode_tool_result(branch.sink[-1])
-            except ValueError:
-                value = branch.sink[-1].text
+        if branch.output_set:
+            value: Any = branch.output
+        else:
+            value = None
+            if branch.sink:
+                try:
+                    value = decode_tool_result(branch.sink[-1])
+                except ValueError:
+                    value = branch.sink[-1].text
         records.append({"model": branch.rm.label, "item": branch.item_index, "result": value})
     return records
 
 
+def _capture_branch_output(
+    branch: _Branch,
+    schema: dict[str, Any],
+    output_id: str,
+    *,
+    agent_ok: bool,
+) -> bool:
+    """Validate one multi-model branch's result against the task's outputs schema.
+
+    Stores the coerced value on ``branch.output`` (``None`` when the branch
+    failed to run or its result violates the contract) and returns whether the
+    branch both ran and produced a schema-valid result. A contract violation is
+    therefore surfaced as a failed branch, which the task's ``completion`` policy
+    (``all`` / ``any``) reduces like any other branch failure.
+    """
+    branch.output_set = True
+    branch.output = None
+    if not agent_ok:
+        return False
+    if not branch.sink:
+        logging.error(
+            "Multi-model branch %r produced no result to validate against 'outputs'",
+            branch.label,
+        )
+        return False
+    try:
+        value = decode_tool_result(branch.sink[-1])
+    except ValueError:
+        value = branch.sink[-1].text
+    try:
+        branch.output = validate_output(schema, value, model_name=f"{output_id}_output")
+        return True
+    except (ValidationError, ValueError) as exc:
+        logging.error(
+            "Multi-model branch %r output violated 'outputs' schema: %s",
+            branch.label,
+            exc,
+        )
+        return False
+
+
 def _resolve_task_models(
     task: TaskDefinition,
     model_keys: list[str],
@@ -1078,7 +1132,7 @@ async def _branch_record(tr: ToolResult) -> None:
                                 on_tool_end=on_tool_end_hook, on_tool_start=on_tool_start_hook
                             )
                             branch_record = record_tool_result
-                        return await deploy_task_agents(
+                        branch_ok = await deploy_task_agents(
                             available_tools,
                             branch.agents,
                             branch.prompt,
@@ -1100,6 +1154,19 @@ async def _branch_record(tr: ToolResult) -> None:
                             agent_hooks=TaskAgentHooks(on_handoff=on_handoff_hook),
                             stream_label=branch.label if multi_model else None,
                         )
+                        # Typed multi-model output: validate this branch's
+                        # result against the task's `outputs` schema. A missing
+                        # or schema-violating result makes it a failed branch,
+                        # folded into the completion policy alongside run
+                        # failures; the coerced value feeds fan-in.
+                        if multi_model and task_output_schema:
+                            return _capture_branch_output(
+                                branch,
+                                task_output_schema,
+                                task_output_id or task_name,
+                                agent_ok=branch_ok,
+                            )
+                        return branch_ok
 
                     work_items = [(branch, branch.rm) for branch in branches]
                     complete = await _fan_out_deploys(

tests/test_models.py

@@ -220,9 +220,12 @@ def test_over_allowed_on_multi_model_repeat(self):
         )
         assert t.over == "outputs.x.items"
 
-    def test_outputs_schema_rejected_on_multi_model(self):
-        with pytest.raises(ValidationError, match="not yet supported on multi-model"):
-            TaskDefinition(user_prompt="hi", models=["a", "b"], outputs={"f": "str"})
+    def test_outputs_schema_allowed_on_multi_model(self):
+        # Typed outputs are now supported on multi-model tasks: the schema is
+        # applied per branch (see runner fan-in), so construction must succeed.
+        t = TaskDefinition(user_prompt="hi", models=["a", "b"], outputs={"f": "str"})
+        assert t.outputs == {"f": "str"}
+        assert [e.model for e in t.models] == ["a", "b"]
 
 
 class TestTaskflowDocument:

tests/test_runner_integration.py

@@ -354,3 +354,60 @@ def test_upstream_skip_then_downstream_if_still_completes(self, monkeypatch, tmp
         data = _session_data(tmp_path)
         assert data["finished"] is True
         assert all(t["skipped"] for t in data["completed_tasks"])
+
+
+class TestTypedMultiModelOutputs:
+    """Typed `outputs` schema applied per branch on multi-model tasks."""
+
+    @staticmethod
+    def _valid_deploy():
+        async def _impl(available_tools, agents, prompt, *, model=None, record_tool_result=None, **kw):
+            if record_tool_result is not None:
+                await record_tool_result(ToolResult(text=json.dumps({"score": 7})))
+            return True
+
+        return _impl
+
+    @staticmethod
+    def _mixed_deploy():
+        # m1 satisfies the schema; every other model violates it (missing `score`).
+        async def _impl(available_tools, agents, prompt, *, model=None, record_tool_result=None, **kw):
+            if record_tool_result is not None:
+                payload = {"score": 1} if model == "m1" else {"wrong": "x"}
+                await record_tool_result(ToolResult(text=json.dumps(payload)))
+            return True
+
+        return _impl
+
+    def test_typed_fanin_validates_each_branch(self, monkeypatch, tmp_path):
+        task = TaskDefinition(
+            id="cmp", agents=["pkg.p"], user_prompt="score X",
+            models=["m1", "m2"], outputs={"score": "int"},
+        )
+        _run(monkeypatch, tmp_path, _taskflow(task), deploy_impl=self._valid_deploy())
+        records = _session_outputs(tmp_path)["cmp"]
+        assert [r["model"] for r in records] == ["m1", "m2"]
+        # Each branch result is the validated/coerced schema value, not the raw envelope.
+        assert all(r["result"] == {"score": 7} for r in records)
+        assert _session_data(tmp_path)["completed_tasks"][0]["result"] is True
+
+    def test_schema_violation_fails_branch_completion_all(self, monkeypatch, tmp_path):
+        task = TaskDefinition(
+            id="cmp", agents=["pkg.p"], user_prompt="score X",
+            models=["m1", "m2"], outputs={"score": "int"}, completion="all",
+        )
+        _run(monkeypatch, tmp_path, _taskflow(task), deploy_impl=self._mixed_deploy())
+        by_model = {r["model"]: r["result"] for r in _session_outputs(tmp_path)["cmp"]}
+        assert by_model["m1"] == {"score": 1}
+        assert by_model["m2"] is None  # violating branch stored as None
+        # completion=all: the invalid branch fails the whole task.
+        assert _session_data(tmp_path)["completed_tasks"][0]["result"] is False
+
+    def test_schema_violation_tolerated_completion_any(self, monkeypatch, tmp_path):
+        task = TaskDefinition(
+            id="cmp", agents=["pkg.p"], user_prompt="score X",
+            models=["m1", "m2"], outputs={"score": "int"}, completion="any",
+        )
+        _run(monkeypatch, tmp_path, _taskflow(task), deploy_impl=self._mixed_deploy())
+        # completion=any: one schema-valid branch is enough for task success.
+        assert _session_data(tmp_path)["completed_tasks"][0]["result"] is True