diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py
index 3b187c8..13d251a 100644
--- a/drb-c2-core/app/config.py
+++ b/drb-c2-core/app/config.py
@@ -78,6 +78,12 @@ class Settings(BaseSettings):
     # browsing session, short enough that a copied link isn't durable access.
     audio_link_ttl_seconds: int = 6 * 60 * 60
 
+    # Two nodes hearing the same transmission start recording within about a
+    # second of each other (measured across node-002/node-PI-2 on TG 9048).
+    # 10s is generous against clock skew while staying well under the gap
+    # between genuinely separate transmissions on a busy dispatch channel.
+    duplicate_window_seconds: int = 10
+
     # CORS — set to your frontend origin(s) in production, e.g. ["https://app.example.com"]
     # Defaults to "*" for local development only.
     cors_origins: list[str] = ["*"]
diff --git a/drb-c2-core/app/internal/dedup.py b/drb-c2-core/app/internal/dedup.py
new file mode 100644
index 0000000..4d7ab1b
--- /dev/null
+++ b/drb-c2-core/app/internal/dedup.py
@@ -0,0 +1,112 @@
+"""
+Cross-node duplicate detection for call recordings.
+
+Two edge nodes within range of the same trunked system both decode and upload
+the same transmission. That is the normal case for a distributed network, not
+an error — but without this, one transmission is transcribed twice, billed
+twice, and correlated twice, and the resulting incident shows two "units"
+where there was one.
+
+CANONICAL SELECTION IS DELIBERATELY NOT "FIRST UPLOAD WINS". Upload order
+depends on encode time and network latency, so it varies run to run; picking
+by it would make which recording is authoritative non-deterministic. The call
+document is created from the MQTT call_start event *before* the upload
+arrives, so by upload time every node's document for the transmission already
+exists and can be ranked. Canonical is the earliest ``started_at``, breaking
+ties on ``call_id`` so both nodes independently reach the same verdict.
+
+The loser keeps its audio — it is ~60 KB and may be the cleaner capture if the
+winner's node had a weak signal — but is excluded from the AI pipeline.
+"""
+from datetime import datetime, timedelta, timezone
+from typing import Awaitable, Callable, Optional
+from app.config import settings
+from app.internal.logger import logger
+
+# Firestore is reached through an injected callable rather than a module-level
+# import. app.internal.firestore initialises firebase-admin at import time,
+# which needs credentials and the SDK present — so importing it here would make
+# this module unimportable in a unit test. Same reasoning as the deferred
+# import in app/internal/auth.py.
+QueryFn = Callable[[str, list], Awaitable[list[dict]]]
+
+
+def _parse_dt(value) -> Optional[datetime]:
+    """Firestore hands back Timestamp, datetime, or ISO string depending on writer."""
+    if not value:
+        return None
+    if isinstance(value, datetime):
+        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def _is_canonical(call: dict, others: list[dict]) -> bool:
+    """True if `call` is the one recording of this transmission that should be processed."""
+    started = _parse_dt(call.get("started_at"))
+    call_id = call.get("call_id") or ""
+    for other in others:
+        other_started = _parse_dt(other.get("started_at"))
+        if not other_started or not started:
+            continue
+        if other_started < started:
+            return False
+        if other_started == started and (other.get("call_id") or "") < call_id:
+            return False
+    return True
+
+
+async def find_duplicate_of(call: dict, query: Optional[QueryFn] = None) -> Optional[str]:
+    """Return the canonical call_id if `call` duplicates another node's recording.
+
+    Returns None when this call is the canonical one, or when there is nothing
+    to compare against (single node in range, or the call lacks the talkgroup
+    and system identifiers the match is keyed on).
+    """
+    system_id = call.get("system_id")
+    talkgroup_id = call.get("talkgroup_id")
+    call_id = call.get("call_id")
+    started = _parse_dt(call.get("started_at"))
+    if not (system_id and talkgroup_id is not None and call_id and started):
+        return None
+
+    if query is None:
+        from app.internal import firestore as fstore
+        query = fstore.collection_where
+
+    window = timedelta(seconds=settings.duplicate_window_seconds)
+    try:
+        # Range-scan on started_at, then filter the rest in Python — Firestore
+        # allows a range on only one field per query.
+        nearby = await query("calls", [
+            ("system_id", "==", system_id),
+            ("started_at", ">=", started - window),
+            ("started_at", "<=", started + window),
+        ])
+    except Exception as e:
+        # Never block an upload on dedup — worst case is the pre-existing
+        # behaviour of processing both copies.
+        logger.warning(f"Duplicate check failed for call {call_id}: {e}")
+        return None
+
+    matches = [
+        c for c in nearby
+        if c.get("call_id") != call_id
+        and c.get("talkgroup_id") == talkgroup_id
+        and c.get("node_id") != call.get("node_id")   # same node twice is a real repeat
+        and not c.get("duplicate_of")                 # never point at another duplicate
+    ]
+    if not matches:
+        return None
+
+    if _is_canonical(call, matches):
+        return None
+
+    canonical = min(
+        matches,
+        key=lambda c: (_parse_dt(c.get("started_at")) or started, c.get("call_id") or ""),
+    )
+    return canonical.get("call_id")
diff --git a/drb-c2-core/app/internal/recorrelation_sweep.py b/drb-c2-core/app/internal/recorrelation_sweep.py
index 521225e..c05b5db 100644
--- a/drb-c2-core/app/internal/recorrelation_sweep.py
+++ b/drb-c2-core/app/internal/recorrelation_sweep.py
@@ -54,6 +54,7 @@ async def _run_sweep_pass() -> None:
         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
         and c.get("corr_sweep_count", 0) < MAX_SWEEP_ATTEMPTS
     ]
 
