refactor(ui): converge interactive history delivery onto REST-first model

Interactive fetched conversation history as an inline SSE `history` event on
every (re)connect — a multi-MB payload — while coord fetches it once via REST
`GET /history` and uses SSE for live deltas only. This converges interactive
onto coord's model so both kinds share one history-delivery pattern, the
prerequisite for lifting the `/command` (rewind/retry) verb to coord.

Backend (server.py, core/session_routes.py):
- `_interactive_events_replay` and the `/command` resume/rewind branches no
  longer emit the inline `history` SSE event; the open/create-resume paths
  emit `clear_ui` only. The `/events` stream no longer carries conversation
  history — REST `GET /history` is the source (acceptable on 1.6.0aN).
- Removed the now-orphaned `events_replay_prepare` hook.
- `_build_history` retained as the canonical wire-shape reference for the
  decoration/parity tests (no production callers post-convergence).

Frontend (ui/static/app.js, ui/static/index.html):
- `_loadHistoryThenConnect` fetches REST `/history`, renders, then opens SSE
  (mirrors coord's `init()` ordering); wired into the seven ws-assign sites.
- `clear_ui` re-renders via a REST refetch and dispatches the edit-and-resend
  latch; `replay_truncated` re-syncs (skipped mid-stream so it cannot clobber
  an in-flight turn). The `case "history"` SSE handler is removed.

New shared module (shared_static/history_normalize.js):
- `normalizeHistoryMessages` converts the raw provider-native REST shape
  (nested tool_calls, `_source`/`_reminders`/`_attachments_meta` side-channels,
  multipart content, no derived flags) into the projected shape `replayHistory`
  renders. Pure/DOM-free and node-unit-tested. This is a transitional bridge —
  a server-side wire-shape unification (folding this projection back into the
  server so interactive, coord, and coord's inline raw-handling collapse onto
  one shape) is planned to replace it.

Tests: backend replay-omits-history regression; a node-executed normalizer
projection test (incl. the orphan->pending and denial-propagation edges); and
REST-first wiring guards in test_app_js.py.

tests/test_app_js.py

@@ -9,12 +9,18 @@ manual testing.
 
 from __future__ import annotations
 
+import json
 import re
+import shutil
+import subprocess
 from pathlib import Path
 
 import pytest
 
 _APP_JS = Path(__file__).resolve().parent.parent / "turnstone/ui/static/app.js"
+_NORMALIZE_JS = (
+    Path(__file__).resolve().parent.parent / "turnstone/shared_static/history_normalize.js"
+)
 
 
 def _pane_method_offset(body: str, name: str) -> int:
@@ -717,8 +723,6 @@ def test_phase8_consent_url_prefix_check_in_click_handler() -> None:
 # smoke; the function is pure (no DOM dependency) so it transplants
 # cleanly into a standalone node invocation.
 
-import subprocess  # noqa: E402
-
 
 def _slice_balanced_body(body: str, anchor: int) -> str | None:
     """Slice ``body`` from ``anchor`` (which must point at or just before
@@ -1397,3 +1401,138 @@ def test_coord_connectsse_onerror_preserves_native_reconnect() -> None:
     assert onerror is not None, "coordinator.js evtSource.onerror not found"
     passed, reason = _onerror_preserves_native_reconnect(onerror, "evtSource")
     assert passed, f"coordinator.js connectSSE.onerror regressed: {reason}"
+
+
+def test_interactive_history_is_rest_first_not_sse() -> None:
+    """PR A converged interactive onto coord's REST-first history
+    model: first paint and post-rewind re-render fetch ``GET /history``
+    over REST (``_loadHistoryThenConnect`` / ``_refetchHistory``), and
+    the server no longer replays the conversation inline over SSE — so
+    the client must no longer consume a ``history`` SSE event. Guards
+    against a regression that re-couples first paint to the removed
+    inline-history replay."""
+    body = _APP_JS.read_text(encoding="utf-8")
+    assert "_loadHistoryThenConnect" in body, (
+        "REST-first first-paint helper missing — interactive must fetch "
+        "history via GET /history before connecting SSE (coord's model)."
+    )
+    assert "_refetchHistory" in body
+    # Pre-PR-A interactive had no REST /history fetch; the quoted URL
+    # segment only appears in the new fetch concatenation.
+    assert '"/history"' in body
+    # The inline SSE ``history`` event is no longer emitted server-side,
+    # so the client must not handle it (history is REST-only now).
+    assert 'case "history":' not in body
+    # The raw REST /history shape is provider-native and differs from the
+    # projected shape replayHistory renders — interactive must normalize it
+    # and must load the shared normalizer script.
+    assert "normalizeHistoryMessages" in body, (
+        "interactive must normalize the raw REST /history shape before "
+        "replayHistory (see history_normalize.js)"
+    )
+    idx = (Path(__file__).resolve().parent.parent / "turnstone/ui/static/index.html").read_text(
+        encoding="utf-8"
+    )
+    assert "/shared/history_normalize.js" in idx, (
+        "index.html must load the shared history normalizer"
+    )
+
+
+@pytest.mark.skipif(shutil.which("node") is None, reason="node not installed")
+def test_normalize_history_projects_rest_shape_to_wire_shape() -> None:
+    """``normalizeHistoryMessages`` (history_normalize.js) must convert the
+    raw provider-native REST ``/history`` payload into the projected shape
+    ``replayHistory`` renders. This pins the JS projection SHAPE — it runs
+    only the JS under node and does NOT execute the Python projection, so
+    it is a shape-regression guard, not a cross-language parity check (a
+    true parity assertion against ``decorate_history_messages`` is left to
+    the planned server-side unification). Run under node (the function is
+    pure / DOM-free) so the projection logic is actually exercised, not
+    just string-present. Guards the interactive render bug:
+    the REST shape nests tool_calls under ``function`` and uses
+    ``_source`` / ``_reminders`` / ``_attachments_meta`` side-channels, which
+    the pre-normalizer renderer choked on (``JSON.parse(undefined)`` →
+    ``TypeError`` → render abort masked as an empty pane)."""
+    raw = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "hi"},
+                {"type": "image_url", "image_url": {}},
+            ],
+            "_source": "system_nudge",
+            "_reminders": [
+                {"type": "correction", "text": "fix", "secret": "x"},
+                {"type": "", "text": ""},
+            ],
+            "_attachments_meta": [{"kind": "image", "filename": "p.png", "mime_type": "image/png"}],
+        },
+        {
+            "role": "assistant",
+            "content": "ok",
+            "reasoning": "think",
+            "tool_calls": [
+                {
+                    "id": "c1",
+                    "type": "function",
+                    "function": {"name": "web_search", "arguments": '{"q":1}'},
+                    "verdict": {"tier": "judge"},
+                }
+            ],
+        },
+        {"role": "tool", "tool_call_id": "c1", "content": "res"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [{"id": "c2", "function": {"name": "bash", "arguments": "{}"}}],
+        },
+        {"role": "tool", "tool_call_id": "c2", "content": "Denied by user: no"},
+        {"role": "tool", "tool_call_id": "cx", "content": "Error: boom"},
+        # mid-conversation orphan: tool_call with no result that is NOT the
+        # last tool turn → must still render (not vanish), so NOT pending.
+        {
+            "role": "assistant",
+            "tool_calls": [{"id": "c_mid", "function": {"name": "g", "arguments": "{}"}}],
+        },
+        # trailing orphan: last tool turn with no result → pending (awaiting).
+        {
+            "role": "assistant",
+            "tool_calls": [{"id": "c3", "function": {"name": "f", "arguments": "{}"}}],
+        },
+    ]
+    node_script = (
+        "const {normalizeHistoryMessages} = require("
+        + json.dumps(str(_NORMALIZE_JS))
+        + "); process.stdout.write(JSON.stringify(normalizeHistoryMessages("
+        + json.dumps(raw)
+        + ")));"
+    )
+    proc = subprocess.run(
+        ["node", "-e", node_script],
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+
+    # tool_calls flattened (the crash fix): name/arguments/verdict top-level
+    assert out[1]["tool_calls"][0]["name"] == "web_search"
+    assert out[1]["tool_calls"][0]["arguments"] == '{"q":1}'
+    assert out[1]["tool_calls"][0]["verdict"]["tier"] == "judge"
+    # multipart user content collapsed; side-channels surfaced top-level
+    assert out[0]["content"] == "hi"
+    assert out[0]["attachments"][0]["filename"] == "p.png"
+    assert out[0]["source"] == "system_nudge"
+    reminder_types = [r["type"] for r in out[0]["reminders"]]
+    assert reminder_types == ["correction"]  # empty entry filtered
+    assert "secret" not in out[0]["reminders"][0]  # unknown key stripped
+    # derived + propagated flags (the REST shape pre-sets none of these)
+    assert out[4]["denied"] is True  # tool deny derived from content prefix
+    assert out[3]["denied"] is True  # propagated to the parent assistant turn
+    assert out[5]["is_error"] is True  # tool error derived from content prefix
+    # pending: ONLY the last tool turn with an orphan (proxy for awaiting);
+    # a mid-conversation orphan still renders its tool block (bug-2 fix).
+    assert out[7].get("pending") is True  # trailing orphan = last tool turn → pending
+    assert "pending" not in out[6]  # mid-conversation orphan → renders, NOT pending
+    assert "pending" not in out[1]  # completed tool_call is not pending

tests/test_server_authz.py

@@ -13,7 +13,7 @@ import json
 import queue
 import threading
 from typing import Any
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 from starlette.testclient import TestClient
@@ -1063,6 +1063,22 @@ class TestInteractiveEventsLifted:
         out = list(_interactive_events_replay(ws, ui, request))
         assert out == []
 
+    def test_events_replay_omits_conversation_history(self):
+        """PR A: conversation history is no longer replayed over SSE.
+        The frontend fetches it via ``GET /history`` (REST) on page
+        load and re-fetches on ``clear_ui``; the replay must neither
+        build nor yield a ``history`` event (which previously shipped a
+        multi-MB message list on every (re)connect)."""
+        from turnstone.server import _interactive_events_replay
+
+        ws, ui, request = _make_interactive_replay_mocks(
+            _pending_approval={"type": "approve_request", "items": []},
+        )
+        with patch("turnstone.server._build_history") as mock_build_history:
+            out = list(_interactive_events_replay(ws, ui, request))
+        mock_build_history.assert_not_called()
+        assert "history" not in {ev["type"] for ev in out}
+
     def test_events_path_keyed_url_resolves_to_404_for_unknown_ws(self, app_client):
         """``GET /v1/api/workstreams/{ws_id}/events`` returns 404 for an
         unknown ws_id. Pre-1.5 the same intent was tested against

tests/test_sse_reconnect_replay.py

@@ -489,7 +489,6 @@ def _wire_events_handler(ui: _ConcreteUI) -> Any:
         not_found_label="Workstream not found",
         audit_action_prefix="workstream",
         events_replay=None,
-        events_replay_prepare=None,
     )
     return make_events_handler(cfg)
 

