fix(OMN-19968): a read through the SQLite adapter no longer takes a write lock

contracts/OMN-19968.yaml

@@ -18,6 +18,9 @@ evidence_requirements:
   - kind: "tests"
     description: "The llm-cost writer, its SQLite schema and the forward migration use the shared usage-source vocabulary, so its rows match on SQLite and on a Postgres infra has migrated."
     command: "uv run pytest tests/unit/test_omn19968_usage_source_vocabulary.py -q"
+  - kind: "tests"
+    description: "A read through the SQLite adapter takes no write lock once the usage-source step is recorded."
+    command: "uv run pytest tests/unit/test_omn19968_usage_source_vocabulary.py -q -k 'no_write_lock or read_only or failed_step_record or stays_as_written'"
 emergency_bypass:
   enabled: false
   justification: ""
@@ -31,3 +34,11 @@ dod_evidence:
         check_value: "uv run pytest tests/unit/test_omn19968_usage_source_vocabulary.py -q"
         cwd: "${OMNI_HOME}/omnimarket"
     binds_ac: ["AC1"]
+  - id: "dod-omn19968-ac1-read-takes-no-write-lock"
+    description: "A read through the SQLite adapter takes no write lock once the usage-source step is recorded, a read-only store is read as it is, and the relabel and its record commit together."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/unit/test_omn19968_usage_source_vocabulary.py -q -k 'no_write_lock or read_only or failed_step_record or stays_as_written'"
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC1"]

src/omnimarket/projection/sqlite_database.py

@@ -22,6 +22,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import sqlite3
 from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime
@@ -35,6 +36,8 @@
     build_upsert_plan,
 )
 
+logger = logging.getLogger(__name__)
+
 _DEFAULT_EVIDENCE_DB_PATH = (
     Path.home() / ".omninode" / "delegation" / "delegation.sqlite"
 )
@@ -300,6 +303,19 @@
     ON llm_call_metrics (input_hash)
 """
 
+# One-time data steps on a local store, the SQLite counterpart of a forward
+# migration: each runs once, committed together with its row here. The table is
+# created by the first step that writes, never on a plain connect, so a
+# read-only store that predates it can still be opened.
+_STORE_STEPS_TABLE = "omnimarket_sqlite_store_steps"
+_STORE_STEPS_DDL = f"""
+CREATE TABLE IF NOT EXISTS {_STORE_STEPS_TABLE} (
+    step       TEXT PRIMARY KEY,
+    applied_at TEXT NOT NULL
+)
+"""
+_USAGE_SOURCE_VOCABULARY_STEP = "omn19968_usage_source_shared_vocabulary"
+
 # JSON-serialized columns: list/dict values are stored as TEXT JSON so the
 # sqlite row round-trips structurally for evidence queries.
 _JSON_COLUMNS = frozenset(
@@ -375,6 +391,7 @@ def _connect(self) -> sqlite3.Connection:
         conn.execute(_TENANT_INFERENCE_CREDENTIALS_DDL)
         conn.execute(_DELEGATION_ROUTING_TENANT_OVERLAY_DDL)
         conn.commit()
+        self._apply_usage_source_vocabulary_step(conn, self._db_path)
         return conn
 
     @staticmethod
@@ -425,16 +442,72 @@ def _reconcile_legacy_llm_call_metrics(cls, conn: sqlite3.Connection) -> None:
             conn.execute(
                 f"ALTER TABLE llm_call_metrics RENAME TO {_LEGACY_LLM_CALL_METRICS_TABLE}"
             )
-        elif "usage_source" in columns:
-            # OMN-19968: rows written before the shared vocabulary move onto it
-            # (EnumUsageSource; omnibase_infra migration 077). Idempotent: it
-            # touches only rows still holding a retired label.
+
+    @staticmethod
+    def _store_step_recorded(conn: sqlite3.Connection, step: str) -> bool:
+        if (
+            conn.execute(
+                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
+                (_STORE_STEPS_TABLE,),
+            ).fetchone()
+            is None
+        ):
+            return False
+        return (
+            conn.execute(
+                f"SELECT 1 FROM {_STORE_STEPS_TABLE} WHERE step = ?", (step,)
+            ).fetchone()
+            is not None
+        )
+
+    @classmethod
+    def _apply_usage_source_vocabulary_step(
+        cls, conn: sqlite3.Connection, db_path: Path
+    ) -> None:
+        """OMN-19968: move rows written before the shared vocabulary onto it, once.
+
+        The SQLite counterpart of migration 0003 (EnumUsageSource; omnibase_infra
+        migration 077). It runs once per store, and the relabel and its record
+        commit together. A store that has recorded it opens with reads only.
+        Every connection goes through here, reads included, and an UPDATE takes a
+        write lock even when it matches nothing.
+
+        The decision reads the store's step record, never llm_call_metrics rows:
+        this module is shared by nodes that do not own that table.
+        """
+        if cls._store_step_recorded(conn, _USAGE_SOURCE_VOCABULARY_STEP):
+            return
+        if "usage_source" not in cls._existing_columns(conn, "llm_call_metrics"):
+            return
+        try:
+            conn.execute("BEGIN IMMEDIATE")
+            conn.execute(_STORE_STEPS_DDL)
             conn.execute(
                 "UPDATE llm_call_metrics SET usage_source = CASE usage_source "
                 "WHEN 'API' THEN 'measured' WHEN 'ESTIMATED' THEN 'estimated' "
                 "WHEN 'MISSING' THEN 'unknown' ELSE usage_source END "
                 "WHERE usage_source IN ('API', 'ESTIMATED', 'MISSING')"
             )
+            conn.execute(
+                f"INSERT OR IGNORE INTO {_STORE_STEPS_TABLE} (step, applied_at) "
+                "VALUES (?, ?)",
+                (_USAGE_SOURCE_VOCABULARY_STEP, datetime.now(UTC).isoformat()),
+            )
+            conn.commit()
+        except sqlite3.Error as exc:
+            conn.rollback()
+            # A store this process cannot write is still read, with the labels
+            # it holds. Anything else, a busy store included, is a real failure.
+            if not (
+                isinstance(exc, sqlite3.OperationalError)
+                and (exc.sqlite_errorcode & 0xFF) == sqlite3.SQLITE_READONLY
+            ):
+                raise
+            logger.warning(
+                "%s is read-only and has not moved llm_call_metrics.usage_source "
+                "onto the shared vocabulary; reading it as it is",
+                db_path,
+            )
 
     @staticmethod
     def _existing_columns(conn: sqlite3.Connection, table: str) -> set[str]: