diff --git a/docs/architecture.md b/docs/architecture.md index df87a24f..0d600d42 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1058,11 +1058,12 @@ warns if the summary was truncated. unhandled promise rejections - **Pending approval across tab switches**: `WebUI._pending_approval` stores the `approve_request` event payload while the session is blocked waiting - for user response. On SSE reconnect (e.g., switching back to the tab), - the event is re-injected after history replay. `_build_history` marks the - pending tool call as `"pending": true` so `replayHistory` skips the - false `✓ approved` badge; the live approval UI is rendered by the - re-injected event instead. + for user response. On tab switch / reconnect the pane reloads history via + REST `GET /history` and then reconnects SSE; the live approval event is + re-injected. The server-side `project_history_messages` projection marks + the trailing orphan tool-call turn `"pending": true` so `replayHistory` + skips the false `✓ approved` badge; the live approval UI is rendered by + the re-injected event instead. - **Browser history integration**: `history.pushState` is called in `switchTab()` with `{turnstone: 'workstream', wsId}`. The initial state is seeded with `history.replaceState({turnstone: 'dashboard'})` on load. The diff --git a/tests/test_app_js.py b/tests/test_app_js.py index f0f0d4cf..e6b8c000 100644 --- a/tests/test_app_js.py +++ b/tests/test_app_js.py @@ -9,18 +9,13 @@ 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: @@ -1430,116 +1425,20 @@ def test_interactive_history_is_rest_first_not_sse() -> None: # 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)" + # The server now projects the canonical wire shape at /history + # (make_history_handler → project_history_messages), so interactive + # feeds the payload straight to replayHistory — the client-side + # normaliser (history_normalize.js) was retired. + assert "normalizeHistoryMessages" not in body, ( + "the client-side history normaliser was retired — interactive must " + "consume the server-projected /history shape directly" + ) + assert "this.replayHistory(data.messages" in body, ( + "interactive must feed the projected REST /history payload straight to replayHistory" ) 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" + assert "/shared/history_normalize.js" not in idx, ( + "the retired history_normalize.js script tag must be removed from index.html" ) - - -@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 diff --git a/tests/test_build_history_reminders.py b/tests/test_build_history_reminders.py deleted file mode 100644 index 511eaa30..00000000 --- a/tests/test_build_history_reminders.py +++ /dev/null @@ -1,266 +0,0 @@ -"""Tests for ``turnstone.server._build_history`` reminder + source surfacing. - -The replay path (``_build_history``) projects the ``_source`` and -``_reminders`` side-channels onto the wire entry the frontend -consumes. Persisted via migration 050 (Commit 1) so multi-tab / -multi-device replay sees the same metacognitive bubble shape the -originating tab saw live. -""" - -from __future__ import annotations - -from types import SimpleNamespace -from typing import Any -from unittest.mock import patch - -from turnstone.server import _build_history - - -def _make_stub_session(messages: list[dict[str, Any]]) -> Any: - """Minimal ChatSession-shaped stub. ``_build_history`` only reads - ``session.messages`` plus calls ``_load_verdict_indexes(ws_id)`` — - the latter we patch out below. - """ - return SimpleNamespace(messages=messages, _ws_id="ws-test") - - -def _build(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: - """Run ``_build_history`` against a stub session, bypassing the - verdicts / output-assessment storage round-trip (no tool_calls in - these tests, so the indexes are unused anyway). - """ - session = _make_stub_session(messages) - with patch( - "turnstone.server._load_verdict_indexes", - return_value=({}, {}), - ): - return _build_history(session) - - -class TestSourceSurfacing: - def test_source_surfaces_when_set(self) -> None: - msg = { - "role": "user", - "content": "", - "_source": "system_nudge", - } - history = _build([msg]) - assert len(history) == 1 - assert history[0]["source"] == "system_nudge" - - def test_source_absent_when_unset(self) -> None: - msg = {"role": "user", "content": "hello"} - history = _build([msg]) - assert "source" not in history[0] - - -class TestRemindersWidening: - def test_watch_triggered_optional_fields_propagate(self) -> None: - """The widened payload (Commit 2) carries watch_name / command / - poll_count / max_polls / is_final on each ``watch_triggered`` - reminder so the frontend renders ``.msg.watch-result``. - """ - msg = { - "role": "user", - "content": "", - "_source": "system_nudge", - "_reminders": [ - { - "type": "watch_triggered", - "text": "$ ls\nfile.txt", - "watch_name": "w1", - "command": "ls", - "poll_count": 2, - "max_polls": 100, - "is_final": False, - } - ], - } - history = _build([msg]) - assert history[0]["source"] == "system_nudge" - assert history[0]["reminders"] == [ - { - "type": "watch_triggered", - "text": "$ ls\nfile.txt", - "watch_name": "w1", - "command": "ls", - "poll_count": 2, - "max_polls": 100, - "is_final": False, - } - ] - - def test_legacy_two_field_reminders_still_work(self) -> None: - """Producers without optional fields (correction / denial / - idle_children) keep the legacy ``{type, text}`` shape — the - widened filter just doesn't add anything beyond that.""" - msg = { - "role": "user", - "content": "noted", - "_reminders": [{"type": "correction", "text": "watch out"}], - } - history = _build([msg]) - assert history[0]["reminders"] == [{"type": "correction", "text": "watch out"}] - - def test_unknown_keys_are_dropped(self) -> None: - """The wire-layer filter projects on a known set of keys so a - future producer accidentally stuffing arbitrary fields can't - leak them through replay. - """ - msg = { - "role": "user", - "content": "x", - "_reminders": [ - { - "type": "correction", - "text": "hi", - "secret": "leak-me", - "internal_id": 42, - } - ], - } - history = _build([msg]) - clean = history[0]["reminders"][0] - assert "secret" not in clean - assert "internal_id" not in clean - assert clean == {"type": "correction", "text": "hi"} - - def test_malformed_reminder_skipped(self) -> None: - """A non-dict / empty entry is filtered out instead of breaking - the rest of the list (mirrors the defensive filter in - ``_apply_reminders_for_provider``). - """ - msg = { - "role": "user", - "content": "x", - "_reminders": [ - "garbage string", - {"type": "", "text": ""}, # empty type + text → drop - {"type": "denial", "text": "ok"}, - ], - } - history = _build([msg]) - assert history[0]["reminders"] == [{"type": "denial", "text": "ok"}] - - -class _StubRegistry: - """Minimal model registry — only ``get_config`` is read by - ``_build_history``.""" - - def __init__(self, surface_persisted_reasoning: bool = True) -> None: - self._cfg = SimpleNamespace(surface_persisted_reasoning=surface_persisted_reasoning) - - def get_config(self, alias: str) -> Any: - return self._cfg - - -def _build_with_registry( - messages: list[dict[str, Any]], - surface_persisted_reasoning: bool = True, -) -> list[dict[str, Any]]: - session = SimpleNamespace( - messages=messages, - _ws_id="ws-test", - _registry=_StubRegistry(surface_persisted_reasoning=surface_persisted_reasoning), - _model_alias="claude-opus-4-7", - ) - with patch( - "turnstone.server._load_verdict_indexes", - return_value=({}, {}), - ): - return _build_history(session) - - -class TestReasoningSurfacing: - """Phase 1 — surface stored Anthropic thinking blocks on the - history payload so refresh-the-page rehydrates the reasoning bubble. - Drives through the real ``AnthropicProvider`` extractor (no mock-of- - extractor) — only the model registry is stubbed. - """ - - def test_reasoning_surfaces_for_anthropic_thinking_msg(self) -> None: - msg = { - "role": "assistant", - "content": "Final answer.", - "_provider_content": [ - {"type": "thinking", "thinking": "let me think", "signature": "s"}, - {"type": "text", "text": "Final answer."}, - ], - } - history = _build_with_registry([msg], surface_persisted_reasoning=True) - assert len(history) == 1 - assert history[0]["reasoning"] == "let me think" - - def test_reasoning_empty_when_persist_flag_false(self) -> None: - msg = { - "role": "assistant", - "content": "Final answer.", - "_provider_content": [ - {"type": "thinking", "thinking": "hidden", "signature": "s"}, - ], - } - history = _build_with_registry([msg], surface_persisted_reasoning=False) - assert "reasoning" not in history[0] - - def test_provider_content_never_in_wire_entry(self) -> None: - # The build path does not copy ``_provider_content`` into the - # entry dict regardless of flag — wire payload stays tight. - msg = { - "role": "assistant", - "content": "Final answer.", - "_provider_content": [ - {"type": "thinking", "thinking": "x", "signature": "s"}, - ], - } - history = _build_with_registry([msg], surface_persisted_reasoning=True) - assert "_provider_content" not in history[0] - - def test_no_reasoning_field_when_provider_content_missing(self) -> None: - msg = {"role": "assistant", "content": "plain answer"} - history = _build_with_registry([msg], surface_persisted_reasoning=True) - assert "reasoning" not in history[0] - - def test_no_reasoning_field_for_non_assistant_messages(self) -> None: - # Defensive — user/tool messages with a stray _provider_content - # do not get the reasoning field stamped. - msgs: list[dict[str, Any]] = [ - {"role": "user", "content": "hi"}, - { - "role": "tool", - "tool_call_id": "c1", - "content": "out", - "_provider_content": [{"type": "thinking", "thinking": "leak", "signature": "s"}], - }, - ] - history = _build_with_registry(msgs, surface_persisted_reasoning=True) - assert "reasoning" not in history[0] - assert "reasoning" not in history[1] - - def test_default_true_when_registry_lookup_raises(self) -> None: - # Conservative default — Phase 1 spec mandates rehydration on - # refresh. A registry/alias mismatch must not silently kill the - # bubble. - class BrokenRegistry: - def get_config(self, alias: str) -> Any: - raise KeyError(alias) - - session = SimpleNamespace( - messages=[ - { - "role": "assistant", - "content": "x", - "_provider_content": [ - {"type": "thinking", "thinking": "still works", "signature": "s"} - ], - } - ], - _ws_id="ws-test", - _registry=BrokenRegistry(), - _model_alias="missing-alias", - ) - with patch( - "turnstone.server._load_verdict_indexes", - return_value=({}, {}), - ): - history = _build_history(session) - assert history[0]["reasoning"] == "still works" diff --git a/tests/test_history_decoration.py b/tests/test_history_decoration.py index 407b5533..a7dad4c0 100644 --- a/tests/test_history_decoration.py +++ b/tests/test_history_decoration.py @@ -1,10 +1,10 @@ """Unit tests for ``turnstone.core.history_decoration``. -The decoration helpers are shared between two surfaces — interactive's -SSE replay (``_build_history``) and the lifted ``/history`` REST -endpoint (``make_history_handler``, used by both interactive and -coord). Pinning the wire shape here lets a future schema/projection -change land in one file rather than spread across the two surfaces. +The decoration helpers compose the single ``/history`` REST projection +pipeline (``make_history_handler``, used by both interactive and coord): +``decorate_history_messages`` + ``extract_reasoning_for_history`` + +``project_history_messages``. Pinning the wire shape here lets a future +schema/projection change land in one file. """ from __future__ import annotations diff --git a/tests/test_history_projection.py b/tests/test_history_projection.py new file mode 100644 index 00000000..775a37a1 --- /dev/null +++ b/tests/test_history_projection.py @@ -0,0 +1,291 @@ +"""Tests for the REST ``/history`` projection helpers. + +``project_history_messages`` does the structural projection — collapse +multipart user content, surface the ``_source`` / ``_reminders`` +side-channels, flatten tool_calls, derive ``denied`` / ``is_error`` / +``pending`` — that the interactive ``replayHistory`` renderer and the +coordinator dashboard both consume directly. ``extract_reasoning_for_history`` +surfaces stored reasoning text and strips the internal ``_provider_content`` +lane. Together they compose the ``make_history_handler`` pipeline. + +Persisted via migration 050 (source / reminders) and migration 052 +(reasoning) so multi-tab / multi-device replay sees the same metacognitive +bubble shape the originating tab saw live. +""" + +from __future__ import annotations + +from typing import Any + +from turnstone.core.history_decoration import ( + extract_reasoning_for_history, + project_history_messages, +) + + +class TestSourceSurfacing: + def test_source_surfaces_when_set(self) -> None: + history = project_history_messages( + [{"role": "user", "content": "", "_source": "system_nudge"}] + ) + assert len(history) == 1 + assert history[0]["source"] == "system_nudge" + + def test_source_absent_when_unset(self) -> None: + history = project_history_messages([{"role": "user", "content": "hello"}]) + assert "source" not in history[0] + + +class TestRemindersWidening: + def test_watch_triggered_optional_fields_propagate(self) -> None: + """The widened payload carries watch_name / command / poll_count / + max_polls / is_final on each ``watch_triggered`` reminder so the + frontend renders ``.msg.watch-result``. + """ + history = project_history_messages( + [ + { + "role": "user", + "content": "", + "_source": "system_nudge", + "_reminders": [ + { + "type": "watch_triggered", + "text": "$ ls\nfile.txt", + "watch_name": "w1", + "command": "ls", + "poll_count": 2, + "max_polls": 100, + "is_final": False, + } + ], + } + ] + ) + assert history[0]["source"] == "system_nudge" + assert history[0]["reminders"] == [ + { + "type": "watch_triggered", + "text": "$ ls\nfile.txt", + "watch_name": "w1", + "command": "ls", + "poll_count": 2, + "max_polls": 100, + "is_final": False, + } + ] + + def test_legacy_two_field_reminders_still_work(self) -> None: + """Producers without optional fields (correction / denial / + idle_children) keep the legacy ``{type, text}`` shape.""" + history = project_history_messages( + [ + { + "role": "user", + "content": "noted", + "_reminders": [{"type": "correction", "text": "watch out"}], + } + ] + ) + assert history[0]["reminders"] == [{"type": "correction", "text": "watch out"}] + + def test_unknown_keys_are_dropped(self) -> None: + """The wire-layer filter projects on a known set of keys so a + future producer accidentally stuffing arbitrary fields can't leak + them through replay.""" + history = project_history_messages( + [ + { + "role": "user", + "content": "x", + "_reminders": [ + { + "type": "correction", + "text": "hi", + "secret": "leak-me", + "internal_id": 42, + } + ], + } + ] + ) + clean = history[0]["reminders"][0] + assert "secret" not in clean + assert "internal_id" not in clean + assert clean == {"type": "correction", "text": "hi"} + + def test_malformed_reminder_skipped(self) -> None: + """A non-dict / empty entry is filtered out instead of breaking the + rest of the list (mirrors the defensive filter in + ``_apply_reminders_for_provider``).""" + history = project_history_messages( + [ + { + "role": "user", + "content": "x", + "_reminders": [ + "garbage string", + {"type": "", "text": ""}, # empty type + text → drop + {"type": "denial", "text": "ok"}, + ], + } + ] + ) + assert history[0]["reminders"] == [{"type": "denial", "text": "ok"}] + + +class TestReasoningSurfacing: + """``extract_reasoning_for_history`` surfaces stored Anthropic thinking + blocks on the assistant message (so refresh-the-page rehydrates the + reasoning bubble) and strips the internal ``_provider_content`` lane. + Drives through the real ``AnthropicProvider`` extractor — only the + surface flag is a parameter (the active-model flag resolution lives in + ``make_history_handler``, covered by its REST tests). + """ + + def test_reasoning_surfaces_for_anthropic_thinking_msg(self) -> None: + msgs: list[dict[str, Any]] = [ + { + "role": "assistant", + "content": "Final answer.", + "_provider_content": [ + {"type": "thinking", "thinking": "let me think", "signature": "s"}, + {"type": "text", "text": "Final answer."}, + ], + } + ] + extract_reasoning_for_history(msgs, surface_persisted_reasoning_flag=True) + assert msgs[0]["reasoning"] == "let me think" + + def test_reasoning_empty_when_persist_flag_false(self) -> None: + msgs: list[dict[str, Any]] = [ + { + "role": "assistant", + "content": "Final answer.", + "_provider_content": [ + {"type": "thinking", "thinking": "hidden", "signature": "s"}, + ], + } + ] + extract_reasoning_for_history(msgs, surface_persisted_reasoning_flag=False) + assert "reasoning" not in msgs[0] + + def test_provider_content_always_stripped(self) -> None: + # The internal lane is stripped regardless of the flag — the wire + # payload never carries it. + for flag in (True, False): + msgs: list[dict[str, Any]] = [ + { + "role": "assistant", + "content": "Final answer.", + "_provider_content": [ + {"type": "thinking", "thinking": "x", "signature": "s"}, + ], + } + ] + extract_reasoning_for_history(msgs, surface_persisted_reasoning_flag=flag) + assert "_provider_content" not in msgs[0] + + def test_no_reasoning_field_when_provider_content_missing(self) -> None: + msgs: list[dict[str, Any]] = [{"role": "assistant", "content": "plain answer"}] + extract_reasoning_for_history(msgs, surface_persisted_reasoning_flag=True) + assert "reasoning" not in msgs[0] + + def test_no_reasoning_field_for_non_assistant_messages(self) -> None: + # Defensive — user/tool messages are skipped entirely; a stray + # _provider_content on them never gets a reasoning field stamped. + msgs: list[dict[str, Any]] = [ + {"role": "user", "content": "hi"}, + { + "role": "tool", + "tool_call_id": "c1", + "content": "out", + "_provider_content": [{"type": "thinking", "thinking": "leak", "signature": "s"}], + }, + ] + extract_reasoning_for_history(msgs, surface_persisted_reasoning_flag=True) + assert "reasoning" not in msgs[0] + assert "reasoning" not in msgs[1] + + +class TestProjectHistoryMessages: + """End-to-end shape test — the Python port of the retired client-side + ``normalizeHistoryMessages`` node test. Feeds the provider-native + ``reconstruct_messages`` storage shape (nested tool_calls, + ``_source`` / ``_reminders`` / ``_attachments_meta`` side-channels, + multipart content, no derived flags) and asserts the canonical + projected wire shape both UIs consume. + """ + + def test_projects_storage_shape_to_wire_shape(self) -> None: + raw: list[dict[str, Any]] = [ + { + "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", # already stamped by extract_reasoning_for_history + "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), 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": "{}"}}], + }, + ] + out = project_history_messages(raw) + + # tool_calls flattened: 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" # _attachments_meta wins + assert out[0]["source"] == "system_nudge" + assert [r["type"] for r in out[0]["reminders"]] == ["correction"] # empty filtered + assert "secret" not in out[0]["reminders"][0] # unknown key stripped + # reasoning passes through (already stamped upstream) + assert out[1]["reasoning"] == "think" + # derived + propagated flags (the storage shape pre-sets none) + 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 on the last orphan turn + assert out[7].get("pending") is True # trailing orphan c3 awaiting + assert out[6].get("pending") is not True # mid-conversation orphan c_mid renders + assert out[1].get("pending") is not True # resolved c1 diff --git a/tests/test_reasoning_audit_log_discipline.py b/tests/test_reasoning_audit_log_discipline.py index be824c92..58acc331 100644 --- a/tests/test_reasoning_audit_log_discipline.py +++ b/tests/test_reasoning_audit_log_discipline.py @@ -2,9 +2,8 @@ Phase 1 of optional reasoning persistence surfaces stored thinking blocks on the ``/history`` payload (UI rehydration). The bytes ride -through the helper (``extract_reasoning_for_history``), through the -provider extractor (``AnthropicProvider.extract_reasoning_text``), and -through the server build path (``_build_history``). +through the helper (``extract_reasoning_for_history``) and the provider +extractor (``AnthropicProvider.extract_reasoning_text``). This test pins the security-sensitive contract: @@ -37,7 +36,6 @@ from turnstone.core.providers._anthropic import AnthropicProvider from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider from turnstone.core.providers._openai_responses import OpenAIResponsesProvider from turnstone.core.providers._protocol import StreamChunk, UsageInfo -from turnstone.server import _build_history _MARKER = "SECRET_REASONING_MARKER_xyz123_unlikely_collision" @@ -169,36 +167,6 @@ class TestReasoningAuditLogDiscipline: f"extract_reasoning_for_history leaked reasoning text into INFO+ logs: {offending}" ) - def test_build_history_does_not_log_reasoning(self) -> None: - registry = SimpleNamespace( - get_config=lambda alias: SimpleNamespace(surface_persisted_reasoning=True) - ) - session = SimpleNamespace( - messages=[self._thinking_msg(_MARKER)], - _ws_id="ws-audit", - _registry=registry, - _model_alias="claude-opus-4-7", - ) - captured, patchers = _capture_log_calls() - for p in patchers: - p.start() - try: - with patch( - "turnstone.server._load_verdict_indexes", - return_value=({}, {}), - ): - history = _build_history(session) - assert history[0]["reasoning"] == _MARKER # UI-bound is allowed - finally: - for p in patchers: - p.stop() - offending = [ - (lvl, args, kwargs) - for lvl, args, kwargs in captured - if _payload_contains_marker(args, kwargs) - ] - assert offending == [], f"_build_history leaked reasoning text into INFO+ logs: {offending}" - # ------------------------------------------------------------------ # Phase 2 + Phase 3 surfaces — added in response to a code-review # finding that the original 4-test coverage missed every code path diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index 19a17b49..ea9dbac4 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -13,7 +13,7 @@ import json import queue import threading from typing import Any -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock import pytest from starlette.testclient import TestClient @@ -1066,17 +1066,15 @@ class TestInteractiveEventsLifted: 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).""" + load and re-fetches on ``clear_ui``; the replay must not 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() + out = list(_interactive_events_replay(ws, ui, request)) 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): diff --git a/tests/test_session.py b/tests/test_session.py index 98e792ff..f0750bfd 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -3847,7 +3847,7 @@ class TestMetacognitiveBuffers: session._queue_user_advisory("correction", "watch out") msg = {"role": "user", "content": "noted"} session._attach_pending_user_reminders(msg) - # on_user_reminder called with the same shape as _build_history + # on_user_reminder called with the same shape project_history_messages # surfaces — list of {type, text} dicts. ``source`` rides as # a kwarg (None for non-wake correction nudges); inspect via # ``call_args.args`` for the positional reminders payload only. @@ -4275,7 +4275,7 @@ class TestApplyRemindersForProvider: """Defensive: a non-dict element in ``_reminders`` (corruption, partial state, future-shape rollback) must be silently skipped rather than aborting ``send`` via ``AttributeError`` on the - ``.get`` call. Mirrors the filter in ``_build_history``.""" + ``.get`` call. Mirrors the filter in ``project_history_messages``.""" session = _make_session() msg = { "role": "user", @@ -5193,7 +5193,7 @@ class TestReminderSidechannelIsolation: class TestSessionUIBaseUserReminderHook: """``on_user_reminder`` enqueues a ``user_reminder`` SSE event with - the same shape ``_build_history`` surfaces, so live tabs and + the same shape ``project_history_messages`` surfaces, so live tabs and reconnecting tabs render the same reminder payload.""" def test_on_user_reminder_enqueues_sse_event(self): diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py index 881fe1d1..75c6f7be 100644 --- a/tests/test_workstream_endpoints.py +++ b/tests/test_workstream_endpoints.py @@ -21,6 +21,10 @@ if TYPE_CHECKING: from starlette.responses import Response from turnstone.core.auth import AuthResult +from turnstone.core.history_decoration import ( + decorate_history_messages, + project_history_messages, +) from turnstone.core.session_routes import ( SessionEndpointConfig, make_detail_handler, @@ -993,21 +997,14 @@ class TestHistoryInteractive: class TestBuildHistoryReminderPropagation: - """``_build_history`` must surface the ``_reminders`` side-channel on - each entry so a tab reconnecting via ``/history`` renders the same - metacognitive nudge bubble the originating tab saw via the live - ``user_reminder`` SSE event. + """``project_history_messages`` must surface the ``_reminders`` + side-channel on each entry so a tab reconnecting via ``/history`` + renders the same metacognitive nudge bubble the originating tab saw + via the live ``user_reminder`` SSE event. """ - def _session_with_messages(self, messages: list[dict]) -> MagicMock: - session = MagicMock() - session.messages = messages - return session - def test_reminders_sidechannel_surfaces_on_entry(self): - from turnstone.server import _build_history - - session = self._session_with_messages( + history = project_history_messages( [ { "role": "user", @@ -1016,28 +1013,19 @@ class TestBuildHistoryReminderPropagation: } ] ) - history = _build_history(session) assert history[0]["content"] == "ah no" assert history[0]["reminders"] == [{"type": "correction", "text": "watch out"}] def test_no_reminders_key_when_sidechannel_absent(self): - from turnstone.server import _build_history - - session = self._session_with_messages([{"role": "user", "content": "just a message"}]) - history = _build_history(session) + history = project_history_messages([{"role": "user", "content": "just a message"}]) assert "reminders" not in history[0] def test_no_reminders_key_when_sidechannel_empty(self): - from turnstone.server import _build_history - - session = self._session_with_messages([{"role": "user", "content": "hi", "_reminders": []}]) - history = _build_history(session) + history = project_history_messages([{"role": "user", "content": "hi", "_reminders": []}]) assert "reminders" not in history[0] def test_multiple_reminders_preserved_in_order(self): - from turnstone.server import _build_history - - session = self._session_with_messages( + history = project_history_messages( [ { "role": "user", @@ -1049,16 +1037,13 @@ class TestBuildHistoryReminderPropagation: } ] ) - history = _build_history(session) assert history[0]["reminders"] == [ {"type": "denial", "text": "FIRST"}, {"type": "correction", "text": "SECOND"}, ] def test_reminders_coexist_with_attachments(self): - from turnstone.server import _build_history - - session = self._session_with_messages( + history = project_history_messages( [ { "role": "user", @@ -1070,17 +1055,14 @@ class TestBuildHistoryReminderPropagation: } ] ) - history = _build_history(session) assert history[0]["content"] == "look" assert history[0]["attachments"] == [{"kind": "image", "filename": "", "mime_type": ""}] assert history[0]["reminders"] == [{"type": "correction", "text": "watch"}] def test_malformed_reminders_filtered_out(self): - """Defensive: a non-dict element in the list (corruption / bug) - is dropped rather than crashing the history serialisation.""" - from turnstone.server import _build_history - - session = self._session_with_messages( + """Defensive: a non-dict element in the list (corruption / bug) is + dropped rather than crashing the history serialisation.""" + history = project_history_messages( [ { "role": "user", @@ -1093,7 +1075,6 @@ class TestBuildHistoryReminderPropagation: } ] ) - history = _build_history(session) # Non-dicts dropped; missing-text fills with empty string. assert history[0]["reminders"] == [ {"type": "correction", "text": "ok"}, @@ -1101,14 +1082,9 @@ class TestBuildHistoryReminderPropagation: ] def test_clean_message_passes_through_unchanged(self): - """No reminders, plain content — _build_history is a no-op for the + """No reminders, plain content — the projection is a no-op for the reminder field and ``content`` rides through verbatim.""" - from turnstone.server import _build_history - - session = self._session_with_messages( - [{"role": "user", "content": "just a normal message"}] - ) - history = _build_history(session) + history = project_history_messages([{"role": "user", "content": "just a normal message"}]) assert history[0]["content"] == "just a normal message" assert "reminders" not in history[0] @@ -1116,160 +1092,87 @@ class TestBuildHistoryReminderPropagation: """Assistant output may legitimately reference the tag (e.g. when the model is explaining the reminder system itself). No transformation should ever apply to assistant content.""" - from turnstone.server import _build_history - content = "Here is a tag in assistant output." - session = self._session_with_messages([{"role": "assistant", "content": content}]) - history = _build_history(session) + history = project_history_messages([{"role": "assistant", "content": content}]) assert history[0]["content"] == content class TestBuildHistoryAdvisoryRoundTrip: - """``_build_history`` must round-trip the persisted - ```` envelope (Seam 1 queued-message splice) to - cleaned content + a wire-shape ``advisories`` array. + """The ``/history`` projection must round-trip the persisted + ```` envelope (Seam 1 queued-message splice) to cleaned + content + a wire-shape ``advisories`` array. - Production realism note: ``session.messages`` never carries an - ``advisories`` key — only ``decorate_history_messages`` mutates - dicts to add it for the REST ``/history`` path, and the SSE replay - surface bypasses that decoration entirely. The earlier - ``TestBuildHistoryAdvisoryPropagation`` class pre-populated - ``advisories`` directly on the session messages, which tested a - passthrough that doesn't exist in production — the SSE replay code - path silently dropped queued messages despite the green tests. - These round-trip tests exercise the production shape (wrapped - envelope on the tool row's ``content``) so a regression in the - inline ``extract_advisories_from_tool_envelope`` call inside - ``_build_history`` surfaces here. + STRING-content envelopes are stripped by ``decorate_history_messages`` + (the first pipeline stage); LIST-content envelopes are stripped by + ``project_history_messages`` (the final stage, which also coerces list + content to a string). These tests drive the relevant stage(s) so a + regression in either surfaces here. """ - def _session_with_messages(self, messages: list[dict]) -> MagicMock: - session = MagicMock() - session.messages = messages - return session - - def test_build_history_round_trips_envelope_to_advisories(self): - """The production-realistic shape: a tool row whose ``content`` - is the wrapped ```` envelope (no ``advisories`` - key set — that's the bug-1 footprint). ``_build_history`` - must extract the advisory back out and ship it on the wire as - cleaned content + ``advisories``. - - Reverting the inline ``extract_advisories_from_tool_envelope`` - call in ``server._build_history``'s tool-message branch breaks - this test. - """ + def test_round_trips_string_envelope_to_advisories(self): + """A tool row whose ``content`` is a wrapped ```` + string envelope: ``decorate_history_messages`` strips it + surfaces + the advisory, then ``project_history_messages`` passes both through + to the wire shape.""" from turnstone.core.tool_advisory import UserInterjection, wrap_tool_result - from turnstone.server import _build_history wrapped = wrap_tool_result( "tool body", [UserInterjection(message="check logs", priority="notice")], ) - session = self._session_with_messages( - [ - { - "role": "tool", - "tool_call_id": "call_a", - "content": wrapped, - } - ] - ) - history = _build_history(session) + msgs: list[dict] = [{"role": "tool", "tool_call_id": "call_a", "content": wrapped}] + decorate_history_messages(msgs, {}, {}) + history = project_history_messages(msgs) # Cleaned content rides on the wire — envelope stripped. assert history[0]["content"] == "tool body" - # Advisory survives as a wire-shape entry the JS can render - # as a user bubble after the tool block. + # Advisory survives as a wire-shape entry the JS renders as a user + # bubble after the tool block. assert history[0]["advisories"] == [ {"type": "user_interjection", "text": "check logs", "priority": "notice"} ] - def test_build_history_round_trips_important_priority(self): - """The ``important`` priority preamble round-trips — pin both - the priority detection in the parser and the projection through - to the wire shape.""" + def test_round_trips_important_priority(self): + """The ``important`` priority preamble round-trips — pin both the + priority detection in the parser and the projection through to the + wire shape.""" from turnstone.core.tool_advisory import UserInterjection, wrap_tool_result - from turnstone.server import _build_history wrapped = wrap_tool_result( "out", [UserInterjection(message="urgent", priority="important")], ) - session = self._session_with_messages( - [{"role": "tool", "tool_call_id": "call_a", "content": wrapped}] - ) - history = _build_history(session) + msgs: list[dict] = [{"role": "tool", "tool_call_id": "call_a", "content": wrapped}] + decorate_history_messages(msgs, {}, {}) + history = project_history_messages(msgs) assert history[0]["content"] == "out" assert history[0]["advisories"] == [ {"type": "user_interjection", "text": "urgent", "priority": "important"} ] - def test_build_history_no_envelope_passes_through_unchanged(self): - """Plain tool content (no ```` prefix) — no - advisories field, content unchanged.""" - from turnstone.server import _build_history - - session = self._session_with_messages( - [{"role": "tool", "tool_call_id": "call_a", "content": "plain output"}] - ) - history = _build_history(session) + def test_no_envelope_passes_through_unchanged(self): + """Plain tool content (no ```` prefix) — no advisories + field, content unchanged.""" + msgs: list[dict] = [{"role": "tool", "tool_call_id": "call_a", "content": "plain output"}] + decorate_history_messages(msgs, {}, {}) + history = project_history_messages(msgs) assert history[0]["content"] == "plain output" assert "advisories" not in history[0] - def test_build_history_round_trip_through_full_decoration_chain(self): - """End-to-end pin: persist a wrapped envelope into ``messages``, - run the full decoration chain (``decorate_history_messages`` - followed by ``_build_history``), assert the wire shape carries - the advisory. This pins the contract every component in the - chain participates in — REST ``/history`` callers go through - ``decorate_history_messages``, and SSE replay goes through - ``_build_history`` — both must produce the same wire shape. - """ - from turnstone.core.history_decoration import decorate_history_messages - from turnstone.core.tool_advisory import UserInterjection, wrap_tool_result - from turnstone.server import _build_history + def test_extracts_advisories_from_list_content_text_part(self): + """List-typed tool output (image / structured MCP results) with a + Seam 1 splice carries the wrap envelope as a separate text part + (``session.py``'s tool-result loop appends + ``{"type": "text", "text": wrap_tool_result("", advisories)}`` when + ``output`` is a list). ``decorate_history_messages`` skips non- + string content, so ``project_history_messages`` owns this: it + extracts the advisory from the carrier part, drops it, and joins + the remaining text parts to a string (non-text parts like + image_url are dropped — the renderers consume a string). - wrapped = wrap_tool_result( - "raw", - [UserInterjection(message="hi", priority="notice")], - ) - # Decorate first — REST /history shape. - rest_messages: list[dict] = [{"role": "tool", "tool_call_id": "call_a", "content": wrapped}] - decorate_history_messages(rest_messages, {}, {}) - # And separately drive _build_history with a fresh undecorated - # message — SSE replay shape. - session = self._session_with_messages( - [{"role": "tool", "tool_call_id": "call_a", "content": wrapped}] - ) - sse_history = _build_history(session) - # Both surfaces produce the same advisory + cleaned content. - assert rest_messages[0]["content"] == "raw" - assert rest_messages[0]["advisories"] == [ - {"type": "user_interjection", "text": "hi", "priority": "notice"} - ] - assert sse_history[0]["content"] == "raw" - assert sse_history[0]["advisories"] == [ - {"type": "user_interjection", "text": "hi", "priority": "notice"} - ] - - def test_build_history_extracts_advisories_from_list_content_text_part(self): - """List-typed tool output (image / structured MCP results) - with a Seam 1 splice carries the wrap envelope as a separate - text part (``session.py``'s tool-result loop appends - ``{"type": "text", "text": wrap_tool_result("", advisories)}`` - when ``output`` is a list). ``_build_history`` must walk the - list parts, extract advisories from any wrap-envelope text - part, and DROP that text part from the projected list — the - cleaned inner content is empty by construction, and leaving - the part would cause the JS replay to render the literal - envelope text as a chunk inside the tool block AND fail to - render the queued message as a user bubble. - - Removing the list-content branch in ``_build_history``'s tool- - message advisory extraction breaks this test. + Removing the list-content branch in ``project_history_messages`` + breaks this test. """ from turnstone.core.tool_advisory import UserInterjection, wrap_tool_result - from turnstone.server import _build_history wrap_text = wrap_tool_result( "", @@ -1280,47 +1183,30 @@ class TestBuildHistoryAdvisoryRoundTrip: {"type": "image_url", "image_url": {"url": "data:image/png;base64,xxx"}}, {"type": "text", "text": wrap_text}, ] - session = self._session_with_messages( + history = project_history_messages( [{"role": "tool", "tool_call_id": "call_a", "content": list_content}] ) - history = _build_history(session) - # Wire-shape content keeps the original text + image parts but - # has the wrap text-part dropped. - wire_content = history[0]["content"] - assert isinstance(wire_content, list) - assert len(wire_content) == 2 - assert wire_content[0] == {"type": "text", "text": "the chart shows X"} - assert wire_content[1] == { - "type": "image_url", - "image_url": {"url": "data:image/png;base64,xxx"}, - } - # Advisory rides on the wire so JS replay renders the user - # bubble after the tool block — same contract as the string- - # content path. + # Content coerced to a string: text parts joined, carrier + non-text + # (image_url) parts dropped — matches the live renderer contract. + assert history[0]["content"] == "the chart shows X" + # Advisory rides on the wire so JS replay renders the user bubble + # after the tool block — same contract as the string-content path. assert history[0]["advisories"] == [ - { - "type": "user_interjection", - "text": "inspect histogram", - "priority": "notice", - } + {"type": "user_interjection", "text": "inspect histogram", "priority": "notice"} ] - def test_build_history_keeps_legitimate_envelope_text_part_with_body(self): - """A tool that legitimately produces output containing a - well-formed ```` envelope as a text part (e.g. - documentation viewer, code analyzer demoing the wrapper, an - echo tool) must NOT have that part dropped on replay. The - list-content drop heuristic must require both an empty cleaned - inner body AND at least one extracted advisory — the - signature of the injected ``wrap_tool_result("", advisories)`` - carrier. A legitimate tool envelope has non-empty inner body - OR no advisories, and stays in the projected list verbatim. - - Removing the ``not cleaned_text and advisories_from_part`` - guard breaks this test (the legitimate envelope gets dropped - from the wire content).""" - from turnstone.server import _build_history + def test_keeps_legitimate_envelope_text_part_with_body(self): + """A tool that legitimately produces output containing a well-formed + ```` envelope as a text part (e.g. documentation + viewer, code analyzer) must NOT have that part dropped on replay. + The drop heuristic requires both an empty cleaned inner body AND at + least one extracted advisory — the signature of the injected + ``wrap_tool_result("", advisories)`` carrier. A legitimate + envelope has a non-empty inner body OR no advisories, so it stays + in the joined content verbatim. + Removing the ``not cleaned_text and advisories_from_part`` guard + breaks this test (the legitimate envelope gets dropped).""" legit_envelope_text = ( "\nThis is what a tool_output envelope looks like.\n" ) @@ -1328,17 +1214,12 @@ class TestBuildHistoryAdvisoryRoundTrip: {"type": "text", "text": "doc preview:"}, {"type": "text", "text": legit_envelope_text}, ] - session = self._session_with_messages( + history = project_history_messages( [{"role": "tool", "tool_call_id": "call_a", "content": list_content}] ) - history = _build_history(session) - # All parts survive — none dropped. - wire_content = history[0]["content"] - assert isinstance(wire_content, list) - assert len(wire_content) == 2 - assert wire_content[1]["text"] == legit_envelope_text - # No advisories surfaced (no system-reminder blocks were - # extracted from the legitimate envelope). + # Both text parts survive (joined to a string) — none dropped. + assert history[0]["content"] == "doc preview:\n" + legit_envelope_text + # No advisories surfaced (no system-reminder blocks were extracted). assert "advisories" not in history[0] diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index 7a1f399b..d979e7d0 100644 --- a/turnstone/api/server_schemas.py +++ b/turnstone/api/server_schemas.py @@ -450,19 +450,21 @@ class WorkstreamHistoryResponse(BaseModel): """Response body for ``GET /v1/api/workstreams/{ws_id}/history``. Renamed and relocated from ``CoordinatorHistoryResponse`` in the - Stage 2 history/detail verb lift. Same OpenAI-like message-row - shape on both kinds; the lift adds the endpoint to interactive as - a feature gain (pre-lift interactive only exposed history through - the SSE replay on ``/events``). + Stage 2 history/detail verb lift. Same projected render shape on + both kinds; the lift adds the endpoint to interactive as a feature + gain (pre-lift interactive only exposed history through the SSE + replay on ``/events``). """ ws_id: str messages: list[dict[str, Any]] = Field( default_factory=list, description=( - "Tail of the workstream's reconstructed message history " - "(provider-fidelity OpenAI-like shape). Bounded by the " - "``limit`` query parameter (default 100, max 500)." + "Tail of the workstream's message history, projected to the " + "canonical render shape (flat tool_calls with verdict / " + "output_assessment, top-level source / reminders / " + "attachments, derived denied / is_error / pending). Bounded " + "by the ``limit`` query parameter (default 100, max 500)." ), ) diff --git a/turnstone/core/history_decoration.py b/turnstone/core/history_decoration.py index d4ea4904..dc0a5324 100644 --- a/turnstone/core/history_decoration.py +++ b/turnstone/core/history_decoration.py @@ -1,16 +1,20 @@ -"""Shared history-replay decoration helpers. +"""Shared history-replay projection + decoration helpers. -Both surfaces that build a history wire payload — interactive's SSE -``_build_history`` and the lifted ``make_history_handler`` REST -endpoint — need the same audit-trail data attached to each -``tool_calls`` entry: the persisted intent verdict (``intent_verdicts`` -table) and the output-guard assessment (``output_assessments`` table). +The REST ``make_history_handler`` (``GET /history``) endpoint is the +single surface that builds the history wire payload for both kinds. Its +pipeline composes the helpers in this module: load the audit-trail +indexes, :func:`decorate_history_messages` (attach the persisted intent +verdict from ``intent_verdicts`` + output-guard assessment from +``output_assessments`` to each ``tool_calls`` entry, strip string +```` envelopes), :func:`extract_reasoning_for_history` +(surface stored reasoning), then :func:`project_history_messages` (the +final structural projection both UIs — interactive ``replayHistory`` +and the coordinator dashboard — consume directly). -Centralising the lookup + decoration here keeps the two surfaces from -drifting on which fields ship to the client and how they're shaped. -The shared helpers also let us project only the fields the UI actually -renders, dropping redundant ones (``call_id``/``func_name`` already -carried on ``tc.id``/``tc.name``) so the wire payload stays tight. +Centralising the projection here is what keeps the wire shape single- +sourced: the helpers project only the fields the UI renders, dropping +redundant ones (``call_id``/``func_name`` already carried on +``tc.id``/``tc.name``) so the wire payload stays tight. All functions are pure I/O or pure transforms — safe to call from either an async caller (via ``asyncio.to_thread``) or a sync hook. @@ -27,6 +31,7 @@ from turnstone.core.tool_advisory import ( _USER_INTERJECTION_IMPORTANT_PREAMBLE, _USER_INTERJECTION_NOTICE_PREAMBLE, ) +from turnstone.core.watch import WATCH_REMINDER_OPTIONAL_KEYS log = get_logger(__name__) @@ -140,14 +145,12 @@ def decorate_tool_call( ) -> None: """Mutate ``tc`` in place, attaching ``verdict`` / ``output_assessment``. - Works on either tool_call shape: - - OpenAI format (``{id, function: {name, arguments}}``) — used by - ``/history`` REST. - - Flattened format (``{id, name, arguments}``) — used by SSE replay. - - Both carry ``id`` at the top level, which is the only field this - helper reads. No-ops cleanly when the call_id has no matching - row (unflagged tools stay clean). + Reads only ``id`` (top-level on every tool_call shape), so it works on + either the OpenAI-nested ``{id, function: {name, arguments}}`` shape — + what ``decorate_history_messages`` passes from the REST ``/history`` + pipeline — or a flattened ``{id, name, arguments}`` shape. No-ops + cleanly when the call_id has no matching row (unflagged tools stay + clean). """ call_id = tc.get("id", "") or "" if not call_id: @@ -362,10 +365,10 @@ def extract_reasoning_text_from_provider_content(provider_content: Any) -> str: drop the reasoning under an index-only check. Same robustness point for Anthropic's hypothetical mixed-order outputs. - Pure transform — safe from any thread. Both history surfaces - (interactive ``_build_history`` and lifted ``make_history_handler``) - call this directly. See ``_BLOCK_TYPE_PROVIDER_FACTORY`` above - for the recognised block types and the providers that own them. + Pure transform — safe from any thread. The REST ``/history`` + reasoning surfacing (:func:`extract_reasoning_for_history`) calls + this directly. See ``_BLOCK_TYPE_PROVIDER_FACTORY`` above for the + recognised block types and the providers that own them. """ if not isinstance(provider_content, list) or not provider_content: return "" @@ -397,12 +400,11 @@ def extract_reasoning_for_history( dispatcher returned non-empty text. Strips ``_provider_content`` unconditionally — the field is internal and never read by either UI. - The interactive ``_build_history`` surface DOES NOT call this - helper; it builds new entry dicts from scratch and calls - :func:`extract_reasoning_text_from_provider_content` directly per - assistant message, stamping ``entry["reasoning"]`` inline. The two - surfaces converge on the same dispatcher; only the mutation shape - differs. + Runs before :func:`project_history_messages` in the ``/history`` + pipeline: this helper stamps ``msg["reasoning"]`` (and strips + ``_provider_content``), then the projection passes that ``reasoning`` + field through to the wire payload — the projection never re-reads + ``_provider_content`` (it is gone by then). Pure transform. Safe to call from ``asyncio.to_thread``. """ @@ -514,3 +516,286 @@ def decorate_history_messages( msg["content"] = cleaned if advisories: msg["advisories"] = advisories + + +def project_history_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Project decorated storage messages into the canonical wire shape. + + This is the SINGLE server-side projection that both the interactive + ``replayHistory`` renderer and the coordinator dashboard's history + rebuild consume. It runs LAST in the ``make_history_handler`` + pipeline — after :func:`decorate_history_messages` (verdict / + output_assessment + STRING ```` advisory stripping) and + :func:`extract_reasoning_for_history` (``reasoning`` stamping + + ``_provider_content`` strip) — and reshapes the provider-native + ``reconstruct_messages`` storage shape into the flat render shape: + + - multipart user ``content`` → plain string + derived ``attachments`` + (the ``_attachments_meta`` side-channel wins when present); + - ``_source`` → ``source``; ``_reminders`` → filtered ``reminders``; + - nested ``tool_calls[].function.{name,arguments}`` → flat + ``{id, name, arguments}`` carrying the decoration (``verdict`` / + ``output_assessment``) already placed on the call by + :func:`decorate_tool_call`; + - ``reasoning`` passes through (already stamped upstream — this + projection NEVER reads ``_provider_content``, which is gone by now); + - tool results: surface ``advisories``, coerce list content to a + string, derive ``denied`` / ``is_error`` from the content prefix; + - ``denied`` propagates from a tool result to its parent assistant + turn; ``pending`` marks ONLY the last assistant tool-call turn that + still has an orphan (a tool_call with no result in the window). + + Two projection responsibilities live HERE and nowhere else: + + * **List-content advisory extraction.** A queued ``UserInterjection`` + spliced into a LIST-typed tool result (image / structured MCP + output) rides as an appended ``wrap_tool_result("", advisories)`` + carrier part. :func:`decorate_history_messages` only handles + STRING content, so the carrier survives to here; this projection + extracts the advisories and drops the carrier part. STRING + envelopes are already stripped upstream, so their ``advisories`` + pass through untouched — we never double-extract. + * **List → string coercion** of tool content: the renderers require a + string (``replayHistory`` calls ``stripAnsi(content).trim()``; + coord joins text parts), so a LIST tool ``content`` is reduced to + its joined text parts here. + + Returns a NEW list of NEW entry dicts (strict 1:1 with *messages*) — + never mutates the input. Pure transform; safe from any thread. + """ + # Pre-scan: which tool_call_ids have a result message? An assistant + # tool_call with no result is an orphan; only the LAST such turn is + # marked "pending" (see the reversed single-turn marking below) — + # a mid-conversation orphan (cancelled / interrupted) must still + # render its tool block, so it is deliberately left unmarked. + resulted_call_ids: set[str] = set() + for msg in messages: + if msg.get("role") == "tool": + cid = msg.get("tool_call_id") + if cid: + resulted_call_ids.add(str(cid)) + + history: list[dict[str, Any]] = [] + for msg in messages: + role = msg.get("role") + content = msg.get("content") + attachments_meta: list[dict[str, Any]] = [] + + # (1) Collapse multipart user content (text + image_url / document + # parts) to a plain string + a derived attachment list. + if role == "user" and isinstance(content, list): + text_parts: list[str] = [] + for part in content: + if not isinstance(part, dict): + continue + ptype = part.get("type") + if ptype == "text": + text_parts.append(str(part.get("text", ""))) + elif ptype == "image_url": + attachments_meta.append({"kind": "image", "filename": "", "mime_type": ""}) + elif ptype == "document": + d = part.get("document", {}) + attachments_meta.append( + { + "kind": "text", + "filename": str(d.get("name", "")), + "mime_type": str(d.get("media_type", "")), + } + ) + content = "\n".join(text_parts) + + # (2) The authoritative ``_attachments_meta`` side-channel wins + # when present (carries image filenames the image_url part + # itself can't express). + side_meta = msg.get("_attachments_meta") + if isinstance(side_meta, list) and side_meta: + attachments_meta = [ + { + "kind": str(m.get("kind") or ""), + "filename": str(m.get("filename") or ""), + "mime_type": str(m.get("mime_type") or ""), + } + for m in side_meta + if isinstance(m, dict) + ] + + entry: dict[str, Any] = {"role": role, "content": content} + if attachments_meta: + entry["attachments"] = attachments_meta + + # (3) ``_source`` side-channel → top-level ``source`` (drives the + # ``.msg.user.system-nudge`` marker on replay). + if msg.get("_source"): + entry["source"] = str(msg["_source"]) + + # (4) ``_reminders`` side-channel → top-level ``reminders``, + # filtered + key-projected. Filter first so an all-malformed + # list doesn't set the field to ``[]`` (absent vs empty mean + # the same on the wire); project on a known key set to narrow + # the blast radius if a producer stuffs extra fields. + reminders = msg.get("_reminders") + if isinstance(reminders, list): + clean_reminders: list[dict[str, Any]] = [] + for r in reminders: + if not isinstance(r, dict): + continue + rtype = str(r.get("type") or "") + rtext = str(r.get("text") or "") + if not rtype and not rtext: + continue + clean: dict[str, Any] = {"type": rtype, "text": rtext} + for opt_key in WATCH_REMINDER_OPTIONAL_KEYS: + if opt_key in r: + clean[opt_key] = r[opt_key] + clean_reminders.append(clean) + if clean_reminders: + entry["reminders"] = clean_reminders + + # (5) Reasoning is already stamped by ``extract_reasoning_for_history`` + # (gated on the active model's surface_persisted_reasoning flag) — + # pass it through. This projection never re-extracts from + # ``_provider_content`` (already stripped upstream). + if msg.get("reasoning"): + entry["reasoning"] = msg["reasoning"] + + # (6) Flatten OpenAI-nested tool_calls ``{id, function:{name, + # arguments}}`` → ``{id, name, arguments}`` that the renderers + # read, carrying the decoration (``verdict`` / + # ``output_assessment``) ``decorate_tool_call`` placed on the + # nested call. + tool_calls = msg.get("tool_calls") + if tool_calls: + tc_entries: list[dict[str, Any]] = [] + for tc in tool_calls: + if not isinstance(tc, dict): + continue + fn = tc.get("function") or {} + arguments = fn.get("arguments") + if arguments is None: + arguments = tc.get("arguments") + if arguments is None: + arguments = "" + tc_entry: dict[str, Any] = { + "id": tc.get("id", "") or "", + "name": fn.get("name") or tc.get("name") or "", + "arguments": arguments, + } + if tc.get("verdict"): + tc_entry["verdict"] = tc["verdict"] + if tc.get("output_assessment"): + tc_entry["output_assessment"] = tc["output_assessment"] + tc_entries.append(tc_entry) + entry["tool_calls"] = tc_entries + + # (7) Tool results: carry ``tool_call_id`` + ``advisories``, coerce + # list content to text (extracting list-envelope advisories + # along the way), and derive ``denied`` / ``is_error`` from the + # content prefix — the storage shape pre-sets none of these. + if role == "tool": + result_call_id = msg.get("tool_call_id") + if result_call_id: + entry["tool_call_id"] = str(result_call_id) + # STRING-content advisories were already surfaced by + # ``decorate_history_messages`` (on ``msg["advisories"]``); + # pass them through. LIST-content envelopes are extracted + # here (decorate skips non-string content). + existing_advisories = msg.get("advisories") + extracted_advisories: list[dict[str, str]] = [] + if isinstance(content, list): + # The Seam 1 splice rides as an appended + # ``wrap_tool_result("", advisories)`` carrier part — the + # inner cleaned content is empty by construction. Drop the + # carrier part ONLY when the parser accepted the envelope + # AND the cleaned inner is empty AND at least one advisory + # survived; a tool legitimately emitting an envelope with a + # non-empty body is left in place. + kept_text: list[str] = [] + for part in content: + text = ( + part.get("text") + if isinstance(part, dict) and part.get("type") == "text" + else None + ) + if isinstance(text, str) and text.startswith("\n"): + try: + extracted = extract_advisories_from_tool_envelope(text) + except Exception: + extracted = None + if extracted is not None: + cleaned_text, advisories_from_part = extracted + if not cleaned_text and advisories_from_part: + extracted_advisories.extend(advisories_from_part) + continue + if isinstance(text, str): + kept_text.append(text) + content = "\n".join(kept_text) + entry["content"] = content + if isinstance(content, str): + if content.startswith("Denied by user") or content.startswith("Blocked"): + entry["denied"] = True + # Persisted flag wins; fall back to the text heuristic for + # historical data that predates ``is_error``. + if ( + msg.get("is_error") + or content.startswith("Error") + or content.startswith("Command timed out") + or content.startswith("Search timed out") + or content.startswith("Unknown tool:") + or content.startswith("JSON parse error:") + or content.startswith("MCP prompt timed out") + or content.startswith("MCP prompt error") + ): + entry["is_error"] = True + # Surface advisories on the wire (string: passed through from + # decorate; list: extracted just above). Project on a known + # key set, mirroring the ``reminders`` filter. + advisories_src: list[dict[str, Any]] = ( + extracted_advisories + if extracted_advisories + else (existing_advisories if isinstance(existing_advisories, list) else []) + ) + if advisories_src: + clean_advisories: list[dict[str, Any]] = [] + for a in advisories_src: + if not isinstance(a, dict): + continue + atype = str(a.get("type") or "") + atext = str(a.get("text") or "") + if not atype or not atext: + continue + clean_advisories.append( + { + "type": atype, + "text": atext, + "priority": str(a.get("priority") or "notice"), + } + ) + if clean_advisories: + entry["advisories"] = clean_advisories + + history.append(entry) + + # (8) Propagate denial from a tool result to its parent assistant turn + # so the tool block renders the denied (not approved) badge. + last_assistant_idx: int | None = None + for idx, entry in enumerate(history): + if entry.get("tool_calls"): + last_assistant_idx = idx + elif entry.get("role") == "tool" and entry.get("denied") and last_assistant_idx is not None: + history[last_assistant_idx]["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". 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 entry in reversed(history): + tcs = entry.get("tool_calls") + if tcs: + has_orphan = any(tc.get("id") and str(tc["id"]) not in resulted_call_ids for tc in tcs) + if has_orphan: + entry["pending"] = True + break + + return history diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 0ca86e74..aed237dc 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -2788,7 +2788,8 @@ class ChatSession: # Defensive filter — only dict entries are valid; a string / # None / other shape from corruption or partial state must # not abort the whole send via an AttributeError on .get(). - # Mirrors the same filter ``_build_history`` applies on the + # Mirrors the same filter ``project_history_messages`` + # (turnstone.core.history_decoration) applies on the # wire-out side. reminders = [r for r in raw_reminders if isinstance(r, dict)] if not reminders: @@ -2878,8 +2879,8 @@ class ChatSession: skip messages with this flag, so the model sees each reminder exactly once (the turn it advised). The flag is a sibling key like ``_reminders`` itself; ``sanitize_messages`` strips both - before the wire and ``_build_history`` ignores the delivered - flag entirely so UI replay parity is preserved across + before the wire and ``project_history_messages`` ignores the + delivered flag entirely so UI replay parity is preserved across reconnects. """ for msg in self.messages: diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 7d390564..d59856d6 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -2489,6 +2489,7 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: decorate_history_messages, extract_reasoning_for_history, load_verdict_indexes, + project_history_messages, ) indexes = await asyncio.to_thread(load_verdict_indexes, ws_id) @@ -2552,6 +2553,17 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: await asyncio.to_thread( extract_reasoning_for_history, messages, surface_persisted_reasoning ) + # Final structural projection: flatten nested tool_calls, + # collapse multipart content, surface the + # ``_source`` / ``_reminders`` / ``_attachments_meta`` + # side-channels top-level, and derive + # ``denied`` / ``is_error`` / ``pending``. Runs last (reads + # decorate's in-place verdict/advisory mutations + the + # stamped ``reasoning``) and returns a fresh list, so the + # wire payload is the canonical render shape both the + # interactive ``replayHistory`` and the coordinator history + # rebuild consume directly — no client-side normaliser. + messages = await asyncio.to_thread(project_history_messages, messages) except Exception: # Operationally interesting: a persistent decoration # failure (missing migration, driver mismatch, schema diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index 5d602370..d0d868d5 100644 --- a/turnstone/core/session_ui_base.py +++ b/turnstone/core/session_ui_base.py @@ -1964,8 +1964,8 @@ class SessionUIBase: SSE consumer (other browser tabs, CLI mirrors, future channel adapters) render the reminder bubble in lockstep with the originating tab. The history-replay path surfaces the same - shape via ``_build_history`` so a tab reconnecting later - renders the same bubble. + shape via ``project_history_messages`` (the REST ``/history`` + projection) so a tab reconnecting later renders the same bubble. ``source`` mirrors the user-message dict's ``_source`` field (today only ``"system_nudge"`` for wake-driven reminders). The diff --git a/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py b/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py index da6fee1a..f06fdb51 100644 --- a/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py +++ b/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py @@ -7,8 +7,8 @@ Adds two boolean (integer-coded) operator knobs: ``provider_data`` and surfaces it on each assistant message dict so a page refresh re-renders the reasoning bubble. **Storage of the reasoning bytes happens regardless of this flag** — it only controls - the extract-and-include step in ``_build_history`` / - ``decorate_history_messages``. The pre-rename column was + the extract-and-include step in ``extract_reasoning_for_history`` + (the REST ``/history`` reasoning surfacing). The pre-rename column was ``persist_reasoning``; the rename to ``surface_persisted_reasoning`` happened in the review-fix wave because the original name implied a storage-control switch when the flag is purely about UI rehydration. diff --git a/turnstone/server.py b/turnstone/server.py index a5c4eff7..b69aadcf 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -54,18 +54,6 @@ from turnstone.core.auth import ( _DenyFilter, jwt_version_slot, ) -from turnstone.core.history_decoration import ( - decorate_tool_call as _decorate_tool_call, -) -from turnstone.core.history_decoration import ( - extract_advisories_from_tool_envelope, -) -from turnstone.core.history_decoration import ( - extract_reasoning_text_from_provider_content as _extract_reasoning_text, -) -from turnstone.core.history_decoration import ( - load_verdict_indexes as _load_verdict_indexes, -) from turnstone.core.log import get_logger from turnstone.core.metrics import metrics as _metrics from turnstone.core.ratelimit import resolve_client_ip @@ -97,7 +85,6 @@ from turnstone.core.session_ui_base import ( fire_judge_verdict_metric, ) from turnstone.core.tools import TOOLS # noqa: F401 — available for introspection -from turnstone.core.watch import WATCH_REMINDER_OPTIONAL_KEYS from turnstone.core.web_helpers import version_html as _version_html from turnstone.core.workstream import ( Workstream, @@ -417,345 +404,6 @@ class WebUI(SessionUIBase): # bookkeeping. -# --------------------------------------------------------------------------- -# History builder -# --------------------------------------------------------------------------- - - -# Verdict + output-assessment decoration helpers (``_decorate_tool_call``, -# ``_load_verdict_indexes``) are imported at module top alongside the -# rest of ``turnstone.core.*``. Both this builder and -# :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( - session: ChatSession, - has_pending_approval: bool = False, - *, - verdicts: dict[str, dict[str, Any]] | None = None, - assessments: dict[str, dict[str, Any]] | None = None, -) -> list[dict[str, Any]]: - """Build a history replay list from ChatSession messages. - - When ``has_pending_approval`` is True, the last assistant entry's - tool_calls are marked ``"pending": True`` so the client renders them - as awaiting approval rather than as already-approved. - - Tool results whose content starts with "Denied by user" are marked - ``"denied": True``, and the corresponding assistant entry that - issued the tool calls is also marked ``"denied": True`` so the - client can render the correct badge. - - ``verdicts`` and ``assessments`` are optional pre-loaded - ``{call_id → row}`` dicts (see :func:`_load_verdict_indexes`). - Async callers should pre-load via ``asyncio.to_thread`` and pass - them in to avoid blocking the event loop on storage I/O. When - omitted, the storage call runs inline (sync call sites). - """ - # Metacognitive nudges live on the message dict's ``_reminders`` - # side-channel — user messages carry user-channel nudges - # (correction / denial / resume / start / completion), tool - # messages carry tool-channel nudges (tool_error / repeat). Both - # are surfaced separately on each entry so the UI can render them - # as their own bubble (live via ``user_reminder`` / - # ``tool_reminder`` SSE events; replay via this propagation). - # ``content`` never carries the ```` envelope — - # that splice is transient, applied to a wire-bound copy in - # ``ChatSession._apply_reminders_for_provider``. - # - # Verdict + output-assessment lookup tables — populated either - # 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 - else: - ws_id = getattr(session, "_ws_id", "") or "" - verdicts_by_call_id, assessments_by_call_id = _load_verdict_indexes(ws_id) - # Active-model ``surface_persisted_reasoning`` flag — single-tier - # resolution (live session's registry only). This path always has - # a live ``ChatSession`` in hand, so the cold-workstream and - # app.state-registry tiers used by ``make_history_handler`` - # (``session_routes.py:2396-2429``) are unreachable here. Default - # True mirrors the migration's server_default and matches the - # conservative rehydration default in the Phase 1 spec. - surface_persisted_reasoning = True - registry = getattr(session, "_registry", None) - model_alias = getattr(session, "_model_alias", "") or "" - if registry is not None and model_alias: - try: - surface_persisted_reasoning = bool( - registry.get_config(model_alias).surface_persisted_reasoning - ) - except Exception: - # Unknown alias / partially-built registry / dataclass drift — - # fall back to the conservative default rather than failing - # the entire history build. - surface_persisted_reasoning = True - history = [] - for msg in session.messages: - content = msg.get("content") - attachments_meta: list[dict[str, Any]] = [] - # User messages with attachments carry list content (text + - # image_url / document parts). The UI wants a plain-text bubble - # plus a derived pill cluster — split the list content here so - # the client never has to interpret provider-shaped parts. - if msg.get("role") == "user" and isinstance(content, list): - text_parts: list[str] = [] - for part in content: - if not isinstance(part, dict): - continue - ptype = part.get("type") - if ptype == "text": - text_parts.append(str(part.get("text", ""))) - elif ptype == "image_url": - attachments_meta.append({"kind": "image", "filename": "", "mime_type": ""}) - elif ptype == "document": - d = part.get("document", {}) - attachments_meta.append( - { - "kind": "text", - "filename": str(d.get("name", "")), - "mime_type": str(d.get("media_type", "")), - } - ) - content = "\n".join(text_parts) - # Prefer the authoritative side-channel (set by - # reconstruct_messages on history replay) — it carries image - # filenames that the image_url part itself can't express. - side_meta = msg.get("_attachments_meta") - if isinstance(side_meta, list) and side_meta: - attachments_meta = [ - { - "kind": str(m.get("kind") or ""), - "filename": str(m.get("filename") or ""), - "mime_type": str(m.get("mime_type") or ""), - } - for m in side_meta - if isinstance(m, dict) - ] - entry = {"role": msg["role"], "content": content} - if attachments_meta: - entry["attachments"] = attachments_meta - # Surface the ``_source`` side-channel so the frontend can apply - # the ``.msg.user.system-nudge`` class on history replay (today - # only the wake-driven empty user turn carries ``"system_nudge"``). - # Persisted via the conversations._source column added in - # migration 050; legacy rows without the column lack the key - # entirely. - if msg.get("_source"): - entry["source"] = str(msg["_source"]) - # Surface the ``_reminders`` side-channel so a tab reconnecting - # via /history renders the same metacognitive nudge bubble the - # originating tab saw live (user-channel reminders via - # ``user_reminder`` SSE; tool-channel via ``tool_reminder``). - # Persisted via the conversations._reminders column added in - # migration 050 — multi-tab / multi-device tabs reconnecting - # later see the same shape now, not just the originating tab. - reminders = msg.get("_reminders") - if isinstance(reminders, list): - # Filter first so an all-malformed _reminders doesn't set the - # field to []; absent vs. empty-list should mean the same - # thing on the wire. Project on a known set of keys — - # narrows the blast radius if a future producer accidentally - # stuffs sensitive fields into the dict. ``watch_triggered`` - # carries the structured watch-card fields (watch_name, - # command, poll_count, max_polls, is_final) and other - # producers leave them unset. - clean_reminders: list[dict[str, Any]] = [] - for r in reminders: - if not isinstance(r, dict): - continue - rtype = str(r.get("type") or "") - rtext = str(r.get("text") or "") - if not rtype and not rtext: - continue - clean: dict[str, Any] = {"type": rtype, "text": rtext} - for opt_key in WATCH_REMINDER_OPTIONAL_KEYS: - if opt_key in r: - clean[opt_key] = r[opt_key] - clean_reminders.append(clean) - if clean_reminders: - entry["reminders"] = clean_reminders - # Surface stored reasoning text on assistant messages for UI - # rehydration (page-refresh path). Sourced from the in-memory - # ``_provider_content`` lane on ``session.messages`` (set - # post-commit at ``session.py:3768-3771``). The lane itself is - # never copied into ``entry`` — the wire payload stays tight. - if msg.get("role") == "assistant" and surface_persisted_reasoning: - reasoning_text = _extract_reasoning_text(msg.get("_provider_content")) - if reasoning_text: - entry["reasoning"] = reasoning_text - if msg.get("tool_calls"): - tc_entries: list[dict[str, Any]] = [] - for tc in msg["tool_calls"]: - tc_entry: dict[str, Any] = { - "id": tc.get("id", "") or "", - "name": tc["function"]["name"], - "arguments": tc["function"].get("arguments", ""), - } - # Decorate with persisted verdict + output_assessment - # via the shared helper (also used by - # ``make_history_handler``). Skips unflagged - # ("risk_level == 'none'") rows so the wire stays - # tight; ships only the fields the UI renders. - _decorate_tool_call( - tc_entry, - verdicts_by_call_id, - assessments_by_call_id, - ) - tc_entries.append(tc_entry) - entry["tool_calls"] = tc_entries - # Detect denied/blocked/errored tool results by their content prefix. - if msg.get("role") == "tool": - content = msg.get("content", "") - # Propagate tool_call_id so replayHistory can anchor the - # rendered output to the specific .ts-approval-tool element - # by data-call-id (mirrors the live appendToolOutput path). - # Without this, multi-tool batches render every result at - # the bottom of the block rather than under each header. - result_call_id = msg.get("tool_call_id") - if result_call_id: - entry["tool_call_id"] = str(result_call_id) - # Extract advisories from a wrapped ```` envelope - # (Seam 1 queued-message splice). ``session.messages`` never - # carries an ``advisories`` key on its own — only - # ``decorate_history_messages`` mutates dicts to add it for - # the REST ``/history`` path, and the SSE replay surface - # bypasses that decoration entirely. Calling the same - # idempotent extraction helper here pins both surfaces to - # the same wire shape: cleaned content + extracted advisories - # ride as a user bubble after the tool block. No-ops cleanly - # for plain (unwrapped) content. - extracted_advisories: list[dict[str, str]] = [] - if isinstance(content, str): - try: - extracted = extract_advisories_from_tool_envelope(content) - except Exception: - extracted = None - if extracted is not None: - cleaned, extracted_advisories = extracted - content = cleaned - entry["content"] = cleaned - elif isinstance(content, list): - # List-typed tool output (image / structured MCP - # results) carries any Seam 1 splice as an appended - # text part produced by ``wrap_tool_result("", ...)`` - # — the inner cleaned content is empty by construction, - # so the part exists only to carry advisories. Walk - # the parts, extract advisories from any wrap - # envelope, and drop those parts from the projected - # list. Without this, the JS replay would render the - # raw envelope as a text bubble inside the tool block - # AND fail to render the queued message as a user - # bubble (no ``advisories`` array). - new_parts: list[Any] = [] - changed = False - for part in content: - text = ( - part.get("text") - if isinstance(part, dict) and part.get("type") == "text" - else None - ) - if isinstance(text, str) and text.startswith("\n"): - try: - extracted = extract_advisories_from_tool_envelope(text) - except Exception: - extracted = None - # Drop the part ONLY when both (a) the parser - # accepted the envelope structure AND (b) the - # cleaned inner content is empty AND (c) at - # least one advisory came out. This is the - # signature of an injected - # ``wrap_tool_result("", advisories)`` carrier - # — the inner is empty by construction. A - # tool legitimately emitting an envelope with - # non-empty body or no advisory blocks is left - # in place rather than silently dropped. - if extracted is not None: - cleaned_text, advisories_from_part = extracted - if not cleaned_text and advisories_from_part: - extracted_advisories.extend(advisories_from_part) - changed = True - continue - new_parts.append(part) - if changed: - content = new_parts - entry["content"] = new_parts - if isinstance(content, str): - if content.startswith("Denied by user") or content.startswith("Blocked"): - entry["denied"] = True - # Use persisted flag if available, fall back to text - # heuristic for historical data that predates is_error. - if ( - msg.get("is_error") - or content.startswith("Error") - or content.startswith("Command timed out") - or content.startswith("Search timed out") - or content.startswith("Unknown tool:") - or content.startswith("JSON parse error:") - or content.startswith("MCP prompt timed out") - or content.startswith("MCP prompt error") - ): - entry["is_error"] = True - # Surface advisories on the wire so the JS replay can render - # them as user bubbles after the tool block. Project on a - # known set of keys — narrows the blast radius if a future - # producer stuffs sensitive fields into the dict. Mirrors - # the ``reminders`` filter above for the same reason. The - # extracted-from-envelope list is the production-realistic - # source: ``decorate_history_messages`` populates the - # ``advisories`` key only on the REST ``/history`` path, but - # ``session.messages`` (the SSE-replay source) never has it. - if extracted_advisories: - clean_advisories: list[dict[str, Any]] = [] - for a in extracted_advisories: - if not isinstance(a, dict): - continue - atype = str(a.get("type") or "") - atext = str(a.get("text") or "") - if not atype or not atext: - continue - clean_advisories.append( - { - "type": atype, - "text": atext, - "priority": str(a.get("priority") or "notice"), - } - ) - if clean_advisories: - entry["advisories"] = clean_advisories - history.append(entry) - - # Propagate denial from tool results to their parent assistant entry. - last_assistant_idx: int | None = None - for idx, entry in enumerate(history): - if entry.get("tool_calls"): - last_assistant_idx = idx - elif entry.get("role") == "tool" and entry.get("denied") and last_assistant_idx is not None: - history[last_assistant_idx]["denied"] = True - - # Mark last assistant tool call as pending if approval is outstanding. - if has_pending_approval: - for entry in reversed(history): - if entry.get("tool_calls"): - entry["pending"] = True - break - return history - - # --------------------------------------------------------------------------- # Pure ASGI middleware (NOT BaseHTTPMiddleware — that breaks SSE streaming) # --------------------------------------------------------------------------- diff --git a/turnstone/shared_static/history_normalize.js b/turnstone/shared_static/history_normalize.js deleted file mode 100644 index 1eb5f450..00000000 --- a/turnstone/shared_static/history_normalize.js +++ /dev/null @@ -1,235 +0,0 @@ -// 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 -// `` 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 -