diff --git a/apps/api/services/workbook/crm_import.py b/apps/api/services/workbook/crm_import.py index b249627..73addd5 100644 --- a/apps/api/services/workbook/crm_import.py +++ b/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)} diff --git a/apps/api/services/workbook/people_search.py b/apps/api/services/workbook/people_search.py index d81367b..8fafbfe 100644 --- a/apps/api/services/workbook/people_search.py +++ b/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, } diff --git a/apps/api/services/workbook/refresh.py b/apps/api/services/workbook/refresh.py index 6ce8e39..759a7a5 100644 --- a/apps/api/services/workbook/refresh.py +++ b/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( diff --git a/apps/api/services/workbook/source_engine.py b/apps/api/services/workbook/source_engine.py index 811520d..0f048cb 100644 --- a/apps/api/services/workbook/source_engine.py +++ b/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, diff --git a/docs/adr/20261012-refresh-dependent-and-new-row-cells.md b/docs/adr/20261012-refresh-dependent-and-new-row-cells.md index 0fe58f4..20160ce 100644 --- a/docs/adr/20261012-refresh-dependent-and-new-row-cells.md +++ b/docs/adr/20261012-refresh-dependent-and-new-row-cells.md @@ -11,14 +11,18 @@ 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. @@ -26,9 +30,13 @@ Running outputs during sourcing would introduce unrequested external dispatch. 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 diff --git a/tests/test_crm_import_sync.py b/tests/test_crm_import_sync.py index dd75631..17422a6 100644 --- a/tests/test_crm_import_sync.py +++ b/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" diff --git a/tests/test_workbook_refresh_cell_scope.py b/tests/test_workbook_refresh_cell_scope.py index eeef88d..0832134 100644 --- a/tests/test_workbook_refresh_cell_scope.py +++ b/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 == []