fix(tui): prompt.submit's truncate_before_user_ordinal reads fresh history

hermes_state.py

@@ -8040,48 +8040,104 @@ def replace_messages(
         differs.
         """
 
-        active_clause = " AND active = 1" if active_only else ""
-
         def _do(conn):
-            session = conn.execute(
-                "SELECT ended_at, end_reason FROM sessions WHERE id = ?",
-                (session_id,),
-            ).fetchone()
-            if (
-                session is not None
-                and session["ended_at"] is not None
-                and session["end_reason"] == "compression"
-            ):
-                raise CompressionSessionClosedError(session_id)
-            if archive_dropped:
-                # Content-preserving UPDATE: the rows keep their FTS entries
-                # (the messages_fts triggers fire on INSERT / DELETE / UPDATE
-                # of content columns, not on `active`), so the replaced turns
-                # stay readable via get_messages(include_inactive=True) and
-                # searchable with include_inactive=True after the rewrite.
-                conn.execute(
-                    "UPDATE messages SET active = 0 "
-                    "WHERE session_id = ? AND active = 1",
-                    (session_id,),
-                )
-            else:
-                conn.execute(
-                    f"DELETE FROM messages WHERE session_id = ?{active_clause}",
-                    (session_id,),
-                )
+            self._replace_messages_in_transaction(
+                conn,
+                session_id,
+                messages,
+                active_only=active_only,
+                archive_dropped=archive_dropped,
+            )
+
+        self._execute_write(_do)
+
+    def _replace_messages_in_transaction(
+        self,
+        conn: sqlite3.Connection,
+        session_id: str,
+        messages: List[Dict[str, Any]],
+        *,
+        active_only: bool,
+        archive_dropped: bool = False,
+    ) -> None:
+        """Replace a transcript using the caller's open write transaction."""
+        session = conn.execute(
+            "SELECT ended_at, end_reason FROM sessions WHERE id = ?",
+            (session_id,),
+        ).fetchone()
+        if (
+            session is not None
+            and session["ended_at"] is not None
+            and session["end_reason"] == "compression"
+        ):
+            raise CompressionSessionClosedError(session_id)
+        if archive_dropped:
+            # Content-preserving UPDATE: the rows keep their FTS entries
+            # because only the active marker changes.
             conn.execute(
-                "UPDATE sessions SET message_count = 0, tool_call_count = 0 WHERE id = ?",
+                "UPDATE messages SET active = 0 "
+                "WHERE session_id = ? AND active = 1",
                 (session_id,),
             )
-            total_messages, total_tool_calls = self._insert_message_rows(
-                conn, session_id, messages
-            )
+        else:
+            active_clause = " AND active = 1" if active_only else ""
             conn.execute(
-                "UPDATE sessions SET message_count = ?, tool_call_count = ? WHERE id = ?",
-                (total_messages, total_tool_calls, session_id),
+                f"DELETE FROM messages WHERE session_id = ?{active_clause}",
+                (session_id,),
             )
+        conn.execute(
+            "UPDATE sessions SET message_count = 0, tool_call_count = 0 WHERE id = ?",
+            (session_id,),
+        )
+        total_messages, total_tool_calls = self._insert_message_rows(
+            conn, session_id, messages
+        )
+        conn.execute(
+            "UPDATE sessions SET message_count = ?, tool_call_count = ? WHERE id = ?",
+            (total_messages, total_tool_calls, session_id),
+        )
 
-        self._execute_write(_do)
+    def replace_active_messages_if_unchanged(
+        self,
+        session_id: str,
+        expected_messages: List[Dict[str, Any]],
+        messages: List[Dict[str, Any]],
+        *,
+        archive_dropped: bool = False,
+    ) -> bool:
+        """Conditionally rewrite the active tip transcript.
+
+        The current model-fed projection is compared with ``expected_messages``
+        inside the same ``BEGIN IMMEDIATE`` transaction that performs the
+        rewrite. If another process appended or rewrote rows after the caller's
+        read, return ``False`` without deleting anything.
+        """
+
+        def _do(conn):
+            rows = conn.execute(
+                f"SELECT {self._CONVERSATION_ROW_COLUMNS} "
+                "FROM messages WHERE session_id = ? AND active = 1 ORDER BY id",
+                (session_id,),
+            ).fetchall()
+            current_messages = self._rows_to_conversation(
+                rows,
+                session_id=session_id,
+                include_ancestors=False,
+                repair_alternation=True,
+                include_row_ids=True,
+            )
+            if current_messages != expected_messages:
+                return False
+            self._replace_messages_in_transaction(
+                conn,
+                session_id,
+                messages,
+                active_only=True,
+                archive_dropped=archive_dropped,
+            )
+            return True
+
+        return bool(self._execute_write(_do))
 
     def has_archived_messages(self, session_id: str) -> bool:
         """Return True if the session has any soft-archived (``active = 0``) rows.

tests/hermes_state/test_replace_messages_archive_siblings.py

@@ -11,7 +11,8 @@
   Now passes ``active_only=True`` unconditionally.
 - ``tui_gateway/methods_prompt.py`` edit/regenerate truncation: bare
   ``replace_messages`` deleted the archived transcript on every
-  edit/regenerate of a compacted session.  Now ``active_only=True``.
+  edit/regenerate of a compacted session. It now uses an atomic conditional
+  active-only rewrite and archives the replaced live rows.
 
 Behavior contract on a fresh (never-compacted) session: every row is
 ``active=1``, so the active-only replace is identical to the full replace —
@@ -125,20 +126,22 @@ def test_fresh_session_active_only_equals_full_replace(self, state_db):
 
 
 class TestTuiPromptTruncationPreservesArchives:
-    def test_truncation_source_uses_active_only(self):
-        """The edit/regenerate persistence call must pass active_only=True."""
+    def test_truncation_source_uses_conditional_active_replace(self):
+        """The edit/regenerate write must stay atomic, active-only, and recoverable."""
         import inspect
         import tui_gateway.methods_prompt as mp
 
         src = inspect.getsource(mp)
-        # The truncation write is the only replace_messages call in the module;
-        # it must carry active_only=True.
+        # The helper name pins active-only behavior; archive_dropped preserves
+        # the replaced live rows instead of hard-deleting them.
         import re
-        calls = re.findall(r"db\.replace_messages\([^)]*\)", src, re.S)
-        assert calls, "expected the truncation replace_messages call"
+        calls = re.findall(
+            r"db\.replace_active_messages_if_unchanged\([^)]*\)", src, re.S
+        )
+        assert calls, "expected the conditional truncation rewrite"
         for call in calls:
-            assert "active_only=True" in call, (
-                f"bare replace_messages in methods_prompt — #80216 class: {call}"
+            assert "archive_dropped=True" in call, (
+                f"hard-delete conditional rewrite in methods_prompt — #80216 class: {call}"
             )
 
     def test_truncation_write_keeps_archived_rows(self, state_db):
@@ -148,10 +151,18 @@ def test_truncation_write_keeps_archived_rows(self, state_db):
         _seed_compacted_session(state_db, sid)
         assert _archived_count(state_db, sid) == 4
 
+        expected, _display = state_db.get_resume_conversations(sid)
         truncated = [{"role": "user", "content": "kept head"}]
-        state_db.replace_messages(sid, truncated, active_only=True)
+        replaced = state_db.replace_active_messages_if_unchanged(
+            sid,
+            expected,
+            truncated,
+            archive_dropped=True,
+        )
 
-        assert _archived_count(state_db, sid) == 4
+        assert replaced is True
+        # Four compacted rows plus the three replaced live rows survive.
+        assert _archived_count(state_db, sid) == 7
         live = [
             m for m in state_db.get_messages_as_conversation(sid)
             if m.get("role") in ("user", "assistant")

tests/test_hermes_state.py

@@ -612,6 +612,65 @@ def test_replace_messages_preserves_timestamps(self, db):
         assert [m["timestamp"] for m in msgs_out] == [100.0, 200.0, 300.0]
         assert self._raw_timestamps(db, "s1") == [100.0, 200.0, 300.0]
 
+    def test_conditional_active_replace_rejects_stale_snapshot(self, db):
+        """A concurrent append must survive a stale transcript rewrite."""
+        db.create_session(session_id="s1", source="cli")
+        db.append_message("s1", role="user", content="first")
+        db.append_message("s1", role="assistant", content="first reply")
+        expected, _display = db.get_resume_conversations("s1")
+
+        db.append_message("s1", role="user", content="concurrent")
+        replaced = db.replace_active_messages_if_unchanged(
+            "s1",
+            expected,
+            [],
+            archive_dropped=True,
+        )
+
+        assert replaced is False
+        assert [
+            message["content"]
+            for message in db.get_messages_as_conversation("s1")
+        ] == ["first", "first reply", "concurrent"]
+        with db._lock:
+            archived_count = db._conn.execute(
+                "SELECT COUNT(*) FROM messages WHERE session_id = ? AND active = 0",
+                ("s1",),
+            ).fetchone()[0]
+        assert archived_count == 0
+
+    def test_conditional_active_replace_commits_matching_snapshot(self, db):
+        """An unchanged active projection may be rewritten atomically."""
+        db.create_session(session_id="s1", source="cli")
+        db.append_message("s1", role="user", content="first")
+        db.append_message("s1", role="assistant", content="first reply")
+        db.append_message("s1", role="user", content="second")
+        expected, _display = db.get_resume_conversations("s1")
+
+        replaced = db.replace_active_messages_if_unchanged(
+            "s1",
+            expected,
+            expected[:2],
+            archive_dropped=True,
+        )
+
+        assert replaced is True
+        assert [
+            message["content"]
+            for message in db.get_messages_as_conversation("s1")
+        ] == ["first", "first reply"]
+        with db._lock:
+            archived = db._conn.execute(
+                "SELECT content FROM messages "
+                "WHERE session_id = ? AND active = 0 ORDER BY id",
+                ("s1",),
+            ).fetchall()
+        assert [row["content"] for row in archived] == [
+            "first",
+            "first reply",
+            "second",
+        ]
+