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 <noreply@anthropic.com>

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)

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,6 +576,15 @@ async def _run_intelligence_pipeline(
     incident_ids: list[str] = []
     all_tags: list[str] = []
     if _flag("correlation_enabled"):
+        # 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,
@@ -539,6 +594,9 @@ async def _run_intelligence_pipeline(
                 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}")

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 == []