fix(index): close SQLite connections and preserve busy rebuild owners

planfile/core/sqlite_index.py

@@ -5,7 +5,8 @@
 import json
 import sqlite3
 import time
-from collections.abc import Iterable
+from collections.abc import Iterable, Iterator
+from contextlib import closing, contextmanager
 from itertools import islice
 from pathlib import Path
 from typing import Any
@@ -18,6 +19,17 @@
 _SOURCE_INDEXED_AT_NS_KEY = "source_indexed_at_ns"
 
 
+def _is_sqlite_contention(error: sqlite3.DatabaseError) -> bool:
+    code = getattr(error, "sqlite_errorcode", None)
+    if code is not None:
+        return isinstance(code, int) and code & 0xFF in (5, 6)
+    # Python 3.10 lacks sqlite_errorcode; do not classify arbitrary messages.
+    message = str(error).lower()
+    return message in {
+        "database is locked", "database table is locked", "database schema is locked"
+    } or message.startswith(("database table is locked: ", "database schema is locked: "))
+
+
 class SQLiteTicketIndex:
     """Indexed projection of YAML/JSONL sources.
 
@@ -35,20 +47,31 @@ def serialize_signature(signature: tuple) -> str:
     def _connect(self) -> sqlite3.Connection:
         self.path.parent.mkdir(parents=True, exist_ok=True)
         connection = sqlite3.connect(self.path, timeout=10)
-        connection.row_factory = sqlite3.Row
-        journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0]
-        if str(journal_mode).lower() != "wal":
-            connection.execute("PRAGMA journal_mode=WAL")
-        connection.execute("PRAGMA synchronous=NORMAL")
-        connection.execute("PRAGMA foreign_keys=ON")
-        initialized = connection.execute(
-            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='meta'"
-        ).fetchone()
-        if initialized is None:
-            self._initialize(connection)
-        self._ensure_routes(connection)
+        try:
+            connection.row_factory = sqlite3.Row
+            journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0]
+            if str(journal_mode).lower() != "wal":
+                connection.execute("PRAGMA journal_mode=WAL")
+            connection.execute("PRAGMA synchronous=NORMAL")
+            connection.execute("PRAGMA foreign_keys=ON")
+            initialized = connection.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='meta'"
+            ).fetchone()
+            if initialized is None:
+                self._initialize(connection)
+            self._ensure_routes(connection)
+        except BaseException:
+            connection.close()
+            raise
         return connection
 
+    @contextmanager
+    def _connection(self) -> Iterator[sqlite3.Connection]:
+        # SQLite's own context manager commits/rolls back but does not close.
+        with closing(self._connect()) as connection:
+            with connection:
+                yield connection
+
     @staticmethod
     def _initialize(connection: sqlite3.Connection) -> None:
         connection.executescript(
@@ -204,7 +227,7 @@ def _route_page(self, where: str, values: list, limit: int,
         if (not isinstance(after, tuple) or len(after) != 2
                 or type(after[0]) is not int or not isinstance(after[1], str)):
             raise ValueError("ticket_index_cursor_invalid")
-        with self._connect() as connection:
+        with self._connection() as connection:
             rows = connection.execute(
                 f"SELECT id, position FROM ticket_routes WHERE {where} "
                 "AND (position, id) > (?, ?) ORDER BY position, id LIMIT ?",
@@ -220,7 +243,7 @@ def is_current(self, signature: tuple) -> bool:
             return False
         expected = self.serialize_signature(signature)
         try:
-            with self._connect() as connection:
+            with self._connection() as connection:
                 row = connection.execute(
                     "SELECT value FROM meta WHERE key='source_signature'"
                 ).fetchone()
@@ -272,7 +295,7 @@ def rebuild(
             raise ValueError("ticket_index_rebuild_batch_size_invalid")
         rows = iter(records)
         count = 0
-        with self._connect() as connection:
+        with self._connection() as connection:
             connection.execute("BEGIN IMMEDIATE")
             connection.execute("DELETE FROM dependencies")
             connection.execute("DELETE FROM tickets")
@@ -303,7 +326,7 @@ def rebuild(
         return count
 
     def get_ticket(self, ticket_id: str) -> dict[str, Any] | None:
-        with self._connect() as connection:
+        with self._connection() as connection:
             row = connection.execute(
                 "SELECT ticket_json FROM tickets WHERE id=?",
                 (ticket_id,),
@@ -320,7 +343,7 @@ def apply(
         """Apply a known-complete store mutation and advance its source signature."""
         rows = list(upserts)
         deleted_ids = list(deletes)
-        with self._connect() as connection:
+        with self._connection() as connection:
             connection.execute("BEGIN IMMEDIATE")
             if deleted_ids:
                 connection.executemany(
@@ -415,7 +438,7 @@ def has_fresh_snapshot(self, max_age_seconds: float) -> bool:
         if max_age_seconds < 0 or not self.path.exists():
             return False
         try:
-            with self._connect() as connection:
+            with self._connection() as connection:
                 rows = connection.execute(
                     "SELECT key, value FROM meta WHERE key IN ('source_signature', ?)",
                     (_SOURCE_INDEXED_AT_NS_KEY,),
@@ -449,7 +472,7 @@ def list_summaries(
             conditions.append(f"{key}=?")
             parameters.append(str(getattr(value, "value", value)))
         where = f" WHERE {' AND '.join(conditions)}" if conditions else ""
-        with self._connect() as connection:
+        with self._connection() as connection:
             total = int(
                 connection.execute(
                     f"SELECT COUNT(*) AS count FROM tickets{where}",
@@ -488,7 +511,7 @@ def list_payloads(
             conditions.append(f"{key}=?")
             parameters.append(str(getattr(value, "value", value)))
         where = f" WHERE {' AND '.join(conditions)}" if conditions else ""
-        with self._connect() as connection:
+        with self._connection() as connection:
             total = int(
                 connection.execute(
                     f"SELECT COUNT(*) AS count FROM tickets{where}",
@@ -527,7 +550,7 @@ def render_payloads(
             conditions.append(f"{key}=?")
             parameters.append(str(getattr(value, "value", value)))
         where = f" WHERE {' AND '.join(conditions)}" if conditions else ""
-        with self._connect() as connection:
+        with self._connection() as connection:
             total = int(
                 connection.execute(
                     f"SELECT COUNT(*) AS count FROM tickets{where}",
@@ -582,7 +605,7 @@ def render_operational_payloads(
         elif offset:
             page_sql += " LIMIT -1 OFFSET ?"
             page_parameters.append(offset)
-        with self._connect() as connection:
+        with self._connection() as connection:
             total = int(
                 connection.execute(
                     f"SELECT COUNT(*) AS count FROM tickets{where}",
@@ -647,7 +670,7 @@ def payload_page_metrics(
             conditions.append(f"{key}=?")
             parameters.append(str(getattr(value, "value", value)))
         where = f" WHERE {' AND '.join(conditions)}" if conditions else ""
-        with self._connect() as connection:
+        with self._connection() as connection:
             total = int(
                 connection.execute(
                     f"SELECT COUNT(*) AS count FROM tickets{where}",
@@ -683,7 +706,7 @@ def status(self, signature: tuple | None = None) -> dict[str, Any]:
                 "path": str(self.path),
             }
         try:
-            with self._connect() as connection:
+            with self._connection() as connection:
                 count = int(
                     connection.execute("SELECT COUNT(*) AS count FROM tickets").fetchone()[
                         "count"

planfile/core/store.py

@@ -2046,17 +2046,9 @@ def _finish_index_mutation(
             # All callers have already committed authoritative files. Only
             # BUSY/LOCKED in this disposable projection may be deferred; other
             # errors and failures before the source write still propagate.
-            code = getattr(exc, "sqlite_errorcode", None)
-            if code is not None:
-                contention = code & 0xFF in (5, 6)  # SQLITE_BUSY / SQLITE_LOCKED
-            else:
-                # Python 3.10 lacks sqlite_errorcode. Match SQLite's own
-                # messages exactly (or its table/schema-qualified variants).
-                message = str(exc).lower()
-                contention = message in {
-                    "database is locked", "database table is locked", "database schema is locked"
-                } or message.startswith(("database table is locked: ", "database schema is locked: "))
-            if not contention:
+            from planfile.core.sqlite_index import _is_sqlite_contention
+
+            if not _is_sqlite_contention(exc):
                 raise
             from planfile.core.fastio import _atomic_write_text
 
@@ -2119,6 +2111,11 @@ def _rebuild_ticket_index_unlocked(self, index, *, force: bool) -> dict:
 
                 if not isinstance(exc, sqlite3.DatabaseError):
                     raise
+                from planfile.core.sqlite_index import _is_sqlite_contention
+
+                if _is_sqlite_contention(exc):
+                    self._ticket_index_rebuild_deferred_until = time.monotonic() + 5.0
+                    raise TicketIndexContentionError("ticket_index_rebuild_busy") from exc
                 index.reset()
                 count = index.rebuild(self._ticket_index_records(), signature_before)
             signature_after = self._ticket_index_signature()

project/TICKETS.md

@@ -161,4 +161,5 @@ This file indexes governance tickets without taking ownership of
 | **ticket-257** | [`README.md`](./ticket-257/README.md) | - | - | - | - | - |
 | **ticket-258** | [`README.md`](./ticket-258/README.md) | - | - | - | - | - |
 | **ticket-259** | [`README.md`](./ticket-259/README.md) | - | - | - | - | - |
+| **ticket-260** | [`README.md`](./ticket-260/README.md) | - | - | - | - | - |
 <!-- AUTO:TICKET_INDEX:END -->

project/ticket-260/README.md

@@ -0,0 +1,23 @@
+# Ticket 260: Close SQLite index connections and preserve busy rebuild owners
+
+- **ID**: ticket-260
+- **Owner**: agent:codex
+- **Status**: IN_PROGRESS
+- **Workflow state**: PUBLICATION
+- **Created**: 2026-10-30
+
+## Goal and scope
+
+SESSION_EXECUTION_AUTHORIZATION: continue the user's autonomous repair, testing and independent publication work. Existing owner semcod/planfile#313, parent PLF-20347 / Willman430. Prior isolated diagnostics proved100status reads retained201additional descriptors while cyclic GC was disabled, and a forced rebuild replaced the index inode while an independent SQLite writer was alive. Fix the component lifetime and contention classification, preserving real corruption recovery and durable authoritative state. Current internal health/summary reads work while archived operational reads exceed deadlines; source changes alone do not prove native service recovery.
+
+## Acceptance criteria
+
+- [x] AC-01: Internal index operations close connections deterministically, including transaction and setup failures, without requiring cyclic GC.
+- [x] AC-02: BUSY/LOCKED rebuilds preserve the active database and owner; independent-process and corruption-recovery regressions pass.
+- [ ] AC-03: Affected checks, protected exact-head CI and independent publication pass; an installed immutable candidate has separate canary evidence and no implicit live rollout.
+
+## Delivery boundary
+
+Never remove a live lock/index, kill a writer, forge a historical publication receipt, hotpatch the live package or equate source tests with deployment. Preserve existing313/316/317 and Validator593 evidence. Allocation uses the installed ID-only allocator; delivery uses the adopted v5 planner and canonical relative worktree.
+
+Independent Validator review correction: the repeated-read regression now retains all 200 real SQLite connections and proves each is closed on every platform. Linux additionally measures `/proc/self/fd`; no regression is skipped. The same 78 affected tests pass with the updated test, and pinned Ruff 0.15.11 reports no findings. Source behavior and the installed artifact bytes are unchanged; fresh exact-head CI and independent approval remain required.

project/ticket-260/intent.json

@@ -0,0 +1,92 @@
+{
+  "schema": "new-project.intent/v3",
+  "ticket": "ticket-260",
+  "summary": "Close SQLite index connections and preserve busy rebuild owners",
+  "workstream": "application",
+  "classification": {
+    "kind": "BUG",
+    "priority": "P1",
+    "origin": "health"
+  },
+  "allowedPaths": [
+    "project/ticket-260/**",
+    "TODO.md",
+    "project/TICKETS.md",
+    "planfile/core/sqlite_index.py",
+    "planfile/core/store.py",
+    "tests/test_sqlite_ticket_index.py"
+  ],
+  "forbiddenPaths": [
+    "project/ticket-*/user-*.md"
+  ],
+  "stacks": [],
+  "dependsOn": [],
+  "conflictsWith": [],
+  "integrationTicket": null,
+  "delivery": {
+    "acceptedBaseSha": "9a221c92dc42a104c2347ad5b507128ded696648",
+    "targetBranch": "main",
+    "outcome": "Close every internal SQLite index connection deterministically on success, failure and connection setup exceptions; treat SQLite BUSY/LOCKED rebuild failures as bounded contention without resetting another writer's index, while retaining genuine corruption recovery.",
+    "nonGoals": [
+      "No live database/lock deletion, live package hotpatch, owner termination, lease takeover or automatic reservation release.",
+      "No claim that a source fix or isolated canary is native runtime activation; prior publication exceptions remain separate."
+    ],
+    "complexity": "S",
+    "estimatedMinutes": 45,
+    "budgets": {
+      "maxImplementationFiles": 3,
+      "maxAffectedComponents": 1,
+      "maxPublicInterfaceChanges": 0,
+      "maxRuntimeDependencies": 0
+    },
+    "architecture": {
+      "status": "accepted",
+      "decision": "Give the existing SQLiteTicketIndex internal operation context deterministic transaction exit and close, close failed setup connections, and share exact SQLite contention classification with the store rebuild/post-commit paths.",
+      "components": [
+        {
+          "name": "sqlite-index-projection",
+          "paths": [
+            "planfile/core/sqlite_index.py",
+            "planfile/core/store.py",
+            "tests/test_sqlite_ticket_index.py"
+          ]
+        }
+      ],
+      "responsibilityChanges": false,
+      "interfaceChanges": [],
+      "dataChanges": [],
+      "ui": {
+        "impact": "none",
+        "states": [],
+        "evidence": []
+      },
+      "rollback": "Revert the source commit; native artifacts/activation require separate exact-target operator receipts."
+    },
+    "runtimeDependencies": [],
+    "validation": [
+      {
+        "criterion": "AC-01",
+        "commands": [
+          "testless scan -s planfile tests/test_sqlite_ticket_index.py"
+        ],
+        "evidence": "Repeated reads release connections without cyclic GC; successful transaction/rollback/setup failures release resources."
+      },
+      {
+        "criterion": "AC-02",
+        "commands": [
+          "testless scan -s planfile tests/test_sqlite_ticket_index.py"
+        ],
+        "evidence": "Independent SQLite writer retains PID/inode/data on BUSY/LOCKED; extended SQLite codes and Python3.10 fallback handled; real corruption recovery retained."
+      },
+      {
+        "criterion": "AC-03",
+        "commands": [
+          "project/governance-check.sh",
+          "protected OneDev exact-head/current-base verification and independent Validator publication",
+          "isolated installed-wheel canary on live native base image without live volume/network"
+        ],
+        "evidence": "Genuine source publication and content-bound runtime candidate; native rollout remains separate."
+      }
+    ]
+  }
+}