feat(task-agent): recall sub-trajectory + per-agent read isolation

Final chunk of the task_agent modernization: rebuild a finished task agent's card from /history (reload / reopen while the workstream is in memory) and isolate each sub-agent's file-read tracking. Recall: _project_agent_steps projects a sub-agent's trajectory into step items (FIFO-per-call_id pairing via _iter_agent_tool_results, shared with _cancel_ledger; output/arguments/count capped); _stash_agent_trajectory keeps them on the UI in an LRU-bounded store; make_history_handler attaches them as agent_steps to each task_agent tool_call, and replayHistory/_replayAgentCard rebuild the collapsed card. In-memory only (durable persistence deferred); a cold/evicted entry renders the flat parent row ("not retained"), never a fabricated 0-step card. Read isolation: _read_files (the blind-overwrite guard's memory) is now per-sub-agent via the _active_read_files contextvar -- _exec_task copies the parent's set on spawn and merges the agent's reads back on completion, so a sibling in the 4-wide pool can't suppress another agent's guard. Also: _exec_task now self-reports the task_agent tool_result on every path (the parent loop only reports error/denied results centrally) -- without it the live card never completed and a failed task recorded is_error=False in the canonical trajectory. is_error flows from _tool_error_flags to the recalled step; on_info suppression is per-thread so a parallel sibling tool's progress isn't dropped.
2026-08-12 23:12:23 -06:00 · 2026-06-28 02:06:20 -07:00
parent ca7958329a
commit 77cb76c006
9 changed files with 868 additions and 97 deletions
@@ -64,9 +64,14 @@ Task-agent harness (/taskagent/livepass.html): the task_agent card — a task
  tool_pending/tool_result/tool_output_chunk/approve_request -> task_agent
  tool_result) so the SSE->card routing (_routeAgentItems / _ensureAgentCard,
  and appendToolOutput finding the nested row by call_id) is exercised, not
-  just the leaf builders.  + &theme=light, &collapsed=1.  document.title stamps
-  TASKAGENT-READY-<steps> on success, TASKAGENT-FAILED-... / TASKAGENT-ERROR
-  when routing breaks, so a broken card can't screenshot green.
+  just the leaf builders.  Query flags: &theme=light; &collapsed=1 (all-auto,
+  no approval -> the natural collapse-by-default state); &parallel=1 (card in a
+  2-tool batch, for the rail-bleed rules); &recall=1 (the RECALL path —
+  replayHistory rebuilding the card from a /history `agent_steps` overlay, i.e.
+  a reload while the ws is in memory); &expand=1 (open every card so a shot
+  shows the nested steps).  document.title stamps TASKAGENT-READY-<steps> on
+  success, TASKAGENT-FAILED-... / TASKAGENT-ERROR when routing breaks, so a
+  broken card can't screenshot green.

 Rebuild after ANY markup change: the dialog blocks are embedded at build
 time. Assets are symlinked, so CSS/JS edits are live on refresh.
@@ -846,6 +851,27 @@ TASKAGENT_TEMPLATE = """<!doctype html>
        pane.setBusy = () => {};
        const ev = (e) => pane.handleEvent(e);

+        // ?recall=1: exercise the RECALL path — replayHistory rebuilding the
+        // card from the /history `agent_steps` overlay (a reload / reopen while
+        // the ws is still in memory), as opposed to the live SSE path below.
+        const recall = q.get("recall") === "1";
+        if (recall) {
+          pane.replayHistory([
+            { role: "user", content: "Find all call sites of resolve_alias and summarize them" },
+            { role: "assistant", tool_calls: [{
+              name: "task_agent", id: "task1",
+              arguments: JSON.stringify({ prompt: "Find call sites of resolve_alias" }),
+              agent_steps: [
+                { id: "task1::c1", name: "search", arguments: JSON.stringify({ query: "resolve_alias" }), output: "12 matches across 4 files", is_error: false },
+                { id: "task1::c2", name: "read_file", arguments: JSON.stringify({ path: "core/registry.py" }), output: "4.1 KB read", is_error: false },
+                { id: "task1::c3", name: "bash", arguments: JSON.stringify({ command: "pytest -k registry" }), output: "12 passed in 1.2s", is_error: false },
+                { id: "task1::c4", name: "notify", arguments: JSON.stringify({ channel: "#eng", message: "post summary" }), output: "posted to #eng", is_error: false },
+              ],
+            }] },
+            { role: "tool", tool_call_id: "task1", content: "resolve_alias has 4 call sites (registry.py:120, session.py:12200, model_registry.py:88, eval.py:54); all pass a validated alias before use." },
+          ]);
+        } else {
+
        // 1. Parent paints the task_agent call (a top-level tool row).
        const taskItem = {
          call_id: "task1", func_name: "task_agent",
@@ -886,6 +912,17 @@ TASKAGENT_TEMPLATE = """<!doctype html>

        // 3. The task agent's own synthesis, rendered below the card.
        ev({ type: "tool_result", call_id: "task1", name: "task_agent", output: "resolve_alias has 4 call sites (registry.py:120, session.py:12200, model_registry.py:88, eval.py:54); all pass a validated alias before use." });
+        }
+
+        // ?expand=1: open every card so a screenshot shows the nested steps
+        // (cards collapse by default; recall has no approval to auto-expand).
+        if (q.get("expand") === "1") {
+          document.querySelectorAll(".conv-agent").forEach(function (c) {
+            c.dataset.collapsed = "false";
+            const t = c.querySelector(".conv-agent-toggle");
+            if (t) t.setAttribute("aria-expanded", "true");
+          });
+        }

        // Loud failure — broken routing must not screenshot green.
        setTimeout(function () {
@@ -530,15 +530,17 @@ def test_phase8_appendtooloutput_dispatches_mcp_error_before_renderer() -> None:
    end = _pane_method_offset(body, "sendMessage")
    fn = body[start:end]
    parse_idx = fn.find("tryParseMcpError(")
-    render_idx = fn.find("renderToolOutput(")
+    # The plain-output render is the shared renderCollapsibleOutput helper; the
+    # ordering invariant is unchanged — MCP dispatch must precede it.
+    render_idx = fn.find("renderCollapsibleOutput(")
    assert parse_idx >= 0, (
        "appendToolOutput must call tryParseMcpError on the error path "
-        "before renderToolOutput, otherwise the consent card never "
+        "before the plain renderer, otherwise the consent card never "
        "replaces the plain JSON output."
    )
-    assert render_idx >= 0, "renderToolOutput call must remain present"
+    assert render_idx >= 0, "renderCollapsibleOutput call must remain present"
    assert parse_idx < render_idx, (
-        "tryParseMcpError must run BEFORE renderToolOutput so the "
+        "tryParseMcpError must run BEFORE the plain renderer so the "
        "interactive card path takes precedence over plain rendering."
    )

@@ -1658,6 +1658,310 @@ class TestAgentChildRegistration:
        session.ui.note_agent_child.assert_called_once_with("task-1::call_1", "task-1")


+class TestProjectAgentSteps:
+    """``_project_agent_steps`` projects a finished sub-agent's trajectory into
+    recall step items for the task card — one per tool call, matched to its
+    result by call_id, landmine-safe on a multimodal result."""
+
+    def test_calls_matched_to_results_in_order(self):
+        from turnstone.core.trajectory import ToolCall, Turn
+
+        turns = [
+            Turn.system("sys"),
+            Turn.user("go"),
+            Turn.assistant(
+                tool_calls=(ToolCall(id="c1", name="search", arguments='{"query":"x"}'),)
+            ),
+            Turn.tool("c1", "12 matches"),
+            Turn.assistant(
+                tool_calls=(ToolCall(id="c2", name="bash", arguments='{"command":"ls"}'),)
+            ),
+            Turn.tool("c2", "boom", is_error=True),
+        ]
+        steps = ChatSession._project_agent_steps(turns)
+        assert [s["id"] for s in steps] == ["c1", "c2"]
+        assert steps[0] == {
+            "id": "c1",
+            "name": "search",
+            "arguments": '{"query":"x"}',
+            "output": "12 matches",
+            "is_error": False,
+        }
+        assert steps[1]["is_error"] is True
+        assert steps[1]["output"] == "boom"
+
+    def test_multimodal_result_placeholdered_not_crashed(self):
+        # A vision tool result is a list[dict] mis-stored as TextBlock.text; the
+        # projection must NOT call Turn.text (would TypeError) — it reads the
+        # payload directly and placeholders a non-str so /history stays text-only.
+        from turnstone.core.trajectory import ToolCall, Turn
+
+        turns = [
+            Turn.assistant(
+                tool_calls=(ToolCall(id="c1", name="read_file", arguments='{"path":"a.png"}'),)
+            ),
+            Turn.tool("c1", [{"type": "image_url"}]),
+        ]
+        steps = ChatSession._project_agent_steps(turns)
+        assert steps[0]["output"] == "[non-text result]"
+
+    def test_output_capped(self):
+        from turnstone.core.session import _AGENT_STEP_OUTPUT_CAP
+        from turnstone.core.trajectory import ToolCall, Turn
+
+        big = "a" * (_AGENT_STEP_OUTPUT_CAP + 500)
+        turns = [
+            Turn.assistant(tool_calls=(ToolCall(id="c1", name="bash", arguments="{}"),)),
+            Turn.tool("c1", big),
+        ]
+        steps = ChatSession._project_agent_steps(turns)
+        assert len(steps[0]["output"]) < len(big)
+        assert "truncated from 2500 chars" in steps[0]["output"]
+
+    def test_unanswered_call_has_empty_output(self):
+        # A tool call with no matching result (cancelled mid-flight) recalls
+        # honestly as empty, not dropped.
+        from turnstone.core.trajectory import ToolCall, Turn
+
+        turns = [Turn.assistant(tool_calls=(ToolCall(id="c1", name="bash", arguments="{}"),))]
+        steps = ChatSession._project_agent_steps(turns)
+        assert steps == [
+            {"id": "c1", "name": "bash", "arguments": "{}", "output": "", "is_error": False}
+        ]
+
+    def test_colliding_ids_paired_fifo_not_last_wins(self):
+        # A local provider reuses id "call_0" across turns; FIFO pairing gives
+        # each call its OWN result, not last-wins (which would show out-B twice).
+        from turnstone.core.trajectory import ToolCall, Turn
+
+        turns = [
+            Turn.assistant(
+                tool_calls=(ToolCall(id="call_0", name="bash", arguments='{"command":"a"}'),)
+            ),
+            Turn.tool("call_0", "out-A"),
+            Turn.assistant(
+                tool_calls=(ToolCall(id="call_0", name="bash", arguments='{"command":"b"}'),)
+            ),
+            Turn.tool("call_0", "out-B"),
+        ]
+        steps = ChatSession._project_agent_steps(turns)
+        assert [s["output"] for s in steps] == ["out-A", "out-B"]
+
+    def test_step_count_capped_with_honest_marker(self):
+        from turnstone.core.session import _AGENT_STEP_COUNT_CAP
+        from turnstone.core.trajectory import ToolCall, Turn
+
+        turns = []
+        for i in range(_AGENT_STEP_COUNT_CAP + 5):
+            turns.append(
+                Turn.assistant(tool_calls=(ToolCall(id=f"c{i}", name="bash", arguments="{}"),))
+            )
+            turns.append(Turn.tool(f"c{i}", f"out{i}"))
+        steps = ChatSession._project_agent_steps(turns)
+        # Capped + one honest LEADING marker, keeping the most RECENT steps (the
+        # tail) — not the earliest — and naming how many earlier ones fell out.
+        assert len(steps) == _AGENT_STEP_COUNT_CAP + 1
+        assert steps[0]["name"] == "…"
+        assert "5 earlier steps not retained" in steps[0]["output"]
+        # c0..c4 dropped; c5 is the first retained, the newest call is last.
+        assert steps[1]["id"] == "c5"
+        assert steps[-1]["id"] == f"c{_AGENT_STEP_COUNT_CAP + 4}"
+
+
+class TestAgentTrajectoryStashWiring:
+    """``_stash_agent_trajectory`` projects + forwards to the UI, getattr-guarded."""
+
+    def test_projects_and_forwards(self):
+        from turnstone.core.trajectory import ToolCall, Turn
+
+        session = _make_session()
+        session.ui = MagicMock()
+        turns = [
+            Turn.assistant(tool_calls=(ToolCall(id="c1", name="bash", arguments="{}"),)),
+            Turn.tool("c1", "ok"),
+        ]
+        session._stash_agent_trajectory("task1", turns)
+        session.ui.stash_agent_trajectory.assert_called_once()
+        cid, steps = session.ui.stash_agent_trajectory.call_args[0]
+        assert cid == "task1"
+        assert steps == [
+            {"id": "c1", "name": "bash", "arguments": "{}", "output": "ok", "is_error": False}
+        ]
+
+    def test_noop_without_call_id(self):
+        session = _make_session()
+        session.ui = MagicMock()
+        session._stash_agent_trajectory(None, [])
+        session.ui.stash_agent_trajectory.assert_not_called()
+
+    def test_noop_on_ui_without_support(self):
+        # NullUI has no stash_agent_trajectory → getattr None → no-op, no raise.
+        _make_session()._stash_agent_trajectory("task1", [])
+
+
+class TestReadFilesIsolation:
+    """A task agent's file-read tracking is isolated from the main session and
+    its pool siblings via ``_active_read_files`` so the blind-overwrite guard
+    can't be cross-contaminated (a sibling's read suppressing another's guard)."""
+
+    def test_defaults_to_main_set(self):
+        session = _make_session()
+        assert session._current_read_files is session._read_files
+
+    def test_active_contextvar_overrides_then_restores(self):
+        from turnstone.core.session import _active_read_files
+
+        session = _make_session()
+        sub: set[str] = set()
+        token = _active_read_files.set(sub)
+        try:
+            assert session._current_read_files is sub
+        finally:
+            _active_read_files.reset(token)
+        assert session._current_read_files is session._read_files
+
+    def test_empty_active_set_is_used_not_main(self):
+        # The resolver guards on `is not None`, not truthiness — an EMPTY
+        # per-agent set must be used, NOT fall through to the main set, or a
+        # fresh agent would inherit the main session's reads and mis-suppress
+        # its own blind-overwrite guard.
+        from turnstone.core.session import _active_read_files
+
+        session = _make_session()
+        session._read_files.add("/main/file")
+        token = _active_read_files.set(set())
+        try:
+            assert session._current_read_files == set()
+        finally:
+            _active_read_files.reset(token)
+
+    def test_exec_task_copies_parent_reads_and_merges_back(self):
+        # Drive the REAL _exec_task wiring (not a hand-rolled contextvar dance):
+        # it copies the parent's reads into an INDEPENDENT per-agent set (so the
+        # agent can edit a file the parent read for it, without leaking mid-run
+        # to a sibling) and merges the agent's own reads back on completion.
+        session = _make_session()
+        session._agent_system_messages = []
+        session._task_tools = []
+        session._read_files.add("/parent/read")
+        seen = {}
+
+        def fake_run_agent(agent_turns, **_kwargs):
+            seen["sees_parent"] = "/parent/read" in session._current_read_files
+            session._current_read_files.add("/child/read")
+            seen["child_isolated"] = "/child/read" not in session._read_files
+            return "done"
+
+        with patch.object(session, "_run_agent", side_effect=fake_run_agent):
+            cid, out = session._exec_task({"call_id": "t1", "prompt": "go"})
+
+        assert (cid, out) == ("t1", "done")
+        assert seen["sees_parent"] is True  # copy-on-spawn: inherits parent's reads
+        assert seen["child_isolated"] is True  # independent set mid-run (no leak)
+        assert "/child/read" in session._read_files  # merged back on completion
+        assert session._current_read_files is session._read_files  # contextvar reset
+
+
+class TestSubAgentErrorRecall:
+    """_run_agent stamps is_error on a sub-tool's Turn from the authoritative
+    _tool_error_flags, so a failed sub-tool recalls styled as an error rather
+    than a green 'done' step (the most serious review finding)."""
+
+    def test_errored_sub_tool_turn_marked_is_error(self):
+        from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
+        from turnstone.core.trajectory import Role
+
+        session = _make_session()
+        session._provider = OpenAIChatCompletionsProvider()
+        calls = [0]
+
+        def fake_create(**_kwargs):
+            calls[0] += 1
+            resp = MagicMock()
+            choice = MagicMock()
+            if calls[0] == 1:
+                choice.finish_reason = "tool_calls"
+                tc = MagicMock()
+                tc.id = "call_1"
+                tc.function.name = "bash"
+                tc.function.arguments = '{"command":"false"}'
+                choice.message.tool_calls = [tc]
+                choice.message.content = None
+            else:
+                choice.finish_reason = "stop"
+                choice.message.tool_calls = None
+                choice.message.content = "done"
+            resp.choices = [choice]
+            resp.usage = MagicMock(prompt_tokens=10, completion_tokens=5)
+            return resp
+
+        session.client.chat.completions.create = fake_create
+
+        def fake_prepare(tc_dict, **_kwargs):
+            cid = tc_dict["id"]
+
+            def _exec(p):
+                # Simulate an errored tool: the real exec records is_error via
+                # _report_tool_result, which sets _tool_error_flags.
+                session._tool_error_flags[p["call_id"]] = True
+                return cid, "boom"
+
+            return {
+                "call_id": cid,
+                "func_name": "bash",
+                "needs_approval": False,
+                "execute": _exec,
+            }
+
+        turns = [Turn.user("run it")]
+        with patch.object(session, "_prepare_tool", side_effect=fake_prepare):
+            session._run_agent(
+                turns,
+                tools=[{"type": "function", "function": {"name": "bash"}}],
+                label="task",
+                auto_tools={"bash"},
+                parent_call_id="t1",
+            )
+
+        tool_turns = [t for t in turns if t.role is Role.TOOL]
+        assert tool_turns, "expected a tool result turn"
+        assert tool_turns[-1].is_error is True
+        # And it carries through the projection to the recalled step.
+        assert ChatSession._project_agent_steps(turns)[-1]["is_error"] is True
+
+
+class TestExecTaskReporting:
+    """_exec_task self-reports the task_agent's OWN result — the live card's
+    only completion signal (the parent loop reports error/denied results
+    centrally but relies on each tool self-reporting its success result)."""
+
+    def _bare_session(self):
+        session = _make_session()
+        session._agent_system_messages = []
+        session._task_tools = []
+        return session
+
+    def test_success_reports_result(self):
+        session = self._bare_session()
+        with (
+            patch.object(session, "_run_agent", return_value="the synthesis"),
+            patch.object(session, "_report_tool_result") as rpt,
+        ):
+            cid, out = session._exec_task({"call_id": "t1", "prompt": "go"})
+        assert (cid, out) == ("t1", "the synthesis")
+        rpt.assert_called_once_with("t1", "task_agent", "the synthesis")
+
+    def test_error_reports_is_error(self):
+        session = self._bare_session()
+        with (
+            patch.object(session, "_run_agent", side_effect=RuntimeError("boom")),
+            patch.object(session, "_report_tool_result") as rpt,
+        ):
+            cid, out = session._exec_task({"call_id": "t1", "prompt": "go"})
+        assert out == "Task error: boom"
+        rpt.assert_called_once_with("t1", "task_agent", "Task error: boom", is_error=True)
+
+
 class TestEvaluateOutputLLMStage:
    """End-to-end coverage of _evaluate_output with the LLM judge stage."""

@@ -18,6 +18,8 @@ import threading
 from typing import Any
 from unittest.mock import MagicMock, patch

+import pytest
+
 from turnstone.core.session_ui_base import SessionUIBase


@@ -2166,8 +2168,19 @@ class TestAgentScopeInfoSuppression:
    """While a task agent runs, its ``on_info`` progress chatter ("[task done] N
    chars", a tool's "fetched N chars") carries no call_id, so it can't nest
    under the task card.  The web pane drops it for the duration rather than let
-    it escape to the top level; the depth counter keeps it correct under the
-    parent's parallel task pool."""
+    it escape to the top level; the per-thread contextvar keeps it correct under
+    the parent's parallel task pool (siblings in other threads aren't suppressed)."""
+
+    @pytest.fixture(autouse=True)
+    def _reset_scope(self):
+        # The scope depth is a module-level contextvar that persists across tests
+        # in the same thread; reset it around each so an unbalanced test (or a
+        # leak from elsewhere) can't bleed suppression into another test.
+        from turnstone.core.session_ui_base import _agent_scope_var
+
+        token = _agent_scope_var.set(0)
+        yield
+        _agent_scope_var.reset(token)

    def test_on_info_suppressed_within_scope(self) -> None:
        ui = _make_ui()
@@ -2211,3 +2224,42 @@ class TestAgentScopeInfoSuppression:
        ui.begin_agent_scope()
        ui.on_info("suppressed")
        assert lq.empty()
+
+
+class TestAgentTrajectoryStash:
+    """The recall store retains a finished task agent's projected sub-trajectory
+    keyed by call_id, LRU-bounded.  A miss is the honest "not retained" signal —
+    /history then renders the flat parent record, never a fabricated 0-step card."""
+
+    def test_stash_and_get_roundtrip(self) -> None:
+        ui = _make_ui()
+        steps = [
+            {"id": "t1::c1", "name": "search", "arguments": "{}", "output": "ok", "is_error": False}
+        ]
+        ui.stash_agent_trajectory("t1", steps)
+        assert ui.get_agent_trajectory("t1") == steps
+
+    def test_missing_returns_none(self) -> None:
+        assert _make_ui().get_agent_trajectory("nope") is None
+
+    def test_empty_call_id_ignored(self) -> None:
+        ui = _make_ui()
+        ui.stash_agent_trajectory("", [{"id": "x"}])
+        assert ui.get_agent_trajectory("") is None
+
+    def test_restash_updates_value(self) -> None:
+        ui = _make_ui()
+        ui.stash_agent_trajectory("k", [{"id": "v1"}])
+        ui.stash_agent_trajectory("k", [{"id": "v2"}])
+        assert ui.get_agent_trajectory("k") == [{"id": "v2"}]
+
+    def test_lru_evicts_oldest(self) -> None:
+        from turnstone.core.session_ui_base import _AGENT_TRAJECTORY_CAP
+
+        ui = _make_ui()
+        for i in range(_AGENT_TRAJECTORY_CAP + 3):
+            ui.stash_agent_trajectory(f"t{i}", [{"id": f"t{i}"}])
+        # The three oldest fell out → honest None; the newest is retained.
+        assert ui.get_agent_trajectory("t0") is None
+        assert ui.get_agent_trajectory("t2") is None
+        assert ui.get_agent_trajectory(f"t{_AGENT_TRAJECTORY_CAP + 2}") is not None
@@ -1117,6 +1117,77 @@ class TestExportInteractive:
        assert "should not leak" not in r.text


+class TestHistoryAgentStepsOverlay:
+    """The history handler attaches a live task agent's stashed sub-trajectory to
+    its ``task_agent`` tool_call (``agent_steps``) so the client rebuilds the
+    card.  A cold ws / evicted entry has none → no overlay (honest flat row)."""
+
+    def _save_task_agent_turn(self, storage, ws_id):
+        storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        storage.save_message(ws_id, "user", "kick off")
+        tc_json = json.dumps(
+            [
+                {
+                    "id": "task1",
+                    "type": "function",
+                    "function": {
+                        "name": "task_agent",
+                        "arguments": '{"prompt":"find call sites"}',
+                    },
+                }
+            ]
+        )
+        storage.save_message(ws_id, "assistant", "Working", tool_calls=tc_json)
+        storage.save_message(ws_id, "tool", "4 call sites found", tool_call_id="task1")
+
+    def test_attaches_agent_steps_from_live_stash(self, _inject_storage):
+        ws_id = "ws-recall-warm"
+        self._save_task_agent_turn(_inject_storage, ws_id)
+        steps = [
+            {
+                "id": "task1::c1",
+                "name": "search",
+                "arguments": "{}",
+                "output": "12 matches",
+                "is_error": False,
+            }
+        ]
+        mock_ws = MagicMock()
+        mock_ws.id = ws_id
+        mock_ws.ui._pending_approval = None
+        mock_ws.ui.get_agent_trajectory = lambda cid: steps if cid == "task1" else None
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = mock_ws
+
+        r = _build_history_app(mock_mgr, _inject_storage).get(
+            f"/v1/api/workstreams/{ws_id}/history"
+        )
+        assert r.status_code == 200
+        assistant = next(m for m in r.json()["messages"] if m.get("role") == "assistant")
+        tc = assistant["tool_calls"][0]
+        assert tc["id"] == "task1"
+        assert tc["agent_steps"] == steps
+
+    def test_no_overlay_when_not_retained(self, _inject_storage):
+        # Cold / evicted: get_agent_trajectory returns None → no agent_steps key,
+        # so the client renders the flat parent record (never a 0-step card).
+        ws_id = "ws-recall-cold"
+        self._save_task_agent_turn(_inject_storage, ws_id)
+        mock_ws = MagicMock()
+        mock_ws.id = ws_id
+        mock_ws.ui._pending_approval = None
+        mock_ws.ui.get_agent_trajectory = lambda cid: None
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = mock_ws
+
+        r = _build_history_app(mock_mgr, _inject_storage).get(
+            f"/v1/api/workstreams/{ws_id}/history"
+        )
+        assert r.status_code == 200
+        assistant = next(m for m in r.json()["messages"] if m.get("role") == "assistant")
+        assert "agent_steps" not in assistant["tool_calls"][0]
+
+
 class TestHistoryInteractive:
    """Interactive parity for the lifted ``GET /v1/api/workstreams/{ws_id}/history``."""

@@ -12,6 +12,7 @@ import base64
 import collections
 import concurrent.futures
 import contextlib
+import contextvars
 import copy
 import dataclasses
 import difflib
@@ -143,6 +144,7 @@ from turnstone.core.tools import (
 from turnstone.core.trajectory import (
    EffectStatus,
    Role,
+    TextBlock,
    ToolCall,
    Turn,
    dicts_from_turns,
@@ -277,6 +279,34 @@ def _encode_image_data_uri(raw: bytes, mime: str) -> str:
 # Upper bound on total skill content injected into system messages
 _MAX_SKILL_CONTENT: int = 32768

+# Cap on a sub-agent tool's RAW output before it enters the trajectory (the
+# in-loop truncation in _run_agent) — bounds what the sub-agent's own model sees
+# on its next turn.  Distinct from (and larger than) the recall per-step cap.
+_AGENT_TOOL_OUTPUT_CAP: int = 16000
+
+# Per-step output/arguments cap for a recalled task-agent sub-trajectory (the
+# projected step items /history attaches for the card rebuild).  Keeps the
+# recall payload small — the card shows a summary, not the full tool output.
+_AGENT_STEP_OUTPUT_CAP: int = 2000
+
+# Cap on the NUMBER of recalled steps per task agent.  With the per-step output
+# cap above, this bounds each stash entry's size (the LRU caps the agent count,
+# not per-entry bytes), so a 100+-tool agent can't blow the memory budget; the
+# overflow is replaced by one honest "(+N not retained)" marker step.
+_AGENT_STEP_COUNT_CAP: int = 100
+
+# Per-task-agent file-read tracking.  ``_read_files`` (the blind-overwrite guard's
+# memory of "files this agent has read") is a single set on the session, but the
+# parent runs task agents in a 4-wide pool — sharing it lets a sibling's read
+# suppress another agent's overwrite guard (a real blind-overwrite hazard).
+# ``_exec_task`` installs a fresh per-run set in this ContextVar for the sub-
+# agent's duration; ``_current_read_files`` reads it.  ``None`` outside a sub-
+# agent → the main session's set.  A ContextVar (not threading.local) so the
+# set/reset is balanced per ``_exec_task`` call and survives pool-thread reuse.
+_active_read_files: contextvars.ContextVar[set[str] | None] = contextvars.ContextVar(
+    "turnstone_active_read_files", default=None
+)
+
 # Cap on the *content portion* (text after ``path:lineno:``) of an
 # emitted search result line. Defends the context budget against
 # pathological lines (minified blobs, base64 data, etc.).
@@ -7466,6 +7496,15 @@ class ChatSession:
            "stop_on_error": args.get("stop_on_error") is True,
        }

+    @property
+    def _current_read_files(self) -> set[str]:
+        """The read-tracking set for the current execution context: a task
+        agent's own per-run set while one is active (so the 4-wide pool can't
+        cross-contaminate the blind-overwrite guard), else the main session's
+        ``_read_files``.  See :data:`_active_read_files`."""
+        active = _active_read_files.get()
+        return active if active is not None else self._read_files
+
    def _prepare_read_file(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
        path = args.get("path", "")
        if not path:
@@ -7519,7 +7558,7 @@ class ChatSession:
                "error": f"Error: limit must be >= 1 (got {limit})",
            }
        # Register early so a same-batch edit_file can pass the read guard.
-        self._read_files.add(resolved)
+        self._current_read_files.add(resolved)
        # Build header showing range if specified
        header = f"\u2699 read_file: {path}"
        if offset is not None or limit is not None:
@@ -7641,7 +7680,7 @@ class ChatSession:
        if mode not in ("overwrite", "append"):
            mode = "overwrite"
        is_append = mode == "append"
-        is_overwrite = exists and resolved not in self._read_files and not is_append
+        is_overwrite = exists and resolved not in self._current_read_files and not is_append

        # Build preview
        preview_parts = []
@@ -7791,7 +7830,7 @@ class ChatSession:
        resolved = os.path.realpath(path)
        is_symlink = os.path.abspath(path) != resolved

-        if resolved not in self._read_files:
+        if resolved not in self._current_read_files:
            return {
                "call_id": call_id,
                "func_name": "edit_file",
@@ -11920,11 +11959,11 @@ class ChatSession:

        all_lines, _, err = self._read_text_lines(path)
        if err:
-            self._read_files.discard(resolved)
+            self._current_read_files.discard(resolved)
            self._report_tool_result(call_id, "read_file", err, is_error=True)
            return call_id, err

-        self._read_files.add(resolved)
+        self._current_read_files.add(resolved)
        total_lines = len(all_lines)

        # Slice if offset/limit specified
@@ -11957,11 +11996,11 @@ class ChatSession:
            try:
                size = os.path.getsize(resolved)
            except OSError as e:
-                self._read_files.discard(resolved)
+                self._current_read_files.discard(resolved)
                msg = f"Error: {path}: {e}"
                self._report_tool_result(call_id, "read_file", msg, is_error=True)
                return call_id, msg
-            self._read_files.add(resolved)
+            self._current_read_files.add(resolved)
            desc = f"image (no vision, {size:,} bytes)"
            self._report_tool_result(call_id, "read_file", desc)
            return call_id, (
@@ -11973,18 +12012,18 @@ class ChatSession:
            with open(resolved, "rb") as f:
                raw = f.read()
        except FileNotFoundError:
-            self._read_files.discard(resolved)
+            self._current_read_files.discard(resolved)
            msg = f"Error: {path} not found"
            self._report_tool_result(call_id, "read_file", msg, is_error=True)
            return call_id, msg
        except Exception as e:
-            self._read_files.discard(resolved)
+            self._current_read_files.discard(resolved)
            msg = f"Error reading {path}: {e}"
            self._report_tool_result(call_id, "read_file", msg, is_error=True)
            return call_id, msg

        if len(raw) > _IMAGE_SIZE_CAP:
-            self._read_files.discard(resolved)
+            self._current_read_files.discard(resolved)
            size_mb = len(raw) / (1024 * 1024)
            cap_mb = _IMAGE_SIZE_CAP / (1024 * 1024)
            msg = (
@@ -11994,7 +12033,7 @@ class ChatSession:
            self._report_tool_result(call_id, "read_file", msg, is_error=True)
            return call_id, msg

-        self._read_files.add(resolved)
+        self._current_read_files.add(resolved)
        mime, _ = mimetypes.guess_type(path)
        if not mime:
            mime = "image/png"
@@ -12183,7 +12222,7 @@ class ChatSession:
        if err:
            self._report_tool_result(call_id, "diff_file", err, is_error=True)
            return call_id, err
-        self._read_files.add(resolved_a)
+        self._current_read_files.add(resolved_a)

        if path_b:
            label_b = path_b
@@ -12191,7 +12230,7 @@ class ChatSession:
            if err:
                self._report_tool_result(call_id, "diff_file", err, is_error=True)
                return call_id, err
-            self._read_files.add(resolved_b)
+            self._current_read_files.add(resolved_b)
        else:
            label_b = "(provided content)"
            lines_b = (content_b or "").splitlines(keepends=True)
@@ -12264,6 +12303,105 @@ class ChatSession:
        if fn is not None:
            fn()

+    @staticmethod
+    def _clip_with_count(text: str, cap: int) -> str:
+        """Head-clip ``text`` to ``cap`` chars with a uniform truncation marker.
+        The one format for sub-agent output clips — the in-loop 16k clip and the
+        recall per-step cap — distinct from the token-budget-aware
+        :meth:`_truncate_output`."""
+        if len(text) <= cap:
+            return text
+        return text[:cap] + f"\n\n... (truncated from {len(text)} chars)"
+
+    @staticmethod
+    def _iter_agent_tool_results(
+        agent_turns: list[Turn],
+    ) -> Iterator[tuple[ToolCall, Turn | None]]:
+        """Yield ``(tool_call, result_turn_or_None)`` for every sub-tool the
+        sub-agent issued, in order, pairing each call to its result FIFO per
+        call_id — a queue per id consumed once, NOT a last-wins dict, so a local
+        provider that reuses ids across turns (``call_0`` …) can't collapse
+        distinct calls onto one result.  Shared by :meth:`_project_agent_steps`
+        (recall) and :meth:`_cancel_ledger` (cancel disposition)."""
+        pending: dict[str, collections.deque[Turn]] = {}
+        for t in agent_turns:
+            if t.role is Role.TOOL and t.tool_call_id:
+                pending.setdefault(t.tool_call_id, collections.deque()).append(t)
+        for t in agent_turns:
+            if t.role is not Role.ASSISTANT:
+                continue
+            for tc in t.tool_calls:
+                q = pending.get(tc.id)
+                yield tc, (q.popleft() if q else None)
+
+    @staticmethod
+    def _project_agent_steps(agent_turns: list[Turn]) -> list[dict[str, Any]]:
+        """Project a finished sub-agent's trajectory into recall step items for
+        the task card — one per sub-tool call, matched to its result (FIFO per
+        call_id via :meth:`_iter_agent_tool_results`).
+
+        Reads the tool turn's payload directly rather than via ``Turn.text``,
+        which raises on a multimodal ``list[dict]`` result (the deferred-finding
+        landmine); a non-``str`` payload becomes a light placeholder so the
+        recall stays text-only and ``/history`` doesn't carry image bytes.  Each
+        step's output AND arguments are capped (:data:`_AGENT_STEP_OUTPUT_CAP`)
+        and the step COUNT (:data:`_AGENT_STEP_COUNT_CAP`); on overflow the most
+        RECENT steps are kept (a sub-agent's latest edits/writes matter more on
+        recall than its opening searches) behind an honest leading marker — so
+        the stash stays memory-bounded even for a 100+-tool agent and never
+        silently under-reports its step count."""
+        pairs = list(ChatSession._iter_agent_tool_results(agent_turns))
+        dropped = max(0, len(pairs) - _AGENT_STEP_COUNT_CAP)
+        # Build dicts only for the retained tail, not the dropped head.
+        tail = pairs[-_AGENT_STEP_COUNT_CAP:] if dropped else pairs
+        steps: list[dict[str, Any]] = []
+        if dropped:
+            # Leading marker naming how many earlier steps fell out — honest, and
+            # correct now that the RETAINED steps are the recent ones.
+            steps.append(
+                {
+                    "id": "",
+                    "name": "…",
+                    "arguments": "{}",
+                    "output": f"(+{dropped} earlier steps not retained)",
+                    "is_error": False,
+                }
+            )
+        for tc, res in tail:
+            output, is_error = "", False
+            if res is not None:
+                is_error = res.is_error
+                raw = (
+                    res.content[0].text
+                    if res.content and isinstance(res.content[0], TextBlock)
+                    else ""
+                )
+                output = raw if isinstance(raw, str) else "[non-text result]"
+            steps.append(
+                {
+                    "id": tc.id,
+                    "name": tc.name,
+                    # Clip arguments too: a write_file/edit_file call carries full
+                    # file content here, which the per-step output cap alone would
+                    # leave unbounded in the in-memory stash.
+                    "arguments": ChatSession._clip_with_count(tc.arguments, _AGENT_STEP_OUTPUT_CAP),
+                    "output": ChatSession._clip_with_count(output, _AGENT_STEP_OUTPUT_CAP),
+                    "is_error": is_error,
+                }
+            )
+        return steps
+
+    def _stash_agent_trajectory(self, call_id: str | None, agent_turns: list[Turn]) -> None:
+        """Retain a finished task agent's projected sub-trajectory for ``/history``
+        card recall (getattr-guarded → no-op on CLI / eval / fixtures).  Called
+        from ``_exec_task``'s ``finally`` so it captures the full record on
+        success and the partial one on cancel/error — both honest."""
+        if not call_id:
+            return
+        fn = getattr(self.ui, "stash_agent_trajectory", None)
+        if fn is not None:
+            fn(call_id, self._project_agent_steps(agent_turns))
+
    def _run_agent(
        self,
        agent_turns: list[Turn],
@@ -12461,9 +12599,16 @@ class ChatSession:
                # output_warning still nests under the task card.
                self._note_agent_child(tc_dict["id"], parent_call_id)

+                # is_error for the recalled step: the guard / prepare / unknown
+                # branches produce explicit error text; the execute paths record
+                # the authoritative flag via ``_report_tool_result`` (consumed
+                # below).  Without this every recalled sub-step reads as success
+                # and a failed sub-tool recalls as a green "done" card.
+                is_tool_error = False
                # Guard 1: block recursive agent calls.
                if tool_name == "task_agent":
                    output = "Error: agents cannot spawn further agents"
+                    is_tool_error = True
                # Guard 2: tool not in this agent's API tool list.
                elif tool_name not in tool_names:
                    output = (
@@ -12471,11 +12616,13 @@ class ChatSession:
                        f"agent mode. "
                        f"Available: {', '.join(sorted(tool_names))}"
                    )
+                    is_tool_error = True
                else:
                    prepared = self._prepare_tool(tc_dict)

                    if prepared.get("error"):
                        output = prepared["error"]
+                        is_tool_error = True
                    # Auto-execute tools in the auto_tools set.
                    elif tool_name in auto_tools:
                        # Paint the step pending under the task card before it
@@ -12484,6 +12631,7 @@ class ChatSession:
                        # paint via approve_tools instead.
                        self._paint_agent_step(parent_call_id, prepared)
                        _, output = prepared["execute"](prepared)
+                        is_tool_error = self._tool_error_flags.pop(tc_dict["id"], False)
                    # Tools not in auto_tools require user approval.
                    elif "execute" in prepared:
                        approved, _ = self.ui.approve_tools([prepared])
@@ -12491,11 +12639,15 @@ class ChatSession:
                            prepared["denied"] = True
                            prepared["denial_msg"] = "Denied by user"
                        if prepared.get("denied"):
+                            # A denial is not an execution error — keep is_error
+                            # False so recall shows the denial text, not red.
                            output = prepared.get("denial_msg", "Denied by user")
                        else:
                            _, output = prepared["execute"](prepared)
+                            is_tool_error = self._tool_error_flags.pop(tc_dict["id"], False)
                    else:
                        output = f"Unknown tool: {tool_name}"
+                        is_tool_error = True

                # Output guard: evaluate before truncation so the guard
                # sees full output (credentials split by truncation would
@@ -12511,8 +12663,8 @@ class ChatSession:
                # Truncate large tool outputs to avoid blowing context limits.
                # Agents operate autonomously; they can refine their queries
                # if truncation loses important detail.
-                if isinstance(output, str) and len(output) > 16000:
-                    output = output[:16000] + f"\n\n... (truncated from {len(output)} chars)"
+                if isinstance(output, str) and len(output) > _AGENT_TOOL_OUTPUT_CAP:
+                    output = self._clip_with_count(output, _AGENT_TOOL_OUTPUT_CAP)

                # NOTE: for a vision tool result ``output`` is a list[dict] of
                # inline content parts (read_file on an image).  It lowers back
@@ -12524,7 +12676,7 @@ class ChatSession:
                # ``create_completion`` (it currently isn't) plus content-
                # addressed byte storage — deferred to the recall/persist work
                # where that attachment path is already in scope.
-                agent_turns.append(Turn.tool(tc_dict["id"], output))
+                agent_turns.append(Turn.tool(tc_dict["id"], output, is_error=is_tool_error))
            turn += 1

        # Exhausted tool turns — force a final synthesis response.
@@ -12606,8 +12758,14 @@ class ChatSession:
            Turn.user(prompt),
        ]
        self._begin_agent_scope()
+        # Per-sub-agent file-read tracking.  COPY the parent's current set in (so
+        # the agent can edit a file the parent already read for it — no spurious
+        # "must read before editing"), but as an INDEPENDENT set, so a pool
+        # sibling's reads can't suppress THIS agent's blind-overwrite guard.  The
+        # agent's own reads merge back to the parent in ``finally``.
+        read_token = _active_read_files.set(set(self._current_read_files))
        try:
-            return call_id, self._run_agent(
+            result = self._run_agent(
                agent_turns,
                label="task",
                tools=self._task_tools,
@@ -12615,6 +12773,14 @@ class ChatSession:
                agent_alias=item.get("model_override"),
                parent_call_id=call_id,
            )
+            # Self-report the task_agent's OWN result.  The parent run-loop only
+            # reports error/denied/exception results centrally; success results
+            # rely on each tool self-reporting (bash, search, … all do), and the
+            # task_agent never did — so without this the live card has no
+            # completion signal (it stays "running" and the synthesis never
+            # renders live, only on reload).
+            self._report_tool_result(call_id, "task_agent", result)
+            return call_id, result
        except GenerationCancelled:
            # Fold back an honest disposition built from the agent's own
            # ledger.  ``agent_turns`` is mutated in place by
@@ -12629,17 +12795,35 @@ class ChatSession:
            # fabricate the acknowledgment but must not fabricate the
            # outcome … unknown, never none").
            self._tool_status[call_id] = self._cancelled_agent_status(agent_turns)
-            return call_id, self._cancelled_agent_disposition(agent_turns, "task")
+            disposition = self._cancelled_agent_disposition(agent_turns, "task")
+            self._report_tool_result(call_id, "task_agent", disposition)
+            return call_id, disposition
        except KeyboardInterrupt:
            # CLI Ctrl-C: keep the terse string and let the outer loop own
            # propagation (unchanged behavior).
            return call_id, "(task interrupted by user)"
        except Exception as e:
-            self.ui.on_info(f"[task error] {e}")
-            return call_id, f"Task error: {e}"
+            # Report as the task_agent's errored result so the card flips to
+            # "failed" and the message renders (replaces the old on_info, which
+            # was suppressed during the agent scope anyway).
+            msg = f"Task error: {e}"
+            self._report_tool_result(call_id, "task_agent", msg, is_error=True)
+            return call_id, msg
        finally:
+            # Teardown first (cheap + critical): merge the agent's reads back to
+            # the parent, restore the contextvar, drop the scope + child tags —
+            # all BEFORE the best-effort stash, so a stash raise can't leak the
+            # scope depth (phantom card nesting) or the child registry.
+            sub_reads = _active_read_files.get()
+            _active_read_files.reset(read_token)
+            if sub_reads:
+                self._current_read_files.update(sub_reads)
            self._end_agent_scope()
            self._clear_agent_children(call_id)
+            try:
+                self._stash_agent_trajectory(call_id, agent_turns)
+            except Exception:
+                log.debug("task_agent.stash_failed call_id=%s", call_id, exc_info=True)

    @staticmethod
    def _cancel_ledger(
@@ -12656,18 +12840,15 @@ class ChatSession:
        it returned, everything after never started. (The LAST gap would invert
        unknown/none on a multi-call turn.) Shared by the disposition string and
        its typed status so the two can't disagree.
+
+        Pairs via :meth:`_iter_agent_tool_results` (FIFO per call_id), so on a
+        provider that reuses ids a half-answered colliding pair is correctly read
+        as one answered + one in-flight gap, not (set-membership) both answered.
        """
