From 71256dfd013b045d5bc0c7c21ada71a29cd566d7 Mon Sep 17 00:00:00 2001 From: rainbowgits <164521089+rainbowgits@users.noreply.github.com> Date: Tue, 18 Aug 2026 20:54:14 +0300 Subject: [PATCH] fix(state): fail loudly when state.db is replaced under a live process Detect same-inode cp via a generation stamp, halt FTS repair, and divert unwritten transcripts to sessions/.jsonl plus the gateway pending spool. Co-authored-by: Cursor --- agent/turn_finalizer.py | 2 +- gateway/session.py | 38 +++- hermes_state.py | 199 ++++++++++++++++++ run_agent.py | 25 +++ .../test_session_db_replaced_fallback.py | 97 +++++++++ .../test_state_db_file_identity.py | 177 ++++++++++++++++ .../test_turn_completion_explainer.py | 12 ++ 7 files changed, 548 insertions(+), 2 deletions(-) create mode 100644 tests/gateway/test_session_db_replaced_fallback.py create mode 100644 tests/hermes_state/test_state_db_file_identity.py diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index bc279c767a..193c56461d 100644 --- a/agent/turn_finalizer.py +++ b/agent/turn_finalizer.py @@ -752,7 +752,7 @@ def finalize_turn( "health (`hermes doctor`), then send your message again" ) # Machine-readable cause for the gateway/desktop: exactly - # 'session_persistence_failed:'. + # 'session_persistence_failed:'. # Never clobber a failure_reason another path already stamped. if "failure_reason" not in result: _cause = getattr(agent, "_last_persistence_error_cause", None) diff --git a/gateway/session.py b/gateway/session.py index 32a343cced..facdb6f4d8 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -3756,7 +3756,43 @@ class SessionStore: try: self._append_transcript_message(session_id, msg) except Exception as exc: - from hermes_state import CompressionSessionClosedError + from hermes_state import CompressionSessionClosedError, StateDbReplacedError + + if isinstance(exc, StateDbReplacedError): + logger.error( + "Session DB was replaced underneath the gateway for %s; " + "stopping SQLite writes and diverting pending " + "transcripts to the on-disk fallback: %s", + session_id, exc, + ) + with self._transcript_retry_lock: + remaining = list(self._dirty_transcripts.get(queue_session_id, [])) + self._dirty_transcripts.pop(queue_session_id, None) + self._transcript_append_failures.pop(session_id, None) + for dropped in remaining: + try: + from gateway.shutdown_flush import ( + spool_dropped_transcript_message, + ) + spool_dropped_transcript_message(session_id, dropped) + except Exception: + logger.warning( + "pending fallback failed for replaced " + "state.db transcript on %s", + session_id, + exc_info=True, + ) + try: + from hermes_state import divert_session_transcript_jsonl + divert_session_transcript_jsonl(session_id, remaining) + except Exception: + logger.warning( + "JSONL divert failed for replaced state.db " + "transcript on %s", + session_id, + exc_info=True, + ) + return if isinstance(exc, CompressionSessionClosedError): # Resolve the full continuation chain via the canonical diff --git a/hermes_state.py b/hermes_state.py index c3dea52e27..e9cfa94349 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -26,9 +26,11 @@ import queue import random import re import sqlite3 +import struct import sys import threading import time +import uuid import weakref from collections import deque from contextlib import contextmanager @@ -2128,6 +2130,7 @@ PERSISTENCE_ERROR_CAUSES = ( "compression_closed", "turn_lease", "corrupt", + "replaced", "disk", "unknown", ) @@ -2173,6 +2176,9 @@ def classify_persistence_error(exc_or_str) -> str: (``database disk image is malformed`` / SQLITE_NOTADB). Distinct from ``"disk"``: freeing space cannot help, the user needs the repair path (``hermes doctor`` / automatic schema surgery). + * ``"replaced"`` — the ``state.db`` path no longer names the file this + process opened (out-of-band ``cp``/``mv``/restore). In-file FTS repair + cannot help; writes to the live handle must stop. * ``"disk"`` — disk full / read-only / permission-shaped failures (delegates the disk-full patterns to :func:`is_disk_full_error` so the two classifiers can never drift apart — e.g. ENOSPC). @@ -2191,6 +2197,8 @@ def classify_persistence_error(exc_or_str) -> str: return "compression_closed" if isinstance(exc_or_str, CompressionSessionBusyError): return "compression" + if isinstance(exc_or_str, StateDbReplacedError): + return "replaced" text = str(exc_or_str).lower() if "turn lease" in text: return "turn_lease" @@ -2198,6 +2206,8 @@ def classify_persistence_error(exc_or_str) -> str: return "compression_closed" if "being compressed" in text or "compression lease" in text: return "compression" + if "was replaced underneath" in text: + return "replaced" # Structural corruption BEFORE the lock and disk buckets: "database disk # image is malformed" contains "disk" (and some wrapped corruption # strings mention "locked" recovery attempts), so later buckets would @@ -4171,6 +4181,82 @@ class SessionTurnLeaseLostError(RuntimeError): """ +class StateDbReplacedError(RuntimeError): + """The state.db path no longer names the file this SessionDB opened. + + Raised when an out-of-band ``cp``/``mv``/restore replaces the database + under a live gateway. In-place FTS repair and fail-open trigger + dropping cannot fix a generation mismatch; they amplify it. + """ + + +# SQLite header: 4-byte big-endian application_id at offset 68. Distinct from +# inode: ``cp`` onto the same path keeps st_ino and truncates+rewrites. +_STATE_DB_APPLICATION_ID_OFFSET = 68 +_STATE_DB_GENERATION_KEY = "db_file_generation" +_STATE_DB_REPLACED_MSG = ( + "FATAL: state.db was replaced underneath the gateway; refusing further " + "writes to this file. Divert transcripts to sessions/.jsonl (and the " + "gateway pending_messages spool) and restore or reopen after operator " + "intervention." +) + + +def divert_session_transcript_jsonl(session_id: str, messages) -> "Optional[Path]": + """Append pending messages as JSON lines under HERMES_HOME/sessions. + + Used when state.db is replaced under a live process so the current + turn is not only in RAM. Returns the jsonl path, or None when there + is nothing to write. + """ + sid = str(session_id or "").strip() + if not sid or not messages: + return None + sessions_dir = get_hermes_home() / "sessions" + sessions_dir.mkdir(parents=True, exist_ok=True) + path = sessions_dir / f"{sid}.jsonl" + with path.open("a", encoding="utf-8") as handle: + for msg in messages: + if isinstance(msg, dict): + handle.write(json.dumps(msg, ensure_ascii=False, default=str) + "\n") + elif msg is not None: + handle.write(json.dumps({"content": str(msg)}, ensure_ascii=False) + "\n") + return path + + +def _read_sqlite_application_id(db_path: Path) -> "Optional[int]": + """Read application_id from the SQLite header without opening a connection.""" + try: + with db_path.open("rb") as handle: + header = handle.read(_STATE_DB_APPLICATION_ID_OFFSET + 4) + except OSError: + return None + if len(header) < _STATE_DB_APPLICATION_ID_OFFSET + 4: + return None + if header[:16] != b"SQLite format 3\x00": + return None + return int( + struct.unpack( + ">I", + header[_STATE_DB_APPLICATION_ID_OFFSET:_STATE_DB_APPLICATION_ID_OFFSET + 4], + )[0] + ) + + +def _stat_db_file_identity(path: Path) -> "Optional[tuple]": + """Return ``(st_dev, st_ino)`` for *path*, or None when identity is unavailable.""" + try: + st = os.stat(path) + except OSError: + return None + # Windows volumes (and some network FS) report st_ino=0; a (0, 0) + # identity would false-positive every check. Skip the inode half of + # the guard there; generation stamp still applies. + if not st.st_dev or not st.st_ino: + return None + return (st.st_dev, st.st_ino) + + def _connect_tracked_db(path, tracking_path=None, **kwargs): """``sqlite3.connect`` that registers the open fd for lock-safety. @@ -4884,6 +4970,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) self._read_open_failed_at = 0.0 self._wal_active = False self._write_count = 0 + # File identity of the state.db this instance opened. Compared on + # every write (and before FTS fail-open / reopen-after-close) so an + # out-of-band replace cannot limp through in-place surgery. + # Inode catches mv/new-file; application_id catches cp onto the + # same path (same inode, truncate+rewrite). + self._db_file_identity: Optional[tuple] = None + self._db_file_application_id: int = 0 + self._db_file_generation_token: str = "" + self._db_replaced = False # 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 @@ -4958,6 +5053,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) except Exception: pass raise + self._record_db_file_identity() initialization_complete = True return @@ -5115,6 +5211,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) # racing session lifecycle and the surprise disk/latency cost on # an unattended open. (An interrupted optimize resumes when the # user re-runs the command.) + self._ensure_db_file_generation() + self._record_db_file_identity() initialization_complete = True except Exception as exc: # Capture the cause so /resume and friends can surface WHY the @@ -5391,6 +5489,12 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) f"SessionDB for {self.db_path} was closed (read-only handle); " f"cannot serve a {context} after close()" ) + # A reopen resolves the PATH again — if the file at that path is no + # longer the one this instance originally opened (out-of-band + # restore/cp/mv), reconnecting would write into the new generation + # through stale WAL/shm assumptions (#89332). Refuse instead. + if self._db_replaced or self._db_file_was_replaced(): + self._halt_db_replaced() logger.warning( "state.db connection for %s was closed while a %s was still in " "flight — reopening (teardown/worker race, #94736)", @@ -5728,6 +5832,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) return "no more rows available" in str(exc).lower() while True: + self._raise_if_db_replaced() try: with self._lock: if self._conn is None: @@ -5793,6 +5898,16 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) except sqlite3.DatabaseError as exc: if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s): continue + # An out-of-band replace of state.db (restore/cp/mv under a + # live process) surfaces as this same corruption error class. + # In-file repair on a NEW file generation amplifies the + # damage (#89332) — halt writes on this handle instead. + if ( + "not a database" in str(exc).lower() + or is_malformed_db_error(exc) + or self._is_fts_write_corruption_error(exc) + ): + self._raise_if_db_replaced() # 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. Never run a @@ -5814,6 +5929,88 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) continue raise + def _ensure_db_file_generation(self) -> None: + """Mint a once-per-file generation stamp (state_meta + application_id). + + First opener wins via INSERT OR IGNORE. application_id is written + only when still 0 so racers converge on the same header value. + PASSIVE checkpoint only — never TRUNCATE (#45383). + """ + if self.read_only or self._conn is None: + return + token = uuid.uuid4().hex + try: + self._conn.execute( + "INSERT OR IGNORE INTO state_meta (key, value) VALUES (?, ?)", + (_STATE_DB_GENERATION_KEY, token), + ) + row = self._conn.execute( + "SELECT value FROM state_meta WHERE key = ?", + (_STATE_DB_GENERATION_KEY,), + ).fetchone() + if row and row[0]: + token = str(row[0]) + self._db_file_generation_token = token + current = 0 + pragma_row = self._conn.execute("PRAGMA application_id").fetchone() + if pragma_row: + current = int(pragma_row[0] or 0) + if current == 0: + app_id = int(token[:8], 16) & 0x7FFFFFFF + if app_id == 0: + app_id = 1 + self._conn.execute(f"PRAGMA application_id={app_id}") + current = app_id + self._db_file_application_id = current + try: + self._conn.execute("PRAGMA wal_checkpoint(PASSIVE)") + except sqlite3.Error: + pass + except sqlite3.Error as exc: + logger.debug("state.db generation stamp skipped: %s", exc) + + def _record_db_file_identity(self) -> None: + """Snapshot inode plus the on-disk generation header when present.""" + self._db_file_identity = _stat_db_file_identity(self.db_path) + disk_id = _read_sqlite_application_id(self.db_path) + if disk_id: + self._db_file_application_id = disk_id + elif self._conn is not None and not self._db_file_application_id: + try: + pragma_row = self._conn.execute("PRAGMA application_id").fetchone() + if pragma_row and pragma_row[0]: + self._db_file_application_id = int(pragma_row[0]) + except sqlite3.Error: + pass + + def _db_file_was_replaced(self) -> bool: + """True when the path no longer names the file this instance opened.""" + recorded = self._db_file_identity + if recorded is not None: + current = _stat_db_file_identity(self.db_path) + if current is None or current != recorded: + return True + recorded_app = int(self._db_file_application_id or 0) + if recorded_app: + disk_app = _read_sqlite_application_id(self.db_path) + # Header 0 means the WAL has not been checkpointed yet — not a + # replace. A copied Hermes DB that minted its own id is nonzero. + if disk_app and disk_app != recorded_app: + return True + return False + + def _halt_db_replaced(self) -> None: + """Stop writes and raise; do not run in-file repair on a new generation.""" + self._db_replaced = True + logger.error(_STATE_DB_REPLACED_MSG) + raise StateDbReplacedError(_STATE_DB_REPLACED_MSG) + + def _raise_if_db_replaced(self) -> None: + if self._db_replaced: + raise StateDbReplacedError(_STATE_DB_REPLACED_MSG) + if self._db_file_was_replaced(): + self._halt_db_replaced() + def _sleep_before_write_retry( self, deadline: float, patience_s: float ) -> bool: @@ -6035,6 +6232,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) """ if not self._fts_enabled or not self._is_fts_write_corruption_error(exc): return False + if self._db_replaced or self._db_file_was_replaced(): + self._halt_db_replaced() try: with self._lock: diff --git a/run_agent.py b/run_agent.py index aaca141527..34712ec919 100644 --- a/run_agent.py +++ b/run_agent.py @@ -2550,10 +2550,24 @@ class AIAgent: # ("storage was busy, send it again") from disk-full/read-only. from hermes_state import ( CompressionSessionClosedError, + StateDbReplacedError, classify_persistence_error, + divert_session_transcript_jsonl, ) self._last_persistence_error_cause = classify_persistence_error(e) + if isinstance(e, StateDbReplacedError): + try: + divert_session_transcript_jsonl( + getattr(self, "session_id", "") or "", + _batch_rows, + ) + except Exception: + logger.warning( + "JSONL divert failed after state.db replace for %s", + getattr(self, "session_id", None), + exc_info=True, + ) if isinstance(e, CompressionSessionClosedError): # Compression race: another path rotated this session while # this turn was still writing against it. The store resolves @@ -4129,6 +4143,17 @@ class AIAgent: "database). Your message should already be saved — " "please send it again in a moment." ) + if cause == "replaced": + return ( + prefix + + "the turn was stopped because the state database file " + "was replaced underneath this process. Do not run " + "`hermes doctor --fix` or in-place FTS repair — stop " + "the process, restore the intended state.db, then " + "restart. Unwritten messages were diverted to " + "sessions/.jsonl and, on the gateway, " + "pending_messages/pending-*.json." + ) if cause == "corrupt": return ( prefix diff --git a/tests/gateway/test_session_db_replaced_fallback.py b/tests/gateway/test_session_db_replaced_fallback.py new file mode 100644 index 0000000000..3f39e29824 --- /dev/null +++ b/tests/gateway/test_session_db_replaced_fallback.py @@ -0,0 +1,97 @@ +"""Gateway SessionStore must not FTS-repair a replaced state.db (#89332).""" + +import json +import os +import shutil + +import pytest + +from gateway.config import GatewayConfig +from gateway.session import SessionStore +from hermes_state import SessionDB + + +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_replaced_state_db_diverts_pending_without_fts_rebuild(tmp_path, monkeypatch): + import hermes_state + + live = tmp_path / "state.db" + other = tmp_path / "other.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-replaced" + store._db.create_session(session_id=sid, source="cli") + store.append_to_transcript( + sid, {"role": "user", "content": "before", "timestamp": 1.0} + ) + if store._db._db_file_identity is None: + store.close_all_db_handles() + 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) + + store.append_to_transcript( + sid, {"role": "user", "content": "after-replace", "timestamp": 2.0} + ) + + assert store._db._db_replaced is True + assert store._db._fts_runtime_rebuild_attempted is False + assert store._fts_rebuild_attempted is False + _assert_diverted(tmp_path, sid, "after-replace") + store.close_all_db_handles() + + +def test_copyfile_replaced_state_db_diverts_pending_without_fts_rebuild( + tmp_path, monkeypatch +): + import hermes_state + + live = tmp_path / "state.db" + other = tmp_path / "other.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-cp-replaced" + store._db.create_session(session_id=sid, source="cli") + store.append_to_transcript( + sid, {"role": "user", "content": "before-cp", "timestamp": 1.0} + ) + if not store._db._db_file_application_id: + store.close_all_db_handles() + pytest.skip("generation stamp not recorded") + + alt = SessionDB(db_path=other) + alt.create_session("other", "cli") + alt.close() + shutil.copyfile(other, live) + + store.append_to_transcript( + sid, {"role": "user", "content": "after-cp", "timestamp": 2.0} + ) + + assert store._db._db_replaced is True + assert store._db._fts_runtime_rebuild_attempted is False + assert store._fts_rebuild_attempted is False + _assert_diverted(tmp_path, sid, "after-cp") + store.close_all_db_handles() diff --git a/tests/hermes_state/test_state_db_file_identity.py b/tests/hermes_state/test_state_db_file_identity.py new file mode 100644 index 0000000000..645306dc4d --- /dev/null +++ b/tests/hermes_state/test_state_db_file_identity.py @@ -0,0 +1,177 @@ +"""File-identity guard on SessionDB writes (#89332). + +When state.db is replaced out-of-band under a live handle, in-place FTS +rebuild / fail-open cannot help: they operate on a generation mismatch. +The store must fail loudly instead of limping. +""" + +import json +import os +import shutil +import sqlite3 +from pathlib import Path + +import pytest + +from hermes_state import ( + SessionDB, + StateDbReplacedError, + classify_persistence_error, + divert_session_transcript_jsonl, +) + + +def _make_db(path: Path, session_id: str, content: str) -> SessionDB: + db = SessionDB(db_path=path) + db.create_session(session_id, "cli") + db.append_message(session_id, role="user", content=content) + return db + + +def _require_identity(db: SessionDB) -> None: + if db._db_file_identity is None: + pytest.skip("filesystem does not expose st_dev/st_ino for identity checks") + + +def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path): + live = tmp_path / "state.db" + other = tmp_path / "other.db" + db = _make_db(live, "live-sess", "original") + _require_identity(db) + alt = _make_db(other, "other-sess", "replacement") + alt.close() + + recorded = db._db_file_identity + assert recorded is not None + os.replace(other, live) + assert _stat_changed(live, recorded) + + with pytest.raises(StateDbReplacedError, match="replaced underneath"): + db.append_message("live-sess", role="user", content="after-replace") + + assert db._db_replaced is True + assert db._fts_runtime_rebuild_attempted is False + assert db._fts_enabled is True + db.close() + + +def test_second_write_after_halt_does_not_attempt_repair(tmp_path): + live = tmp_path / "state.db" + other = tmp_path / "other.db" + db = _make_db(live, "s", "a") + _require_identity(db) + alt = _make_db(other, "t", "b") + alt.close() + os.replace(other, live) + with pytest.raises(StateDbReplacedError): + db.append_message("s", role="user", content="first") + with pytest.raises(StateDbReplacedError): + db.append_message("s", role="user", content="second") + assert db._fts_runtime_rebuild_attempted is False + db.close() + + +def test_same_file_fts_corruption_still_rebuilds(tmp_path): + """Identity guard must not disable genuine in-file FTS recovery.""" + db = _make_db(tmp_path / "state.db", "s1", "hello world") + _require_identity(db) + identity = db._db_file_identity + raw = sqlite3.connect(str(tmp_path / "state.db")) + raw.execute( + "UPDATE messages_fts_data SET block = X'DEADBEEFDEADBEEFDEADBEEFDEADBEEF'" + ) + raw.commit() + raw.close() + db.append_message("s1", role="user", content="healed append") + assert db._db_file_identity == identity + assert db._db_replaced is False + assert db._fts_runtime_rebuild_attempted is True + db.close() + + +def test_classify_replaced_is_not_disk_or_fts_repair(): + err = StateDbReplacedError( + "FATAL: state.db was replaced underneath the gateway; refusing further writes" + ) + assert classify_persistence_error(err) == "replaced" + assert classify_persistence_error(str(err)) == "replaced" + + +def test_new_sessiondb_on_replaced_path_records_new_identity(tmp_path): + live = tmp_path / "state.db" + other = tmp_path / "other.db" + db = _make_db(live, "s", "a") + old_id = db._db_file_identity + _require_identity(db) + db.close() + alt = _make_db(other, "t", "b") + alt.close() + os.replace(other, live) + reopened = SessionDB(db_path=live) + try: + assert reopened._db_file_identity != old_id + reopened.append_message("t", role="user", content="adopted after reopen") + assert reopened._db_replaced is False + finally: + reopened.close() + + +def test_malformed_error_on_replaced_file_skips_fts_rebuild(tmp_path): + """Even if SQLite surfaces malformed, identity mismatch blocks repair.""" + live = tmp_path / "state.db" + other = tmp_path / "other.db" + db = _make_db(live, "s", "a") + _require_identity(db) + alt = _make_db(other, "t", "b") + alt.close() + os.replace(other, live) + + with pytest.raises(StateDbReplacedError): + db._try_runtime_fts_rebuild( + sqlite3.DatabaseError("database disk image is malformed") + ) + assert db._fts_runtime_rebuild_attempted is False + db.close() + + +def test_copyfile_same_inode_fails_loudly_without_fts_repair(tmp_path): + """``cp`` keeps st_ino; generation stamp must still halt (#89332).""" + live = tmp_path / "state.db" + other = tmp_path / "other.db" + db = _make_db(live, "live-sess", "original") + alt = _make_db(other, "other-sess", "replacement") + live_app = db._db_file_application_id + other_app = alt._db_file_application_id + if not live_app or not other_app: + alt.close() + db.close() + pytest.skip("generation stamp not recorded on this filesystem") + assert live_app != other_app + recorded = db._db_file_identity + alt.close() + shutil.copyfile(other, live) + if recorded is not None: + st = os.stat(live) + assert (st.st_dev, st.st_ino) == recorded + with pytest.raises(StateDbReplacedError, match="replaced underneath"): + db.append_message("live-sess", role="user", content="after-cp") + assert db._db_replaced is True + assert db._fts_runtime_rebuild_attempted is False + db.close() + + +def test_divert_session_transcript_jsonl_appends(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + path = divert_session_transcript_jsonl( + "sess-jsonl", + [{"role": "user", "content": "hello-jsonl"}], + ) + assert path == tmp_path / "sessions" / "sess-jsonl.jsonl" + lines = path.read_text(encoding="utf-8").strip().splitlines() + assert json.loads(lines[-1])["content"] == "hello-jsonl" + assert divert_session_transcript_jsonl("sess-jsonl", []) is None + + +def _stat_changed(path: Path, recorded) -> bool: + st = os.stat(path) + return (st.st_dev, st.st_ino) != recorded diff --git a/tests/run_agent/test_turn_completion_explainer.py b/tests/run_agent/test_turn_completion_explainer.py index 6e2c889de4..99b052c7f9 100644 --- a/tests/run_agent/test_turn_completion_explainer.py +++ b/tests/run_agent/test_turn_completion_explainer.py @@ -154,6 +154,17 @@ def test_explanation_persistence_corrupt_cause_never_says_free_space(): assert "full disk" not in lower +def test_explanation_persistence_replaced_cause_forbids_inplace_repair(): + out = AIAgent._format_turn_completion_explanation( + "session_persistence_failed", "replaced" + ) + lower = out.lower() + assert "replaced" in lower + assert "doctor --fix" in lower or "in-place" in lower + assert "free some space" not in lower + assert "full disk" not in lower + + def test_explanation_persistence_unknown_cause_is_neutral(): """None/'unknown' cause must not claim disk-full — point at diagnostics.""" for cause in (None, "unknown"): @@ -305,6 +316,7 @@ def test_persistence_error_causes_tuple_matches_classifier(): "Session 'abc' is being compressed by another writer", "Session turn lease lost; refusing transcript write for 'abc'", "database disk image is malformed", + "FATAL: state.db was replaced underneath the gateway", "database or disk is full", "something else entirely", None,