feat(delegation): mark max_iterations-truncated subagent results for the parent (#86641)

A delegated subagent that exhausts its per-child iteration budget
(delegation.max_iterations) still returns a summary, so the result carries
status='completed' even though the child's exit_reason is 'max_iterations' and
its work was cut off mid-task. The parent then reads 'completed', trusts the
partial summary, and only discovers the truncation by parsing the prose (where
the child happens to mention 'hit the iteration limit'). That wastes parent
turns and risks acting on incomplete work.

exit_reason is already computed authoritatively and threaded to every
parent-visible surface; it just wasn't reflected anywhere the parent reads at a
glance. This surfaces it:

- delegate_tool.py: add a parent-visible boolean 'truncated' (= exit_reason ==
  'max_iterations') to each task entry, alongside the existing exit_reason.
- process_registry._format_async_delegation: for both the batch and single-task
  paths, when truncated -> use a warning icon, append
  'TRUNCATED: hit max_iterations — work may be incomplete' to the header/Status
  line, and prefix the summary with an unmissable truncation notice. status
  semantics are left unchanged (stays 'completed') so existing icon/summary
  branch logic and ~10 tests asserting status=='completed' stay valid.

Tests: single-task truncated -> banner; single-task clean -> no banner; batch
marks only the truncated task, not its clean sibling. 23/23 in the async-
delegation suite.

Co-authored-by: Teknium <teknium1@users.noreply.github.com>

tests/tools/test_async_delegation.py

@@ -761,3 +761,67 @@ def test_gateway_cli_origin_event_left_unrouted():
     runner._enrich_async_delegation_routing(evt)
     assert "platform" not in evt
 
+
+def test_single_task_truncation_banner_when_max_iterations():
+    """A single async subagent that hit its iteration cap (exit_reason=
+    max_iterations) must surface a TRUNCATED marker in the formatted result,
+    even though status stays 'completed' (a summary exists)."""
+    evt = _make_async_evt(
+        status="completed",
+        summary="Did part of the work then ran out of budget.",
+        exit_reason="max_iterations",
+    )
+    text = format_process_notification(evt)
+    assert text is not None
+    assert "TRUNCATED" in text
+    assert "max_iterations" in text
+    # The summary is still shown, just flagged.
+    assert "Did part of the work" in text
+
+
+def test_single_task_no_banner_when_clean():
+    """A cleanly-finished subagent must NOT get a truncation banner."""
+    evt = _make_async_evt(status="completed", summary="All done.", exit_reason="completed")
+    text = format_process_notification(evt)
+    assert text is not None
+    assert "TRUNCATED" not in text
+
+
+def test_batch_truncation_banner_marks_only_truncated_task():
+    """In a batch, only the task that hit max_iterations gets the TRUNCATED
+    marker; a clean sibling keeps the normal check icon."""
+    evt = _make_async_evt(
+        is_batch=True,
+        goals=["clean task", "truncated task"],
+        results=[
+            {
+                "task_index": 0,
+                "status": "completed",
+                "summary": "finished cleanly",
+                "api_calls": 5,
+                "exit_reason": "completed",
+                "truncated": False,
+            },
+            {
+                "task_index": 1,
+                "status": "completed",
+                "summary": "cut off mid-work",
+                "api_calls": 250,
+                "exit_reason": "max_iterations",
+                "truncated": True,
+            },
+        ],
+    )
+    text = format_process_notification(evt)
+    assert text is not None
+    assert "TRUNCATED" in text
+    # The clean task's summary and the truncated one's both render...
+    assert "finished cleanly" in text
+    assert "cut off mid-work" in text
+    # ...but the banner is tied to the truncated task, not the clean one.
+    trunc_pos = text.index("cut off mid-work")
+    clean_pos = text.index("finished cleanly")
+    banner_pos = text.index("TRUNCATED")
+    # The header banner for task 2 appears after task 1's summary.
+    assert banner_pos > clean_pos
+

tools/delegate_tool.py

@@ -2870,6 +2870,13 @@ def _run_single_child(
             "duration_seconds": duration,
             "model": _model if isinstance(_model, str) else None,
             "exit_reason": exit_reason,
+            # Explicit, parent-visible truncation flag. A subagent that
+            # exhausts its per-child iteration budget still returns a summary,
+            # so `status` stays "completed" (see above) — without this the
+            # parent can't tell truncated-but-summarized from cleanly-finished
+            # work except by parsing the summary prose. exit_reason is computed
+            # authoritatively from the child's `completed` flag.
+            "truncated": exit_reason == "max_iterations",
             "tokens": {
                 "input": (
                     _input_tokens if isinstance(_input_tokens, (int, float)) else 0

tools/process_registry.py

@@ -2695,6 +2695,7 @@ def _format_async_delegation(evt: dict) -> str:
     error = evt.get("error")
     api_calls = evt.get("api_calls", 0)
     duration = evt.get("duration_seconds", "?")
+    truncated = evt.get("truncated") or evt.get("exit_reason") == "max_iterations"
     dispatched_at = evt.get("dispatched_at")
     completed_at = evt.get("completed_at") or _time.time()
 
@@ -2735,7 +2736,8 @@ def _format_async_delegation(evt: dict) -> str:
             r_summary = r.get("summary")
             r_error = r.get("error")
             r_goal = goals[idx] if idx < len(goals) else r.get("goal", "")
-            icon = "✓" if r_status in ("completed", "success") else "✗"
+            r_truncated = r.get("truncated") or r.get("exit_reason") == "max_iterations"
+            icon = "⚠" if r_truncated else ("✓" if r_status in ("completed", "success") else "✗")
             lines.append("")
             header = f"--- {icon} TASK {idx + 1}/{n}"
             if r_goal:
@@ -2745,9 +2747,17 @@ def _format_async_delegation(evt: dict) -> str:
                 header += f", api_calls={r['api_calls']}"
             if r.get("duration_seconds") is not None:
                 header += f", {r['duration_seconds']}s"
+            if r_truncated:
+                header += ", TRUNCATED: hit max_iterations — work may be incomplete"
             header += ") ---"
             lines.append(header)
             if r_status in ("completed", "success") and r_summary:
+                if r_truncated:
+                    lines.append(
+                        "[TRUNCATED — subagent hit its iteration cap; the "
+                        "summary below may be incomplete. Verify before relying "
+                        "on it, or re-dispatch the unfinished part.]"
+                    )
                 lines.append(r_summary)
             elif r_summary:
                 if r_error:
@@ -2787,9 +2797,16 @@ def _format_async_delegation(evt: dict) -> str:
     if toolsets:
         lines.append(f"Toolsets: {', '.join(toolsets)}")
     lines.append(f"Role: {role}   Model: {model}")
-    lines.append(f"Status: {status}   API calls: {api_calls}   Duration: {duration}s")
+    _trunc = " [TRUNCATED: hit max_iterations — work may be incomplete]" if truncated else ""
+    lines.append(f"Status: {status}   API calls: {api_calls}   Duration: {duration}s{_trunc}")
     lines.append("--- RESULT ---")
     if status in ("completed", "success") and summary:
+        if truncated:
+            lines.append(
+                "[TRUNCATED — subagent hit its iteration cap; the summary below "
+                "may be incomplete. Verify before relying on it, or re-dispatch "
+                "the unfinished part.]"
+            )
         lines.append(summary)
     elif status == "interrupted":
         lines.append(