turnstone/core/session_routes.py

@@ -395,15 +395,6 @@ class SessionEndpointConfig:
     # separate ``/history`` endpoint and doesn't render the per-tab
     # status bar). Kinds that don't need pre-replay wire ``None``.
     events_replay: EventsReplay | None = None
-    # async (ws, ui, request) -> None. Kind-specific async pre-step
-    # the lifted ``events`` body awaits BEFORE iterating
-    # ``events_replay``. Lets a kind move blocking storage I/O off
-    # the event loop (via ``asyncio.to_thread``) and stash results
-    # on ``request.state`` for the sync replay generator to read.
-    # Interactive uses it to pre-load intent_verdicts +
-    # output_assessments so ``_build_history``'s decoration stays
-    # off the hot path. Coord wires ``None``.
-    events_replay_prepare: Callable[..., Any] | None = None
     # (request) -> Executor for the SSE live-loop's blocking
     # ``queue.get`` wait. Interactive returns the dedicated
     # ``request.app.state.sse_executor`` (200-thread pool) so SSE
@@ -1341,13 +1332,10 @@ def make_open_handler(
         # emit_rehydrated path).
         if cfg.open_post_load is not None:
             try:
-                # Off-loop: interactive's post_load runs the sync
-                # ``_build_history`` (storage I/O for verdict
-                # indexes + message reconstruction) — without the
-                # to_thread wrap this blocks the event loop on every
-                # workstream open, mirroring the SSE replay path
-                # that's already protected via
-                # ``events_replay_prepare``.
+                # Off-loop: interactive's post_load does blocking
+                # storage I/O (a workstream display-name lookup) that
+                # would otherwise stall the event loop on every
+                # workstream open.
                 await asyncio.to_thread(cfg.open_post_load, request, ws)
             except Exception:
                 # Post-load is observational — never let a hook bug
@@ -1648,19 +1636,6 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler:
                     # last BUFFERED id (or none on truly-fresh
                     # connect), which is what the server can replay.
                     if replay_cb is not None:
-                        # Kind-specific async prep — runs before the
-                        # sync replay generator iterates so blocking
-                        # storage I/O lands in the executor pool
-                        # rather than the event loop's hot path.
-                        if cfg.events_replay_prepare is not None:
-                            try:
-                                await cfg.events_replay_prepare(ws, ui, request)
-                            except Exception:
-                                log.debug(
-                                    "ws.events.replay_prepare_failed ws=%s",
-                                    ws_id[:8],
-                                    exc_info=True,
-                                )
                         try:
                             for ev in replay_cb(ws, ui, request):
                                 yield {"data": json.dumps(ev)}

turnstone/server.py

@@ -428,6 +428,15 @@ class WebUI(SessionUIBase):
 # :func:`make_history_handler` (the /history REST endpoint coord uses
 # as its primary history loader) share them so the two surfaces don't
 # drift on the wire shape they emit.
+#
+# NB: ``_build_history`` has NO production callers post-convergence — the
+# live interactive + coord history projection is ``make_history_handler``
+# (``reconstruct_messages`` + ``decorate_history_messages``), and the
+# browser mirrors that shape in ``shared_static/history_normalize.js``.
+# This function is retained as the canonical wire-shape REFERENCE that the
+# decoration / parity tests assert against; if the projected wire shape
+# changes, update ``history_normalize.js`` (and ``decorate_history_messages``)
+# to match — editing this builder alone touches test-only code.
 
 
 def _build_history(
@@ -466,10 +475,9 @@ def _build_history(
     # ``ChatSession._apply_reminders_for_provider``.
     #
     # Verdict + output-assessment lookup tables — populated either
-    # inline (sync call sites) or pre-loaded by an async caller via
-    # asyncio.to_thread (see _load_verdict_indexes).  Pre-loading is
-    # what keeps _build_history off the event loop's hot path on the
-    # SSE replay generator path.
+    # inline (default) or from pre-loaded dicts passed by an async
+    # caller via asyncio.to_thread (see _load_verdict_indexes), which
+    # keeps the storage I/O off the event loop when called from one.
     if verdicts is not None and assessments is not None:
         verdicts_by_call_id = verdicts
         assessments_by_call_id = assessments
@@ -953,45 +961,23 @@ def _audit_close_workstream(
     )
 
 
-async def _interactive_events_replay_prepare(ws: Workstream, ui: Any, request: Request) -> None:
-    """Async pre-step run before ``_interactive_events_replay`` iterates.
-
-    Loads ``intent_verdicts`` + ``output_assessments`` for the
-    workstream off the event loop (via ``asyncio.to_thread``) and
-    stashes the result on ``request.state.verdict_indexes``. The sync
-    replay generator reads from there and passes the dicts into
-    ``_build_history`` so the storage I/O never blocks the event loop
-    on the SSE replay path.
-
-    Best-effort: if the workstream has no session or no ws_id, leaves
-    ``request.state.verdict_indexes`` unset and ``_build_history``
-    falls back to the inline storage call (sync path).
-    """
-    del ui  # not needed; lookup is keyed on ws.session._ws_id
-    session = ws.session
-    if session is None:
-        return
-    ws_id = getattr(session, "_ws_id", "") or ""
-    if not ws_id:
-        return
-    indexes = await asyncio.to_thread(_load_verdict_indexes, ws_id)
-    request.state.verdict_indexes = indexes
-
-
 def _interactive_events_replay(
     ws: Workstream, ui: Any, request: Request
 ) -> Iterable[dict[str, Any]]:
     """Initial SSE replay payload for interactive ``events`` connections.
 
-    Pre-lift ``events_sse`` yielded five things on connect: a
-    ``connected`` event with model + skip_permissions; a ``status``
-    event with the workstream's last token usage + context %; the
-    full conversation ``history`` (with pending-approval flagging on
-    the last assistant entry's tool calls); the pending approval
-    prompt + cached intent verdicts (if a prompt is pending); the
-    pending plan-review (if a review is pending). The lifted
-    ``make_events_handler`` body delegates that yield sequence to
-    this callback so the kind-specific shape stays in this module.
+    Yields a ``connected`` event (model + skip_permissions), a
+    ``status`` event with the workstream's last token usage + context %
+    (when a turn has completed), the pending approval prompt + cached
+    intent verdicts (if a prompt is pending), and the pending
+    plan-review (if a review is pending). The lifted
+    ``make_events_handler`` body delegates that yield sequence to this
+    callback so the kind-specific shape stays in this module.
+
+    Conversation history is NOT replayed over SSE: the frontend fetches
+    it via ``GET /history`` on page load and re-fetches on the
+    ``clear_ui`` signal (coord's REST-first model), keeping a multi-MB
+    message list off every (re)connect.
 
     Pure read — never mutates ``ws`` / ``ui`` / ``session``.
     """
@@ -1007,29 +993,9 @@ def _interactive_events_replay(
     # field add.
     yield from session_replay_preamble(session, ui)
 
-    # History replay — pending-approval flag rides on the last
-    # assistant entry's tool_calls so the client renders them as
-    # awaiting approval rather than already approved.  Verdict /
-    # assessment indexes were pre-loaded off the event loop by
-    # _interactive_events_replay_prepare; passing them in here keeps
-    # _build_history's storage I/O out of the sync generator path.
-    pending_approval = getattr(ui, "_pending_approval", None)
-    cached_indexes = getattr(request.state, "verdict_indexes", None)
-    if isinstance(cached_indexes, tuple) and len(cached_indexes) == 2:
-        verdicts, assessments = cached_indexes
-    else:
-        verdicts, assessments = None, None
-    history = _build_history(
-        session,
-        has_pending_approval=pending_approval is not None,
-        verdicts=verdicts,
-        assessments=assessments,
-    )
-    if history:
-        yield {"type": "history", "messages": history}
-
     # Pending approval re-injection (so a reconnecting tab sees the
     # prompt) + cached LLM verdicts received since the prompt fired.
+    pending_approval = getattr(ui, "_pending_approval", None)
     if pending_approval is not None:
         yield pending_approval
         with ui._ws_lock:
@@ -1055,10 +1021,11 @@ def _interactive_open_post_load(request: Request, ws: Workstream) -> None:
     1. Sync the workstream's name to the persisted display alias
        (a user-renamed workstream stores its alias separately from
        the manager's in-memory name).
-    2. Replay clear_ui + history onto the per-workstream UI listener
-       queue so a freshly-connected browser tab sees the conversation
-       state. Only fires when ``ws.session.messages`` is non-empty
-       (resume succeeded and there's history to show).
+    2. Emit ``clear_ui`` onto the per-workstream UI listener queue so a
+       connected browser tab re-fetches conversation state over REST
+       ``GET /history`` (the REST-first model — history is no longer
+       replayed inline over SSE). Only fires when ``ws.session.messages``
+       is non-empty (resume succeeded and there's history to show).
     3. Enqueue ``ws_created`` onto the global SSE queue so dashboards
        and other multi-workstream consumers see the rehydrate. The
        handler-side emission is the load-bearing path on interactive;
@@ -1072,9 +1039,6 @@ def _interactive_open_post_load(request: Request, ws: Workstream) -> None:
     session = ws.session
     if isinstance(ui, WebUI) and session is not None and session.messages:
         ui._enqueue({"type": "clear_ui"})
-        history = _build_history(session)
-        if history:
-            ui._enqueue({"type": "history", "messages": history})
 
     gq: queue.Queue[dict[str, Any]] | None = getattr(request.app.state, "global_queue", None)
     if gq is not None:
@@ -1704,20 +1668,15 @@ async def command(request: Request) -> JSONResponse:
         if cmd_word in ("/clear", "/new"):
             ui._enqueue({"type": "clear_ui"})
         elif cmd_word == "/resume":
+            # clear_ui signals the frontend to re-fetch history via REST.
             ui._enqueue({"type": "clear_ui"})
-            history = await asyncio.to_thread(_build_history, ws.session)
-            if history:
-                ui._enqueue({"type": "history", "messages": history})
         elif cmd_word in ("/rewind", "/retry"):
-            # Refresh frontend with truncated history. Always emit the
-            # history event even when empty: editing the first message
-            # rewinds to zero messages, and the frontend dispatches the
-            # queued edit-and-resend from the history handler — skipping
-            # the event on an empty list orphans `_pendingEditSend` and
-            # leaves the composer stuck in busy.
+            # clear_ui signals the frontend to re-fetch the (now
+            # truncated) history via REST and dispatch any queued
+            # edit-and-resend once it lands. Fires even on a rewind to
+            # zero messages: the frontend keys the resend off this
+            # signal, not an inline history payload.
             ui._enqueue({"type": "clear_ui"})
-            history = await asyncio.to_thread(_build_history, ws.session)
-            ui._enqueue({"type": "history", "messages": history})
             # Audit trail
             storage = getattr(request.app.state, "auth_storage", None)
             if storage:
@@ -2189,10 +2148,8 @@ async def _interactive_create_post_install(
                 ws.name = user_name
             ui = ws.ui
             if isinstance(ui, WebUI):
+                # clear_ui signals the frontend to re-fetch history via REST.
                 ui._enqueue({"type": "clear_ui"})
-                history = await asyncio.to_thread(_build_history, ws.session)
-                if history:
-                    ui._enqueue({"type": "history", "messages": history})
             with contextlib.suppress(queue.Full):
                 gq.put_nowait({"type": "ws_rename", "ws_id": ws.id, "name": ws.name})
 
@@ -4047,7 +4004,6 @@ def create_app(
         open_resolve_alias=_resolve_workstream_alias,
         open_post_load=_interactive_open_post_load,
         events_replay=_interactive_events_replay,
-        events_replay_prepare=_interactive_events_replay_prepare,
         # Pre-lift ``events_sse`` used the dedicated 200-thread
         # ``sse_executor`` so SSE polling stayed isolated from
         # every other ``asyncio.to_thread`` caller in the process

turnstone/shared_static/history_normalize.js

@@ -0,0 +1,235 @@
+// Shared history normaliser — converts the raw `reconstruct_messages`
+// shape returned by `GET /v1/api/workstreams/{ws_id}/history` into the
+// projected wire shape that `_build_history` (turnstone/server.py) emits
+// and the interactive `replayHistory` renderer consumes.
+//
+// Why this exists: the REST `/history` endpoint returns provider-native
+// OpenAI-shaped messages — nested `tool_calls[].function.{name,arguments}`,
+// side-channel `_source` / `_reminders` / `_attachments_meta`, multipart
+// user content, and NO derived `denied` / `is_error` / `pending` flags.
+// This normaliser does that projection client-side so the REST-fetched
+// history renders identically to the legacy SSE replay.
+//
+// LIVE PROJECTION REFERENCE: the production REST `/history` shape comes
+// from `reconstruct_messages` + `decorate_history_messages` /
+// `extract_reasoning_for_history` (turnstone/core/history_decoration.py) —
+// THAT is the shape this mirror must track. `server.py::_build_history` is
+// the historical SSE projection that originally defined the target wire
+// shape, but it has NO production callers post-convergence (test-only
+// reference impl) — do not "keep lockstep" by editing it alone.
+// (Transitional bridge: a planned server-side wire-shape unification will
+// fold this projection back into the server so both kinds consume it.)
+//
+// `reasoning` and per-tool-call `verdict` / `output_assessment` are already
+// projected top-level by the REST decoration and pass through unchanged.
+// `advisories` pass through for STRING tool content (decorate strips the
+// `<tool_output>` envelope server-side); LIST-content tool results are a
+// KNOWN GAP — their envelope is not stripped here and a carried advisory is
+// dropped (uncommon: image/structured MCP result + concurrent interjection;
+// closed by the planned server-side unification).
+//
+// Pure + DOM-free so it can be unit-tested under node
+// (tests/test_app_js.py::test_normalize_history_*).
+
+const _WATCH_REMINDER_OPTIONAL_KEYS = [
+  "watch_name",
+  "command",
+  "poll_count",
+  "max_polls",
+  "is_final",
+];
+
+const _TOOL_ERROR_PREFIXES = [
+  "Error",
+  "Command timed out",
+  "Search timed out",
+  "Unknown tool:",
+  "JSON parse error:",
+  "MCP prompt timed out",
+  "MCP prompt error",
+];
+
+function normalizeHistoryMessages(messages) {
+  if (!Array.isArray(messages)) return [];
+
+  // Pre-scan: which tool_call_ids have a result message? An assistant
+  // tool_call with no result is an orphan → its turn is "pending"
+  // (awaiting approval on a live reconnect, or interrupted) and must not
+  // render a fake "approved" badge. Mirrors `_build_history`'s
+  // has_pending_approval handling for the realistic cases.
+  const resultedCallIds = new Set();
+  for (const m of messages) {
+    if (m && m.role === "tool" && m.tool_call_id) {
+      resultedCallIds.add(String(m.tool_call_id));
+    }
+  }
+
+  const out = [];
+  for (const msg of messages) {
+    if (!msg || typeof msg !== "object") continue;
+    const role = msg.role;
+    let content = msg.content;
+    let attachments = null;
+
+    // (1) Collapse multipart user content (text + image_url + document
+    //     parts) to a plain string + a derived attachment list.
+    if (role === "user" && Array.isArray(content)) {
+      const textParts = [];
+      const meta = [];
+      for (const part of content) {
+        if (!part || typeof part !== "object") continue;
+        if (part.type === "text") {
+          textParts.push(String(part.text || ""));
+        } else if (part.type === "image_url") {
+          meta.push({ kind: "image", filename: "", mime_type: "" });
+        } else if (part.type === "document") {
+          const d = part.document || {};
+          meta.push({
+            kind: "text",
+            filename: String(d.name || ""),
+            mime_type: String(d.media_type || ""),
+          });
+        }
+      }
+      content = textParts.join("\n");
+      if (meta.length) attachments = meta;
+    }
+
+    // (2) The authoritative `_attachments_meta` side-channel wins when
+    //     present (carries image filenames the image_url part can't).
+    const sideMeta = msg._attachments_meta;
+    if (Array.isArray(sideMeta) && sideMeta.length) {
+      attachments = sideMeta
+        .filter((x) => x && typeof x === "object")
+        .map((x) => ({
+          kind: String(x.kind || ""),
+          filename: String(x.filename || ""),
+          mime_type: String(x.mime_type || ""),
+        }));
+    }
+
+    const entry = { role: role, content: content };
+    if (attachments && attachments.length) entry.attachments = attachments;
+
+    // (3) `_source` side-channel → top-level `source` (drives the
+    //     `.msg.user.system-nudge` marker).
+    if (msg._source) entry.source = String(msg._source);
+
+    // (4) `_reminders` side-channel → top-level `reminders`, filtered +
+    //     key-projected (mirrors `_build_history`'s reminder filter).
+    if (Array.isArray(msg._reminders)) {
+      const clean = [];
+      for (const r of msg._reminders) {
+        if (!r || typeof r !== "object") continue;
+        const rtype = String(r.type || "");
+        const rtext = String(r.text || "");
+        if (!rtype && !rtext) continue;
+        const c = { type: rtype, text: rtext };
+        for (const k of _WATCH_REMINDER_OPTIONAL_KEYS) {
+          if (k in r) c[k] = r[k];
+        }
+        clean.push(c);
+      }
+      if (clean.length) entry.reminders = clean;
+    }
+
+    // (5) Reasoning is already projected top-level by the REST
+    //     decoration — pass it through.
+    if (msg.reasoning) entry.reasoning = msg.reasoning;
+
+    // (6) Flatten OpenAI-nested tool_calls `{id, function:{name,
+    //     arguments}}` → `{id, name, arguments}` that `replayHistory`
+    //     reads, carrying the decoration (`verdict` / `output_assessment`,
+    //     which `decorate_tool_call` placed top-level on the call).
+    if (Array.isArray(msg.tool_calls) && msg.tool_calls.length) {
+      const tcs = [];
+      for (const tc of msg.tool_calls) {
+        if (!tc || typeof tc !== "object") continue;
+        const fn = tc.function || {};
+        const id = tc.id || "";
+        const flat = {
+          id: id,
+          name: fn.name || tc.name || "",
+          arguments:
+            fn.arguments != null
+              ? fn.arguments
+              : tc.arguments != null
+                ? tc.arguments
+                : "",
+        };
+        if (tc.verdict) flat.verdict = tc.verdict;
+        if (tc.output_assessment) flat.output_assessment = tc.output_assessment;
+        tcs.push(flat);
+      }
+      entry.tool_calls = tcs;
+    }
+
+    // (7) Tool results: carry `tool_call_id` + `advisories` (already
+    //     top-level on the REST shape), coerce list content to text, and
+    //     derive `denied` / `is_error` from the content prefix — the REST
+    //     path does NOT pre-set these (mirrors `_build_history`).
+    if (role === "tool") {
+      if (msg.tool_call_id) entry.tool_call_id = String(msg.tool_call_id);
+      if (Array.isArray(msg.advisories) && msg.advisories.length) {
+        entry.advisories = msg.advisories;
+      }
+      if (Array.isArray(content)) {
+        content = content
+          .filter((p) => p && p.type === "text")
+          .map((p) => String(p.text || ""))
+          .join("\n");
+        entry.content = content;
+      }
+      const c = typeof content === "string" ? content : "";
+      if (c.startsWith("Denied by user") || c.startsWith("Blocked")) {
+        entry.denied = true;
+      }
+      if (msg.is_error || _TOOL_ERROR_PREFIXES.some((p) => c.startsWith(p))) {
+        entry.is_error = true;
+      }
+    }
+
+    out.push(entry);
+  }
+
+  // (8) Propagate denial from a tool result to its parent assistant turn
+  //     so the tool block renders the denied (not approved) badge.
+  //     Mirrors `_build_history`'s last-assistant propagation.
+  let lastAssistantIdx = -1;
+  for (let i = 0; i < out.length; i++) {
+    if (out[i].tool_calls) {
+      lastAssistantIdx = i;
+    } else if (
+      out[i].role === "tool" &&
+      out[i].denied &&
+      lastAssistantIdx >= 0
+    ) {
+      out[lastAssistantIdx].denied = true;
+    }
+  }
+
+  // (9) Mark `pending` ONLY on the LAST assistant tool-call turn, and only
+  //     when it has an orphan (a tool_call with no result in the loaded
+  //     window) — the proxy for "awaiting approval". Mirrors
+  //     `_build_history`'s reversed single-turn pending. A mid-conversation
+  //     cancelled/interrupted tool call is also an orphan but must still
+  //     render its tool block (not vanish), so it is NOT marked pending.
+  for (let i = out.length - 1; i >= 0; i--) {
+    const e = out[i];
+    if (e.tool_calls && e.tool_calls.length) {
+      const hasOrphan = e.tool_calls.some(
+        (tc) => tc.id && !resultedCallIds.has(String(tc.id)),
+      );
+      if (hasOrphan) e.pending = true;
+      break;
+    }
+  }
+
+  return out;
+}
+
+// Dual-mode: a browser global (loaded via <script>) AND a CommonJS export
+// so node-based unit tests can require it without a DOM.
+if (typeof module !== "undefined" && module.exports) {
+  module.exports = { normalizeHistoryMessages };
+}

turnstone/ui/static/app.js

@@ -738,6 +738,45 @@ class Pane {
     };
   }
 
+  _loadHistoryThenConnect(wsId) {
+    // Mirror coord's init() ordering: render history from REST first,
+    // THEN open the live stream. Disconnect any existing stream up
+    // front so stray events from the previously-assigned ws don't paint
+    // into the pane mid-fetch. History is no longer replayed over SSE,
+    // so this REST fetch is the sole first-paint source; connectSSE's
+    // in_progress_snapshot still covers a generation that lands between
+    // the fetch and the stream opening.
+    this.disconnectSSE();
+    this._refetchHistory(wsId).finally(() => this.connectSSE(wsId));
+  }
+
+  async _refetchHistory(wsId) {
+    // Fetch conversation history over REST. Used for first paint (before
+    // connecting SSE) and to re-render after a clear_ui signal (rewind /
+    // retry / resume / open). The FETCH is wrapped (network/parse failure
+    // → empty pane); the render is deliberately OUTSIDE the catch so a
+    // render bug surfaces loudly instead of being masked as an empty pane.
+    const id = wsId || this.wsId;
+    let data = null;
+    try {
+      const r = await authFetch(
+        "/v1/api/workstreams/" + encodeURIComponent(id) + "/history",
+      );
+      if (r && r.ok) data = await r.json();
+    } catch (err) {
+      data = null;
+    }
+    if (data) {
+      // The REST /history payload is provider-native (nested tool_calls,
+      // `_source`/`_reminders`/`_attachments_meta` side-channels, multipart
+      // content, no derived denied/is_error/pending). Normalize it to the
+      // projected shape replayHistory renders — see history_normalize.js.
+      this.replayHistory(normalizeHistoryMessages(data.messages || []));
+    } else {
+      this.showEmptyState();
+    }
+  }
+
   _refetchWorkstreamsAndReassign() {
     // Lifted from the pre-PR-D ``onerror`` body.  Triggered when the
     // focused pane sees its EventSource enter the error state — pulls
@@ -809,7 +848,7 @@ class Pane {
               setTimeout(
                 ((pp) => {
                   return () => {
-                    pp.connectSSE(pp.wsId);
+                    pp._loadHistoryThenConnect(pp.wsId);
                   };
                 })(p3),
                 this.retryDelay,
@@ -1135,30 +1174,51 @@ class Pane {
         }
         break;
 
-      case "history":
-        this.replayHistory(evt.messages);
-        // Dispatch pending edit-and-resend after rewind history arrives
-        if (this._pendingEditSend) {
-          const editText = this._pendingEditSend;
-          this._pendingEditSend = null;
-          this.setBusy(true);
-          this.addUserMessage(editText);
-          authFetch(
-            "/v1/api/workstreams/" + encodeURIComponent(this.wsId) + "/send",
-            {
-              method: "POST",
-              headers: { "Content-Type": "application/json" },
-              body: JSON.stringify({ message: editText }),
-            },
-          ).catch((err) => {
-            this.addErrorMessage("Connection error: " + err.message);
+      case "clear_ui":
+        // Conversation was structurally reset (rewind / retry / resume /
+        // open / fork). Empty the pane for immediate feedback, then
+        // re-render from REST and dispatch any queued edit-and-resend
+        // once the (possibly truncated) history lands. The resend keys
+        // off this signal rather than an inline history SSE event.
+        this.messagesEl.replaceChildren();
+        this._refetchHistory(this.wsId)
+          .then(() => {
+            if (!this._pendingEditSend) return;
+            const editText = this._pendingEditSend;
+            this._pendingEditSend = null;
+            this.setBusy(true);
+            this.addUserMessage(editText);
+            authFetch(
+              "/v1/api/workstreams/" + encodeURIComponent(this.wsId) + "/send",
+              {
+                method: "POST",
+                headers: { "Content-Type": "application/json" },
+                body: JSON.stringify({ message: editText }),
+              },
+            ).catch((err) => {
+              this.addErrorMessage("Connection error: " + err.message);
+              this.setBusy(false);
+            });
+          })
+          .catch((err) => {
+            // The render runs outside _refetchHistory's try/catch by design;
+            // if it throws, don't strand the queued edit-and-resend — clear
+            // the latch + busy so the composer recovers.
+            this._pendingEditSend = null;
             this.setBusy(false);
+            this.addErrorMessage("Failed to reload history: " + err.message);
           });
-        }
         break;
 
-      case "clear_ui":
-        this.messagesEl.replaceChildren();
+      case "replay_truncated":
+        // Reconnect buffer evicted past our last-seen event id — the
+        // live recovery replay no longer carries history, so re-sync
+        // from REST. Skip while a turn is mid-stream: the recovery
+        // floor's in_progress_snapshot already paints it, and an async
+        // refetch's replaceChildren() would detach the live bubble so
+        // content deltas render nowhere. Re-syncs on the next clean
+        // (re)connect.
+        if (!this.currentAssistantEl) this._refetchHistory(this.wsId);
         break;
     }
   }
@@ -1328,7 +1388,8 @@ class Pane {
     const turnsToRewind = userMsgs.length - idx;
 
     this.setBusy(true);
-    // Store pending send — dispatched when the rewind history event arrives
+    // Store pending send — dispatched from the clear_ui handler once
+    // the rewind's truncated history is re-fetched over REST.
     this._pendingEditSend = newText;
     authFetch("/v1/api/command", {
       method: "POST",
@@ -1480,7 +1541,9 @@ class Pane {
                   cmd.textContent = parts.join("\n");
                 }
               } catch (e) {
-                cmd.textContent = tc.arguments.substring(0, 100);
+                // Defensive: never let a non-string `arguments` (or a
+                // parse failure) escalate into a render-aborting throw.
+                cmd.textContent = String(tc.arguments || "").substring(0, 100);
               }
               div.appendChild(cmd);
               // Verdict badge — anchor to THIS tool's row (div) rather
@@ -2332,7 +2395,7 @@ function splitPane(paneId, direction) {
   renderLayout();
   setFocusedPane(newPane.id);
   newPane.showEmptyState();
-  newPane.connectSSE(newWsId);
+  newPane._loadHistoryThenConnect(newWsId);
 }
 
 function closePane(paneId) {
@@ -3270,7 +3333,7 @@ function switchTab(wsId) {
   pane.showEmptyState();
   pane.updateWsName();
   renderTabBar();
-  pane.connectSSE(wsId);
+  pane._loadHistoryThenConnect(wsId);
 
   if (!_historyNavigation) {
     history.pushState({ turnstone: "workstream", wsId: wsId }, "");
@@ -3702,7 +3765,7 @@ function _reassignPanesForClosedWs(closedWsId, tabIdsBeforeClose) {
         dp.messagesEl.replaceChildren();
         dp.showEmptyState();
         dp.updateWsName();
-        dp.connectSSE(remaining[0]);
+        dp._loadHistoryThenConnect(remaining[0]);
       }
     }
     if (focusedPaneId && panes[focusedPaneId]) {
@@ -3790,7 +3853,7 @@ function _reassignPanesForClosedWs(closedWsId, tabIdsBeforeClose) {
       p.messagesEl.replaceChildren();
       p.showEmptyState();
       p.updateWsName();
-      p.connectSSE(newWsId);
+      p._loadHistoryThenConnect(newWsId);
       usedWsIds[newWsId] = true;
     } else if (countLeaves(splitRoot) > 1) {
       // No unused workstream available — close redundant pane
@@ -3803,7 +3866,7 @@ function _reassignPanesForClosedWs(closedWsId, tabIdsBeforeClose) {
         p.messagesEl.replaceChildren();
         p.showEmptyState();
         p.updateWsName();
-        p.connectSSE(remaining[0]);
+        p._loadHistoryThenConnect(remaining[0]);
       }
     }
   }
@@ -6682,7 +6745,7 @@ function initWorkstreams() {
       for (let id in panes) {
         if (!panes[id].evtSource) {
           panes[id].showEmptyState();
-          panes[id].connectSSE(panes[id].wsId);
+          panes[id]._loadHistoryThenConnect(panes[id].wsId);
         }
       }
       const params = new URLSearchParams(location.search);

turnstone/ui/static/index.html

@@ -620,6 +620,7 @@
       ];
     </script>
     <script src="/shared/utils.js"></script>
+    <script src="/shared/history_normalize.js"></script>
     <script src="/shared/cards.js"></script>
     <script src="/shared/toast.js"></script>
     <script src="/shared/composer.js"></script>