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/tests/tools/test_delegate.py b/tests/tools/test_delegate.py index bb995dda599ce..a8a6c3eda30a3 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,137 @@ 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_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": "harmless"}, + {"role": "tool", "tool_call_id": "t2", "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, fail_closed=True + ) + + 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 +1880,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/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. diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index e4f4766fca875..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): @@ -368,10 +380,35 @@ 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: + 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; skipping this entry", + exc_info=True, + ) + 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. - 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 +2436,56 @@ 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, + # 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) + 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 +2511,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 +2535,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)