-        answered: set[str] = set()
-        for t in agent_turns:
-            if t.role is Role.TOOL and t.tool_call_id:
-                answered.add(t.tool_call_id)
-        issued: list[tuple[str, bool]] = []
-        for t in agent_turns:
-            if t.role is not Role.ASSISTANT:
-                continue
-            for tc in t.tool_calls:
-                name = (tc.name or "tool").strip()
-                issued.append((name, tc.id in answered))
+        issued = [
+            ((tc.name or "tool").strip(), res is not None)
+            for tc, res in ChatSession._iter_agent_tool_results(agent_turns)
+        ]
        first_gap = next((i for i, (_n, ans) in enumerate(issued) if not ans), None)
        return issued, first_gap

@@ -13472,7 +13653,7 @@ class ChatSession:
            os.makedirs(os.path.dirname(resolved) or ".", exist_ok=True)
            with open(resolved, "a" if is_append else "w") as f:
                f.write(content)
-            self._read_files.add(resolved)
+            self._current_read_files.add(resolved)
            verb = "Appended" if is_append else "Wrote"
            msg = f"{verb} {len(content)} chars to {path}"
            self._report_tool_result(call_id, "write_file", msg)
@@ -3240,6 +3240,29 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
                messages = await asyncio.to_thread(
                    project_history_messages, to_project, awaiting_approval
                )
