fix(state): make the FTS write-health probe flush segments and catch IntegrityError

`_db_opens_cleanly` drove one probe row through the messages_fts* triggers
and rolled back. FTS5 only buffers that row in an in-memory segment until
commit, so the probe never wrote to `<fts>_idx`/`_data` and could not hit a
stale `messages_fts_trigram_idx` row waiting at the next segid — the class
where PRAGMA integrity_check, the FTS5 integrity-check command and MATCH all
report clean while every committed append fails with
`IntegrityError: constraint failed`. The probe also caught only
OperationalError; IntegrityError is a DatabaseError sibling, so even a
colliding probe would have escaped and been reported as healthy.

Now the probe issues `INSERT INTO <fts>(<fts>) VALUES('flush')` for every
FTS family inside the rolled-back transaction (capability / not-built errors
stay benign), catches sqlite3.DatabaseError, and always rolls back in a
finally. `hermes doctor` and `hermes sessions repair --check-only` surface
the corruption and `repair_state_db_schema` heals it via the FTS rebuild
strategy (verified with a real stale-segid fixture).

Refs #100227
Reported-by: #100227

hermes_state_repair.py

@@ -788,7 +788,11 @@ def _db_opens_cleanly(db_path: Path) -> Optional[str]:
                         # image is malformed") while reads of the FTS5 table itself parse fine.
                         return f"fts5 read probe failed on {fts_table}: {exc}"
             # FTS write probe: drive a row through the messages_fts* triggers in a transaction that is always
-            # rolled back.
+            # rolled back. The trigger INSERT alone only buffers the row in FTS5's in-memory segment; the
+            # ``flush`` command writes that segment to ``<fts>_idx``/``_data`` exactly as a committed append
+            # would, so a stale ``_idx`` row at the next segid (IntegrityError "constraint failed", the #100227
+            # class: integrity_check and MATCH both clean, every real append fails) is hit here rather than
+            # by the user's next message.
             probe_session_id = f"_hermes_fts_health_probe_{time.time_ns()}"
             try:
                 conn.execute("BEGIN IMMEDIATE")
@@ -796,16 +800,24 @@ def _db_opens_cleanly(db_path: Path) -> Optional[str]:
                              (probe_session_id, "_health_probe", time.time()))
                 conn.execute("INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
                              (probe_session_id, "user", "_fts_health_probe", time.time()))
-                conn.execute("ROLLBACK")
-            except sqlite3.OperationalError as exc:
-                with contextlib.suppress(sqlite3.Error):
-                    conn.execute("ROLLBACK")
+                for fts_table in _FTS_TABLES:
+                    try:
+                        conn.execute(f"INSERT INTO {fts_table}({fts_table}) VALUES('flush')")
+                    except sqlite3.OperationalError as exc:
+                        if not (SessionDB._is_fts5_unavailable_error(exc) or _schema_not_built(exc)):
+                            raise
+            except sqlite3.DatabaseError as exc:
+                # IntegrityError is a DatabaseError sibling of OperationalError, not a child: catching only the
+                # latter let the trigram-segment collision report "healthy".
                 # Missing messages/sessions tables = brand new file mid-init, not corruption. "no such tokenizer":
                 # this process lacks the cjk extension the DB's index needs — capability gap; a tokenizer-less
                 # SessionDB drops the triggers itself.
                 if _schema_not_built(exc) or "no such tokenizer: cjk_unicode61" in str(exc).lower():
                     return None
-                return str(exc)
+                return f"fts5 write probe failed: {exc}"
+            finally:
+                with contextlib.suppress(sqlite3.Error):
+                    conn.execute("ROLLBACK")
             return None
     except sqlite3.DatabaseError as exc:
         return str(exc)

tests/test_state_db_fts_segment_collision_probe.py

@@ -0,0 +1,76 @@
+"""Segment-dependent FTS5 trigram corruption (#100227).
+
+A stale ``messages_fts_trigram_idx`` row sitting at the segid FTS5 allocates next makes every
+committed append fail with ``IntegrityError: constraint failed`` while ``PRAGMA integrity_check``,
+the FTS5 ``integrity-check`` command and ``MATCH`` all report healthy. The write probe must
+report it (and the repair must heal it) without any mocked detector.
+"""
+import sqlite3
+import time
+import uuid
+from pathlib import Path
+
+import pytest
+
+from hermes_state import SessionDB
+from hermes_state_repair import _db_opens_cleanly, repair_state_db_schema
+
+
+def _build_db_with_trigram(db_path: Path) -> str:
+    db = SessionDB(db_path=db_path)
+    if not db._trigram_available:
+        db.close()
+        pytest.skip("trigram tokenizer unavailable in this SQLite build")
+    sid = db.create_session(session_id=str(uuid.uuid4()), source="cli")
+    for i in range(60):
+        db.append_message(sid, role="user", content=f"quick brown fox {i} lorem ipsum dolor {i * 7}")
+    db.close()
+    return sid
+
+
+def _plant_stale_trigram_segment(db_path: Path) -> None:
+    """Leave an index row at the next free segid: the shape an aborted segment write leaves behind."""
+    conn = sqlite3.connect(str(db_path), isolation_level=None)
+    used = {r[0] for r in conn.execute("SELECT segid FROM messages_fts_trigram_idx")}
+    stale = next(s for s in range(1, 1 << 20) if s not in used)
+    conn.execute("INSERT INTO messages_fts_trigram_idx(segid, term, pgno) VALUES (?, X'', 2)", (stale,))
+    conn.close()
+
+
+def _real_append_fails(db_path: Path, sid: str) -> bool:
+    conn = sqlite3.connect(str(db_path), isolation_level=None)
+    try:
+        conn.execute("INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
+                     (sid, "user", "zebra yak xylophone wombat", time.time()))
+        return False
+    except sqlite3.IntegrityError:
+        return True
+    finally:
+        conn.close()
+
+
+def test_write_probe_reports_segment_collision_that_integrity_check_misses(tmp_path):
+    db_path = tmp_path / "state.db"
+    sid = _build_db_with_trigram(db_path)
+    _plant_stale_trigram_segment(db_path)
+
+    assert sqlite3.connect(str(db_path)).execute("PRAGMA integrity_check").fetchall() == [("ok",)]
+    assert _real_append_fails(db_path, sid), "fixture must break real appends"
+
+    reason = _db_opens_cleanly(db_path)
+    assert reason is not None and "constraint failed" in reason
+    # The probe rolls back: it must not have added rows or moved the FTS state.
+    assert sqlite3.connect(str(db_path)).execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 1
+
+
+def test_repair_heals_segment_collision_and_restores_appends(tmp_path):
+    db_path = tmp_path / "state.db"
+    sid = _build_db_with_trigram(db_path)
+    _plant_stale_trigram_segment(db_path)
+
+    report = repair_state_db_schema(db_path, backup=False)
+    assert report.get("repaired"), report
+    assert _db_opens_cleanly(db_path) is None
+    assert not _real_append_fails(db_path, sid)
+    with SessionDB(db_path=db_path) as db:
+        assert db._conn.execute("SELECT COUNT(*) FROM messages WHERE session_id = ?", (sid,)).fetchone()[0] == 61