perf(workbook): select stale refresh cells with bounded SQL

.github/workflows/ci.yml

@@ -62,6 +62,7 @@ jobs:
           tests/test_entity_identity_pg.py
           tests/test_pg_tenancy_rls.py
           tests/test_lead_paging_pg.py
+          tests/test_workbook_refresh_stale_sql_pg.py
 
   frontend:
     name: Frontend lint, tests and build

apps/api/core/config.py

@@ -163,6 +163,12 @@ class Settings(BaseSettings):
     # Cloud deployments may populate this; comma-separated when set via env.
     AUTOMATIONS_WEBHOOK_DOMAIN_ALLOWLIST: list[str] = []
 
+    # ── Living workbook refresh ──
+    # Rows per keyset page when a refresh selects stale enrichment cells in SQL.
+    # Bounds memory/query size per round trip; every page is still scanned, so
+    # the selected cells do not depend on this value.
+    WORKBOOK_REFRESH_SCAN_BATCH_SIZE: int = 1000
+
     # ── Signal scanner (legacy hiring/JobSpy scan → unified ORM signals) ──
     # The scanner now iterates EVERY workspace and fans out provider I/O per
     # lead, so per-workspace and global caps keep a multi-tenant scan polite.

apps/api/services/workbook/refresh.py

@@ -52,37 +52,93 @@ def _interval_minutes(policy: dict) -> Optional[int]:
     return INTERVAL_MINUTES.get(str(interval).lower())
 
 
-def _stale_row_columns(db, workbook_id: str, enrichment_cols: List[dict], ttl_map: dict) -> dict[str, list[str]]:
-    """Exact enrichment cells missing, failed or older than their field's TTL."""
-    rows = db.query(WorkbookRow).filter(WorkbookRow.workbook_id == workbook_id).all()
-    if not rows:
-        return {}
-    # Legacy overlay keys follow the execution key (linked lead id or row id).
-    overlays = db.query(WorkbookEnrichment).filter(
-        WorkbookEnrichment.workbook_id == workbook_id
-    ).all()
-    seen = {(o.lead_id, o.column_id): o for o in overlays}
+def _stale_row_columns(db, workbook_id: str, enrichment_cols: List[dict], ttl_map: dict,
+                       batch_size: Optional[int] = None) -> dict[str, list[str]]:
+    """Exact enrichment cells missing, failed or older than their field's TTL.
+
+    Selected in SQL, keyset-paginated over ``workbook_rows.id`` so a refresh
+    never loads the whole workbook (or its overlays) into Python. Each page
+    cross-joins the page's rows with the enrichment columns and LEFT JOINs the
+    overlay on the execution key (linked lead id, else row id); only stale
+    (row, column) pairs come back. A cell is stale when its overlay is missing,
+    its status is anything but ``complete`` (including NULL), its
+    ``updated_at`` is NULL, or ``updated_at`` is older than
+    ``now - staleness_ttl_days[target_field or id]`` (default 30 days).
+
+    Every page is scanned, so the result is independent of ``batch_size``.
+    Rows are returned in id order with columns in config order. Tenancy: the
+    explicit ``workbook_id`` filter plus RLS on both tables, as before.
+    """
+    from sqlalchemy import DateTime, Integer, String, and_, func, literal, or_, select, true, union_all
+
+    from apps.api.core.config import settings
 
-    stale = {}
+    # Column ids are NOT NULL on overlays; de-dup preserves config order (the
+    # caller de-dups the per-row lists anyway).
+    cutoffs: dict[str, datetime] = {}
     now = _now()
