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

# 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}")


async def _run_sweep_pass() -> None:
    cutoff = datetime.now(timezone.utc) - 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)
    ]

    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