feat(storage): persist _source + _reminders side-channels on conversations

Adds two TEXT-NULL columns to the conversations table so multi-tab /
multi-device replay sees the same metacognitive bubble shape the
originating tab saw live.  Until now, reminders lived only on the
in-memory ChatSession.messages dict, and the wake-driven empty user
turn was not persisted at all (skip at session.py:2685-2686) — a
second tab connecting via /history saw the assistant turn with no
preceding wake context, and missed every other tab's reminder
bubbles besides.

Single Alembic revision 050 (head was 049) adds:
  * conversations._source — today only "system_nudge" for wake rows
  * conversations._reminders — JSON-encoded reminder list

Both backends (sqlite + postgresql) thread the columns through
save_message / save_messages_bulk / load_messages.  reconstruct_messages
unpacks the row tuple as 9 elements (was 7), JSON-decoding _reminders
on the user AND tool branches with the same contextlib.suppress guard
the existing provider_data / tool_calls decode uses.  Tool-row
reminders ride the same column so tool_error / repeat replay shape
matches user-channel parity.

session.py:2685-2686 wake-row persist skip is dropped; _append_user_turn
JSON-encodes user_msg["_reminders"] and passes both source + reminders
to save_message.  The tool-message save site at session.py:3014-3020
mirrors with metacog_reminders.

Plan reference: docs/design/watch-card-ux.md §4 Steps 1-5 (Commit 1).

(cherry picked from commit f64c3e7b10)

tests/test_reconstruct_messages.py

@@ -15,9 +15,26 @@ def _row(
     tc_id=None,
     pdata=None,
     tool_calls=None,
+    source=None,
+    reminders=None,
 ):
-    """Build a 7-element conversation row tuple (id, role, ...)."""
-    return (next(_row_ids), role, content, tool_name, tc_id, pdata, tool_calls)
+    """Build a 9-element conversation row tuple (id, role, ...).
+
+    Trailing ``source`` / ``reminders`` mirror the persisted twins of
+    the in-memory ``_source`` / ``_reminders`` side-channels added in
+    migration 050.
+    """
+    return (
+        next(_row_ids),
+        role,
+        content,
+        tool_name,
+        tc_id,
+        pdata,
+        tool_calls,
+        source,
+        reminders,
+    )
 
 
 class TestAssistantWithToolCalls:

tests/test_session.py

@@ -3187,6 +3187,72 @@ class TestDeliverWakeNudge:
         # Wake tag cleared even on exception (finally block).
         assert session._wake_source_tag == ""
 
