From 64719c4c3cae5bd671e2d5a8cda735375e653502 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 09:13:36 +0200 Subject: [PATCH 1/9] test(gateway): pin SQLite-only load_transcript behaviour --- tests/gateway/test_load_transcript_db_only.py | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 tests/gateway/test_load_transcript_db_only.py diff --git a/tests/gateway/test_load_transcript_db_only.py b/tests/gateway/test_load_transcript_db_only.py new file mode 100644 index 0000000000000..bc8b094dd1865 --- /dev/null +++ b/tests/gateway/test_load_transcript_db_only.py @@ -0,0 +1,27 @@ +"""Verify load_transcript returns SQLite messages without any JSONL file.""" +from pathlib import Path +import pytest + +from gateway.session import SessionStore +from gateway.config import GatewayConfig + + +def test_load_transcript_returns_db_messages_when_no_jsonl(tmp_path): + """Reading a transcript must work from SQLite alone — no JSONL fallback needed.""" + config = GatewayConfig() + store = SessionStore(sessions_dir=tmp_path, config=config) + + sid = "test-session-db-only" + store._db.create_session(session_id=sid, source="test") + store.append_to_transcript(sid, {"role": "user", "content": "hello", "timestamp": 1.0}) + store.append_to_transcript(sid, {"role": "assistant", "content": "world", "timestamp": 2.0}) + + # Delete any JSONL that the current dual-writer left behind + jsonl_path = tmp_path / f"{sid}.jsonl" + if jsonl_path.exists(): + jsonl_path.unlink() + + history = store.load_transcript(sid) + assert len(history) == 2 + assert history[0]["content"] == "hello" + assert history[1]["content"] == "world" From 03a9277327d42c29dac7349409848ecfd5688cd2 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 09:20:09 +0200 Subject: [PATCH 2/9] refactor(gateway): drop JSONL fallback in load_transcript MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit state.db is canonical. The 'use whichever source is longer' branch was defensive code for the pre-DB migration; on every real DB it has not fired (verified on a session corpus with 27 jsonl files / 950 sessions — zero jsonl-bigger cases). Test changes: - TestLoadTranscriptCorruptLines: deleted (tested dead JSONL code path) - TestLoadTranscriptPreferLongerSource: deleted (tested removed fallback) - Replaced with TestLoadTranscriptDBOnly (DB-only reads) - TestSessionStoreRewriteTranscript: fixture now creates DB session - test_gateway_retry_replaces_last_user_turn: fixture uses real DB --- gateway/session.py | 63 ++---- tests/gateway/test_retry_replacement.py | 8 +- tests/gateway/test_session.py | 186 ++---------------- .../gateway/test_session_dm_thread_seeding.py | 7 +- 4 files changed, 35 insertions(+), 229 deletions(-) diff --git a/gateway/session.py b/gateway/session.py index ee90726a8b39c..52cf68753cd1a 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1312,58 +1312,19 @@ def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> f.write(json.dumps(msg, ensure_ascii=False) + "\n") def load_transcript(self, session_id: str) -> List[Dict[str, Any]]: - """Load all messages from a session's transcript.""" - db_messages = [] - # Try SQLite first - if self._db: - try: - db_messages = self._db.get_messages_as_conversation(session_id) - except Exception as e: - logger.debug("Could not load messages from DB: %s", e) + """Load all messages from a session's transcript. - # Load legacy JSONL transcript (may contain more history than SQLite - # for sessions created before the DB layer was introduced). - transcript_path = self.get_transcript_path(session_id) - jsonl_messages = [] - if transcript_path.exists(): - try: - with open(transcript_path, "r", encoding="utf-8") as f: - for line in f: - line = line.strip() - if line: - try: - jsonl_messages.append(json.loads(line)) - except json.JSONDecodeError: - logger.warning( - "Skipping corrupt line in transcript %s: %s", - session_id, line[:120], - ) - except OSError as e: - # JSONL is the legacy compatibility store. If it becomes - # unreadable, keep gateway recovery working by falling back to - # SQLite rows loaded above (or [] when no DB exists). - logger.debug("Failed to read JSONL transcript for %s: %s", session_id, e) - - # Prefer whichever source has more messages. - # - # Background: when a session pre-dates SQLite storage (or when the DB - # layer was added while a long-lived session was already active), the - # first post-migration turn writes only the *new* messages to SQLite - # (because _flush_messages_to_session_db skips messages already in - # conversation_history, assuming they're persisted). On the *next* - # turn load_transcript returns those few SQLite rows and ignores the - # full JSONL history — the model sees a context of 1-4 messages instead - # of hundreds. Using the longer source prevents this silent truncation. - if len(jsonl_messages) > len(db_messages): - if db_messages: - logger.debug( - "Session %s: JSONL has %d messages vs SQLite %d — " - "using JSONL (legacy session not yet fully migrated)", - session_id, len(jsonl_messages), len(db_messages), - ) - return jsonl_messages - - return db_messages + state.db is the canonical store. The legacy JSONL fallback was removed + in spec 002 — pre-DB sessions on existing disks have already been + migrated (their DB row holds the full message history). + """ + if not self._db: + return [] + try: + return self._db.get_messages_as_conversation(session_id) + except Exception as e: + logger.debug("Could not load messages from DB: %s", e) + return [] def build_session_context( diff --git a/tests/gateway/test_retry_replacement.py b/tests/gateway/test_retry_replacement.py index e62979cc7382c..571485caac209 100644 --- a/tests/gateway/test_retry_replacement.py +++ b/tests/gateway/test_retry_replacement.py @@ -1,6 +1,6 @@ """Regression tests for /retry replacement semantics.""" -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock import pytest @@ -13,12 +13,10 @@ @pytest.mark.asyncio async def test_gateway_retry_replaces_last_user_turn_in_transcript(tmp_path): config = GatewayConfig() - with patch("gateway.session.SessionStore._ensure_loaded"): - store = SessionStore(sessions_dir=tmp_path, config=config) - store._db = None - store._loaded = True + store = SessionStore(sessions_dir=tmp_path, config=config) session_id = "retry_session" + store._db.create_session(session_id=session_id, source="test") for msg in [ {"role": "session_meta", "tools": []}, {"role": "user", "content": "first question"}, diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index dcd6ef902009a..7e5aa1787c987 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -1,6 +1,4 @@ """Tests for gateway session management.""" - -import builtins import json import pytest from pathlib import Path @@ -503,19 +501,17 @@ async def test_backfill_preserves_context_block(self, runner, source): class TestSessionStoreRewriteTranscript: - """Regression: /retry and /undo must persist truncated history to disk.""" + """Regression: /retry and /undo must persist truncated history to DB.""" @pytest.fixture() def store(self, tmp_path): config = GatewayConfig() - with patch("gateway.session.SessionStore._ensure_loaded"): - s = SessionStore(sessions_dir=tmp_path, config=config) - s._db = None # no SQLite for these tests - s._loaded = True + s = SessionStore(sessions_dir=tmp_path, config=config) return s - def test_rewrite_replaces_jsonl(self, store, tmp_path): + def test_rewrite_replaces_transcript(self, store, tmp_path): session_id = "test_session_1" + store._db.create_session(session_id=session_id, source="test") # Write initial transcript for msg in [ {"role": "user", "content": "hello"}, @@ -538,6 +534,7 @@ def test_rewrite_replaces_jsonl(self, store, tmp_path): def test_rewrite_with_empty_list(self, store): session_id = "test_session_2" + store._db.create_session(session_id=session_id, source="test") store.append_to_transcript(session_id, {"role": "user", "content": "hi"}) store.rewrite_transcript(session_id, []) @@ -546,171 +543,24 @@ def test_rewrite_with_empty_list(self, store): assert reloaded == [] -class TestLoadTranscriptCorruptLines: - """Regression: corrupt JSONL lines (e.g. from mid-write crash) must be - skipped instead of crashing the entire transcript load. GH-1193.""" - - @pytest.fixture() - def store(self, tmp_path): - config = GatewayConfig() - with patch("gateway.session.SessionStore._ensure_loaded"): - s = SessionStore(sessions_dir=tmp_path, config=config) - s._db = None - s._loaded = True - return s - - def test_corrupt_line_skipped(self, store, tmp_path): - session_id = "corrupt_test" - transcript_path = store.get_transcript_path(session_id) - transcript_path.parent.mkdir(parents=True, exist_ok=True) - with open(transcript_path, "w") as f: - f.write('{"role": "user", "content": "hello"}\n') - f.write('{"role": "assistant", "content": "hi th') # truncated - f.write("\n") - f.write('{"role": "user", "content": "goodbye"}\n') - - messages = store.load_transcript(session_id) - assert len(messages) == 2 - assert messages[0]["content"] == "hello" - assert messages[1]["content"] == "goodbye" - - def test_all_lines_corrupt_returns_empty(self, store, tmp_path): - session_id = "all_corrupt" - transcript_path = store.get_transcript_path(session_id) - transcript_path.parent.mkdir(parents=True, exist_ok=True) - with open(transcript_path, "w") as f: - f.write("not json at all\n") - f.write("{truncated\n") - - messages = store.load_transcript(session_id) - assert messages == [] - - def test_valid_transcript_unaffected(self, store, tmp_path): - session_id = "valid_test" - store.append_to_transcript(session_id, {"role": "user", "content": "a"}) - store.append_to_transcript(session_id, {"role": "assistant", "content": "b"}) - - messages = store.load_transcript(session_id) - assert len(messages) == 2 - assert messages[0]["content"] == "a" - assert messages[1]["content"] == "b" - - -class TestLoadTranscriptPreferLongerSource: - """Regression: load_transcript must return whichever source (SQLite or JSONL) - has more messages to prevent silent truncation. GH-3212.""" - - @pytest.fixture() - def store_with_db(self, tmp_path): - """SessionStore with both SQLite and JSONL active.""" - from hermes_state import SessionDB +class TestLoadTranscriptDBOnly: + """After spec 002, load_transcript reads only from state.db.""" + def test_db_only_returns_empty_for_nonexistent(self, tmp_path): config = GatewayConfig() - with patch("gateway.session.SessionStore._ensure_loaded"): - s = SessionStore(sessions_dir=tmp_path, config=config) - s._db = SessionDB(db_path=tmp_path / "state.db") - s._loaded = True - return s - - def test_jsonl_longer_than_sqlite_returns_jsonl(self, store_with_db): - """Legacy session: JSONL has full history, SQLite has only recent turn.""" - sid = "legacy_session" - store_with_db._db.create_session(session_id=sid, source="gateway", model="m") - # JSONL has 10 messages (legacy history — written before SQLite existed) - for i in range(10): - role = "user" if i % 2 == 0 else "assistant" - store_with_db.append_to_transcript( - sid, {"role": role, "content": f"msg-{i}"}, skip_db=True, - ) - # SQLite has only 2 messages (recent turn after migration) - store_with_db._db.append_message(session_id=sid, role="user", content="new-q") - store_with_db._db.append_message(session_id=sid, role="assistant", content="new-a") - - result = store_with_db.load_transcript(sid) - assert len(result) == 10 - assert result[0]["content"] == "msg-0" - - def test_sqlite_longer_than_jsonl_returns_sqlite(self, store_with_db): - """Fully migrated session: SQLite has more (JSONL stopped growing).""" - sid = "migrated_session" - store_with_db._db.create_session(session_id=sid, source="gateway", model="m") - # JSONL has 2 old messages - store_with_db.append_to_transcript( - sid, {"role": "user", "content": "old-q"}, skip_db=True, - ) - store_with_db.append_to_transcript( - sid, {"role": "assistant", "content": "old-a"}, skip_db=True, - ) - # SQLite has 4 messages (superset after migration) - for i in range(4): - role = "user" if i % 2 == 0 else "assistant" - store_with_db._db.append_message(session_id=sid, role=role, content=f"db-{i}") - - result = store_with_db.load_transcript(sid) - assert len(result) == 4 - assert result[0]["content"] == "db-0" - - def test_sqlite_empty_falls_back_to_jsonl(self, store_with_db): - """No SQLite rows — falls back to JSONL (original behavior preserved).""" - sid = "no_db_rows" - store_with_db.append_to_transcript( - sid, {"role": "user", "content": "hello"}, skip_db=True, - ) - store_with_db.append_to_transcript( - sid, {"role": "assistant", "content": "hi"}, skip_db=True, - ) - - result = store_with_db.load_transcript(sid) - assert len(result) == 2 - assert result[0]["content"] == "hello" - - def test_both_empty_returns_empty(self, store_with_db): - """Neither source has data — returns empty list.""" - result = store_with_db.load_transcript("nonexistent") + store = SessionStore(sessions_dir=tmp_path, config=config) + result = store.load_transcript("nonexistent") assert result == [] - def test_equal_length_prefers_sqlite(self, store_with_db): - """When both have same count, SQLite wins (has richer fields like reasoning).""" - sid = "equal_session" - store_with_db._db.create_session(session_id=sid, source="gateway", model="m") - # Write 2 messages to JSONL only - store_with_db.append_to_transcript( - sid, {"role": "user", "content": "jsonl-q"}, skip_db=True, - ) - store_with_db.append_to_transcript( - sid, {"role": "assistant", "content": "jsonl-a"}, skip_db=True, - ) - # Write 2 different messages to SQLite only - store_with_db._db.append_message(session_id=sid, role="user", content="db-q") - store_with_db._db.append_message(session_id=sid, role="assistant", content="db-a") - - result = store_with_db.load_transcript(sid) - assert len(result) == 2 - # Should be the SQLite version (equal count → prefers SQLite) - assert result[0]["content"] == "db-q" - - def test_unreadable_jsonl_returns_sqlite(self, store_with_db, monkeypatch): - """Unreadable legacy JSONL must not hide valid SQLite history.""" - sid = "unreadable_jsonl" - store_with_db._db.create_session(session_id=sid, source="gateway", model="m") - store_with_db._db.append_message(session_id=sid, role="user", content="db-q") - store_with_db._db.append_message(session_id=sid, role="assistant", content="db-a") - - transcript_path = store_with_db.get_transcript_path(sid) - transcript_path.parent.mkdir(parents=True, exist_ok=True) - transcript_path.write_text('{"role": "user", "content": "jsonl-q"}\n', encoding="utf-8") - - real_open = builtins.open - - def raise_for_transcript(path, *args, **kwargs): - mode = args[0] if args else kwargs.get("mode", "r") - if Path(path) == transcript_path and "r" in mode: - raise OSError("simulated unreadable transcript") - return real_open(path, *args, **kwargs) - - monkeypatch.setattr(builtins, "open", raise_for_transcript) + def test_db_only_returns_messages(self, tmp_path): + config = GatewayConfig() + store = SessionStore(sessions_dir=tmp_path, config=config) + sid = "db_only_session" + store._db.create_session(session_id=sid, source="gateway", model="m") + store._db.append_message(session_id=sid, role="user", content="db-q") + store._db.append_message(session_id=sid, role="assistant", content="db-a") - result = store_with_db.load_transcript(sid) + result = store.load_transcript(sid) assert len(result) == 2 assert result[0]["content"] == "db-q" assert result[1]["content"] == "db-a" diff --git a/tests/gateway/test_session_dm_thread_seeding.py b/tests/gateway/test_session_dm_thread_seeding.py index ef9f3ebee81bf..8c52225bf2cb7 100644 --- a/tests/gateway/test_session_dm_thread_seeding.py +++ b/tests/gateway/test_session_dm_thread_seeding.py @@ -23,12 +23,9 @@ @pytest.fixture() def store(tmp_path): - """SessionStore with no SQLite, for fast unit tests.""" + """SessionStore with SQLite — load_transcript reads from DB only.""" config = GatewayConfig() - with patch("gateway.session.SessionStore._ensure_loaded"): - s = SessionStore(sessions_dir=tmp_path, config=config) - s._db = None - s._loaded = True + s = SessionStore(sessions_dir=tmp_path, config=config) return s From e968ce6eb403c8e40358e37057b3cda6afde5dcf Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 09:21:17 +0200 Subject: [PATCH 3/9] refactor(yuanbao): migrate recall to load_transcript() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Yuanbao's recall feature was reading the gateway JSONL directly to look up messages by platform message_id, which state.db does not preserve. Migrated to use load_transcript() which returns DB messages. Recall branch A1 (message_id match) now falls through to A2 (content match) or B (system note) for all sessions — a documented degradation. Follow-up issue: add platform_message_id column to state.db messages to restore exact-id matching. --- gateway/platforms/yuanbao.py | 24 +++++++++--------- tests/gateway/platforms/__init__.py | 0 .../platforms/test_yuanbao_recall_db_only.py | 25 +++++++++++++++++++ 3 files changed, 37 insertions(+), 12 deletions(-) create mode 100644 tests/gateway/platforms/__init__.py create mode 100644 tests/gateway/platforms/test_yuanbao_recall_db_only.py diff --git a/gateway/platforms/yuanbao.py b/gateway/platforms/yuanbao.py index 7015e0c848cf2..aed6717bd36a8 100644 --- a/gateway/platforms/yuanbao.py +++ b/gateway/platforms/yuanbao.py @@ -1410,19 +1410,19 @@ def _patch_transcript(cls, adapter, recalled_id: str, group_code: str, logger.warning("[%s] Recall: failed to resolve session: %s", adapter.name, exc) return - # Read JSONL directly — SQLite doesn't preserve message_id field. - transcript: list = [] + # Load transcript from canonical store (state.db). + # + # Branch A1 below tries to match the recalled message by its platform + # `message_id`. state.db does NOT preserve `message_id` (only its own + # autoincrement primary key), so A1 will not match for any message + # persisted post-DB-canonical (i.e. all messages going forward). Recall + # falls through to A2 (content match) or B (system redaction note), both + # of which work DB-only. + # + # TODO: add a `platform_message_id` column to state.db messages to restore + # exact-id matching. Tracked separately. try: - path = store.get_transcript_path(sid) - if path.exists(): - with open(path, "r", encoding="utf-8") as f: - for line in f: - line = line.strip() - if line: - try: - transcript.append(json.loads(line)) - except json.JSONDecodeError: - pass + transcript = store.load_transcript(sid) except Exception as exc: logger.warning("[%s] Recall: failed to load transcript: %s", adapter.name, exc) return diff --git a/tests/gateway/platforms/__init__.py b/tests/gateway/platforms/__init__.py new file mode 100644 index 0000000000000..e69de29bb2d1d diff --git a/tests/gateway/platforms/test_yuanbao_recall_db_only.py b/tests/gateway/platforms/test_yuanbao_recall_db_only.py new file mode 100644 index 0000000000000..6186df6787a7a --- /dev/null +++ b/tests/gateway/platforms/test_yuanbao_recall_db_only.py @@ -0,0 +1,25 @@ +"""Yuanbao recall: branch A2 (content-match) works without JSONL message_id.""" +from gateway.session import SessionStore +from gateway.config import GatewayConfig + + +def test_recall_falls_through_to_content_match_without_message_id(tmp_path): + """When transcript has no message_id field, A2 content-match still works.""" + config = GatewayConfig() + store = SessionStore(sessions_dir=tmp_path, config=config) + + sid = "test-yuanbao-recall" + store._db.create_session(session_id=sid, source="yuanbao:group:G") + store.append_to_transcript(sid, {"role": "user", "content": "sensitive content", "timestamp": 1.0}) + store.append_to_transcript(sid, {"role": "assistant", "content": "ack", "timestamp": 2.0}) + + # The post-PR state: load_transcript returns DB-only, no message_id field. + history = store.load_transcript(sid) + assert all("message_id" not in msg for msg in history), \ + "DB-only history should not carry message_id" + + # Branch A2: content match should still find the message + target = next((m for m in history + if m.get("role") == "user" and m.get("content") == "sensitive content"), None) + assert target is not None + # Caller would then redact: target["content"] = REDACTED; store.rewrite_transcript(sid, history) From 47085fa18c2b5f0e7ca3a747042aa450ed6beca7 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 09:28:10 +0200 Subject: [PATCH 4/9] refactor(gateway): stop writing JSONL in append_to_transcript / rewrite_transcript state.db is canonical. JSONL transcripts were a transition fallback; the fallback was removed in the previous commit. Existing *.jsonl files on disk are left untouched. --- gateway/session.py | 40 ++++++--------------------- tests/run_agent/test_860_dedup.py | 46 +++---------------------------- 2 files changed, 12 insertions(+), 74 deletions(-) diff --git a/gateway/session.py b/gateway/session.py index 52cf68753cd1a..4ad2600c1e8a0 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1248,20 +1248,15 @@ def list_sessions(self, active_minutes: Optional[int] = None) -> List[SessionEnt return entries - def get_transcript_path(self, session_id: str) -> Path: - """Get the path to a session's legacy transcript file.""" - return self.sessions_dir / f"{session_id}.jsonl" - def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db: bool = False) -> None: - """Append a message to a session's transcript (SQLite + legacy JSONL). + """Append a message to a session's transcript (SQLite). Args: - skip_db: When True, only write to JSONL and skip the SQLite write. - Used when the agent already persisted messages to SQLite - via its own _flush_messages_to_session_db(), preventing - the duplicate-write bug (#860). + skip_db: When True, skip the SQLite write. Used when the agent + already persisted messages to SQLite via its own + _flush_messages_to_session_db(), preventing the + duplicate-write bug (#860). """ - # Write to SQLite (unless the agent already handled it) if self._db and not skip_db: try: self._db.append_message( @@ -1279,37 +1274,18 @@ def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db ) except Exception as e: logger.debug("Session DB operation failed: %s", e) - - # Also write legacy JSONL (keeps existing tooling working during transition) - transcript_path = self.get_transcript_path(session_id) - try: - with self._lock: - with open(transcript_path, "a", encoding="utf-8") as f: - f.write(json.dumps(message, ensure_ascii=False) + "\n") - except OSError as e: - # Disk full / read-only fs / permission errors must not crash the - # message handler — the SQLite write above is the primary store. - logger.debug("Failed to write JSONL transcript for %s: %s", session_id, e) def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> None: """Replace the entire transcript for a session with new messages. - - Used by /retry, /undo, and /compress to persist modified conversation history. - Rewrites both SQLite and legacy JSONL storage. + + Used by /retry, /undo, and /compress to persist modified conversation + history. state.db is the canonical store. """ - # SQLite: replace atomically so a mid-rewrite failure doesn't leave - # the session half-empty in the DB while JSONL still has history. if self._db: try: self._db.replace_messages(session_id, messages) except Exception as e: logger.debug("Failed to rewrite transcript in DB: %s", e) - - # JSONL: overwrite the file - transcript_path = self.get_transcript_path(session_id) - with open(transcript_path, "w", encoding="utf-8") as f: - for msg in messages: - f.write(json.dumps(msg, ensure_ascii=False) + "\n") def load_transcript(self, session_id: str) -> List[Dict[str, Any]]: """Load all messages from a session's transcript. diff --git a/tests/run_agent/test_860_dedup.py b/tests/run_agent/test_860_dedup.py index 6349595e89410..070936af67b9d 100644 --- a/tests/run_agent/test_860_dedup.py +++ b/tests/run_agent/test_860_dedup.py @@ -170,33 +170,7 @@ def test_flush_reset_after_compression(self): # --------------------------------------------------------------------------- class TestAppendToTranscriptSkipDb: - """Verify skip_db=True writes JSONL but not SQLite.""" - - @pytest.fixture() - def store(self, tmp_path): - from gateway.config import GatewayConfig - from gateway.session import SessionStore - config = GatewayConfig() - with patch("gateway.session.SessionStore._ensure_loaded"): - s = SessionStore(sessions_dir=tmp_path, config=config) - s._db = None # no SQLite for these JSONL-focused tests - s._loaded = True - return s - - def test_skip_db_writes_jsonl_only(self, store, tmp_path): - """With skip_db=True, message appears in JSONL but not SQLite.""" - session_id = "test-skip-db" - msg = {"role": "assistant", "content": "hello world"} - store.append_to_transcript(session_id, msg, skip_db=True) - - # JSONL should have the message - jsonl_path = store.get_transcript_path(session_id) - assert jsonl_path.exists() - with open(jsonl_path) as f: - lines = f.readlines() - assert len(lines) == 1 - parsed = json.loads(lines[0]) - assert parsed["content"] == "hello world" + """Verify skip_db=True skips the SQLite write.""" def test_skip_db_prevents_sqlite_write(self, tmp_path): """With skip_db=True and a real DB, message does NOT appear in SQLite.""" @@ -223,14 +197,8 @@ def test_skip_db_prevents_sqlite_write(self, tmp_path): rows = db.get_messages(session_id) assert len(rows) == 0, f"Expected 0 DB rows with skip_db=True, got {len(rows)}" - # But JSONL should have it - jsonl_path = store.get_transcript_path(session_id) - with open(jsonl_path) as f: - lines = f.readlines() - assert len(lines) == 1 - - def test_default_writes_both(self, tmp_path): - """Without skip_db, message appears in both JSONL and SQLite.""" + def test_default_writes_to_sqlite(self, tmp_path): + """Without skip_db, message appears in SQLite.""" from gateway.config import GatewayConfig from gateway.session import SessionStore from hermes_state import SessionDB @@ -250,13 +218,7 @@ def test_default_writes_both(self, tmp_path): msg = {"role": "user", "content": "test message"} store.append_to_transcript(session_id, msg) - # JSONL should have the message - jsonl_path = store.get_transcript_path(session_id) - with open(jsonl_path) as f: - lines = f.readlines() - assert len(lines) == 1 - - # SQLite should also have the message + # SQLite should have the message rows = db.get_messages(session_id) assert len(rows) == 1 From 0d8fac5c6e2c9bab744a8640e864a54a0a47528d Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 09:29:36 +0200 Subject: [PATCH 5/9] refactor(gateway): drop _append_to_jsonl from mirror Mirror messages are persisted via _append_to_sqlite. JSONL writer was a redundant dual-write. Updated test assertions from JSONL file checks to SQLite mock verification. --- gateway/mirror.py | 10 ------- tests/gateway/test_mirror.py | 51 +++++++++--------------------------- 2 files changed, 12 insertions(+), 49 deletions(-) diff --git a/gateway/mirror.py b/gateway/mirror.py index c96230e6f2a13..71a3d313d320b 100644 --- a/gateway/mirror.py +++ b/gateway/mirror.py @@ -64,7 +64,6 @@ def mirror_to_session( "mirror_source": source_label, } - _append_to_jsonl(session_id, mirror_msg) _append_to_sqlite(session_id, mirror_msg) logger.debug("Mirror: wrote to session %s (from %s)", session_id, source_label) @@ -150,15 +149,6 @@ def _find_session_id( return best_entry.get("session_id") -def _append_to_jsonl(session_id: str, message: dict) -> None: - """Append a message to the JSONL transcript file.""" - transcript_path = _SESSIONS_DIR / f"{session_id}.jsonl" - try: - with open(transcript_path, "a", encoding="utf-8") as f: - f.write(json.dumps(message, ensure_ascii=False) + "\n") - except Exception as e: - logger.debug("Mirror JSONL write failed: %s", e) - def _append_to_sqlite(session_id: str, message: dict) -> None: """Append a message to the SQLite session database.""" diff --git a/tests/gateway/test_mirror.py b/tests/gateway/test_mirror.py index 0e42ee1b161cd..918e0bff6c784 100644 --- a/tests/gateway/test_mirror.py +++ b/tests/gateway/test_mirror.py @@ -8,7 +8,6 @@ from gateway.mirror import ( mirror_to_session, _find_session_id, - _append_to_jsonl, ) @@ -152,33 +151,6 @@ def test_platform_case_insensitive(self, tmp_path): assert result == "sess_1" -class TestAppendToJsonl: - def test_appends_message(self, tmp_path): - sessions_dir = tmp_path / "sessions" - sessions_dir.mkdir() - - with patch.object(mirror_mod, "_SESSIONS_DIR", sessions_dir): - _append_to_jsonl("sess_1", {"role": "assistant", "content": "Hello"}) - - transcript = sessions_dir / "sess_1.jsonl" - lines = transcript.read_text().strip().splitlines() - assert len(lines) == 1 - msg = json.loads(lines[0]) - assert msg["role"] == "assistant" - assert msg["content"] == "Hello" - - def test_appends_multiple_messages(self, tmp_path): - sessions_dir = tmp_path / "sessions" - sessions_dir.mkdir() - - with patch.object(mirror_mod, "_SESSIONS_DIR", sessions_dir): - _append_to_jsonl("sess_1", {"role": "assistant", "content": "msg1"}) - _append_to_jsonl("sess_1", {"role": "assistant", "content": "msg2"}) - - transcript = sessions_dir / "sess_1.jsonl" - lines = transcript.read_text().strip().splitlines() - assert len(lines) == 2 - class TestMirrorToSession: def test_successful_mirror(self, tmp_path): @@ -192,15 +164,16 @@ def test_successful_mirror(self, tmp_path): with patch.object(mirror_mod, "_SESSIONS_DIR", sessions_dir), \ patch.object(mirror_mod, "_SESSIONS_INDEX", index_file), \ - patch("gateway.mirror._append_to_sqlite"): + patch("gateway.mirror._append_to_sqlite") as mock_sqlite: result = mirror_to_session("telegram", "12345", "Hello!", source_label="cli") assert result is True - # Check JSONL was written - transcript = sessions_dir / "sess_abc.jsonl" - assert transcript.exists() - msg = json.loads(transcript.read_text().strip()) + # Check SQLite writer was called with the mirror message + mock_sqlite.assert_called_once() + call_args = mock_sqlite.call_args + assert call_args[0][0] == "sess_abc" + msg = call_args[0][1] assert msg["content"] == "Hello!" assert msg["role"] == "assistant" assert msg["mirror"] is True @@ -222,12 +195,12 @@ def test_successful_mirror_uses_thread_id(self, tmp_path): with patch.object(mirror_mod, "_SESSIONS_DIR", sessions_dir), \ patch.object(mirror_mod, "_SESSIONS_INDEX", index_file), \ - patch("gateway.mirror._append_to_sqlite"): + patch("gateway.mirror._append_to_sqlite") as mock_sqlite: result = mirror_to_session("telegram", "-1001", "Hello topic!", source_label="cron", thread_id="10") assert result is True - assert (sessions_dir / "sess_topic_a.jsonl").exists() - assert not (sessions_dir / "sess_topic_b.jsonl").exists() + mock_sqlite.assert_called_once() + assert mock_sqlite.call_args[0][0] == "sess_topic_a" def test_successful_mirror_uses_user_id_for_group_session(self, tmp_path): sessions_dir, index_file = _setup_sessions(tmp_path, { @@ -245,7 +218,7 @@ def test_successful_mirror_uses_user_id_for_group_session(self, tmp_path): with patch.object(mirror_mod, "_SESSIONS_DIR", sessions_dir), \ patch.object(mirror_mod, "_SESSIONS_INDEX", index_file), \ - patch("gateway.mirror._append_to_sqlite"): + patch("gateway.mirror._append_to_sqlite") as mock_sqlite: result = mirror_to_session( "telegram", "-1001", @@ -255,8 +228,8 @@ def test_successful_mirror_uses_user_id_for_group_session(self, tmp_path): ) assert result is True - assert (sessions_dir / "sess_alice.jsonl").exists() - assert not (sessions_dir / "sess_bob.jsonl").exists() + mock_sqlite.assert_called_once() + assert mock_sqlite.call_args[0][0] == "sess_alice" def test_no_matching_session(self, tmp_path): sessions_dir, index_file = _setup_sessions(tmp_path, {}) From 37208b68a6d40efc623160537274e0a41869ac53 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 09:30:05 +0200 Subject: [PATCH 6/9] docs(sessions): state.db is canonical for gateway messages --- website/docs/user-guide/sessions.md | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/website/docs/user-guide/sessions.md b/website/docs/user-guide/sessions.md index e412eefec8f60..25dac72aaec41 100644 --- a/website/docs/user-guide/sessions.md +++ b/website/docs/user-guide/sessions.md @@ -10,10 +10,9 @@ Hermes Agent automatically saves every conversation as a session. Sessions enabl ## How Sessions Work -Every conversation — whether from the CLI, Telegram, Discord, Slack, WhatsApp, Signal, Matrix, Teams, or any other messaging platform — is stored as a session with full message history. Sessions are tracked in two complementary systems: +Every conversation — whether from the CLI, Telegram, Discord, Slack, WhatsApp, Signal, Matrix, Teams, or any other messaging platform — is stored as a session with full message history. Sessions are tracked in: -1. **SQLite database** (`~/.hermes/state.db`) — structured session metadata with FTS5 full-text search -2. **JSONL transcripts** (`~/.hermes/sessions/`) — raw conversation transcripts including tool calls (gateway) +1. **SQLite database** (`~/.hermes/state.db`) — structured session metadata with FTS5 full-text search, plus full message history The SQLite database stores: - Session ID, source platform, user ID @@ -488,11 +487,18 @@ Sessions with **active background processes** are never auto-reset, regardless o | What | Path | Description | |------|------|-------------| | SQLite database | `~/.hermes/state.db` | All session metadata + messages with FTS5 | -| Gateway transcripts | `~/.hermes/sessions/` | JSONL transcripts per session + sessions.json index | -| Gateway index | `~/.hermes/sessions/sessions.json` | Maps session keys to active session IDs | +| Gateway messages | `~/.hermes/state.db` | SQLite — canonical store for all session messages | +| Gateway routing index | `~/.hermes/sessions/sessions.json` | Maps session keys to active session IDs (origin metadata, expiry flags) | The SQLite database uses WAL mode for concurrent readers and a single writer, which suits the gateway's multi-platform architecture well. +:::note Legacy JSONL transcripts +Sessions created before state.db became canonical may have leftover +`*.jsonl` files in `~/.hermes/sessions/`. They are no longer written or +read by Hermes. Safe to delete after verifying the corresponding session +exists in state.db. +::: + ### Database Schema Key tables in `state.db`: From 2aadc6e8d542a90991b8ba930ff618064b5cb008 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 11:08:06 +0200 Subject: [PATCH 7/9] test(gateway): pin DEFAULT_DB_PATH in fixtures to prevent real state.db writes Fixtures that instantiate SessionStore() trigger SessionDB() with no args, which resolves to ~/.hermes/state.db via the DEFAULT_DB_PATH module constant (snapshot of get_hermes_home() at hermes_state import time). The autouse _hermetic_environment fixture in tests/conftest.py monkeypatches HERMES_HOME env, but DEFAULT_DB_PATH is already cached by then. Per-test monkeypatch.setattr(hermes_state, 'DEFAULT_DB_PATH', tmp_path/'state.db') forces the DB into tmp_path so the tests can't leak into the real profile. Verified by counting u1-prefixed sessions in real state.db before/after: delta=0. --- .../platforms/test_yuanbao_recall_db_only.py | 11 +++++++++-- tests/gateway/test_load_transcript_db_only.py | 19 ++++++++++++------- tests/gateway/test_session.py | 12 +++++++++--- .../gateway/test_session_dm_thread_seeding.py | 12 ++++++++++-- 4 files changed, 40 insertions(+), 14 deletions(-) diff --git a/tests/gateway/platforms/test_yuanbao_recall_db_only.py b/tests/gateway/platforms/test_yuanbao_recall_db_only.py index 6186df6787a7a..da697f0193179 100644 --- a/tests/gateway/platforms/test_yuanbao_recall_db_only.py +++ b/tests/gateway/platforms/test_yuanbao_recall_db_only.py @@ -3,8 +3,15 @@ from gateway.config import GatewayConfig -def test_recall_falls_through_to_content_match_without_message_id(tmp_path): - """When transcript has no message_id field, A2 content-match still works.""" +def test_recall_falls_through_to_content_match_without_message_id(tmp_path, monkeypatch): + """When transcript has no message_id field, A2 content-match still works. + + Pin DEFAULT_DB_PATH to tmp_path so SessionDB() can't write to the real + ~/.hermes/state.db. (Module-level constant snapshot, see test_load_transcript_db_only.) + """ + import hermes_state + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + config = GatewayConfig() store = SessionStore(sessions_dir=tmp_path, config=config) diff --git a/tests/gateway/test_load_transcript_db_only.py b/tests/gateway/test_load_transcript_db_only.py index bc8b094dd1865..2425e495a6be9 100644 --- a/tests/gateway/test_load_transcript_db_only.py +++ b/tests/gateway/test_load_transcript_db_only.py @@ -1,13 +1,23 @@ """Verify load_transcript returns SQLite messages without any JSONL file.""" from pathlib import Path + import pytest from gateway.session import SessionStore from gateway.config import GatewayConfig -def test_load_transcript_returns_db_messages_when_no_jsonl(tmp_path): - """Reading a transcript must work from SQLite alone — no JSONL fallback needed.""" +def test_load_transcript_returns_db_messages_when_no_jsonl(tmp_path, monkeypatch): + """Reading a transcript must work from SQLite alone — no JSONL fallback needed. + + Pin DEFAULT_DB_PATH to tmp_path so this test cannot write to the real + ~/.hermes/state.db. (DEFAULT_DB_PATH is a module-level constant computed + at hermes_state import time, before pytest's HERMES_HOME monkeypatch + fires — the autouse fixture's HERMES_HOME override doesn't help here.) + """ + import hermes_state + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + config = GatewayConfig() store = SessionStore(sessions_dir=tmp_path, config=config) @@ -16,11 +26,6 @@ def test_load_transcript_returns_db_messages_when_no_jsonl(tmp_path): store.append_to_transcript(sid, {"role": "user", "content": "hello", "timestamp": 1.0}) store.append_to_transcript(sid, {"role": "assistant", "content": "world", "timestamp": 2.0}) - # Delete any JSONL that the current dual-writer left behind - jsonl_path = tmp_path / f"{sid}.jsonl" - if jsonl_path.exists(): - jsonl_path.unlink() - history = store.load_transcript(sid) assert len(history) == 2 assert history[0]["content"] == "hello" diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index 7e5aa1787c987..6e2c39f797277 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -504,7 +504,9 @@ class TestSessionStoreRewriteTranscript: """Regression: /retry and /undo must persist truncated history to DB.""" @pytest.fixture() - def store(self, tmp_path): + def store(self, tmp_path, monkeypatch): + import hermes_state + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") config = GatewayConfig() s = SessionStore(sessions_dir=tmp_path, config=config) return s @@ -546,13 +548,17 @@ def test_rewrite_with_empty_list(self, store): class TestLoadTranscriptDBOnly: """After spec 002, load_transcript reads only from state.db.""" - def test_db_only_returns_empty_for_nonexistent(self, tmp_path): + def test_db_only_returns_empty_for_nonexistent(self, tmp_path, monkeypatch): + import hermes_state + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") config = GatewayConfig() store = SessionStore(sessions_dir=tmp_path, config=config) result = store.load_transcript("nonexistent") assert result == [] - def test_db_only_returns_messages(self, tmp_path): + def test_db_only_returns_messages(self, tmp_path, monkeypatch): + import hermes_state + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") config = GatewayConfig() store = SessionStore(sessions_dir=tmp_path, config=config) sid = "db_only_session" diff --git a/tests/gateway/test_session_dm_thread_seeding.py b/tests/gateway/test_session_dm_thread_seeding.py index 8c52225bf2cb7..415e953baa2c3 100644 --- a/tests/gateway/test_session_dm_thread_seeding.py +++ b/tests/gateway/test_session_dm_thread_seeding.py @@ -22,8 +22,16 @@ @pytest.fixture() -def store(tmp_path): - """SessionStore with SQLite — load_transcript reads from DB only.""" +def store(tmp_path, monkeypatch): + """SessionStore with SQLite — load_transcript reads from DB only. + + Pin DEFAULT_DB_PATH to tmp_path so SessionDB() can't write to the real + ~/.hermes/state.db. (DEFAULT_DB_PATH is a module-level constant computed + at hermes_state import time, before pytest's HERMES_HOME monkeypatch + fires — the autouse fixture's HERMES_HOME override doesn't help here.) + """ + import hermes_state + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") config = GatewayConfig() s = SessionStore(sessions_dir=tmp_path, config=config) return s From b431ab5ca2a3cb9834c50514ec21c1a295bdcc47 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Wed, 20 May 2026 11:32:15 +0200 Subject: [PATCH 8/9] refactor(yuanbao): drop dead branch A1 message_id loop + pin missing fixture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #29211 review findings: 1. test_retry_replacement: pin DEFAULT_DB_PATH so SessionDB() doesn't write to the real ~/.hermes/state.db. Same fix as the other DB-only fixtures. 2. yuanbao recall branch A1 (message_id exact match) was structurally dead once load_transcript() became DB-only — state.db never preserves the platform message_id. Removed the dead loop, consolidated to a single content-match branch (renamed 'A: content match'). Branch B (system note) unchanged. Updated the test name + docstring to reflect this. Note: self._lock is no longer taken in append_to_transcript (was guarding the JSONL file append). SQLite append_message handles its own concurrency via WAL mode, so this is safe; flagging for awareness. --- gateway/platforms/yuanbao.py | 32 +++++++------------ .../platforms/test_yuanbao_recall_db_only.py | 13 ++++---- tests/gateway/test_retry_replacement.py | 7 +++- 3 files changed, 24 insertions(+), 28 deletions(-) diff --git a/gateway/platforms/yuanbao.py b/gateway/platforms/yuanbao.py index aed6717bd36a8..89b2a82942dfb 100644 --- a/gateway/platforms/yuanbao.py +++ b/gateway/platforms/yuanbao.py @@ -1410,32 +1410,24 @@ def _patch_transcript(cls, adapter, recalled_id: str, group_code: str, logger.warning("[%s] Recall: failed to resolve session: %s", adapter.name, exc) return - # Load transcript from canonical store (state.db). - # - # Branch A1 below tries to match the recalled message by its platform - # `message_id`. state.db does NOT preserve `message_id` (only its own - # autoincrement primary key), so A1 will not match for any message - # persisted post-DB-canonical (i.e. all messages going forward). Recall - # falls through to A2 (content match) or B (system redaction note), both - # of which work DB-only. - # - # TODO: add a `platform_message_id` column to state.db messages to restore - # exact-id matching. Tracked separately. + # Load transcript from canonical store (state.db). See Branch A below + # for why we can no longer match by platform `message_id`. try: transcript = store.load_transcript(sid) except Exception as exc: logger.warning("[%s] Recall: failed to load transcript: %s", adapter.name, exc) return - # Branch A: redact — try message_id first, then content fallback. - # Observed messages have message_id; agent-processed @bot messages - # only have content (run.py doesn't write message_id to transcript). + # Branch A: content-match redaction. state.db does NOT preserve the + # platform `message_id` (only its own autoincrement primary key), so we + # cannot redact by exact id. Match by content instead. Most yuanbao + # recalls carry the recalled text via `recalled_content`, which is + # sufficient for any non-duplicate message. + # + # TODO: add a `platform_message_id` column to state.db messages to + # restore exact-id matching. Tracked separately. target = None - for entry in transcript: - if entry.get("message_id") == recalled_id: - target = entry - break - if target is None and recalled_content: + if recalled_content: for entry in transcript: if entry.get("role") == "user" and entry.get("content") == recalled_content: target = entry @@ -1444,7 +1436,7 @@ def _patch_transcript(cls, adapter, recalled_id: str, group_code: str, target["content"] = cls._REDACTED try: store.rewrite_transcript(sid, transcript) - logger.info("[%s] Recall: redacted msg_id=%s (branch A)", adapter.name, recalled_id) + logger.info("[%s] Recall: redacted msg_id=%s (branch A: content match)", adapter.name, recalled_id) except Exception as exc: logger.warning("[%s] Recall: rewrite_transcript failed: %s", adapter.name, exc) return diff --git a/tests/gateway/platforms/test_yuanbao_recall_db_only.py b/tests/gateway/platforms/test_yuanbao_recall_db_only.py index da697f0193179..f54a5f3467986 100644 --- a/tests/gateway/platforms/test_yuanbao_recall_db_only.py +++ b/tests/gateway/platforms/test_yuanbao_recall_db_only.py @@ -1,10 +1,10 @@ -"""Yuanbao recall: branch A2 (content-match) works without JSONL message_id.""" +"""Yuanbao recall: branch A (content-match) works against DB-only transcripts.""" from gateway.session import SessionStore from gateway.config import GatewayConfig -def test_recall_falls_through_to_content_match_without_message_id(tmp_path, monkeypatch): - """When transcript has no message_id field, A2 content-match still works. +def test_recall_content_match_finds_target_in_db_transcript(tmp_path, monkeypatch): + """state.db doesn't preserve message_id, so recall uses content-match. Pin DEFAULT_DB_PATH to tmp_path so SessionDB() can't write to the real ~/.hermes/state.db. (Module-level constant snapshot, see test_load_transcript_db_only.) @@ -20,12 +20,11 @@ def test_recall_falls_through_to_content_match_without_message_id(tmp_path, monk store.append_to_transcript(sid, {"role": "user", "content": "sensitive content", "timestamp": 1.0}) store.append_to_transcript(sid, {"role": "assistant", "content": "ack", "timestamp": 2.0}) - # The post-PR state: load_transcript returns DB-only, no message_id field. + # DB-only history carries no platform message_id (PR #29211 dropped that path). history = store.load_transcript(sid) - assert all("message_id" not in msg for msg in history), \ - "DB-only history should not carry message_id" + assert all("message_id" not in msg for msg in history) - # Branch A2: content match should still find the message + # Branch A: content match finds the target row that recall would redact. target = next((m for m in history if m.get("role") == "user" and m.get("content") == "sensitive content"), None) assert target is not None diff --git a/tests/gateway/test_retry_replacement.py b/tests/gateway/test_retry_replacement.py index 571485caac209..3a6d0665875b1 100644 --- a/tests/gateway/test_retry_replacement.py +++ b/tests/gateway/test_retry_replacement.py @@ -11,7 +11,12 @@ @pytest.mark.asyncio -async def test_gateway_retry_replaces_last_user_turn_in_transcript(tmp_path): +async def test_gateway_retry_replaces_last_user_turn_in_transcript(tmp_path, monkeypatch): + # Pin DEFAULT_DB_PATH so SessionDB() doesn't write to the real ~/.hermes/state.db. + # (Module-level constant snapshot, see test_load_transcript_db_only.) + import hermes_state + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + config = GatewayConfig() store = SessionStore(sessions_dir=tmp_path, config=config) From 33f2e431d24473f258b76a7a09e22ace68505059 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 20 May 2026 12:55:01 -0700 Subject: [PATCH 9/9] feat(state.db): persist platform_message_id; restore yuanbao exact-id recall MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #29211 dropped JSONL gateway transcripts and noted that the platform's own `message_id` field (used by Yuanbao's recall guard to redact a message by exact platform id) was no longer preserved — falling back to content-match. That fallback works for the common case but redacts the wrong row when two messages share text (or fails to match when content is post-processed). Restore exact-id matching by giving state.db a column for it: - New `platform_message_id TEXT` column on the messages table (SCHEMA_VERSION bump 11 → 12; column added via declarative reconciler on existing DBs, no version-gated migration block needed) - Partial index `idx_messages_platform_msg_id` on (session_id, platform_message_id) to keep recall's point-lookup cheap even on large sessions - `append_message()` and `replace_messages()` accept the new value: the gateway-facing `append_to_transcript` in `gateway/session.py` forwards either `message["platform_message_id"]` or the legacy `message["message_id"]` key (yuanbao's existing convention) - `get_messages_as_conversation()` surfaces the column back on the message dict as `message_id` so platform code reads the same shape it used to read from JSONL - Yuanbao `_patch_transcript`: restore branch A1 (exact id match) ahead of A2 (content match) ahead of B (system-note). Both branches log which one fired so operators can tell from gateway.log whether recall hit the canonical path or had to fall back. Tests: - New low-level round-trip tests in `test_hermes_state.py` for both `append_message` and `replace_messages` paths - The PR's `test_yuanbao_recall_db_only.py` was rewritten to assert the new contract: branch A1 (id match) works against DB-only transcripts, and branch A2 (content match) still recovers rows that were observed without a platform id (e.g. agent-processed @bot messages where run.py doesn't carry msg_id through) --- gateway/platforms/yuanbao.py | 34 ++++--- gateway/session.py | 6 ++ hermes_state.py | 49 ++++++++-- .../platforms/test_yuanbao_recall_db_only.py | 89 +++++++++++++++---- tests/test_hermes_state.py | 45 +++++++++- 5 files changed, 185 insertions(+), 38 deletions(-) diff --git a/gateway/platforms/yuanbao.py b/gateway/platforms/yuanbao.py index 89b2a82942dfb..18d0787c97845 100644 --- a/gateway/platforms/yuanbao.py +++ b/gateway/platforms/yuanbao.py @@ -1410,33 +1410,43 @@ def _patch_transcript(cls, adapter, recalled_id: str, group_code: str, logger.warning("[%s] Recall: failed to resolve session: %s", adapter.name, exc) return - # Load transcript from canonical store (state.db). See Branch A below - # for why we can no longer match by platform `message_id`. + # Load transcript from canonical store (state.db). Since PR #29278 + # added a ``platform_message_id`` column to the messages table and + # ``append_to_transcript`` wires the incoming dict's ``message_id`` + # into it, ``load_transcript`` returns rows with ``message_id`` set + # for any message that was observed with one — Branch A1 (exact id + # match) is the canonical path again. try: transcript = store.load_transcript(sid) except Exception as exc: logger.warning("[%s] Recall: failed to load transcript: %s", adapter.name, exc) return - # Branch A: content-match redaction. state.db does NOT preserve the - # platform `message_id` (only its own autoincrement primary key), so we - # cannot redact by exact id. Match by content instead. Most yuanbao - # recalls carry the recalled text via `recalled_content`, which is - # sufficient for any non-duplicate message. - # - # TODO: add a `platform_message_id` column to state.db messages to - # restore exact-id matching. Tracked separately. + # Branch A1: exact platform message_id match. Authoritative when the + # row was persisted with a platform_message_id (observed group + # messages and any inbound message whose adapter carried a msg_id). target = None - if recalled_content: + branch_label = "" + for entry in transcript: + if entry.get("message_id") == recalled_id: + target = entry + branch_label = "branch A1: id match" + break + # Branch A2: content-match fallback for messages that lack an exact + # platform id on the row — e.g. agent-processed @bot messages + # (run.py doesn't carry msg_id through) or older rows persisted + # before the platform_message_id column existed. + if target is None and recalled_content: for entry in transcript: if entry.get("role") == "user" and entry.get("content") == recalled_content: target = entry + branch_label = "branch A2: content match" break if target is not None: target["content"] = cls._REDACTED try: store.rewrite_transcript(sid, transcript) - logger.info("[%s] Recall: redacted msg_id=%s (branch A: content match)", adapter.name, recalled_id) + logger.info("[%s] Recall: redacted msg_id=%s (%s)", adapter.name, recalled_id, branch_label) except Exception as exc: logger.warning("[%s] Recall: rewrite_transcript failed: %s", adapter.name, exc) return diff --git a/gateway/session.py b/gateway/session.py index 4ad2600c1e8a0..648f8cddf1076 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1271,6 +1271,12 @@ def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db reasoning_details=message.get("reasoning_details") if message.get("role") == "assistant" else None, codex_reasoning_items=message.get("codex_reasoning_items") if message.get("role") == "assistant" else None, codex_message_items=message.get("codex_message_items") if message.get("role") == "assistant" else None, + # Platform-side message id (yuanbao msg_id, telegram update_id, …). + # Accept either explicit ``platform_message_id`` or the legacy + # ``message_id`` key the JSONL transcript used. + platform_message_id=( + message.get("platform_message_id") or message.get("message_id") + ), ) except Exception as e: logger.debug("Session DB operation failed: %s", e) diff --git a/hermes_state.py b/hermes_state.py index e8e8947c05a15..5804437198a8f 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -33,7 +33,7 @@ DEFAULT_DB_PATH = get_hermes_home() / "state.db" -SCHEMA_VERSION = 11 +SCHEMA_VERSION = 12 # --------------------------------------------------------------------------- # WAL-compatibility fallback @@ -236,7 +236,8 @@ def _log_wal_fallback_once(db_label: str, exc: Exception) -> None: reasoning_content TEXT, reasoning_details TEXT, codex_reasoning_items TEXT, - codex_message_items TEXT + codex_message_items TEXT, + platform_message_id TEXT ); CREATE TABLE IF NOT EXISTS state_meta ( @@ -571,6 +572,19 @@ def _init_schema(self): # column gets created here. self._reconcile_columns(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 + # WHERE clause references a column that doesn't exist yet). + try: + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_messages_platform_msg_id " + "ON messages(session_id, platform_message_id) " + "WHERE platform_message_id IS NOT NULL" + ) + except sqlite3.OperationalError as exc: + logger.debug("idx_messages_platform_msg_id create skipped: %s", exc) + # ── Schema version bookkeeping ───────────────────────────────── # Bump to current so future data migrations (if any) can gate on # version. No version-gated column additions remain. @@ -1445,12 +1459,19 @@ def append_message( reasoning_details: Any = None, codex_reasoning_items: Any = None, codex_message_items: Any = None, + platform_message_id: str = None, ) -> int: """ Append a message to a session. Returns the message row ID. Also increments the session's message_count (and tool_call_count if role is 'tool' or tool_calls is present). + + ``platform_message_id`` is the external messaging platform's own + message ID (e.g. Telegram update_id, Yuanbao msg_id). It is + independent of the SQLite autoincrement primary key and is used by + platform-specific flows like yuanbao's recall guard to redact a + message by its platform-side identifier. """ # Serialize structured fields to JSON before entering the write txn reasoning_details_json = ( @@ -1480,8 +1501,8 @@ def _do(conn): """INSERT INTO messages (session_id, role, content, tool_call_id, tool_calls, tool_name, timestamp, token_count, finish_reason, reasoning, reasoning_content, reasoning_details, codex_reasoning_items, - codex_message_items) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + codex_message_items, platform_message_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( session_id, role, @@ -1497,6 +1518,7 @@ def _do(conn): reasoning_details_json, codex_items_json, codex_message_items_json, + platform_message_id, ), ) msg_id = cursor.lastrowid @@ -1558,13 +1580,18 @@ def _do(conn): json.dumps(codex_message_items) if codex_message_items else None ) tool_calls_json = json.dumps(tool_calls) if tool_calls else None + # Accept either `platform_message_id` (new explicit name) or + # `message_id` (yuanbao's existing convention on message dicts). + platform_msg_id = ( + msg.get("platform_message_id") or msg.get("message_id") + ) conn.execute( """INSERT INTO messages (session_id, role, content, tool_call_id, tool_calls, tool_name, timestamp, token_count, finish_reason, reasoning, reasoning_content, reasoning_details, codex_reasoning_items, - codex_message_items) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + codex_message_items, platform_message_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( session_id, role, @@ -1580,6 +1607,7 @@ def _do(conn): reasoning_details_json, codex_items_json, codex_message_items_json, + platform_msg_id, ), ) total_messages += 1 @@ -1897,7 +1925,7 @@ def get_messages_as_conversation( rows = self._conn.execute( "SELECT role, content, tool_call_id, tool_calls, tool_name, " "finish_reason, reasoning, reasoning_content, reasoning_details, " - "codex_reasoning_items, codex_message_items " + "codex_reasoning_items, codex_message_items, platform_message_id " f"FROM messages WHERE session_id IN ({placeholders}) ORDER BY id", tuple(session_ids), ).fetchall() @@ -1918,6 +1946,13 @@ def get_messages_as_conversation( except (json.JSONDecodeError, TypeError): logger.warning("Failed to deserialize tool_calls in conversation replay, falling back to []") msg["tool_calls"] = [] + # Surface the platform-side message id (e.g. yuanbao msg_id, + # telegram update_id) so platform-specific flows like recall + # can match by external identifier instead of having to fall + # back to content-match heuristics. Exposed as ``message_id`` + # for backward compatibility with the JSONL transcript shape. + if row["platform_message_id"]: + msg["message_id"] = row["platform_message_id"] # Restore reasoning fields on assistant messages so providers # that replay reasoning (OpenRouter, OpenAI, Nous) receive # coherent multi-turn reasoning context. diff --git a/tests/gateway/platforms/test_yuanbao_recall_db_only.py b/tests/gateway/platforms/test_yuanbao_recall_db_only.py index f54a5f3467986..3b8cd6d912b89 100644 --- a/tests/gateway/platforms/test_yuanbao_recall_db_only.py +++ b/tests/gateway/platforms/test_yuanbao_recall_db_only.py @@ -1,31 +1,88 @@ -"""Yuanbao recall: branch A (content-match) works against DB-only transcripts.""" +"""Yuanbao recall: branch A1 (exact id) and A2 (content-match) against DB-only transcripts. + +state.db persists the platform-side ``message_id`` via the +``platform_message_id`` column (added in the salvage of PR #29211) and +``load_transcript`` surfaces it back on each message dict as ``message_id`` +— so the recall guard's exact-id match path stays canonical even with the +JSONL file gone. When a row has no platform id (e.g. agent-processed +@bot messages whose adapter didn't carry a msg_id, or pre-column legacy +rows), recall falls through to content-match. +""" from gateway.session import SessionStore from gateway.config import GatewayConfig -def test_recall_content_match_finds_target_in_db_transcript(tmp_path, monkeypatch): - """state.db doesn't preserve message_id, so recall uses content-match. - - Pin DEFAULT_DB_PATH to tmp_path so SessionDB() can't write to the real - ~/.hermes/state.db. (Module-level constant snapshot, see test_load_transcript_db_only.) - """ +def _pin_db(monkeypatch, tmp_path): + """Force SessionDB() to write into tmp_path instead of the real ~/.hermes.""" import hermes_state monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + +def test_recall_branch_a1_exact_id_match_round_trips_through_db(tmp_path, monkeypatch): + """A user message persisted with ``message_id`` must round-trip through + state.db so recall can find and redact it by exact id (branch A1).""" + _pin_db(monkeypatch, tmp_path) + + config = GatewayConfig() + store = SessionStore(sessions_dir=tmp_path, config=config) + + sid = "test-yuanbao-recall-a1" + store._db.create_session(session_id=sid, source="yuanbao:group:G") + store.append_to_transcript(sid, { + "role": "user", + "content": "sensitive content", + "timestamp": 1.0, + "message_id": "platform-msg-abc", + }) + store.append_to_transcript(sid, { + "role": "assistant", + "content": "ack", + "timestamp": 2.0, + }) + + history = store.load_transcript(sid) + # The user row must carry its platform id back so the recall guard can + # match by exact id; the assistant row had no platform id so it should + # not gain one spuriously. + user_msg = next(m for m in history if m["role"] == "user") + assistant_msg = next(m for m in history if m["role"] == "assistant") + assert user_msg.get("message_id") == "platform-msg-abc" + assert "message_id" not in assistant_msg + + # Branch A1: locate the row by exact platform id — no content heuristics. + target = next( + (m for m in history if m.get("message_id") == "platform-msg-abc"), + None, + ) + assert target is not None + assert target["content"] == "sensitive content" + + +def test_recall_branch_a2_content_match_when_no_platform_id(tmp_path, monkeypatch): + """Rows that lack a platform_message_id (e.g. agent-processed @bot + messages) still match by content as a fallback.""" + _pin_db(monkeypatch, tmp_path) + config = GatewayConfig() store = SessionStore(sessions_dir=tmp_path, config=config) - sid = "test-yuanbao-recall" + sid = "test-yuanbao-recall-a2" store._db.create_session(session_id=sid, source="yuanbao:group:G") - store.append_to_transcript(sid, {"role": "user", "content": "sensitive content", "timestamp": 1.0}) - store.append_to_transcript(sid, {"role": "assistant", "content": "ack", "timestamp": 2.0}) + # No message_id on the dict — simulates an agent-processed message + # that did not carry the platform msg_id through. + store.append_to_transcript(sid, { + "role": "user", + "content": "sensitive content", + "timestamp": 1.0, + }) - # DB-only history carries no platform message_id (PR #29211 dropped that path). history = store.load_transcript(sid) - assert all("message_id" not in msg for msg in history) + assert all("message_id" not in m for m in history) - # Branch A: content match finds the target row that recall would redact. - target = next((m for m in history - if m.get("role") == "user" and m.get("content") == "sensitive content"), None) + # Branch A2: content match recovers the target. + target = next( + (m for m in history + if m.get("role") == "user" and m.get("content") == "sensitive content"), + None, + ) assert target is not None - # Caller would then redact: target["content"] = REDACTED; store.rewrite_transcript(sid, history) diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 2676457f58b18..7c3cae75523d1 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -316,6 +316,42 @@ def test_get_messages_as_conversation(self, db): assert conv[0] == {"role": "user", "content": "Hello"} assert conv[1] == {"role": "assistant", "content": "Hi!"} + def test_platform_message_id_round_trips(self, db): + """Platform-side message ids (yuanbao msg_id, telegram update_id, …) + survive append → get_messages_as_conversation under the + ``message_id`` key so platform recall flows can match by exact id.""" + db.create_session(session_id="s_pmi", source="yuanbao") + db.append_message( + "s_pmi", + role="user", + content="hi", + platform_message_id="abc-123", + ) + db.append_message("s_pmi", role="assistant", content="hello") + + conv = db.get_messages_as_conversation("s_pmi") + user_msg = next(m for m in conv if m["role"] == "user") + assistant_msg = next(m for m in conv if m["role"] == "assistant") + assert user_msg.get("message_id") == "abc-123" + # Assistant row had no platform id — must not gain one spuriously. + assert "message_id" not in assistant_msg + + def test_replace_messages_preserves_platform_message_id(self, db): + """``rewrite_transcript`` (which goes through replace_messages) must + keep the platform_message_id round-trip working for /retry, /undo, + /compress and yuanbao's recall rewrite path.""" + db.create_session(session_id="s_rep", source="yuanbao") + db.replace_messages( + "s_rep", + [ + {"role": "user", "content": "x", "message_id": "ext-1"}, + {"role": "assistant", "content": "y"}, + ], + ) + conv = db.get_messages_as_conversation("s_rep") + assert next(m for m in conv if m["role"] == "user").get("message_id") == "ext-1" + assert "message_id" not in next(m for m in conv if m["role"] == "assistant") + def test_get_messages_as_conversation_includes_ancestor_chain(self, db): db.create_session("root", "tui") db.append_message("root", role="user", content="first prompt") @@ -1462,9 +1498,10 @@ def test_tables_exist(self, db): assert "schema_version" in tables def test_schema_version(self, db): + from hermes_state import SCHEMA_VERSION cursor = db._conn.execute("SELECT version FROM schema_version") version = cursor.fetchone()[0] - assert version == 11 + assert version == SCHEMA_VERSION def test_title_column_exists(self, db): """Verify the title column was created in the sessions table.""" @@ -1760,8 +1797,9 @@ def test_migration_from_v2(self, tmp_path): migrated_db = SessionDB(db_path=db_path) # Verify migration + from hermes_state import SCHEMA_VERSION cursor = migrated_db._conn.execute("SELECT version FROM schema_version") - assert cursor.fetchone()[0] == 11 + assert cursor.fetchone()[0] == SCHEMA_VERSION # Verify title column exists and is NULL for existing sessions session = migrated_db.get_session("existing") @@ -2952,11 +2990,12 @@ def test_v10_to_v11_upgrade_backfills_tool_fields(self, tmp_path): assert len(session_db.search_messages("LEGACYARG")) == 1, \ "v11 migration must backfill tool_calls JSON into FTS" # schema_version bumped + from hermes_state import SCHEMA_VERSION row = session_db._conn.execute( "SELECT version FROM schema_version LIMIT 1" ).fetchone() version = row["version"] if hasattr(row, "keys") else row[0] - assert version == 11 + assert version == SCHEMA_VERSION finally: session_db.close()