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. Confirmed in review:
_update_incident/_create_incident append call_id to an incident's call_ids
unconditionally, with no cross-incident dedup guard -- preventing the second
correlation attempt is the only lever available at this layer.

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 15
minutes old (raised from an initial 5 -- see below), 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. Also
covers the /calls/{id}/reprocess path, which calls the same
_run_intelligence_pipeline.

15 min, not 5: neither the OpenAI Whisper client nor the Gemini call in
llm_correlator.py sets a request timeout (filed server-26#153), so 5 min was
a guess against an unbounded tail -- drb-correlation-review flagged this.
Raising it is free on the recovery side: a call that finished processing
(linked or genuinely orphaned) always has corr_path set and is already
excluded by the sweep's other filter, so this constant only ever delays
calls that are still actually running. DEFERRED.md row 52 (outside this
repo, Version 5C root) updated to flag its ~6 min timing figure as stale.

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/routers/upload.py

@@ -413,6 +413,25 @@ async def _run_intelligence_pipeline(
     """
     from app.internal import transcription, intelligence, incident_correlator, alerter, talkgroups
 
+    # server-26#131: mark that real-time processing has started for this call
+    # BEFORE any of the slow steps below (STT, scene extraction, correlation).
+    # The re-correlation sweep (internal/recorrelation_sweep.py) scans for
+    # calls that still look orphaned within a wide window (recorrelation_scan_
+    # minutes, default 60) — with no guard here, a call whose real-time
+    # pipeline is still mid-flight (still transcribing, still waiting on a
+    # Gemini call) has no incident_id/corr_path written yet, so the sweep's
+    # orphan filter can't tell "never processed" from "processing right now"
+    # and correlates it a second time, independently, sometimes landing on a
+    # different incident than the real-time path — the exact duplicate-link
+    # bug #131 found (same call in two incidents' call_ids, ~2% of linked
+    # calls). Best-effort: a write failure here must not abort the pipeline.
+    try:
+        await fstore.doc_set("calls", call_id, {
+            "intelligence_started_at": datetime.now(timezone.utc).isoformat()
+        })
+    except Exception as e:
+        logger.warning(f"Could not mark intelligence_started_at for call {call_id}: {e}")
+
     # The node only sends talkgroup_name when OP25 had it in the loaded tags
     # file, so it arrives empty for exactly the talkgroups C2 can name from the
     # system config. Resolve it once, here, at the single funnel both /upload