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/incident_correlator.py

@@ -1461,21 +1461,6 @@ async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]:
                 "transcript":   ctx.get("scene_transcript"),
                 "incident_id":  incident_id,
                 "corr_debug":   corr_debug,
-                # server-26#139: this scene's OWN extracted incident_type/
-                # severity, as read by _call_is_substanceless's ctx at
-                # decision time — not the call doc's flat top-level field,
-                # which is last-scene-wins (server-26#96) and was the reason
-                # #138's "type" veto couldn't be told apart from cross-scene
-                # contamination without re-guessing from a live dump.
-                # NOTE: unlike incident_type, call_severity is already
-                # coerced to "routine" when extraction emitted nothing
-                # (ctx build: `severity or "routine"`) — a scene reading
-                # "routine" here doesn't distinguish "extraction said
-                # routine" from "extraction said nothing". Don't split a
-                # severity veto the way #138 splits the type veto without
-                # accounting for that.
-                "incident_type": ctx.get("incident_type"),
-                "severity":      ctx.get("call_severity"),
             }
         }
         try:

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

@@ -32,17 +32,7 @@ from app.config import settings
 # 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
-# 15, not 5: neither OpenAI's Whisper client nor Gemini's call in
-# llm_correlator.py sets a request timeout (server-26#153), so a hung call can
-# run well past a few minutes on SDK-default retries, and this constant is a
-# guess against that unbounded tail, not a measured bound. Raising it costs
-# nothing on the recovery side: a call that finished processing (linked OR
-# genuinely orphaned) always has corr_path set (_apply_and_log writes it even
-# on the orphan action), so it's already excluded by the
-# `not c.get("corr_path")` filter below and never reaches this check at all —
-# this constant only ever delays calls that are still actually running.
-MIN_MINUTES_SINCE_PIPELINE_START = 15
 
 # Standard link-only retry budget before a call is tombstoned corr_path="unlinked".
 MAX_SWEEP_ATTEMPTS = 3

drb-c2-core/app/routers/admin.py

@@ -109,11 +109,6 @@ async def debug_correlation(
             "scene_index":            scene_index,
             "transcript":             scene.get("transcript"),
             "incident_id":            scene.get("incident_id"),
-            # server-26#139: this scene's OWN incident_type/severity, as seen
-            # by _call_is_substanceless at decision time — not the call doc's
-            # flat top-level field, which is last-scene-wins (server-26#96).
-            "incident_type":          scene.get("incident_type"),
-            "severity":               scene.get("severity"),
             "corr_path":              corr_debug.get("corr_path"),
             "corr_incident_idle_min": corr_debug.get("corr_incident_idle_min"),
             "corr_distance_km":       corr_debug.get("corr_distance_km"),

drb-c2-core/tests/test_per_scene_call_doc.py

@@ -123,66 +123,10 @@ async def test_single_scene_call_still_gets_a_scenes_map_equivalent_to_flat_fiel
             "transcript": "10-4",
             "incident_id": None,
             "corr_debug": {"corr_path": "fast/thin", "corr_consensus": "rules_only"},
-            "incident_type": None,
-            "severity": None,
         }
     }
 
 
-@pytest.mark.asyncio
-async def test_scene_entry_captures_its_own_incident_type_not_a_sibling_scenes():
-    """
-    server-26#139: _call_is_substanceless's "type" veto reads ctx["incident_type"]
-    at decision time, but that value was never persisted per-scene — only the
-    last-scene-wins flat field, which #138's dump analysis couldn't
-    distinguish from cross-scene contamination. Pins _apply_and_log's write
-    side: each scene's own scenes.<n> entry carries its own incident_type/
-    severity, distinct from any other scene on the same call. Does NOT cover
-    whether the ctx handed to _call_is_substanceless is the same object that
-    reaches here — that linkage is pinned by test_consensus_gate.py and
-    test_incident_identity.py, not this file.
-    """
-    docs: dict[tuple, dict] = {}
-
-    async def fake_doc_set(collection, doc_id, data, merge=True):
-        docs.setdefault((collection, doc_id), {})
-        _merge(docs[(collection, doc_id)], data)
-
-    decision0 = {
-        "action": "orphan", "matched_incident": None, "incident_type": None,
-        "corr_debug": {"corr_path": "new", "corr_consensus": "tiebreak", "corr_gate_veto": "type"},
-    }
-    ctx0 = {
-        "call_id": "call-5", "scene_index": 0, "scene_transcript": "10-4, clear",
-        "incident_type": "traffic-stop", "call_severity": "routine",
-    }
-
-    decision1 = {
-        "action": "orphan", "matched_incident": None, "incident_type": None,
-        "corr_debug": {"corr_path": "new", "corr_consensus": "agreed"},
-    }
-    ctx1 = {
-        "call_id": "call-5", "scene_index": 1, "scene_transcript": "roll call",
-        "incident_type": None, "call_severity": "moderate",
-    }
-
-    with patch.object(incident_correlator, "fstore") as mock_fstore:
-        mock_fstore.doc_set = fake_doc_set
-        await incident_correlator._apply_and_log(decision0, ctx0)
-        await incident_correlator._apply_and_log(decision1, ctx1)
-
-    doc = docs[("calls", "call-5")]
-    scenes = doc["scenes"]
-    assert scenes["0"]["incident_type"] == "traffic-stop"
-    assert scenes["0"]["severity"] == "routine"
-    assert scenes["1"]["incident_type"] is None
-    assert scenes["1"]["severity"] == "moderate"
-    # _apply_and_log only ever flat-merges corr_debug's own keys (:1460) — a
-    # future corr_debug["incident_type"] would silently clobber
-    # intelligence.py's flat field, so this is asserted, not just commented.
-    assert "incident_type" not in doc
-
-
 @pytest.mark.asyncio
 async def test_empty_corr_debug_writes_nothing_same_as_before():
     """Preserve the pre-#96 short-circuit: no corr_debug means no write at