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