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

"""
Re-correlation sweep.

Runs every summary_interval_minutes (same tick as the summarizer).  Each pass
finds calls that are:
  - recently ended  (ended_at within the last recorrelation_scan_minutes)
  - still orphaned  (incident_id is null)

and re-runs the incident correlator against currently-active incidents, using
the call's own started_at as the time anchor so the window is correct regardless
of when the sweep fires.

Never creates new incidents — link-only.  Zero LLM tokens (uses pre-computed
talkgroup strings, haversine math, and stored embeddings).
"""
import asyncio
from datetime import datetime, timezone, timedelta
from typing import Optional
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.
#
# 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
# server-26#115 — a call the consensus LLM-orphan gate parked (llm=orphan vs
# rules=new, no substance) gets a longer budget. The gate fires before any
# incident for the job may exist, so the substantive call that would justify
# linking can land well after the standard ~6 min. Still link-only: a genuinely
# thin call must not mint an incident, and the rules creation gate would re-orphan
# it anyway.
GATED_ORPHAN_SWEEP_ATTEMPTS = 10


def _max_sweep_attempts(call: dict) -> int:
    if call.get("corr_consensus") == "llm_orphan_gate":
        return GATED_ORPHAN_SWEEP_ATTEMPTS
    return MAX_SWEEP_ATTEMPTS


async def recorrelation_loop() -> None:
    interval = settings.summary_interval_minutes * 60
    logger.info(
        f"Re-correlation sweep started — "
        f"interval: {settings.summary_interval_minutes}m, "
        f"scan window: {settings.recorrelation_scan_minutes}m"
    )
    while True:
        await asyncio.sleep(interval)
        try:
            await _run_sweep_pass()
        except Exception as e:
            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:
    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).
    # This keeps the fetched set small regardless of total collection size.
    recent_ended = await fstore.collection_where("calls", [
        ("status",   "==",  "ended"),
        ("ended_at", ">=",  cutoff),
    ])
    # corr_path="unlinked" is written after the attempt budget is exhausted.
    # Allows a few retries so a welfare-check call can link to an escalation
    # incident that is created a few minutes later, without sweeping 30× forever.
    orphans = [
        c for c in recent_ended
        if not c.get("incident_ids") and not c.get("incident_id")
        and not c.get("corr_path")  # skip calls already exhausted
        and not c.get("duplicate_of")  # another node's copy — never processed by design
        # /upload deliberately skips correlation for garbage and too-short
        # transcripts (routers/upload.py) because they carry no signal. The sweep
        # was not applying the same guard, so those fragments came back in through
        # the thin path minutes later and attached to whatever was most recent —
        # 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:
        return

    logger.info(f"Re-correlation sweep: {len(orphans)} orphaned call(s) to check")
    linked = 0
    for call in orphans:
        if await _recorrelate_orphan(call):
            linked += 1

    if linked:
        logger.info(f"Re-correlation sweep: linked {linked}/{len(orphans)} orphaned call(s)")


async def _recorrelate_orphan(call: dict) -> bool:
    """
    Attempt to link a single orphaned call to an existing incident.
    Returns True if a match was found and the call was linked.
    """
    from app.internal import incident_correlator

    call_id      = call.get("call_id")
    started_at   = _parse_dt(call.get("started_at"))
    if not call_id or not started_at:
        return False

    # All data needed for correlation was stored by the first-pass extraction.
    # embedding/severity are no longer read from the call doc inside
    # _build_context (server-26#80/#95) — the sweep re-links a whole call, not a
    # scene, so it passes the call doc's stored (primary-scene) values here. It
    # is link-only (create_if_new=False), so a borrowed severity cannot open a
    # new incident off this path.
    incident_id = await incident_correlator.correlate_call(
        call_id        = call_id,
        node_id        = call.get("node_id", ""),
        system_id      = call.get("system_id"),
        talkgroup_id   = call.get("talkgroup_id"),
        talkgroup_name = call.get("talkgroup_name"),
        tags           = call.get("tags") or [],
        incident_type  = call.get("incident_type"),
        location       = call.get("location"),
        location_coords= call.get("location_coords"),
        cleared_units  = call.get("cleared_units") or [],
        embedding      = call.get("embedding"),
        severity       = call.get("severity"),
        transcript     = call.get("transcript_corrected") or call.get("transcript"),
        reference_time = started_at,   # anchor window to when the call happened
        create_if_new  = False,        # never create — link-only
    )

    if incident_id:
        await fstore.doc_set("calls", call_id, {"incident_ids": [incident_id]})
        logger.info(
            f"Re-correlation: linked orphaned call {call_id} → incident {incident_id}"
        )
        return True

    # Increment the attempt counter. Once the budget is reached the orphan filter
    # above will stop picking this call up, and we write corr_path="unlinked" as
    # a permanent tombstone.
    attempts = call.get("corr_sweep_count", 0) + 1
    update: dict = {"corr_sweep_count": attempts}
    if attempts >= _max_sweep_attempts(call):
        update["corr_path"] = "unlinked"
    await fstore.doc_set("calls", call_id, update)
    return False


def _parse_dt(value) -> Optional[datetime]:
    if not value:
        return None
    try:
        dt = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except Exception:
        return None