+    def test_wake_row_persists_with_source_column(self, tmp_db):
+        """The wake's synthesised empty user turn now persists with
+        ``_source = "system_nudge"`` (post-#484 the skip at
+        ``session.py:2685-2686`` is dropped).  Without persistence,
+        a second tab connecting via /history would see the assistant
+        turn with no preceding wake context.
+        """
+        from turnstone.core.storage import get_storage
+
+        session = _make_session()
+        session._title_generated = True
+        session._queue_user_advisory("denial", "leftover")
+        with (
+            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
+            patch.object(
+                session,
+                "_stream_response",
+                return_value={"role": "assistant", "content": "ok"},
+            ),
+            patch.object(session, "_full_messages", return_value=[]),
+            patch.object(session, "_update_token_table"),
+            patch.object(session, "_print_status_line"),
+            patch.object(session, "_emit_state"),
+            patch.object(session, "_visible_memory_count", return_value=0),
+        ):
+            session.deliver_wake_nudge_from_queue()
+        rows = get_storage().load_messages(session._ws_id)
+        wake_rows = [
+            r for r in rows if r.get("role") == "user" and r.get("_source") == "system_nudge"
+        ]
+        assert len(wake_rows) == 1
+        assert wake_rows[0]["content"] == ""
+
+    def test_user_reminder_persists_with_widened_payload(self, tmp_db):
+        """User-channel reminders attached to the wake's synthetic
+        empty turn round-trip through storage including their full
+        payload (the ``denial`` text here; later steps add optional
+        fields like ``watch_name`` for ``watch_triggered``).
+        """
+        from turnstone.core.storage import get_storage
+
+        session = _make_session()
+        session._title_generated = True
+        session._queue_user_advisory("denial", "do not do that")
+        with (
+            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
+            patch.object(
+                session,
+                "_stream_response",
+                return_value={"role": "assistant", "content": "ok"},
+            ),
+            patch.object(session, "_full_messages", return_value=[]),
+            patch.object(session, "_update_token_table"),
+            patch.object(session, "_print_status_line"),
+            patch.object(session, "_emit_state"),
+            patch.object(session, "_visible_memory_count", return_value=0),
+        ):
+            session.deliver_wake_nudge_from_queue()
+        rows = get_storage().load_messages(session._ws_id)
+        wake_rows = [
+            r for r in rows if r.get("role") == "user" and r.get("_source") == "system_nudge"
+        ]
+        assert len(wake_rows) == 1
+        reminders = wake_rows[0].get("_reminders")
+        assert reminders == [{"type": "denial", "text": "do not do that"}]
+
 
 class TestReminderSidechannelIsolation:
     """The side-channel design's load-bearing guarantee: any reader of

tests/test_storage_source_reminders.py

@@ -0,0 +1,129 @@
+"""Tests for ``_source`` / ``_reminders`` round-tripping through both
+storage backends.
+
+Persisting the in-memory side-channels lets multi-tab / multi-device
+replay show the same metacognitive bubble shape the originating tab
+saw live — see ``docs/design/watch-card-ux.md`` §1.
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+
+from turnstone.core.storage._schema import conversations
+
+
+class TestSourceRoundtrip:
+    def test_source_roundtrip(self, backend):
+        backend.register_workstream("s1")
+        backend.save_message("s1", "user", "", source="system_nudge")
+        msgs = backend.load_messages("s1")
+        assert len(msgs) == 1
+        assert msgs[0]["role"] == "user"
+        assert msgs[0]["content"] == ""
+        assert msgs[0].get("_source") == "system_nudge"
+
+    def test_source_absent_when_not_set(self, backend):
+        backend.register_workstream("s1")
+        backend.save_message("s1", "user", "hello")
+        msgs = backend.load_messages("s1")
+        assert "_source" not in msgs[0]
+
+
+class TestRemindersRoundtrip:
+    def test_reminders_roundtrip(self, backend):
+        backend.register_workstream("s1")
+        payload = [
+            {
+                "type": "watch_triggered",
+                "text": "$ ls\nfile.txt\n",
+                "watch_name": "w1",
+                "command": "ls",
+                "poll_count": 2,
+                "max_polls": 100,
+                "is_final": False,
+            }
+        ]
+        backend.save_message(
+            "s1",
+            "user",
+            "",
+            source="system_nudge",
+            reminders=json.dumps(payload, separators=(",", ":")),
+        )
+        msgs = backend.load_messages("s1")
+        assert msgs[0].get("_reminders") == payload
+        # Optional fields preserved verbatim.
+        rem = msgs[0]["_reminders"][0]
+        assert rem["watch_name"] == "w1"
+        assert rem["command"] == "ls"
+        assert rem["poll_count"] == 2
+        assert rem["max_polls"] == 100
+        assert rem["is_final"] is False
+
+    def test_reminders_null_renders_as_no_key(self, backend):
+        """Absent vs. empty-list should map to the same shape on the
+        load side: ``_reminders`` simply not present in the dict.
+        Mirrors the ``_attachments_meta`` precedent in
+        ``reconstruct_messages``.
+        """
+        backend.register_workstream("s1")
+        backend.save_message("s1", "user", "hello")
+        msgs = backend.load_messages("s1")
+        assert "_reminders" not in msgs[0]
+
+    def test_tool_reminders_roundtrip(self, backend):
+        backend.register_workstream("s1")
+        # Build a minimal valid history: assistant turn with one
+        # tool_call followed by the tool result that carries the
+        # tool-channel reminder.  Without the assistant turn the
+        # tool row would be orphaned and stripped by the repair pass.
+        tc_json = json.dumps(
+            [
+                {
+                    "id": "c1",
+                    "type": "function",
+                    "function": {"name": "bash", "arguments": "{}"},
+                }
+            ]
+        )
+        backend.save_message("s1", "user", "go")
+        backend.save_message("s1", "assistant", None, tool_calls=tc_json)
+        payload = [{"type": "tool_error", "text": "command failed"}]
+        backend.save_message(
+            "s1",
+            "tool",
+            "boom",
+            tool_call_id="c1",
+            reminders=json.dumps(payload, separators=(",", ":")),
+        )
+        msgs = backend.load_messages("s1")
+        # Find the tool message and assert reminders survived load.
+        tool_msgs = [m for m in msgs if m.get("role") == "tool"]
+        assert len(tool_msgs) == 1
+        assert tool_msgs[0].get("_reminders") == payload
+
+    def test_malformed_reminders_json_does_not_crash_load(self, backend):
+        """A garbage string in the column must not abort the whole
+        load — mirrors the ``provider_data`` JSON-decode-suppress
+        pattern.  Concretely: write a row with valid columns BUT a
+        corrupted ``_reminders`` value via raw SQL, then verify the
+        load returns the message with no ``_reminders`` key (rather
+        than raising or surfacing the garbage).
+        """
+        backend.register_workstream("s1")
+        msg_id = backend.save_message("s1", "user", "hello")
+        with backend._engine.connect() as conn:
+            conn.execute(
+                sa.update(conversations)
+                .where(conversations.c.id == msg_id)
+                .values(_reminders="this is not json {{")
+            )
+            conn.commit()
+        msgs = backend.load_messages("s1")
+        assert len(msgs) == 1
+        # Garbage suppressed silently — key absent, content intact.
+        assert "_reminders" not in msgs[0]
+        assert msgs[0]["content"] == "hello"

turnstone/core/memory.py

@@ -41,11 +41,18 @@ def save_message(
     tool_call_id: str | None = None,
     provider_data: str | None = None,
     tool_calls: str | None = None,
+    source: str | None = None,
+    reminders: str | None = None,
 ) -> int:
     """Log a message to the conversations table.
 
     Returns the inserted row id, or ``0`` on failure (preserving the
     module's no-raise contract).
+
+    ``source`` / ``reminders`` mirror the in-memory ``_source`` /
+    ``_reminders`` side-channels (``reminders`` JSON-encoded).  Both
+    default to ``None`` for the common case where no metacog payload
+    rides the row.
     """
     try:
         return get_storage().save_message(
@@ -56,6 +63,8 @@ def save_message(
             tool_call_id,
             provider_data,
             tool_calls=tool_calls,
+            source=source,
+            reminders=reminders,
         )
     except Exception:
         log.warning("Failed to save message for ws=%s role=%s", ws_id, role, exc_info=True)

turnstone/core/session.py

@@ -2692,15 +2692,26 @@ class ChatSession:
         # leaves pending rows that the UI's chip rehydration can still
         # surface so the user can clear or resend them.
         #
-        # Skip the persist for the wake's synthesized empty turn — the
-        # row would carry no content (the system-reminder lives on the
-        # ``_reminders`` sibling, stripped before persist) and the
-        # ``_source`` audit tag isn't column-backed.  Replays can
-        # re-derive the wake event from the conversation flow without
-        # this empty row.
-        if from_wake and not user_input and not attachments:
-            return 0
-        message_id = save_message(self._ws_id, "user", user_input)
+        # The wake's synthesised empty turn DOES persist now: the
+        # ``_source`` and ``_reminders`` columns mirror the in-memory
+        # side-channels so a tab reconnecting via /history sees the
+        # same system-nudge marker + reminder bubbles the originating
+        # tab rendered live.  Without persistence, multi-tab / multi-
+        # device replay shows the assistant's response with no
+        # preceding context — the wake event looks like it came out of
+        # nowhere.
+        source = user_msg.get("_source")
+        reminders_payload = user_msg.get("_reminders")
+        reminders_json = (
+            json.dumps(reminders_payload, separators=(",", ":")) if reminders_payload else None
+        )
+        message_id = save_message(
+            self._ws_id,
+            "user",
+            user_input,
+            source=source if isinstance(source, str) and source else None,
+            reminders=reminders_json,
+        )
         if attachments and message_id:
             mark_attachments_consumed(
                 [a.attachment_id for a in attachments],
@@ -3027,12 +3038,23 @@ class ChatSession:
                         )[:TOOL_RESULT_STORAGE_CAP]
                     else:
                         store_text = raw_output[:TOOL_RESULT_STORAGE_CAP]
+                    # Mirror the user-channel persistence: tool-channel
+                    # reminders (``tool_error`` / ``repeat``) ride the same
+                    # ``_reminders`` JSON column so a tab reconnecting via
+                    # /history sees the below-the-tool bubble the
+                    # originating tab rendered live.
+                    tool_reminders_json = (
+                        json.dumps(metacog_reminders, separators=(",", ":"))
+                        if metacog_reminders
+                        else None
+                    )
                     save_message(
                         self._ws_id,
                         "tool",
                         store_text,
                         _tname,
                         tool_call_id=tc_id,
+                        reminders=tool_reminders_json,
                     )
                 # Inject user feedback from approval prompt (e.g. "y, use full path")
                 if user_feedback:

turnstone/core/storage/_postgresql.py

@@ -178,6 +178,8 @@ class PostgreSQLBackend:
         tool_call_id: str | None = None,
         provider_data: str | None = None,
         tool_calls: str | None = None,
+        source: str | None = None,
+        reminders: str | None = None,
     ) -> int:
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         content = sanitize_text(content)
@@ -194,6 +196,8 @@ class PostgreSQLBackend:
                     tool_call_id=tool_call_id,
                     provider_data=provider_data,
                     tool_calls=tool_calls,
