Avoid archive scans in autonomous queue status reads

planfile/core/sqlite_index.py

@@ -59,12 +59,26 @@ def _connect(self) -> sqlite3.Connection:
             ).fetchone()
             if initialized is None:
                 self._initialize(connection)
+            self._ensure_status_index(connection)
             self._ensure_routes(connection)
         except BaseException:
             connection.close()
             raise
         return connection
 
+    @staticmethod
+    def _ensure_status_index(connection: sqlite3.Connection) -> None:
+        # Cross-sprint controllers filter by status, then paginate by position.
+        # A sprint-leading index cannot serve that query; scanning position can
+        # read the entire archive even when the requested status has no rows.
+        if connection.execute(
+            "SELECT 1 FROM sqlite_master WHERE type='index' AND name='tickets_status_position'"
+        ).fetchone():
+            return
+        connection.execute(
+            "CREATE INDEX IF NOT EXISTS tickets_status_position ON tickets(status, position)"
+        )
+
     @contextmanager
     def _connection(self) -> Iterator[sqlite3.Connection]:
         # SQLite's own context manager commits/rolls back but does not close.

project/ticket-262/README.md

@@ -0,0 +1,29 @@
+# Ticket 262: Index cross-sprint status reads for autonomous queue snapshots
+
+- **ID**: ticket-262
+- **Owner**: agent:codex
+- **Status**: IN_PROGRESS
+- **Workflow state**: PUBLICATION
+- **Created**: 2026-10-30
+
+## Goal and scope
+
+SESSION_EXECUTION_AUTHORIZATION: autonomously repair and publish the requested
+ecosystem throughput work. Current Control lifecycle reads fail as
+`planfile_unavailable` even when health and open-ticket queries succeed.
+Cross-sprint in-progress and blocked reads exceeded 6–10 seconds at limit one.
+The deployed SQLite query plan scans tickets using the position index;
+there is no index beginning with status. Add the missing ordered status index
+and migrate existing disposable projections without changing ticket records.
+
+## Acceptance criteria
+
+- [x] AC-01: New and existing projections search by status in ticket order without a whole-ticket scan; data and pagination remain unchanged.
+- [x] AC-02: Concurrent opens migrate idempotently, existing reads stay read-only after migration, and index regression checks pass.
+
+Production rollout requires exact-head independent publication and an observed
+runtime exercise. Tests or a source merge alone do not prove deployment.
+
+Validation: 82 focused SQLite index and connection-lifetime tests passed.
+Ruff and diff whitespace checks passed. Protected exact-head verification and
+independent publication remain required before runtime adoption.

project/ticket-262/intent.json

@@ -0,0 +1,80 @@
+{
+  "schema": "new-project.intent/v3",
+  "ticket": "ticket-262",
+  "summary": "Index cross-sprint status reads for autonomous queue snapshots",
+  "workstream": "application",
+  "classification": {
+    "kind": "SERVICE",
+    "priority": "P1",
+    "origin": "requested"
+  },
+  "allowedPaths": [
+    "project/ticket-262/**",
+    "TODO.md",
+    "project/TICKETS.md",
+    "planfile/core/sqlite_index.py",
+    "tests/test_sqlite_status_index.py"
+  ],
+  "forbiddenPaths": [
+    "project/ticket-*/user-*.md"
+  ],
+  "stacks": [],
+  "dependsOn": [],
+  "conflictsWith": [],
+  "integrationTicket": null,
+  "delivery": {
+    "targetBranch": "main",
+    "acceptedBaseSha": "57d085698cd2a92f6da7bf69a4e6a270571aec0e",
+    "outcome": "Status-filtered cross-sprint queue reads use an ordered SQLite index, including databases created before this change, without altering ticket data or admission decisions.",
+    "complexity": "S",
+    "estimatedMinutes": 45,
+    "budgets": {
+      "maxImplementationFiles": 2,
+      "maxAffectedComponents": 1,
+      "maxPublicInterfaceChanges": 0,
+      "maxRuntimeDependencies": 0
+    },
+    "architecture": {
+      "status": "accepted",
+      "decision": "Ensure one additive status-position index when opening the disposable ticket projection. Already upgraded databases keep the read path; no source-data rewrite or ticket-state change. Regression evidence checks query plans, bounded SQLite work, unchanged payload order and concurrent migration.",
+      "components": [
+        {
+          "name": "sqlite-ticket-projection",
+          "paths": [
+            "planfile/core/sqlite_index.py",
+            "tests/test_sqlite_status_index.py"
+          ]
+        }
+      ],
+      "interfaceChanges": [],
+      "responsibilityChanges": false,
+      "dataChanges": [],
+      "rollback": "Revert the bounded source change through independent publication. The additive index remains harmless and ticket sources are unchanged.",
+      "ui": {
+        "impact": "none",
+        "states": [],
+        "evidence": []
+      }
+    },
+    "runtimeDependencies": [],
+    "nonGoals": [
+      "No queue gate bypass, status change, execution grant, unbounded concurrency increase or runtime rewrite."
+    ],
+    "validation": [
+      {
+        "criterion": "AC-01",
+        "commands": [
+          "python -m pytest -q tests/test_sqlite_status_index.py"
+        ],
+        "evidence": "Fresh and existing databases use indexed ordered status reads and retain exact ticket payloads."
+      },
+      {
+        "criterion": "AC-02",
+        "commands": [
+          "python -m pytest -q tests/test_sqlite_ticket_index.py tests/test_sqlite_index_and_retry_perf.py"
+        ],
+        "evidence": "Connections close correctly, concurrent migration is idempotent and existing ticket index behavior is preserved."
+      }
+    ]
+  }
+}

