fix(runner): capture a plain task's output from its own result, not the store

_capture_task_output read the run-global store.last(), so a non-fan-out task
that produced no tool result silently captured the previous task's result as
its own outputs.<id> (and could validate it against this task's schema). Pass
the task's own last result explicitly -- its single branch's last tool result,
or the shell result -- so an empty producer yields None (or a hard failure when
a schema is declared) instead of leaking prior-task data. The fan-out path
already read per-branch sinks and was unaffected.

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

Author: anticomputer

src/seclab_taskflow_agent/runner.py

@@ -408,16 +408,19 @@ def _capture_task_output(
     output_id: str,
     schema: dict[str, Any],
     task_name: str,
+    last: ToolResult | None,
 ) -> None:
-    """Capture a task's final tool result as a named typed output.
+    """Capture a non-fan-out task's own final tool result as a named output.
 
-    The task's output is its most recent tool result. When *schema* is
-    non-empty the decoded value is validated against it (raising a
+    *last* is this task's produced result (its branch's last tool result, or
+    the shell result), **not** the run-global store's last entry -- so a task
+    that produced nothing yields ``None`` (or a hard failure when a schema is
+    declared) rather than silently leaking the previous task's result. When
+    *schema* is non-empty the decoded value is validated against it (raising a
     clear error on mismatch); otherwise the decoded value is stored as-is,
     falling back to the raw text when it is not JSON. The result is exposed to
     later tasks as ``outputs.<output_id>``.
     """
-    last = store.last()
     if last is None:
         if schema:
             raise ValueError(
@@ -1329,9 +1332,19 @@ async def _branch_record(tr: ToolResult) -> None:
                     if fans_out:
                         store.set_output(task_output_id, _aggregate_fanin(captured_branches))
                     else:
+                        # This task's own last result: the single branch's last
+                        # tool result for a model task, or the shell result for a
+                        # shell task (which has no branches and wrote directly to
+                        # the store). Never the run-global store.last(), so an
+                        # empty producer does not leak the previous task's result.
+                        if captured_branches:
+                            _sink = captured_branches[-1].sink
+                            _own_last = _sink[-1] if _sink else None
+                        else:
+                            _own_last = store.last()
                         try:
                             _capture_task_output(
-                                store, task_output_id, task_output_schema, task_name
+                                store, task_output_id, task_output_schema, task_name, _own_last
                             )
                         except Exception as exc:
                             logging.error("Task %r output capture failed: %s", task_name, exc)

tests/test_runner.py

@@ -814,13 +814,13 @@ class TestCaptureTaskOutput:
     def test_capture_schemaless_decodes_json(self):
         store = ResultStore()
         store.record(ToolResult(text=json.dumps({"functions": [1, 2]})))
-        _capture_task_output(store, "list_fns", {}, "task-0")
+        _capture_task_output(store, "list_fns", {}, "task-0", store.last())
         assert store.outputs["list_fns"] == {"functions": [1, 2]}
 
     def test_capture_schemaless_falls_back_to_text(self):
         store = ResultStore()
         store.record(ToolResult(text="a plain answer"))
-        _capture_task_output(store, "answer", {}, "task-0")
+        _capture_task_output(store, "answer", {}, "task-0", store.last())
         assert store.outputs["answer"] == "a plain answer"
 
     def test_capture_with_schema_validates(self):
@@ -840,7 +840,7 @@ def test_capture_with_schema_validates(self):
             },
             "required": ["functions"],
         }
-        _capture_task_output(store, "list_fns", schema, "task-0")
+        _capture_task_output(store, "list_fns", schema, "task-0", store.last())
         assert store.outputs["list_fns"] == {"functions": [{"name": "f", "body": "b"}]}
 
     def test_capture_with_schema_validation_error(self):
@@ -854,23 +854,23 @@ def test_capture_with_schema_validation_error(self):
             "required": ["functions"],
         }
         with pytest.raises(ValidationError):
-            _capture_task_output(store, "list_fns", schema, "task-0")
+            _capture_task_output(store, "list_fns", schema, "task-0", store.last())
 
     def test_capture_no_result_with_schema_raises(self):
         store = ResultStore()
         schema = {"type": "object", "properties": {"f": {"type": "string"}}, "required": ["f"]}
         with pytest.raises(ValueError, match="produced no tool result"):
-            _capture_task_output(store, "x", schema, "task-0")
+            _capture_task_output(store, "x", schema, "task-0", None)
 
     def test_capture_no_result_schemaless_sets_none(self):
         store = ResultStore()
-        _capture_task_output(store, "x", {}, "task-0")
+        _capture_task_output(store, "x", {}, "task-0", None)
         assert store.outputs["x"] is None
 
     def test_capture_prefers_structured(self):
         store = ResultStore()
         store.record(ToolResult(structured={"functions": ["a"]}))
-        _capture_task_output(store, "out", {}, "task-0")
+        _capture_task_output(store, "out", {}, "task-0", store.last())
         assert store.outputs["out"] == {"functions": ["a"]}
 
 

tests/test_runner_integration.py

@@ -531,3 +531,21 @@ def test_shell_task_never_fans_out(self, monkeypatch, tmp_path):
         task = TaskDefinition(id="sh", run='echo \'{"x": 1}\'', models=["a", "b"])
         _run(monkeypatch, tmp_path, _taskflow(task))
         assert _session_outputs(tmp_path)["sh"] == {"x": 1}
+
+    def test_plain_task_with_no_result_does_not_leak_prior(self, monkeypatch, tmp_path):
+        # Task A produces a result; Task B produces no tool result. B's output
+        # must be None (its own empty result), never A's result via the
+        # run-global store.
+        async def deploy(available_tools, agents, prompt, *, model=None, record_tool_result=None, **kw):
+            if record_tool_result is not None and prompt.startswith("produce"):
+                await record_tool_result(ToolResult(text=json.dumps({"a": 1})))
+            return True
+
+        tasks = [
+            TaskDefinition(id="a", agents=["pkg.p"], user_prompt="produce a"),
+            TaskDefinition(id="b", agents=["pkg.p"], user_prompt="answer without a tool"),
+        ]
+        _run(monkeypatch, tmp_path, _taskflow_tasks(tasks), deploy_impl=deploy)
+        outputs = _session_outputs(tmp_path)
+        assert outputs["a"] == {"a": 1}
+        assert outputs["b"] is None