fix(delegation): redact credentials in live subagent transcripts

tests/tools/test_delegation_live_log.py

@@ -481,3 +481,131 @@ def test_delegate_task_proceeds_when_transcripts_unavailable(monkeypatch):
 if __name__ == "__main__":
     import sys
     sys.exit(pytest.main([__file__, "-v"]))
+
+
+# ---------------------------------------------------------------------------
+# Credential redaction
+# ---------------------------------------------------------------------------
+#
+# These transcripts land under ``cache/delegation``, which delegate_tool mounts
+# READ-ONLY into remote terminal backends — so a line written here is readable
+# from inside the sandbox. The rendered events are exactly the secret-bearing
+# surfaces (tool args, tool results, streamed assistant text), and every other
+# sink for that data already routes through the canonical redactor.
+
+_BEARER = "sk-ant-api03-" + "R" * 24
+_ENV_KEY = "sk-proj-" + "L" * 24
+_AWS = "wJalrXUtnFEMIK7MDENG" + "bPxRfiCY"
+
+
+def test_tool_args_are_redacted_before_hitting_disk():
+    w = LiveTranscriptWriter("deleg_redact_args", 0, "g")
+    w.observe(
+        "tool.started",
+        "terminal",
+        f'curl -H "Authorization: Bearer {_BEARER}" https://api.internal',
+        None,
+    )
+    body = w.path.read_text(encoding="utf-8")
+    assert _BEARER not in body
+    assert "terminal" in body, "redaction must not gut the operational detail"
+
+
+def test_tool_results_are_redacted_before_hitting_disk():
+    w = LiveTranscriptWriter("deleg_redact_result", 0, "g")
+    w.observe(
+        "tool.completed",
+        "terminal",
+        None,
+        None,
+        result=f"OPENAI_API_KEY={_ENV_KEY}\nAWS_SECRET_ACCESS_KEY={_AWS}",
+        duration=0.4,
+    )
+    body = w.path.read_text(encoding="utf-8")
+    assert _ENV_KEY not in body
+    assert _AWS not in body
+    assert "OPENAI_API_KEY" in body, "key NAMES stay — only the values are masked"
+
+
+def test_streamed_assistant_text_is_redacted():
+    w = LiveTranscriptWriter("deleg_redact_stream", 0, "g")
+    w.observe("subagent.text", None, f"the key is {_ENV_KEY}")
+    w.flush_stream()
+    assert _ENV_KEY not in w.path.read_text(encoding="utf-8")
+
+
+def test_goal_header_is_redacted():
+    """The header bypasses event(); a pasted key in the goal must not survive."""
+    w = LiveTranscriptWriter("deleg_redact_goal", 0, f"deploy using {_BEARER}")
+    body = w.path.read_text(encoding="utf-8")
+    assert _BEARER not in body
+    assert "deploy using" in body
+
+
+def test_manifest_goal_is_redacted():
+    """manifest.json shares the mounted dir with the .log files.
+
+    Redacting the log header while ``_write_manifest`` serialises the same goal
+    verbatim would leave the credential exposed one file over — both sinks in
+    ``cache/delegation/live/<id>/`` are readable from inside a sandbox.
+    """
+    delegation_id, _writers, _paths = create_live_transcripts(
+        [{"goal": f"deploy using {_BEARER}"}]
+    )
+
+    manifest = json.loads(
+        (live_transcript_root() / delegation_id / "manifest.json").read_text(
+            encoding="utf-8"
+        )
+    )
+    goal = manifest["tasks"][0]["goal"]
+
+    assert _BEARER not in goal
+    assert "deploy using" in goal, "redaction must not blank the goal entirely"
+
+
+def test_no_file_in_the_dispatch_directory_carries_the_raw_key():
+    """Whole-directory sweep: every artefact dispatch writes is covered."""
+    delegation_id, _writers, _paths = create_live_transcripts(
+        [{"goal": f"deploy using {_BEARER}"}, {"goal": "second task"}]
+    )
+
+    directory = live_transcript_root() / delegation_id
+    written = sorted(p.name for p in directory.iterdir())
+
+    assert "manifest.json" in written
+    assert any(name.endswith(".log") for name in written)
+    for path in directory.iterdir():
+        assert _BEARER not in path.read_text(encoding="utf-8"), (
+            f"{path.name} leaked the credential"
+        )
+
+
+def test_thinking_text_is_redacted():
+    w = LiveTranscriptWriter("deleg_redact_think", 0, "g")
+    w.observe("_thinking", f"I should use {_ENV_KEY} here")
+    assert _ENV_KEY not in w.path.read_text(encoding="utf-8")
+
+
+def test_redaction_covers_every_helper_via_the_event_chokepoint():
+    """Any helper that reaches disk goes through event(), so all are covered."""
+    w = LiveTranscriptWriter("deleg_redact_all", 0, "g")
+    w.assistant_text(f"a {_ENV_KEY}")
+    w.thinking(f"b {_ENV_KEY}")
+    w.tool_start("terminal", f"c {_ENV_KEY}")
+    w.tool_result("terminal", result=f"d {_ENV_KEY}")
+    w.marker(f"e {_ENV_KEY}")
+    w.finalize({"status": "error", "error": f"f {_ENV_KEY}"})
+    body = w.path.read_text(encoding="utf-8")
+    assert _ENV_KEY not in body, "a write path escaped the redactor"
+
+
+def test_benign_transcript_content_is_untouched():
+    """Redaction must not mangle ordinary transcript text."""
+    w = LiveTranscriptWriter("deleg_redact_benign", 0, "refactor the parser")
+    w.observe("tool.started", "read_file", "src/parser.py", None)
+    w.observe("tool.completed", "read_file", None, None, result="def parse(x): ...", duration=1.5)
+    body = w.path.read_text(encoding="utf-8")
+    assert "src/parser.py" in body
+    assert "def parse(x)" in body
+    assert "refactor the parser" in body

