From 6fbb5e52e0921495debd5209d2edef488a32f061 Mon Sep 17 00:00:00 2001 From: Mibayy Date: Thu, 26 Mar 2026 10:24:18 +0000 Subject: [PATCH] fix(session-db): survive CLI/gateway concurrent write contention Closes #3139 Three layered fixes for the scenario where CLI and gateway write to state.db concurrently, causing create_session() to fail with 'database is locked' and permanently disabling session_search on the gateway side. 1. Increase SQLite connection timeout: 10s -> 30s hermes_state.py: longer window for the WAL writer to finish a batch flush before the other process gives up entirely. 2. INSERT OR IGNORE in create_session hermes_state.py: prevents IntegrityError on duplicate session IDs (e.g. gateway restarts while CLI session is still alive). 3. Don't null out _session_db on create_session failure (main fix) run_agent.py: a transient lock at agent startup must not permanently disable session_search for the lifetime of that agent instance. _session_db now stays alive so subsequent flushes and searches work once the lock clears. 4. New ensure_session() helper + call it during flush hermes_state.py: INSERT OR IGNORE for a minimal session row. run_agent.py _flush_messages_to_session_db: calls ensure_session() before appending messages, so the FK constraint is satisfied even when create_session() failed at startup. No-op when the row exists. --- hermes_state.py | 28 +++++++++++++++-- run_agent.py | 19 ++++++++++-- tests/test_hermes_state.py | 63 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 4 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index eb62bc2ba8..8f651750b6 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -124,7 +124,10 @@ class SessionDB: self._conn = sqlite3.connect( str(self.db_path), check_same_thread=False, - timeout=10.0, + # 30s gives the WAL writer (CLI or gateway) time to finish a batch + # flush before the concurrent reader/writer gives up. 10s was too + # short when the CLI is doing frequent memory flushes. + timeout=30.0, ) self._conn.row_factory = sqlite3.Row self._conn.execute("PRAGMA journal_mode=WAL") @@ -255,7 +258,7 @@ class SessionDB: """Create a new session record. Returns the session_id.""" with self._lock: self._conn.execute( - """INSERT INTO sessions (id, source, user_id, model, model_config, + """INSERT OR IGNORE INTO sessions (id, source, user_id, model, model_config, system_prompt, parent_session_id, started_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( @@ -351,6 +354,27 @@ class SessionDB: ) self._conn.commit() + def ensure_session( + self, + session_id: str, + source: str = "unknown", + model: str = None, + ) -> None: + """Ensure a session row exists, creating it with minimal metadata if absent. + + Used by _flush_messages_to_session_db to recover from a failed + create_session() call (e.g. transient SQLite lock at agent startup). + INSERT OR IGNORE is safe to call even when the row already exists. + """ + with self._lock: + self._conn.execute( + """INSERT OR IGNORE INTO sessions + (id, source, model, started_at) + VALUES (?, ?, ?, ?)""", + (session_id, source, model, time.time()), + ) + self._conn.commit() + def get_session(self, session_id: str) -> Optional[Dict[str, Any]]: """Get a session by ID.""" with self._lock: diff --git a/run_agent.py b/run_agent.py index 3a75d040d3..f2f0239597 100644 --- a/run_agent.py +++ b/run_agent.py @@ -893,8 +893,15 @@ class AIAgent: user_id=None, ) except Exception as e: - logger.warning("Session DB create_session failed — messages will NOT be indexed: %s", e) - self._session_db = None # prevent silent data loss on every subsequent flush + # Transient SQLite lock contention (e.g. CLI and gateway writing + # concurrently) must NOT permanently disable session_search for + # this agent. Keep _session_db alive — subsequent message + # flushes and session_search calls will still work once the + # lock clears. The session row may be missing from the index + # for this run, but that is recoverable (flushes upsert rows). + logger.warning( + "Session DB create_session failed (session_search still available): %s", e + ) # In-memory todo list for task planning (one per agent/session) from tools.todo_tool import TodoStore @@ -1578,6 +1585,14 @@ class AIAgent: return self._apply_persist_user_message_override(messages) try: + # If create_session() failed at startup (e.g. transient lock), the + # session row may not exist yet. ensure_session() uses INSERT OR + # IGNORE so it is a no-op when the row is already there. + self._session_db.ensure_session( + self.session_id, + source=self.platform or "cli", + model=self.model, + ) start_idx = len(conversation_history) if conversation_history else 0 flush_from = max(start_idx, self._last_flushed_db_idx) for msg in messages[flush_from:]: diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 381bb9d199..d4712450bf 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -1116,3 +1116,66 @@ class TestResolveSessionByNameOrId: db.set_session_title("s1", "my project") result = db.resolve_session_by_title("my project") assert result == "s1" + + +# ========================================================================= +# Concurrent write safety / lock contention fixes (#3139) +# ========================================================================= + +class TestConcurrentWriteSafety: + def test_create_session_insert_or_ignore_is_idempotent(self, db): + """create_session with the same ID twice must not raise (INSERT OR IGNORE).""" + db.create_session(session_id="dup-1", source="cli", model="m") + # Second call should be silent — no IntegrityError + db.create_session(session_id="dup-1", source="gateway", model="m2") + session = db.get_session("dup-1") + # Row should exist (first write wins with OR IGNORE) + assert session is not None + assert session["source"] == "cli" + + def test_ensure_session_creates_missing_row(self, db): + """ensure_session must create a minimal row when the session doesn't exist.""" + assert db.get_session("orphan-session") is None + db.ensure_session("orphan-session", source="gateway", model="test-model") + row = db.get_session("orphan-session") + assert row is not None + assert row["source"] == "gateway" + assert row["model"] == "test-model" + + def test_ensure_session_is_idempotent(self, db): + """ensure_session on an existing row must be a no-op (no overwrite).""" + db.create_session(session_id="existing", source="cli", model="original-model") + db.ensure_session("existing", source="gateway", model="overwrite-model") + row = db.get_session("existing") + # First write wins — ensure_session must not overwrite + assert row["source"] == "cli" + assert row["model"] == "original-model" + + def test_ensure_session_allows_append_message_after_failed_create(self, db): + """Messages can be flushed even when create_session failed at startup. + + Simulates the #3139 scenario: create_session raises (lock), then + ensure_session is called during flush, then append_message succeeds. + """ + # Simulate failed create_session — row absent + db.ensure_session("late-session", source="gateway", model="gpt-4") + db.append_message( + session_id="late-session", + role="user", + content="hello after lock", + ) + msgs = db.get_messages("late-session") + assert len(msgs) == 1 + assert msgs[0]["content"] == "hello after lock" + + def test_sqlite_timeout_is_at_least_30s(self, db): + """Connection timeout should be >= 30s to survive CLI/gateway contention.""" + # Access the underlying connection timeout via sqlite3 introspection. + # There is no public API, so we check the kwarg via the module default. + import sqlite3 + import inspect + from hermes_state import SessionDB as _SessionDB + src = inspect.getsource(_SessionDB.__init__) + assert "30" in src, ( + "SQLite timeout should be at least 30s to handle CLI/gateway lock contention" + )