fix(gateway): move a session's context-breakdown key on an in-place row rewrite

conversation_revision keyed on the active rows' COUNT(*) and MAX(id).
set_user_message_content (prompt expansion, multimodal context), the
api_content writes and the redaction pass rewrite rows in place and move
neither, so a breakdown cached before the rewrite kept the old size; it is
usually masked by last_prompt_tokens moving, but not when that turn's call
fails (a context overflow after an @-file expansion), when the meter matters.

sessions gains transcript_revision, bumped by an AFTER UPDATE trigger on
the message columns the transcript loader reads, so every writer (older
builds included) moves it; conversation_revision folds the lineage's sum
into its key. Additive: the column is reconciled onto existing stores and
the trigger lives with the deferred DDL that runs after reconcile.

Test: test_context_breakdown_follows_an_in_place_row_rewrite (dokterdok's
probe), red on the previous head.

Review finding by @kshitijk4poor on #140968
Review finding by @dokterdok on #140968
Refs #134889 (GW-18)
Refs #140719

hermes_state_common.py

@@ -507,6 +507,7 @@ CREATE TABLE IF NOT EXISTS sessions (
     hidden INTEGER NOT NULL DEFAULT 0,
     last_read_at REAL,
     tool_names TEXT,
+    transcript_revision INTEGER NOT NULL DEFAULT 0,
     FOREIGN KEY (parent_session_id) REFERENCES sessions(id),
     FOREIGN KEY (system_prompt_hash) REFERENCES system_prompts(hash)
 );
@@ -767,6 +768,16 @@ WHEN EXISTS (SELECT 1 FROM session_admissions WHERE target_session_id = old.id)
 BEGIN
     SELECT RAISE(ABORT, 'Refused to delete a session that has gateway runtime history: state.db was upgraded by a newer Hermes, and this older Hermes cannot delete such sessions. Downgrading is unsupported. Run hermes update to go back to the newer release, or restore the pre-update snapshot: https://hermes-agent.nousresearch.com/docs/getting-started/updating#rolling-back-to-an-older-release');
 END;
+DROP TRIGGER IF EXISTS messages_transcript_revision_update;
+CREATE TRIGGER IF NOT EXISTS messages_transcript_revision_update
+AFTER UPDATE OF role, content, api_content, tool_calls, tool_call_id, tool_name, reasoning, reasoning_content,
+    reasoning_details ON messages
+BEGIN
+    -- An in-place rewrite (prompt expansion, multimodal context, a redaction) moves neither the
+    -- active row count nor the highest id; this write-side revision is what a transcript-derived
+    -- cache keys on (conversation_revision). Whoever wrote the row, old builds included.
+    UPDATE sessions SET transcript_revision = transcript_revision + 1 WHERE id = new.session_id;
+END;
 CREATE INDEX IF NOT EXISTS idx_messages_session_active
     ON messages(session_id, active, timestamp);
 CREATE INDEX IF NOT EXISTS idx_messages_display_page

hermes_state_messages.py

@@ -1534,13 +1534,16 @@ class SessionMessagesMixin(SessionDetachedResultsMixin):
 
     def conversation_revision(self, session_id: str, include_ancestors: bool = False) -> tuple:
         """A cheap fingerprint of what ``get_messages_as_conversation`` (live rows) would load: the
-        lineage ids plus the active rows' count and highest id. Any append, rewind, undo or compaction
-        moves it, so a payload derived from the transcript can be reused while it holds."""
+        lineage ids, the active rows' count and highest id, and the sessions' write-side
+        ``transcript_revision`` (bumped by a trigger on every in-place row rewrite). Any append,
+        rewind, undo, compaction or rewrite moves it, so a payload derived from the transcript can be
+        reused while it holds."""
         ids = self._resume_lineage_ids(session_id) if include_ancestors else [session_id]
+        marks = _placeholders(ids)
         row = self._read_one(
-            f"SELECT COUNT(*), COALESCE(MAX(id), 0) FROM messages WHERE session_id IN ({_placeholders(ids)})"
-            " AND active = 1", tuple(ids))
-        return (tuple(ids), int(row[0]), int(row[1]))
+            f"SELECT COUNT(*), COALESCE(MAX(id), 0), (SELECT COALESCE(SUM(transcript_revision), 0) FROM sessions"
+            f" WHERE id IN ({marks})) FROM messages WHERE session_id IN ({marks}) AND active = 1", (*ids, *ids))
+        return (tuple(ids), int(row[0]), int(row[1]), int(row[2]))
 
     def _dedupe_replayed_user(self, messages, msg, exact_user_clones) -> tuple[bool, Any]:
         """Ancestor-lineage dedupe of one decoded user *msg* -> ``(skip, exact_clone_key)``. Rotation

tests/gateway/test_tui_session_reads.py

@@ -169,3 +169,31 @@ async def test_context_breakdown_is_reused_until_the_transcript_moves(owner, mon
     assert computed == [0, 1] and moved['estimated_total'] == 1
     await owner.call('session.context_breakdown', session_id=sid)
     assert computed == [0, 1]
+
+
+@pytest.mark.asyncio
+async def test_context_breakdown_follows_an_in_place_row_rewrite(owner, monkeypatch):
+    """The turn prologue rewrites an already-written user row in place (``set_user_message_content``:
+    an @-file expansion, multimodal context), which moves neither the row count nor the highest id.
+    A breakdown cached before the rewrite must not keep the old size (probe by @dokterdok)."""
+    sid = (await owner.call('session.create', request_id='r', source='cli', cwd=str(owner.workdir)))['session_id']
+    from agent import context_breakdown, context_file_sources
+    agent = SimpleNamespace(model='fake-model', provider='custom')
+    owner.authority.runner._cached_agent_for = lambda route: agent
+
+    def compute(a, messages):
+        chars = sum(len(str(m.get('content') or '')) for m in messages)
+        return {'categories': [], 'context_max': 128000, 'context_used': chars, 'context_percent': 0,
+                'estimated_total': chars, 'context_estimated': True, 'context_source': 'local_estimate',
+                'model': 'fake-model'}
+    monkeypatch.setattr(context_breakdown, 'compute_session_context_breakdown', compute)
+    monkeypatch.setattr(context_file_sources, 'context_file_sources_for_agent', lambda a: [])
+    from gateway.session_tui_reads import _stored_id
+    stored = _stored_id(owner.authority, SimpleNamespace(session_id=sid))
+    db = owner.authority.db
+    row_id = db.append_message(stored, 'user', 'explain @file:big.py')
+    before = await owner.call('session.context_breakdown', session_id=sid)
+    expanded = 'explain @file:big.py\n' + 'x' * 20000
+    assert db.set_user_message_content(stored, row_id, expanded) == 1
+    after = await owner.call('session.context_breakdown', session_id=sid)
+    assert after['estimated_total'] == len(expanded) > before['estimated_total']