CLI keeps the api_content sidecar when a turn was persisted early — prompt cache survives (#102194, salvage #102411)

agent/session_persistence.py

@@ -7,7 +7,7 @@
 import re
 from contextlib import nullcontext
 
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from agent.context_compressor import (
     COMPRESSED_SUMMARY_METADATA_KEY,
@@ -74,6 +74,19 @@ def _override_replaces_content(msg: Dict, content: Any, override: Any) -> bool:
     )
 
 
+def durable_user_row_content(agent, msg: Dict, content: Any, api_content: Any) -> Tuple[Any, Any]:
+    """``(content, api_content)`` as the current turn's user row is written: the persist override is the
+    clean transcript, the live content is what the wire sent — so when they differ and nothing else was
+    injected, the live bytes ARE the sidecar. Shared by the flush and the turn-start stamp so the stamp
+    matches the row the flush wrote."""
+    override = getattr(agent, "_persist_user_message_override", None)
+    if _override_replaces_content(msg, content, override):
+        if api_content is None and isinstance(content, str) and content != override:
+            api_content = content
+        content = override
+    return content, api_content
+
+
 def _summary_display_kind(msg: Dict) -> Any:
     """Standalone handoffs are hidden so they never occupy the active user slot in retry/undo dispatch;
     merge-into-tail carriers keep their prior visibility."""
@@ -143,12 +156,7 @@ def _db_flush_row(agent, msg: Dict, is_current_turn_user: bool) -> Dict[str, Any
     api_content = msg.get("api_content") if isinstance(msg.get("api_content"), str) else None
     timestamp = msg.get("timestamp")
     if is_current_turn_user and role == "user":
-        override = getattr(agent, "_persist_user_message_override", None)
-        if _override_replaces_content(msg, content, override):
-            # Live content is what the wire sent, the override is the clean transcript; keep the sent bytes.
-            if api_content is None and isinstance(content, str) and content != override:
-                api_content = content
-            content = override
+        content, api_content = durable_user_row_content(agent, msg, content, api_content)
         ov_timestamp = getattr(agent, "_persist_user_message_timestamp", None)
         timestamp = timestamp if ov_timestamp is None else ov_timestamp
     if api_content == content:

agent/turn_context.py

@@ -728,30 +728,44 @@ def _stamp_api_content_sidecar(
     """api_content sidecar — persist what you send: injected context lives only in the
     API copy, so stamp the exact sent bytes on the live dict for replay."""
     _turn_user_msg = messages[current_turn_user_idx]
-    _api_content = compose_user_api_content(
-        _turn_user_msg.get("content", ""), ext_prefetch_cache, plugin_user_context
+    live_content = _turn_user_msg.get("content")
+    from agent.session_persistence import _persist_lock, durable_user_row_content
+    # Match the row the flush wrote (persist override = clean transcript), not the live bytes.
+    durable_content, _api_content = durable_user_row_content(
+        agent, _turn_user_msg, live_content,
+        compose_user_api_content(live_content or "", ext_prefetch_cache, plugin_user_context),
     )
-    if _api_content is None or _api_content == _turn_user_msg.get("content"):
+    if _api_content is None or _api_content == durable_content:
         return
     _turn_user_msg["api_content"] = _api_content
-    # In-place preflight compaction already inserted this turn's user row and the
-    # crash persist identity-skips compacted dicts, so backfill the stamp onto the row
-    # directly. Rotation mode flushes to the child session later.
-    if not (preflight_compressed and getattr(agent, "_last_compaction_in_place", False)):
-        return
-    _db = getattr(agent, "_session_db", None)
-    if _db is not None:
+
+    # When another writer materialized this turn's user row BEFORE the sidecar existed — in-place
+    # preflight compaction, or a close/early flush that raced the prologue (#102194) — the crash
+    # persist marker-skips the message and the stamp never reaches the DB, so the next turn replays
+    # clean content and the request prefix diverges here. Both writers stamp ``_row_id`` on the live
+    # dict, which is at once the proof a row exists and the address to update.
+    #
+    # Never widen this to an unconditional positional backfill — see set_latest_user_api_content.
+    #
+    # ``_row_id`` is read under ``_session_persist_lock``: a close flush holds it while it commits
+    # the row and only then writes ``_row_id`` back (``sync_flushed_message_markers``). Read outside
+    # it, the stamp can land in between, see no id, return — and the flush then marks the message
+    # persisted with ``api_content = NULL``, leaving no writer to correct the row.
+    with _persist_lock(agent):
+        _row_id = _turn_user_msg.get("_row_id")
+        _in_place_compacted = preflight_compressed and bool(getattr(agent, "_last_compaction_in_place", False))
+        _db = getattr(agent, "_session_db", None)
+        if _db is None or not (isinstance(_row_id, int) or _in_place_compacted):
+            return
         try:
-            _db.set_latest_user_api_content(
-                agent.session_id, _turn_user_msg.get("content"), _api_content
-            )
+            if isinstance(_row_id, int):
+                _db.set_message_api_content(agent.session_id, _row_id, durable_content, _api_content)
+            else:
+                # Compacted copies carry no row id; positional is safe only because
+                # archive_and_compact just made this message the newest active user row.
+                _db.set_latest_user_api_content(agent.session_id, durable_content, _api_content)
         except Exception:
-            logger.warning(
-                "in-place compaction api_content backfill failed "
-                "for session=%s",
-                agent.session_id or "none",
-                exc_info=True,
-            )
+            logger.warning("api_content backfill failed for session=%s", agent.session_id or "none", exc_info=True)
 
 
 def _persist_turn_start(

hermes_state_messages.py

@@ -576,13 +576,53 @@ def _message_column_names(self, conn) -> List[str]:
     def set_latest_user_api_content(self, session_id: str, content: Any, api_content: str) -> int:
         """Backfill the ``api_content`` sidecar onto the newest ACTIVE user row (0/1 rows). Preflight compaction
         inserts that row BEFORE the sidecar exists and the later persist identity-skips compacted dicts;
-        without this a reload reopens the prompt-cache divergence. ``content`` match guards a racing rewrite."""
+        without this a reload reopens the prompt-cache divergence. ``content`` match guards a racing rewrite.
+
+        POSITIONAL, and only safe when the caller already knows the newest
+        active user row IS the message it stamped. The content match is NOT
+        sufficient on its own: repeated identical user turns ("ok", "y",
+        "continue") make an OLDER row compare equal, so calling this before
+        the current turn's row exists overwrites the previous turn's sidecar
+        with this turn's bytes — durable wrong-bytes replay, a worse cache
+        break than the missing sidecar. When the caller holds the durable row
+        id (``_row_id``, synced onto the live dict by
+        ``sync_flushed_message_markers`` and stamped by
+        :meth:`_insert_message_rows`), use :meth:`set_message_api_content`
+        instead — it addresses the exact row and cannot land on a neighbour.
+        """
         return self._write_rowcount(
             "UPDATE messages SET api_content = ? WHERE id = (SELECT id FROM messages "
             "WHERE session_id = ? AND role = 'user' AND active = 1 ORDER BY id DESC LIMIT 1"
             ") AND content IS ?",
             (_scrub_surrogates(api_content), session_id, self._encode_content(content)))
 
+    def set_message_api_content(
+        self, session_id: str, row_id: int, content: Any, api_content: str
+    ) -> int:
+        """Backfill the ``api_content`` sidecar onto ONE known durable row.
+
+        Row-addressed counterpart to :meth:`set_latest_user_api_content`: the
+        caller passes the ``_row_id`` the write path stamped on the live
+        message dict, so the update cannot drift onto a neighbouring row that
+        merely carries the same text.
+
+        Used by the turn prologue whenever the current turn's user row was
+        already materialized before the sidecar could be composed (in-place
+        preflight compaction, a close/early flush that raced the prologue).
+        The crash persist then marker-skips that message, so this is the only
+        way the stamped bytes reach the store.
+
+        ``active = 1`` and the ``content`` match stay as defensive guards: a
+        row the compaction archived, or one a racing rewrite changed, is left
+        untouched.
+        """
+        if not session_id or isinstance(row_id, bool) or not isinstance(row_id, int) or row_id <= 0:
+            return 0
+        return self._write_rowcount(
+            "UPDATE messages SET api_content = ? WHERE id = ? AND session_id = ? "
+            "AND role = 'user' AND active = 1 AND content IS ?",
+            (_scrub_surrogates(api_content), row_id, session_id, self._encode_content(content)))
+
     def _dedupe_display_generations(self, rows):
         """Collapse compaction generations so each logical message appears once (the protected tail is copied
         into each generation: same role/content/timestamp, different ``active``/id); prefer the live row, then