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
import math
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.
#
# Derived, not guessed (server-26#153). Every AI request the real-time
# pipeline makes now has an explicit timeout (config.py: stt_timeout_seconds,
# openai_timeout_seconds, openai_max_retries, gemini_timeout_seconds), so the
# longest a call can stay "started but no corr_path yet" has an upper bound:
# pipeline_worst_case_seconds() below. The hold-back is that bound rounded up,
# never below the old hand-picked 15.
#
# 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, or whose pipeline died after writing the marker.

# Scenes whose embeddings are computed before the first scene's correlation
# commits (intelligence.extract_scenes embeds every scene, then upload.py
# correlates them in order; scene 0's commit writes corr_path). Extraction
# rarely returns more than 2; 3 is the budget, not a hard cap.
_BOUND_SCENES = 3
# Non-AI work on the same path: GCS download, Firestore reads/writes, geocoding
# and place verification (httpx, 5s timeouts each).
_NON_AI_SLACK_SECONDS = 120.0
_FLOOR_MINUTES = 15


def pipeline_worst_case_seconds(n_scenes: int = _BOUND_SCENES) -> float:
    """Upper bound on intelligence_started_at -> first corr_path write, from
    the configured AI timeouts. Order follows routers/upload.py
    _run_intelligence_pipeline."""
    openai_attempts = 1 + max(0, settings.openai_max_retries)
    stt_rounds = 2 if settings.stt_retry_on_degenerate else 1
    stt = stt_rounds * openai_attempts * settings.stt_timeout_seconds
    # Transcript correction (inside transcribe_call). gemini.generate_json only
    # re-issues a request when the model rejects a thinking level — a fast
    # 400, not a timeout — and google-genai itself does not retry.
    correction = settings.gemini_timeout_seconds
    extraction = openai_attempts * settings.openai_timeout_seconds
    embeddings = n_scenes * openai_attempts * settings.openai_timeout_seconds
    correlation = settings.gemini_timeout_seconds  # at most one LLM call, scene 0 (#299)
    return stt + correction + extraction + embeddings + correlation + _NON_AI_SLACK_SECONDS


def _derive_min_minutes() -> int:
    return max(_FLOOR_MINUTES, math.ceil(pipeline_worst_case_seconds() / 60))


# 21 with the defaults (stt 120s, openai 60s x2 attempts, gemini 60s, 3 scenes).
MIN_MINUTES_SINCE_PIPELINE_START = _derive_min_minutes()


def _last_chance(call: dict, now: datetime) -> bool:
    """
    server-26#27 — the retry budget is wall-clock, not an attempt count.

    The old budget was 3 attempts, which with the #131 hold-back expired ~19
    minutes after the pipeline started, while the incident the orphan belongs
    to may be opened later than that (`f321b047` was tombstoned three minutes
    after its own incident was created). An orphan is now retried on every
    pass for as long as the scan query can see it, and tombstoned
    corr_path="unlinked" only on its LAST visible pass: the one after which
    its ended_at falls out of `recorrelation_scan_minutes`. A call with no
    parseable ended_at gets no further retries.
    """
    ended = _parse_dt(call.get("ended_at"))
    if ended is None:
        return True
    remaining = timedelta(minutes=settings.recorrelation_scan_minutes) - (now - ended)
    return remaining <= timedelta(minutes=settings.summary_interval_minutes)


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, "
        f"pipeline hold-back: {MIN_MINUTES_SINCE_PIPELINE_START}m"
    )
    if MIN_MINUTES_SINCE_PIPELINE_START >= settings.recorrelation_scan_minutes:
        # A call is held back longer than the window that finds it, so the
        # sweep can never recover a call whose pipeline died mid-flight.
        logger.warning(
            f"Re-correlation sweep: pipeline hold-back ({MIN_MINUTES_SINCE_PIPELINE_START}m) "
            f">= scan window ({settings.recorrelation_scan_minutes}m) — lower the AI "
            f"timeouts or raise RECORRELATION_SCAN_MINUTES"
        )
    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:
    from app.internal import clock
    now = clock.now()
    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 the tombstone written on an orphan's last
    # visible pass (_last_chance); the scan window itself is the budget.
    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 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, now):
            linked += 1

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


async def _recorrelate_orphan(call: dict, now: datetime) -> 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

    # The attempt count is kept for the admin debug view only; the budget is
    # the scan window (_last_chance), after which the call is tombstoned.
    update: dict = {"corr_sweep_count": call.get("corr_sweep_count", 0) + 1}
    if _last_chance(call, now):
        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