diff --git a/sdk/typescript/src/events.ts b/sdk/typescript/src/events.ts index 2fba228b..abb60797 100644 --- a/sdk/typescript/src/events.ts +++ b/sdk/typescript/src/events.ts @@ -22,6 +22,10 @@ export interface HistoryEvent { * - `source`: the operator-context kind on a `system` turn (`output_guard` / * `user_interjection` / `tool_error` / ...), or `system_nudge` on a * wake-driven empty user turn + * - `meta`: structured per-kind fields on an operator-context `system` turn + * (e.g. `watch_triggered`'s `{watch_name, command, poll_count, max_polls, + * is_final}`) so the renderer can rebuild per-kind UI (the watch-result + * card); absent for kinds with no structured data * - `attachments`: per-attachment metadata `{kind, filename, mime_type, size_bytes}` * - `reasoning`: concatenated reasoning text for assistant turns that * round-tripped a thinking-block lane (Anthropic-with-thinking today; diff --git a/tests/test_app_js.py b/tests/test_app_js.py index c093b614..b3a82537 100644 --- a/tests/test_app_js.py +++ b/tests/test_app_js.py @@ -270,7 +270,10 @@ def test_replay_renders_system_turn_via_add_system_context() -> None: assert 'msg.role === "system"' in fn, ( "replayHistory must have a system-role branch for first-class operator-context turns." ) - assert "addSystemContext(msg.content" in fn, ( + # Whitespace-tolerant: the call carries a 3rd ``meta`` arg now, so the + # formatter wraps it across lines — match the call + first arg, not a + # brittle contiguous substring. + assert re.search(r"addSystemContext\(\s*msg\.content", fn), ( "the system-role branch must route the turn through addSystemContext " "so it renders as an operator bubble." ) diff --git a/tests/test_coordinator_idle_observer.py b/tests/test_coordinator_idle_observer.py index 67cd5628..b18e508a 100644 --- a/tests/test_coordinator_idle_observer.py +++ b/tests/test_coordinator_idle_observer.py @@ -169,6 +169,34 @@ class TestEnqueueOnIdle: assert "child-a" in text assert "child-b" in text + def test_idle_children_carries_structured_meta(self, coord_setup): + # The nudge rides the structured child list as ``metadata`` so the FE + # rebuilds the idle-children card; the same list ``format_idle_children + # _nudge`` rendered into ``text`` (one source, no drift). + mgr, storage, ws = coord_setup + _add_active_child(storage, ws_id="child-a", name="research", state="running") + _add_active_child(storage, ws_id="child-b", name="deploy", state="thinking") + ws.session.messages = turns_from_dicts( + [ + {"role": "user", "content": "go"}, + {"role": "assistant", "content": "ok"}, + ] + ) + + observer = CoordinatorIdleObserver(mgr, storage) + observer.start() + mgr.fire_state(ws.id, WorkstreamState.IDLE) + + snap = ws.session._nudge_queue.pending_with_metadata(channel="any") + assert len(snap) == 1 + meta = snap[0][2] + assert meta == { + "children": [ + {"ws_id": "child-a", "name": "research", "state": "running"}, + {"ws_id": "child-b", "name": "deploy", "state": "thinking"}, + ] + } + def test_idle_with_no_active_children_no_enqueue(self, coord_setup): mgr, storage, ws = coord_setup # storage.children is empty diff --git a/tests/test_history_projection.py b/tests/test_history_projection.py index be180fac..7f1b1a4f 100644 --- a/tests/test_history_projection.py +++ b/tests/test_history_projection.py @@ -56,6 +56,28 @@ class TestSystemTurnProjection: assert history[0]["source"] == "user_interjection" assert history[0]["content"] == "check the logs" + def test_system_turn_source_meta_projects(self) -> None: + # ``_source_meta`` → ``meta`` so a reconnecting tab rebuilds the same + # per-kind card (the watch-result card etc.) the live SSE event drives. + history = project_history_messages( + [ + { + "role": "system", + "_source": "watch_triggered", + "content": "ci failed", + "_source_meta": {"watch_name": "ci", "poll_count": 3}, + } + ] + ) + assert history[0]["source"] == "watch_triggered" + assert history[0]["meta"] == {"watch_name": "ci", "poll_count": 3} + + def test_system_turn_without_meta_omits_meta_field(self) -> None: + history = project_history_messages( + [{"role": "system", "_source": "correction", "content": "watch out"}] + ) + assert "meta" not in history[0] + def test_legacy_reminders_column_not_projected(self) -> None: """A pre-migration row that still carries ``_reminders`` must NOT surface a ``reminders`` field — the projection dropped that lane.""" diff --git a/tests/test_idle_nudge_wake_integration.py b/tests/test_idle_nudge_wake_integration.py index 640a8aef..acedabd1 100644 --- a/tests/test_idle_nudge_wake_integration.py +++ b/tests/test_idle_nudge_wake_integration.py @@ -70,7 +70,7 @@ class _FakeUI: def on_state_change(self, state: str) -> None: self.events.append(("state", state)) - def on_system_turn(self, content: str, source: str) -> None: + def on_system_turn(self, content: str, source: str, meta: dict | None = None) -> None: self.events.append(("system_turn", content, source)) def on_error(self, message: str) -> None: diff --git a/tests/test_reconstruct_messages.py b/tests/test_reconstruct_messages.py index 07c2ae4c..f70fc917 100644 --- a/tests/test_reconstruct_messages.py +++ b/tests/test_reconstruct_messages.py @@ -372,6 +372,61 @@ class TestSystemTurns: assert msgs[2]["role"] == "system" assert "_source" not in msgs[2] + def test_system_row_with_meta_column_reconstructed(self): + # A system row carrying the JSON ``meta`` column (migration 060) rehydrates + # the structured operator meta onto the dict as ``_source_meta`` — the + # source the FE watch-result card derives from. Full 11-tuple: + # (id, role, content, tool_name, tc_id, pdata, tool_calls, source, + # event_id, is_error, meta). + meta_json = json.dumps({"watch_name": "ci", "command": "make test", "poll_count": 3}) + row = ( + 1, + "system", + "ci failed", + None, + None, + None, + None, + "watch_triggered", + None, + False, + meta_json, + ) + msgs = reconstruct_messages([row], "ws1") + assert msgs[0] == { + "role": "system", + "content": "ci failed", + "_source": "watch_triggered", + "_source_meta": {"watch_name": "ci", "command": "make test", "poll_count": 3}, + } + + def test_legacy_short_row_without_meta_column_valid(self): + # A pre-meta 8-tuple row reconstructs fine (defensive length check) and + # carries no ``_source_meta`` key. + rows = [_row("system", "old note", source="output_guard")] + msgs = reconstruct_messages(rows, "ws1") + assert msgs[0]["_source"] == "output_guard" + assert "_source_meta" not in msgs[0] + + def test_malformed_meta_column_dropped_not_crashed(self): + # A non-JSON / non-object meta column is dropped (the human-readable body + # still rides ``content``), never raised. + row = ( + 1, + "system", + "note", + None, + None, + None, + None, + "output_guard", + None, + False, + "{bad json", + ) + msgs = reconstruct_messages([row], "ws1") + assert "_source_meta" not in msgs[0] + def test_trailing_system_after_incomplete_assistant_strips_both(self): """A nudge appended after an interrupted tool-call turn must not leave the orphaned assistant — the strip walks through the trailing system diff --git a/tests/test_session.py b/tests/test_session.py index f782bb84..da1fbcbd 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -61,7 +61,7 @@ class NullUI: def on_error(self, message): pass - def on_system_turn(self, content, source): + def on_system_turn(self, content, source, meta=None): pass def on_state_change(self, state): @@ -2942,7 +2942,15 @@ class TestMetacognitiveBuffers: assert "HIGH" in content assert "API key detected" in content assert "redacted" in content.lower() - assert meta == {} + # The structured finding rides as meta (the source the FE card and the + # rendered ``content`` both derive from); ``redacted`` is the boolean + # projection of ``sanitized is not None``. + assert meta == { + "flags": ["credential_leak"], + "risk_level": "high", + "annotations": ["API key detected"], + "redacted": True, + } def test_collect_advisories_drains_queued_messages_on_last_result(self, tmp_db): """Queued user messages drain into a ``user_interjection`` spec on @@ -2958,7 +2966,10 @@ class TestMetacognitiveBuffers: assert len(specs) == 1 source, content, meta = specs[0] assert source == "user_interjection" - assert meta == {"priority": "notice"} + # Meta carries the priority AND the user's raw words, so the FE renders + # a clean "queued message" bubble while ``content`` keeps the framed, + # model-facing wording. + assert meta == {"priority": "notice", "message": "hows it going?"} # Framed as the user's words (known #2 — keeps user, not operator, # authority, especially on the native path), not the raw text. assert content.endswith("User message: hows it going?") @@ -3523,9 +3534,11 @@ class TestMetacognitiveBuffers: with patch("turnstone.core.session.save_message"): session._emit_pending_user_nudges() assert session.ui.on_system_turn.call_count == 1 - content, source = session.ui.on_system_turn.call_args.args + content, source, meta = session.ui.on_system_turn.call_args.args assert content == "watch out" assert source == "correction" + # ``correction`` is a static nudge — no structured per-kind meta. + assert meta is None def test_emit_user_nudges_swallows_on_system_turn_failure(self, tmp_db): """A UI hook that raises (queue full, unexpected bug) must not abort @@ -4388,9 +4401,24 @@ class TestSessionUIBaseSystemTurnHook: ui = _RecordingUI() ui.on_system_turn("watch out", "correction") + # A static-text kind carries no structured meta → ``meta`` is None. assert ui.events == [ - {"type": "system_turn", "content": "watch out", "source": "correction"} + { + "type": "system_turn", + "content": "watch out", + "source": "correction", + "meta": None, + } ] + # A structured kind rides its per-kind meta on the event so the FE + # rebuilds the card live, in lockstep with /history replay. + ui.on_system_turn("ci failed", "watch_triggered", {"watch_name": "ci", "poll_count": 3}) + assert ui.events[-1] == { + "type": "system_turn", + "content": "ci failed", + "source": "watch_triggered", + "meta": {"watch_name": "ci", "poll_count": 3}, + } def test_on_system_turn_carries_each_source_kind(self): from turnstone.core.session_ui_base import SessionUIBase diff --git a/tests/test_storage_sqlite.py b/tests/test_storage_sqlite.py index fc118915..d012c30d 100644 --- a/tests/test_storage_sqlite.py +++ b/tests/test_storage_sqlite.py @@ -36,6 +36,36 @@ class TestSaveAndLoadMessages: assert msgs[1]["role"] == "assistant" assert msgs[1]["content"] == "world" + def test_system_turn_meta_roundtrip(self, backend): + # The structured per-kind operator meta saved on a ``system`` row + # round-trips back as ``_source_meta`` through the real insert + SELECT + # (the ``conversations.meta`` column, migration 060). + import json + + backend.register_workstream("s1") + backend.save_message("s1", "user", "go") + backend.save_message( + "s1", + "system", + "ci failed", + source="watch_triggered", + meta=json.dumps({"watch_name": "ci", "command": "make test", "poll_count": 3}), + ) + msgs = backend.load_messages("s1") + assert msgs[1] == { + "role": "system", + "content": "ci failed", + "_source": "watch_triggered", + "_source_meta": {"watch_name": "ci", "command": "make test", "poll_count": 3}, + } + + def test_ordinary_row_has_null_meta(self, backend): + # A non-operator row carries no meta — no ``_source_meta`` on reload. + backend.register_workstream("s1") + backend.save_message("s1", "user", "hello") + msgs = backend.load_messages("s1") + assert "_source_meta" not in msgs[0] + def test_tool_call_grouping(self, backend): import json diff --git a/tests/test_tool_advisory.py b/tests/test_tool_advisory.py index 48ff4c35..ea7b3383 100644 --- a/tests/test_tool_advisory.py +++ b/tests/test_tool_advisory.py @@ -8,6 +8,7 @@ from turnstone.core.tool_advisory import ( SYSTEM_TURN_SOURCES, make_system_turn, parse_priority, + render_output_guard_text, render_user_interjection, ) @@ -23,31 +24,40 @@ class TestMakeSystemTurn: "content": "check auth too", } - def test_meta_keys_underscore_prefixed(self) -> None: - turn = make_system_turn( - "watch_triggered", "ci failed", watch_name="ci", priority="important" - ) - assert turn["_watch_name"] == "ci" - assert turn["_priority"] == "important" - assert turn["role"] == "system" - assert turn["_source"] == "watch_triggered" - assert turn["content"] == "ci failed" + def test_meta_carried_as_source_meta_dict(self) -> None: + # Meta rides as ONE ``_source_meta`` dict (not scattered ``_``-prefixed + # siblings) — one carrier mapping to one storage column / one FE field. + turn = make_system_turn("watch_triggered", "ci failed", watch_name="ci", poll_count=3) + assert turn == { + "role": "system", + "_source": "watch_triggered", + "content": "ci failed", + "_source_meta": {"watch_name": "ci", "poll_count": 3}, + } - def test_already_underscored_meta_kept(self) -> None: - turn = make_system_turn("repeat", "stop repeating", _event_id=7) - assert turn["_event_id"] == 7 + def test_no_meta_omits_source_meta(self) -> None: + # A kind with no structured fields carries no ``_source_meta`` key. + turn = make_system_turn("output_guard", "flag (HIGH)") + assert turn == { + "role": "system", + "_source": "output_guard", + "content": "flag (HIGH)", + } + assert "_source_meta" not in turn def test_unknown_source_rejected(self) -> None: with pytest.raises(ValueError, match="unknown system-turn source"): make_system_turn("bogus", "x") - def test_meta_key_colliding_with_source_rejected(self) -> None: - # `source=` can't even reach **meta — it's a named parameter, so Python - # raises TypeError first. The only reachable collision is an explicit - # ``_source=`` meta key, which the builder rejects rather than letting it - # silently clobber the validated source. - with pytest.raises(ValueError, match="collides with reserved"): - make_system_turn("repeat", "x", _source="evil") + def test_meta_namespaced_cannot_clobber_reserved_keys(self) -> None: + # Because meta rides namespaced inside ``_source_meta``, a meta key that + # mirrors a reserved top-level key (``role`` / ``_source``) lands inside + # the dict and cannot overwrite the validated turn fields — so the old + # collision guard is no longer needed (the shape makes it impossible). + turn = make_system_turn("repeat", "x", role="evil", _source="spoof") + assert turn["role"] == "system" + assert turn["_source"] == "repeat" + assert turn["_source_meta"] == {"role": "evil", "_source": "spoof"} def test_vocabulary_mirrors_nudge_map_both_directions(self) -> None: # The nudge-derived sources must equal _NUDGE_MAP exactly: a new nudge @@ -60,6 +70,58 @@ class TestMakeSystemTurn: assert nudge_sources == set(_NUDGE_MAP) +class TestRenderOutputGuardText: + """render_output_guard_text() projects the structured guard meta to prose.""" + + def test_full_finding(self) -> None: + text = render_output_guard_text( + { + "flags": ["aws_key", "private_key"], + "risk_level": "high", + "annotations": ["matched AKIA…", "PEM header"], + "redacted": True, + } + ) + assert text == ( + "Output guard: aws_key, private_key (HIGH)\n" + " matched AKIA…\n" + " PEM header\n" + "Credentials have been redacted. Do not attempt to reconstruct redacted values." + ) + + def test_no_annotations_no_redaction(self) -> None: + text = render_output_guard_text( + {"flags": ["pii"], "risk_level": "low", "annotations": [], "redacted": False} + ) + assert text == "Output guard: pii (LOW)" + + def test_missing_keys_default_gracefully(self) -> None: + # Defensive: a partial meta dict never raises. + assert render_output_guard_text({}) == "Output guard: (NONE)" + + +class TestMetaIsWireStripped: + """The structured operator meta must never reach the LLM wire.""" + + def test_source_meta_stripped_by_sanitize(self) -> None: + from turnstone.core.providers._openai_common import sanitize_messages + + out = sanitize_messages( + [ + {"role": "user", "content": "hi"}, + { + "role": "system", + "_source": "watch_triggered", + "content": "ci failed", + "_source_meta": {"command": "rm -rf /", "watch_name": "ci"}, + }, + ] + ) + # No leading-underscore side channel survives to the wire. + assert all(not k.startswith("_") for m in out for k in m) + assert out[1] == {"role": "system", "content": "ci failed"} + + class TestParsePriority: """parse_priority() extracts !!! prefix as priority signal.""" diff --git a/tests/test_trajectory.py b/tests/test_trajectory.py index 81b6beca..29d86851 100644 --- a/tests/test_trajectory.py +++ b/tests/test_trajectory.py @@ -90,6 +90,13 @@ _ROUNDTRIP: list[dict[str, Any]] = [ }, {"role": "system", "content": "guard note", "_source": "output_guard"}, {"role": "system", "content": "you are an assistant"}, # base prompt, no _source + # Operator turn with structured per-kind meta (the watch-result card source). + { + "role": "system", + "content": "ci failed", + "_source": "watch_triggered", + "_source_meta": {"watch_name": "ci", "command": "make test", "poll_count": 3}, + }, ] @@ -109,6 +116,30 @@ def test_turns_from_dicts_preserves_order_and_count() -> None: # --------------------------------------------------------------------------- # # Field mapping — the side channels become typed Turn fields. # --------------------------------------------------------------------------- # +def test_source_meta_maps_to_meta_extra() -> None: + # The per-kind operator meta rides a single ``_source_meta`` dict and lands + # in ``Turn.meta.extra["source_meta"]`` (and round-trips back out). + t = turn_from_dict( + { + "role": "system", + "content": "ci failed", + "_source": "watch_triggered", + "_source_meta": {"watch_name": "ci", "poll_count": 3}, + } + ) + assert t.meta.extra["source_meta"] == {"watch_name": "ci", "poll_count": 3} + assert turn_to_dict(t)["_source_meta"] == {"watch_name": "ci", "poll_count": 3} + + +def test_empty_source_meta_omitted() -> None: + # An empty meta dict is not carried (no key on the Turn, none re-emitted). + t = turn_from_dict( + {"role": "system", "content": "n", "_source": "correction", "_source_meta": {}} + ) + assert "source_meta" not in t.meta.extra + assert "_source_meta" not in turn_to_dict(t) + + def test_source_maps_to_source_field() -> None: t = turn_from_dict({"role": "system", "content": "n", "_source": "tool_error"}) assert t.role is Role.SYSTEM diff --git a/tests/test_watch.py b/tests/test_watch.py index 2ef6297e..16c9de1a 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -336,6 +336,10 @@ class TestBuildWatchReminder: assert reminder["type"] == "watch_triggered" assert reminder["watch_name"] == "pr-review" assert reminder["command"] == "gh pr view --json state" + # The raw shell output rides as its own field so the FE card body shows + # it alone (no header / command repeat); the wire ``text`` keeps the + # full prose for the model. + assert reminder["output"] == '{"state": "MERGED"}' assert reminder["poll_count"] == 5 assert reminder["max_polls"] == 100 assert reminder["is_final"] is True diff --git a/turnstone/cli.py b/turnstone/cli.py index d370a339..d1eba511 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -301,12 +301,14 @@ class TerminalUI(SessionUI): sys.stdout.write(f"{RED}{message}{RESET}\n") sys.stdout.flush() - def on_system_turn(self, content: str, source: str) -> None: + def on_system_turn(self, content: str, source: str, meta: dict[str, Any] | None = None) -> None: """Render a first-class operator-context system turn as an ``[operator · source] text`` line in the terminal — the CLI's equivalent of the web UI's operator bubble. Terminal output is anchored by flow (the line lands directly after the turn it - relates to), so no DOM anchoring is needed. + relates to), so no DOM anchoring is needed. ``meta`` (the structured + per-kind fields the web card uses) is unused here: the formatted body + in ``content`` already reads well in a terminal. """ label = "operator" + (f" · {source}" if source else "") sys.stdout.write(f"{YELLOW}[{label}]{RESET} {content}\n") diff --git a/turnstone/console/coordinator_idle_observer.py b/turnstone/console/coordinator_idle_observer.py index 606e0949..03e35d13 100644 --- a/turnstone/console/coordinator_idle_observer.py +++ b/turnstone/console/coordinator_idle_observer.py @@ -39,6 +39,7 @@ from turnstone.core.log import get_logger from turnstone.core.metacognition import ( _cooldown_allows, format_idle_children_nudge, + sanitize_name, should_nudge, ) from turnstone.core.trajectory import Role @@ -205,6 +206,21 @@ class CoordinatorIdleObserver: if not text: # belt-and-braces: formatter empty-input guard return + # Structured ``_source_meta`` for the FE idle-children card — the same + # child list ``format_idle_children_nudge`` rendered into ``text`` above, + # so the card and the model-facing prose derive from one source. Names + # are sanitized identically (``sanitize_name``) so a steering-vector / + # angle-bracket name can't reach the operator card any more than the + # text; the FE additionally renders every field via ``textContent``. + children_meta = [ + { + "ws_id": str(c.get("ws_id", "")), + "name": sanitize_name(str(c.get("name", ""))), + "state": str(c.get("state", "")), + } + for c in active + ] + # Bind ws.id + user_id by closure so the predicate captures the # workstream identity (not the live ``ws`` reference, which # could mutate). The predicate runs at drain time outside the @@ -235,6 +251,7 @@ class CoordinatorIdleObserver: text, "any", valid_until=_still_has_active_children, + metadata={"children": children_meta}, ) with self._fire_counts_lock: ws_caps = self._fire_counts.setdefault(ws_id, {}) diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index 6ad350c1..77d20cd7 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -391,6 +391,180 @@ return appendMsg(role, esc(text), opts); } + // Structured ``.msg.watch-result`` card for a ``watch_triggered`` + // operator-context system turn — command-preview header + shell-output body + // + poll-counter footer. ``content`` is the formatted watch body (the system + // turn's content); ``meta`` carries the structured fields (``watch_name`` / + // ``command`` / ``poll_count`` / ``max_polls`` / ``is_final``) delivered live + // on the ``system_turn`` SSE event and on the ``/history`` projection. All + // text goes through ``textContent`` so shell output containing angle brackets + // / scripts / steering bytes renders inertly. Mirrors the interactive pane's + // _buildWatchResultBubble. + function appendWatchResult(meta, content) { + const el = document.createElement("div"); + el.className = "msg watch-result"; + el.setAttribute("role", "article"); + el.setAttribute("data-ts-role", "watch"); + el.setAttribute("aria-label", "watch"); + const header = document.createElement("div"); + header.className = "msg-watch-header"; + header.textContent = + "watch" + (meta.watch_name ? " · " + String(meta.watch_name) : ""); + el.appendChild(header); + if (meta.command) { + const cmd = document.createElement("div"); + cmd.className = "msg-watch-cmd"; + cmd.textContent = "$ " + String(meta.command); + el.appendChild(cmd); + } + const body = document.createElement("pre"); + body.className = "msg-watch-body"; + // Prefer the structured ``output`` (raw shell output alone) so the body + // doesn't re-print the header / ``$ command`` lines the chrome already + // shows; fall back to the full turn ``content`` for legacy turns that + // predate the ``output`` meta field (migration 060 is additive). + body.textContent = + meta.output != null ? String(meta.output) : content || ""; + el.appendChild(body); + if (meta.poll_count != null && meta.max_polls != null) { + const footer = document.createElement("div"); + footer.className = "msg-watch-footer"; + const finalSuffix = meta.is_final ? " · final" : ""; + footer.textContent = + "poll " + + String(meta.poll_count) + + "/" + + String(meta.max_polls) + + finalSuffix; + el.appendChild(footer); + } + messagesEl.appendChild(el); + _scheduleScroll(); + return el; + } + + // Structured ``.msg.guard-finding`` card for an ``output_guard`` + // operator-context system turn. ``meta`` carries ``{flags, risk_level, + // annotations, redacted}``. Reuses the tool-row warning chip's risk / flags + // / redaction vocabulary (``.coord-tool-row-warning``) so guard findings read + // identically wherever they surface, then appends the annotations the inline + // tool chip omits. All text via textContent. Mirrors the interactive pane's + // _buildGuardFindingBubble. + function appendGuardFinding(meta) { + const el = document.createElement("div"); + el.className = "msg guard-finding"; + el.setAttribute("role", "article"); + el.setAttribute("data-ts-role", "output_guard"); + el.setAttribute("aria-label", "output guard"); + const risk = String(meta.risk_level || "medium"); + const warn = document.createElement("div"); + warn.className = "coord-tool-row-warning coord-tool-row-warning--" + risk; + warn.setAttribute("role", "status"); + const label = document.createElement("span"); + label.className = "coord-tool-row-warning-label"; + label.textContent = "⚠ " + risk.toUpperCase(); + warn.appendChild(label); + const flags = Array.isArray(meta.flags) ? meta.flags : []; + if (flags.length) { + warn.appendChild(document.createTextNode(" " + flags.join(", "))); + } + if (meta.redacted) { + const redacted = document.createElement("span"); + redacted.className = "coord-tool-row-warning-redacted"; + redacted.textContent = " (credentials redacted)"; + warn.appendChild(redacted); + } + el.appendChild(warn); + const anns = Array.isArray(meta.annotations) ? meta.annotations : []; + for (let i = 0; i < anns.length; i++) { + const a = document.createElement("div"); + a.className = "msg-guard-annotation"; + a.textContent = String(anns[i]); + el.appendChild(a); + } + messagesEl.appendChild(el); + _scheduleScroll(); + return el; + } + + // Structured ``.msg.idle-children`` card for the coordinator-only + // ``idle_children`` operator-context system turn — lists the child + // workstreams still running while the coordinator went idle. ``meta.children`` + // is ``[{ws_id, name, state}]`` (names already ``sanitize_name``-cleaned at the + // producer); rendered via textContent so a hostile workstream name is inert. + function appendIdleChildren(meta) { + const el = document.createElement("div"); + el.className = "msg idle-children"; + el.setAttribute("role", "article"); + el.setAttribute("data-ts-role", "idle_children"); + el.setAttribute("aria-label", "idle children"); + const children = Array.isArray(meta.children) ? meta.children : []; + const header = document.createElement("div"); + header.className = "msg-idle-header"; + header.textContent = + "idle · " + + children.length + + (children.length === 1 + ? " child still running" + : " children still running"); + el.appendChild(header); + const list = document.createElement("ul"); + list.className = "msg-idle-list"; + for (let i = 0; i < children.length; i++) { + const c = children[i] || {}; + const li = document.createElement("li"); + li.className = "msg-idle-child"; + const name = document.createElement("span"); + name.className = "msg-idle-child-name"; + name.textContent = String(c.name || c.ws_id || "child"); + li.appendChild(name); + if (c.state) { + const state = document.createElement("span"); + state.className = "msg-idle-child-state"; + state.textContent = String(c.state); + li.appendChild(state); + } + list.appendChild(li); + } + el.appendChild(list); + messagesEl.appendChild(el); + _scheduleScroll(); + return el; + } + + // "queued message" bubble for a ``user_interjection`` system turn — shows the + // user's raw words (``meta.message``) rather than the model-directed framing + // baked into ``content``, with brighter emphasis for ``!!!``-important + // interjections. Reuses ``appendText`` (→ ``.msg.system-context``) + a class. + function appendInterjection(meta, content) { + const important = meta && meta.priority === "important"; + const text = + meta && meta.message != null ? String(meta.message) : content || ""; + const el = appendText("system", text, { + label: important ? "queued message · important" : "queued message", + }); + el.classList.add("interjection"); + if (important) el.classList.add("important"); + return el; + } + + // Dispatch a first-class operator-context system turn to the right renderer. + // Shared by the live ``system_turn`` SSE handler and history replay so the + // two can't drift on which kinds get structured cards. ``watch_triggered`` / + // ``output_guard`` / ``idle_children`` carry structured ``meta`` → cards; + // ``user_interjection`` → a "queued message" bubble; everything else → the + // labeled operator bubble. + function renderSystemTurn(source, content, meta) { + const m = meta && typeof meta === "object" ? meta : null; + if (source === "watch_triggered" && m) + return appendWatchResult(m, content || ""); + if (source === "output_guard" && m) return appendGuardFinding(m); + if (source === "idle_children" && m) return appendIdleChildren(m); + if (source === "user_interjection") + return appendInterjection(m, content || ""); + return appendText("system", content || "", { label: source || "system" }); + } + // User-message bubble with attachment-pill cluster appended below // the text. Mirrors Pane.addUserMessage in the interactive UI so // live-send and history-replay both render the same chip strip the @@ -2251,14 +2425,12 @@ break; case "system_turn": // First-class operator-context system turn (output-guard finding, - // user interjection, metacognitive nudge — see make_system_turn). - // Consolidates the legacy user_reminder / tool_reminder events into - // one operator bubble rendered in trajectory sequence (it FOLLOWS - // the turn it advises). ``ev.source`` carries the kind; the - // ``system`` _MSG_VARIANTS entry gives it the operator styling. - appendText("system", ev.content || "", { - label: ev.source || "system", - }); + // user interjection, metacognitive nudge, watch result — see + // make_system_turn). Rendered in trajectory sequence (it FOLLOWS the + // turn it advises). ``renderSystemTurn`` routes by ``ev.source`` to the + // structured card (watch / guard / idle-children) or the operator bubble + // (carrying ``ev.meta`` so cards rebuild identically live and on replay). + renderSystemTurn(ev.source || "", ev.content || "", ev.meta); break; case "connected": // First yield from _coord_events_replay — populates the @@ -4626,11 +4798,12 @@ label: role, }); } else if (role === "system") { - // First-class operator-context system turn — label with the - // ``source`` kind (output_guard / user_interjection / ...); - // the ``system`` _MSG_VARIANTS entry gives it operator styling. + // First-class operator-context system turn — ``renderSystemTurn`` + // routes by ``m.source`` to the structured card (watch / guard / + // idle-children) or the operator bubble, reading ``m.meta`` from the + // ``/history`` projection so replay matches the live render exactly. if (!content) return; - appendText(role, content, { label: m.source || "system" }); + renderSystemTurn(m.source || "", content, m.meta); } else { if (!content) return; appendText(role, content, { label: role }); diff --git a/turnstone/core/history_decoration.py b/turnstone/core/history_decoration.py index eb24a8a1..c3cea117 100644 --- a/turnstone/core/history_decoration.py +++ b/turnstone/core/history_decoration.py @@ -560,6 +560,17 @@ def project_history_messages( if msg.get("_source"): entry["source"] = str(msg["_source"]) + # (3b) ``_source_meta`` side-channel → top-level ``meta``. The + # operator turn's structured per-kind fields (``watch_triggered``'s + # ``watch_name`` / command / poll counters) — the FE branches on + # ``source`` and uses these to rebuild per-kind rendering (the + # watch-result card) instead of a plain operator bubble. Mirrors + # the live ``on_system_turn`` SSE event's ``meta`` field so a + # reconnecting tab renders identically. + source_meta = msg.get("_source_meta") + if isinstance(source_meta, dict) and source_meta: + entry["meta"] = source_meta + # (4) 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 diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index 7510cb1f..9f86874c 100644 --- a/turnstone/core/memory.py +++ b/turnstone/core/memory.py @@ -47,6 +47,7 @@ def save_message( event_id: int | None = None, is_error: bool = False, producer: str | None = None, + meta: str | None = None, ) -> int: """Log a message to the conversations table. @@ -61,6 +62,13 @@ def save_message( time (``SessionUIBase._event_id``); the caller in ``session.py`` passes ``self.ui._event_id`` so ``/history`` can return it as the ``Last-Event-ID`` resume cursor. ``None`` for offline / bulk saves. + + ``meta`` is the pre-serialized JSON of a first-class ``system`` turn's + structured per-kind operator-context fields (e.g. ``watch_triggered``'s + ``watch_name`` / ``command`` / poll counters) — the persisted twin of the + in-memory ``Turn.meta.extra["source_meta"]`` / ``_source_meta`` side + channel. ``None`` for ordinary rows and operator turns with no extra + fields. Opaque to storage (like ``tool_calls`` / ``provider_data``). """ try: return get_storage().save_message( @@ -75,6 +83,7 @@ def save_message( event_id=event_id, is_error=is_error, producer=producer, + meta=meta, ) except Exception: log.warning("Failed to save message for ws=%s role=%s", ws_id, role, exc_info=True) diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 274e94fe..ed1b3f1e 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -116,6 +116,7 @@ from turnstone.core.storage._utils import ( ) from turnstone.core.tool_advisory import ( make_system_turn, + render_output_guard_text, render_user_interjection, ) from turnstone.core.tool_search import ToolSearchManager @@ -800,7 +801,9 @@ class SessionUI(Protocol): def on_status(self, usage: dict[str, Any], context_window: int, effort: str) -> None: ... def on_info(self, message: str) -> None: ... def on_error(self, message: str) -> None: ... - def on_system_turn(self, content: str, source: str) -> None: ... + def on_system_turn( + self, content: str, source: str, meta: dict[str, Any] | None = None + ) -> None: ... def on_state_change(self, state: str) -> None: ... def on_rename(self, name: str) -> None: ... def on_intent_verdict(self, verdict: dict[str, Any]) -> None: @@ -2602,6 +2605,11 @@ class ChatSession: except (TypeError, ValueError): pd_str = None src = msg.get("_source") + # Operator-context per-kind meta (``_source_meta`` dict) rides + # the fork too, so a forked watch-result keeps its structured + # card. Serialized to JSON for the ``conversations.meta`` column. + sm = msg.get("_source_meta") + meta_json = json.dumps(sm) if isinstance(sm, dict) and sm else None bulk_rows.append( { "ws_id": self._ws_id, @@ -2614,6 +2622,7 @@ class ChatSession: "source": src if isinstance(src, str) and src else None, "is_error": bool(msg.get("is_error", False)), "producer": msg.get("_producer"), + "meta": meta_json, } ) save_messages_bulk(bulk_rows) @@ -3728,21 +3737,29 @@ class ChatSession: throwing here must not abort the turn. *source* must be one of :data:`tool_advisory.SYSTEM_TURN_SOURCES`; - extra *meta* rides as leading-underscore sibling keys (stripped - before the wire by ``sanitize_messages``). + extra *meta* is the turn's structured per-kind data (e.g. + ``watch_triggered``'s ``watch_name`` / command / poll counters). It + rides three boundaries in lockstep: the in-memory ``Turn`` (as + ``meta.extra["source_meta"]`` via :func:`make_system_turn` → + ``turn_from_dict``), the persisted ``conversations.meta`` column (JSON), + and the live ``on_system_turn`` SSE hook — so a reconnecting tab and a + live mirror both rebuild the same per-kind bubble. It is stripped before + the LLM wire (``_source_meta`` is a ``_``-prefixed key). """ turn = make_system_turn(source, content, **meta) self.messages.append(turn_from_dict(turn)) self._msg_tokens.append(max(1, int(self._msg_char_count(turn) / self._chars_per_token))) + meta_json = json.dumps(meta) if meta else None save_message( self._ws_id, "system", content, source=source, event_id=self._ui_event_id(), + meta=meta_json, ) try: - self.ui.on_system_turn(content, source) + self.ui.on_system_turn(content, source, meta or None) except Exception: log.warning("ui.on_system_turn failed; system turn still appended", exc_info=True) @@ -5887,19 +5904,19 @@ class ChatSession: """ specs: list[tuple[str, str, dict[str, Any]]] = [] - # Output guard advisory — rendered inline (the wire/UI content the - # model and operator see). Mirrors the legacy GuardAdvisory.render. + # Output guard advisory. The structured finding is the source of + # truth: build ``meta`` (flags / risk / annotations / redaction), then + # derive the wire/UI text ``content`` from it via + # ``render_output_guard_text`` so the prose the model reads and the FE + # guard-finding card cannot drift. Mirrors the legacy GuardAdvisory. if assessment is not None: - lines = [ - f"Output guard: {', '.join(assessment.flags)} ({assessment.risk_level.upper()})" - ] - for ann in assessment.annotations: - lines.append(f" {ann}") - if assessment.sanitized is not None: - lines.append( - "Credentials have been redacted. Do not attempt to reconstruct redacted values." - ) - specs.append(("output_guard", "\n".join(lines), {})) + guard_meta: dict[str, Any] = { + "flags": list(assessment.flags), + "risk_level": assessment.risk_level, + "annotations": list(assessment.annotations), + "redacted": assessment.sanitized is not None, + } + specs.append(("output_guard", render_output_guard_text(guard_meta), guard_meta)) # Last-result-in-batch drain seams: queued user messages (Seam 1) # and tool/any-channel metacog nudges. Both fire once per batch so a @@ -5917,8 +5934,14 @@ class ChatSession: # where it enters as a real role=system message. if not text.strip(): continue + # ``framed`` (preamble + "User message: …") is the model-facing + # content — it keeps the user's authority framing on the wire. + # The structured meta carries the user's RAW words + priority so + # the FE renders a clean "queued message" bubble (the operator + # reads the message, not the model-directed preamble) with + # priority emphasis. Both derive from ``(text, priority)``. framed = render_user_interjection(text, priority) - specs.append(("user_interjection", framed, {"priority": priority})) + specs.append(("user_interjection", framed, {"priority": priority, "message": text})) # Metacognitive tool-channel drain. Queued by # ``_queue_tool_advisory`` from the tool_error / repeat diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index 9f1105dc..2c4abbe3 100644 --- a/turnstone/core/session_ui_base.py +++ b/turnstone/core/session_ui_base.py @@ -2363,23 +2363,29 @@ class SessionUIBase: def on_error(self, message: str) -> None: self._enqueue({"type": "error", "message": message}) - def on_system_turn(self, content: str, source: str) -> None: + def on_system_turn(self, content: str, source: str, meta: dict[str, Any] | None = None) -> None: """Surface a first-class operator-context system turn as its own UI element. Operator context (output-guard findings, user interjections, - metacognitive nudges) lives in the conversation trajectory as a - real ``{"role": "system", "_source": , ...}`` turn (see - ``tool_advisory.make_system_turn``). This event lets every + metacognitive nudges, watch results) lives in the conversation + trajectory as a real ``{"role": "system", "_source": , ...}`` + turn (see ``tool_advisory.make_system_turn``). This event lets every connected SSE consumer (other browser tabs, CLI mirrors, future channel adapters) render the operator bubble in lockstep with the originating tab's optimistic render. The history-replay path surfaces the same shape via ``project_history_messages`` (the REST ``/history`` projection) so a tab reconnecting later renders the same bubble. ``source`` carries the turn's ``_source`` kind so - the frontend can label / style the bubble. + the frontend can label / style the bubble; ``meta`` carries the + turn's structured per-kind fields (``watch_triggered``'s + ``watch_name`` / command / poll counters) so the frontend can rebuild + per-kind rendering (the watch-result card). ``None`` for kinds with + no structured data. """ - self._enqueue({"type": "system_turn", "content": content, "source": source}) + self._enqueue( + {"type": "system_turn", "content": content, "source": source, "meta": meta or None} + ) # ------------------------------------------------------------------ # Broadcast hooks — kind-specific transport. diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index 42a17c41..297c6ed6 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -308,6 +308,7 @@ class PostgreSQLBackend: event_id: int | None = None, is_error: bool = False, producer: str | None = None, + meta: str | None = None, ) -> int: now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") content = sanitize_text(content) @@ -330,6 +331,7 @@ class PostgreSQLBackend: _source=source, event_id=event_id, is_error=is_error, + meta=meta, ) .returning(conversations.c.id) ) @@ -366,6 +368,7 @@ class PostgreSQLBackend: "tool_calls": row.get("tool_calls"), "_source": sanitize_text(row.get("source")), "is_error": bool(row.get("is_error", False)), + "meta": row.get("meta"), } ) with self._conn() as conn: @@ -382,7 +385,7 @@ class PostgreSQLBackend: """Fetch a ws's conversation rows + resolved attachment map (shared by :meth:`load_messages` and :meth:`load_message_turns`). The trailing ``attachments`` ref-list column is split off and is NOT part of the - positional tuple ``reconstruct_*`` unpacks (id..is_error).""" + positional tuple ``reconstruct_*`` unpacks (id..meta).""" _cols = ( conversations.c.id, conversations.c.role, @@ -394,6 +397,7 @@ class PostgreSQLBackend: conversations.c._source, conversations.c.event_id, conversations.c.is_error, + conversations.c.meta, conversations.c.attachments, ) with self._conn() as conn: @@ -412,7 +416,7 @@ class PostgreSQLBackend: .order_by(conversations.c.id) ).fetchall() attachments = self._resolve_row_attachments(rows) - msg_rows = [tuple(r)[:10] for r in rows] + msg_rows = [tuple(r)[:11] for r in rows] return msg_rows, (attachments or None) def load_messages( @@ -438,7 +442,7 @@ class PostgreSQLBackend: attachment_refs: dict[int, list[str]] = {} all_ids: set[str] = set() for r in rows: - ids = _parse_attachment_refs(r[10]) + ids = _parse_attachment_refs(r[11]) if ids: attachment_refs[r[0]] = ids all_ids.update(ids) diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index 1b1279e5..241e8457 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -162,6 +162,7 @@ class StorageBackend(Protocol): event_id: int | None = None, is_error: bool = False, producer: str | None = None, + meta: str | None = None, ) -> int: """Log a message to the conversations table. @@ -178,6 +179,10 @@ class StorageBackend(Protocol): time (``SessionUIBase._event_id``) — the ``Last-Event-ID`` resume cursor space, distinct from the returned ``id`` PK. NULL when the caller has no live UI counter (offline / bulk / fork re-saves). + + ``meta`` is the pre-serialized JSON of an operator-context ``system`` + turn's structured per-kind fields (the ``_source_meta`` side channel); + opaque to storage and NULL for ordinary rows. """ ... @@ -187,8 +192,8 @@ class StorageBackend(Protocol): Each dict must include ``ws_id``, ``role``, and ``content`` (which may be ``None`` for assistant messages with only tool_calls). Optional keys: ``tool_name``, ``tool_call_id``, ``provider_data``, - ``tool_calls``, ``source``. Timestamp and workstream updated-at - are handled internally. + ``tool_calls``, ``source``, ``meta``. Timestamp and workstream + updated-at are handled internally. """ ... diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py index 0a90cec3..4189d720 100644 --- a/turnstone/core/storage/_schema.py +++ b/turnstone/core/storage/_schema.py @@ -65,6 +65,16 @@ conversations = sa.Table( # (JSON; NULL for turns with no attachments) — the sole message->blob link in # the content-addressed model; bytes resolve from workstream_attachments by id. sa.Column("attachments", sa.Text, nullable=True), + # Structured per-kind operator-context metadata for a first-class ``system`` + # turn (JSON object; NULL for ordinary turns and for operator turns with no + # extra fields). The persisted twin of the in-memory + # ``Turn.meta.extra["source_meta"]`` / the ``_source_meta`` side channel: + # ``watch_triggered`` carries ``{watch_name, command, poll_count, max_polls, + # is_final}`` so ``/history`` can rebuild the structured watch-result card; + # other kinds (``user_interjection`` → ``{priority}``) ride generically. + # Stripped before the LLM wire (it is a ``_``-prefixed key by the time it + # reaches a provider). Added in migration 060. + sa.Column("meta", sa.Text, nullable=True), ) sa.Index("idx_conversations_timestamp", conversations.c.timestamp) diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index 7eb9f9ae..4c7513a2 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -342,6 +342,7 @@ class SQLiteBackend: event_id: int | None = None, is_error: bool = False, producer: str | None = None, + meta: str | None = None, ) -> int: now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") content = sanitize_text(content) @@ -364,6 +365,7 @@ class SQLiteBackend: "_source": source, "event_id": event_id, "is_error": is_error, + "meta": meta, }, ) if result.lastrowid is None: @@ -414,6 +416,7 @@ class SQLiteBackend: "tool_calls": row.get("tool_calls"), "_source": sanitize_text(row.get("source")), "is_error": bool(row.get("is_error", False)), + "meta": row.get("meta"), } ) with self._conn() as conn: @@ -442,7 +445,7 @@ class SQLiteBackend: Shared by :meth:`load_messages` (→ dicts, resolved for display) and :meth:`load_message_turns` (→ canonical Turns for resume). The trailing ``attachments`` ref-list column is split off to resolve blobs and is NOT - part of the positional tuple ``reconstruct_*`` unpacks (id..is_error). + part of the positional tuple ``reconstruct_*`` unpacks (id..meta). """ _cols = ( conversations.c.id, @@ -455,6 +458,7 @@ class SQLiteBackend: conversations.c._source, conversations.c.event_id, conversations.c.is_error, + conversations.c.meta, conversations.c.attachments, ) with self._conn() as conn: @@ -477,7 +481,7 @@ class SQLiteBackend: ).fetchall() attachments = self._resolve_row_attachments(rows) - msg_rows = [tuple(r)[:10] for r in rows] + msg_rows = [tuple(r)[:11] for r in rows] return msg_rows, (attachments or None) def load_messages( @@ -506,7 +510,7 @@ class SQLiteBackend: attachment_refs: dict[int, list[str]] = {} all_ids: set[str] = set() for r in rows: - ids = _parse_attachment_refs(r[10]) + ids = _parse_attachment_refs(r[11]) if ids: attachment_refs[r[0]] = ids all_ids.update(ids) diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py index bec2e99c..0fd350aa 100644 --- a/turnstone/core/storage/_utils.py +++ b/turnstone/core/storage/_utils.py @@ -538,15 +538,17 @@ def reconstruct_messages( ) -> list[dict[str, Any]]: """Reconstruct OpenAI message format from stored conversation rows. - Each *row* is an 8- or 9-tuple ``(id, role, content, tool_name, - tool_call_id, provider_data, tool_calls_json, source [, event_id])``, - ordered chronologically by row id. ``source`` is rehydrated as the - ``_source`` side channel. (The legacy ``_reminders`` column that used to - ride here was dropped in migration 060 — operator context lives in - first-class ``system`` turns now.) The optional 9th element - ``event_id`` (migration 059, the per-ws SSE ``Last-Event-ID`` resume - cursor) is surfaced as the ``_event_id`` side-channel; legacy 9-tuple - fixtures omit it (handled by the defensive unpack below). + Each *row* is an 8-to-11-tuple ``(id, role, content, tool_name, + tool_call_id, provider_data, tool_calls_json, source [, event_id [, is_error + [, meta]]])``, ordered chronologically by row id. ``source`` is rehydrated + as the ``_source`` side channel. (The legacy ``_reminders`` column that used + to ride here was dropped in migration 060 — operator context lives in + first-class ``system`` turns now.) The trailing optional elements — + ``event_id`` (migration 059, the per-ws SSE ``Last-Event-ID`` resume cursor → + ``_event_id``), ``is_error`` (migration 060, the persisted tool-result error + flag), and ``meta`` (migration 060, an operator-context turn's structured + ``_source_meta``) — are handled by the defensive unpack so shorter legacy + fixtures stay valid. When ``attachments_by_msg`` is provided (keyed by row id, each value an ordered list of content-addressed attachment rows resolved from the @@ -624,6 +626,24 @@ def _native_from_provider_data(provider_data: str | None) -> ProviderNative | No return None +def _source_meta_from_json(meta_json: str | None) -> dict[str, Any] | None: + """Decode the stored ``meta`` column into an operator-context meta dict. + + The persisted twin of ``Turn.meta.extra["source_meta"]`` — a first-class + ``system`` turn's structured per-kind fields (e.g. ``watch_triggered``'s + ``watch_name`` / ``command`` / poll counters). A decode failure or a + non-object payload yields ``None`` (the meta is dropped — the human-readable + body still lives in ``content``). + """ + if not meta_json: + return None + try: + parsed = json.loads(meta_json) + except (json.JSONDecodeError, TypeError): + return None + return parsed if isinstance(parsed, dict) and parsed else None + + def _tool_calls_from_json(tool_calls_json: str | None) -> tuple[ToolCall, ...]: """Decode the stored ``tool_calls`` column into typed :class:`ToolCall`s.""" if not tool_calls_json: @@ -663,11 +683,15 @@ def reconstruct_turns( for row in rows: (row_id, role, content, _tool_name, tc_id, provider_data, tool_calls_json, source) = row[:8] # event_id (col 9, migration 059) — the per-ws SSE Last-Event-ID cursor; - # is_error (col 10, migration 060) rides last. Defensive length checks - # keep pre-event_id / pre-is_error fixtures valid. + # is_error (col 10, migration 060); meta (col 11, also migration 060 — the + # operator-context per-kind ``source_meta``) rides last. Defensive + # length checks keep pre-event_id / pre-is_error / pre-meta fixtures valid. event_id = int(row[8]) if len(row) > 8 and row[8] is not None else None is_error = bool(row[9]) if len(row) > 9 else False meta = TurnMeta(event_id=event_id) + source_meta = _source_meta_from_json(row[10]) if len(row) > 10 else None + if source_meta is not None: + meta.extra["source_meta"] = source_meta src = str(source) if source else None if role == "user": diff --git a/turnstone/core/storage/migrations/versions/060_unwrap_tool_envelopes.py b/turnstone/core/storage/migrations/versions/060_unwrap_tool_envelopes.py index 9712aa60..4c3c5496 100644 --- a/turnstone/core/storage/migrations/versions/060_unwrap_tool_envelopes.py +++ b/turnstone/core/storage/migrations/versions/060_unwrap_tool_envelopes.py @@ -28,7 +28,10 @@ This migration retires both carriers: This migration also carries the additive front of the canonical-trajectory storage cut: it adds the ``is_error`` column (persisting the tool-result error flag, previously an -in-memory-only message key) and tags legacy bare-list ``provider_data`` rows with their +in-memory-only message key), the ``meta`` column (a first-class ``system`` turn's +structured per-kind operator-context fields — ``watch_triggered``'s ``watch_name`` / +``command`` / poll counters etc. — so ``/history`` rebuilds the structured watch-result +card; additive, no backfill), and tags legacy bare-list ``provider_data`` rows with their generating provider as the ``{producer, blocks}`` envelope (producer inferred from block types — see ``_infer_producer``, which must match the live save's ``provider_name`` values), so the lowering layer can replay the native lane verbatim only to its producer. @@ -271,6 +274,15 @@ def upgrade() -> None: # backfill below (step 3) fills it from the legacy message_id link and # then drops message_id/reserved_* (step 4). batch_op.add_column(sa.Column("attachments", sa.Text, nullable=True)) + # Structured per-kind operator-context metadata for a first-class + # ``system`` turn (JSON object; NULL otherwise) — ``watch_triggered``'s + # ``{watch_name, command, poll_count, max_polls, is_final}`` etc., the + # persisted twin of the in-memory ``_source_meta`` side channel — so + # ``/history`` and the live SSE replay rebuild the structured + # watch-result card. Additive, no backfill (operator turns predating + # this stay plain text bubbles; their structured fields are unrecoverable + # from the flattened ``content``). + batch_op.add_column(sa.Column("meta", sa.Text, nullable=True)) with op.batch_alter_table("workstream_attachments") as batch_op: batch_op.add_column( sa.Column("refcount", sa.Integer, nullable=False, server_default=sa.text("0")) @@ -423,6 +435,7 @@ def downgrade() -> None: batch_op.add_column(sa.Column("_reminders", sa.Text, nullable=True)) batch_op.drop_column("is_error") batch_op.drop_column("attachments") + batch_op.drop_column("meta") with op.batch_alter_table("workstream_attachments") as batch_op: batch_op.add_column(sa.Column("message_id", sa.Integer, nullable=True)) batch_op.add_column(sa.Column("reserved_for_msg_id", sa.Text, nullable=True)) diff --git a/turnstone/core/tool_advisory.py b/turnstone/core/tool_advisory.py index b34cb24f..016165fd 100644 --- a/turnstone/core/tool_advisory.py +++ b/turnstone/core/tool_advisory.py @@ -54,6 +54,29 @@ _USER_INTERJECTION_IMPORTANT_PREAMBLE: Final = ( _USER_INTERJECTION_BODY_MARKER: Final = "\n\nUser message: " +def render_output_guard_text(meta: dict[str, Any]) -> str: + """Render an ``output_guard`` system turn's text from its structured *meta*. + + *meta* carries ``{flags, risk_level, annotations, redacted}``. Returns the + operator/model-facing prose — the flag list + risk level, one indented line + per annotation, and (when credentials were redacted) the do-not-reconstruct + notice. This is the text projection of the structured guard finding: the + producer (``ChatSession._collect_advisories``) builds *meta* and derives the + turn ``content`` from it via this function, so the wire text and the FE + guard-finding card cannot drift (both read the same *meta*). + """ + flags = meta.get("flags") or [] + risk_level = str(meta.get("risk_level") or "none") + lines = [f"Output guard: {', '.join(flags)} ({risk_level.upper()})"] + for ann in meta.get("annotations") or []: + lines.append(f" {ann}") + if meta.get("redacted"): + lines.append( + "Credentials have been redacted. Do not attempt to reconstruct redacted values." + ) + return "\n".join(lines) + + def render_user_interjection(message: str, priority: str) -> str: """Frame a queued user *message* as user-authored operator context. @@ -108,14 +131,22 @@ SYSTEM_TURN_SOURCES: Final = frozenset( def make_system_turn(source: str, content: str, **meta: Any) -> dict[str, Any]: """Build a first-class operator-context system turn for ``self.messages``. - Returns ``{"role": "system", "_source": source, "content": content}``. - *source* must be one of :data:`SYSTEM_TURN_SOURCES`. Extra keyword *meta* - fields are attached as leading-underscore sibling keys (e.g. - ``watch_name="ci"`` → ``_watch_name``) so they ride the persisted record - and the UI projection but are stripped before the LLM wire by - ``sanitize_messages``. Keys already underscore-prefixed are kept as-is; a - meta key that normalises onto a reserved field (``_source``) is rejected so - the validated source can't be silently clobbered. + Returns ``{"role": "system", "_source": source, "content": content}``, plus a + ``_source_meta`` dict of the *meta* keyword fields when any are supplied. + *source* must be one of :data:`SYSTEM_TURN_SOURCES`. + + The *meta* fields are the turn's structured per-kind data (e.g. + ``watch_triggered``'s ``watch_name`` / ``command`` / ``poll_count`` / + ``max_polls`` / ``is_final``; ``user_interjection``'s ``priority``). They + ride as ONE ``_source_meta`` dict — a single carrier that maps cleanly to the + one ``conversations.meta`` storage column, the one ``Turn.meta.extra + ["source_meta"]`` field (via :func:`turnstone.core.trajectory.turn_from_dict`), + and the one ``meta`` field the live ``on_system_turn`` SSE event / the + ``/history`` projection hand the frontend so it can rebuild per-kind rendering + (the ``watch_triggered`` card). Being a ``_``-prefixed key, ``_source_meta`` + is stripped before the LLM wire by ``sanitize_messages`` (and the Anthropic + native mid-conversation path copies only ``role`` + ``content``), so the + structured fields never reach the model. ``content`` is stored and — on the native mid-conversation-system path (claude-opus-4-8) — sent to the model verbatim, so fence-escaping is NOT @@ -128,9 +159,6 @@ def make_system_turn(source: str, content: str, **meta: Any) -> dict[str, Any]: if source not in SYSTEM_TURN_SOURCES: raise ValueError(f"unknown system-turn source: {source!r}") turn: dict[str, Any] = {"role": "system", "_source": source, "content": content} - for key, value in meta.items(): - norm = key if key.startswith("_") else f"_{key}" - if norm in turn: - raise ValueError(f"system-turn meta key {key!r} collides with reserved {norm!r}") - turn[norm] = value + if meta: + turn["_source_meta"] = dict(meta) return turn diff --git a/turnstone/core/trajectory.py b/turnstone/core/trajectory.py index a2675b6e..c501a6c4 100644 --- a/turnstone/core/trajectory.py +++ b/turnstone/core/trajectory.py @@ -89,7 +89,11 @@ class TurnMeta: """Sidecar metadata: never reaches the wire, never read by the lowering layer. ``event_id`` is the per-ws SSE ``Last-Event-ID`` resume cursor; ``extra`` holds - open operator-turn metadata (e.g. ``watch_name``).""" + open metadata under well-known keys — ``"source_meta"`` (an operator-context + ``system`` turn's structured per-kind fields, e.g. ``watch_triggered``'s + ``watch_name`` / ``command`` / poll counters; persisted in the + ``conversations.meta`` column, surfaced to the FE for per-kind rendering) and + ``"attachments_meta"`` (display metadata for by-reference attachments).""" event_id: int | None = None extra: dict[str, Any] = field(default_factory=dict) @@ -240,6 +244,12 @@ def turn_from_dict(msg: dict[str, Any]) -> Turn: am = msg.get("_attachments_meta") if am is not None: meta.extra["attachments_meta"] = am + # Operator-context per-kind structured fields (``watch_triggered`` etc.). + # Carried as ONE dict so it maps to one ``conversations.meta`` column / one + # FE ``meta`` field, rather than scattered ``_``-prefixed siblings. + sm = msg.get("_source_meta") + if sm: + meta.extra["source_meta"] = sm return Turn( role=role, @@ -280,6 +290,9 @@ def turn_to_dict(turn: Turn) -> dict[str, Any]: am = turn.meta.extra.get("attachments_meta") if am is not None: msg["_attachments_meta"] = am + sm = turn.meta.extra.get("source_meta") + if sm: + msg["_source_meta"] = sm return msg diff --git a/turnstone/core/watch.py b/turnstone/core/watch.py index 5895bfd4..8f6f2f2d 100644 --- a/turnstone/core/watch.py +++ b/turnstone/core/watch.py @@ -195,9 +195,15 @@ def format_watch_message( return "\n".join(lines) +# The structured fields that ride the ``watch_triggered`` system turn's +# ``_source_meta`` (delivered to the FE for the watch-result card). ``output`` +# is the raw (sanitized) shell output so the card body renders it alone, without +# re-showing the header / command lines that ``format_watch_message`` bakes into +# the turn's text ``content`` (which is what the model reads on the wire). WATCH_REMINDER_OPTIONAL_KEYS = ( "watch_name", "command", + "output", "poll_count", "max_polls", "is_final", @@ -217,13 +223,18 @@ def build_watch_reminder( ) -> dict[str, Any]: """Build a structured ``watch_triggered`` reminder dict. - Returns a dict with ``{type, text, watch_name, command, poll_count, - max_polls, is_final}``. The ``text`` field is the formatted body + Returns a dict with ``{type, text, watch_name, command, output, + poll_count, max_polls, is_final}``. The ``text`` field is the formatted body (same content :func:`format_watch_message` produces) and becomes the - content of the first-class ``{"role": "system", "_source": - "watch_triggered"}`` turn the drain seam emits; the remaining fields - ride as sibling metadata. Compaction / channel adapters keep seeing - the human-readable shell output via the ``text`` field. + ``content`` of the first-class ``{"role": "system", "_source": + "watch_triggered"}`` turn the drain seam emits — the model-facing prose, with + the watch header / ``$ command`` / output all baked in. The remaining fields + ride as the turn's structured ``_source_meta`` so the FE rebuilds the + watch-result card from them; ``output`` is carried separately so the card + body shows the raw shell output alone (without re-printing the header / + command the chrome already renders). Both ``text`` and the structured fields + derive from the same inputs, so they cannot drift. Compaction / channel + adapters keep seeing the human-readable shell output via the ``text`` field. """ return { "type": "watch_triggered", @@ -240,6 +251,7 @@ def build_watch_reminder( ), "watch_name": name, "command": command, + "output": output, "poll_count": poll_count, "max_polls": max_polls, "is_final": is_final, diff --git a/turnstone/eval.py b/turnstone/eval.py index ca340b8a..1bf61797 100644 --- a/turnstone/eval.py +++ b/turnstone/eval.py @@ -143,7 +143,7 @@ class NullUI: def on_error(self, message: str) -> None: pass - def on_system_turn(self, content: str, source: str) -> None: + def on_system_turn(self, content: str, source: str, meta: dict[str, Any] | None = None) -> None: pass def on_state_change(self, state: str) -> None: diff --git a/turnstone/sdk/events.py b/turnstone/sdk/events.py index d3a7b03a..ee78c7fd 100644 --- a/turnstone/sdk/events.py +++ b/turnstone/sdk/events.py @@ -62,6 +62,11 @@ class HistoryEvent(ServerEvent): * ``source`` — the operator-context kind on a ``system`` turn (``output_guard`` / ``user_interjection`` / ``tool_error`` / ...), or ``"system_nudge"`` on a wake-driven empty user turn + * ``meta`` — structured per-kind fields on an operator-context + ``system`` turn (e.g. ``watch_triggered``'s ``{watch_name, command, + poll_count, max_polls, is_final}``) so the renderer can rebuild + per-kind UI (the watch-result card); absent when the kind carries no + structured data * ``reasoning`` — concatenated reasoning text for assistant turns that round-tripped a thinking-block lane (Anthropic-with-thinking today; OpenAI Responses + Gemini in later phases). Present only diff --git a/turnstone/shared_static/chat.css b/turnstone/shared_static/chat.css index ed43108c..a2f2dea8 100644 --- a/turnstone/shared_static/chat.css +++ b/turnstone/shared_static/chat.css @@ -1003,11 +1003,13 @@ } /* First-class operator-context system turn (role="system" in the trajectory) — the consolidation of the metacognition reminder / - user-interjection / output-guard / watch-result bubbles into one turn - type that FOLLOWS the turn it advises. Shared so the interactive UI and - the console coord viewer render it identically. Yellow accent reads as - "operator metadata" against the amber user colour and the cyan tool - cards; deliberately quieter so it doesn't compete for attention. */ + user-interjection / output-guard bubbles into one turn type that FOLLOWS + the turn it advises. Shared so the interactive UI and the console coord + viewer render it identically. Yellow accent reads as "operator metadata" + against the amber user colour and the cyan tool cards; deliberately quieter + so it doesn't compete for attention. The ``watch_triggered`` kind carries + structured per-kind meta and branches into the richer ``.msg.watch-result`` + card below instead of this plain bubble. */ .msg.system-context { border-left-color: var(--yellow); color: var(--fg-dim); @@ -1025,6 +1027,107 @@ white-space: pre-wrap; } +/* Structured watch-result card — the ``watch_triggered`` operator-context + system turn renders here (command-preview header / shell-output body / poll + counter footer) rather than as the plain ``.msg.system-context`` bubble, + because it carries structured per-kind meta. Cyan accent ties it to the + tool-surface vocabulary (it IS shell output). Built by + ``_buildWatchResultBubble`` (interactive) / ``appendWatchResult`` (coord). */ +.msg.watch-result { + border-left-color: var(--cyan); + background: var(--panel); + padding: 8px 12px; + margin: 6px 0; + width: 100%; +} +.msg.watch-result .msg-watch-header { + color: var(--cyan); + font-family: var(--font-mono); + font-size: 11px; + font-weight: 600; + margin-bottom: 4px; + text-transform: lowercase; +} +.msg.watch-result .msg-watch-cmd { + color: var(--ink-3); + font-family: var(--font-mono); + font-size: 12px; + margin-bottom: 6px; +} +.msg.watch-result .msg-watch-body { + font-family: var(--font-mono); + font-size: 12px; + margin: 0; + white-space: pre-wrap; + /* Shell output can be wide (URLs, JSON, file lists) — break inside + long words so the off-canvas mobile drawer doesn't blow horizontal + layout. */ + word-break: break-word; +} +.msg.watch-result .msg-watch-footer { + color: var(--ink-3); + font-size: 11px; + margin-top: 6px; +} + +/* Guard-finding card — the ``output_guard`` operator-context system turn. The + inner warning element (``.output-warning`` interactive / ``.coord-tool-row- + warning`` coord) carries the risk colour + flags + redaction; this wrapper + hosts it as a standalone message row and styles the annotation detail lines + the terse inline tool chip omits. Built by ``_buildGuardFindingBubble`` + (interactive) / ``appendGuardFinding`` (coord). */ +.msg.guard-finding { + border-left-color: var(--yellow); + padding: 6px 10px; +} +.msg.guard-finding .msg-guard-annotation { + color: var(--ink-3); + font-size: 11px; + margin-top: 4px; + margin-left: 8px; + white-space: pre-wrap; + word-break: break-word; +} + +/* Idle-children card — the coordinator-only ``idle_children`` operator-context + system turn lists the child workstreams still running while the coordinator + went idle. Built by ``appendIdleChildren`` (coord). */ +.msg.idle-children { + border-left-color: var(--yellow); + padding: 6px 10px; +} +.msg.idle-children .msg-idle-header { + color: var(--yellow); + font-weight: 600; + font-size: 11px; + text-transform: lowercase; + margin-bottom: 4px; +} +.msg.idle-children .msg-idle-list { + margin: 0; + padding-left: 16px; +} +.msg.idle-children .msg-idle-child { + font-size: 12px; + color: var(--ink-3); +} +.msg.idle-children .msg-idle-child-state { + color: var(--ink-4); + margin-left: 6px; + font-size: 11px; +} + +/* Important user interjection — a ``!!!``-prefixed queued message that drained + mid-turn. Brighter accent + heavier weight so an urgent operator interjection + doesn't read as a quiet metadata note. ``notice`` interjections keep the + default ``.system-context`` styling. */ +.msg.system-context.important { + border-left-color: var(--accent); +} +.msg.system-context.important .msg-system-context-label { + color: var(--accent); +} + /* System-nudge marker — visible-but-thin replacement for the wake's synthetic empty user turn, so the operator-context `system` turns that follow it have a visible anchor instead of trailing an old user message. diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js index c244121d..9fa4b032 100644 --- a/turnstone/ui/static/app.js +++ b/turnstone/ui/static/app.js @@ -244,24 +244,49 @@ class Pane { return el; } - addSystemContext(content, source) { + addSystemContext(content, source, meta) { // First-class operator-context system turn — the consolidation of the // legacy metacognition reminders / user interjections / output-guard // notes into one role="system" trajectory turn. Rendered as a distinct // operator bubble in sequence (it FOLLOWS the turn it advises); // `source` carries the kind (user_interjection / output_guard / - // tool_error / ...) for the bubble label. + // tool_error / ...) for the bubble label. `watch_triggered` additionally + // carries structured `meta` (watch_name / command / poll counters) → the + // richer `.msg.watch-result` card instead of the plain operator bubble. this.removeEmptyState(); + if (source === "watch_triggered" && meta && typeof meta === "object") { + const card = _buildWatchResultBubble(meta, content || ""); + this.messagesEl.appendChild(card); + this.scrollToBottom(true); + return card; + } + if (source === "output_guard" && meta && typeof meta === "object") { + const card = _buildGuardFindingBubble(meta); + this.messagesEl.appendChild(card); + this.scrollToBottom(true); + return card; + } + // user_interjection renders as a "queued message" bubble showing the user's + // RAW words (`meta.message`) rather than the model-directed framing in + // `content`, with brighter emphasis for `!!!`-important interjections. + const isInterjection = + source === "user_interjection" && meta && typeof meta === "object"; + const important = isInterjection && meta.priority === "important"; const el = document.createElement("div"); - el.className = "msg system-context"; + el.className = "msg system-context" + (important ? " important" : ""); const body = document.createElement("div"); body.className = "msg-body"; const labelEl = document.createElement("span"); labelEl.className = "msg-system-context-label"; - labelEl.textContent = "operator" + (source ? " · " + String(source) : ""); + labelEl.textContent = isInterjection + ? "queued message" + (important ? " · important" : "") + : "operator" + (source ? " · " + String(source) : ""); const textEl = document.createElement("span"); textEl.className = "msg-system-context-text"; - textEl.textContent = content || ""; + textEl.textContent = + isInterjection && meta.message != null + ? String(meta.message) + : content || ""; body.appendChild(labelEl); body.appendChild(textEl); el.appendChild(body); @@ -1181,8 +1206,13 @@ class Pane { // user_reminder / tool_reminder events into one operator bubble // rendered in trajectory sequence (it FOLLOWS the turn it advises, // so by the time this SSE event arrives the related turn already - // rendered). ``evt.source`` carries the kind for the bubble label. - this.addSystemContext(evt.content || "", evt.source || ""); + // rendered). ``evt.source`` carries the kind for the bubble label; + // ``evt.meta`` the structured per-kind fields (watch-result card). + this.addSystemContext( + evt.content || "", + evt.source || "", + evt.meta || null, + ); break; case "message_queued": @@ -2124,8 +2154,13 @@ class Pane { // interjection, metacognitive nudge — see make_system_turn). These // now FOLLOW the turn they advise as their own rows; tool-channel // nudges and queued interjections that used to splice into the tool - // result render here in sequence. `source` is the kind. - this.addSystemContext(msg.content || "", msg.source || ""); + // result render here in sequence. `source` is the kind; `meta` the + // structured per-kind fields (watch-result card) from /history. + this.addSystemContext( + msg.content || "", + msg.source || "", + msg.meta || null, + ); lastToolBlock = null; } } @@ -2712,6 +2747,79 @@ class Pane { } } +// Build a structured ``.msg.watch-result`` card for a ``watch_triggered`` +// operator-context system turn — command-preview header + shell-output body + +// poll-counter footer. ``content`` is the formatted watch body (the system +// turn's content); ``meta`` carries the structured fields (``watch_name`` / +// ``command`` / ``poll_count`` / ``max_polls`` / ``is_final``) delivered live on +// the ``system_turn`` SSE event and on the ``/history`` projection. All text +// goes through ``textContent`` so shell output containing angle brackets / +// scripts / steering bytes renders inertly. Mirrors the coordinator pane's +// buildWatchResultBubble. +function _buildWatchResultBubble(meta, content) { + const el = document.createElement("div"); + el.className = "msg watch-result"; + el.setAttribute("role", "article"); + el.setAttribute("data-ts-role", "watch"); + el.setAttribute("aria-label", "watch"); + const header = document.createElement("div"); + header.className = "msg-watch-header"; + header.textContent = + "watch" + (meta.watch_name ? " · " + String(meta.watch_name) : ""); + el.appendChild(header); + if (meta.command) { + const cmd = document.createElement("div"); + cmd.className = "msg-watch-cmd"; + cmd.textContent = "$ " + String(meta.command); + el.appendChild(cmd); + } + const body = document.createElement("pre"); + body.className = "msg-watch-body"; + // Prefer the structured ``output`` (raw shell output alone) so the body + // doesn't re-print the header / ``$ command`` lines the chrome already shows; + // fall back to the full turn ``content`` for legacy turns that predate the + // ``output`` meta field (migration 060 is additive — no backfill). + body.textContent = meta.output != null ? String(meta.output) : content || ""; + el.appendChild(body); + if (meta.poll_count != null && meta.max_polls != null) { + const footer = document.createElement("div"); + footer.className = "msg-watch-footer"; + const finalSuffix = meta.is_final ? " · final" : ""; + footer.textContent = + "poll " + + String(meta.poll_count) + + "/" + + String(meta.max_polls) + + finalSuffix; + el.appendChild(footer); + } + return el; +} + +// Build a structured ``.msg.guard-finding`` card for an ``output_guard`` +// operator-context system turn. ``meta`` carries the structured finding +// ``{flags, risk_level, annotations, redacted}``. Reuses the tool-result +// warning chip (``_buildOutputWarningEl``) for the risk / flags / redaction +// header so the operator-context finding speaks the same visual vocabulary, +// then appends the annotations (matched-pattern detail) — which the inline +// tool chip omits to stay terse. All text via textContent. +function _buildGuardFindingBubble(meta) { + const el = document.createElement("div"); + el.className = "msg guard-finding"; + el.setAttribute("role", "article"); + el.setAttribute("data-ts-role", "output_guard"); + el.setAttribute("aria-label", "output guard"); + el.appendChild(_buildOutputWarningEl(meta)); + const anns = Array.isArray(meta.annotations) ? meta.annotations : []; + for (let i = 0; i < anns.length; i++) { + const a = document.createElement("div"); + a.className = "msg-guard-annotation"; + a.textContent = String(anns[i]); + el.appendChild(a); + } + return el; +} + // Shared output-warning DOM builder — used by both replayHistory // (saved-workstream rendering) and the live appendToolOutput path // via showOutputWarning. Single source of truth keeps the two