fix(store): retry the open-time schema window on SQLITE_SCHEMA

CHANGELOG/v4.md

@@ -47,6 +47,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Opening a store while another process opened the same store could raise `database schema has changed` ([#1310](https://github.com/robotrocketscience/aelfrice/issues/1310)).** SQLite raises `SQLITE_SCHEMA` when the schema cookie moves between a statement's prepare and its step, which is exactly what two processes running the open-time `CREATE TABLE IF NOT EXISTS` battery against one `.git/aelfrice/memory.db` do to each other. `busy_timeout=5000` does not cover it — that pragma retries `database is locked`, a different error — so the constructor failed outright and took a required CI check down at random. The whole open-time window (the stale-`ingest_log` probe, `_SCHEMA`, `_MIGRATIONS`, the post-migration indexes, the `schema_meta` seed reads and writes, and the local scope-id resolve) now runs under one bounded re-prepare retry rather than only the DDL loop the traceback named: the reads are exposed to the same race, and a per-call-site patch would leave gaps as the constructor grows. The window is idempotent by construction (`IF NOT EXISTS` / `OR IGNORE` / marker-gated), so re-running it is a no-op. Only `schema has changed` is retried — every other `OperationalError` still propagates, and a persistently changing schema fails after a bounded number of attempts instead of spinning. Contrary to the issue's reasoning, the `_MIGRATIONS` catch three lines below did **not** already tolerate this class: it admits only `duplicate column name` and `no such column`, both of which are re-raised for anything else.
 - **`[relationship_detector]` was half-honoured at ingest: the flag took effect, the thresholds did not ([#1299](https://github.com/robotrocketscience/aelfrice/issues/1299)).** `ingest.py` resolved `auto_detect` from `.aelfrice.toml` and then called `write_semantic_edges(store, new_belief_ids=...)` with no threshold arguments, so `jaccard_min` / `confidence_min` / `max_candidate_pairs` reached the read-only `aelf doctor --relationships` audits and were silently ignored on the one path that actually mutates the graph. Adjacent keys in the same section, opposite reach, no trace on stderr — and the asymmetry ran the risky direction, with the audit tunable and the writer pinned at the module defaults. Ingest now threads the resolved config through. The three keys are parsed from the **same** `.aelfrice.toml` read that resolves `auto_detect` (new `resolve_ingest_relationship_config`), so the per-turn config-probe count is unchanged — measured 11 probes with no config file and 4 with one four directories up, before and after — rather than adding a second filesystem walk to a hot path ([#1289](https://github.com/robotrocketscience/aelfrice/issues/1289)/[#1298](https://github.com/robotrocketscience/aelfrice/issues/1298)). That measurement is env-unset; the env var is the other half of the precedence and it moves the count, so it is stated separately. `AELFRICE_AUTO_RELATIONSHIPS=0` decides the question without needing any threshold, and the flag-only resolver short-circuited on env before touching the filesystem — so that install paid **0** probes per turn and still does, the resolver returning before the walk rather than reading a config whose only consumer will not run. With the var set truthy the walk is real and new (0 -> 11 on a deep tree): the thresholds are then actually used, which is the point of the fix, so that one is a cost and not a regression. Precedence for `auto_detect` is unchanged (env > TOML > default-off), and default-off means a fresh install is byte-identical. `residual_overlap_min` and `max_edges_per_belief` still have no TOML key at all; `docs/user/CONFIG.md` now documents the section with per-key reach so which is which is readable.
 - **PHILOSOPHY and the write-log memo claimed `edges` are a materialized projection of the log; they never have been ([#1283](https://github.com/robotrocketscience/aelfrice/issues/1283)).** All six `derive()` return paths emit `edges=[]`, so `ingest_log.derived_edge_ids` is NULL on **every** row (0 of 139,592 on the development store), and every real edge is written outside the log by `ingest.py`, `temporal_spine.py` and the relationship / contradiction detectors. The substrate claim therefore overreached for the graph: a replay from empty cannot reconstruct the L3 BFS graph, the temporal spine or the `CONTRADICTS` substrate, and `replay.py` reports edge divergence only in the bucket documented as *"never promoted into `has_drift`"*. Both documents now state the ratified contract (edges are **log-derived**, recompute keyed on `(created_at, ingest_log ULID)`, operator ruling 2026-08-01) separately from the shipped state, and name what is still missing. The key is the log's ULID rather than anything read off the belief table because the writer actually orders by `(created_at, rowid)` and `rowid` is implicit here — VACUUM may renumber it; measured, the ULID key reproduces **93.7%** of the live `TEMPORAL_NEXT` set against **7.4%** for a belief-table key (`benchmarks/spine_order_provenance.py`). Docs only — no code, no defaults, and the recompute itself is not built.
 - **LIMITATIONS claimed a residual exposure-as-evidence path that has not existed since #1162 ([#1267](https://github.com/robotrocketscience/aelfrice/issues/1267)).** The sharp-edges entry said retrieval "still enqueues each surfaced belief" and that `aelf sweep-feedback` "applies a small alpha bump (default +0.05)", citing [#1091](https://github.com/robotrocketscience/aelfrice/issues/1091) as having only *flagged* the sweep for audit-only treatment. All three were stale: the sweeper has written nothing since [#1162](https://github.com/robotrocketscience/aelfrice/issues/1162) (it classifies what it *would* have applied and returns `mutated=False`), and the enqueue inside `retrieve()` is gated on `AELFRICE_IMPLICIT_FEEDBACK_ENQUEUE`, default off. A reader auditing where their posteriors come from was pointed at a mutation path that no longer fires. New `benchmarks/posterior_channel_audit.py` drives all three `apply_feedback` routes against a fresh store and fails non-zero if any default moves, so the entry cannot go stale silently again.

src/aelfrice/store.py

@@ -20,7 +20,15 @@
 import sqlite3
 from contextlib import contextmanager
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Callable, Final, Iterable, Iterator, Sequence
+from typing import (
+    TYPE_CHECKING,
+    Callable,
+    Final,
+    Iterable,
+    Iterator,
+    Sequence,
+    TypeVar,
+)
 
 if TYPE_CHECKING:
     # #1126 belief categories. Imported lazily at runtime inside the
@@ -1125,6 +1133,61 @@ def _drop_stale_ingest_log(conn: sqlite3.Connection) -> None:
     conn.execute("DROP TABLE ingest_log")
 
 
+_SCHEMA_CHANGED_MSG: Final[str] = "schema has changed"
+_SCHEMA_RETRY_ATTEMPTS: Final[int] = 3
+
+_RetryT = TypeVar("_RetryT")
+
+
+def _retry_on_schema_change(
+    op: Callable[[], _RetryT], *, attempts: int = _SCHEMA_RETRY_ATTEMPTS
+) -> _RetryT:
+    """Run `op`, retrying only `SQLITE_SCHEMA` ("schema has changed").
+
+    #1310. SQLite raises `OperationalError: database schema has changed`
+    when the schema cookie moves between a statement's prepare and its
+    step — i.e. another connection committed DDL in between. Two
+    processes opening the same store both run the open-time
+    `CREATE TABLE IF NOT EXISTS` battery, so this is reachable on every
+    multi-worktree open. `busy_timeout` does not cover it: that pragma
+    retries `database is locked`, a different error.
+
+    Re-preparing against the new cookie is the entire fix, so `op` must
+    be idempotent. Every other `OperationalError` propagates unchanged —
+    a malformed statement must stay loud rather than be retried into
+    silence. `attempts` is bounded so a schema that keeps changing
+    fails instead of spinning forever.
+    """
+    for i in range(attempts):
+        try:
+            return op()
+        except sqlite3.OperationalError as e:
+            if _SCHEMA_CHANGED_MSG not in str(e) or i == attempts - 1:
+                raise
+    # Unreachable: the loop either returns or raises on the last pass.
+    raise AssertionError("attempts must be >= 1")
+
+
+def _execute_reprepare(
+    conn: sqlite3.Connection,
+    stmt: str,
+    *,
+    attempts: int = _SCHEMA_RETRY_ATTEMPTS,
+) -> sqlite3.Cursor:
+    """Execute one parameterless statement, re-preparing on SQLITE_SCHEMA.
+
+    #1310. Statement-level sibling of `_retry_on_schema_change`, used by
+    the open-time DDL loops so the common case re-runs one `CREATE TABLE
+    IF NOT EXISTS` rather than restarting the whole battery. The battery
+    is wrapped as well — reads and parameterised writes are exposed to
+    the same race, and a per-call-site patch would leave gaps as the
+    constructor grows.
+    """
+    return _retry_on_schema_change(
+        lambda: conn.execute(stmt), attempts=attempts
+    )
+
+
 def _ingest_row_to_dict(row: sqlite3.Row) -> dict[str, object]:
     """Decode an `ingest_log` sqlite row into a Python dict.
 
@@ -1223,68 +1286,18 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None:
         # aelfrice/memory.db. Per the v1.1.0 #89 concurrency tests.
         self._conn.execute("PRAGMA busy_timeout=5000")
         self._conn.execute("PRAGMA foreign_keys=ON")
-        _drop_stale_ingest_log(self._conn)
-        for stmt in _SCHEMA:
-            self._conn.execute(stmt)
-        for stmt in _MIGRATIONS:
-            try:
-                self._conn.execute(stmt)
-            except sqlite3.OperationalError as e:
-                # Idempotency catches:
-                #   - "duplicate column name: X" — ADD COLUMN already
-                #     present (fresh v1.2 DB or prior migration pass).
-                #   - "no such column: X" — DROP COLUMN already done
-                #     (fresh DB that never had the column, or prior
-                #     migration pass).
-                msg = str(e)
-                if (
-                    "duplicate column name" not in msg
-                    and "no such column" not in msg
-                ):
-                    raise
-        for stmt in _POST_MIGRATION_INDEXES:
-            self._conn.execute(stmt)
-        # #1135: one-shot. Ran unguarded on every open pre-v4.2; the
-        # marker matches the other schema_meta-gated passes. Rides the
-        # single open commit below.
-        # #1135: seed the durable mutation counter. Read-first: an
-        # unconditional INSERT OR IGNORE would take the write lock on
-        # every open even when the row exists, blocking behind any
-        # concurrent writer's open transaction. Rides the single open
-        # commit below. Two connections racing the first-ever seed both
-        # pass the SELECT; OR IGNORE makes the second insert a no-op.
-        seeded = self._conn.execute(
-            "SELECT 1 FROM schema_meta WHERE key = ?",
-            (SCHEMA_META_STORE_GENERATION,),
-        ).fetchone()
-        if seeded is None:
-            self._conn.execute(
-                "INSERT OR IGNORE INTO schema_meta (key, value) "
-                "VALUES (?, '0')",
-                (SCHEMA_META_STORE_GENERATION,),
-            )
-        marker = self._conn.execute(
-            "SELECT value FROM schema_meta WHERE key = ?",
-            (SCHEMA_META_ORIGIN_BACKFILL,),
-        ).fetchone()
-        if marker is None:
-            for stmt in _BACKFILL_STATEMENTS:
-                self._conn.execute(stmt)
-            self._conn.execute(
-                "INSERT OR REPLACE INTO schema_meta (key, value) "
-                "VALUES (?, ?)",
-                (
-                    SCHEMA_META_ORIGIN_BACKFILL,
-                    datetime.now(timezone.utc).isoformat(),
-                ),
-            )
-        self._commit()
         self._invalidation_callbacks: list[Callable[[], None]] = []
-        # v1.5.0 #204 federation forward-compat. Resolve (or
-        # generate) the local scope id BEFORE any belief/edge
-        # write path runs — write hooks consume `_local_scope_id`
-        # to bump the version-vector counter.
-        self._local_scope_id: str = self._resolve_local_scope_id()
+        # #1310: the whole open-time schema window runs under one
+        # schema-cookie retry, not just the DDL loops. Every statement
+        # in it — the `_drop_stale_ingest_log` reads, the schema_meta
+        # queries, the backfill writes, and `_resolve_local_scope_id` —
+        # is exposed to a concurrent opener committing DDL between
+        # prepare and step. The window is idempotent by construction
+        # (IF NOT EXISTS / OR IGNORE / marker-gated), so re-running it
+        # is safe. See `_retry_on_schema_change`.
+        self._local_scope_id: str = _retry_on_schema_change(
+            self._apply_open_schema
+        )
         # #1161: every one-shot below runs through `_run_guarded_migration`
         # so a raising pass cannot make the store unopenable. Ordering is
         # unchanged and the guard does not alter the success path — see
@@ -1356,6 +1369,80 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None:
         self._peer_handles: dict[str, sqlite3.Connection] = {}
         self._peer_deps_loaded: bool = False
 
+    def _apply_open_schema(self) -> str:
+        """Run the open-time DDL + seed + backfill window; return scope id.
+
+        Extracted from `__init__` (#1310) so the whole window can be
+        re-run as a unit when a concurrent opener moves the schema
+        cookie — see `_retry_on_schema_change`, which is the only
+        intended caller. Idempotent: every statement here is
+        `IF NOT EXISTS`, `OR IGNORE`/`OR REPLACE`, or gated on a
+        `schema_meta` marker, so a second pass is a no-op.
+        """
+        _drop_stale_ingest_log(self._conn)
+        for stmt in _SCHEMA:
+            _execute_reprepare(self._conn, stmt)
+        for stmt in _MIGRATIONS:
+            try:
+                _execute_reprepare(self._conn, stmt)
+            except sqlite3.OperationalError as e:
+                # Idempotency catches:
+                #   - "duplicate column name: X" — ADD COLUMN already
+                #     present (fresh v1.2 DB or prior migration pass).
+                #   - "no such column: X" — DROP COLUMN already done
+                #     (fresh DB that never had the column, or prior
+                #     migration pass).
+                # It does NOT catch "database schema has changed" — that
+                # class is handled by the retry above/around, not here.
+                msg = str(e)
+                if (
+                    "duplicate column name" not in msg
+                    and "no such column" not in msg
+                ):
+                    raise
+        for stmt in _POST_MIGRATION_INDEXES:
+            _execute_reprepare(self._conn, stmt)
+        # #1135: one-shot. Ran unguarded on every open pre-v4.2; the
+        # marker matches the other schema_meta-gated passes. Rides the
+        # single open commit below.
+        # #1135: seed the durable mutation counter. Read-first: an
+        # unconditional INSERT OR IGNORE would take the write lock on
+        # every open even when the row exists, blocking behind any
+        # concurrent writer's open transaction. Rides the single open
+        # commit below. Two connections racing the first-ever seed both
+        # pass the SELECT; OR IGNORE makes the second insert a no-op.
+        seeded = self._conn.execute(
+            "SELECT 1 FROM schema_meta WHERE key = ?",
+            (SCHEMA_META_STORE_GENERATION,),
+        ).fetchone()
+        if seeded is None:
+            self._conn.execute(
+                "INSERT OR IGNORE INTO schema_meta (key, value) "
+                "VALUES (?, '0')",
+                (SCHEMA_META_STORE_GENERATION,),
+            )
+        marker = self._conn.execute(
+            "SELECT value FROM schema_meta WHERE key = ?",
+            (SCHEMA_META_ORIGIN_BACKFILL,),
+        ).fetchone()
+        if marker is None:
+            for stmt in _BACKFILL_STATEMENTS:
+                self._conn.execute(stmt)
+            self._conn.execute(
+                "INSERT OR REPLACE INTO schema_meta (key, value) "
+                "VALUES (?, ?)",
+                (
+                    SCHEMA_META_ORIGIN_BACKFILL,
+                    datetime.now(timezone.utc).isoformat(),
+                ),
+            )
+        self._commit()
+        # v1.5.0 #204 federation forward-compat. Resolve (or generate)
+        # the local scope id BEFORE any belief/edge write path runs —
+        # write hooks consume `_local_scope_id` to bump the
+        # version-vector counter.
+        return self._resolve_local_scope_id()
+
     def close(self) -> None:
         for conn in self._peer_handles.values():
             try: