server-26/drb-c2-core/tests/test_recorrelation_sweep.py

"""
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"]