fix(state): four session-state fixes — safe close tracking, flush-cursor class fix, row-retry, usage-PK healer

agent/context_compressor.py

@@ -2166,6 +2166,10 @@ def __init__(
         self._micro_compact_consecutive_failures: int = 0
         self._micro_compact_last_failure_cursor: int = -1
         self._micro_compact_defrag_threshold_tokens: int = 2000
+        # Set by _defrag_rolling_summary when it pops _DB_PERSISTED_MARKER
+        # from a live dict in place; consumed by finalize_turn to invalidate
+        # the agent's bounded flush-scan cursor (sibling of the #75170 site).
+        self._flush_scan_cursor_invalidated: bool = False
         self._micro_compact_passes: int = 0
         self._micro_compact_tokens_saved_total: int = 0
         # Cadence: run a pass every Nth completed turn. Each pass rewrites
@@ -5322,6 +5326,15 @@ def _defrag_rolling_summary(
                 # Content changed after a possible flush — clear the persisted
                 # stamp so the DB sync/flush rewrites the row.
                 entry.pop(_DB_PERSISTED_MARKER, None)
+                # Sibling of the finalize_turn pop site (#75170): this pop
+                # also strips the marker from a LIVE dict in place, so the
+                # bounded flush-scan cursor would identity-skip the rewritten
+                # marker and the defragged summary would never reach state.db.
+                # The compressor holds no agent reference, so raise a flag the
+                # finalizer consumes to invalidate agent._db_flush_scan_prefix.
+                # (The pop sites at module scope — fresh copies in
+                # strip-marker helpers — break identity and need no flag.)
+                self._flush_scan_cursor_invalidated = True
                 break
         logger.info(
             "Micro-compaction defrag: rolling summary re-summarized "

agent/turn_finalizer.py

@@ -339,6 +339,12 @@ def finalize_turn(
                 # otherwise ``/resume`` reloads ``content=""`` and the bug
                 # resurfaces cross-session.
                 _tail.pop("_db_persisted", None)
+                # The bounded flush-scan cursor (run_agent.py) skips the
+                # identity-matched prefix of its previous snapshot on the
+                # assumption that no live dict loses the marker in place —
+                # this pop is the one place that does. Invalidate it so the
+                # filled row is re-examined instead of skipped.
+                agent._db_flush_scan_prefix = None
 
         # The model has completed its request, so replace API-local
         # voice/model/skill guidance with the clean user input before writing the
@@ -378,6 +384,18 @@ def finalize_turn(
                 ):
                     _before = len(messages)
                     _compacted = _compressor._micro_compact(messages)
+                    # Micro-compaction defrag rewrites the newest MICRO
+                    # marker's content and pops _db_persisted from the live
+                    # dict in place — the sibling of the pop site above. The
+                    # compressor has no agent reference, so it raises a flag
+                    # for us to invalidate the bounded flush-scan cursor;
+                    # otherwise the rewritten marker row is identity-skipped
+                    # and the stale summary persists to state.db.
+                    if getattr(
+                        _compressor, "_flush_scan_cursor_invalidated", False
+                    ):
+                        _compressor._flush_scan_cursor_invalidated = False
+                        agent._db_flush_scan_prefix = None
                     if isinstance(_compacted, list) and _compacted:
                         messages[:] = _compacted
                     _after = len(messages)

contributors/emails/dasilva.daniel6@gmail.com

@@ -0,0 +1 @@
+Dannou

contributors/emails/peace@trippyogi.com

@@ -0,0 +1 @@
+trippyogi

hermes_cli/sqlite_safe_read.py

@@ -40,7 +40,7 @@
 each of which spans the syscall *and* the registry mutation:
 
 * open + register (:func:`connect_tracked`)
-* unregister + close (:meth:`TrackedConnection.close`)
+* close + unregister (:meth:`TrackedConnection.close`)
 * check + ``open``/``read``/``close`` (:func:`read_header_bytes_preopen`)
 
 Without that, a thread could pass the "no live connection" check, a second
@@ -154,10 +154,14 @@ class _TrackingMixin:
     def close(self) -> None:  # type: ignore[misc]
         with _live_lock:
             path = getattr(self, "_hermes_tracked_path", None)
+            # Close first; untrack only once the descriptor is actually gone.
+            # Untracking before a failing close (e.g. cross-thread
+            # ProgrammingError) leaves the FD open while the byte-probe
+            # guard thinks nothing is live — see #75629.
+            super().close()  # type: ignore[misc]
             if path is not None:
                 self._hermes_tracked_path = None
                 untrack_connection(path)
-            super().close()  # type: ignore[misc]
 
 
 class TrackedConnection(_TrackingMixin, sqlite3.Connection):
@@ -169,9 +173,11 @@ class TrackedConnection(_TrackingMixin, sqlite3.Connection):
     method every close path must go through — keeps the registry from
     drifting upward and permanently disabling byte-probes.
 
-    The unregister and the real ``close()`` happen together under
+    The real ``close()`` and the unregister happen together under
     ``_live_lock`` so a concurrent probe can never observe "no live
-    connection" while this descriptor is still open.
+    connection" while this descriptor is still open. Unregister runs only
+    after ``close()`` succeeds; a raising close leaves the connection
+    tracked so the byte-probe guard keeps refusing.
 
     Note ``with conn:`` does NOT close a sqlite3 connection (it only commits or
     rolls back), so this hook is not fired spuriously by transaction scopes.

hermes_state.py

@@ -2405,6 +2405,17 @@ def _execute_write(
         # Set on the first compression-busy collision so the short wait is
         # measured from then, not from the start of the write.
         compression_deadline: Optional[float] = None
+
+        # Transient engine-level error observed on contended WAL appends
+        # (dual gateway/agent writers; FTS5 trigram sync holds the write
+        # lock). The identical write succeeds standalone, so it is
+        # retryable like locked/busy. The exception CLASS varies with the
+        # SQLite build — some surface it as InterfaceError, which lives
+        # OUTSIDE DatabaseError and escaped the retry net entirely on
+        # attempt 0 — so the check is message-scoped, not class-scoped.
+        def _is_no_more_rows(exc: sqlite3.Error) -> bool:
+            return "no more rows available" in str(exc).lower()
+
         while True:
             try:
                 with self._lock:
@@ -2460,9 +2471,13 @@ def _execute_write(
                         "a large WAL checkpoint, or an older pre-update "
                         "process; the database itself is healthy)"
                     ) from exc
-                # Non-lock error — propagate.
+                if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s):
+                    continue
+                # Non-lock error or patience exhausted — propagate.
                 raise
             except sqlite3.DatabaseError as exc:
+                if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s):
+                    continue
                 # Corrupt FTS shadow tables make every write raise the
                 # malformed/corrupt error class through the FTS sync triggers
                 # while the canonical messages table is intact. The gateway
@@ -2475,6 +2490,15 @@ def _execute_write(
                 if not self._try_runtime_fts_rebuild(exc):
                     raise
                 continue
+            except sqlite3.Error as exc:
+                # Catch-all for builds that surface 'no more rows available'
+                # as InterfaceError (a sibling of DatabaseError, not a
+                # subclass) or another sqlite3.Error class outside the two
+                # handlers above. Message-scoped: anything else propagates
+                # untouched.
+                if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s):
+                    continue
+                raise
 
     def _sleep_before_write_retry(
         self, deadline: float, patience_s: float

hermes_state_schema.py

@@ -290,6 +290,126 @@ def _col(row, idx, name):
         )
         cursor.execute("DROP TABLE gateway_routing_legacy_pk")
 
+    def _heal_session_model_usage_pk(self, cursor: sqlite3.Cursor) -> None:
+        """Rebuild ``session_model_usage`` when its PRIMARY KEY lacks ``task``.
+
+        Installs whose ``state.db`` reached ``schema_version >= 22`` before
+        the ``task`` dimension was added carry a 5-column PRIMARY KEY
+        ``(session_id, model, billing_provider, billing_base_url,
+        billing_mode)``.  ``_reconcile_columns()`` ADDs the ``task`` column
+        as a bare nullable, but SQLite cannot ALTER a primary key, so the
+        shipped composite 6-column key never lands.  The version-gated v22
+        rebuild is unreachable on those installs (``current_version < 22``
+        is already false), so every upsert in ``_record_model_usage()``
+        fails with "ON CONFLICT clause does not match any PRIMARY KEY or
+        UNIQUE constraint" — aborting the enclosing write transaction and
+        silently zeroing all token *and* cost accounting (#73823).
+
+        Idempotent; runs unconditionally on every open, same pattern as
+        :meth:`_heal_gateway_routing_pk` above.  On healthy databases the
+        PRAGMA check short-circuits and this is a no-op.
+        """
+        try:
+            rows = cursor.execute(
+                'PRAGMA table_info("session_model_usage")'
+            ).fetchall()
+        except sqlite3.OperationalError:
+            return
+        if not rows:
+            # Table doesn't exist yet — SCHEMA_SQL creates it correctly.
+            return
+
+        def _col(row, idx, name):
+            return row[idx] if isinstance(row, (tuple, list)) else row[name]
+
+        pk_cols = {
+            _col(r, 1, "name") for r in rows if _col(r, 5, "pk")
+        }
+        if "task" in pk_cols:
+            # task is already in the PK — healthy.
+            return
+
+        logger.info(
+            "session_model_usage has legacy primary key %r (missing task); "
+            "rebuilding with composite 6-column key",
+            sorted(pk_cols),
+        )
+        # FK-off window: the connection enables PRAGMA foreign_keys=ON
+        # before _init_schema runs, and session_model_usage.session_id
+        # REFERENCES sessions(id).  INSERT OR IGNORE does NOT suppress
+        # foreign-key violations (OR IGNORE only covers uniqueness/NOT
+        # NULL conflicts), so an orphaned usage row — possible after a
+        # partial prune while accounting was broken — would abort the
+        # whole rebuild.  Disable FK enforcement for the copy and restore
+        # it afterwards.  PRAGMA foreign_keys is a no-op inside a
+        # transaction, which is fine here: _init_schema runs on an
+        # isolation_level=None connection with no transaction open.
+        cursor.execute("PRAGMA foreign_keys=OFF")
+        try:
+            cursor.execute(
+                "ALTER TABLE session_model_usage "
+                "RENAME TO session_model_usage_legacy_pk"
+            )
+            cursor.execute(
+                """CREATE TABLE session_model_usage (
+    session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
+    model TEXT NOT NULL,
+    billing_provider TEXT NOT NULL DEFAULT '',
+    billing_base_url TEXT NOT NULL DEFAULT '',
+    billing_mode TEXT NOT NULL DEFAULT '',
+    task TEXT NOT NULL DEFAULT '',
+    api_call_count INTEGER NOT NULL DEFAULT 0,
+    input_tokens INTEGER NOT NULL DEFAULT 0,
+    output_tokens INTEGER NOT NULL DEFAULT 0,
+    cache_read_tokens INTEGER NOT NULL DEFAULT 0,
+    cache_write_tokens INTEGER NOT NULL DEFAULT 0,
+    reasoning_tokens INTEGER NOT NULL DEFAULT 0,
+    estimated_cost_usd REAL NOT NULL DEFAULT 0,
+    actual_cost_usd REAL NOT NULL DEFAULT 0,
+    cost_status TEXT,
+    cost_source TEXT,
+    first_seen REAL,
+    last_seen REAL,
+    PRIMARY KEY (session_id, model, billing_provider, billing_base_url, billing_mode, task)
+)"""
+            )
+            # OR IGNORE: while the PK was wrong the reconciler may have left
+            # ``task`` NULL on old rows; COALESCE to '' can theoretically
+            # collide with a genuine ''-task row — keep the first, drop the
+            # duplicate rather than fail the heal.
+            cursor.execute(
+                """INSERT OR IGNORE INTO session_model_usage (
+                       session_id, model, billing_provider, billing_base_url,
+                       billing_mode, task, api_call_count, input_tokens,
+                       output_tokens, cache_read_tokens, cache_write_tokens,
+                       reasoning_tokens, estimated_cost_usd, actual_cost_usd,
+                       cost_status, cost_source, first_seen, last_seen
+                   )
+                   SELECT session_id, model,
+                          COALESCE(billing_provider, ''),
+                          COALESCE(billing_base_url, ''),
+                          COALESCE(billing_mode, ''),
+                          COALESCE(task, ''),
+                          api_call_count, input_tokens,
+                          output_tokens, cache_read_tokens, cache_write_tokens,
+                          reasoning_tokens, estimated_cost_usd, actual_cost_usd,
+                          cost_status, cost_source, first_seen, last_seen
+                   FROM session_model_usage_legacy_pk"""
+            )
+            cursor.execute("DROP TABLE session_model_usage_legacy_pk")
+            cursor.execute(
+                "CREATE INDEX IF NOT EXISTS idx_session_model_usage_session "
+                "ON session_model_usage(session_id)"
+            )
+            cursor.execute(
+                "CREATE INDEX IF NOT EXISTS idx_session_model_usage_model "
+                "ON session_model_usage(model)"
+            )
+        except sqlite3.OperationalError as exc:
+            logger.debug("session_model_usage PK heal skipped: %s", exc)
+        finally:
+            cursor.execute("PRAGMA foreign_keys=ON")
+
     def _init_schema(self):
         """Create tables and FTS if they don't exist, reconcile columns.
 
@@ -319,6 +439,12 @@ def _init_schema(self):
         # the one table-shape repair reconciliation can't express.
         self._heal_gateway_routing_pk(cursor)
 
+        # Rebuild session_model_usage if its PRIMARY KEY lacks the ``task``
+        # column (5-column PK on installs already at v22+ when the column
+        # landed — the version-gated rebuild is unreachable there, #73823).
+        # Same PK-rebuild constraint as gateway_routing above.
+        self._heal_session_model_usage_pk(cursor)
+
         # Indexes that reference reconciler-added columns must be created
         # AFTER _reconcile_columns runs — declaring them in SCHEMA_SQL
         # makes the initial executescript fail on legacy DBs (the index's

tests/agent/test_micro_compaction.py

@@ -752,3 +752,64 @@ def test_splice_preserves_db_persisted_stamps(self):
         assert not unstamped, (
             "splice must not strip _db_persisted from surviving messages"
         )
+
+
+class TestDefragFlushCursorInvalidation:
+    """Sibling of the finalize_turn pop site (#75170): defrag pops
+    _DB_PERSISTED_MARKER from the live marker dict in place, so the bounded
+    flush-scan cursor must be invalidated or the rewritten summary is
+    identity-skipped and never re-persisted."""
+
+    def _defrag_setup(self):
+        from agent.context_compressor import _DB_PERSISTED_MARKER
+
+        cc = _compressor(summary="FRESH DEFRAGGED SUMMARY")
+        messages = _conversation(exchanges=8)
+        messages = cc._micro_compact(list(messages))
+        # Simulate an incremental flush having stamped the marker row.
+        for m in messages:
+            if m.get(COMPRESSED_SUMMARY_METADATA_KEY):
+                m[_DB_PERSISTED_MARKER] = True
+        cc._micro_compact_rolling_summary = "x" * 40_000  # force defrag
+        return cc, messages
+
+    def test_defrag_marker_pop_raises_invalidation_flag(self):
+        from agent.context_compressor import _DB_PERSISTED_MARKER
+
+        cc, messages = self._defrag_setup()
+        assert cc._flush_scan_cursor_invalidated is False
+
+        result = cc._micro_compact(list(messages))
+
+        markers = _summary_markers(result)
+        assert len(markers) == 1
+        # The pop happened in place on the live dict...
+        assert not markers[0].get(_DB_PERSISTED_MARKER)
+        # ...so the compressor must flag the flush-scan cursor stale.
+        assert cc._flush_scan_cursor_invalidated is True
+
+    def test_no_defrag_no_flag(self):
+        cc = _compressor()
+        messages = _conversation(exchanges=8)
+        cc._micro_compact(list(messages))
+        assert cc._flush_scan_cursor_invalidated is False
+
+    def test_finalizer_consumes_flag_and_invalidates_agent_cursor(self):
+        """finalize_turn's micro-compaction block must translate the
+        compressor flag into agent._db_flush_scan_prefix = None (and reset
+        the flag) so the next flush re-examines the rewritten marker row."""
+        import inspect
+
+        from agent import turn_finalizer
+
+        src = inspect.getsource(turn_finalizer.finalize_turn)
+        micro_block = src.split("Post-turn micro-compaction", 1)[1]
+        micro_block = micro_block.split("agent._persist_session", 1)[0]
+        assert "_flush_scan_cursor_invalidated" in micro_block, (
+            "finalize_turn must consume the compressor's cursor-invalidation "
+            "flag raised by the defrag marker pop"
+        )
+        assert "agent._db_flush_scan_prefix = None" in micro_block, (
+            "finalize_turn must invalidate the bounded flush-scan cursor "
+            "when the defrag pop stripped a live marker's stamp"
+        )