fix(poller): consume funding and executive filings in one batch

Signed-off-by: Rudy Celekli <47457359+rudycelekli@users.noreply.github.com>

Author: rudycelekli

apps/api/services/poller/engine.py

@@ -502,14 +502,14 @@ def _poll_one_source(store, watch_id, workspace_id, src, fire_key, lead_id, back
         if src in ("feed", "job_change", "account_group"):
             bill_src = src
         else:
-            bill_src = "funding" if src in ("funding", "exec") else "hiring"
+            bill_src = "funding" if src in ("funding", "exec", "funding_exec") else "hiring"
         if not _debit_source(db, workspace_id, bill_src, fire_key):
             watch.last_error = "insufficient_credits"
             return False
 
-        if src == "funding" or src == "exec":
+        if src in ("funding", "exec", "funding_exec"):
             events, patch = sources.fetch_funding_and_exec(
-                watch, want_funding=(src == "funding"), want_exec=(src == "exec"),
+                watch, want_funding=(src != "exec"), want_exec=(src != "funding"),
                 backfill=backfill,
             )
         elif src == "hiring" or src == "tech":
@@ -570,7 +570,8 @@ def _poll_one_source(store, watch_id, workspace_id, src, fire_key, lead_id, back
 
 def _source_set(watch) -> list:
     """Sources to fan in for a watch kind. company → funding+exec+hiring+tech.
-    Each entry is one independent-txn source."""
+    Each entry is one independent-txn source. SEC event types share a watermark
+    and must consume their filing batch together."""
     kind = watch.kind
     types = set(watch.signal_types or [])
 
@@ -584,13 +585,13 @@ def wants(*sts):
         settings, "TECH_STACK_WEBSITE_FETCH_ENABLED", False
     ) else "tech"
 
+    # Both SEC detectors consume sec_last_accession. Separate transactions let
+    # the first detector advance it before the second sees the same filings.
+    sec_sources = (["funding_exec"] if wants("company_funded") and wants("executive_hired")
+                   else ["funding"] if wants("company_funded")
+                   else ["exec"] if wants("executive_hired") else [])
     if kind == "funding":
-        out = []
-        if wants("company_funded"):
-            out.append("funding")
-        if wants("executive_hired"):
-            out.append("exec")
-        return out or ["funding"]
+        return sec_sources or ["funding"]
     if kind == "hiring":
         out = []
         if wants("hiring_surge"):
@@ -605,16 +606,12 @@ def wants(*sts):
     if kind == "account_group":
         return ["account_group"]
     if kind == "company":
-        out = []
-        if wants("company_funded"):
-            out.append("funding")
-        if wants("executive_hired"):
-            out.append("exec")
+        out = list(sec_sources)
         if wants("hiring_surge"):
             out.append("hiring")
         if wants("new_tech_adopted"):
             out.append(tech_src)
-        return out or ["funding", "exec", "hiring", tech_src]
+        return out or ["funding_exec", "hiring", tech_src]
     return []
 
 

tests/test_poller_sec_batch.py

@@ -0,0 +1,65 @@
+"""Funding and executive detectors must consume their shared filing batch together."""
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+
+from apps.api.core.tenancy import current_workspace_var
+from apps.api.database import SessionLocal
+from apps.api.services.leadgen.enrichment.providers import sec_edgar
+from apps.api.services.leadgen.orm_models import LeadRow, SignalRow
+from apps.api.services.poller import engine
+from apps.api.services.poller.models import WatchSubscription
+from apps.api.services.signals.store import SignalStore
+
+
+@pytest.mark.parametrize("kind,types,expected", [
+    ("company", ["company_funded", "executive_hired"], {"company_funded", "executive_hired"}),
+    ("funding", ["company_funded", "executive_hired"], {"company_funded", "executive_hired"}),
+    ("funding", ["company_funded"], {"company_funded"}),
+    ("company", ["executive_hired"], {"executive_hired"}),
+])
+def test_sec_poll_preserves_enabled_event_types_and_replay(kind, types, expected, monkeypatch):
+    workspace, watch_id = f"sec-batch-{uuid4()}", str(uuid4())
+    filing = SimpleNamespace(cik="0001234567", accession="acc101", filing_date="2026-01-01",
+                             fields={"funding_amount": 5000000},
+                             related_persons=[{"name": "Synthetic Person", "title": "Chief Executive Officer"}])
+    calls = []
+    class Provider:
+        async def list_form_d_since(self, target, since):
+            calls.append(since)
+            return [filing] if not since or filing.accession > since else []
+    monkeypatch.setattr(sec_edgar, "SecEdgarProvider", Provider)
+    monkeypatch.setattr(engine, "_debit_source", lambda *args: True)
+    with SessionLocal() as db:
+        lead = LeadRow(workspace_id=workspace, company="Synthetic SEC Company")
+        db.add(lead)
+        db.flush()
+        lead_id = lead.id
+        db.add(WatchSubscription(id=watch_id, workspace_id=workspace, kind=kind,
+                                target=lead.company, lead_id=lead_id, interval="daily", enabled=True,
+                                signal_types=types, cursor={"bootstrapped": True, "sec_last_accession": "acc100"}))
+        db.commit()
+    store = SignalStore(workspace)
+    token = current_workspace_var.set(workspace)
+    try:
+        for attempt in ("first", "replay"):
+            with SessionLocal() as db:
+                watch = db.get(WatchSubscription, watch_id)
+                kinds = engine._source_set(watch)
+            for source in kinds:
+                engine._poll_one_source(store, watch_id, workspace, source, attempt, lead_id, False)
+            with SessionLocal() as db:
+                signals = db.query(SignalRow).filter_by(workspace_id=workspace).all()
+                assert {signal.signal_type for signal in signals} == expected
+                assert len(signals) == len(expected)
+                assert {signal.lead_id for signal in signals} == {lead_id}
+                assert db.get(WatchSubscription, watch_id).cursor["sec_last_accession"] == "acc101"
+        assert calls == ["acc100", "acc101"]
+    finally:
+        with SessionLocal() as db:
+            db.query(SignalRow).filter_by(workspace_id=workspace).delete()
+            db.query(WatchSubscription).filter_by(id=watch_id).delete()
+            db.query(LeadRow).filter_by(workspace_id=workspace).delete()
+            db.commit()
+        current_workspace_var.reset(token)