+                    _source=source,
+                    _reminders=reminders,
                 )
                 .returning(conversations.c.id)
             )
@@ -223,6 +227,8 @@ class PostgreSQLBackend:
                     "tool_call_id": row.get("tool_call_id"),
                     "provider_data": sanitize_text(row.get("provider_data")),
                     "tool_calls": row.get("tool_calls"),
+                    "_source": row.get("source"),
+                    "_reminders": row.get("reminders"),
                 }
             )
         with self._conn() as conn:
@@ -245,6 +251,8 @@ class PostgreSQLBackend:
                         conversations.c.tool_call_id,
                         conversations.c.provider_data,
                         conversations.c.tool_calls,
+                        conversations.c._source,
+                        conversations.c._reminders,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id.desc())
@@ -261,6 +269,8 @@ class PostgreSQLBackend:
                         conversations.c.tool_call_id,
                         conversations.c.provider_data,
                         conversations.c.tool_calls,
+                        conversations.c._source,
+                        conversations.c._reminders,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id)

turnstone/core/storage/_protocol.py

@@ -115,12 +115,20 @@ class StorageBackend(Protocol):
         tool_call_id: str | None = None,
         provider_data: str | None = None,
         tool_calls: str | None = None,
+        source: str | None = None,
+        reminders: str | None = None,
     ) -> int:
         """Log a message to the conversations table.
 
         Returns the inserted row's ``id`` (autoincrement PK).  Callers
         that need to link side tables (e.g. ``workstream_attachments``)
         use this to associate the row after save.
+
+        ``source`` is the persisted twin of the in-memory ``_source``
+        side-channel (today only ``"system_nudge"`` for wake-driven
+        empty user turns).  ``reminders`` is a JSON-encoded list mirroring
+        ``_reminders``; both are NULL for the common case where a
+        message carries no metacog payload.
         """
         ...
 