tools/delegation_live_log.py

@@ -80,6 +80,35 @@ def _one_line(text: Any, limit: int) -> str:
     return s
 
 
+def _redact(text: str) -> str:
+    """Mask credentials before anything reaches the transcript file.
+
+    These logs live under ``cache/delegation``, which ``delegate_tool`` mounts
+    READ-ONLY into remote terminal backends — so every line written here is
+    readable from inside the sandbox. The events rendered here carry exactly
+    the data that tends to hold secrets: tool args (a bearer header on a
+    curl), tool results (a ``.env`` dump, a provider error echoing the key
+    back) and streamed assistant text. Every other sink for that data already
+    routes through this same redactor — search results via
+    ``redact_sensitive_text``, terminal output via ``redact_terminal_output``
+    — so a transcript that skipped it is the one place the operator's keys
+    land in plaintext.
+
+    ``force=True``: this is a safety boundary, so it must redact even when the
+    global toggle is off. Withholds the line rather than emitting raw text if
+    the redactor is somehow unavailable — losing a debug line costs less than
+    writing a live credential into a sandbox-readable file.
+    """
+    if not text:
+        return text
+    try:
+        from agent.redact import redact_sensitive_text
+
+        return redact_sensitive_text(text, force=True) or ""
+    except Exception:  # pragma: no cover - core module; never leak on failure
+        return "[line withheld: redaction unavailable]"
+
+
 class LiveTranscriptWriter:
     """Append-only human-readable event log for ONE subagent task.
 
@@ -103,7 +132,9 @@ def __init__(self, delegation_id: str, task_index: int, goal: str,
             header = [
                 "=== Hermes subagent live transcript ===",
                 f"delegation: {delegation_id}   task: {task_index}",
-                f"goal: {_one_line(goal, _KICKOFF_MAX)}",
+                # Header bypasses event(), so redact here too — a goal string
+                # can carry a key the caller pasted into the task.
+                f"goal: {_redact(_one_line(goal, _KICKOFF_MAX))}",
                 f"started: {time.strftime('%Y-%m-%d %H:%M:%S')}",
                 "(append-only; streams while the subagent runs — tail -f me)",
                 "=" * 40,
@@ -122,7 +153,10 @@ def event(self, role: str, text: str) -> None:
         """Append one ``HH:MM:SS role ⟩ text`` line. Flushed per event."""
         if not self._ok or self.path is None:
             return
-        line = f"{time.strftime('%H:%M:%S')} {role:<9}| {text}\n"
+        # Single choke point: every typed helper funnels through here, so
+        # redacting once covers args, results, thinking and streamed text —
+        # and a helper added later can't bypass it.
+        line = f"{time.strftime('%H:%M:%S')} {role:<9}| {_redact(text)}\n"
         try:
             with self._lock:
                 # Append mode per write: no held handle, survives child crash,
@@ -326,7 +360,12 @@ def _write_manifest(delegation_id: str, task_list: List[Dict[str, Any]],
             "tasks": [
                 {
                     "index": i,
-                    "goal": str(t.get("goal", ""))[:500],
+                    # manifest.json sits in the same mounted
+                    # cache/delegation/live/<id>/ directory as the .log files,
+                    # so it needs the same treatment — redacting the header
+                    # while serialising the goal verbatim here would leave the
+                    # credential exposed one file over.
+                    "goal": _redact(str(t.get("goal", ""))[:500]),
                     "log": paths[i] if i < len(paths) else None,
                     "status": "running",
                 }