ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-28 06:44:51 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor(history): project the /history wire shape server-side

Browse Source 
Collapse the three hand-synced "raw storage -> render shape" projections
into one server-side projection. The projection previously lived in a
test-only `_build_history` (SSE-era reference impl), a client-side JS
normaliser (`history_normalize.js`, the transitional bridge), and coord's
inline `init()` handling -- drifting silently with no parity test.

Add `project_history_messages` to `history_decoration.py` and run it as the
final step of the `make_history_handler` pipeline (load_messages -> decorate
-> extract_reasoning -> project), so `GET /history` emits the canonical
render shape directly: flat tool_calls (with verdict / output_assessment),
top-level source / reminders / attachments, collapsed multipart content,
derived denied / is_error / pending, reasoning, and advisories. Interactive
`replayHistory` now consumes the payload verbatim.

Close two gaps the JS bridge deferred:
- list-content <tool_output> advisory extraction (decorate handles only
  string content; the projection extracts list-carrier advisories, then
  joins remaining text parts to the string the renderers require);
- orphan->pending marks ONLY the last orphan tool-call turn, so a
  mid-conversation cancelled tool still renders instead of vanishing.

Delete `history_normalize.js` (+ its <script> tag and node test) and the
test-only `_build_history` (+ orphaned imports); retarget its direct tests
onto the projection helpers. Update the WorkstreamHistoryResponse
description and the Web UI Resilience architecture note to the projected
shape.

Coord's `init()` still reads the raw side-channels; migrating it to the
projected shape is the next commit, browser-verified separately.

Refs #549.
This commit is contained in:
Patrick Buckley
2026-05-28 16:50:12 -07:00
parent 91ffbf8692
commit ee8dc7c1c3
19 changed files with 757 additions and 1272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture.md
+6 -5
View File
@@ -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
tests/test_app_js.py
+12 -113
View File
@@ -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
tests/test_build_history_reminders.py
-266
View File
@@ -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"
tests/test_history_decoration.py
+5 -5
View File
@@ -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
tests/test_history_projection.py
+291
View File
@@ -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
tests/test_reasoning_audit_log_discipline.py
+2 -34
View File
@@ -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
tests/test_server_authz.py
+5 -7
View File
@@ -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):
tests/test_session.py
+3 -3
View File
@@ -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):
tests/test_workstream_endpoints.py
+84 -203
View File
@@ -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 <system-reminder> 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
``<tool_output>`` envelope (Seam 1 queued-message splice) to
cleaned content + a wire-shape ``advisories`` array.
"""The ``/history`` projection must round-trip the persisted
``<tool_output>`` 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 ``<tool_output>`` 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 ``<tool_output>``
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 ``<tool_output>`` 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 ``<tool_output>`` 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 ``<tool_output>`` 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
``<tool_output>`` 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 = (
"<tool_output>\nThis is what a tool_output envelope looks like.\n</tool_output>"
)
@@ -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]
turnstone/api/server_schemas.py
+9 -7
View File
@@ -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)."
),
)
turnstone/core/history_decoration.py
+314 -29
View File
@@ -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
``<tool_output>`` 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 ``<tool_output>`` 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("<tool_output>\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
turnstone/core/session.py
+4 -3
View File
@@ -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:
turnstone/core/session_routes.py
+12
View File
@@ -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
turnstone/core/session_ui_base.py
+2 -2
View File
@@ -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
turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py
+2 -2
View File
@@ -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.
turnstone/server.py
-352
View File
@@ -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 ``<system-reminder>`` 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 ``<tool_output>`` 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("<tool_output>\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)
# ---------------------------------------------------------------------------
turnstone/shared_static/history_normalize.js
-235
View File
@@ -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
// `<tool_output>` envelope server-side); LIST-content tool results are a
// KNOWN GAP — their envelope is not stripped here and a carried advisory is
// dropped (uncommon: image/structured MCP result + concurrent interjection;
// closed by the planned server-side unification).
//
// Pure + DOM-free so it can be unit-tested under node
// (tests/test_app_js.py::test_normalize_history_*).
const _WATCH_REMINDER_OPTIONAL_KEYS = [
"watch_name",
"command",
"poll_count",
"max_polls",
"is_final",
];
const _TOOL_ERROR_PREFIXES = [
"Error",
"Command timed out",
"Search timed out",
"Unknown tool:",
"JSON parse error:",
"MCP prompt timed out",
"MCP prompt error",
];
function normalizeHistoryMessages(messages) {
if (!Array.isArray(messages)) return [];
// Pre-scan: which tool_call_ids have a result message? An assistant
// tool_call with no result is an orphan → its turn is "pending"
// (awaiting approval on a live reconnect, or interrupted) and must not
// render a fake "approved" badge. Mirrors `_build_history`'s
// has_pending_approval handling for the realistic cases.
const resultedCallIds = new Set();
for (const m of messages) {
if (m && m.role === "tool" && m.tool_call_id) {
resultedCallIds.add(String(m.tool_call_id));
}
}
const out = [];
for (const msg of messages) {
if (!msg || typeof msg !== "object") continue;
const role = msg.role;
let content = msg.content;
let attachments = null;
// (1) Collapse multipart user content (text + image_url + document
// parts) to a plain string + a derived attachment list.
if (role === "user" && Array.isArray(content)) {
const textParts = [];
const meta = [];
for (const part of content) {
if (!part || typeof part !== "object") continue;
if (part.type === "text") {
textParts.push(String(part.text || ""));
} else if (part.type === "image_url") {
meta.push({ kind: "image", filename: "", mime_type: "" });
} else if (part.type === "document") {
const d = part.document || {};
meta.push({
kind: "text",
filename: String(d.name || ""),
mime_type: String(d.media_type || ""),
});
}
}
content = textParts.join("\n");
if (meta.length) attachments = meta;
}
// (2) The authoritative `_attachments_meta` side-channel wins when
// present (carries image filenames the image_url part can't).
const sideMeta = msg._attachments_meta;
if (Array.isArray(sideMeta) && sideMeta.length) {
attachments = sideMeta
.filter((x) => x && typeof x === "object")
.map((x) => ({
kind: String(x.kind || ""),
filename: String(x.filename || ""),
mime_type: String(x.mime_type || ""),
}));
}
const entry = { role: role, content: content };
if (attachments && attachments.length) entry.attachments = attachments;
// (3) `_source` side-channel → top-level `source` (drives the
// `.msg.user.system-nudge` marker).
if (msg._source) entry.source = String(msg._source);
// (4) `_reminders` side-channel → top-level `reminders`, filtered +
// key-projected (mirrors `_build_history`'s reminder filter).
if (Array.isArray(msg._reminders)) {
const clean = [];
for (const r of msg._reminders) {
if (!r || typeof r !== "object") continue;
const rtype = String(r.type || "");
const rtext = String(r.text || "");
if (!rtype && !rtext) continue;
const c = { type: rtype, text: rtext };
for (const k of _WATCH_REMINDER_OPTIONAL_KEYS) {
if (k in r) c[k] = r[k];
}
clean.push(c);
}
if (clean.length) entry.reminders = clean;
}
// (5) Reasoning is already projected top-level by the REST
// decoration — pass it through.
if (msg.reasoning) entry.reasoning = msg.reasoning;
// (6) Flatten OpenAI-nested tool_calls `{id, function:{name,
// arguments}}` → `{id, name, arguments}` that `replayHistory`
// reads, carrying the decoration (`verdict` / `output_assessment`,
// which `decorate_tool_call` placed top-level on the call).
if (Array.isArray(msg.tool_calls) && msg.tool_calls.length) {
const tcs = [];
for (const tc of msg.tool_calls) {
if (!tc || typeof tc !== "object") continue;
const fn = tc.function || {};
const id = tc.id || "";
const flat = {
id: id,
name: fn.name || tc.name || "",
arguments:
fn.arguments != null
? fn.arguments
: tc.arguments != null
? tc.arguments
: "",
};
if (tc.verdict) flat.verdict = tc.verdict;
if (tc.output_assessment) flat.output_assessment = tc.output_assessment;
tcs.push(flat);
}
entry.tool_calls = tcs;
}
// (7) Tool results: carry `tool_call_id` + `advisories` (already
// top-level on the REST shape), coerce list content to text, and
// derive `denied` / `is_error` from the content prefix — the REST
// path does NOT pre-set these (mirrors `_build_history`).
if (role === "tool") {
if (msg.tool_call_id) entry.tool_call_id = String(msg.tool_call_id);
if (Array.isArray(msg.advisories) && msg.advisories.length) {
entry.advisories = msg.advisories;
}
if (Array.isArray(content)) {
content = content
.filter((p) => p && p.type === "text")
.map((p) => String(p.text || ""))
.join("\n");
entry.content = content;
}
const c = typeof content === "string" ? content : "";
if (c.startsWith("Denied by user") || c.startsWith("Blocked")) {
entry.denied = true;
}
if (msg.is_error || _TOOL_ERROR_PREFIXES.some((p) => c.startsWith(p))) {
entry.is_error = true;
}
}
out.push(entry);
}
// (8) Propagate denial from a tool result to its parent assistant turn
// so the tool block renders the denied (not approved) badge.
// Mirrors `_build_history`'s last-assistant propagation.
let lastAssistantIdx = -1;
for (let i = 0; i < out.length; i++) {
if (out[i].tool_calls) {
lastAssistantIdx = i;
} else if (
out[i].role === "tool" &&
out[i].denied &&
lastAssistantIdx >= 0
) {
out[lastAssistantIdx].denied = true;
}
}
// (9) Mark `pending` ONLY on the LAST assistant tool-call turn, and only
// when it has an orphan (a tool_call with no result in the loaded
// window) — the proxy for "awaiting approval". Mirrors
// `_build_history`'s reversed single-turn pending. A mid-conversation
// cancelled/interrupted tool call is also an orphan but must still
// render its tool block (not vanish), so it is NOT marked pending.
for (let i = out.length - 1; i >= 0; i--) {
const e = out[i];
if (e.tool_calls && e.tool_calls.length) {
const hasOrphan = e.tool_calls.some(
(tc) => tc.id && !resultedCallIds.has(String(tc.id)),
);
if (hasOrphan) e.pending = true;
break;
}
}
return out;
}
// Dual-mode: a browser global (loaded via <script>) AND a CommonJS export
// so node-based unit tests can require it without a DOM.
if (typeof module !== "undefined" && module.exports) {
module.exports = { normalizeHistoryMessages };
}
turnstone/ui/static/app.js
+6 -5
View File
@@ -786,11 +786,12 @@ class Pane {
// paint the wrong ws's history into the pane.
if (token !== undefined && token !== this._historyLoadToken) return;
if (data) {
// The REST /history payload is provider-native (nested tool_calls,
// `_source`/`_reminders`/`_attachments_meta` side-channels, multipart
// content, no derived denied/is_error/pending). Normalize it to the
// projected shape replayHistory renders — see history_normalize.js.
this.replayHistory(normalizeHistoryMessages(data.messages || []));
// The REST /history payload is already the canonical projected wire
// shape (server-side projection in make_history_handler:
// flat tool_calls, top-level source/reminders/attachments, collapsed
// content, derived denied/is_error/pending) — feed it straight to
// replayHistory. No client-side normalisation.
this.replayHistory(data.messages || []);
} else {
this.showEmptyState();
}
turnstone/ui/static/index.html
-1
View File
@@ -620,7 +620,6 @@
];
</script>
<script src="/shared/utils.js"></script>
<script src="/shared/history_normalize.js"></script>
<script src="/shared/cards.js"></script>
<script src="/shared/toast.js"></script>
<script src="/shared/composer.js"></script>