-    for row in rows:
-        row_id = row.id
-        lead_id = row.lead_id or row_id
-        for col in enrichment_cols:
-            cid = col.get("id")
-            ttl_days = ttl_map.get(col.get("target_field") or cid, DEFAULT_STALENESS_DAYS)
-            o = seen.get((lead_id, cid))
-            if o is None or o.status != "complete":
-                stale.setdefault(str(row_id), []).append(cid)
-                continue
-            upd = o.updated_at
-            if upd is None:
-                stale.setdefault(str(row_id), []).append(cid)
-                continue
-            if upd.tzinfo is None:
-                upd = upd.replace(tzinfo=timezone.utc)
-            if upd < now - timedelta(days=ttl_days):
-                stale.setdefault(str(row_id), []).append(cid)
+    for col in enrichment_cols:
+        cid = col.get("id")
+        if not cid or cid in cutoffs:
+            continue
+        ttl_days = ttl_map.get(col.get("target_field") or cid, DEFAULT_STALENESS_DAYS)
+        # Overlay timestamps are naive UTC (DateTime without tz on both
+        # dialects); compare naive UTC to avoid session-timezone coercion on
+        # PostgreSQL and offset-suffixed strings on SQLite.
+        cutoffs[cid] = (now - timedelta(days=ttl_days)).astimezone(timezone.utc).replace(tzinfo=None)
+    if not cutoffs:
+        return {}
+
+    columns = union_all(*(
+        select(literal(cid, String).label("cid"),
+               literal(ordinal, Integer).label("ord"),
+               literal(cutoff, DateTime()).label("cutoff"))
+        for ordinal, (cid, cutoff) in enumerate(cutoffs.items())
+    )).subquery("refresh_cols")
+
+    size = max(1, int(batch_size or settings.WORKBOOK_REFRESH_SCAN_BATCH_SIZE))
+    R, E = WorkbookRow, WorkbookEnrichment
+    stale: dict[str, list[str]] = {}
+    after = None
+    while True:
+        # Keyset page bounds: the next `size` row ids after the previous page.
+        page_q = select(R.id).where(R.workbook_id == workbook_id)
+        if after is not None:
+            page_q = page_q.where(R.id > after)
+        page = page_q.order_by(R.id).limit(size).subquery("refresh_page")
+        hi, count = db.execute(select(func.max(page.c.id), func.count())).one()
+        if not count:
+            break
+
+        in_page = [R.workbook_id == workbook_id, R.id <= hi]
+        if after is not None:
+            in_page.append(R.id > after)
+        overlay_on = and_(
+            E.workbook_id == workbook_id,
+            E.lead_id == func.coalesce(R.lead_id, R.id),
+            E.column_id == columns.c.cid,
+        )
+        rows = db.execute(
+            select(R.id, columns.c.cid)
+            .select_from(R)
+            .join(columns, true())
+            .outerjoin(E, overlay_on)
+            .where(*in_page)
+            .where(or_(
+                E.id.is_(None),
+                E.status.is_(None),
+                E.status != "complete",
+                E.updated_at.is_(None),
+                E.updated_at < columns.c.cutoff,
+            ))
+            .order_by(R.id, columns.c.ord)
+        ).all()
+        for row_id, cid in rows:
+            stale.setdefault(str(row_id), []).append(cid)
+
+        if count < size:
+            break
+        after = hi
     return stale
 
 

docs/adr/20261012-refresh-dependent-and-new-row-cells.md

@@ -38,6 +38,25 @@ writer during sourcing are left alone (covered by
 `test_concurrently_inserted_row_is_not_treated_as_sourced`). No extra row-ID
 queries are issued.
 
+## Addendum (2026-10-13): stale selection in bounded SQL (part of #33)
+
+Stale-cell selection no longer loads every workbook row and overlay into
+Python. `_stale_row_columns` keyset-paginates `workbook_rows.id` (page size
+`WORKBOOK_REFRESH_SCAN_BATCH_SIZE`, default 1000) and, per page, cross-joins
+the rows with the enrichment column ids (each carrying its own cutoff,
+`now - staleness_ttl_days[target_field or id]`, default 30 days) and LEFT JOINs
+`workbook_enrichments` on `(workbook_id, COALESCE(lead_id, id), column_id)`.
+Only stale pairs are returned: overlay missing, status not `complete` (NULL
+included), `updated_at` NULL, or `updated_at` older than the cutoff. Cutoffs
+are bound as naive UTC to match the naive-UTC `updated_at` column on both
+SQLite and PostgreSQL. Every page is scanned, so the selected cells are
+identical to the previous Python loop regardless of page size; it is a bound
+on per-query work, not a cap on refresh scope. Column entries without an id
+(malformed config) are skipped rather than reported as a `None` column.
+Parity against the previous implementation is covered by
+`tests/test_workbook_refresh_stale_sql.py` and, under FORCE RLS on
+PostgreSQL, `tests/test_workbook_refresh_stale_sql_pg.py`.
+
 ## Validation
 
 Python 3.12.3 and the repository's frozen uv dependencies: both new regressions

tests/refresh_stale_support.py