+                # Task-agent recall: attach each task_agent tool_call's stashed
+                # sub-trajectory (projected step items) so the client's
+                # ``replayHistory`` can rebuild the collapsible card.  Live
+                # in-memory session only — a cold/closed ws, or an entry evicted
+                # past the LRU cap, has none, so the card renders the flat parent
+                # record ("not retained"), never a fabricated 0-step card.
+                # [[HYPOTHESIS]] an unobserved sub-trajectory is unknown, not none.
+                get_traj = getattr(getattr(live_session, "ui", None), "get_agent_trajectory", None)
+                if get_traj is not None:
+                    for msg in messages:
+                        for tc in msg.get("tool_calls") or ():
+                            # Only task_agent calls ever stash — skip the rest so
+                            # we don't take the agent-state lock once per tool_call
+                            # on a long history for ids that can never match.
+                            if tc.get("name") != "task_agent":
+                                continue
+                            steps = get_traj(tc.get("id") or "")
+                            # Attach only a well-formed, non-empty list — the
+                            # ``get_agent_trajectory`` contract is ``list | None``,
+                            # and the guard keeps a malformed result out of the
+                            # JSON payload (a non-list can't be serialized).
+                            if isinstance(steps, list) and steps:
+                                tc["agent_steps"] = steps
            except Exception:
                # Operationally interesting: a persistent decoration
                # failure (missing migration, driver mismatch, schema
@@ -25,6 +25,7 @@ from __future__ import annotations

 import collections
 import contextlib
+import contextvars
 import copy
 import json
 import math
@@ -44,6 +45,25 @@ log = get_logger(__name__)
 # from bloating memory.
 _DEFAULT_LISTENER_QUEUE_MAX = 500

+# Recall: how many finished task agents' projected sub-trajectories to retain
+# in memory for /history card rebuilds.  LRU-bounded so a marathon workstream
+# can't grow it without limit; eviction (and a cold reopen, which starts empty)
+# recalls honestly as "not retained" rather than a fabricated 0-step card.
+_AGENT_TRAJECTORY_CAP = 256
+
+# Sub-agent scope, PER-THREAD.  A task agent's progress chatter reaches the UI as
+# ``on_info`` lines ("[task done] N chars", a tool's "fetched N chars") that
+# carry no call_id — they can't nest under the task card and would escape to the
+# top level, so the web pane drops them while a sub-agent runs.  A contextvar,
+# not a session-global counter, so the suppression follows the sub-agent's OWN
+# thread: a parallel sibling tool running in another pool thread keeps its info
+# lines.  Incremented/decremented by begin_/end_agent_scope around each
+# ``_run_agent``; the CLI overrides ``on_info`` and is unaffected (no card → the
+# lines are its only signal).
+_agent_scope_var: contextvars.ContextVar[int] = contextvars.ContextVar(
+    "turnstone_agent_scope_depth", default=0
+)
+

 def _resolve_event_buffer_max() -> int:
    """Read ``TURNSTONE_SSE_EVENT_BUFFER_MAX`` env override at import time.
@@ -262,16 +282,20 @@ class SessionUIBase:
        # own lock so the hot fan-out path never serializes on ``_listeners_lock``.
        self._agent_children: dict[str, str] = {}
        self._agent_children_lock = threading.Lock()
-        # Sub-agent scope depth.  A task agent's progress chatter reaches the UI
-        # as ``on_info`` lines ("[task done] N chars", a tool's "fetched N
-        # chars, extracting...") that carry no call_id — so they can't nest under
-        # the task card and would escape to the top level.  While any ``_run_agent``
-        # is in flight this is > 0 and the web pane drops those info lines (the
-        # card already shows the steps + result).  A depth, not a flag, so the
-        # parent's 4-wide parallel task pool nests correctly; guarded by
-        # ``_agent_children_lock``.  The CLI overrides ``on_info`` and is
-        # unaffected (it has no card, so the lines are its only signal).
-        self._agent_scope_depth = 0
+        # Recall store: a finished task agent's projected sub-trajectory (step
+        # items: id/name/arguments/output/is_error), keyed by its (parent)
+        # call_id, so /history can rebuild the collapsible card after a fresh
+        # connect / reopen while the workstream is still in memory.  LRU-bounded
+        # (oldest evicted past the cap); IN-MEMORY ONLY — durable persistence is
+        # deferred, so a cold reopen (new session) finds it empty and renders the
+        # flat parent record.  Guarded by ``_agent_children_lock`` (same low-rate
+        # agent-state path).  [[HYPOTHESIS]] the sub-trajectory is the ledger;
+        # an absent one is unknown ("not retained"), never none ("0 steps").
+        self._agent_trajectories: collections.OrderedDict[str, list[dict[str, Any]]] = (
+            collections.OrderedDict()
+        )
+        # (Sub-agent ``on_info`` suppression is per-thread via the module-level
+        # ``_agent_scope_var`` contextvar — no instance field.)
        # Approval blocking — the worker thread calls approve_tools
        # which waits on _approval_event; the /approve endpoint sets
        # it via resolve_approval.
@@ -554,17 +578,40 @@ class SessionUIBase:

    def begin_agent_scope(self) -> None:
        """Enter a task agent's execution.  Until the matching
