fix(workbook): refresh only rows created by its own sourcing

apps/api/services/workbook/crm_import.py

@@ -222,4 +222,5 @@ async def materialize_crm_import(
         f"CRM import {workbook_id}/{column_id} ({crm}): "
         f"found={found} added={added} skipped={skipped}"
     )
-    return {"found": found, "added": added, "skipped": skipped, "crm": crm}
+    return {"found": found, "added": added, "skipped": skipped, "crm": crm,
+            "added_row_ids": list(_added_row_ids)}

apps/api/services/workbook/people_search.py

@@ -427,6 +427,7 @@ async def materialize_people_search(
     )
     result = {
         "found": found, "added": added, "skipped": skipped,
+        "added_row_ids": list(_added_row_ids),
         "companies": len(companies), "searches_used": searches_used,
         "search_failures": search_failures,
     }

apps/api/services/workbook/refresh.py

@@ -116,15 +116,16 @@ async def _refresh_workbook_impl(workbook_id: str, reason: str, workspace_id: st
         source_cols = [c for c in cols if c.get("type") == "source"]
         enrichment_cols = [c for c in cols if c.get("type") in ("enrichment", "waterfall", "ai_formula")]
         ttl_map = policy.get("staleness_ttl_days") or {}
-        previous_row_ids = {row.id for row in db.query(WorkbookRow.id).filter(
-            WorkbookRow.workbook_id == workbook_id
-        ).all()}
 
-    # 1) Re-source (append-only via entity dedup)
+    # 1) Re-source (append-only via entity dedup). Track exactly the rows our
+    # own sourcing inserted: rows added concurrently by other writers (CSV
+    # import, API insert, another job) must not get every column run here.
     sourced = 0
+    added_row_ids: list = []
     for col in source_cols:
         res = await materialize_source(workbook_id, col["id"], workspace_id)
         sourced += res.get("added", 0)
+        added_row_ids.extend(res.get("added_row_ids") or [])
 
     # 2) Refresh stale cells and their reactive downstream dependencies.
     def is_reactive(column: dict) -> bool:
@@ -140,14 +141,11 @@ def is_reactive(column: dict) -> bool:
     reenriched = 0
     with SessionLocal() as db:
         stale = _stale_row_columns(db, workbook_id, enrichment_cols, ttl_map) if enrichment_cols else {}
-        added_row_ids = {row.id for row in db.query(WorkbookRow.id).filter(
-            WorkbookRow.workbook_id == workbook_id
-        ).all()} - previous_row_ids
     for row_id, column_ids in stale.items():
         dependents = downstream_columns(cols, set(column_ids), eligible=is_reactive)
         stale[row_id] = list(dict.fromkeys([*column_ids, *(c["id"] for c in dependents if c.get("id"))]))
     if new_row_columns:
-        for row_id in added_row_ids:
+        for row_id in dict.fromkeys(added_row_ids):
             stale[str(row_id)] = list(new_row_columns)
     if stale:
         result = await run_workbook_enrichment(

apps/api/services/workbook/source_engine.py

@@ -76,7 +76,9 @@ async def materialize_source(
     ``workspace_id`` is the tenant this run belongs to. It arrives out-of-band
     (job payload / caller) — never read back off the workbook row — and binds the
     RLS scope for the whole run, so the wb/row reads and the lead-store read-back
-    are all tenant-correct. Returns a summary {found, added, skipped, query}.
+    are all tenant-correct. Returns a summary {found, added, skipped, query,
+    added_row_ids}; ``added_row_ids`` lists exactly the rows this run inserted
+    (absent on early-error returns, so callers should ``.get(..., [])``).
     """
     from apps.api.core.tenancy import workspace_scope
 
@@ -384,6 +386,7 @@ async def _materialize_source_impl(
         "shortfall": max(0, requested_count - total_rows_after),
         "found": found,
         "added": added,
+        "added_row_ids": list(_added_row_ids),
         "skipped": skipped,
         "rejected_by_reason": rejected_by_reason,
         "exhausted_sources": exhausted_sources,

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

@@ -11,24 +11,32 @@ and research cells uninitialized.
 
 Expand each existing row's stale-cell scope through the existing transitive
 `downstream_columns` helper, using the row-edit path's reactive eligibility
-policy. Snapshot persisted row IDs before sourcing and compare them afterward.
-New rows initialize every runnable non-output column. Pass the resulting exact
+policy. The source engine reports the IDs of the rows each run inserted
+(`added_row_ids` in the `materialize_source` summary); those rows, and only
+those, initialize every runnable non-output column. Pass the resulting exact
 per-row scopes through the existing enrichment runner.
 
 ## Alternatives
 
-Returning added IDs from the source engine requires changing its return contract
-and each caller. Running all columns on every stale row would restore dependent
+The original version of this decision snapshotted persisted row IDs before
+sourcing and diffed afterward, avoiding a source-engine contract change. That
+misclassified rows inserted concurrently by other writers (CSV import, API
+insert, another job) as sourced, running paid enrichments on them; it was
+replaced by the additive `added_row_ids` key. Running all columns on every stale row would restore dependent
 values but discard this PR's cost-saving behavior for unrelated fresh cells.
 Running outputs during sourcing would introduce unrequested external dispatch.
 
 ## Consequences
 
 Changed inputs propagate through derived columns while unrelated fresh cells
 remain intact. Explicitly non-reactive dependents stay excluded; outputs remain
-non-reactive by default and new-row initialization excludes outputs. Row-ID
-comparison adds two scoped ID reads and identifies rows added during the cycle.
-It is not a concurrent-writer snapshot isolation guarantee.
+non-reactive by default and new-row initialization excludes outputs.
+
+Resolved limitation: the snapshot diff was not concurrent-writer safe. Refresh
+now uses the IDs the source engine itself inserted, so rows added by another
+writer during sourcing are left alone (covered by
+`test_concurrently_inserted_row_is_not_treated_as_sourced`). No extra row-ID
+queries are issued.
 
 ## Validation
 

tests/test_crm_import_sync.py

@@ -174,10 +174,12 @@ def test_hubspot_import_paging_mapping_and_sync_key(db, monkeypatch, flag_on, hu
     col_id = _mk_workbook(db, source={"kind": "crm_import", "crm": "hubspot"})
 
     res = _run(se.materialize_source("wb1", col_id, WS_A))
+    added_ids = res.pop("added_row_ids")
     assert res == {"found": 3, "added": 3, "skipped": 0, "crm": "hubspot"}
     assert len(calls) == 2  # two pages
 
     rows = _rows(db)
+    assert added_ids == [r["id"] for r in rows]  # exactly the rows this run inserted
     assert [r["crm_external_id"] for r in rows] == ["1", "2", "3"]
     assert all(r["crm_type"] == "hubspot" and r["crm_object"] == "contact" for r in rows)
     assert rows[0]["email"] == "alice@x.com"
@@ -200,7 +202,7 @@ def test_hubspot_import_rerun_dedups_on_external_id(db, monkeypatch, flag_on, hu
     first = _run(se.materialize_source("wb1", col_id, WS_A))
     again = _run(se.materialize_source("wb1", col_id, WS_A))
     assert first["added"] == 3
-    assert again == {"found": 3, "added": 0, "skipped": 3, "crm": "hubspot"}
+    assert again == {"found": 3, "added": 0, "skipped": 3, "crm": "hubspot", "added_row_ids": []}
     assert len(_rows(db)) == 3
 
 
@@ -321,6 +323,7 @@ def handler(request: httpx.Request) -> httpx.Response:
         "kind": "crm_import", "crm": "salesforce", "filter": "Email != null",
     })
     res = _run(se.materialize_source("wb1", col_id, WS_A))
+    assert len(res.pop("added_row_ids")) == 2
     assert res == {"found": 2, "added": 2, "skipped": 0, "crm": "salesforce"}
     assert seen["soql"].startswith(
         "SELECT Id, Email, FirstName, LastName, Account.Name, Phone, Title FROM Contact"

tests/test_workbook_refresh_cell_scope.py

@@ -143,9 +143,9 @@ def test_refresh_recomputes_transitive_dependents_without_unrelated_spend(refres
     assert sent == []
 
 
-def test_new_source_row_runs_every_runnable_column_without_dispatch(refresh_runner, monkeypatch):
+def _sourced_workbook(sessions, monkeypatch, *, concurrent_row: bool):
+    """Workbook whose sourcing adds row 202; optionally another writer adds 303 meanwhile."""
     from apps.api.services.workbook import source_engine, http_column, research_column, agent_column
-    sessions, now, calls, sent = refresh_runner
     columns = [
         {"id": "source", "type": "source"},
         {"id": "email", "type": "input"},
@@ -169,8 +169,12 @@ async def source(_workbook_id, _column_id, _workspace_id):
         with sessions() as db:
             db.add(WorkbookRow(id=202, workbook_id=wid, workspace_id="refresh-scope", position=1,
                                data={"email": "new@example.test"}))
+            if concurrent_row:
+                # A CSV import / API insert landing while sourcing is in flight.
+                db.add(WorkbookRow(id=303, workbook_id=wid, workspace_id="refresh-scope", position=2,
+                                   data={"email": "imported@example.test"}))
             db.commit()
-        return {"added": 1}
+        return {"added": 1, "added_row_ids": [202]}
     async def http(*args, **kwargs):
         return {"value": "HTTP", "error": None}
     async def research(*args, **kwargs):
@@ -181,6 +185,12 @@ async def agent(*args, **kwargs):
     monkeypatch.setattr(http_column, "execute_http_column", http)
     monkeypatch.setattr(research_column, "execute_research_column", research)
     monkeypatch.setattr(agent_column, "run_agent_cell", agent)
+    return wid
+
+
+def test_new_source_row_runs_every_runnable_column_without_dispatch(refresh_runner, monkeypatch):
+    sessions, now, calls, sent = refresh_runner
+    wid = _sourced_workbook(sessions, monkeypatch, concurrent_row=False)
     result = asyncio.run(refresh.refresh_workbook(wid, workspace_id="refresh-scope"))
     with sessions() as db:
         added = db.get(WorkbookRow, 202)
@@ -190,3 +200,22 @@ async def agent(*args, **kwargs):
         assert "send" not in added.enrichments
     assert result == {"sourced": 1, "reenriched": 4, "stale_rows": 1}
     assert sent == calls == []
+
+
+def test_concurrently_inserted_row_is_not_treated_as_sourced(refresh_runner, monkeypatch):
+    """Only rows the refresh's own sourcing created get every column run.
+
+    A row another writer inserts while sourcing is in flight is not ours: the
+    scheduled refresh must not spend paid enrichments on it.
+    """
+    sessions, now, calls, sent = refresh_runner
+    wid = _sourced_workbook(sessions, monkeypatch, concurrent_row=True)
+    result = asyncio.run(refresh.refresh_workbook(wid, workspace_id="refresh-scope"))
+    with sessions() as db:
+        added = db.get(WorkbookRow, 202)
+        assert {cid: added.enrichments[cid]["value"] for cid in ("formula", "http", "agent", "research")} == {
+            "formula": "new@example.test", "http": "HTTP", "agent": "AGENT", "research": "RESEARCH"}
+        assert not db.get(WorkbookRow, 303).enrichments
+        assert not db.get(WorkbookRow, 101).enrichments
+    assert result == {"sourced": 1, "reenriched": 4, "stale_rows": 1}
+    assert sent == calls == []