@@ -0,0 +1,115 @@
+"""Shared fixtures for the SQL stale-cell selection parity tests (SQLite + PG).
+
+``legacy_stale_row_columns`` is the pre-#33 Python implementation, kept
+verbatim as the oracle: the SQL selection must return exactly what it did.
+"""
+
+import random
+from datetime import timedelta, timezone
+
+from apps.api.services.workbook.models import Workbook, WorkbookEnrichment, WorkbookRow
+from apps.api.services.workbook.refresh import DEFAULT_STALENESS_DAYS
+
+COLUMNS = [
+    {"id": "email", "type": "input"},
+    {"id": "summary", "type": "ai_formula", "prompt": "s {email}"},       # ttl 7
+    {"id": "pitch", "type": "ai_formula", "prompt": "p {email}"},         # ttl 30
+    {"id": "domain_col", "type": "enrichment", "target_field": "domain"},  # ttl via target_field: 2
+    {"id": "no_ttl", "type": "waterfall"},                                 # default 30
+    {"id": "quick", "type": "ai_formula", "prompt": "q {email}"},         # ttl 0
+]
+# "domain_col" also has its own entry: target_field must take precedence.
+TTL_MAP = {"summary": 7, "pitch": 30, "domain": 2, "quick": 0, "domain_col": 999}
+ENRICHMENT_COLS = [c for c in COLUMNS if c["type"] in ("enrichment", "waterfall", "ai_formula")]
+
+
+def legacy_stale_row_columns(db, workbook_id, enrichment_cols, ttl_map, now):
+    rows = db.query(WorkbookRow).filter(WorkbookRow.workbook_id == workbook_id).all()
+    if not rows:
+        return {}
+    overlays = db.query(WorkbookEnrichment).filter(
+        WorkbookEnrichment.workbook_id == workbook_id
+    ).all()
+    seen = {(o.lead_id, o.column_id): o for o in overlays}
+    stale = {}
+    for row in rows:
+        row_id = row.id
+        lead_id = row.lead_id or row_id
+        for col in enrichment_cols:
+            cid = col.get("id")
+            ttl_days = ttl_map.get(col.get("target_field") or cid, DEFAULT_STALENESS_DAYS)
+            o = seen.get((lead_id, cid))
+            if o is None or o.status != "complete":
+                stale.setdefault(str(row_id), []).append(cid)
+                continue
+            upd = o.updated_at
+            if upd is None:
+                stale.setdefault(str(row_id), []).append(cid)
+                continue
+            if upd.tzinfo is None:
+                upd = upd.replace(tzinfo=timezone.utc)
+            if upd < now - timedelta(days=ttl_days):
+                stale.setdefault(str(row_id), []).append(cid)
+    return stale
+
+
+def _age(now, days, hours=0):
+    return (now - timedelta(days=days, hours=hours)).astimezone(timezone.utc).replace(tzinfo=None)
+
+
+STATES = ["fresh", "stale", "missing", "error", "pending", "null_status",
+          "null_updated", "edge_fresh", "edge_stale"]
+
+
+def populate_random(db, ws, now, *, n_rows=40, seed=33, name="Stale SQL parity"):
+    """A workbook with a randomized mix of fresh/stale/missing/failed cells.
+
+    Also writes a sibling workbook whose complete, fresh overlays share the
+    same execution keys; they must never leak into the first workbook's
+    selection. Returns (workbook_id, states_seen).
+    """
+    rng = random.Random(seed)
+    wb = Workbook(name=name, workspace_id=ws, columns_config=COLUMNS,
+                  refresh_policy={"staleness_ttl_days": TTL_MAP})
+    other = Workbook(name=f"{name} sibling", workspace_id=ws, columns_config=COLUMNS)
+    db.add_all([wb, other])
+    db.flush()
+    seen_states = set()
+    for i in range(n_rows):
+        row = WorkbookRow(workbook_id=wb.id, workspace_id=ws, position=i,
+                          lead_id=(100000 + i) if rng.random() < 0.5 else None,
+                          data={"email": f"r{i}@example.test"}, enrichments={})
+        db.add(row)
+        db.flush()
+        key = row.lead_id or row.id
+        for col in ENRICHMENT_COLS:
+            db.add(WorkbookEnrichment(workbook_id=other.id, workspace_id=ws, lead_id=key,
+                                      column_id=col["id"], status="complete",
+                                      updated_at=_age(now, 0)))
+            ttl = TTL_MAP.get(col.get("target_field") or col["id"], DEFAULT_STALENESS_DAYS)
+            state = rng.choice(STATES)
+            seen_states.add(state)
+            if state == "missing":
+                continue
+            status, upd = "complete", _age(now, 0)
+            if state == "stale":
+                upd = _age(now, ttl + 5)
+            elif state == "error":
+                status = "error"
+            elif state == "pending":
+                status = "pending"
+            elif state == "edge_fresh":
+                upd = _age(now, ttl, hours=-1)
+            elif state == "edge_stale":
+                upd = _age(now, ttl, hours=1)
+            cell = WorkbookEnrichment(workbook_id=wb.id, workspace_id=ws, lead_id=key,
+                                      column_id=col["id"], status=status, updated_at=upd)
+            db.add(cell)
+            db.flush()
+            if state in ("null_status", "null_updated"):
+                # Bypass ORM/server defaults to persist genuine NULLs.
+                db.query(WorkbookEnrichment).filter(WorkbookEnrichment.id == cell.id).update(
+                    {"status": None} if state == "null_status" else {"updated_at": None},
+                    synchronize_session=False)
+    db.commit()
+    return wb.id, seen_states

tests/test_workbook_refresh_cell_scope.py

@@ -103,6 +103,19 @@ def test_all_fresh_cells_make_refresh_a_noop(refresh_runner, linked):
     assert calls == sent == []
 
 
+@pytest.mark.parametrize("linked", [False, True])
+def test_refresh_scope_is_independent_of_scan_page_size(refresh_runner, monkeypatch, linked):
+    from apps.api.core.config import settings
+
+    sessions, now, calls, sent = refresh_runner
+    monkeypatch.setattr(settings, "WORKBOOK_REFRESH_SCAN_BATCH_SIZE", 1)
+    wid = populate(sessions, now, linked)
+    result = asyncio.run(refresh.refresh_workbook(wid, workspace_id="refresh-scope"))
+    assert sorted(calls) == [("pitch {email}", "row202@example.test"),
+                             ("summary {email}", "row101@example.test")]
+    assert result == {"sourced": 0, "reenriched": 2, "stale_rows": 2}
+
+
 def test_refresh_recomputes_transitive_dependents_without_unrelated_spend(refresh_runner):
     sessions, now, calls, sent = refresh_runner
     columns = [