tests/test_sqlite_status_index.py

@@ -0,0 +1,96 @@
+"""Ordered cross-sprint status reads must not scan the terminal archive."""
+from __future__ import annotations
+
+import json
+import sqlite3
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from planfile.core.sqlite_index import SQLiteTicketIndex
+
+
+def _records(count=1200):
+    for position in range(count):
+        status = "blocked" if position % 400 == 399 else "done"
+        ticket = {"id": f"PLF-{position + 1}", "status": status,
+                  "name": f"Ticket {position}", "history": ["not operational"]}
+        yield {"id": ticket["id"], "sprint": f"archive-{position % 10}",
+               "status": status, "priority": "high", "source": "local",
+               "queue": "default", "created_at": None, "updated_at": None,
+               "position": position, "ticket_json": json.dumps(ticket),
+               "summary_json": json.dumps(ticket), "blocked_by": []}
+
+
+def _legacy(index):
+    index.rebuild(_records(), signature=("unchanged",))
+    with sqlite3.connect(index.path) as connection:
+        connection.execute("DROP INDEX tickets_status_position")
+        return connection.execute(
+            "SELECT id, status, position, ticket_json FROM tickets ORDER BY position"
+        ).fetchall()
+
+
+@pytest.mark.parametrize("legacy", [False, True])
+def test_status_page_searches_index_without_sorting_or_archive_scan(tmp_path, legacy):
+    index = SQLiteTicketIndex(tmp_path / "tickets.sqlite3")
+    if legacy:
+        _legacy(index)
+    else:
+        index.rebuild(_records(), signature=("unchanged",))
+    with index._connection() as connection:
+        query = "SELECT position, ticket_json FROM tickets WHERE status=? ORDER BY position LIMIT 1"
+        plans = [row[3] for row in connection.execute("EXPLAIN QUERY PLAN " + query, ("blocked",))]
+        assert any("SEARCH tickets USING INDEX tickets_status_position" in plan for plan in plans)
+        assert not any("SCAN tickets" in plan or "TEMP B-TREE" in plan for plan in plans)
+        # SQLite VM work, rather than wall-clock timing, proves an absent status
+        # does not inspect every archived row on a loaded host.
+        steps = 0
+
+        def progress():
+            nonlocal steps
+            steps += 1
+            return int(steps > 10)
+
+        connection.set_progress_handler(progress, 100)
+        assert connection.execute(query, ("in_progress",)).fetchall() == []
+        assert connection.execute("SELECT COUNT(*) FROM tickets WHERE status=?",
+                                  ("in_progress",)).fetchone()[0] == 0
+        connection.set_progress_handler(None, 0)
+    body, total, count = index.render_operational_payloads(
+        sprint="all", filters={"status": "blocked"}, offset=1, limit=1)
+    assert total == 3 and count == 1
+    assert json.loads(body) == [{"id": "PLF-800", "status": "blocked", "name": "Ticket 799"}]
+
+
+def test_existing_database_migrates_once_and_preserves_payloads_and_signature(tmp_path):
+    index = SQLiteTicketIndex(tmp_path / "tickets.sqlite3")
+    before = _legacy(index)
+    assert index.is_current(("unchanged",))
+    with index._connection() as connection:
+        after = connection.execute(
+            "SELECT id, status, position, ticket_json FROM tickets ORDER BY position"
+        ).fetchall()
+        assert [tuple(row) for row in after] == before
+    # Subsequent opens must not perform schema writes or require a writer lock.
+    with sqlite3.connect(index.path) as writer:
+        writer.execute("BEGIN IMMEDIATE")
+        assert index.is_current(("unchanged",))
+
+
+def test_concurrent_existing_database_opens_share_additive_migration(tmp_path):
+    index = SQLiteTicketIndex(tmp_path / "tickets.sqlite3")
+    before = _legacy(index)
+
+    def observe(_):
+        local = SQLiteTicketIndex(index.path)
+        body, total, count = local.render_operational_payloads(
+            sprint="all", filters={"status": "blocked"}, offset=0, limit=1)
+        return json.loads(body)[0]["id"], total, count
+
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        assert list(pool.map(observe, range(8))) == [("PLF-400", 3, 1)] * 8
+    with index._connection() as connection:
+        assert connection.execute("SELECT COUNT(*) FROM tickets").fetchone()[0] == len(before)
+        assert len([row for row in connection.execute("PRAGMA index_list(tickets)")
+                    if row[1] == "tickets_status_position"]) == 1