@@ -130,8 +138,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``.  Timestamp and workstream
-        updated-at are handled internally.
+        ``tool_calls``, ``source``, ``reminders``.  Timestamp and
+        workstream updated-at are handled internally.
         """
         ...
 

turnstone/core/storage/_schema.py

@@ -38,6 +38,14 @@ conversations = sa.Table(
     sa.Column("tool_call_id", sa.Text),
     sa.Column("provider_data", sa.Text),
     sa.Column("tool_calls", sa.Text),
+    # Sibling-key columns mirroring the in-memory ``_source`` /
+    # ``_reminders`` side-channels.  ``_source`` audits which producer
+    # synthesised the row (today only ``"system_nudge"`` for wake-driven
+    # empty user turns); ``_reminders`` stores the JSON-encoded reminder
+    # list ``[{type, text, ...optional}]`` so multi-tab / multi-device
+    # replay sees the same bubble shape the originating tab saw live.
+    sa.Column("_source", sa.Text),
+    sa.Column("_reminders", sa.Text),
 )
 
 sa.Index("idx_conversations_timestamp", conversations.c.timestamp)

turnstone/core/storage/_sqlite.py

@@ -218,6 +218,8 @@ class SQLiteBackend:
         tool_call_id: str | None = None,
         provider_data: str | None = None,
         tool_calls: str | None = None,
+        source: str | None = None,
+        reminders: str | None = None,
     ) -> int:
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         content = sanitize_text(content)
@@ -234,6 +236,8 @@ class SQLiteBackend:
                     "tool_call_id": tool_call_id,
                     "provider_data": provider_data,
                     "tool_calls": tool_calls,
+                    "_source": source,
+                    "_reminders": reminders,
                 },
             )
             if result.lastrowid is None:
@@ -277,6 +281,8 @@ class SQLiteBackend:
                     "tool_call_id": row.get("tool_call_id"),
                     "provider_data": sanitize_text(row.get("provider_data")),
                     "tool_calls": row.get("tool_calls"),
+                    "_source": row.get("source"),
+                    "_reminders": row.get("reminders"),
                 }
             )
         with self._conn() as conn:
@@ -312,6 +318,8 @@ class SQLiteBackend:
                         conversations.c.tool_call_id,
                         conversations.c.provider_data,
                         conversations.c.tool_calls,
+                        conversations.c._source,
+                        conversations.c._reminders,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id.desc())
@@ -328,6 +336,8 @@ class SQLiteBackend:
                         conversations.c.tool_call_id,
                         conversations.c.provider_data,
                         conversations.c.tool_calls,
+                        conversations.c._source,
+                        conversations.c._reminders,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id)

turnstone/core/storage/_utils.py

@@ -289,9 +289,12 @@ def reconstruct_messages(
 ) -> list[dict[str, Any]]:
     """Reconstruct OpenAI message format from stored conversation rows.
 