diff --git a/drb-c2-core/app/internal/transcription.py b/drb-c2-core/app/internal/transcription.py
index 0076f45..6f8513c 100644
--- a/drb-c2-core/app/internal/transcription.py
+++ b/drb-c2-core/app/internal/transcription.py
@@ -5,6 +5,7 @@ Audio is downloaded from GCS then sent to the Whisper API. Falls back to
 returning None on any failure so the intelligence pipeline can still run.
 """
 import asyncio
+import re
 import tempfile
 import os
 from typing import Optional
@@ -14,15 +15,83 @@ from app.internal import firestore as fstore
 # Whisper treats `prompt` as preceding transcript text, not instructions.
 # Writing it as actual radio speech primes the vocabulary toward P25 codes
 # and phrasing before the model hears the audio.
+#
+# DO NOT put an enumerated run of ten-codes in here. The original version of
+# this prompt opened with "10-4. 10-23. 10-20. 10-97. 10-8. ..." and Whisper,
+# treating that as text it should continue, filled noisy or silent audio with
+# sequences like "10-4. 10-5. 10-6. ... 10-99." Those hallucinations sailed
+# straight past the no_speech_prob filter below, because the model is highly
+# confident the continuation it invented is speech. Codes appear here only
+# singly and inside a sentence, where there is no series to extend.
 _WHISPER_PROMPT = (
-    "10-4. 10-23. 10-20. 10-97. 10-8. 10-7. 10-34. 10-50. 10-52. "
-    "Post 4, I'm out. Post 3. En route. On scene. In route. "
-    "Copy. Negative. Stand by. Be advised. Go ahead. "
-    "Units responding. Dispatch. Talkgroup. "
-    "Engine. Ladder. Medic. Rescue. Car. Unit. "
-    "MVA. MVC. Structure fire. Working fire."
+    "Dispatch, go ahead. Copy that, en route. Show me on scene. "
+    "Be advised, units responding. Negative, stand by. "
+    "Post 4, I'm out. Received, thank you. "
+    "Engine and ladder responding to a structure fire. "
+    "Medic on scene with one patient. "
+    "Vehicle accident with injuries, MVA. "
+    "Show me 10-8 and clear."
 )
 
+# Degenerate-output detection (see _is_degenerate). Tuned to catch Whisper's
+# repetition failure mode without discarding terse but real radio traffic.
+_MIN_CODES_FOR_RUN = 6       # ten-codes needed before a run is even considered
+_RUN_RATIO = 0.7             # share of consecutive pairs that must step by +1
+_MIN_SEGMENTS_FOR_REPEAT = 6  # segments needed before repetition is considered
+_UNIQUE_RATIO = 0.25         # unique/total segment texts at or below this is degenerate
+_MAX_PHRASE_REPEATS = 8      # identical consecutive phrase repeats allowed in one blob
+
+
+def _ten_code_run(text: str) -> bool:
+    """True if the text is mostly a counting run of ten-codes.
+
+    Real traffic uses ten-codes constantly, but never in ascending order — a
+    dispatcher does not say "10-4, 10-5, 10-6". An arithmetic series is the
+    signature of Whisper continuing a pattern rather than hearing one.
+    """
+    numbers = [int(n) for n in re.findall(r"\b10-(\d{1,2})\b", text)]
+    if len(numbers) < _MIN_CODES_FOR_RUN:
+        return False
+    steps = [b - a for a, b in zip(numbers, numbers[1:])]
+    ascending = sum(1 for s in steps if s == 1)
+    return steps and (ascending / len(steps)) >= _RUN_RATIO
+
+
+def _phrase_loop(text: str) -> bool:
+    """True if one short phrase repeats far more than speech plausibly would.
+
+    Catches the other repetition mode, e.g. "Dispatch, do you copy?" emitted
+    a dozen times over static.
+    """
+    parts = [p.strip().lower() for p in re.split(r"[.!?]", text) if p.strip()]
+    if len(parts) <= _MAX_PHRASE_REPEATS:
+        return False
+    repeats = 1
+    for prev, cur in zip(parts, parts[1:]):
+        repeats = repeats + 1 if cur == prev else 1
+        if repeats > _MAX_PHRASE_REPEATS:
+            return True
+    return False
+
+
+def _is_degenerate(text: str, segments: list[dict]) -> bool:
+    """True if a transcript looks like Whisper output rather than radio traffic.
+
+    Applied AFTER the per-segment no_speech_prob filter, which does not catch
+    these: the model reports high confidence in text it invented by continuing
+    a pattern, so the only tell is the shape of the output itself.
+    """
+    if not text:
+        return False
+    if _ten_code_run(text) or _phrase_loop(text):
+        return True
+    # Near-identical segments repeated across the whole recording.
+    if len(segments) >= _MIN_SEGMENTS_FOR_REPEAT:
+        normalised = {s["text"].strip().lower() for s in segments}
+        if len(normalised) / len(segments) <= _UNIQUE_RATIO:
+            return True
+    return False
+
 
 async def transcribe_call(
     call_id: str,
@@ -76,7 +145,10 @@ def _sync_transcribe(
 
     if not settings.openai_api_key:
         logger.warning("OPENAI_API_KEY not set — transcription disabled.")
-        return None
+        # Tuple, not a bare None: the caller unpacks two values, so returning
+        # None here raised a TypeError that surfaced as a misleading
+        # "Transcription failed" instead of the real missing-key warning.
+        return None, []
 
     without_scheme = gcs_uri[len("gs://"):]
     bucket_name, blob_path = without_scheme.split("/", 1)
@@ -145,11 +217,17 @@ def _sync_transcribe(
             # in sync.  If every segment was filtered, text becomes None which prevents
             # the intelligence pipeline from running on hallucinated content.
             text = " ".join(s["text"] for s in segments) or None
+            if _is_degenerate(text or "", segments):
+                logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}")
+                return None, []
             return text, segments
         else:
             # json format returns just {"text": "..."} — no segments or timestamps.
             # Intelligence extraction falls back to treating the whole transcript as one block.
             text = (response.text or "").strip() or None
+            if _is_degenerate(text or "", []):
+                logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}")
+                return None, []
             return text, []
     finally:
         try:
diff --git a/drb-c2-core/app/models.py b/drb-c2-core/app/models.py
index 10e8b3d..1b54bb7 100644
--- a/drb-c2-core/app/models.py
+++ b/drb-c2-core/app/models.py
@@ -64,6 +64,7 @@ class CallRecord(BaseModel):
     ended_at: Optional[datetime] = None
     audio_gcs_uri: Optional[str] = None   # canonical gs:// object location
     audio_url: Optional[str] = None       # NOT stored — minted per read, see internal/storage.py
+    duplicate_of: Optional[str] = None    # another node recorded this same transmission first
     transcript: Optional[str] = None      # populated later by STT
     incident_ids: List[str] = []          # one per scene detected in the recording
     location: Optional[Dict[str, float]] = None  # {lat, lng}
diff --git a/drb-c2-core/app/routers/admin.py b/drb-c2-core/app/routers/admin.py
index 307d1f3..a5dcd34 100644
--- a/drb-c2-core/app/routers/admin.py
+++ b/drb-c2-core/app/routers/admin.py
@@ -132,6 +132,7 @@ async def debug_correlation(
         _call_summary(c) for c in recent_calls
         if c.get("status") == "ended"
         and not c.get("incident_ids") and not c.get("incident_id")
+        and not c.get("duplicate_of")  # another node's copy — never meant to correlate
         and c.get("system_id") in ai_systems
     ]
     orphans.sort(key=lambda c: c.get("started_at", ""), reverse=True)
diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py
index dca6ff3..b7e4d63 100644
--- a/drb-c2-core/app/routers/upload.py
+++ b/drb-c2-core/app/routers/upload.py
@@ -2,6 +2,7 @@ from typing import Optional
 from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form, HTTPException, Security
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from app.internal.storage import upload_audio
+from app.internal import dedup
 from app.internal import firestore as fstore
 from app.internal.logger import logger
 from app.config import settings
@@ -57,6 +58,19 @@ async def upload_call_audio(
         except Exception as e:
             logger.warning(f"Could not update call {call_id} with audio_gcs_uri: {e}")
 
+        # Another node in range recorded the same transmission. Keep the audio
+        # (it may be the cleaner capture) but don't transcribe or correlate it
+        # a second time — see app/internal/dedup.py.
+        call_doc = await fstore.doc_get("calls", call_id)
+        duplicate_of = await dedup.find_duplicate_of(call_doc) if call_doc else None
+        if duplicate_of:
+            await fstore.doc_set("calls", call_id, {"duplicate_of": duplicate_of})
+            logger.info(
+                f"Call {call_id} from {node_id} duplicates {duplicate_of} "
+                f"— audio kept, AI pipeline skipped."
+            )
+            return {"url": gcs_uri, "duplicate_of": duplicate_of}
+
         background_tasks.add_task(
             _run_intelligence_pipeline,
             call_id=call_id,
diff --git a/drb-c2-core/tests/test_dedup.py b/drb-c2-core/tests/test_dedup.py
new file mode 100644
index 0000000..0473138
--- /dev/null
+++ b/drb-c2-core/tests/test_dedup.py
@@ -0,0 +1,158 @@
+"""
+Unit tests for cross-node duplicate detection.
+
+Fixture timings come from real production data: node-002 and node-PI-2 both
+recorded TG 9048 on 2026-08-16, starting ~1.1s apart.
+"""
+import pytest
+from datetime import datetime, timezone, timedelta
+from app.internal.dedup import _parse_dt, _is_canonical, find_duplicate_of
+
+BASE = datetime(2026, 8, 16, 19, 31, 46, tzinfo=timezone.utc)
+
+
+def _query_returning(*calls):
+    """Stand-in for fstore.collection_where."""
+    async def _q(_collection, _conditions):
+        return list(calls)
+    return _q
+
+
+def _query_raising(exc):
+    async def _q(_collection, _conditions):
+        raise exc
+    return _q
+
+
+def _call(call_id, node_id, offset_seconds=0.0, talkgroup_id=9048, **extra):
+    return {
+        "call_id": call_id,
+        "node_id": node_id,
+        "system_id": "sys-1",
+        "talkgroup_id": talkgroup_id,
+        "started_at": BASE + timedelta(seconds=offset_seconds),
+        **extra,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Timestamp parsing — Firestore returns three different shapes
+# ---------------------------------------------------------------------------
+
+def test_parse_dt_accepts_aware_datetime():
+    assert _parse_dt(BASE) == BASE
+
+
+def test_parse_dt_assumes_utc_for_naive_datetime():
+    naive = datetime(2026, 8, 16, 19, 31, 46)
+    assert _parse_dt(naive) == BASE
+
+
+def test_parse_dt_accepts_iso_string_with_z():
+    assert _parse_dt("2026-08-16T19:31:46Z") == BASE
+
+
+def test_parse_dt_returns_none_for_junk():
+    assert _parse_dt("not a date") is None
+    assert _parse_dt(None) is None
+
+
+# ---------------------------------------------------------------------------
+# Canonical selection
+# ---------------------------------------------------------------------------
+
+def test_earlier_start_wins():
+    early = _call("a", "node-002", 0.0)
+    late = _call("b", "node-PI-2", 1.1)
+    assert _is_canonical(early, [late]) is True
+    assert _is_canonical(late, [early]) is False
+
+
+def test_identical_starts_break_tie_on_call_id():
+    first = _call("aaa", "node-002", 0.0)
+    second = _call("bbb", "node-PI-2", 0.0)
+    assert _is_canonical(first, [second]) is True
+    assert _is_canonical(second, [first]) is False
+
+
+def test_both_nodes_reach_the_same_verdict():
+    """The whole point: the decision must not depend on upload order."""
+    a = _call("a", "node-002", 0.0)
+    b = _call("b", "node-PI-2", 1.1)
+    verdicts = [_is_canonical(a, [b]), _is_canonical(b, [a])]
+    assert verdicts.count(True) == 1, "exactly one recording must be canonical"
+
+
+# ---------------------------------------------------------------------------
+# find_duplicate_of
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_returns_canonical_id_for_later_recording():
+    canonical = _call("canon", "node-002", 0.0)
+    later = _call("later", "node-PI-2", 1.1)
+    q = _query_returning(canonical, later)
+    assert await find_duplicate_of(later, query=q) == "canon"
+
+
+@pytest.mark.asyncio
+async def test_returns_none_for_the_canonical_recording():
+    canonical = _call("canon", "node-002", 0.0)
+    later = _call("later", "node-PI-2", 1.1)
+    q = _query_returning(canonical, later)
+    assert await find_duplicate_of(canonical, query=q) is None
+
+
+@pytest.mark.asyncio
+async def test_same_node_is_never_a_duplicate():
+    """Back-to-back transmissions from one node are real, separate calls."""
+    first = _call("a", "node-002", 0.0)
+    second = _call("b", "node-002", 2.0)
+    q = _query_returning(first, second)
+    assert await find_duplicate_of(second, query=q) is None
+
+
+@pytest.mark.asyncio
+async def test_different_talkgroup_is_not_a_duplicate():
+    other_tg = _call("a", "node-002", 0.0, talkgroup_id=9600)
+    mine = _call("b", "node-PI-2", 1.0, talkgroup_id=9048)
+    q = _query_returning(other_tg, mine)
+    assert await find_duplicate_of(mine, query=q) is None
+
+
+@pytest.mark.asyncio
+async def test_never_chains_onto_another_duplicate():
+    """A third node must point at the original, not at a duplicate of it."""
+    canonical = _call("canon", "node-002", 0.0)
+    already_dupe = _call("dupe", "node-PI-2", 0.5, duplicate_of="canon")
+    third = _call("third", "node-003", 1.0)
+    q = _query_returning(canonical, already_dupe, third)
+    assert await find_duplicate_of(third, query=q) == "canon"
+
+
+@pytest.mark.asyncio
+async def test_no_match_returns_none():
+    lonely = _call("only", "node-002", 0.0)
+    q = _query_returning(lonely)
+    assert await find_duplicate_of(lonely, query=q) is None
+
+
+@pytest.mark.asyncio
+async def test_missing_identifiers_skip_the_check():
+    called = False
+
+    async def _q(_collection, _conditions):
+        nonlocal called
+        called = True
+        return []
+
+    incomplete = {"call_id": "x", "node_id": "node-002", "started_at": BASE}
+    assert await find_duplicate_of(incomplete, query=_q) is None
+    assert called is False, "must bail out before querying"
+
+
+@pytest.mark.asyncio
+async def test_query_failure_never_blocks_the_upload():
+    call = _call("a", "node-002", 0.0)
+    q = _query_raising(RuntimeError("firestore down"))
+    assert await find_duplicate_of(call, query=q) is None