perf: gate-receipts and loop-session projections stop shipping fat rows

/api/snapshots ran 33-60s even after retention pruned the events table —
profiling each projection found query_gate_receipts (21s) and
query_loop_session (10s) fetching thousands of full payload rows to
Python to keep 100 (or ONE). Both now select newest-first with SQL
LIMIT / single-column reads: 0.12s and 0.07s against prod. Remaining
projections (watch_states 6s, daily_performance 5s, approval_queue 3.5s)
are the same class — later pass if the snapshot still feels slow.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

storage/projections.py

@@ -499,22 +499,36 @@ def query_gate_receipts(store, symbol: str | None = None, limit: int = 100,
     in-Python filter idiom rather than store.query_events' exact-match one)."""
     since, until = _hist_window(for_date)
     sym_upper = symbol.upper() if symbol else None
+    # newest-first LIMIT in SQL: the old limit=None fetch parsed EVERY receipt
+    # in the window (unbounded on live) to keep 100 (~20s of the snapshot
+    # build). Over-fetch when symbol-filtering in Python, then trim.
+    fetch = limit if sym_upper is None else max(limit * 20, 1000)
+    conds, params = ["event_type = 'entry_gate_evaluated'"], []
+    if since is not None:
+        conds.append("timestamp >= ?"); params.append(since)
+    if until is not None:
+        conds.append("timestamp < ?"); params.append(until)
+    params.append(int(fetch))
+    with store._lock:
+        raw = store.con.execute(
+            "SELECT timestamp, payload_json FROM events WHERE " + " AND ".join(conds)
+            + " ORDER BY timestamp DESC LIMIT ?", params).fetchall()
     rows: list[dict] = []
-    for event in store.query_events(event_type="entry_gate_evaluated", since=since,
-                                    until=until, limit=None):
-        payload = json.loads(event.get("payload_json", "{}"))
+    for ts, pj in raw:
+        payload = json.loads(pj or "{}")
         if sym_upper is not None and str(payload.get("symbol") or "").upper() != sym_upper:
             continue
         rows.append({
-            "timestamp": event.get("timestamp"),
+            "timestamp": str(ts),
             "symbol": payload.get("symbol"),
             "source": payload.get("source"),
             "passed": payload.get("passed"),
             "reject_reason": payload.get("reject_reason"),
             "gates": payload.get("gates") or {},
         })
-    rows.sort(key=lambda r: str(r.get("timestamp") or ""), reverse=True)
-    return rows[:limit]
+        if len(rows) >= limit:
+            break
+    return rows
 
 
 def _r_multiple(entry, stop, exit_price, side: str) -> float | None:
@@ -949,13 +963,15 @@ def query_loop_session(store) -> dict:
     bounded range scan with no per-session lookup."""
     from datetime import datetime, timedelta
     since = (datetime.now() - timedelta(minutes=15)).isoformat()
-    events = store.query_events(since=since, limit=None)
-    sid = None
-    for e in reversed(events):
-        cid = str(e.get("correlation_id") or "")
-        if cid.startswith("paper-"):
-            sid = cid
-            break
+    # one indexed row, correlation_id only — the old type-agnostic
+    # query_events(limit=None) shipped every fat payload of the last 15 min
+    # to Python just to read one column (~10s of the snapshot build)
+    with store._lock:
+        rows = store.con.execute(
+            "SELECT correlation_id FROM events WHERE timestamp >= ? "
+            "AND correlation_id LIKE 'paper-%' "
+            "ORDER BY timestamp DESC LIMIT 1", [since]).fetchall()
+    sid = str(rows[0][0]) if rows else None
     if not sid:
         return {"session_id": None, "session_date": None, "stale": False}
     parts = sid.split("-")                      # paper-YYYY-MM-DD-hash