-    Each *row* is a 7-tuple ``(id, role, content, tool_name,
-    tool_call_id, provider_data, tool_calls_json)``, ordered
-    chronologically by row id.
+    Each *row* is a 9-tuple ``(id, role, content, tool_name,
+    tool_call_id, provider_data, tool_calls_json, source,
+    reminders_json)``, ordered chronologically by row id.  The trailing
+    two columns mirror the ``_source`` / ``_reminders`` in-memory side
+    channels so multi-tab / multi-device replay sees the same bubble
+    shape the originating tab saw live.
 
     When ``attachments_by_msg`` is provided, any user row whose id has
     attachments is rebuilt with multipart list content (text +
@@ -299,7 +302,17 @@ def reconstruct_messages(
     """
     messages: list[dict[str, Any]] = []
     for row in rows:
-        row_id, role, content, _tool_name, tc_id, provider_data, tool_calls_json = row
+        (
+            row_id,
+            role,
+            content,
+            _tool_name,
+            tc_id,
+            provider_data,
+            tool_calls_json,
+            source,
+            reminders_json,
+        ) = row
 
         if role == "user":
             parts: list[dict[str, Any]] = []
@@ -325,9 +338,17 @@ def reconstruct_messages(
                 umsg: dict[str, Any] = {"role": "user", "content": user_content}
                 if meta:
                     umsg["_attachments_meta"] = meta
-                messages.append(umsg)
             else:
-                messages.append({"role": "user", "content": content or ""})
+                umsg = {"role": "user", "content": content or ""}
+            if source:
+                umsg["_source"] = str(source)
+            if reminders_json:
+                # Mirrors the ``provider_data`` / ``tool_calls`` JSON
+                # decode pattern below: malformed JSON in the column is
+                # swallowed silently rather than aborting load.
+                with contextlib.suppress(json.JSONDecodeError, TypeError):
+                    umsg["_reminders"] = json.loads(reminders_json)
+            messages.append(umsg)
 
         elif role == "assistant":
             msg: dict[str, Any] = {"role": "assistant", "content": content or ""}
@@ -340,13 +361,18 @@ def reconstruct_messages(
             messages.append(msg)
 
         elif role == "tool":
-            messages.append(
-                {
-                    "role": "tool",
-                    "tool_call_id": tc_id or "",
-                    "content": content or "",
-                }
-            )
+            tmsg: dict[str, Any] = {
+                "role": "tool",
+                "tool_call_id": tc_id or "",
+                "content": content or "",
+            }
+            if reminders_json:
+                # Tool-channel reminders (``tool_error`` / ``repeat``)
+                # ride the same column so replay shows the same below-
+                # the-tool bubble the originating tab rendered live.
+                with contextlib.suppress(json.JSONDecodeError, TypeError):
+                    tmsg["_reminders"] = json.loads(reminders_json)
+            messages.append(tmsg)
 
     # Repair: strip trailing incomplete tool call turns
     while messages:

turnstone/core/storage/migrations/versions/050_conversations_source_and_reminders.py

@@ -0,0 +1,40 @@
+"""Add ``_source`` and ``_reminders`` columns to ``conversations``.
+
+Persisting these two fields lets multi-tab / multi-device replay show
+the same metacognitive bubble shape the originating tab saw live.
+Until now, the side-channel keys lived only in the in-memory
+``ChatSession.messages`` list, so a second browser tab connecting via
+``/history`` saw a synthetic empty wake row missing entirely (skipped
+at save time) and any preceding tab's reminder bubbles missing as well.
+
+* ``_source`` — TEXT NULL.  Today only ``"system_nudge"`` is written
+  (the wake-driven empty user turn marker).  Future producers can
+  extend (e.g. ``"external_webhook"``) without another migration.
+* ``_reminders`` — TEXT NULL holding a JSON array of reminder dicts
+  ``{type, text, ...optional}``.  Empty / missing column means no
+  reminders for that row.
+
+Revision ID: 050
+Revises: 049
+Create Date: 2026-05-06
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "050"
+down_revision = "049"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("conversations", sa.Column("_source", sa.Text, nullable=True))
+    op.add_column("conversations", sa.Column("_reminders", sa.Text, nullable=True))
+
+
+def downgrade() -> None:
+    # Reverse order of upgrade.  ``op.drop_column`` works on SQLite via
+    # Alembic batch-mode auto-rebuild and on PostgreSQL natively.
+    op.drop_column("conversations", "_reminders")
+    op.drop_column("conversations", "_source")