fix(core): bound fleet census lock waits (#1336)

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [SemVer](https://semver.org/) starting from v3.1.2.
 - **Advisory offline linear BEAM throughput benchmark (#901).** A fixed synthetic workload on a temporary database reports warm per-call latency, sequential throughput, raw samples and environment/revision metadata as JSON, with compatible same-machine baseline comparisons and a documented retention/interpretation policy. No runtime recall changes or CI timing threshold.
 
 ### Fixed
+- **Fleet census bounds SQLite lock waits for read-only candidate banks (#1336).** Each locked file previously consumed SQLite's five-second default busy timeout. Census reads now wait up to 50 milliseconds per candidate; unreadable files remain in the report and keep the cross-bank bound unmeasured. Discovery and bank contents are unchanged.
 - **Opt-in input caps now cover local embeddings as well as the API (Refs #1153).** Local document and query calls apply the existing character and UTF-8 byte caps to their prefixed text before fastembed runs. Both caps remain disabled by default, and query-cache entries remain keyed by effective cap and prefix. Enhanced recall snapshots both effective input caps for its cache key and embedding calls, with generation 18 invalidating results from the current Main predecessor, generation 17. Changing caps does not rebuild stored vectors; reindex when applying the new policy to existing documents. This limits input size only; it does not establish an ONNX memory ceiling or close the reported arena issue.
 
 - **Polyphonic graph recall retains the strongest eligible traversal to a memory (#1303).** When multiple entity seeds reach the same memory, a later shorter path can replace an earlier weaker one. Scoped recall checks gist/fact seeds against a visible, unsuperseded and unexpired backing row satisfying the caller's date, source and topic filters before traversal; unscoped callers keep their prior behavior. The graph voice still emits each traversed memory once and excludes seed IDs. Enhanced recall cache version 17 invalidates results from the actual Main predecessor, generation 16.

mnemosyne/core/fleet_census.py

@@ -186,6 +186,11 @@ def _norm_pair(a: str, b: str) -> tuple:
     return (a, b) if a <= b else (b, a)
 
 
+# A foreign SQLite lock must not consume the session-end sleep budget.
+# An unreadable candidate is reported and makes the measured bound fail closed.
+_CENSUS_READ_TIMEOUT = 0.05
+
+
 def _read_bank(bank: Path) -> Optional[Dict[str, Any]]:
     """Read one bank's conflict rows.
 
@@ -195,7 +200,9 @@ def _read_bank(bank: Path) -> Optional[Dict[str, Any]]:
     of aborting the census. The connection is opened ``mode=ro`` — a
     read-only URI that SQLite refuses to write through.
     """
-    connection = sqlite3.connect(f"{bank.as_uri()}?mode=ro", uri=True)
+    connection = sqlite3.connect(
+        f"{bank.as_uri()}?mode=ro", uri=True, timeout=_CENSUS_READ_TIMEOUT
+    )
     try:
         if not connection.execute(
             "SELECT 1 FROM sqlite_master WHERE type='table' AND name='conflicts'"

tests/test_fleet_conflict_census.py

@@ -19,6 +19,7 @@
 import hashlib
 import re
 import sqlite3
+import time
 from datetime import datetime, timedelta
 from pathlib import Path
 
@@ -412,6 +413,98 @@ def _schema_and_counts(path: Path):
 
 
 class TestReadOnly:
+    def test_census_read_connection_has_short_timeout_and_rejects_writes(
+        self, fleet_env, monkeypatch
+    ):
+        """Pin the per-file budget without relying on scheduler timing."""
+        bank = _make_bank(fleet_env / "bank.db", [_TWIN + (None,)])
+        before = _file_digest(bank)
+        connect = sqlite3.connect
+        observed = []
+
+        def inspect_connection(*args, **kwargs):
+            conn = connect(*args, **kwargs)
+            try:
+                observed.append(conn.execute("PRAGMA busy_timeout").fetchone()[0])
+                with pytest.raises(sqlite3.OperationalError, match="readonly"):
+                    conn.execute("DELETE FROM conflicts")
+            except BaseException:
+                conn.close()
+                raise
+            return conn
+
+        monkeypatch.setattr(fleet_census.sqlite3, "connect", inspect_connection)
+        report = fleet_census.census(banks=[bank])
+        assert observed == [50]
+        assert report["banks_scanned"] == 1
+        assert report["unreadable"] == []
+        assert report["bound_holds"] is True
+        assert _file_digest(bank) == before
+
+    def test_foreign_exclusive_locks_do_not_delay_the_fleet(self, fleet_env):
+        """Browser databases must not consume the session-end census budget."""
+        bank = _make_bank(fleet_env / "custom" / "bank.db", [_TWIN + (None,)])
+        browser = fleet_env / "browser-chrome-profile"
+        foreign = [browser / f"profile-{i}" / f"cache-{i}.db" for i in range(3)]
+        locks = []
+        for path in foreign:
+            path.parent.mkdir(parents=True)
+            conn = sqlite3.connect(path)
+            conn.execute("PRAGMA journal_mode=DELETE")
+            conn.execute("CREATE TABLE foreign_data (value TEXT)")
+            conn.commit()
+            conn.execute("BEGIN EXCLUSIVE")
+            locks.append(conn)
+        before = {path: _file_digest(path) for path in [bank, *foreign]}
+        try:
+            assert set(foreign).issubset(fleet_census.discover_banks(fleet_env))
+            start = time.monotonic()
+            report = fleet_census.census()
+            elapsed = time.monotonic() - start
+        finally:
+            for conn in locks:
+                conn.rollback()
+                conn.close()
+
+        assert elapsed < 1.0, elapsed
+        assert report["banks_scanned"] == 1
+        assert len(report["unreadable"]) == 3
+        assert {entry["path"] for entry in report["unreadable"]} == {
+            str(path) for path in foreign
+        }
+        assert report["bound_holds"] is False
+        assert fleet_census.census()["bound_holds"] is True
+        assert {path: _file_digest(path) for path in [bank, *foreign]} == before
+
+    def test_locked_real_bank_fails_closed_then_recovers(self, fleet_env):
+        first = _make_bank(fleet_env / "home-a" / "bank.db", [_TWIN + (None,)])
+        second = _make_bank(
+            fleet_env / "unusual-location" / "other.db",
+            [(_TWIN[1], _TWIN[0], None)],
+        )
+        before = {path: _file_digest(path) for path in [first, second]}
+        lock = sqlite3.connect(second)
+        lock.execute("PRAGMA journal_mode=DELETE")
+        lock.execute("BEGIN EXCLUSIVE")
+        try:
+            start = time.monotonic()
+            partial = fleet_census.census()
+            elapsed = time.monotonic() - start
+        finally:
+            lock.rollback()
+            lock.close()
+
+        assert elapsed < 1.0, elapsed
+        assert partial["banks_scanned"] == 1
+        assert partial["unreadable"][0]["path"] == str(second)
+        assert partial["home_home_twin_count"] == 0
+        assert partial["bound_holds"] is False
+        complete = fleet_census.census()
+        assert complete["unreadable"] == []
+        assert complete["banks_scanned"] == 2
+        assert complete["home_home_twin_count"] == 1
+        assert complete["bound_holds"] is False
+        assert {path: _file_digest(path) for path in [first, second]} == before
 
     def test_census_writes_nothing_to_the_banks_it_reads(self, fleet_env):
         _build_fleet(fleet_env, {
@@ -704,4 +797,4 @@ def test_anything_else_stays_on(self, value, monkeypatch):
 
     def test_unset_means_on(self, monkeypatch):
         monkeypatch.delenv(fleet_census.CENSUS_ENV, raising=False)
-        assert fleet_census.census_enabled() is True
+        assert fleet_census.census_enabled() is True