fix(state): narrow FTS UPDATE triggers with AFTER UPDATE OF + migration

hermes_state.py

@@ -1486,7 +1486,8 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
     VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls);
 END;
 
-CREATE TRIGGER IF NOT EXISTS messages_fts_cjk_update AFTER UPDATE ON messages
+CREATE TRIGGER IF NOT EXISTS messages_fts_cjk_update
+AFTER UPDATE OF content, tool_name, tool_calls, role ON messages
 WHEN (old.content IS NOT new.content
     OR old.tool_name IS NOT new.tool_name
     OR old.tool_calls IS NOT new.tool_calls

hermes_state_common.py

@@ -357,7 +357,11 @@ def _ephemeral_child_sql(alias: str = "s") -> str:
     VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls);
 END;
 
-CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE ON messages
+-- UPDATE OF skips the trigger entirely for non-content column writes
+-- (status/compacted/observed/etc.), which is stronger than the WHEN gate
+-- alone and avoids FTS I/O saturation on large state.db (#68858 / #73639).
+CREATE TRIGGER IF NOT EXISTS messages_fts_update
+AFTER UPDATE OF content, tool_name, tool_calls ON messages
 WHEN (old.content IS NOT new.content
     OR old.tool_name IS NOT new.tool_name
     OR old.tool_calls IS NOT new.tool_calls)
@@ -425,7 +429,8 @@ def _ephemeral_child_sql(alias: str = "s") -> str:
     VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls);
 END;
 
-CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update AFTER UPDATE ON messages
+CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update
+AFTER UPDATE OF content, tool_name, tool_calls, role ON messages
 WHEN (old.content IS NOT new.content
     OR old.tool_name IS NOT new.tool_name
     OR old.tool_calls IS NOT new.tool_calls
@@ -486,7 +491,8 @@ def _ephemeral_child_sql(alias: str = "s") -> str:
     DELETE FROM messages_fts WHERE rowid = old.id;
 END;
 
-CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE ON messages BEGIN
+CREATE TRIGGER IF NOT EXISTS messages_fts_update
+AFTER UPDATE OF content, tool_name, tool_calls ON messages BEGIN
     DELETE FROM messages_fts WHERE rowid = old.id;
     INSERT INTO messages_fts(rowid, content) VALUES (
         new.id,
@@ -513,7 +519,8 @@ def _ephemeral_child_sql(alias: str = "s") -> str:
     DELETE FROM messages_fts_trigram WHERE rowid = old.id;
 END;
 
-CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update AFTER UPDATE ON messages BEGIN
+CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update
+AFTER UPDATE OF content, tool_name, tool_calls ON messages BEGIN
     DELETE FROM messages_fts_trigram WHERE rowid = old.id;
     INSERT INTO messages_fts_trigram(rowid, content) VALUES (
         new.id,

hermes_state_schema.py

@@ -16,6 +16,7 @@
 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,
@@ -56,6 +57,146 @@ def _fts_trigger_count(cursor: sqlite3.Cursor) -> int:
         ).fetchone()
         return int(row[0] if not isinstance(row, sqlite3.Row) else row[0])
 
+
+    @staticmethod
+    def _fts_update_trigger_needs_narrowing(sql: Optional[str]) -> bool:
+        """True when trigger SQL is missing AFTER UPDATE OF (still broad)."""
+        if not sql:
+            return False
+        # Collapse whitespace so multi-line DDL still matches.
+        compact = " ".join(sql.split()).upper()
+        # Already narrowed.
+        if "AFTER UPDATE OF " in compact:
+            return False
+        # Broad UPDATE trigger that we still need to replace.
+        return "AFTER UPDATE ON " in compact
+
+    def _migrate_broad_fts_update_triggers(self, cursor: sqlite3.Cursor) -> int:
+        """Replace broad AFTER UPDATE FTS triggers with AFTER UPDATE OF variants.
+
+        ``CREATE TRIGGER IF NOT EXISTS`` will not replace an existing broad
+        trigger, so installs that already created ``AFTER UPDATE ON messages``
+        would keep firing on every messages row touch (status/compaction
+        writes included). Inspect ``sqlite_master``, drop any still-broad
+        UPDATE triggers, and re-apply the current DDL constants.
+
+        No FTS rebuild: content correctness was already gated by WHEN clauses
+        on modern installs; OF only skips unnecessary trigger evaluation.
+
+        Returns the number of triggers dropped (0 when already converged).
+        """
+        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",
+        )
+        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 "
+            f"WHERE type = 'trigger' AND name IN ({placeholders})",
+            update_names,
+        ).fetchall()
+        to_drop = []
+        for row in rows:
+            name = row[0] if not isinstance(row, sqlite3.Row) else row["name"]
+            sql = row[1] if not isinstance(row, sqlite3.Row) else row["sql"]
+            if self._fts_update_trigger_needs_narrowing(sql):
+                to_drop.append(name)
+        if not to_drop:
+            return 0
+
+        for name in to_drop:
+            # Trigger names are from our fixed allowlist, not user input.
+            if not _re.fullmatch(r"[A-Za-z0-9_]+", name):
+                continue
+            cursor.execute(f"DROP TRIGGER IF EXISTS {name}")
+
+        # Re-apply current DDL so CREATE TRIGGER installs the OF variants.
+        # Choose legacy vs v23 the same way _init_schema does.
+        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
+            )
+        else:
+            self._ensure_fts_schema(cursor, "messages_fts", FTS_SQL)
+            self._ensure_fts_schema(
+                cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL
+            )
+            # CJK triggers live on the host SessionDB; only recreate one that
+            # this migration actually dropped. ``_ensure_fts_cjk_schema`` is
+            # documented never-raises and soft-fails OperationalError by
+            # clearing availability — raise-path handling alone is not
+            # enough. After ensure, require a narrowed CJK UPDATE trigger or
+            # durable quarantine (stale breadcrumb + unavailable).
+            if "messages_fts_cjk_update" in to_drop:
+                try:
+                    self._ensure_fts_cjk_schema(cursor)
+                except Exception:
+                    self._quarantine_cjk_after_update_of_migration(cursor)
+                    logger.exception(
+                        "CJK FTS re-ensure after UPDATE OF migration failed"
+                    )
+                    raise
+                if not self._cjk_update_trigger_is_narrowed(cursor):
+                    self._quarantine_cjk_after_update_of_migration(cursor)
+                    logger.warning(
+                        "CJK FTS UPDATE trigger missing or still broad after "
+                        "UPDATE OF migration; marked stale and unavailable"
+                    )
+
+        logger.info(
+            "Migrated %d broad FTS UPDATE trigger(s) to AFTER UPDATE OF "
+            "(no rebuild required)",
+            len(to_drop),
+        )
+        return len(to_drop)
+
+    def _cjk_update_trigger_is_narrowed(self, cursor: sqlite3.Cursor) -> bool:
+        """True when messages_fts_cjk_update exists with AFTER UPDATE OF."""
+        row = cursor.execute(
+            "SELECT sql FROM sqlite_master "
+            "WHERE type = 'trigger' AND name = ?",
+            ("messages_fts_cjk_update",),
+        ).fetchone()
+        if not row:
+            return False
+        sql = row[0] if not isinstance(row, sqlite3.Row) else row["sql"]
+        return not self._fts_update_trigger_needs_narrowing(sql)
+
+    def _quarantine_cjk_after_update_of_migration(
+        self, cursor: sqlite3.Cursor
+    ) -> None:
+        """Fail-closed after dropping CJK UPDATE during OF migration.
+
+        Clears availability, persists ``fts_cjk_stale``, and drops any
+        residual broad/partial CJK UPDATE trigger so a later open cannot
+        ``CREATE TRIGGER IF NOT EXISTS`` a gap without rebuild.
+        """
+        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,
+            )
+        try:
+            cursor.execute("DROP TRIGGER IF EXISTS messages_fts_cjk_update")
+        except Exception:
+            logger.debug(
+                "Could not drop residual CJK UPDATE trigger after quarantine",
+                exc_info=True,
+            )
+
+
     @staticmethod
     def _rebuild_fts_indexes(
         cursor: sqlite3.Cursor,
@@ -853,6 +994,11 @@ def _init_schema(self):
                     # the surfaces above and gated on the loadable tokenizer:
                     self._ensure_fts_cjk_schema(cursor)
 
+            # Replace any pre-existing broad AFTER UPDATE triggers with
+            # AFTER UPDATE OF variants. IF NOT EXISTS cannot rewrite them.
+            if getattr(self, "_fts_enabled", False):
+                self._migrate_broad_fts_update_triggers(cursor)
+
         self._conn.commit()
 
     def _backfill_gateway_metadata_from_sessions_json(