diff --git a/hermes_state_schema.py b/hermes_state_schema.py index f9d25208b9a4e..ce9c48e857b86 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -16,6 +16,7 @@ from typing import Dict, Optional from hermes_constants import get_hermes_home from hermes_state_common import ( DEFERRED_INDEX_SQL, + FTS_CJK_STALE_KEY, FTS_SQL, FTS_STORAGE_VERSION, FTS_TRIGRAM_SQL, @@ -86,14 +87,20 @@ class SessionSchemaMixin: """ import re as _re + # CJK is a v23-only surface. Decide the layout before selecting + # destructive candidates so the legacy branch never drops a trigger + # it does not recreate. + legacy_layout = self._db_has_legacy_inline_fts(cursor) update_names = ( "messages_fts_update", "messages_fts_trigram_update", - "messages_fts_cjk_update", ) + if not legacy_layout and hasattr(self, "_ensure_fts_cjk_schema"): + update_names += ("messages_fts_cjk_update",) + placeholders = ", ".join("?" for _ in update_names) rows = cursor.execute( "SELECT name, sql FROM sqlite_master " - "WHERE type = 'trigger' AND name IN (?, ?, ?)", + f"WHERE type = 'trigger' AND name IN ({placeholders})", update_names, ).fetchall() to_drop = [] @@ -113,7 +120,7 @@ class SessionSchemaMixin: # Re-apply current DDL so CREATE TRIGGER installs the OF variants. # Choose legacy vs v23 the same way _init_schema does. - if self._db_has_legacy_inline_fts(cursor): + if legacy_layout: self._ensure_fts_schema(cursor, "messages_fts", LEGACY_FTS_SQL) self._ensure_fts_schema( cursor, "messages_fts_trigram", LEGACY_FTS_TRIGRAM_SQL @@ -123,15 +130,27 @@ class SessionSchemaMixin: self._ensure_fts_schema( cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL ) - # CJK triggers live on the host SessionDB; only recreate if present. - if hasattr(self, "_ensure_fts_cjk_schema"): + # CJK triggers live on the host SessionDB; only recreate one that + # this migration actually dropped. Unexpected ensure failures + # must not leave the host advertising a now-triggerless index. + if ( + "messages_fts_cjk_update" in to_drop + ): try: self._ensure_fts_cjk_schema(cursor) except Exception: - logger.debug( - "CJK FTS re-ensure after UPDATE OF migration skipped", - exc_info=True, + self._fts_cjk_available = False + try: + self.set_meta(FTS_CJK_STALE_KEY, "1", cursor=cursor) + except Exception: + logger.debug( + "Could not persist CJK FTS stale breadcrumb", + exc_info=True, + ) + logger.exception( + "CJK FTS re-ensure after UPDATE OF migration failed" ) + raise logger.info( "Migrated %d broad FTS UPDATE trigger(s) to AFTER UPDATE OF " diff --git a/tests/test_fts_update_of_narrowing.py b/tests/test_fts_update_of_narrowing.py index 48c3e2e4d532a..f66c10e7eed5b 100644 --- a/tests/test_fts_update_of_narrowing.py +++ b/tests/test_fts_update_of_narrowing.py @@ -9,6 +9,7 @@ from pathlib import Path import pytest from hermes_state import SessionDB +from hermes_state_common import FTS_CJK_STALE_KEY from hermes_state_schema import SessionSchemaMixin @@ -20,6 +21,59 @@ def _trigger_sql(conn: sqlite3.Connection, name: str) -> str | None: return row[0] if row else None +def _assert_nonindexed_updates_bypass_missing_fts_target( + db: SessionDB, message_id: int +) -> None: + """Prove the UPDATE OF gate, not the trigger's content-change WHEN.""" + db._conn.execute("DROP TABLE messages_fts") + assert _trigger_sql(db._conn, "messages_fts_update") is not None + + db._conn.execute( + "UPDATE messages SET active = 0, compacted = 1, observed = 1 " + "WHERE id = ?", + (message_id,), + ) + with pytest.raises(sqlite3.OperationalError, match=r"no such table.*messages_fts"): + db._conn.execute( + "UPDATE messages SET content = 'changed' WHERE id = ?", + (message_id,), + ) + + +def _install_legacy_inline_base_fts(db: SessionDB) -> None: + """Replace v23 FTS with the broad inline shape shipped by v11..v22.""" + db._drop_fts_triggers(db._conn) + db._conn.executescript( + """ + DROP TABLE IF EXISTS messages_fts; + DROP TABLE IF EXISTS messages_fts_trigram; + DROP VIEW IF EXISTS messages_fts_trigram_src; + + CREATE VIRTUAL TABLE messages_fts USING fts5(content); + CREATE TRIGGER messages_fts_insert AFTER INSERT ON messages BEGIN + INSERT INTO messages_fts(rowid, content) VALUES ( + new.id, + COALESCE(new.content, '') || ' ' || + COALESCE(new.tool_name, '') || ' ' || + COALESCE(new.tool_calls, '') + ); + END; + CREATE TRIGGER messages_fts_delete AFTER DELETE ON messages BEGIN + DELETE FROM messages_fts WHERE rowid = old.id; + END; + CREATE TRIGGER messages_fts_update AFTER UPDATE ON messages BEGIN + DELETE FROM messages_fts WHERE rowid = old.id; + INSERT INTO messages_fts(rowid, content) VALUES ( + new.id, + COALESCE(new.content, '') || ' ' || + COALESCE(new.tool_name, '') || ' ' || + COALESCE(new.tool_calls, '') + ); + END; + """ + ) + + def test_fresh_db_installs_update_of_triggers(tmp_path: Path): db = SessionDB(db_path=tmp_path / "state.db") try: @@ -79,18 +133,114 @@ def test_needs_narrowing_helper(): assert not SessionSchemaMixin._fts_update_trigger_needs_narrowing(None) -def test_status_only_update_does_not_require_content_change(tmp_path: Path): - """Smoke: DB opens and accepts message updates under narrowed triggers.""" +def test_v23_status_only_update_bypasses_fts_trigger_body(tmp_path: Path): db = SessionDB(db_path=tmp_path / "state.db") try: sid = "s1" db.create_session(sid, source="test") mid = db.append_message(sid, role="user", content="hello searchable") - db._conn.execute( - "UPDATE messages SET content = content WHERE id = ?", - (mid,), - ) - db._conn.commit() - assert _trigger_sql(db._conn, "messages_fts_update") + _assert_nonindexed_updates_bypass_missing_fts_target(db, mid) + finally: + db.close() + + +def test_legacy_status_only_update_bypasses_migrated_fts_trigger(tmp_path: Path): + db = SessionDB(db_path=tmp_path / "state.db") + try: + sid = "s1" + db.create_session(sid, source="test") + mid = db.append_message(sid, role="user", content="legacy searchable") + _install_legacy_inline_base_fts(db) + + assert db._db_has_legacy_inline_fts(db._conn) + assert db._migrate_broad_fts_update_triggers(db._conn) >= 1 + assert "AFTER UPDATE OF" in " ".join( + _trigger_sql(db._conn, "messages_fts_update").split() + ).upper() + + _assert_nonindexed_updates_bypass_missing_fts_target(db, mid) + finally: + db.close() + + +def test_cjk_ensure_failure_marks_unavailable_and_propagates( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + path = tmp_path / "state.db" + db = SessionDB(db_path=path) + try: + db._conn.execute("DROP TRIGGER IF EXISTS messages_fts_cjk_update") + db._conn.execute( + "CREATE TRIGGER messages_fts_cjk_update " + "AFTER UPDATE ON messages BEGIN SELECT 1; END" + ) + db._fts_cjk_available = True + + def _fail_cjk_ensure(_cursor): + raise sqlite3.DatabaseError("injected CJK ensure failure") + + monkeypatch.setattr(db, "_ensure_fts_cjk_schema", _fail_cjk_ensure) + + with pytest.raises(sqlite3.DatabaseError, match="injected CJK ensure failure"): + db._migrate_broad_fts_update_triggers(db._conn) + assert db._fts_cjk_available is False + assert _trigger_sql(db._conn, "messages_fts_cjk_update") is None + + # The DROP is autocommitted. Fail-closed therefore needs a durable + # breadcrumb that other processes can observe, not just an instance + # flag on the SessionDB whose initialization is aborting. + with sqlite3.connect(path) as observer: + stale = observer.execute( + "SELECT value FROM state_meta WHERE key = ?", + (FTS_CJK_STALE_KEY,), + ).fetchone() + assert stale == ("1",) + assert _trigger_sql(observer, "messages_fts_cjk_update") is None + finally: + db.close() + + +def test_cjk_broad_trigger_is_restored_as_update_of( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + db = SessionDB(db_path=tmp_path / "state.db") + try: + db._conn.execute("DROP TRIGGER IF EXISTS messages_fts_cjk_update") + db._conn.execute( + "CREATE TRIGGER messages_fts_cjk_update " + "AFTER UPDATE ON messages BEGIN SELECT 1; END" + ) + + def _restore_cjk_update_trigger(cursor): + cursor.execute( + "CREATE TRIGGER messages_fts_cjk_update " + "AFTER UPDATE OF content, tool_name, tool_calls ON messages " + "BEGIN SELECT 1; END" + ) + + monkeypatch.setattr(db, "_ensure_fts_cjk_schema", _restore_cjk_update_trigger) + + assert db._migrate_broad_fts_update_triggers(db._conn) == 1 + cjk_sql = _trigger_sql(db._conn, "messages_fts_cjk_update") + assert cjk_sql is not None + assert "AFTER UPDATE OF" in " ".join(cjk_sql.split()).upper() + finally: + db.close() + + +def test_legacy_migration_does_not_drop_cjk_trigger(tmp_path: Path): + db = SessionDB(db_path=tmp_path / "state.db") + try: + _install_legacy_inline_base_fts(db) + db._conn.execute("DROP TRIGGER IF EXISTS messages_fts_cjk_update") + db._conn.execute( + "CREATE TRIGGER messages_fts_cjk_update " + "AFTER UPDATE ON messages BEGIN SELECT 1; END" + ) + + assert db._migrate_broad_fts_update_triggers(db._conn) >= 1 + cjk_sql = _trigger_sql(db._conn, "messages_fts_cjk_update") + assert cjk_sql is not None + assert "AFTER UPDATE OF" not in " ".join(cjk_sql.split()).upper() finally: db.close()