-        :meth:`end_agent_scope`, the web pane drops ``on_info`` lines (the task
-        card carries the sub-agent's visible output, so the info chatter would
-        only escape to the top level).  Depth-counted for the parent's parallel
-        task pool; the session brackets each ``_run_agent`` with this pair."""
-        with self._agent_children_lock:
-            self._agent_scope_depth += 1
+        :meth:`end_agent_scope`, the web pane drops ``on_info`` lines from THIS
+        thread (the task card carries the sub-agent's visible output, so the info
+        chatter would only escape to the top level).  Per-thread via
+        :data:`_agent_scope_var` so a parallel SIBLING tool in another pool
+        thread isn't suppressed; depth-counted for safety though task agents
+        don't nest.  The session brackets each ``_run_agent`` with this pair."""
+        _agent_scope_var.set(_agent_scope_var.get() + 1)

    def end_agent_scope(self) -> None:
        """Leave a task agent's execution (see :meth:`begin_agent_scope`)."""
+        _agent_scope_var.set(max(0, _agent_scope_var.get() - 1))
+
+    def stash_agent_trajectory(self, call_id: str, steps: list[dict[str, Any]]) -> None:
+        """Retain a finished task agent's projected sub-trajectory (step items)
+        keyed by its call_id so ``/history`` can rebuild the card on a fresh
+        connect / reopen while the workstream is in memory.  LRU-bounded; the
+        newest write is most-recently-used, oldest evicted past the cap.  See
+        :data:`_AGENT_TRAJECTORY_CAP` and the field comment in ``__init__``."""
+        if not call_id:
+            return
        with self._agent_children_lock:
