diff --git a/tests/test_reconstruct_messages.py b/tests/test_reconstruct_messages.py index 2edba237..9095eec8 100644 --- a/tests/test_reconstruct_messages.py +++ b/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: diff --git a/tests/test_session.py b/tests/test_session.py index a1333763..95f5e852 100644 --- a/tests/test_session.py +++ b/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 diff --git a/tests/test_storage_source_reminders.py b/tests/test_storage_source_reminders.py new file mode 100644 index 00000000..07774121 --- /dev/null +++ b/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" diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index 5debefdf..8c7bd6c2 100644 --- a/turnstone/core/memory.py +++ b/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) diff --git a/turnstone/core/session.py b/turnstone/core/session.py index e2f76d60..b192d7b2 100644 --- a/turnstone/core/session.py +++ b/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: diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index b040a099..c018467b 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/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) diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index fbda5063..01b807af 100644 --- a/turnstone/core/storage/_protocol.py +++ b/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. """ ... diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py index 5f679cb5..e4a34e39 100644 --- a/turnstone/core/storage/_schema.py +++ b/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) diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index fbc44084..f08f2515 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/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) diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py index 422bffb3..d1cd7542 100644 --- a/turnstone/core/storage/_utils.py +++ b/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: diff --git a/turnstone/core/storage/migrations/versions/050_conversations_source_and_reminders.py b/turnstone/core/storage/migrations/versions/050_conversations_source_and_reminders.py new file mode 100644 index 00000000..636ad697 --- /dev/null +++ b/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")