""" server-26#131 — the re-correlation sweep's orphan filter checked incident_id/ incident_ids/corr_path but had no way to tell "never processed" apart from "real-time pipeline (routers/upload.py _run_intelligence_pipeline) is still mid-flight". Racing the sweep against an in-flight real-time correlation could land the same call on two different incidents — the exact duplicate-link bug #131 found in 3 live dumps (~2% of linked calls). This pins the fix: a call whose intelligence_started_at marker is recent is held back from the sweep regardless of how orphaned it otherwise looks. """ from datetime import datetime, timezone, timedelta from unittest.mock import patch import pytest from app.internal import recorrelation_sweep def _iso(dt: datetime) -> str: return dt.isoformat() class TestPipelineLikelyStillRunning: def test_recent_marker_is_still_running(self): now = datetime.now(timezone.utc) call = {"intelligence_started_at": _iso(now - timedelta(minutes=1))} assert recorrelation_sweep._pipeline_likely_still_running(call, now) is True def test_old_marker_is_not_still_running(self): now = datetime.now(timezone.utc) call = {"intelligence_started_at": _iso(now - timedelta(minutes=30))} assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False def test_marker_exactly_at_the_threshold_is_not_held_back(self): # age_minutes < MIN_MINUTES_SINCE_PIPELINE_START (strict), so exactly # at the threshold is old enough to release — pins the boundary so it # can't drift to <= by accident and silently double the hold time. now = datetime.now(timezone.utc) threshold = recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START call = {"intelligence_started_at": _iso(now - timedelta(minutes=threshold))} assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False def test_no_marker_at_all_is_not_held_back(self): """A pre-#131 call doc, or the marker write itself failed — absence isn't evidence of an in-flight pipeline, so the sweep must still be able to pick these up (that's its whole job).""" now = datetime.now(timezone.utc) assert recorrelation_sweep._pipeline_likely_still_running({}, now) is False def test_unparseable_marker_is_not_held_back(self): now = datetime.now(timezone.utc) call = {"intelligence_started_at": "not-a-timestamp"} assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False @pytest.mark.asyncio async def test_sweep_pass_skips_a_call_whose_pipeline_just_started(): """Integration-shaped: a call that looks orphaned by every OTHER filter (no incident_ids, no corr_path, no skip_reason, under the attempt budget) but has a fresh intelligence_started_at must not reach correlate_call — that's the race #131 found.""" now = datetime.now(timezone.utc) racing_call = { "call_id": "call-racing", "started_at": _iso(now - timedelta(minutes=2)), "ended_at": _iso(now - timedelta(minutes=1)), "intelligence_started_at": _iso(now - timedelta(seconds=30)), } genuinely_orphaned_call = { "call_id": "call-genuine-orphan", "started_at": _iso(now - timedelta(minutes=20)), "ended_at": _iso(now - timedelta(minutes=19)), "intelligence_started_at": _iso(now - timedelta(minutes=19)), } async def fake_collection_where(collection, clauses): assert collection == "calls" return [racing_call, genuinely_orphaned_call] correlate_calls: list[str] = [] async def fake_correlate_call(**kwargs): correlate_calls.append(kwargs["call_id"]) return None # no match — exercises the "not linked" branch too doc_sets: list[tuple] = [] async def fake_doc_set(collection, doc_id, data, merge=True): doc_sets.append((collection, doc_id, data)) with patch.object(recorrelation_sweep, "fstore") as mock_fstore, \ patch("app.internal.incident_correlator.correlate_call", fake_correlate_call): mock_fstore.collection_where = fake_collection_where mock_fstore.doc_set = fake_doc_set await recorrelation_sweep._run_sweep_pass() assert correlate_calls == ["call-genuine-orphan"]