fix(runner): address review feedback on outputs docs, empty iterables, exit code

- Correct stale "validated/coerced" wording in comments/docstrings/grammar docs:
  output-schema validation is strict and does not coerce.
- Convert the remaining old-DSL `outputs` snippet in the `if` docs to JSON Schema.
- Fix the `_build_prompts_to_run` docstring: a non-iterable derived value raises
  TypeError (from iter/list), not ValueError.
- Materialize a repeat_prompt `over:` iterable before the emptiness check so a
  one-shot generator (e.g. Jinja map/select) is detected as empty instead of
  silently producing zero prompts.
- Raise instead of break on a must_complete task failure so run_main propagates
  and the CLI exits non-zero, matching the other hard-failure paths; the session
  is still saved for --resume.

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

Author: anticomputer

doc/GRAMMAR.md

@@ -178,7 +178,7 @@ Notes and semantics:
   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 applied per branch on multi-model tasks: each
-  branch's result is validated/coerced against the schema and stored as the
+  branch's result is validated 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").
@@ -212,7 +212,11 @@ as skipped and not run); otherwise it runs normally.
       user_prompt: |
         Audit this code and report findings as JSON: {"findings": [...]}
       outputs:
-        findings: list[any]
+        type: object
+        properties:
+          findings:
+            type: array
+        required: [findings]
   - task:
       # only remediate when the audit actually found something
       if: "outputs.audit.findings | length > 0"

src/seclab_taskflow_agent/models.py

@@ -122,7 +122,7 @@ class TaskDefinition(BaseModel):
     model: str = ""
     model_settings: dict[str, Any] = Field(default_factory=dict)
     # Typed named outputs (M2): declare a schema for this task's produced
-    # value; when set, the captured output is validated/coerced and exposed to
+    # value; when set, the captured output is validated and exposed to
     # later tasks as ``outputs.<id>``.
     outputs: dict[str, Any] = Field(default_factory=dict)
     # Explicit iterable selector for repeat_prompt: a Jinja expression

src/seclab_taskflow_agent/runner.py

@@ -235,7 +235,7 @@ class _Branch:
     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
+    # each branch's result is validated and the 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.
@@ -250,7 +250,7 @@ def _aggregate_fanin(branches: list[_Branch]) -> list[dict[str, Any]]:
     "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). When the task declares an ``outputs`` schema, ``result`` is
-    the per-branch validated/coerced value instead (``None`` for a branch that
+    the per-branch validated 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>``.
     """
@@ -375,7 +375,7 @@ def _capture_task_output(
     """Capture a task's final tool result as a named typed output.
 
     The task's output is its most recent tool result. When *schema* is
-    non-empty the decoded value is validated/coerced against it (raising a
+    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>``.
@@ -501,7 +501,8 @@ async def _build_prompts_to_run(
 
     Raises:
         IndexError: If the legacy path has no previous tool result.
-        ValueError: If the derived value is not valid JSON or not iterable.
+        ValueError: If the legacy path's tool result is not valid JSON.
+        TypeError: If the derived value is not iterable.
     """
     outputs = outputs or {}
     prompts_to_run: list[str] = []
@@ -530,17 +531,22 @@ async def _build_prompts_to_run(
                 raise IndexError("No last tool result available for repeat_prompt")
             iterable_result = decode_tool_result(last)
 
+        # Materialise before testing/iterating: an ``over:`` expression may
+        # yield a one-shot iterator (e.g. Jinja ``map``/``select`` filters),
+        # which is always truthy and would defeat the emptiness check and be
+        # consumed by a single pass. ``list(...)`` also raises TypeError for a
+        # genuinely non-iterable value.
         try:
-            iter(iterable_result)
+            items = list(iterable_result)
         except TypeError:
             logging.critical("repeat_prompt iterable is not iterable: %r", iterable_result)
             raise
 
-        if not iterable_result:
+        if not items:
             await render_model_output("** 🤖❗repeat_prompt iterable is empty!\n")
         else:
-            logging.debug("Rendering templated prompts for results: %s", iterable_result)
-            for value in iterable_result:
+            logging.debug("Rendering templated prompts for results: %s", items)
+            for value in items:
                 try:
                     rendered_prompt = render_template(
                         template_str=task_prompt,
@@ -1240,11 +1246,15 @@ async def _branch_record(tr: ToolResult) -> None:
                         f"** 🤖💾 Session saved: {session.session_id}\n"
                         f"** 🤖💡 Resume with: --resume {session.session_id}\n"
                     )
-                    break
+                    # Raise (rather than break) so run_main propagates the
+                    # failure and the CLI exits non-zero, matching the other
+                    # hard-failure paths; the session is already saved for
+                    # --resume.
+                    raise RuntimeError(f"Required task {task_name!r} did not complete")
 
                 # Capture this task's typed named output (M2). The task's
                 # output is its final tool result; when an ``outputs`` schema
-                # is declared it is validated/coerced before being stored under
+                # is declared it is validated before being stored under
                 # ``outputs.<id>`` for downstream tasks to consume by name. A
                 # declared output contract that the produced value violates is
                 # a hard failure, surfaced with the same session-saved / resume

tests/test_runner.py

@@ -455,6 +455,31 @@ def test_repeat_with_empty_iterable(self):
         )
         assert prompts == []
 
+    def test_repeat_over_empty_generator_emits_empty_notice(self):
+        """An `over:` expression yielding an empty one-shot generator (e.g. a
+        Jinja `map`/`select` filter) is detected as empty rather than silently
+        producing zero prompts (a bare generator is always truthy)."""
+        rendered: list[str] = []
+
+        async def _capture(text: str, **_kw: Any) -> None:
+            rendered.append(text)
+
+        with patch("seclab_taskflow_agent.runner.render_model_output", _capture):
+            prompts = asyncio.run(
+                _build_prompts_to_run(
+                    task_prompt="do {{ result }}",
+                    repeat_prompt=True,
+                    store=ResultStore(),
+                    available_tools=_mock_available_tools(),
+                    global_variables={},
+                    inputs={},
+                    outputs={"items": []},
+                    over="outputs.items | map('upper')",
+                )
+            )
+        assert prompts == []
+        assert any("iterable is empty" in t for t in rendered)
+
     def test_raises_index_error_when_no_last_result(self):
         """IndexError when the store has no previous result."""
         with pytest.raises(IndexError):

tests/test_runner_integration.py

@@ -17,6 +17,8 @@
 from typing import ClassVar
 from unittest.mock import MagicMock
 
+import pytest
+
 from seclab_taskflow_agent.models import (
     PersonalityDocument,
     TaskDefinition,
@@ -425,3 +427,20 @@ def test_schema_violation_tolerated_completion_any(self, monkeypatch, tmp_path):
         _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
+
+
+class TestMustCompleteFailure:
+    """A failed `must_complete` task aborts non-silently so the CLI exits non-zero."""
+
+    def test_must_complete_failure_raises_and_marks_failed(self, monkeypatch, tmp_path):
+        async def failing_deploy(available_tools, agents, prompt, *, model=None, record_tool_result=None, **kw):
+            return False  # the task never completes
+
+        task = TaskDefinition(agents=["pkg.p"], user_prompt="hi", must_complete=True)
+        with pytest.raises(RuntimeError, match="did not complete"):
+            _run(monkeypatch, tmp_path, _taskflow(task), deploy_impl=failing_deploy)
+
+        # The session is still saved and marked failed (so --resume works).
+        data = _session_data(tmp_path)
+        assert data["finished"] is False
+        assert "did not complete" in data["error"]