From bcc2e6581819bebd46dfd594de4e08ff3e1a7965 Mon Sep 17 00:00:00 2001 From: leomcamilo Date: Wed, 2 Sep 2026 05:22:41 -0300 Subject: [PATCH] fix(state): quarantine SessionDB handle after structural corruption A bare SQLITE_CORRUPT/NOTADB on a live write (not FTS-scoped, not a replaced file) now sets a sticky per-instance flag: later writes fail fast with StateDbCorruptError, the handle never reopens after close(), and close() skips its explicit PASSIVE WAL checkpoint. Gateway and agent flush paths divert pending transcripts to JSONL/spool like the replaced case instead of retrying forever. Field evidence: a handle that kept writing for ~50 minutes after the first structural error checkpointed 15 pages under the wrong page numbers on shutdown (page 1 <- messages_fts_trigram_data leaf), turning "malformed" into "file is not a database". Refs #90837, #90950, #97940, #89332, #45383 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CNX8rNYHqA5pT4tAGSzXtb --- docs/state-db-recovery.md | 38 ++++ gateway/session.py | 19 +- hermes_state.py | 123 +++++++++- run_agent.py | 9 +- .../test_session_db_corrupt_fallback.py | 70 ++++++ .../test_state_db_corrupt_quarantine.py | 212 ++++++++++++++++++ .../test_flush_diverts_on_corrupt_state_db.py | 70 ++++++ .../test_turn_completion_explainer.py | 5 + tests/state/test_fts_runtime_rebuild.py | 23 +- tests/test_state_db_notadb_fail_closed.py | 7 +- 10 files changed, 565 insertions(+), 11 deletions(-) create mode 100644 tests/gateway/test_session_db_corrupt_fallback.py create mode 100644 tests/hermes_state/test_state_db_corrupt_quarantine.py create mode 100644 tests/run_agent/test_flush_diverts_on_corrupt_state_db.py diff --git a/docs/state-db-recovery.md b/docs/state-db-recovery.md index c56b5b5f7a..c776aaea9f 100644 --- a/docs/state-db-recovery.md +++ b/docs/state-db-recovery.md @@ -23,6 +23,44 @@ cross-process admission lock and foreign-holder guard. If that guarded rebuild cannot run, FTS remains detached, canonical writes stay available, and `hermes doctor` reports the explicit repair command. +## Live behavior when the file itself is corrupt + +If a live write reports bare `SQLITE_CORRUPT` / `SQLITE_NOTADB` (`database +disk image is malformed`, `file is not a database`) with no FTS provenance, +the damage is in a canonical B-tree, the schema, or the freelist. `SessionDB` +then quarantines that handle (`StateDbCorruptError`): + +1. the failing write propagates the typed error and nothing is retried; +2. later writes on the handle fail immediately without touching the file; +3. the handle never reopens its connection after `close()`; and +4. `close()` skips its explicit WAL checkpoint. + +Stopping the writes is the protection. In the field, a handle that kept +writing for ~50 minutes after the first structural error checkpointed 15 +pages under the wrong page numbers on shutdown (page 1 received a +`messages_fts_trigram_data` leaf) and turned a damaged-but-readable file into +one that no longer opened at all. Skipping the explicit checkpoint is the +second line of defence; SQLite may still run its own last-connection +checkpoint when the connection closes, so copy `state.db`, `state.db-wal` and +`state.db-shm` together before restarting anything. + +The gateway and the agent flush path treat the quarantine like a replaced +file: pending transcripts go to `sessions/.jsonl` and the gateway +`pending_messages/` spool instead of the retry queue, and the FTS one-shot +rebuild never runs on the damaged file. The quarantine is per process — the +shared handle stays poisoned for every holder until the process restarts on a +repaired or restored file. Do not run `hermes doctor --fix` while the gateway +is still up. Next steps: + +```bash +hermes gateway stop +HERMES_HOME="$HOME/.hermes" hermes sessions recover --source "$HOME/.hermes/state.db" --inspect-only +# if recoverable: +HERMES_HOME="$HOME/.hermes" hermes sessions recover --source "$HOME/.hermes/state.db" --output "$HOME/recovered-state.db" +``` + +or restore the newest snapshot from `state-snapshots/`. + ## Explicit repair Stop every process that can open the profile database before repairing it. diff --git a/gateway/session.py b/gateway/session.py index a6c76bb3d4..a38b9c965c 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -3981,14 +3981,23 @@ class SessionStore: try: self._append_transcript_message(session_id, msg) except Exception as exc: - from hermes_state import CompressionSessionClosedError, StateDbReplacedError + from hermes_state import ( + CompressionSessionClosedError, + StateDbCorruptError, + StateDbReplacedError, + ) - if isinstance(exc, StateDbReplacedError): + if isinstance(exc, (StateDbReplacedError, StateDbCorruptError)): + # Both classes mean "this handle must not touch the file + # again": replaced generation (#89332) or structural + # corruption (quarantine). Retrying cannot succeed, and + # the FTS one-shot rebuild below must never run on a + # damaged file. Divert instead. logger.error( - "Session DB was replaced underneath the gateway for %s; " - "stopping SQLite writes and diverting pending " + "Session DB refused further writes on this handle for " + "%s (%s); stopping SQLite writes and diverting pending " "transcripts to the on-disk fallback: %s", - session_id, exc, + session_id, type(exc).__name__, exc, ) with self._transcript_retry_lock: remaining = list(self._dirty_transcripts.get(queue_session_id, [])) diff --git a/hermes_state.py b/hermes_state.py index 3cd6f152af..3d2066c9b1 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -2265,6 +2265,8 @@ def classify_persistence_error(exc_or_str) -> str: return "compression" if isinstance(exc_or_str, StateDbReplacedError): return "replaced" + if isinstance(exc_or_str, StateDbCorruptError): + return "corrupt" text = str(exc_or_str).lower() if "turn lease" in text: return "turn_lease" @@ -4361,6 +4363,40 @@ _STATE_DB_REPLACED_MSG = ( ) +class StateDbCorruptError(sqlite3.DatabaseError): + """A live SessionDB observed structural (non-FTS) corruption and is quarantined. + + Raised once a write on this handle reports bare ``SQLITE_CORRUPT`` / + ``SQLITE_NOTADB`` that is neither FTS-scoped (``_is_fts_write_corruption_error``) + nor a replaced-file case (``StateDbReplacedError``). Subclasses + ``sqlite3.DatabaseError`` so every existing ``except sqlite3.Error`` + degrade path keeps working; ``sqlite_errorcode``/``sqlite_errorname`` + are copied from the originating error. + + The quarantine is sticky for the life of the handle: later writes fail + fast, the handle never reopens after ``close()``, and ``close()`` skips + its own WAL checkpoint. Field evidence (the #90837 lost/reordered-page + signature, the #90950 page-1 clobber): a handle that kept writing for ~50 + minutes after the first structural error checkpointed 15 pages under the + wrong page numbers on shutdown, turning a still-readable file into + ``file is not a database``. Stopping the writes is what prevents that; + skipping the explicit checkpoint is the second line of defence (SQLite + may still run its own last-connection checkpoint on close — Python's + ``sqlite3`` does not expose ``SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE``). The + recovery boundary is a process restart on a repaired or restored file. + """ + + +_STATE_DB_CORRUPT_MSG = ( + "FATAL: state.db reported structural corruption (database disk image is " + "malformed outside the FTS shadow tables) on a live handle; refusing further " + "writes, automatic reopen, and the close-time WAL checkpoint on this file. " + "Stop the gateway, then run `hermes sessions recover --source " + "--inspect-only` or restore a snapshot. Unwritten transcripts are diverted to " + "sessions/.jsonl (and the gateway pending_messages spool)." +) + + def divert_session_transcript_jsonl(session_id: str, messages) -> "Optional[Path]": """Append pending messages as JSON lines under HERMES_HOME/sessions. @@ -5242,6 +5278,12 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) self._db_file_application_id: int = 0 self._db_file_generation_token: str = "" self._db_replaced = False + # Sticky: set once a write on THIS handle reports bare SQLITE_CORRUPT / + # NOTADB that is not FTS-scoped and not a replaced-file case. Never + # cleared; the recovery boundary is a process restart on a repaired or + # restored file (see StateDbCorruptError). + self._db_corrupt = False + self._db_corrupt_reason = "" # One-shot guard for the usermerge-floor config write on the # incremental FTS merge cadence (see _merge_fts_incrementally). self._fts_usermerge_floor_applied = False @@ -5783,6 +5825,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) # through stale WAL/shm assumptions (#89332). Refuse instead. if self._db_replaced or self._db_file_was_replaced(): self._halt_db_replaced() + # A quarantined handle must never come back: reopening would hand a + # fresh connection (and its own close-time checkpoint) to a file we + # already know is structurally damaged. + if self._db_corrupt: + raise self._corrupt_error( + f"state.db connection for {self.db_path} is quarantined after " + f"structural corruption; refusing to reopen for a {context} " + "after close(). " + ) logger.warning( "state.db connection for %s was closed while a %s was still in " "flight — reopening (teardown/worker race, #94736)", @@ -6127,6 +6178,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) return "no more rows available" in str(exc).lower() while True: + self._raise_if_db_corrupt() self._raise_if_db_replaced() fn_started = False try: @@ -6229,6 +6281,11 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) # explicit repair paths retain rebuild ownership. if self._enter_fts_fail_open(exc): continue + # Bare SQLITE_CORRUPT / NOTADB that survived the replaced-file + # check and the FTS-scoped fail-open is structural damage: + # quarantine the handle (see StateDbCorruptError). + if self._is_structural_corruption_error(exc): + self._halt_db_corrupt(exc) raise except sqlite3.Error as exc: # Catch-all for builds that surface 'no more rows available' @@ -6324,6 +6381,54 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) if self._db_file_was_replaced(): self._halt_db_replaced() + @classmethod + def _is_structural_corruption_error(cls, exc: BaseException) -> bool: + """Bare SQLITE_CORRUPT/NOTADB with no FTS provenance. + + ``_is_fts_write_corruption_error`` is the positive FTS classifier; + everything else in the ``corrupt`` bucket of + ``classify_persistence_error`` is damage to a canonical B-tree, the + schema, or the freelist — never repairable from the live write path. + """ + if not isinstance(exc, sqlite3.DatabaseError): + return False + if isinstance(exc, StateDbCorruptError): + return False + if cls._is_fts_write_corruption_error(exc): + return False + return classify_persistence_error(exc) == "corrupt" + + def _corrupt_error(self, prefix: str = "") -> "StateDbCorruptError": + """Build the quarantine error for this handle (message assembled once).""" + return StateDbCorruptError( + f"{prefix}{_STATE_DB_CORRUPT_MSG} (cause: {self._db_corrupt_reason})" + ) + + def _halt_db_corrupt(self, exc: BaseException) -> None: + """Quarantine this handle and raise; never run in-file repair here.""" + self._db_corrupt = True + self._db_corrupt_reason = str(exc) + logger.error( + "state.db %s reported structural corruption outside the FTS " + "indexes (%s); quarantining this handle: no further writes, no " + "automatic reopen, no explicit WAL checkpoint at close. Stop the " + "gateway and run `hermes sessions recover --source %s " + "--inspect-only`.", + self.db_path, + exc, + self.db_path, + ) + err = self._corrupt_error() + for attr in ("sqlite_errorcode", "sqlite_errorname"): + value = getattr(exc, attr, None) + if value is not None: + setattr(err, attr, value) + raise err from exc + + def _raise_if_db_corrupt(self) -> None: + if self._db_corrupt: + raise self._corrupt_error() + def _sleep_before_write_retry( self, deadline: float, patience_s: float ) -> bool: @@ -6545,6 +6650,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) """ if not self._fts_enabled or not self._is_fts_write_corruption_error(exc): return False + self._raise_if_db_corrupt() if self._db_replaced or self._db_file_was_replaced(): self._halt_db_replaced() @@ -6612,6 +6718,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) databases (65K+ pages) due to the exclusive-lock I/O pressure from checkpointing thousands of frames at once (issue #45383). """ + if self._db_corrupt: + return # quarantined: never checkpoint over a damaged image try: with self._lock: result = self._conn.execute( @@ -6703,7 +6811,20 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) self._close_read_conn(conn) with self._lock: if self._conn: - if not self.read_only: + if self._db_corrupt: + # Quarantined handle (see StateDbCorruptError): no explicit + # checkpoint over a damaged page image. + logger.warning( + "Skipping the close-time WAL checkpoint for %s: this " + "handle observed structural corruption (%s). Take a " + "snapshot of state.db, -wal and -shm before restarting, " + "then run `hermes sessions recover --source %s " + "--inspect-only`.", + self.db_path, + self._db_corrupt_reason, + self.db_path, + ) + elif not self.read_only: # PASSIVE, not TRUNCATE. Every cron run_agent opens+closes a # transient SessionDB, so a TRUNCATE here fires a full WAL # reset many times/hour, racing the gateway's long-lived diff --git a/run_agent.py b/run_agent.py index 86b62db372..7b93887914 100644 --- a/run_agent.py +++ b/run_agent.py @@ -2670,13 +2670,17 @@ class AIAgent: # ("storage was busy, send it again") from disk-full/read-only. from hermes_state import ( CompressionSessionClosedError, + StateDbCorruptError, StateDbReplacedError, classify_persistence_error, divert_session_transcript_jsonl, ) self._last_persistence_error_cause = classify_persistence_error(e) - if isinstance(e, StateDbReplacedError): + if isinstance(e, (StateDbReplacedError, StateDbCorruptError)): + # Replaced generation or quarantined (structurally corrupt) + # handle: SQLite will not take this batch again, so keep it + # on disk instead of only in RAM. try: divert_session_transcript_jsonl( getattr(self, "session_id", "") or "", @@ -2684,7 +2688,8 @@ class AIAgent: ) except Exception: logger.warning( - "JSONL divert failed after state.db replace for %s", + "JSONL divert failed after state.db %s for %s", + self._last_persistence_error_cause, getattr(self, "session_id", None), exc_info=True, ) diff --git a/tests/gateway/test_session_db_corrupt_fallback.py b/tests/gateway/test_session_db_corrupt_fallback.py new file mode 100644 index 0000000000..9a6be9a114 --- /dev/null +++ b/tests/gateway/test_session_db_corrupt_fallback.py @@ -0,0 +1,70 @@ +"""Gateway SessionStore must divert, not retry forever, after structural corruption. + +Mirrors ``test_session_db_replaced_fallback.py``: once the SessionDB handle +is quarantined (``StateDbCorruptError``) the pending transcript goes to the +JSONL/spool fallback and no FTS surgery runs on the damaged file. +""" + +import json +import sqlite3 + +from gateway.config import GatewayConfig +from gateway.session import SessionStore + + +class _MalformedConn: + def __init__(self, real_conn): + self._real = real_conn + + def execute(self, *args, **kwargs): + raise sqlite3.DatabaseError("database disk image is malformed") + + def __getattr__(self, name): + return getattr(self._real, name) + + +def _assert_diverted(tmp_path, sid, needle): + pending = list((tmp_path / "pending_messages").glob("pending-*.json")) + assert pending, "expected pending_messages/pending-*.json spool" + spooled = False + for path in pending: + payload = json.loads(path.read_text(encoding="utf-8")) + message = (payload.get("data") or {}).get("message") or {} + if needle in str(message.get("content", "")): + spooled = True + break + assert spooled, f"{needle!r} missing from pending spool" + jsonl = tmp_path / "sessions" / f"{sid}.jsonl" + assert jsonl.is_file() + assert needle in jsonl.read_text(encoding="utf-8") + + +def test_corrupt_state_db_diverts_pending_without_fts_rebuild(tmp_path, monkeypatch): + import hermes_state + + live = tmp_path / "state.db" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", live) + + store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + sid = "gw-corrupt" + store._db.create_session(session_id=sid, source="cli") + store.append_to_transcript( + sid, {"role": "user", "content": "before", "timestamp": 1.0} + ) + real_conn = store._db._conn + store._db._conn = _MalformedConn(real_conn) + try: + store.append_to_transcript( + sid, {"role": "user", "content": "after-corrupt", "timestamp": 2.0} + ) + assert store._db._db_corrupt is True + # No FTS surgery ran on either layer. + assert store._db._fts_enabled is True + assert store._db._fts_stale is False + assert store._fts_rebuild_attempted is False + assert sid not in store._dirty_transcripts + _assert_diverted(tmp_path, sid, "after-corrupt") + finally: + store._db._conn = real_conn + store.close_all_db_handles() diff --git a/tests/hermes_state/test_state_db_corrupt_quarantine.py b/tests/hermes_state/test_state_db_corrupt_quarantine.py new file mode 100644 index 0000000000..258672bb77 --- /dev/null +++ b/tests/hermes_state/test_state_db_corrupt_quarantine.py @@ -0,0 +1,212 @@ +"""Quarantine of a live SessionDB handle after structural (non-FTS) corruption. + +Field evidence (the #90837 lost/reordered-page-write class): a gateway kept +retrying writes for ~50 minutes after ``gateway_routing`` reported +``database disk image is malformed``; on SIGTERM the close-time +``PRAGMA wal_checkpoint(PASSIVE)`` then wrote 15 pages to the wrong page +numbers (page 1 received a ``messages_fts_trigram_data`` leaf) and the file +stopped opening at all. Once structural corruption is observed on a handle +the only safe policy is to stop touching the file. +""" + +import sqlite3 + +import pytest + +from hermes_state import SessionDB, StateDbCorruptError + + +class _MalformedConn: + """Connection proxy whose every execute reports bare SQLITE_CORRUPT.""" + + def __init__(self, real_conn): + self._real = real_conn + + def execute(self, *args, **kwargs): + raise sqlite3.DatabaseError("database disk image is malformed") + + def __getattr__(self, name): + return getattr(self._real, name) + + +class TestQuarantineAfterStructuralCorruption: + def test_structural_corruption_sets_sticky_flag_and_raises_typed(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + real_conn = db._conn + try: + db.create_session(session_id="s1", source="cli", model="test") + db._conn = _MalformedConn(real_conn) + with pytest.raises(StateDbCorruptError, match="malformed") as excinfo: + db.create_session(session_id="s2", source="cli", model="test") + assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError) + assert db._db_corrupt is True + # Structural damage must never be mistaken for FTS-scoped damage. + assert db._fts_stale is False + finally: + db._conn = real_conn + db.close() + + +class _RecordingConn: + """Connection proxy that records every SQL text and delegates.""" + + def __init__(self, real_conn): + self._real = real_conn + self.recorded = [] + + def execute(self, sql, *args, **kwargs): + self.recorded.append(str(sql)) + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + +def _quarantined_db(tmp_path): + """A SessionDB whose first corrupt write already tripped the quarantine.""" + db = SessionDB(db_path=tmp_path / "state.db") + real_conn = db._conn + db.create_session(session_id="s1", source="cli", model="test") + db._conn = _MalformedConn(real_conn) + with pytest.raises(StateDbCorruptError): + db.create_session(session_id="s2", source="cli", model="test") + db._conn = real_conn + assert db._db_corrupt is True + return db, real_conn + + +class TestQuarantinedHandleStopsTouchingTheFile: + def test_subsequent_writes_fail_fast_without_touching_connection(self, tmp_path): + db, real_conn = _quarantined_db(tmp_path) + recorder = _RecordingConn(real_conn) + db._conn = recorder + try: + with pytest.raises(StateDbCorruptError): + db.create_session(session_id="s3", source="cli", model="test") + assert recorder.recorded == [] + finally: + db._conn = real_conn + db.close() + + def test_close_skips_wal_checkpoint_when_quarantined(self, tmp_path, caplog): + db, real_conn = _quarantined_db(tmp_path) + recorder = _RecordingConn(real_conn) + db._conn = recorder + with caplog.at_level("WARNING", logger="hermes_state"): + db.close() + assert not any("wal_checkpoint" in sql for sql in recorder.recorded) + assert db._conn is None + assert any( + "Skipping the close-time WAL checkpoint" in rec.getMessage() + and "hermes sessions recover" in rec.getMessage() + for rec in caplog.records + ) + + def test_reopen_after_close_refused_when_quarantined(self, tmp_path, monkeypatch): + from unittest.mock import MagicMock + + db, real_conn = _quarantined_db(tmp_path) + db.close() + reopen = MagicMock() + monkeypatch.setattr("hermes_state._connect_tracked_db", reopen) + with pytest.raises(StateDbCorruptError, match="structural corruption"): + db.create_session(session_id="s4", source="cli", model="test") + reopen.assert_not_called() + # The read fallback after close() goes through the same reopen path. + with pytest.raises(StateDbCorruptError, match="refusing to reopen"): + db.get_session("s1") + reopen.assert_not_called() + + +class TestQuarantineScope: + def test_fts_scoped_corruption_does_not_trip_flag(self, tmp_path): + """Corrupt FTS shadow tables keep the existing fail-open detach path.""" + path = tmp_path / "state.db" + db = SessionDB(db_path=path) + db.create_session(session_id="s1", source="cli", model="test") + db.append_message("s1", role="user", content="hello world") + raw = sqlite3.connect(str(path)) + raw.execute( + "UPDATE messages_fts_data SET block = X'DEADBEEFDEADBEEFDEADBEEFDEADBEEF'" + ) + raw.commit() + raw.close() + try: + db.append_message("s1", role="user", content="healed append") + assert db._db_corrupt is False + assert db._fts_stale is True + assert db._fts_enabled is False + finally: + db.close() + + def test_replaced_file_takes_precedence_over_corrupt(self, tmp_path): + import os + + from hermes_state import StateDbReplacedError + + live = tmp_path / "state.db" + other = tmp_path / "other.db" + db = SessionDB(db_path=live) + real_conn = db._conn + try: + db.create_session(session_id="s1", source="cli", model="test") + if db._db_file_identity is None: + pytest.skip("filesystem does not expose st_dev/st_ino") + alt = SessionDB(db_path=other) + alt.create_session("other", "cli") + alt.close() + os.replace(other, live) + db._conn = _MalformedConn(real_conn) + with pytest.raises(StateDbReplacedError): + db.create_session(session_id="s2", source="cli", model="test") + assert db._db_replaced is True + assert db._db_corrupt is False + finally: + db._conn = real_conn + db.close() + + def test_classify_persistence_error_maps_quarantine_to_corrupt(self): + from hermes_state import _STATE_DB_CORRUPT_MSG, classify_persistence_error + + assert classify_persistence_error(StateDbCorruptError("x")) == "corrupt" + # The stringified form (RPC boundaries) must classify the same way. + assert classify_persistence_error(_STATE_DB_CORRUPT_MSG) == "corrupt" + + +@pytest.fixture +def _clean_registry(): + import hermes_state_registry as registry + + registry.close_all() + registry._generations.clear() + registry._retired.clear() + yield registry + registry.close_all() + registry._generations.clear() + registry._retired.clear() + + +class TestSharedRegistry: + def test_holders_share_quarantine_and_close_all_skips_checkpoint( + self, tmp_path, _clean_registry + ): + registry = _clean_registry + path = tmp_path / "state.db" + holder_a = registry.acquire(path) + holder_b = registry.acquire(path) + assert holder_a is holder_b + real_conn = holder_a._conn + holder_a.create_session(session_id="s1", source="cli", model="test") + + holder_a._conn = _MalformedConn(real_conn) + with pytest.raises(StateDbCorruptError): + holder_a.create_session(session_id="s2", source="cli", model="test") + recorder = _RecordingConn(real_conn) + holder_b._conn = recorder + + with pytest.raises(StateDbCorruptError): + holder_b.create_session(session_id="s3", source="cli", model="test") + + registry.close_all() + assert not any("wal_checkpoint" in sql for sql in recorder.recorded) + assert holder_a._conn is None diff --git a/tests/run_agent/test_flush_diverts_on_corrupt_state_db.py b/tests/run_agent/test_flush_diverts_on_corrupt_state_db.py new file mode 100644 index 0000000000..26e941c495 --- /dev/null +++ b/tests/run_agent/test_flush_diverts_on_corrupt_state_db.py @@ -0,0 +1,70 @@ +"""Agent flush path: a quarantined (structurally corrupt) SessionDB diverts to JSONL. + +Mirrors the replaced-file contract: the batch that SQLite will never take +again is kept on disk under ``sessions/.jsonl`` instead of only in RAM, +the flush fails closed (no retry loop), and the turn-end explanation gets the +``corrupt`` cause. +""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +from hermes_state import SessionDB, StateDbCorruptError +from run_agent import AIAgent + + +def _flush_agent(db, session_id): + agent = SimpleNamespace( + _session_db=db, + _session_db_created=True, + _persist_disabled=False, + session_id=session_id, + _session_persist_lock=None, + _flushed_db_message_ids=set(), + _flushed_db_message_session_id=None, + _last_flushed_db_idx=0, + _db_flush_scan_prefix=None, + _persist_user_message_idx=None, + _persist_user_message_override=None, + _persist_user_message_timestamp=None, + _pending_cli_user_message=None, + _active_session_turn_lease_holder=None, + _last_persistence_error_cause=None, + _compression_adoption_failed=False, + ) + agent._ensure_db_session = lambda: None + agent._flush_messages_to_session_db = ( + AIAgent._flush_messages_to_session_db.__get__(agent, AIAgent) + ) + agent._flush_messages_to_session_db_unlocked = ( + AIAgent._flush_messages_to_session_db_unlocked.__get__(agent, AIAgent) + ) + return agent + + +def test_flush_diverts_batch_to_jsonl_when_handle_is_quarantined( + tmp_path: Path, monkeypatch +) -> None: + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + db = SessionDB(db_path=tmp_path / "state.db") + try: + db.create_session("live", source="cli") + agent = _flush_agent(db, "live") + + def _quarantined(self, *, session_id, messages, **kwargs): + raise StateDbCorruptError("database disk image is malformed (quarantined)") + + monkeypatch.setattr(SessionDB, "append_messages_batch", _quarantined) + + messages = [{"role": "user", "content": "kept-on-disk-after-corruption"}] + result = agent._flush_messages_to_session_db(messages, []) + + assert result is False + assert agent._last_persistence_error_cause == "corrupt" + jsonl = tmp_path / "sessions" / "live.jsonl" + assert jsonl.is_file() + assert "kept-on-disk-after-corruption" in jsonl.read_text(encoding="utf-8") + finally: + db.close() diff --git a/tests/run_agent/test_turn_completion_explainer.py b/tests/run_agent/test_turn_completion_explainer.py index 99b052c7f9..3a7f6b62e0 100644 --- a/tests/run_agent/test_turn_completion_explainer.py +++ b/tests/run_agent/test_turn_completion_explainer.py @@ -438,3 +438,8 @@ def test_run_conversation_partial_stream_recovery_surfaces_explanation(): assert result["response_previewed"] is False +def test_classify_persistence_error_quarantined_handle_is_corrupt() -> None: + """A quarantined SessionDB raises the typed error; it stays in the corrupt bucket.""" + from hermes_state import StateDbCorruptError, classify_persistence_error + + assert classify_persistence_error(StateDbCorruptError("quarantined")) == "corrupt" diff --git a/tests/state/test_fts_runtime_rebuild.py b/tests/state/test_fts_runtime_rebuild.py index 8779186e72..ac030f4cb6 100644 --- a/tests/state/test_fts_runtime_rebuild.py +++ b/tests/state/test_fts_runtime_rebuild.py @@ -332,7 +332,15 @@ class TestRuntimeFtsRebuild: with pytest.raises(sqlite3.DatabaseError) as caught: db._execute_write(lambda _conn: (_ for _ in ()).throw(structural)) - assert caught.value is structural + # Structural corruption quarantines the handle: the typed error wraps + # the original (cause preserved, SQLite result code copied) and the + # sticky flag is set, so later writes fail fast. + from hermes_state import StateDbCorruptError + + assert isinstance(caught.value, StateDbCorruptError) + assert caught.value.__cause__ is structural + assert caught.value.sqlite_errorcode == sqlite3.SQLITE_CORRUPT + assert db._db_corrupt is True assert rebuild_called is False assert db._fts_stale is False assert _meta_value(tmp_path / "state.db", FTS_STALE_KEY) is None @@ -831,6 +839,19 @@ class TestPhysicalCorruptionAcceptance: # The misdiagnosis message from the field incident must be gone. assert "canonical message rows are preserved" not in caplog.text assert "attempting one-shot in-place FTS rebuild" not in caplog.text + # Structural damage quarantines the handle: typed error, sticky + # flag, later writes fail fast, and close() must not checkpoint + # the WAL over a damaged page image (the #90950 page-1 clobber). + from hermes_state import StateDbCorruptError + + assert isinstance(caught.value, StateDbCorruptError) + assert db._db_corrupt is True + with pytest.raises(StateDbCorruptError): + db.append_message("s1", "user", "second write after corruption") + caplog.clear() + with caplog.at_level("WARNING", logger="hermes_state"): + db.close() + assert "Skipping the close-time WAL checkpoint" in caplog.text finally: db.close() diff --git a/tests/test_state_db_notadb_fail_closed.py b/tests/test_state_db_notadb_fail_closed.py index 6f5c787ea0..9fce19e3e1 100644 --- a/tests/test_state_db_notadb_fail_closed.py +++ b/tests/test_state_db_notadb_fail_closed.py @@ -14,7 +14,7 @@ from unittest.mock import MagicMock import pytest -from hermes_state import SessionDB, _on_disk_journal_mode +from hermes_state import SessionDB, StateDbCorruptError, _on_disk_journal_mode class _NotADbOnce: @@ -42,9 +42,12 @@ class TestFailClosedAfterNotADb: reopen = MagicMock() monkeypatch.setattr("hermes_state._connect_tracked_db", reopen) db._conn = _NotADbOnce(real_conn) - with pytest.raises(sqlite3.DatabaseError, match="not a database"): + with pytest.raises(sqlite3.DatabaseError, match="not a database") as excinfo: db.create_session(session_id="s2", source="cli", model="test") reopen.assert_not_called() + # NOTADB on a live write is structural: the handle is quarantined. + assert isinstance(excinfo.value, StateDbCorruptError) + assert db._db_corrupt is True finally: db._conn = real_conn db.close()