correlator: stop the re-correlation sweep racing an in-flight upload (#131)

server-26#131: same call_id ends up in TWO incidents' call_ids, byte-identical
extracted data, ~2% of linked calls across 3 live dumps. Root cause: the
sweep's orphan filter (incident_id/incident_ids/corr_path all absent) can't
tell 'never processed' apart from 'real-time pipeline is still mid-flight' --
a call whose STT/scene-extraction/correlation chain (routers/upload.py
_run_intelligence_pipeline) hasn't finished yet has none of those fields set,
so the sweep picks it up and correlates it independently, sometimes onto a
different incident than the real-time path lands on.

Fix: _run_intelligence_pipeline marks intelligence_started_at on the call doc
before any slow step; the sweep holds back any call whose marker is under 5
minutes old, regardless of how orphaned it otherwise looks. No marker at all
(pre-#131 call doc, or the marker write itself failed) is not held back --
absence isn't evidence of an in-flight pipeline, and that's #131's own
pre-existing population.

recorrelation_sweep.py had zero test coverage before this. New file covers
the guard function's boundary (age < threshold vs exactly-at vs old vs
missing vs unparseable) and one integration-shaped test proving a racing call
never reaches correlate_call while a genuinely-orphaned call still does.

Sandboxed pytest: 381 -> 387.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix

drb-c2-core/app/internal/recorrelation_sweep.py

@@ -20,6 +20,20 @@ from app.internal.logger import logger
 from app.internal import firestore as fstore
 from app.config import settings
 
+# server-26#131: minimum time since the real-time pipeline (routers/upload.py
+# _run_intelligence_pipeline) marked intelligence_started_at before the sweep
+# will touch a call, even though it already looks orphaned. STT + scene
+# extraction + correlation is a multi-second-to-low-minutes chain (Whisper,
+# then a Gemini call per scene); without this buffer the sweep could pick up
+# a call mid-pipeline — no incident_id/corr_path written yet — and correlate
+# it a second time, independently, sometimes onto a different incident than
+# the real-time path lands on. That race is what #131 found (same call in
+# two incidents' call_ids, ~2% of linked calls). A call with no
+# intelligence_started_at at all (pre-#131 call doc, or the marker write
+# itself failed) is NOT held back by this — absence isn't evidence of an
+# in-flight pipeline, and #131's own bug predates this field existing.
+MIN_MINUTES_SINCE_PIPELINE_START = 5
+
 # Standard link-only retry budget before a call is tombstoned corr_path="unlinked".
 MAX_SWEEP_ATTEMPTS = 3
 # server-26#115 — a call the consensus LLM-orphan gate parked (llm=orphan vs
@@ -52,8 +66,22 @@ async def recorrelation_loop() -> None:
             logger.error(f"Re-correlation sweep failed: {e}")
 
 
+def _pipeline_likely_still_running(call: dict, now: datetime) -> bool:
+    """server-26#131 — True when the real-time pipeline marked
+    intelligence_started_at recently enough that it's probably still mid-flight
+    (STT / scene extraction / correlation), so the sweep should not race it.
+    No marker at all (older call doc, or the marker write itself failed)
+    returns False — absence isn't evidence of an in-flight pipeline."""
+    started = _parse_dt(call.get("intelligence_started_at"))
+    if not started:
+        return False
+    age_minutes = (now - started).total_seconds() / 60
+    return age_minutes < MIN_MINUTES_SINCE_PIPELINE_START
+
+
 async def _run_sweep_pass() -> None:
-    cutoff = datetime.now(timezone.utc) - timedelta(minutes=settings.recorrelation_scan_minutes)
+    now = datetime.now(timezone.utc)
+    cutoff = now - timedelta(minutes=settings.recorrelation_scan_minutes)
 
     # Server-side range query: only calls that ended within the scan window.
     # Filter incident_id=null client-side (Firestore can't query for missing fields).
@@ -77,6 +105,7 @@ async def _run_sweep_pass() -> None:
         # a second route into the over-merge the thin fix above addresses.
         and not c.get("skip_reason")
         and c.get("corr_sweep_count", 0) < _max_sweep_attempts(c)
+        and not _pipeline_likely_still_running(c, now)
     ]
 
     if not orphans: