From 07699f74010aebe1955a9e6bd8cffc62caa1be81 Mon Sep 17 00:00:00 2001
From: Charles Cha <92324143+ypwcharles@users.noreply.github.com>
Date: Fri, 17 Jul 2026 00:51:32 +0800
Subject: [PATCH 1/3] fix(delegation): preserve redacted output on timeout

When a delegate_task child times out after completing real work, preserve a bounded tail of completed tool results so the parent can continue from verified evidence instead of receiving only an empty timeout summary.

Redact each complete tool result with force=True before truncation, cap the evidence to 8 entries and 600 characters per preview, and fail closed at the timeout caller if sanitization cannot be completed. Preserve the legacy schema for timeouts before the first model call and for ordinary exception paths.

Render the sanitized tail through the existing foreground and background notification paths. The original #65824 already covered background batch and single-result rendering; this current-main rebuild preserves those paths rather than introducing them as a separate salvage feature.

Author and implementation ownership remain with Charles Cha (@ypwcharles), from the original #65824.
---
 tests/tools/test_delegate.py                  | 345 ++++++++++++++++++
 ...st_delegate_subagent_timeout_diagnostic.py | 217 ++++++++++-
 tools/delegate_tool.py                        |  67 +++-
 tools/process_registry.py                     |  21 ++
 4 files changed, 639 insertions(+), 11 deletions(-)

diff --git a/tests/tools/test_delegate.py b/tests/tools/test_delegate.py
index bb995dda599ce..0619b1937f42b 100644
--- a/tests/tools/test_delegate.py
+++ b/tests/tools/test_delegate.py
@@ -23,6 +23,7 @@
     DelegateEvent,
     _get_max_concurrent_children,
     _load_config,
+    _extract_output_tail,
     delegate_task,
     _build_child_agent,
     _build_child_progress_callback,
@@ -467,6 +468,45 @@ def test_tool_trace_handles_list_content_blocks(self):
             self.assertEqual(trace[0]["status"], "ok")
             self.assertGreater(trace[0]["result_bytes"], 0)
 
+
+    def test_output_tail_preserves_plain_output_and_redacts_secrets(self):
+        secret = "sk-" + ("s" * 30)
+        result = {
+            "messages": [
+                {"role": "assistant", "tool_calls": [
+                    {"id": "t1", "function": {"name": "terminal", "arguments": "{}"}},
+                    {"id": "t2", "function": {"name": "terminal", "arguments": "{}"}},
+                ]},
+                {"role": "tool", "tool_call_id": "t1", "content": "focused test passed"},
+                {"role": "tool", "tool_call_id": "t2", "content": f"token={secret}"},
+            ]
+        }
+
+        tail = _extract_output_tail(result, max_entries=8, max_chars=600)
+
+        self.assertEqual(tail[0]["preview"], "focused test passed")
+        self.assertEqual(tail[1]["preview"], "token=***")
+        self.assertNotIn(secret, str(tail))
+
+    def test_output_tail_redaction_error_drops_entire_tail(self):
+        result = {
+            "messages": [
+                {"role": "assistant", "tool_calls": [
+                    {"id": "t1", "function": {"name": "terminal", "arguments": "{}"}},
+                ]},
+                {"role": "tool", "tool_call_id": "t1", "content": "sensitive output"},
+            ]
+        }
+
+        with patch(
+            "agent.redact.redact_sensitive_text",
+            side_effect=RuntimeError("redactor unavailable"),
+        ):
+            tail = _extract_output_tail(result, max_entries=8, max_chars=600)
+
+        self.assertEqual(tail, [])
+
+
     def test_parallel_tool_calls_paired_correctly(self):
         """Parallel tool calls should each get their own result via tool_call_id matching."""
         parent = _make_mock_parent(depth=0)
@@ -1748,5 +1788,310 @@ def test_child_gets_no_fallback_when_parent_chain_empty(self):
         self.assertIsNone(kwargs["fallback_model"])
 
 
