fix(api): bound indexed queue read amplification

planfile/api/server.py

@@ -375,6 +375,8 @@ def _validate_completion_receipt(receipt: dict[str, Any] | None, ticket_id: str)
 
 
 def _ticket_snapshot_signature(pf, sprint: str) -> tuple:
+    if pf.store.ticket_index_enabled():
+        return pf.store._cached_ticket_index_signature()
     return pf.store.sprint_signature(sprint), pf.store._evidence_revision()
 
 
@@ -536,17 +538,25 @@ def _bounded_stale_index_response(
                 offset=offset,
                 limit=limit,
             )
+        elif view == "operational":
+            body, total, count = index.render_operational_payloads(
+                sprint=sprint,
+                filters=filters,
+                offset=offset,
+                limit=limit,
+            )
+            payload = None
         else:
             payload, total = index.list_payloads(
                 sprint=sprint,
                 filters=filters,
                 offset=offset,
                 limit=limit,
             )
-            payload = [_ticket_operational_payload(ticket) for ticket in payload]
     except (json.JSONDecodeError, OSError, sqlite3.DatabaseError):
         return None
-    body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    if view != "operational":
+        body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
     return Response(
         content=body,
         media_type="application/json",
@@ -555,7 +565,7 @@ def _bounded_stale_index_response(
             "X-Planfile-View": view,
             "X-Planfile-Index-State": "stale",
             "X-Total-Count": str(total),
-            "X-Result-Count": str(len(payload)),
+            "X-Result-Count": str(count if view == "operational" else len(payload)),
         },
     )
 
@@ -608,6 +618,7 @@ def _ticket_list_response(
                             offset=offset,
                             limit=limit,
                             repair=False,
+                            signature=signature,
                         )
                         count = len(payload)
                     elif view == "full":
@@ -617,6 +628,7 @@ def _ticket_list_response(
                             offset=offset,
                             limit=limit,
                             repair=False,
+                            signature=signature,
                         )
                         response_limit = _ticket_list_response_byte_limit()
                         if estimated_bytes > response_limit:
@@ -648,6 +660,16 @@ def _ticket_list_response(
                             offset=offset,
                             limit=limit,
                             repair=False,
+                            signature=signature,
+                        )
+                    elif view == "operational":
+                        body, total, count = pf.store.indexed_ticket_operational_response(
+                            sprint=sprint,
+                            filters=filters,
+                            offset=offset,
+                            limit=limit,
+                            repair=False,
+                            signature=signature,
                         )
                     else:
                         payload, total = pf.store.indexed_ticket_payloads(
@@ -656,9 +678,8 @@ def _ticket_list_response(
                             offset=offset,
                             limit=limit,
                             repair=False,
+                            signature=signature,
                         )
-                        if view == "operational":
-                            payload = [_ticket_operational_payload(ticket) for ticket in payload]
                         count = len(payload)
                 except TicketIndexContentionError:
                     if (
@@ -709,8 +730,10 @@ def _ticket_list_response(
                 ]
             if body is None:
                 body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
-            # Do not retain a response assembled across a concurrent file change.
-            if signature == _ticket_snapshot_signature(pf, sprint):
+            # SQLite reads are transactionally coherent and version-keyed. The
+            # durable-file fallback still needs a second source check because it
+            # can span multiple independently replaced YAML files.
+            if pf.store.ticket_index_enabled() or signature == _ticket_snapshot_signature(pf, sprint):
                 _cache_ticket_list_response(
                     query_key=query_key,
                     versioned_key=key,

planfile/core/sqlite_index.py

@@ -406,6 +406,79 @@ def render_payloads(
             body.extend(b"]")
         return bytes(body), total, count
 
+    def render_operational_payloads(
+        self,
+        *,
+        sprint: str,
+        filters: dict[str, Any],
+        offset: int,
+        limit: int | None,
+    ) -> tuple[bytes, int, int]:
+        """Project and render queue payloads inside SQLite's JSON runtime."""
+        conditions = []
+        parameters: list[Any] = []
+        if sprint != "all":
+            conditions.append("sprint=?")
+            parameters.append(sprint)
+        for key in ("status", "priority", "source"):
+            value = filters.get(key)
+            if value is None:
+                continue
+            conditions.append(f"{key}=?")
+            parameters.append(str(getattr(value, "value", value)))
+        where = f" WHERE {' AND '.join(conditions)}" if conditions else ""
+        page_sql = f"SELECT position, ticket_json FROM tickets{where} ORDER BY position"
+        page_parameters = list(parameters)
+        if limit is not None:
+            page_sql += " LIMIT ? OFFSET ?"
+            page_parameters.extend((limit, offset))
+        elif offset:
+            page_sql += " LIMIT -1 OFFSET ?"
+            page_parameters.append(offset)
+        with self._connect() as connection:
+            total = int(
+                connection.execute(
+                    f"SELECT COUNT(*) AS count FROM tickets{where}",
+                    parameters,
+                ).fetchone()["count"]
+            )
+            cursor = connection.execute(
+                """
+                WITH page AS (
+                    """ + page_sql + """
+                ), stripped AS (
+                    SELECT position, json_remove(
+                        ticket_json,
+                        '$.history', '$.dsl', '$.file', '$.files',
+                        '$.integration', '$.llm_hints', '$.sync',
+                        '$.source.context', '$.outputs.notes', '$.outputs.artifacts'
+                    ) AS payload
+                    FROM page
+                ), without_empty_source AS (
+                    SELECT position,
+                        CASE WHEN json_extract(payload, '$.source') = '{}'
+                            THEN json_remove(payload, '$.source') ELSE payload END AS payload
+                    FROM stripped
+                ), normalized AS (
+                    SELECT position,
+                        CASE WHEN json_extract(payload, '$.outputs') = '{}'
+                            THEN json_remove(payload, '$.outputs') ELSE payload END AS payload
+                    FROM without_empty_source
+                )
+                SELECT payload FROM normalized ORDER BY position
+                """,
+                page_parameters,
+            )
+            body = bytearray(b"[")
+            count = 0
+            for row in cursor:
+                if count:
+                    body.extend(b",")
+                body.extend(row["payload"].encode("utf-8"))
+                count += 1
+            body.extend(b"]")
+        return bytes(body), total, count
+
     def payload_page_metrics(
         self,
         *,

planfile/core/store.py

@@ -10,6 +10,7 @@
 from datetime import UTC, datetime, timedelta
 from enum import Enum
 from pathlib import Path
+from threading import RLock
 
 import yaml
 from pydantic import BaseModel
@@ -67,6 +68,8 @@ def __init__(self, directory: str | Path):
         self._ticket_index_path = self.base_dir / "index" / "tickets.sqlite3"
         self._ticket_index_rebuild_lock_path = self.base_dir / "index" / ".rebuild.lock"
         self._ticket_index_rebuild_deferred_until = 0.0
+        self._ticket_index_signature_cache: tuple[float, tuple, tuple] | None = None
+        self._ticket_index_signature_cache_lock = RLock()
         self._history_locations_path = self.base_dir / "index" / "history-locations.yaml"
 
     def _storage_config(self) -> dict:
@@ -494,8 +497,10 @@ def mutation_lock(self):
                 fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
             except ImportError:
                 pass
+            self._invalidate_ticket_index_signature_cache()
             yield
         finally:
+            self._invalidate_ticket_index_signature_cache()
             try:
                 import fcntl
 
@@ -1181,6 +1186,53 @@ def _sqlite_ticket_index(self):
     def _ticket_index_signature(self) -> tuple:
         return self.sprint_signature("all"), self._evidence_revision()
 
+    @staticmethod
+    def _ticket_index_signature_cache_seconds() -> float:
+        """Bound source-signature reuse during a concurrent API read burst."""
+        try:
+            configured = float(
+                os.environ.get("PLANFILE_TICKET_INDEX_SIGNATURE_CACHE_SECONDS", "0.25")
+            )
+        except (TypeError, ValueError):
+            configured = 0.25
+        return max(0.0, min(configured, 3.0))
+
+    def _cached_ticket_index_signature(self) -> tuple:
+        """Coalesce expensive filesystem scans without hiding local mutations.
+
+        A large store may contain thousands of evidence files. FastAPI executes
+        synchronous reads in parallel, so scanning that tree independently in
+        every worker amplifies one dashboard refresh into CPU and I/O
+        starvation. The mutation lock invalidates this process-local snapshot
+        before and after every write; the short TTL bounds observation of
+        changes made by another process.
+        """
+        now = time.monotonic()
+        probe = self._ticket_index_signature_cache_probe()
+        with self._ticket_index_signature_cache_lock:
+            cached = self._ticket_index_signature_cache
+            if (
+                cached is not None
+                and cached[2] == probe
+                and now - cached[0] <= self._ticket_index_signature_cache_seconds()
+            ):
+                return cached[1]
+            signature = self._ticket_index_signature()
+            self._ticket_index_signature_cache = (time.monotonic(), signature, probe)
+            return signature
+
+    def _ticket_index_signature_cache_probe(self) -> tuple:
+        """Cheaply detect the active-queue edits that must never wait for TTL."""
+        try:
+            evidence_dir_mtime = self._evidence_dir.stat().st_mtime_ns
+        except OSError:
+            evidence_dir_mtime = -1
+        return self.sprint_signature("current"), evidence_dir_mtime
+
+    def _invalidate_ticket_index_signature_cache(self) -> None:
+        with self._ticket_index_signature_cache_lock:
+            self._ticket_index_signature_cache = None
+
     def _ticket_index_records(self):
         from planfile.core.fastio import read_yaml_fast
 
@@ -1298,7 +1350,7 @@ def ensure_ticket_index(self, *, force: bool = False) -> dict:
                 return index.status(signature) | {"rebuilt": False}
             return self._rebuild_ticket_index_unlocked(index, force=force)
 
-    def require_current_ticket_index(self) -> dict:
+    def require_current_ticket_index(self, *, signature: tuple | None = None) -> dict:
         """Validate the projection without rebuilding it on the caller's thread.
 
         Latency-sensitive API reads use this guard.  A stale projection is a
@@ -1308,7 +1360,8 @@ def require_current_ticket_index(self) -> dict:
         index = self._sqlite_ticket_index()
         if not self.ticket_index_enabled():
             raise TicketIndexContentionError("ticket_index_disabled")
-        signature = self._ticket_index_signature()
+        if signature is None:
+            signature = self._cached_ticket_index_signature()
         if not index.is_current(signature):
             raise TicketIndexContentionError("ticket_index_stale")
         return index.status(signature) | {"rebuilt": False}
@@ -1400,8 +1453,12 @@ def indexed_ticket_summaries(
         offset: int,
         limit: int | None,
         repair: bool = True,
+        signature: tuple | None = None,
     ) -> tuple[list[dict], int]:
-        (self.ensure_ticket_index if repair else self.require_current_ticket_index)()
+        if repair:
+            self.ensure_ticket_index()
+        else:
+            self.require_current_ticket_index(signature=signature)
         return self._sqlite_ticket_index().list_summaries(
             sprint=sprint,
             filters=filters,
@@ -1417,9 +1474,13 @@ def indexed_ticket_payloads(
         offset: int,
         limit: int | None,
         repair: bool = True,
+        signature: tuple | None = None,
     ) -> tuple[list[dict], int]:
         """Read full ticket JSON directly from the disposable SQLite projection."""
-        (self.ensure_ticket_index if repair else self.require_current_ticket_index)()
+        if repair:
+            self.ensure_ticket_index()
+        else:
+            self.require_current_ticket_index(signature=signature)
         return self._sqlite_ticket_index().list_payloads(
             sprint=sprint,
             filters=filters,
@@ -1435,9 +1496,13 @@ def indexed_ticket_json_response(
         offset: int,
         limit: int | None,
         repair: bool = True,
+        signature: tuple | None = None,
     ) -> tuple[bytes, int, int]:
         """Render full ticket JSON from SQLite without a Python object graph."""
-        (self.ensure_ticket_index if repair else self.require_current_ticket_index)()
+        if repair:
+            self.ensure_ticket_index()
+        else:
+            self.require_current_ticket_index(signature=signature)
         return self._sqlite_ticket_index().render_payloads(
             sprint=sprint,
             filters=filters,
@@ -1453,16 +1518,42 @@ def indexed_ticket_json_metrics(
         offset: int,
         limit: int | None,
         repair: bool = True,
+        signature: tuple | None = None,
     ) -> tuple[int, int, int]:
         """Measure a full-ticket page before materializing its JSON body."""
-        (self.ensure_ticket_index if repair else self.require_current_ticket_index)()
+        if repair:
+            self.ensure_ticket_index()
+        else:
+            self.require_current_ticket_index(signature=signature)
         return self._sqlite_ticket_index().payload_page_metrics(
             sprint=sprint,
             filters=filters,
             offset=offset,
             limit=limit,
         )
 
+    def indexed_ticket_operational_response(
+        self,
+        *,
+        sprint: str,
+        filters: dict,
+        offset: int,
+        limit: int | None,
+        repair: bool = True,
+        signature: tuple | None = None,
+    ) -> tuple[bytes, int, int]:
+        """Render a bounded operational page without Python JSON object graphs."""
+        if repair:
+            self.ensure_ticket_index()
+        else:
+            self.require_current_ticket_index(signature=signature)
+        return self._sqlite_ticket_index().render_operational_payloads(
+            sprint=sprint,
+            filters=filters,
+            offset=offset,
+            limit=limit,
+        )
+
     def migrate_to_sharded_yaml(
         self,
         *,