From ec35d8058021d79195dad92b3423812331fc86b7 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:26:48 -0400 Subject: [PATCH] upload: correlate late uploads on the call's own clock; make /upload idempotent on call_id (server-26#301, server-26#302) The edge node now spools and retries uploads for up to 72h. #301: when a call's started_at is more than late_upload_anchor_minutes behind the real clock, extraction+correlation run under clock.pin(started_at) (the replay mechanism), so recency gates and every timestamp written are on the call's timeline. On-time uploads are unchanged. Also applies to /calls/{id}/reprocess, which shares _run_intelligence_pipeline. #302: an upload for a call_id whose audio is already stored returns the stored URL without rewriting GCS or re-running Whisper/extraction/ correlation. Not atomic: two concurrent retries can both pass the check. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/config.py | 4 + drb-c2-core/app/routers/upload.py | 76 ++++++++++++++++--- .../tests/test_correlation_decision_flow.py | 65 ++++++++++++++++ 3 files changed, 136 insertions(+), 9 deletions(-) diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index b796974..0d76d23 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -147,6 +147,10 @@ class Settings(BaseSettings): # location_proximity_km the slow path accepts). The LLM tier had no # distance check at all, which is how one incident drew a 37 km path. incident_max_spread_km: float = 2.0 + # server-26#301: an upload arriving more than this long after the call + # started is correlated as of its started_at (clock pin), not now. Well + # above normal upload + STT lag so on-time calls are never back-dated. + late_upload_anchor_minutes: int = 10 location_proximity_km: float = 0.5 # radius for location-proximity matching geocode_max_km: float = 40.0 # reject geocode results farther than this from the node incident_auto_resolve_minutes: int = 90 # auto-resolve after N minutes with no new calls (major / unknown severity) diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index 965a15a..cc69509 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -1,4 +1,5 @@ import secrets +from datetime import datetime, timedelta, timezone from typing import Optional from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form, HTTPException, Security from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials @@ -66,6 +67,24 @@ async def upload_call_audio( if len(data) > settings.upload_max_bytes: raise HTTPException(413, f"File too large (max {settings.upload_max_bytes // (1024*1024)} MB).") + # server-26#302: the edge node retries uploads (spooled for up to 72h), and + # a retry arrives whenever the node lost the response to an upload C2 had + # already stored. call_id is the idempotency key: audio already stored + # means this upload was already processed, so + # re-running Whisper/extraction/correlation would double the AI spend and + # double-link the call (the server-26#131 shape). Not atomic: two retries + # racing inside one request's lifetime can both pass this check. + # Keyed on stored audio only: the pipeline marker alone would refuse a + # retry forever (and hand the node a null URL to keep retrying) after a + # pipeline crash; such a call is recovered by the sweep or a reprocess. + existing = await fstore.doc_get("calls", call_id) or {} + if existing.get("audio_gcs_uri"): + logger.info( + f"Upload for call {call_id} from {node_id} is a retry of one already " + f"processed — returning the stored audio, pipeline not re-run" + ) + return {"url": existing.get("audio_gcs_uri"), "already_uploaded": True} + gcs_uri = await upload_audio(data, file.filename or "", call_id=call_id) if gcs_uri: @@ -456,6 +475,33 @@ async def _extract_and_correlate( return incident_ids, all_tags, scenes +def _late_upload_anchor(call_doc: Optional[dict]) -> Optional[datetime]: + """ + The call's own started_at when its upload is late (server-26#301), else + None. "Late" is `late_upload_anchor_minutes` behind the real clock: an + on-time call keeps being judged against now exactly as before, so the live + path's recency windows are unchanged; only a spooled/retried upload is + back-dated. A started_at in the future or unparseable is ignored. + """ + raw = (call_doc or {}).get("started_at") + if not raw: + return None + if isinstance(raw, datetime): # mqtt_handler stores a native timestamp + started = raw + else: + try: + started = datetime.fromisoformat(str(raw).replace("Z", "+00:00")) + except ValueError: + return None + if started.tzinfo is None: + started = started.replace(tzinfo=timezone.utc) + lag = clock.now() - started + if lag > timedelta(minutes=settings.late_upload_anchor_minutes): + logger.info(f"Late upload: call started {lag} ago — correlating as of {started.isoformat()}") + return started + return None + + async def _run_intelligence_pipeline( call_id: str, node_id: str, @@ -530,15 +576,27 @@ async def _run_intelligence_pipeline( incident_ids: list[str] = [] all_tags: list[str] = [] if _flag("correlation_enabled"): - incident_ids, all_tags, _ = await _extract_and_correlate( - call_id=call_id, - node_id=node_id, - system_id=system_id, - talkgroup_id=talkgroup_id, - talkgroup_name=talkgroup_name, - transcript=transcript, - segments=segments, - ) + # server-26#301: a late upload (the node spools and retries for up to + # 72h) is correlated as of when the call happened, not as of now — the + # same clock pin replay uses, so every recency gate and every + # started_at/updated_at/resolved_at it writes is on the call's own + # timeline. Without it a call hours late linked into whatever incident + # was live on its talkgroup now, or opened one at the wrong time. + anchor = _late_upload_anchor(_call_doc) + tok = clock.pin(anchor) if anchor else None + try: + incident_ids, all_tags, _ = await _extract_and_correlate( + call_id=call_id, + node_id=node_id, + system_id=system_id, + talkgroup_id=talkgroup_id, + talkgroup_name=talkgroup_name, + transcript=transcript, + segments=segments, + ) + finally: + if tok is not None: + clock.unpin(tok) else: scope = "globally" if not flags["correlation_enabled"] else f"system {system_id}" logger.info(f"Correlation disabled ({scope}) — skipping scene extraction and correlation for call {call_id}") diff --git a/drb-c2-core/tests/test_correlation_decision_flow.py b/drb-c2-core/tests/test_correlation_decision_flow.py index 99b99f6..98a087f 100644 --- a/drb-c2-core/tests/test_correlation_decision_flow.py +++ b/drb-c2-core/tests/test_correlation_decision_flow.py @@ -10,6 +10,8 @@ Also pins the section fixes that ride on the new shape: #189 weak links refused past the span cap; LLM links distance-guarded #190 severity steps down on an explicit stand-down, the peak never does #27 sweep budget is wall-clock + #301 late uploads are correlated on the call's own clock + #302 a retried upload does not re-run the pipeline """ from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, patch @@ -365,3 +367,66 @@ async def test_sweep_keeps_retrying_past_the_old_three_attempt_budget(): fs.doc_set = AsyncMock() assert await recorrelation_sweep._recorrelate_orphan(call, NOW) is False assert fs.doc_set.await_args.args[2] == {"corr_sweep_count": 6} + + +# ── #301 / #302 ───────────────────────────────────────────────────────────── + +def test_late_upload_is_anchored_to_started_at_and_on_time_is_not(): + with patch("app.internal.clock.now", return_value=NOW): + late = NOW - timedelta(hours=3) + assert upload._late_upload_anchor({"started_at": late.isoformat()}) == late + assert upload._late_upload_anchor({"started_at": late}) == late # native timestamp + assert upload._late_upload_anchor({"started_at": (NOW - timedelta(minutes=1)).isoformat()}) is None + assert upload._late_upload_anchor({}) is None + + +@pytest.mark.asyncio +async def test_late_upload_correlates_on_the_calls_own_clock(): + from app.internal import clock + seen = {} + started = NOW - timedelta(hours=3) + + async def fake_extract(**kw): + seen["now"] = clock.now() + return [], [], [] + + flags = ({"stt_enabled": False, "correlation_enabled": True}, lambda k: k == "correlation_enabled") + with patch.object(upload, "fstore") as fs, \ + patch.object(upload, "_resolve_flags", AsyncMock(return_value=flags)), \ + patch.object(upload, "_extract_and_correlate", fake_extract), \ + patch("app.internal.talkgroups.resolve", AsyncMock(return_value="Ch 1")), \ + patch("app.internal.alerter.check_and_dispatch", AsyncMock()): + fs.doc_get = AsyncMock(return_value={"started_at": started.isoformat(), "talkgroup_name": "Ch 1"}) + fs.doc_set = AsyncMock() + await upload._run_intelligence_pipeline("call-1", "n", "sys-1", 383, "Ch 1", None) + assert seen["now"] == started + assert clock.now() != started # pin released + + +@pytest.mark.asyncio +async def test_retried_upload_returns_stored_audio_without_rerunning_the_pipeline(): + from fastapi import BackgroundTasks + from fastapi.security import HTTPAuthorizationCredentials + + docs = { + ("node_keys", "n"): {"api_key": "k"}, + ("nodes", "n"): {"approval_status": "approved"}, + ("calls", "call-1"): {"audio_gcs_uri": "gs://b/call-1.mp3"}, + } + + class _File: + filename = "a.mp3" + + async def read(self): + return b"audio" + + bt = BackgroundTasks() + with patch.object(upload, "fstore") as fs, patch.object(upload, "upload_audio", AsyncMock()) as up: + fs.doc_get = AsyncMock(side_effect=lambda coll, doc_id: docs.get((coll, doc_id))) + out = await upload.upload_call_audio( + background_tasks=bt, file=_File(), call_id="call-1", node_id="n", + credentials=HTTPAuthorizationCredentials(scheme="Bearer", credentials="k"), + ) + assert out == {"url": "gs://b/call-1.mp3", "already_uploaded": True} + up.assert_not_awaited() + assert bt.tasks == []