+class TestFormatAsyncDelegationPartialTail(unittest.TestCase):
+    """Regression tests for partial_output_tail rendering in _format_async_delegation.
+
+    When a timed-out subagent has produced tool output before the timeout,
+    the completion event carries a ``partial_output_tail`` list of
+    ``{tool, preview, is_error}`` dicts.  ``_format_async_delegation`` must
+    render these into the re-injection block so the parent agent sees what
+    the child accomplished before it was killed.
+    """
+
+    def _batch_evt(self, results, goals=None):
+        """Build a minimal batch async_delegation event."""
+        return {
+            "type": "async_delegation",
+            "delegation_id": "test-batch-001",
+            "is_batch": True,
+            "goals": goals or [f"Goal {i}" for i in range(len(results))],
+            "results": results,
+            "dispatched_at": 1700000000,
+            "completed_at": 1700000060,
+        }
+
+    def _single_evt(self, **overrides):
+        """Build a minimal single-task async_delegation event."""
+        base = {
+            "type": "async_delegation",
+            "delegation_id": "test-single-001",
+            "goal": "Research something",
+            "status": "timeout",
+            "error": "Subagent timed out after 120s",
+            "api_calls": 3,
+            "duration_seconds": 120.5,
+            "dispatched_at": 1700000000,
+            "completed_at": 1700000120,
+        }
+        base.update(overrides)
+        return base
+
+    def test_batch_timeout_renders_partial_output_tail(self):
+        """A timed-out batch result with partial_output_tail must render it."""
+        from tools.process_registry import _format_async_delegation
+
+        tail = [
+            {"tool": "terminal", "preview": "Build succeeded", "is_error": False},
+            {"tool": "web_search", "preview": "Found 3 results", "is_error": False},
+        ]
+        results = [
+            {
+                "task_index": 0,
+                "status": "timeout",
+                "summary": None,
+                "error": "Subagent timed out",
+                "api_calls": 2,
+                "duration_seconds": 120,
+                "partial_output_tail": tail,
+            },
+        ]
+        out = _format_async_delegation(self._batch_evt(results))
+
+        assert "Partial output (redacted):" in out
+        assert "[terminal] Build succeeded" in out
+        assert "[web_search] Found 3 results" in out
+
+    def test_batch_success_without_tail_omits_tail_section(self):
+        """A successful batch result must not contain partial_output_tail text."""
+        from tools.process_registry import _format_async_delegation
+
+        results = [
+            {
+                "task_index": 0,
+                "status": "completed",
+                "summary": "All done",
+                "api_calls": 3,
+                "duration_seconds": 10,
+            },
+        ]
+        out = _format_async_delegation(self._batch_evt(results))
+
+        assert "Partial output (redacted):" not in out
+
+    def test_batch_timeout_without_tail_omits_tail_section(self):
+        """A timed-out result with no partial_output_tail must not render it."""
+        from tools.process_registry import _format_async_delegation
+
+        results = [
+            {
+                "task_index": 0,
+                "status": "timeout",
+                "summary": None,
+                "error": "Timed out",
+                "api_calls": 0,
+                "duration_seconds": 120,
+            },
+        ]
+        out = _format_async_delegation(self._batch_evt(results))
+
+        assert "Partial output (redacted):" not in out
+
+    def test_batch_mixed_statuses_render_tail_only_for_timed_out(self):
+        """Only the timed-out task should get tail rendering, not the completed one."""
+        from tools.process_registry import _format_async_delegation
+
+        tail = [{"tool": "terminal", "preview": "partial work", "is_error": False}]
+        results = [
+            {
+                "task_index": 0,
+                "status": "completed",
+                "summary": "Task A done",
+                "api_calls": 2,
+                "duration_seconds": 5,
+            },
+            {
+                "task_index": 1,
+                "status": "timeout",
+                "summary": None,
+                "error": "Timed out",
+                "api_calls": 1,
+                "duration_seconds": 120,
+                "partial_output_tail": tail,
+            },
+        ]
+        out = _format_async_delegation(self._batch_evt(results))
+
+        # Completed task should not have tail
+        assert "Task A done" in out
+        # Timed-out task should have tail
+        assert "[terminal] partial work" in out
+
+    def test_single_timeout_renders_partial_output_tail(self):
+        """A single-task timeout with partial_output_tail must render it."""
+        from tools.process_registry import _format_async_delegation
+
+        tail = [
+            {"tool": "terminal", "preview": "npm install done", "is_error": False},
+            {"tool": "terminal", "preview": "Error: build failed", "is_error": True},
+        ]
+        evt = self._single_evt(partial_output_tail=tail)
+        out = _format_async_delegation(evt)
+
+        assert "Partial output (redacted):" in out
+        assert "[terminal] npm install done" in out
+        assert "[terminal] Error: build failed" in out
+
+    def test_single_success_without_tail_omits_tail_section(self):
+        """A successful single-task event must not render partial_output_tail."""
+        from tools.process_registry import _format_async_delegation
+
+        evt = self._single_evt(
+            status="completed",
+            summary="All done",
+            error=None,
+        )
+        out = _format_async_delegation(evt)
+
+        assert "Partial output (redacted):" not in out
+
+    def test_single_timeout_without_tail_omits_tail_section(self):
+        """A timed-out single task with no partial_output_tail must not render it."""
+        from tools.process_registry import _format_async_delegation
+
+        evt = self._single_evt(partial_output_tail=None)
+        out = _format_async_delegation(evt)
+
+        assert "Partial output (redacted):" not in out
+
+    def test_single_error_with_tail_renders_tail(self):
+        """An errored single-task event with partial_output_tail must also render it."""
+        from tools.process_registry import _format_async_delegation
+
+        tail = [{"tool": "web_search", "preview": "query results", "is_error": False}]
+        evt = self._single_evt(
+            status="error",
+            error="RuntimeError: connection lost",
+            partial_output_tail=tail,
+        )
+        out = _format_async_delegation(evt)
+
+        assert "[web_search] query results" in out
+
+    def test_batch_empty_tail_list_omits_tail_section(self):
+        """An empty partial_output_tail list should not render the section."""
+        from tools.process_registry import _format_async_delegation
+
+        results = [
+            {
+                "task_index": 0,
+                "status": "timeout",
+                "summary": None,
+                "error": "Timed out",
+                "api_calls": 0,
+                "duration_seconds": 120,
+                "partial_output_tail": [],
+            },
+        ]
+        out = _format_async_delegation(self._batch_evt(results))
+
+        assert "Partial output (redacted):" not in out
+
+    def test_render_partial_output_tail_helper_format(self):
+        """The _render_partial_output_tail helper produces correct format."""
+        from tools.process_registry import _render_partial_output_tail
+
+        tail = [
+            {"tool": "terminal", "preview": "output line 1", "is_error": False},
+            {"tool": "web_search", "preview": "result", "is_error": False},
+        ]
+        lines = _render_partial_output_tail(tail)
+
+        assert lines[0] == "Partial output (redacted):"
+        assert lines[1] == "  [terminal] output line 1"
+        assert lines[2] == "  [web_search] result"
+
+    def test_batch_tail_preserves_existing_success_rendering(self):
+        """Successful results still show their summary, not a tail section."""
+        from tools.process_registry import _format_async_delegation
+
+        results = [
+            {
+                "task_index": 0,
+                "status": "completed",
+                "summary": "Research complete",
+                "api_calls": 5,
+                "duration_seconds": 30,
+            },
+        ]
+        out = _format_async_delegation(self._batch_evt(results))
+
+        assert "Research complete" in out
+        assert "Partial output (redacted):" not in out
+
+    def test_batch_tail_preserves_existing_interrupted_rendering(self):
+        """Interrupted results still show their partial output, plus tail if present."""
+        from tools.process_registry import _format_async_delegation
+
+        tail = [{"tool": "terminal", "preview": "interrupted work", "is_error": False}]
+        results = [
+            {
+                "task_index": 0,
+                "status": "interrupted",
+                "summary": "some output",
+                "error": "User interrupted",
+                "api_calls": 2,
+                "duration_seconds": 10,
+                "partial_output_tail": tail,
+            },
+        ]
+        out = _format_async_delegation(self._batch_evt(results))
+
+        # The existing "Partial output:" section is preserved
+        assert "Partial output:" in out
+        assert "some output" in out
+        # And the tail is also rendered
+        assert "Partial output (redacted):" in out
+        assert "[terminal] interrupted work" in out
+
+    # ------------------------------------------------------------------
+    # End-to-end regression through format_process_notification()
+    # ------------------------------------------------------------------
+
+    def test_format_process_notification_timeout_with_tail(self):
+        """Single-task timeout event routed through format_process_notification()
+        must render partial_output_tail as 'Partial output (redacted)'."""
+        from tools.process_registry import format_process_notification
+
+        tail = [
+            {"tool": "terminal", "preview": "Build succeeded", "is_error": False},
+            {"tool": "web_search", "preview": "Found 3 results", "is_error": False},
+        ]
+        evt = self._single_evt(partial_output_tail=tail)
+        out = format_process_notification(evt)
+
+        assert out is not None
+        assert "Partial output (redacted):" in out
+        assert "[terminal] Build succeeded" in out
+        assert "[web_search] Found 3 results" in out
+
+    def test_format_process_notification_batch_timeout_with_tail(self):
+        """Batch timeout event routed through format_process_notification()
+        must render partial_output_tail as 'Partial output (redacted)'."""
+        from tools.process_registry import format_process_notification
+
+        tail = [
+            {"tool": "terminal", "preview": "npm install done", "is_error": False},
+            {"tool": "terminal", "preview": "Error: build failed", "is_error": True},
+        ]
+        results = [
+            {
+                "task_index": 0,
+                "status": "timeout",
+                "summary": None,
+                "error": "Subagent timed out",
+                "api_calls": 2,
+                "duration_seconds": 120,
+                "partial_output_tail": tail,
+            },
+        ]
+        evt = self._batch_evt(results)
+        out = format_process_notification(evt)
+
+        assert out is not None
+        assert "Partial output (redacted):" in out
+        assert "[terminal] npm install done" in out
+        assert "[terminal] Error: build failed" in out
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/tools/test_delegate_subagent_timeout_diagnostic.py b/tests/tools/test_delegate_subagent_timeout_diagnostic.py
index 9d0fcad8c8bf5..c04d0b9bba1d3 100644
--- a/tests/tools/test_delegate_subagent_timeout_diagnostic.py
+++ b/tests/tools/test_delegate_subagent_timeout_diagnostic.py
@@ -41,6 +41,8 @@ def __init__(
         hang_seconds: float = 5.0,
         subagent_id: str = "sa-0-stubabc",
         tool_schema=None,
+        session_messages=None,
+        progress_callback=None,
     ):
         self._subagent_id = subagent_id
         self._delegate_depth = 1
