feat(operator-context): deliver structured per-kind meta to the UI

Operator-context system turns (watch results, output-guard findings, idle
children, user interjections) carried their kind (_source) and a flattened
text content, but the structured per-kind fields were dropped at every
persist/deliver boundary — so the UI rendered every kind as one generic
operator bubble and the structured watch-result card was lost.

Wire the structured meta through as the single source of truth:

- Storage: new conversations.meta JSON column (migration 060); threaded
  through save_message/save_messages_bulk (facade + protocol + both backends)
  and rehydrated in reconstruct_turns onto Turn.meta.extra["source_meta"].
- Canonical: make_system_turn carries meta as one _source_meta dict;
  turn_from_dict/turn_to_dict bridge it to/from Turn.meta.extra.
- Live + history: widen on_system_turn(content, source, meta) across all
  impls + the SSE payload; surface _source_meta -> meta in the /history
  projection. SDK HistoryEvent docs note the field.
- Producers derive both the model-facing content text AND the card from one
  meta dict, so they cannot drift: render_output_guard_text, build_watch_
  reminder carrying output, idle_children and user_interjection metadata.
- Frontend: addSystemContext / renderSystemTurn dispatch by source to the
  watch-result, guard-finding, idle-children, and queued-message cards in
  both the interactive and coordinator panes; every untrusted field renders
  via textContent.

The meta is a leading-underscore key, stripped before the wire (sanitize_
messages and the native mid-conversation path copy only role+content), so the
per-provider wire payloads stay byte-identical. Additive column, no backfill:
operator turns predating it reload as plain text bubbles.

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;

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."
     )

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

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."""

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:

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

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

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
 

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."""
 

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

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

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")

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, {})

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 });

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

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)

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

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": <kind>, ...}`` 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": <kind>, ...}``
+        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.

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)

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.
         """
         ...
 

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)

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)

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":

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))

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

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
 
 

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,

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:

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

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.

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