-            self._agent_scope_depth = max(0, self._agent_scope_depth - 1)
+            self._agent_trajectories[call_id] = steps
+            self._agent_trajectories.move_to_end(call_id)
+            while len(self._agent_trajectories) > _AGENT_TRAJECTORY_CAP:
+                self._agent_trajectories.popitem(last=False)
+
+    def get_agent_trajectory(self, call_id: str) -> list[dict[str, Any]] | None:
+        """Read a stashed sub-trajectory, or ``None`` if not retained (evicted,
+        or a cold reopen with an empty store).  ``None`` is the honest "unknown"
+        signal — the caller renders the flat parent record, never a 0-step card."""
+        if not call_id:
+            return None
+        with self._agent_children_lock:
+            return self._agent_trajectories.get(call_id)

    def _register_listener(
        self, maxsize: int = _DEFAULT_LISTENER_QUEUE_MAX
@@ -2504,12 +2551,13 @@ class SessionUIBase:
            log.warning("Failed to record usage event", exc_info=True)

    def on_info(self, message: str) -> None:
-        # Inside a task agent, progress chatter ("[task done] N chars", a tool's
-        # "fetched N chars, extracting...") carries no call_id, so it can't nest
-        # under the task card — drop it on the web pane rather than let it escape
-        # to the top level.  The card shows the steps + result; the CLI overrides
-        # this method and keeps the lines (no card there).
-        if self._agent_scope_depth > 0:
+        # Inside a task agent (on THIS thread), progress chatter ("[task done] N
+        # chars", a tool's "fetched N chars") carries no call_id, so it can't
+        # nest under the task card — drop it on the web pane rather than let it
+        # escape to the top level.  Per-thread, so a parallel sibling tool's info
+        # still shows.  The card shows the steps + result; the CLI overrides this
+        # method and keeps the lines (no card there).
+        if _agent_scope_var.get() > 0:
            return
        self._enqueue({"type": "info", "message": message})

@@ -1979,6 +1979,10 @@ class Pane {
    // Replaces a JSON.stringify→dataset→JSON.parse round-trip with an
    // in-memory map keyed by call_id.
    const pendingAssessments = {};
+    // Task-agent recall: call_id -> card .conv-agent wrap, so the tool-result
+    // branch can flip the card's done/error state from the task's own result
+    // (mirroring the live appendToolOutput), not from sub-step errors.
+    const agentCardWraps = {};
    let lastToolBlock = null;
    for (let i = 0; i < messages.length; i++) {
      const msg = messages[i];
@@ -2054,30 +2058,8 @@ class Pane {
            msg.tool_calls.forEach((tc, idx) => {
              // Synthesize the live `item` shape from the stored tool_call so
              // replay renders the SAME .conv-row as the live path.
-              let header = "";
-              try {
-                const args = JSON.parse(tc.arguments);
-                if (tc.name === "bash") {
-                  header = String(Object.values(args)[0] || "");
-                } else {
-                  const parts = [];
-                  const keys = Object.keys(args);
-                  for (let k = 0; k < keys.length; k++) {
-                    const val = args[keys[k]];
-                    let valStr =
-                      val === null || val === undefined ? "null" : String(val);
-                    if (valStr.length > 80)
-                      valStr = valStr.substring(0, 77) + "...";
-                    parts.push(keys[k] + ": " + valStr);
-                  }
-                  header = parts.join("\n");
-                }
-              } catch (e) {
-                header = String(tc.arguments || "").substring(0, 100);
-              }
-              const item = { func_name: tc.name, call_id: tc.id || "", header };
              const row = buildToolDiv(
-                item,
+                synthToolItem(tc),
                indexLabel(idx, msg.tool_calls.length),
              );
              // Verdict anchored to THIS row; replay verdicts are final.
@@ -2087,6 +2069,13 @@ class Pane {
                );
              }
              block.appendChild(row);
+              // Task-agent recall: rebuild the collapsible card under this row
+              // from its stashed sub-trajectory (the /history `agent_steps`
+              // overlay).  Absent ⇒ flat parent row (cold / not-retained).
+              if (tc.agent_steps && tc.agent_steps.length) {
+                const wrap = this._replayAgentCard(row, tc.agent_steps);
+                if (tc.id) agentCardWraps[tc.id] = wrap;
+              }
              // Output-guard finding — deferred until the tool result lands so
              // it anchors under the output (mirrors live showOutputWarning).
              if (
@@ -2152,11 +2141,7 @@ class Pane {
            if (media) {
              insertChained(buildMediaEmbed(media, stripped));
            } else {
-              const out = renderToolOutput(stripped, isToolError);
-              if (out.textContent.split("\n").length > 10) {
-                makeCollapsible(out);
-              }
-              insertChained(out);
+              insertChained(renderCollapsibleOutput(stripped, isToolError));
            }
          }
          if (
@@ -2177,6 +2162,14 @@ class Pane {
              delete pendingAssessments[msg.tool_call_id];
            }
          }
+          // Task-agent recall: flip its card done/error from the task's OWN
+          // result (matching the live appendToolOutput) — NOT from sub-step
+          // errors, since a sub-tool can fail and the agent still synthesize.
+          if (msg.tool_call_id && agentCardWraps[msg.tool_call_id]) {
+            agentCardWraps[msg.tool_call_id].dataset.state = msg.is_error
+              ? "error"
+              : "done";
+          }
        }
      } else if (msg.role === "system") {
        // First-class operator-context turn (output-guard finding, user
@@ -2553,6 +2546,28 @@ class Pane {
    card.label.textContent = n === 1 ? "1 step" : n + " steps";
  }

+  _replayAgentCard(row, steps) {
+    // Rebuild a finished task agent's card from its recalled step items
+    // (/history `agent_steps`), mirroring the live _routeAgentItems nesting so a
+    // reload looks identical: a collapsed body of step rows, each with its
+    // result.  Recall is terminal — no live approval affordances.  Card state
+    // defaults to "done"; the caller flips it to "error" from the task's OWN
+    // result (the role==="tool" branch), matching the live path — sub-step
+    // errors don't decide it (an agent can recover and synthesize fine).
+    const card = buildAgentCardBody();
+    steps.forEach((step) => {
+      card.body.appendChild(buildToolDiv(synthToolItem(step), ""));
+      const out = stripAnsi(String(step.output || "")).trim();
+      if (out) {
+        card.body.appendChild(renderCollapsibleOutput(out, !!step.is_error));
+      }
+    });
+    card.wrap.dataset.state = "done";
+    this._updateAgentLabel(card);
+    row.appendChild(card.wrap);
+    return card.wrap;
+  }
+
  appendToolOutput(callId, name, output, isError) {
    // Capture pin before the streamEl removal + result insertion change
    // scrollHeight — see announceToolBlock.  The result block is the other
@@ -2650,7 +2665,10 @@ class Pane {
      }
    }

-    const out = renderToolOutput(stripped, isError);
+    // The media / MCP-error dispatch above both early-return, so by here it's
+    // the plain-output path — the shared helper applies (test_app_js pins that
+    // tryParseMcpError precedes this renderer call).
+    const out = renderCollapsibleOutput(stripped, isError);

    // Mark the parent approval block as errored
    if (
@@ -2662,10 +2680,6 @@ class Pane {
      appendToolErrorBadge(parentBlock);
    }

-    if (out.textContent.split("\n").length > 10) {
-      makeCollapsible(out);
-    }
-
    target.after(out);
    this.scrollToBottom(stick);
  }
@@ -3314,6 +3328,35 @@ function buildToolDiv(item, indexLabel) {
  return row;
 }

+// Synthesize the live `item` shape ({func_name, call_id, header}) from a stored
+// tool_call ({name, id, arguments}) so /history replay AND task-agent card
+// recall render the SAME .conv-row as the live path.  Header derivation mirrors
+// the live serialize: bash shows the command; other tools show `key: value`
+// lines, values clipped at 80 chars; unparseable args fall back to a raw clip.
+function synthToolItem(tc) {
+  tc = tc || {};
+  let header = "";
+  try {
+    const args = JSON.parse(tc.arguments);
+    if (tc.name === "bash") {
+      header = String(Object.values(args)[0] || "");
+    } else {
+      const parts = [];
+      const keys = Object.keys(args);
+      for (let k = 0; k < keys.length; k++) {
+        const val = args[keys[k]];
+        let valStr = val === null || val === undefined ? "null" : String(val);
+        if (valStr.length > 80) valStr = valStr.substring(0, 77) + "...";
+        parts.push(keys[k] + ": " + valStr);
+      }
+      header = parts.join("\n");
+    }
+  } catch (e) {
+    header = String(tc.arguments || "").substring(0, 100);
+  }
+  return { func_name: tc.name, call_id: tc.id || "", header };
+}
+
 function renderVerdictBadge(verdict, judgePending) {
  // Thin wrapper over the shared builder (returns a fragment [badge, detail]).
  return buildConvVerdict(verdict, { judgePending });
@@ -3381,6 +3424,16 @@ function renderToolOutput(stripped, isError) {
  return out;
 }

+// Shared recipe: render tool output and auto-collapse past 10 lines.  The one
+// source for the live appendToolOutput's plain-output path, the /history
+// tool-result replay, and the task-agent card recall — so the collapse
+// threshold can't drift between them.
+function renderCollapsibleOutput(stripped, isError) {
+  const out = renderToolOutput(stripped, isError);
+  if (out.textContent.split("\n").length > 10) makeCollapsible(out);
+  return out;
+}
+
 function buildMediaEmbed(media, rawJson) {
  const wrapper = document.createElement("div");
  wrapper.className = "media-embed";