fix(compression): fence durable transcript revisions (#6554)

Co-authored-by: Anthony Ruiz <ruizanthony@users.noreply.github.com>
Co-authored-by: nesquena-hermes <nesquena+hermes@gmail.com>

api/models.py

@@ -13,7 +13,9 @@ import threading
 import time
 import uuid
 from contextlib import closing, contextmanager
+from dataclasses import dataclass
 from pathlib import Path
+from typing import Literal, overload
 
 try:  # pragma: no cover - platform-specific imports.
     import fcntl as _fcntl
@@ -8107,6 +8109,43 @@ def _json_loads_if_string(value):
         return value
 
 
+@dataclass(frozen=True)
+class StateDBSessionMessagesSnapshot:
+    """Internal message projection paired with its durable SQLite revision."""
+
+    messages: list
+    revision: dict | None
+
+
+def _state_db_session_messages_result(messages, revision, *, with_revision):
+    if with_revision:
+        return StateDBSessionMessagesSnapshot(messages=list(messages or []), revision=revision)
+    return list(messages or [])
+
+
+def _state_db_active_rows_digest(rows) -> str:
+    """Match the Agent fence for model-facing fields mutable in place."""
+    digest = hashlib.sha256()
+    stamped = False
+    for row in rows:
+        keys = row.keys()
+        api_content = row['api_content'] if 'api_content' in keys else None
+        if api_content is None:
+            continue
+        stamped = True
+        digest.update(
+            json.dumps(
+                [row['id'], api_content],
+                ensure_ascii=False,
+                separators=(',', ':'),
+                default=str,
+            ).encode('utf-8', errors='surrogatepass')
+        )
+        digest.update(b'\n')
+    return digest.hexdigest() if stamped else ''
+
+
+@overload
 def get_state_db_session_messages(
     sid,
     *,
@@ -8115,7 +8154,33 @@ def get_state_db_session_messages(
     since_timestamp=None,
     include_inactive: bool = False,
     limit=None,
-) -> list:
+    with_revision: Literal[False] = False,
+) -> list: ...
+
+
+@overload
+def get_state_db_session_messages(
+    sid,
+    *,
+    stitch_continuations: bool = False,
+    profile=None,
+    since_timestamp=None,
+    include_inactive: bool = False,
+    limit=None,
+    with_revision: Literal[True],
+) -> StateDBSessionMessagesSnapshot: ...
+
+
+def get_state_db_session_messages(
+    sid,
+    *,
+    stitch_continuations: bool = False,
+    profile=None,
+    since_timestamp=None,
+    include_inactive: bool = False,
+    limit=None,
+    with_revision: bool = False,
+):
     """Read messages for a Hermes session from state.db.
 
     When *profile* is supplied, reads from that profile's state.db; otherwise
@@ -8145,20 +8210,23 @@ def get_state_db_session_messages(
     ``active=0`` archive rows back in resurrects pre-compaction history and can
     make every later turn re-trigger compression. Pass ``include_inactive=True``
     only for explicit recovery/audit views.
+
+    ``with_revision=True`` returns a :class:`StateDBSessionMessagesSnapshot`.
+    Its revision is derived from the exact rows fetched by the same SQLite
+    query and is available only for an unbounded, active, current-segment read.
+    Existing callers keep the historical list return by default.
     """
     try:
         import sqlite3
     except ImportError:
-        return []
+        return _state_db_session_messages_result([], None, with_revision=with_revision)
 
     if isinstance(profile, str) and profile:
         db_path = _get_profile_home(profile) / 'state.db'
-        if not db_path.exists():
-            db_path = _active_state_db_path()
     else:
         db_path = _active_state_db_path()
     if not db_path.exists():
-        return []
+        return _state_db_session_messages_result([], None, with_revision=with_revision)
 
     try:
         with closing(open_state_db_readonly(db_path)) as conn:
@@ -8168,7 +8236,7 @@ def get_state_db_session_messages(
             available = {str(row['name']) for row in cur.fetchall()}
             required = {'role', 'content', 'timestamp'}
             if not required.issubset(available):
-                return []
+                return _state_db_session_messages_result([], None, with_revision=with_revision)
             optional = [
                 'tool_call_id',
                 'tool_calls',
@@ -8185,7 +8253,18 @@ def get_state_db_session_messages(
                 'api_content',
             ]
             id_col = ['id'] if 'id' in available else []
-            selected = id_col + ['role', 'content', 'timestamp'] + [c for c in optional if c in available]
+            revision_cols = []
+            if with_revision:
+                if 'active' in available:
+                    revision_cols.append('active')
+                if 'api_content' in available:
+                    revision_cols.append('api_content')
+            selected = (
+                id_col
+                + ['role', 'content', 'timestamp']
+                + revision_cols
+                + [c for c in optional if c in available]
+            )
 
             session_chain = [str(sid)]
             if stitch_continuations:
@@ -8244,6 +8323,7 @@ def get_state_db_session_messages(
             active_clause = ""
             if 'active' in available and not include_inactive:
                 active_clause = " AND (active IS NULL OR active != 0)"
+            durable_order_column = 'id' if 'id' in available else 'timestamp'
             # Defensive row cap (backstop only — see docstring). Applied as a
             # SQL LIMIT bound parameter (?) so the tail (newest) rows are
             # retained and a pathological state.db can't materialize unbounded
@@ -8255,10 +8335,9 @@ def get_state_db_session_messages(
                 except (TypeError, ValueError):
                     limit_int = None
                 if limit_int is not None:
-                    # The query orders ASC (oldest first); to keep the NEWEST
+                    # Keep the newest durable rows under the cap, then restore
-                    # rows under the cap, take a descending-ordered subquery and
+                    # canonical append order for the caller.
-                    # re-sort ascending — a plain LIMIT would keep the oldest.
+                    limit_clause = f" ORDER BY {durable_order_column} DESC LIMIT ?"
-                    limit_clause = " ORDER BY timestamp DESC, id DESC LIMIT ?"
                     params.append(limit_int)
             if limit_clause:
                 cur.execute(f"""
@@ -8269,7 +8348,7 @@ def get_state_db_session_messages(
                         {since_clause}
                         {active_clause}
                         {limit_clause}
-                    ) ORDER BY timestamp ASC, id ASC
+                    ) ORDER BY {durable_order_column} ASC
                 """, params)
             else:
                 cur.execute(f"""
@@ -8278,10 +8357,34 @@ def get_state_db_session_messages(
                     WHERE session_id IN ({placeholders})
                     {since_clause}
                     {active_clause}
-                    ORDER BY timestamp ASC, id ASC
+                    ORDER BY {durable_order_column} ASC
                 """, params)
+            rows = cur.fetchall()
+            revision = None
+            if (
+                with_revision
+                and 'id' in available
+                and 'active' in available
+                and len(session_chain) == 1
+                and str(session_chain[0]) == str(sid)
+                and since_timestamp is None
+                and not include_inactive
+                and limit is None
+                and all(row['active'] == 1 for row in rows)
+            ):
+                revision = {
+                    'session_id': str(sid),
+                    'active_message_count': len(rows),
+                    'max_active_message_id': max(
+                        (int(row['id']) for row in rows),
+                        default=0,
+                    ),
+                }
+                if 'api_content' in available:
+                    revision['active_rows_digest'] = _state_db_active_rows_digest(rows)
+
             msgs = []
-            for row in cur.fetchall():
+            for row in rows:
                 msg = {
                     'role': row['role'],
                     'content': row['content'],
@@ -8315,8 +8418,8 @@ def get_state_db_session_messages(
                     msg['name'] = msg['tool_name']
                 msgs.append(msg)
     except Exception:
-        return []
+        return _state_db_session_messages_result([], None, with_revision=with_revision)
-    return msgs
+    return _state_db_session_messages_result(msgs, revision, with_revision=with_revision)
 
 
 def get_state_db_session_message_prefix_summary(
@@ -10210,12 +10313,40 @@ def merge_session_messages_append_only(
     return merged_messages
 
 
+@overload
 def reconciled_state_db_messages_for_session(
-    session, *, prefer_context: bool = False, state_messages: list | None = None
+    session,
-) -> list:
+    *,
+    prefer_context: bool = False,
+    state_messages: list | StateDBSessionMessagesSnapshot | None = None,
+    with_revision: Literal[False] = False,
+) -> list: ...
+
+
+@overload
+def reconciled_state_db_messages_for_session(
+    session,
+    *,
+    prefer_context: bool = False,
+    state_messages: list | StateDBSessionMessagesSnapshot | None = None,
+    with_revision: Literal[True],
+) -> StateDBSessionMessagesSnapshot: ...
+
+
+def reconciled_state_db_messages_for_session(
+    session,
+    *,
+    prefer_context: bool = False,
+    state_messages: list | StateDBSessionMessagesSnapshot | None = None,
+    with_revision: bool = False,
+):
     """Return append-only messages reconciled with state.db for a WebUI session."""
     if session is None:
-        return []
+        return _state_db_session_messages_result([], None, with_revision=with_revision)
+    state_revision = None
+    if isinstance(state_messages, StateDBSessionMessagesSnapshot):
+        state_revision = state_messages.revision
+        state_messages = state_messages.messages
     local_messages = []
     using_context_messages = False
     if prefer_context:
@@ -10226,7 +10357,28 @@ def reconciled_state_db_messages_for_session(
     if not local_messages:
         local_messages = getattr(session, 'messages', None) or []
     if state_messages is None:
-        state_messages = get_state_db_session_messages(getattr(session, 'session_id', None))
+        session_id = getattr(session, 'session_id', None)
+        session_profile = getattr(session, 'profile', None)
+        if with_revision:
+            state_result = get_state_db_session_messages(
+                session_id,
+                profile=session_profile,
+                with_revision=True,
+            )
+        elif session_profile:
+            state_result = get_state_db_session_messages(
+                session_id,
+                profile=session_profile,
+            )
+        else:
+            # Preserve the historical one-argument call contract for ordinary
+            # reconciliation and for integrations that replace this reader.
+            state_result = get_state_db_session_messages(session_id)
+        if isinstance(state_result, StateDBSessionMessagesSnapshot):
+            state_revision = state_result.revision
+            state_messages = state_result.messages
+        else:
+            state_messages = state_result
     if prefer_context and local_messages:
         if using_context_messages:
             sidecar_messages = getattr(session, 'messages', None) or []
@@ -10253,22 +10405,35 @@ def reconciled_state_db_messages_for_session(
                             "Compressed context for session %s has no compression anchor; using context_messages only",
                             getattr(session, "session_id", None),
                         )
-                        return list(local_messages)
+                        return _state_db_session_messages_result(
+                            local_messages,
+                            state_revision,
+                            with_revision=with_revision,
+                        )
                     anchor_index = _state_db_anchor_index(state_messages, anchor_key)
                     if anchor_index is None:
                         logger.debug(
                             "Compressed context for session %s has an unverifiable compression anchor; using context_messages only",
                             getattr(session, "session_id", None),
                         )
-                        return list(local_messages)
+                        return _state_db_session_messages_result(
+                            local_messages,
+                            state_revision,
+                            with_revision=with_revision,
+                        )
                     state_messages = list(state_messages or [])[anchor_index + 1 :]
         state_messages = state_db_delta_after_context(local_messages, state_messages)
-    return merge_session_messages_append_only(
+    reconciled_messages = merge_session_messages_append_only(
         local_messages,
         state_messages,
         truncation_watermark=getattr(session, "truncation_watermark", None),
         truncation_boundary=getattr(session, "truncation_boundary", None),
     )
+    return _state_db_session_messages_result(
+        reconciled_messages,
+        state_revision,
+        with_revision=with_revision,
+    )
 
 
 def get_cli_session_messages(sid, *, profile=None) -> list:

tests/test_issue5121_provider_auth_terminal_error.py

@@ -285,11 +285,19 @@ def test_auth_401_classification_receives_stringified_probe_text(tmp_path, monke
     observed = {}
     real_classify = streaming._classify_provider_error
 
-    def _spy_classify_provider_error(err_str, exc=None, *, silent_failure=False):
+    def _spy_classify_provider_error(
+        err_str, exc=None, *, silent_failure=False, result=None
+    ):
         observed["err_str"] = err_str
         observed["exc"] = exc
         observed["silent_failure"] = silent_failure
-        return real_classify(err_str, exc, silent_failure=silent_failure)
+        observed["result"] = result
+        return real_classify(
+            err_str,
+            exc,
+            silent_failure=silent_failure,
+            result=result,
+        )
 
     with mock.patch.object(streaming, "_classify_provider_error", side_effect=_spy_classify_provider_error):
         _run_stream(monkeypatch, session, "stream_auth_probe_text", agent_cls, workspace=str(tmp_path))
@@ -415,6 +423,77 @@ def test_auth_retry_success_does_not_append_error_turn(tmp_path, monkeypatch):
     assert not any(msg.get("_error") for msg in saved.messages)
 
 
+def test_auth_exception_retry_structured_failure_still_emits_error(
+    tmp_path,
+    monkeypatch,
+):
+    session = _prepare_session(
+        "auth_retry_structured_failure",
+        "stream_auth_retry_structured_failure",
+        pending_user_message="Please retry",
+    )
+
+    class ExceptionThenStructuredFailureAgent(MockAgent):
+        runs = 0
+
+        def run_conversation(self, **kwargs):
+            type(self).runs += 1
+            if type(self).runs == 1:
+                raise RuntimeError("401 unauthorized")
+            if self.stream_delta_callback is not None:
+                self.stream_delta_callback("partial retry output")
+            return {
+                "error": {
+                    "type": "authentication_error",
+                    "status_code": 401,
+                    "message": "retry returned a structured failure",
+                },
+                "messages": list(kwargs.get("conversation_history") or []),
+            }
+
+    fake_queue = queue.Queue()
+    streaming.STREAMS["stream_auth_retry_structured_failure"] = fake_queue
+    config.STREAM_PARTIAL_TEXT["stream_auth_retry_structured_failure"] = ""
+    heal_rt = {
+        "provider": "test-provider",
+        "api_key": "fresh-key",
+        "base_url": None,
+    }
+
+    with mock.patch.object(streaming, "get_session", return_value=session), \
+         mock.patch.object(
+             streaming,
+             "_get_ai_agent",
+             return_value=ExceptionThenStructuredFailureAgent,
+         ), \
+         mock.patch.object(
+             streaming,
+             "resolve_model_provider",
+             return_value=("test-model", "test-provider", None),
+         ), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]), \
+         mock.patch.object(
+             streaming,
+             "_attempt_credential_self_heal",
+             return_value=heal_rt,
+         ):
+        streaming._run_agent_streaming(
+            session_id=session.session_id,
+            msg_text=session.pending_user_message,
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id="stream_auth_retry_structured_failure",
+        )
+
+    saved = Session.load("auth_retry_structured_failure")
+    assert saved is not None
+    events = _queue_events(fake_queue)
+    assert any(event == "apperror" for event, _ in events)
+    assert not any(event == "done" for event, _ in events)
+    assert saved.messages[-1]["_error"] is True
+
+
 def test_success_repeated_assistant_text_stays_successful_current_turn(tmp_path, monkeypatch):
     session = _prepare_session("repeat_success", "stream_repeat_success", pending_user_message="Please say it again")
     _seed_prior_turn(

tests/test_state_db_read_backstop.py

@@ -117,3 +117,49 @@ def test_limit_composes_with_since_timestamp(tmp_path, monkeypatch):
     msgs = get_state_db_session_messages("s1", since_timestamp=50.0, limit=10)
     assert len(msgs) == 10
     assert [m["content"] for m in msgs] == [f"msg{i}" for i in range(90, 100)]
+
+
+def test_reader_uses_durable_id_order_when_timestamps_are_non_monotonic(
+    tmp_path,
+    monkeypatch,
+):
+    db = tmp_path / "state.db"
+    conn = sqlite3.connect(str(db))
+    conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY)")
+    conn.execute(
+        """
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            role TEXT,
+            content TEXT,
+            timestamp REAL,
+            active INTEGER DEFAULT 1
+        )
+        """
+    )
+    conn.execute("INSERT INTO sessions (id) VALUES ('s1')")
+    conn.executemany(
+        "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
+        [
+            ("s1", "user", "id-1 timestamp-300", 300.0),
+            ("s1", "assistant", "id-2 timestamp-100", 100.0),
+            ("s1", "user", "id-3 timestamp-200", 200.0),
+        ],
+    )
+    conn.commit()
+    conn.close()
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db)
+
+    full = get_state_db_session_messages("s1")
+    limited = get_state_db_session_messages("s1", limit=2)
+
+    assert [message["content"] for message in full] == [
+        "id-1 timestamp-300",
+        "id-2 timestamp-100",
+        "id-3 timestamp-200",
+    ]
+    assert [message["content"] for message in limited] == [
+        "id-2 timestamp-100",
+        "id-3 timestamp-200",
+    ]