@@ -62,6 +64,8 @@ def __init__(
             {"name": "terminal", "description": "shell"},
         ]
         self._api_call_count = api_call_count
+        self._session_messages = list(session_messages or [])
+        self.tool_progress_callback = progress_callback
         self._hang = threading.Event()
         self._hang_seconds = hang_seconds
 
@@ -116,7 +120,7 @@ def test_writes_log_with_expected_sections(self, hermes_home):
         assert p.name.startswith("subagent-timeout-sa-7-abc123-")
         assert p.suffix == ".log"
 
-        content = p.read_text()
+        content = p.read_text(encoding="utf-8")
         # Header references the issue for future grep-ability
         assert "issue #14726" in content
         # Timeout facts
@@ -158,7 +162,7 @@ def test_returns_none_on_unwritable_logs_dir(self, tmp_path, monkeypatch):
         # so mkdir(exist_ok=True) → NotADirectoryError and we fall through.
         bogus.parent.mkdir(parents=True, exist_ok=True)
         bogus.mkdir()
-        (bogus / "logs").write_text("not a dir")
+        (bogus / "logs").write_text("not a dir", encoding="utf-8")
         result = _dump_subagent_timeout_diagnostic(
             child=child,
             task_index=0,
@@ -238,3 +242,212 @@ def _boom(*a, **kw):
         assert result["timeout_seconds"] is None
         assert result["timed_out_after_seconds"] is None
         assert result["timeout_phase"] is None
+    @staticmethod
+    def _tool_messages(outputs):
+        messages = []
+        for index, output in enumerate(outputs):
+            tool_call_id = f"tool-{index}"
+            messages.extend(
+                [
+                    {
+                        "role": "assistant",
+                        "tool_calls": [
+                            {
+                                "id": tool_call_id,
+                                "function": {
+                                    "name": "terminal",
+                                    "arguments": "{}",
+                                },
+                            }
+                        ],
+                    },
+                    {
+                        "role": "tool",
+                        "tool_call_id": tool_call_id,
+                        "content": output,
+                    },
+                ]
+            )
+        return messages
+
+    def test_timeout_preserves_completed_tool_output_as_partial_tail(
+        self, hermes_home, monkeypatch
+    ):
+        child = _StubChild(
+            api_call_count=1,
+            hang_seconds=10.0,
+            session_messages=self._tool_messages(["focused test passed"]),
+        )
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        assert result["status"] == "timeout"
+        assert result["partial"] is True
+        assert result["partial_output_tail"] == [
+            {
+                "tool": "terminal",
+                "preview": "focused test passed",
+                "is_error": False,
+            }
+        ]
+
+    def test_timeout_redacts_secrets_from_partial_tail(
+        self, hermes_home, monkeypatch
+    ):
+        secret = "sk-" + ("a" * 30)
+        child = _StubChild(
+            api_call_count=1,
+            hang_seconds=10.0,
+            session_messages=self._tool_messages([f"token={secret}"]),
+        )
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        serialized_tail = str(result["partial_output_tail"])
+        assert secret not in serialized_tail
+        assert "token=" in serialized_tail
+
+    def test_timeout_partial_tail_is_bounded_by_count_and_preview_size(
+        self, hermes_home, monkeypatch
+    ):
+        child = _StubChild(
+            api_call_count=12,
+            hang_seconds=10.0,
+            session_messages=self._tool_messages(
+                [f"output-{index}:" + ("x" * 1000) for index in range(12)]
+            ),
+        )
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        tail = result["partial_output_tail"]
+        assert len(tail) == 8
+        assert all(0 < len(entry["preview"]) <= 600 for entry in tail)
+        assert tail[0]["preview"].startswith("output-4:")
+        assert tail[-1]["preview"].startswith("output-11:")
+
+    def test_timeout_without_tool_output_keeps_old_schema(
+        self, hermes_home, monkeypatch
+    ):
+        child = _StubChild(api_call_count=0, hang_seconds=10.0)
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        assert result["status"] == "timeout"
+        assert "partial" not in result
+        assert "partial_output_tail" not in result
+
+    def test_timeout_malformed_live_messages_fails_closed(
+        self, hermes_home, monkeypatch
+    ):
+        secret = "sk-" + ("m" * 30)
+        child = _StubChild(
+            api_call_count=1,
+            hang_seconds=10.0,
+            session_messages=[
+                {"role": "assistant", "tool_calls": [f"malformed-{secret}"]},
+                {"role": "tool", "content": f"must not leak {secret}"},
+            ],
+        )
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        assert result["status"] == "timeout"
+        assert "partial" not in result
+        assert "partial_output_tail" not in result
+        assert secret not in str(result)
+
+    def test_timeout_live_snapshot_error_fails_closed(
+        self, hermes_home, monkeypatch
+    ):
+        class _SnapshotErrorChild(_StubChild):
+            @property
+            def _session_messages(self):
+                raise RuntimeError("live transcript unavailable")
+
+            @_session_messages.setter
+            def _session_messages(self, value):
+                pass
+
+        child = _SnapshotErrorChild(api_call_count=1, hang_seconds=10.0)
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        assert result["status"] == "timeout"
+        assert "partial" not in result
+        assert "partial_output_tail" not in result
+
+    def test_timeout_output_tail_helper_error_fails_closed(
+        self, hermes_home, monkeypatch
+    ):
+        from tools import delegate_tool
+
+        secret = "sk-" + ("h" * 30)
+        events = []
+
+        def capture(event_type, **kwargs):
+            events.append((event_type, kwargs))
+
+        child = _StubChild(
+            api_call_count=1,
+            hang_seconds=10.0,
+            session_messages=self._tool_messages([f"must not leak {secret}"]),
+            progress_callback=capture,
+        )
+        monkeypatch.setattr(
+            delegate_tool,
+            "_extract_output_tail",
+            MagicMock(side_effect=RuntimeError("extractor failed")),
+        )
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        assert result["status"] == "timeout"
+        assert "partial" not in result
+        assert "partial_output_tail" not in result
+        complete_events = [payload for kind, payload in events if kind == "subagent.complete"]
+        assert len(complete_events) == 1
+        assert "output_tail" not in complete_events[0]
+        assert secret not in str((result, events))
+
+    def test_non_timeout_exception_never_exposes_partial_output(
+        self, hermes_home, monkeypatch
+    ):
+        class _ErrorChild(_StubChild):
+            def run_conversation(self, *args, **kwargs):
+                raise RuntimeError("provider failed")
+
+        child = _ErrorChild(
+            api_call_count=1,
+            session_messages=self._tool_messages(["completed before provider error"]),
+        )
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        assert result["status"] == "error"
+        assert "partial" not in result
+        assert "partial_output_tail" not in result
+
+    def test_timeout_progress_event_receives_only_redacted_output(
+        self, hermes_home, monkeypatch
+    ):
+        secret = "sk-" + ("a" * 30)
+        events = []
+
+        def capture(event_type, **kwargs):
+            events.append((event_type, kwargs))
+
+        child = _StubChild(
+            api_call_count=1,
+            hang_seconds=10.0,
+            session_messages=self._tool_messages([f"result with {secret}"]),
+            progress_callback=capture,
+        )
+
+        result = self._invoke_with_short_timeout(child, monkeypatch)
+
+        complete_events = [payload for kind, payload in events if kind == "subagent.complete"]
+        assert len(complete_events) == 1
+        event_tail = complete_events[0]["output_tail"]
+        assert event_tail == result["partial_output_tail"]
+        assert secret not in str(event_tail)
diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py
index e4f4766fca875..e6f34e88df11f 100644
--- a/tools/delegate_tool.py
+++ b/tools/delegate_tool.py
@@ -368,10 +368,23 @@ def _extract_output_tail(
         content = _stringify_tool_content(msg.get("content") or "")
         is_error = _looks_like_error_output(content)
         tool_name = pending_call_by_id.get(msg.get("tool_call_id") or "", "tool")
+        # Redact the complete tool output before truncating it. Truncating first
+        # could split a credential at the boundary and leave an unrecognisable
+        # (therefore unredacted) secret fragment in the preview.
+        try:
+            from agent.redact import redact_sensitive_text
+
+            safe_content = redact_sensitive_text(content, force=True)
+        except Exception:
+            logger.warning(
+                "Failed to redact delegated tool output; dropping output tail",
+                exc_info=True,
+            )
+            return []
         # Preserve line structure so the overlay's wrapped scroll region can
         # show real output rather than a whitespace-collapsed blob. We still
         # cap the payload size to keep events bounded.
-        preview = content[:max_chars]
+        preview = safe_content[:max_chars]
         tail.append({"tool": tool_name, "preview": preview, "is_error": is_error})
 
     tail.reverse()  # restore chronological order for display
@@ -2399,19 +2412,52 @@ def _run_with_thread_capture():
                         diagnostic_path,
                     )
 
+            # A timed-out child may already have completed useful tool calls.
+            # Snapshot its live in-memory transcript, then reuse the normal
+            # bounded/redacted overlay extractor. Never expose this evidence for
+            # ordinary exceptions: those retain the existing error schema.
+            partial_output_tail: List[Dict[str, Any]] = []
+            if is_timeout:
+                try:
+                    session_messages = getattr(child, "_session_messages", None)
+                    if isinstance(session_messages, list):
+                        partial_output_tail = [
+                            item
+                            for item in _extract_output_tail(
+                                {"messages": list(session_messages)},
+                                max_entries=8,
+                                max_chars=600,
+                            )
+                            if isinstance(item, dict)
+                            and isinstance(item.get("preview"), str)
+                            and item["preview"].strip()
+                        ]
+                except Exception:
+                    # This path is handling an existing timeout and must never
+                    # turn observability damage into a new error or expose an
+                    # unredacted transcript. Preserve the legacy timeout schema.
+                    logger.warning(
+                        "Failed to extract redacted delegated timeout output; "
+                        "dropping output tail",
+                        exc_info=True,
+                    )
+                    partial_output_tail = []
+
             if child_progress_cb:
                 try:
-                    child_progress_cb(
-                        "subagent.complete",
-                        preview=(
+                    complete_kwargs: Dict[str, Any] = {
+                        "preview": (
                             f"Timed out after {duration}s"
                             if is_timeout
                             else str(_timeout_exc)
                         ),
-                        status="timeout" if is_timeout else "error",
-                        duration_seconds=duration,
-                        summary="",
-                    )
+                        "status": "timeout" if is_timeout else "error",
+                        "duration_seconds": duration,
+                        "summary": "",
+                    }
+                    if partial_output_tail:
+                        complete_kwargs["output_tail"] = partial_output_tail
+                    child_progress_cb("subagent.complete", **complete_kwargs)
                 except Exception:
                     pass
 
@@ -2437,7 +2483,7 @@ def _run_with_thread_capture():
             else:
                 _err = str(_timeout_exc)
 
-            _error_entry = {
+            _error_entry: Dict[str, Any] = {
                 "task_index": task_index,
                 "status": "timeout" if is_timeout else "error",
                 "summary": None,
@@ -2461,6 +2507,9 @@ def _run_with_thread_capture():
                     " [steer did not land before the subagent stopped: "
                     f"{_late_pending_steer}]"
                 )
+            if partial_output_tail:
+                _error_entry["partial"] = True
+                _error_entry["partial_output_tail"] = partial_output_tail
             return _error_entry
         finally:
             # Shut down executor without waiting — if the child thread
diff --git a/tools/process_registry.py b/tools/process_registry.py
index 6d61b3ab690d3..88a8910d1e328 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -2660,6 +2660,21 @@ def _format_age(seconds: float) -> str:
     return f"{h}h" if m == 0 else f"{h}h{m}m"
 
 
+def _render_partial_output_tail(tail: list) -> list:
+    """Render a bounded partial_output_tail into human-readable lines.
+
+    Each entry is ``{tool, preview, is_error}`` produced by
+    ``_extract_output_tail`` in delegate_tool.py.  Returns a list of
+    formatted strings ready to join into the parent block.
+    """
+    out = ["Partial output (redacted):"]
+    for entry in tail:
+        tool_name = entry.get("tool", "tool")
+        preview = entry.get("preview", "")
+        out.append(f"  [{tool_name}] {preview}")
+    return out
+
+
 def _format_async_delegation(evt: dict) -> str:
     """Format an async-delegation completion into a self-contained re-injection.
 
@@ -2748,6 +2763,9 @@ def _format_async_delegation(evt: dict) -> str:
                     + (f": {r_error}" if r_error else "")
                     + ")"
                 )
+            r_tail = r.get("partial_output_tail")
+            if r_tail:
+                lines.extend(_render_partial_output_tail(r_tail))
             r_live = r.get("live_transcript")
             if r_live:
                 lines.append(
@@ -2796,6 +2814,9 @@ def _format_async_delegation(evt: dict) -> str:
         if summary:
             lines.append("Partial output:")
             lines.append(summary)
+    partial_tail = evt.get("partial_output_tail")
+    if partial_tail:
+        lines.extend(_render_partial_output_tail(partial_tail))
     return "\n".join(lines)
 
 

From d9fb954a8aee5f9f3a99b7f1461422184bdece4d Mon Sep 17 00:00:00 2001
From: Matvii Sakhnenko <matvey.sakhnenko03@icloud.com>
Date: Tue, 11 Aug 2026 22:41:41 +0200
Subject: [PATCH 2/3] fix(delegation): scope fail-closed tail to the timeout
 caller
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review on #84085 caught a real regression in the previous commit.

`_extract_output_tail` is SHARED. It powers the ordinary delegation
Output overlay on the non-timeout success path (delegate_tool.py, the
cc-swarm-parity feature), not only the new timeout tail. The blanket
`return []` on redaction failure meant a single redactor hiccup — a
config issue or one pattern edge case — silently blanked the entire
Output section for ALL delegations, where upstream degraded gracefully.

Confirmed by reproduction before changing anything: with a redactor
patched to raise, `_extract_output_tail` returned `[]` and the overlay
lost output it had shown moments earlier with a healthy redactor.

Add a `fail_closed` parameter, default False:

- Default (display callers, ordinary overlay): best-effort. Skip only
  the tool result that could not be redacted and keep rendering the
  rest, so one bad entry cannot destroy a working feature.
- `fail_closed=True` (timeout evidence call-site): unchanged hard fail.
  Evidence surfaced from a FAILED child is a security boundary, not a
  display nicety, so it emits nothing rather than risk an unredacted
  preview.

Verified with a three-result fixture where the redactor raises on the
middle one: the overlay keeps both healthy findings and drops the
unredactable one, while the fail-closed caller emits nothing.

Also address the review's test-gap suggestion with a truncation-boundary
case. The first attempt was VACUOUS — it passed even under
truncate-before-redact sabotage, because `sk-`/`ghp_`-style patterns
still match after truncation. Searched for a case that actually bites:
only patterns with a minimum-length quantifier are vulnerable. The Codex
pattern `gAAAA[A-Za-z0-9_=-]{20,}` cut to 25 chars drops below the
20-char body minimum, stops matching, and leaks
`credential gAAAABBBBBBBBB` in the clear.

Note the redactor elides the middle of a matched token
(`gAAAAB...BBBB`) rather than deleting it, so the test asserts on the
surviving raw body run rather than absence of the prefix. Non-vacuity
is proven: sabotaging the implementation to
`redact(content[:max_chars])` fails this test.

Tests: the previous fail-closed test is split into a default best-effort
case and an explicit `fail_closed=True` case, plus the boundary test.
`tests/tools/test_delegate.py` 81 passed; focused trio 168 passed / 0
failed. Full `tests/tools/` 6057 passed / 73 failed, the identical 73
failing on unmodified origin/main on this host — no new failures. The
end-to-end timeout reproduction still returns `partial=True` with the
child's real tool output.

Co-authored-by: Charles Cha <92324143+ypwcharles@users.noreply.github.com>
---
 tests/tools/test_delegate.py | 98 ++++++++++++++++++++++++++++++++++--
 tools/delegate_tool.py       | 32 +++++++++++-
 2 files changed, 125 insertions(+), 5 deletions(-)

diff --git a/tests/tools/test_delegate.py b/tests/tools/test_delegate.py
index 0619b1937f42b..a8a6c3eda30a3 100644
--- a/tests/tools/test_delegate.py
+++ b/tests/tools/test_delegate.py
@@ -488,13 +488,103 @@ def test_output_tail_preserves_plain_output_and_redacts_secrets(self):
         self.assertEqual(tail[1]["preview"], "token=***")
         self.assertNotIn(secret, str(tail))
 
-    def test_output_tail_redaction_error_drops_entire_tail(self):
+    def test_output_tail_redacts_secret_split_across_truncation_boundary(self):
+        """A secret cut below its pattern's minimum length must not leak.
+
+        This is the case that motivates redact-before-truncate, and it only
+        bites for patterns carrying a minimum-length quantifier. The Codex
+        encrypted-token pattern is ``gAAAA[A-Za-z0-9_=-]{20,}``: truncating
+        first can cut the body below 20 chars, at which point the fragment
+        stops matching, passes the redactor untouched, and lands in the
+        preview in the clear.
+
+        The redactor elides the middle of a matched token rather than
+        deleting it (``gAAAAB...BBBB``), so the invariant under test is the
+        length of the surviving RAW body run, not the absence of the prefix.
+
+        Non-vacuity verified: swapping the implementation to
+        ``redact(content[:max_chars])`` makes this assertion fail with
+        ``'credential gAAAABBBBBBBBB'``.
+        """
+        secret = "gAAAA" + "B" * 40
+        content = f"credential {secret} trailing"
+        # Cut inside the token, below the pattern's 20-char body minimum.
+        max_chars = 25
+        result = {
+            "messages": [
+                {"role": "assistant", "tool_calls": [
+                    {"id": "t1", "function": {"name": "terminal", "arguments": "{}"}},
+                ]},
+                {"role": "tool", "tool_call_id": "t1", "content": content},
+            ]
+        }
+
+        tail = _extract_output_tail(result, max_entries=8, max_chars=max_chars)
+
+        preview = tail[0]["preview"]
+        self.assertLessEqual(len(preview), max_chars)
+        # The token was recognised and collapsed before truncation ran.
+        self.assertIn("...", preview)
+
+        def _longest_run(text: str, ch: str) -> int:
+            best = cur = 0
+            for c in text:
+                cur = cur + 1 if c == ch else 0
+                best = max(best, cur)
+            return best
+
+        # Truncate-first leaves a 9-char raw body run here; redact-first
+        # leaves only the short retained edge of the elided token.
+        self.assertLess(_longest_run(preview, "B"), 8)
+
+    def test_output_tail_redaction_error_skips_only_that_entry_by_default(self):
+        """Default is best-effort: one unredactable result must not blank all.
+
+        ``_extract_output_tail`` is shared with the ordinary delegation
+        Output overlay (the non-timeout success path). Dropping the whole
+        tail on any redactor hiccup would turn a transient failure into a
+        total loss of a working display feature.
+        """
+        poison = "UNREDACTABLE"
+        result = {
+            "messages": [
+                {"role": "assistant", "tool_calls": [
+                    {"id": "t1", "function": {"name": "terminal", "arguments": "{}"}},
+                    {"id": "t2", "function": {"name": "terminal", "arguments": "{}"}},
+                ]},
+                {"role": "tool", "tool_call_id": "t1", "content": "keep me"},
+                {"role": "tool", "tool_call_id": "t2", "content": poison},
+            ]
+        }
+
+        from agent.redact import redact_sensitive_text as real
+
+        def _flaky(text, **kwargs):
+            if poison in text:
+                raise RuntimeError("redactor unavailable")
+            return real(text, **kwargs)
+
+        with patch("agent.redact.redact_sensitive_text", side_effect=_flaky):
+            tail = _extract_output_tail(result, max_entries=8, max_chars=600)
+
+        self.assertEqual([e["preview"] for e in tail], ["keep me"])
+        self.assertNotIn(poison, str(tail))
+
+    def test_output_tail_redaction_error_drops_entire_tail_when_fail_closed(self):
+        """fail_closed=True (timeout evidence) still drops everything.
+
+        Evidence surfaced from a FAILED child is a security boundary rather
+        than a display nicety, so it must emit nothing rather than risk an
+        unredacted preview.
+        """
         result = {
             "messages": [
                 {"role": "assistant", "tool_calls": [
                     {"id": "t1", "function": {"name": "terminal", "arguments": "{}"}},
+                    {"id": "t2", "function": {"name": "terminal", "arguments": "{}"}},
                 ]},
-                {"role": "tool", "tool_call_id": "t1", "content": "sensitive output"},
+                {"role": "tool", "tool_call_id": "t1", "content": "harmless"},
+                {"role": "tool", "tool_call_id": "t2", "content": "sensitive output"},
             ]
         }
 
@@ -502,7 +592,9 @@ def test_output_tail_redaction_error_drops_entire_tail(self):
             "agent.redact.redact_sensitive_text",
             side_effect=RuntimeError("redactor unavailable"),
         ):
-            tail = _extract_output_tail(result, max_entries=8, max_chars=600)
+            tail = _extract_output_tail(
+                result, max_entries=8, max_chars=600, fail_closed=True
+            )
 
         self.assertEqual(tail, [])
 
diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py
index e6f34e88df11f..b286068e8e5e1 100644
--- a/tools/delegate_tool.py
+++ b/tools/delegate_tool.py
@@ -328,6 +328,7 @@ def _extract_output_tail(
     *,
     max_entries: int = 12,
     max_chars: int = 8000,
+    fail_closed: bool = False,
 ) -> List[Dict[str, Any]]:
     """Pull the last N tool-call results from a child's conversation.
 
@@ -335,6 +336,17 @@ def _extract_output_tail(
     We reuse the same messages list the trajectory saver walks, taking
     only the tail to keep event payloads small.  Each entry is
     ``{tool, preview, is_error}``.
+
+    Tool output is redacted BEFORE truncation so a credential cannot be
+    split at the ``max_chars`` boundary into an unrecognisable — and
+    therefore unredacted — fragment.
+
+    ``fail_closed`` controls what happens if the redactor itself raises.
+    The default (False) is best-effort: that single tool result is skipped
+    and the rest of the tail still renders, so a redactor hiccup cannot
+    blank the whole ordinary Output overlay.  Callers surfacing evidence
+    from a *failed* child — where the entry is a security boundary rather
+    than a display nicety — pass True to drop the entire tail instead.
     """
     messages = result.get("messages") if isinstance(result, dict) else None
     if not isinstance(messages, list):
@@ -376,11 +388,23 @@ def _extract_output_tail(
 
             safe_content = redact_sensitive_text(content, force=True)
         except Exception:
+            if fail_closed:
+                # Security-boundary caller (timeout evidence): never emit a
+                # possibly-unredacted preview, drop the whole tail.
+                logger.warning(
+                    "Failed to redact delegated tool output; dropping output tail",
+                    exc_info=True,
+                )
+                return []
+            # Display caller (ordinary Output overlay): skip only this entry.
+            # Blanking the entire section because one result could not be
+            # redacted would turn a redactor hiccup into a total loss of a
+            # working feature, which upstream never did.
             logger.warning(
-                "Failed to redact delegated tool output; dropping output tail",
+                "Failed to redact delegated tool output; skipping this entry",
                 exc_info=True,
             )
-            return []
+            continue
         # Preserve line structure so the overlay's wrapped scroll region can
         # show real output rather than a whitespace-collapsed blob. We still
         # cap the payload size to keep events bounded.
@@ -2427,6 +2451,10 @@ def _run_with_thread_capture():
                                 {"messages": list(session_messages)},
                                 max_entries=8,
                                 max_chars=600,
+                                # Evidence from a FAILED child is a security
+                                # boundary, not a display nicety: if redaction
+                                # cannot be guaranteed, emit nothing at all.
+                                fail_closed=True,
                             )
                             if isinstance(item, dict)
                             and isinstance(item.get("preview"), str)

From f02bbbd5f78c7ccca5a46b99077d477d69f30b29 Mon Sep 17 00:00:00 2001
From: Charles Cha <92324143+ypwcharles@users.noreply.github.com>
Date: Wed, 12 Aug 2026 21:01:16 +0800
Subject: [PATCH 3/3] fix(delegation): preserve single async timeout evidence

Carry timeout partial_output_tail through the real single-background dispatch, completion-queue, durable-state, and process-notification path.

Treat the runner result as untrusted at the event boundary: accept evidence only for timeout + partial results, validate every entry, retain at most the final 8 entries, force-redact complete previews before truncating to 600 characters, normalize untrusted tool labels, and fail closed on malformed data or redaction errors. Ordinary and non-timeout results retain the legacy event schema.

Persist the sanitized event evidence for restart delivery, but omit the pre-sanitized partial_output_tail from durable result_json so status reads cannot re-expose raw runner data.

Add transport- and durable-level RED/GREEN regressions covering the valid path, timeout-only scoping, malformed input, redaction and bounds, redactor failure, and durable-state sanitization.
---
 tests/tools/test_async_delegation.py | 242 +++++++++++++++++++++++++++
 tools/async_delegation.py            |  53 +++++-
 2 files changed, 294 insertions(+), 1 deletion(-)

diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py
index 238ed6f15982a..6e265ed42f7d6 100644
--- a/tests/tools/test_async_delegation.py
+++ b/tests/tools/test_async_delegation.py
@@ -175,6 +175,248 @@ def runner():
     assert evt["delegation_id"] == res["delegation_id"]
 
 
+def test_single_timeout_partial_tail_survives_dispatch_to_notification():
+    """The real single-task transport must preserve timeout evidence.
+
+    Rendering a hand-built event is insufficient: dispatch_async_delegation()
+    normalizes the child result through _push_completion_event() before the
+    process notification formatter receives it.
+    """
+    safe_tail = [
+        {"tool": "terminal", "preview": "REDACTED_FINDING", "is_error": False}
+    ]
+
+    def runner():
+        return {
+            "status": "timeout",
+            "summary": None,
+            "error": "child timed out",
+            "api_calls": 2,
+            "duration_seconds": 3.0,
+            "partial": True,
+            "partial_output_tail": safe_tail,
+        }
+
+    res = ad.dispatch_async_delegation(
+        goal="research then time out",
+        context=None,
+        toolsets=["terminal"],
+        role="leaf",
+        model="test-model",
+        session_key="agent:main:cli:dm:local",
+        runner=runner,
+        max_async_children=3,
+    )
+    assert res["status"] == "dispatched"
+
+    evt = _drain_for(res["delegation_id"])
+    assert evt is not None
+    assert evt["status"] == "timeout"
+    assert evt["partial"] is True
+    assert evt["partial_output_tail"] == [
+        {"tool": "tool", "preview": "REDACTED_FINDING", "is_error": False}
+    ]
+
+    text = format_process_notification(evt)
+    assert text is not None
+    assert "Partial output (redacted):" in text
+    assert "REDACTED_FINDING" in text
+
+
+def test_single_timeout_malformed_partial_tail_keeps_legacy_event_schema():
+    """Malformed timeout evidence must not cross into the formatter."""
+
+    def runner():
+        return {
+            "status": "timeout",
+            "summary": None,
+            "error": "child timed out",
+            "partial": True,
+            "partial_output_tail": ["not-a-tail-entry"],
+        }
+
+    res = ad.dispatch_async_delegation(
+        goal="malformed timeout evidence",
+        context=None,
+        toolsets=None,
+        role="leaf",
+        model="test-model",
+        session_key="agent:main:cli:dm:local",
+        runner=runner,
+        max_async_children=3,
+    )
+    evt = _drain_for(res["delegation_id"])
+
+    assert evt is not None
+    assert "partial" not in evt
+    assert "partial_output_tail" not in evt
+    text = format_process_notification(evt)
+    assert text is not None
+    assert "Partial output (redacted):" not in text
+
+
+def test_single_non_timeout_partial_tail_keeps_legacy_event_schema():
+    """Only timeout results may expose bounded partial evidence."""
+    safe_tail = [
+        {"tool": "terminal", "preview": "safe output", "is_error": False}
+    ]
+
+    def runner():
+        return {
+            "status": "completed",
+            "summary": "done",
+            "partial": True,
+            "partial_output_tail": safe_tail,
+        }
+
+    res = ad.dispatch_async_delegation(
+        goal="completed task with stray evidence",
+        context=None,
+        toolsets=None,
+        role="leaf",
+        model="test-model",
+        session_key="agent:main:cli:dm:local",
+        runner=runner,
+        max_async_children=3,
+    )
+    evt = _drain_for(res["delegation_id"])
+
+    assert evt is not None
+    assert evt["status"] == "completed"
+    assert "partial" not in evt
+    assert "partial_output_tail" not in evt
+
+
+def test_single_timeout_transport_force_redacts_then_bounds_tail():
+    """Shape-valid runner output is still untrusted at the event boundary."""
+    secret = "sk-proj-" + "A" * 48
+    raw_preview = f"token={secret}\n" + "x" * 900
+    raw_tail = [
+        {
+            "tool": f"terminal-{secret}",
+            "preview": raw_preview,
+            "is_error": False,
+        }
+        for _ in range(10)
+    ]
+
+    def runner():
+        return {
+            "status": "timeout",
+            "summary": None,
+            "error": "child timed out",
+            "partial": True,
+            "partial_output_tail": raw_tail,
+        }
+
+    res = ad.dispatch_async_delegation(
+        goal="untrusted timeout evidence",
+        context=None,
+        toolsets=None,
+        role="leaf",
+        model="test-model",
+        session_key="agent:main:cli:dm:local",
+        runner=runner,
+        max_async_children=3,
+    )
+    evt = _drain_for(res["delegation_id"])
+
+    assert evt is not None
+    transported = evt["partial_output_tail"]
+    assert len(transported) == 8
+    assert all(len(entry["preview"]) <= 600 for entry in transported)
+    assert all(
+        secret not in entry["preview"] for entry in transported
+    ), repr([entry["preview"][:100] for entry in transported])
+    assert all(
+        secret not in entry["tool"] for entry in transported
+    ), repr([entry["tool"] for entry in transported])
+    assert all(len(entry["tool"]) <= 120 for entry in transported)
+    assert all("token=***" in entry["preview"] for entry in transported)
+
+
+def test_single_timeout_transport_drops_tail_when_redaction_fails(monkeypatch):
+    """The transport boundary must fail closed if forced redaction raises."""
+    from agent import redact
+
+    def fail_redaction(_text, *, force=False):
+        assert force is True
+        raise RuntimeError("redactor unavailable")
+
+    monkeypatch.setattr(redact, "redact_sensitive_text", fail_redaction)
+
+    def runner():
+        return {
+            "status": "timeout",
+            "summary": None,
+            "error": "child timed out",
+            "partial": True,
+            "partial_output_tail": [
+                {"tool": "terminal", "preview": "raw secret", "is_error": False}
+            ],
+        }
+
+    res = ad.dispatch_async_delegation(
+        goal="redaction failure",
+        context=None,
+        toolsets=None,
+        role="leaf",
+        model="test-model",
+        session_key="agent:main:cli:dm:local",
+        runner=runner,
+        max_async_children=3,
+    )
+    evt = _drain_for(res["delegation_id"])
+
+    assert evt is not None
+    assert "partial" not in evt
+    assert "partial_output_tail" not in evt
+
+
+def test_single_timeout_durable_result_drops_raw_partial_tail(tmp_path, monkeypatch):
+    """Durable status data must not retain the untrusted pre-sanitized tail."""
+    monkeypatch.setattr(ad, "_db_path", lambda: tmp_path / "state.db")
+    secret = "sk-proj-" + "B" * 48
+
+    def runner():
+        return {
+            "status": "timeout",
+            "summary": None,
+            "error": "child timed out",
+            "partial": True,
+            "partial_output_tail": [
+                {
+                    "tool": f"terminal-{secret}",
+                    "preview": f"token={secret}",
+                    "is_error": False,
+                }
+            ],
+        }
+
+    res = ad.dispatch_async_delegation(
+        goal="durable timeout evidence",
+        context=None,
+        toolsets=None,
+        role="leaf",
+        model="test-model",
+        session_key="agent:main:cli:dm:local",
+        runner=runner,
+        max_async_children=3,
+    )
+    evt = _drain_for(res["delegation_id"])
+    durable = ad.get_durable_delegation(res["delegation_id"])
+
+    assert evt is not None
+    assert evt["partial_output_tail"] == [
+        {"tool": "tool", "preview": "token=***", "is_error": False}
+    ]
+    assert secret not in str(evt)
+    assert durable is not None
+    assert durable["result"]["status"] == "timeout"
+    assert "partial_output_tail" not in durable["result"]
+    assert secret not in str(durable)
+
+
 def test_rich_reinjection_block_is_self_contained():
     def runner():
         return {"status": "completed", "summary": "The answer is 42.",
diff --git a/tools/async_delegation.py b/tools/async_delegation.py
index 4965363a07eef..4786296708ddb 100644
--- a/tools/async_delegation.py
+++ b/tools/async_delegation.py
@@ -314,13 +314,20 @@ def _prune_durable_records() -> None:
 
 def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None:
     now = time.time()
+    # ``result`` comes from the injected runner. The public completion event
+    # already carries the independently sanitized, bounded timeout evidence;
+    # retaining the pre-sanitized tail in result_json would re-expose it via
+    # get_durable_delegation(). Preserve all legacy result metadata, but never
+    # persist this untrusted display payload.
+    durable_result = dict(result)
+    durable_result.pop("partial_output_tail", None)
     with _DB_LOCK, _transaction() as conn:
         conn.execute(
             """UPDATE async_delegations SET state=?, completed_at=?, updated_at=?,
                event_json=?, result_json=?, delivery_state='pending'
                WHERE delegation_id=?""",
             (event.get("status", "completed"), event.get("completed_at", now), now,
-             json.dumps(event), json.dumps(result), event["delegation_id"]),
+             json.dumps(event), json.dumps(durable_result), event["delegation_id"]),
         )
 
 
@@ -985,6 +992,50 @@ def _push_completion_event(
         "completed_at": completed_at,
         "exit_reason": result.get("exit_reason"),
     }
+    # A single background timeout is normalized through this event boundary
+    # before process_registry formats it. Treat the runner result as untrusted:
+    # validate the shape, force-redact complete strings before truncating, and
+    # fail closed on any malformed entry or redactor failure.
+    partial_tail = result.get("partial_output_tail")
+    if (
+        status == "timeout"
+        and result.get("partial") is True
+        and isinstance(partial_tail, list)
+        and partial_tail
+    ):
+        try:
+            from agent.redact import redact_sensitive_text
+
+            safe_partial_tail = []
+            for entry in partial_tail[-8:]:
+                if (
+                    not isinstance(entry, dict)
+                    or not isinstance(entry.get("tool"), str)
+                    or not isinstance(entry.get("preview"), str)
+                    or not isinstance(entry.get("is_error"), bool)
+                ):
+                    raise ValueError("invalid partial output tail entry")
+                # ``tool`` is display-only here and comes from the untrusted
+                # runner result. Do not echo it across the security boundary:
+                # arbitrary identifier-looking text can still be a secret.
+                safe_partial_tail.append(
+                    {
+                        "tool": "tool",
+                        "preview": redact_sensitive_text(
+                            entry["preview"], force=True
+                        )[:600],
+                        "is_error": entry["is_error"],
+                    }
+                )
+        except Exception:
+            logger.warning(
+                "Failed to sanitize async delegation timeout evidence; "
+                "dropping output tail",
+                exc_info=True,
+            )
+        else:
+            evt["partial"] = True
+            evt["partial_output_tail"] = safe_partial_tail
     # Routing origin captured at dispatch (see _capture_routing_origin):
     # additive, lets the gateway reconstruct a full SessionSource (incl.
     # scope_id for relay tenant egress) when its own caches are cold.