diff --git a/core/application/delivery_reconciler.py b/core/application/delivery_reconciler.py
index f1396e7c..da95f7c5 100644
--- a/core/application/delivery_reconciler.py
+++ b/core/application/delivery_reconciler.py
@@ -32,6 +32,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import time
 from typing import Any, Dict, List, Optional
 
@@ -353,7 +354,15 @@ async def _check_unknown_deliveries(self, owner_user_id: Optional[int]) -> List[
         A send that timed out is marked UNKNOWN because neither success nor
         failure is known. Staying UNKNOWN forever is the one state with no
         recovery path of its own, so it must be visible to an operator.
+
+        When the Go Worker owns delivery (ATF_WORKER_ENABLED=1) this check
+        stands down: the worker's own reconciler is the single authority
+        over UNKNOWN fences/jobs, and re-flagging the same rows here every
+        sweep is a second voice over state it does not own — production ran
+        two competing reconcilers, one deciding and one repeating.
         """
+        if str(os.environ.get("ATF_WORKER_ENABLED", "0")).strip().lower() in ("1", "true", "yes", "on"):
+            return []
         scope, params = await self._scope(owner_user_id)
         db = self._db()
         if db is None:
diff --git a/core/infrastructure/persistence/postgres_repositories.py b/core/infrastructure/persistence/postgres_repositories.py
index 58db556e..de160d04 100644
--- a/core/infrastructure/persistence/postgres_repositories.py
+++ b/core/infrastructure/persistence/postgres_repositories.py
@@ -85,6 +85,21 @@ def _safe_dict(val: Any, default: dict) -> dict:
     return default
 
 
+def _replay_job_kind(payload_json: str) -> str:
+    """Derive the delivery_jobs.job_kind a dead-letter replay must rebuild.
+
+    The business key on delivery_jobs includes job_kind, so the replay has to
+    restore the SAME deliverable it dead-lettered: a keyword_reply entry must
+    not come back as kind=forward (it would also let the kind-less lookup
+    reset a forward sibling that shares rule/source/message/target).
+    """
+    try:
+        raw = json.loads(payload_json) if isinstance(payload_json, str) else (payload_json or {})
+    except Exception:
+        raw = {}
+    return "keyword_reply" if isinstance(raw, dict) and raw.get("is_reply") else "forward"
+
+
 def _format_sql(query: str, db: Any) -> str:
     is_asyncpg = "asyncpg" in type(db).__module__ or hasattr(db, "_protocol")
     if is_asyncpg and "%s" in query:
@@ -1323,14 +1338,21 @@ async def replay_dead_letter(
         # delivery_jobs is UNIQUE(rule_id, source_chat_id, source_message_id,
         # target_chat_id), so a replay wakes the one delivery up instead of
         # adding a second: that is what makes it unable to double-post.
+        # The delivery_jobs business key includes job_kind: a replay must
+        # wake its OWN deliverable, never reset a keyword_reply sibling that
+        # happens to share (rule, source, message, target).
+        payload = entry.get("payload_json") or "{}"
+        if isinstance(payload, (dict, list)):
+            payload = json.dumps(payload)
+        replay_kind = _replay_job_kind(payload)
         original = None
         if job_id:
             original = await self._fetch_one("SELECT * FROM delivery_jobs WHERE id=%s", (job_id,))
         if original is None:
             original = await self._fetch_one(
                 "SELECT * FROM delivery_jobs WHERE rule_id=%s AND source_chat_id=%s"
-                " AND source_message_id=%s AND target_chat_id=%s",
-                (rule_id, src_chat, src_msg, tgt_chat),
+                " AND source_message_id=%s AND target_chat_id=%s AND job_kind=%s",
+                (rule_id, src_chat, src_msg, tgt_chat, replay_kind),
             )
 
         now = int(time.time())
@@ -1352,20 +1374,24 @@ async def replay_dead_letter(
                 )
             else:
                 max_attempts = max(1, int(entry.get("attempts") or 0) or DeliveryJob().max_attempts)
-                payload = entry.get("payload_json") or "{}"
-                if isinstance(payload, (dict, list)):
-                    payload = json.dumps(payload)
+                # The business key is UNIQUE(rule_id, source_chat_id,
+                # source_message_id, target_chat_id, job_kind): the conflict
+                # target and the inserted row must carry job_kind, otherwise
+                # PostgreSQL rejects the whole INSERT with "no unique or
+                # exclusion constraint matching the ON CONFLICT specification"
+                # and every replay of a purged row fails (and a keyword_reply
+                # entry would silently rebuild itself as kind=forward).
                 await _execute_db(
                     self._db,
                     "INSERT INTO delivery_jobs ("
                     " id, rule_id, source_chat_id, source_message_id, target_chat_id,"
-                    " payload_json, status, attempts, max_attempts, next_retry_at,"
+                    " job_kind, payload_json, status, attempts, max_attempts, next_retry_at,"
                     " lease_until, worker_id, error_detail, created_at, updated_at"
-                    ") VALUES (%s, %s, %s, %s, %s, %s, 'PENDING', 0, %s, 0, 0, '', '', %s, %s)"
-                    " ON CONFLICT (rule_id, source_chat_id, source_message_id, target_chat_id) DO NOTHING",
+                    ") VALUES (%s, %s, %s, %s, %s, %s, %s, 'PENDING', 0, %s, 0, 0, '', '', %s, %s)"
+                    " ON CONFLICT (rule_id, source_chat_id, source_message_id, target_chat_id, job_kind) DO NOTHING",
                     (
                         job_id or f"replay_{entry_id}", rule_id, src_chat, src_msg, tgt_chat,
-                        payload, max(1, max_attempts), now, now,
+                        replay_kind, payload, max(1, max_attempts), now, now,
                     ),
                 )
                 # Restore the placeholder mapping row the enqueue path
diff --git a/core/infrastructure/persistence/sqlite_repositories.py b/core/infrastructure/persistence/sqlite_repositories.py
index a060a859..4054d9f0 100644
--- a/core/infrastructure/persistence/sqlite_repositories.py
+++ b/core/infrastructure/persistence/sqlite_repositories.py
@@ -141,6 +141,21 @@ def _safe_list(raw, default=None):
         return default or []
 
 
+def _replay_job_kind(payload_json: str) -> str:
+    """Derive the delivery_jobs.job_kind a dead-letter replay must rebuild.
+
+    The business key on delivery_jobs includes job_kind, so the replay has to
+    restore the SAME deliverable it dead-lettered: a keyword_reply entry must
+    not come back as kind=forward (it would also let the kind-less lookup
+    reset a forward sibling that shares rule/source/message/target).
+    """
+    try:
+        raw = json.loads(payload_json) if isinstance(payload_json, str) else (payload_json or {})
+    except Exception:
+        raw = {}
+    return "keyword_reply" if isinstance(raw, dict) and raw.get("is_reply") else "forward"
+
+
 def _rule_meta_json(rule) -> str:
     """Serialise rule metadata, mirroring name/description into it.
 
@@ -2407,14 +2422,18 @@ async def replay_dead_letter(
         # deliberate dedup, not an accident. So a replay can never be a second
         # row — it either wakes the existing one up or is refused. That
         # constraint is also why a replay cannot double-post.
+        # The delivery_jobs business key includes job_kind: a replay must
+        # wake its OWN deliverable, never reset a keyword_reply sibling that
+        # happens to share (rule, source, message, target).
+        replay_kind = _replay_job_kind(payload)
         original = None
         if job_id:
             original = self._db.query_one("SELECT * FROM delivery_jobs WHERE id=?", (job_id,))
         if original is None:
             original = self._db.query_one(
                 "SELECT * FROM delivery_jobs WHERE rule_id=? AND source_chat_id=?"
-                " AND source_message_id=? AND target_chat_id=?",
-                (rule_id, src_chat, src_msg, tgt_chat),
+                " AND source_message_id=? AND target_chat_id=? AND job_kind=?",
+                (rule_id, src_chat, src_msg, tgt_chat, replay_kind),
             )
 
         now = _now()
@@ -2482,14 +2501,15 @@ def _unit(conn) -> bool:
                     """
                     INSERT OR IGNORE INTO delivery_jobs (
                         id, rule_id, source_chat_id, source_message_id, target_chat_id,
-                        payload_json, status, attempts, max_attempts, next_retry_at,
+                        job_kind, payload_json, status, attempts, max_attempts, next_retry_at,
                         lease_until, worker_id, error_detail, created_at, updated_at,
                         intermediate_chat_id, intermediate_message_id, delivery_stage
-                    ) VALUES (?, ?, ?, ?, ?, ?, 'PENDING', 0, ?, 0, 0, '', '', ?, ?, NULL, NULL, 'DIRECT')
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, 'PENDING', 0, ?, 0, 0, '', '', ?, ?, NULL, NULL, 'DIRECT')
                     """,
                     (
                         job.id, job.rule_id, str(job.source_chat_id), int(job.source_message_id),
-                        str(job.target_chat_id), SqliteDatabase.dumps(job.payload_data),
+                        str(job.target_chat_id), replay_kind,
+                        SqliteDatabase.dumps(job.payload_data),
                         int(job.max_attempts), int(job.created_at), int(job.updated_at),
                     ),
                 )
diff --git a/core/tests/test_delivery_reconciliation.py b/core/tests/test_delivery_reconciliation.py
index 57133b85..0c4e998f 100644
--- a/core/tests/test_delivery_reconciliation.py
+++ b/core/tests/test_delivery_reconciliation.py
@@ -248,6 +248,24 @@ async def test_an_unresolved_unknown_delivery_is_flagged(db, queue, mmap, rules,
     assert "timed out" in unknown[0]["error_detail"], unknown[0]["error_detail"]
 
 
+async def test_unknown_deliveries_stand_down_when_go_worker_owns_delivery(
+    db, queue, mmap, rules, attempts, gaps, monkeypatch
+):
+    """ATF_WORKER_ENABLED=1: the Go Worker's reconciler owns UNKNOWN state.
+
+    Flagging Go-owned rows here every sweep made Python a second, repeating
+    voice over state it does not resolve (production warned about the same 3
+    rows every sweep while the Go reconciler legitimately kept them UNKNOWN).
+    """
+    await rules.add(make_rule())
+    _job(db, status="UNKNOWN", error_detail="timed out")
+    monkeypatch.setenv("ATF_WORKER_ENABLED", "1")
+    r = await _reconciler(queue, mmap, rules, attempts, gaps).reconcile()
+    assert r["checks"]["unknown_deliveries"] == [], r["checks"]["unknown_deliveries"]
+    # The other checks keep working — only the Go-owned one stands down.
+    assert "orphan_mappings" in r["checks"]
+
+
 # --------------------------------------------------------------------------- #
 # Repeated failures per destination / per source
 # --------------------------------------------------------------------------- #
diff --git a/docs/reliability/phase2-progress-ledger.md b/docs/reliability/phase2-progress-ledger.md
new file mode 100644
index 00000000..a0c5fa17
--- /dev/null
+++ b/docs/reliability/phase2-progress-ledger.md
@@ -0,0 +1,44 @@
+# NSE Phase 2 — Progress Ledger (CSV finding IDs → outcome)
+
+Branch `fix/delivery-dup-and-ownership` · baseline master `fcc0dc1` (PR #340) · round 2026-11-06.
+Production evidence was collected **read-only** (sqlite `.backup` snapshot + live queries); the one
+production mutation is item F-01 (executed under the runbook's safeguards, verified delivered).
+Raw per-record exports (server-db-audit.csv / server-db-affected-records.csv) are kept OUT of the
+repository — they are production evidence, not source.
+
+| Finding | Severity | Evidence basis | Fix | Status |
+|---|---|---|---|---|
+| F-01 12× `PHOTO_EXT_INVALID` FAILED (attempts=0) | P0 | snapshot + live dry-run gates 12/0/0 | conditional fail-closed requeue (runbook §4) | **EXECUTED 2026-11-06 — 12/12 delivered exactly-once, 0 WARN/ERROR** |
+| F-02 3× UNKNOWN `e2e-go*` (naming: e2e-go/e2e-go2/e2e-go3) | P2 | snapshot; prior ledger pattern `e2e-go-%` matched only 1 of 3 | corrected explicit-ID DELETE (runbook §7) | PREPARED — irreversible, awaiting authorization |
+| F-03 127 orphan SENT fences (no retention) | P2 | snapshot | `PurgeAgedFences` (terminal-only, UNKNOWN never swept) + hourly sweep | IMPLEMENTED + tested; production purge starts post-deploy |
+| F-04 45 orphan mappings (15 in-window) | P3 | snapshot | recurrence prevented by `run_matrix` namespace cleanup (#339); residue deletion prepared | PREPARED (residue) — awaiting authorization |
+| F-05 6 SENT-without-fence rows (dual write authority) | P2 | snapshot | single-owner architecture: worker enabled, Python dispatch dormant, fence ordering | VERIFIED-SOUND for current runtime; rows kept as history |
+| F-07 webhook add-on enabled on 1 rule, no engine (G2) | P3 | snapshot `forward_rules.metadata` | implement-or-enforce (product decision) | TRACKED (PR #339 ledger §6.3) |
+| F-08 Python reconciler repeats Go-owned UNKNOWN WARNs | P3 | live logger post-deploy delta | `_check_unknown_deliveries` stands down at `ATF_WORKER_ENABLED=1` | IMPLEMENTED + tested |
+| F-09 SQLite fence writers unguarded (SENT overwritable) | P1 | Agent-B sequence (needs F-10) | `status <> 'SENT'` guards, parity with PG + Python | IMPLEMENTED + tested |
+| F-10 same-worker fence claim ignores lease (double dispatch) | P1 | Agent-B exact sequence | FenceClaim pre-read state machine (sqlite+pg), FAILED re-claim preserved | IMPLEMENTED + tested (sqlite + real PG) |
+| F-11 ambiguous album group copy → per-item re-upload dup; LAST-id mapping; opt-out ignored | P1 | Agent-B sequence + code read | IsAmbiguous guard, first-item mapping id, opt-out honored | IMPLEMENTED + tested |
+| F-12 bot-route media refusal → permanent loss (no USER rescue) | P1 | Agent-B sequence + legacy parity | `botMediaRefused` USER fallback (opt-in aware) | IMPLEMENTED + tested |
+| F-13 staged media never deleted; MediaQuotaBytes dead | P2 | grep + compose topology | `defer os.Remove` all outcomes; fake writes real files to test | IMPLEMENTED + tested |
+| F-14 PG DLQ replay dead code (4-col ON CONFLICT vs 5-col key) | P2 | Agent-B + schema read | 5-col conflict target + job_kind in INSERT/lookup (both stores) | IMPLEMENTED + tested on real PG |
+| F-15 reconciler strands dial-phase Unavailable | P2 | Agent-B both sequences + prod rows | `Error while dialing` marker; bare Unavailable stays UNKNOWN | IMPLEMENTED + tested |
+| F-16 16 `mtx-*` SENT residue rows | P3 | snapshot | none — ages out at 7d (~Nov 11) | NO-ACTION (self-resolving) |
+| F-17 timezone-dependent autoreply gate test | P3 | red on +0330 host, green on UTC CI | derive window from the account-offset clock | IMPLEMENTED + tested |
+| F-19 stale stores beside live DBs (logger.db, backup-preclear) | P3 | volume listing (PR #339 §7, still true) | one-time removal | PREPARED — awaiting authorization |
+
+## Verified-sound this round (no action)
+
+- Preflight-before-fence ordering; `FenceAlreadySent` close-out; atomic `ReconcileUnknown` both stores;
+  PG claim fails closed; split-album resume; DLQ write-once; runtime bot-token ownership checks;
+  keyword-reply mapping isolation; no byte corruption in the extension chain (Agent A/B negative space).
+- Post-deploy logger delta (Nov 5 20:45+): only 4 benign WARN lines — the #339 log-collapse prediction held.
+- Python suite 4292 pass / 27 skip (1 environmental `is_checkout` failure: no real `.git` in export; CI green).
+- Go suite + `-race` green; PG queue suite green against real PostgreSQL 17.
+
+## Awaiting authorization (all prepared, none executed)
+
+1. F-02 artifact deletion (corrected 3-ID list).
+2. F-04 residue mapping cleanup (45 rows) — optional; aged ones are by-design retention.
+3. F-19 stale-store file removal.
+4. Merge/deploy of this PR (deploy also starts the F-03 retention sweep in production).
+5. Phase-6 live re-upload test (docs/reliability/phase6-live-test-spec.md).
diff --git a/docs/reliability/phase4-recovery-runbook.md b/docs/reliability/phase4-recovery-runbook.md
new file mode 100644
index 00000000..e60993c7
--- /dev/null
+++ b/docs/reliability/phase4-recovery-runbook.md
@@ -0,0 +1,158 @@
+# Phase 4 — Production Recovery Runbook (PREPARED, NOT EXECUTED)
+
+**Status: dry-run validated against snapshot `/tmp/atf-phase2-snap.db` (taken 2026-11-06 22:58 local, read-only `.backup`). No production write has been performed or will be performed without explicit authorization.**
+
+## 1. Revalidated current state (2026-11-06 snapshot)
+
+- Exactly **12** rows match `error_detail LIKE '%PHOTO_EXT_INVALID%'` — unchanged from the historical count.
+- All 12: `status='FAILED'`, `attempts=0`, `max_attempts=5`.
+- Rule `40edad5a-a9b2-45f9-8eca-1f6f65619c64`: `is_active=1`, target `-1001854636317` (unchanged).
+- Delivery fences (delivery_attempts) for all 12 source messages: `status='FAILED'`, `target_message_id IS NULL` → Telegram never accepted a send (server-side rejection before persist).
+- message_map for all 12: `delivery_status='PENDING'`, `target_message_id IS NULL` → no delivery side effect recorded.
+- Delta vs Nov 05: **none** for this group (the 10 rule-family UNKNOWN rows WERE reconciled to FAILED post-deploy — reconciler confirmed working in production; their requeue from the old ledger §5(b) remains a separate user-gated decision, out of this task's scope).
+
+### The 12 exact job IDs
+
+```
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18321:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18323:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18326:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18332:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18336:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18340:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18347:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18355:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18361:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18362:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18366:-1001854636317
+40edad5a-a9b2-45f9-8eca-1f6f65619c64:-1001957768371:18375:-1001854636317
+```
+
+(msg ids 18321–18375, created Nov 05 09:18–14:35, source `-1001957768371` → target `-1001854636317`)
+
+## 2. Safety properties (why requeue is duplicate-safe)
+
+Each job's whole failure history is `400 PHOTO_EXT_INVALID` from `messages.SendMedia` — a **server-side rejection before persist**: Telegram never accepted any upload for these. Fences confirm (`target_message_id` NULL), mappings confirm (NULL/PENDING), `attempts=0`. The fixed pipeline (worker `fcc0dc1`: stagedExtension + target-side-rejection fallback) will stage with the real extension and deliver exactly once.
+
+## 3. Dry-run mode (run first — makes zero writes)
+
+```sql
+-- DRY-RUN A: expected exactly 12 rows, all preconditions true
+SELECT j.id, j.status, j.attempts,
+       (SELECT a.target_message_id FROM delivery_attempts a
+         WHERE a.source_chat_id = j.source_chat_id
+           AND a.source_message_id = j.source_message_id
+           AND a.target_chat_id    = j.target_chat_id) AS fence_tmsg,
+       (SELECT m.target_message_id FROM message_map m
+         WHERE m.source_chat_id = j.source_chat_id
+           AND m.source_message_id = j.source_message_id
+           AND m.target_chat_id    = j.target_chat_id) AS map_tmsg
+FROM delivery_jobs j
+WHERE j.error_detail LIKE '%PHOTO_EXT_INVALID%'
+  AND j.status = 'FAILED' AND j.attempts = 0;
+
+-- DRY-RUN B (fail-closed gate): must return exactly 12 and 12
+SELECT
+  (SELECT COUNT(*) FROM delivery_jobs
+    WHERE error_detail LIKE '%PHOTO_EXT_INVALID%'
+      AND status='FAILED' AND attempts=0)                                   AS eligible,
+  (SELECT COUNT(*) FROM delivery_jobs
+    WHERE error_detail LIKE '%PHOTO_EXT_INVALID%'
+      AND status='FAILED' AND attempts=0
+      AND EXISTS (SELECT 1 FROM delivery_attempts a
+                  WHERE a.source_chat_id=delivery_jobs.source_chat_id
+                    AND a.source_message_id=delivery_jobs.source_message_id
+                    AND a.target_chat_id=delivery_jobs.target_chat_id
+                    AND a.target_message_id IS NOT NULL))                   AS fence_blockers,
+  (SELECT COUNT(*) FROM delivery_jobs
+    WHERE error_detail LIKE '%PHOTO_EXT_INVALID%'
+      AND status='FAILED' AND attempts=0
+      AND EXISTS (SELECT 1 FROM message_map m
+                  WHERE m.source_chat_id=delivery_jobs.source_chat_id
+                    AND m.source_message_id=delivery_jobs.source_message_id
+                    AND m.target_chat_id=delivery_jobs.target_chat_id
+                    AND m.target_message_id IS NOT NULL))                   AS map_blockers;
+-- ABORT unless eligible=12, fence_blockers=0, map_blockers=0
+```
+
+Snapshot result of dry-run: **eligible=12, fence_blockers=0, map_blockers=0** ✅
+
+## 4. Apply mode (CONDITIONAL — every row guarded individually; REQUIRES AUTHORIZATION)
+
+```sql
+BEGIN IMMEDIATE;
+
+CREATE TEMP TABLE _recovery_rows AS
+  SELECT id, status, error_detail, updated_at FROM delivery_jobs
+  WHERE error_detail LIKE '%PHOTO_EXT_INVALID%'
+    AND status='FAILED' AND attempts=0
+    AND NOT EXISTS (SELECT 1 FROM delivery_attempts a
+                    WHERE a.source_chat_id=delivery_jobs.source_chat_id
+                      AND a.source_message_id=delivery_jobs.source_message_id
+                      AND a.target_chat_id=delivery_jobs.target_chat_id
+                      AND a.target_message_id IS NOT NULL)
+    AND NOT EXISTS (SELECT 1 FROM message_map m
+                    WHERE m.source_chat_id=delivery_jobs.source_chat_id
+                      AND m.source_message_id=delivery_jobs.source_message_id
+                      AND m.target_chat_id=delivery_jobs.target_chat_id
+                      AND m.target_message_id IS NOT NULL);
+
+-- fail-closed: exactly 12 or roll everything back
+SELECT CASE WHEN (SELECT COUNT(*) FROM _recovery_rows) = 12
+            THEN 'OK-PROCEED' ELSE (SELECT RAISE(ABORT, 'precondition count mismatch')) END;
+
+UPDATE delivery_jobs
+SET status='PENDING', next_retry_at=0, lease_until=0, worker_id='',
+    error_detail = error_detail || ' | requeued-by:phase2-runbook-2026-11-06',
+    updated_at=strftime('%s','now')
+WHERE id IN (SELECT id FROM _recovery_rows)
+  AND status='FAILED' AND attempts=0;   -- re-check under the transaction
+
+SELECT changes();  -- must be 12, otherwise ROLLBACK
+COMMIT;
+```
+
+Notes:
+- `error_detail` is **appended**, not blanked — exhausted-operation evidence preserved (no evidence destruction).
+- The UPDATE predicate re-checks status/attempts under the write lock → no race with the worker.
+- Rows re-enter the normal claim path; the fixed worker applies the extension staging path.
+
+## 5. Post-apply verification
+
+```sql
+SELECT status, COUNT(*) FROM delivery_jobs
+ WHERE id IN (SELECT id FROM _recovery_rows) GROUP BY status;   -- expect 12 PENDING (then SENT as worker drains)
+-- later: no PHOTO_EXT_INVALID re-failures, no duplicate message_map rows:
+SELECT COUNT(*) FROM message_map
+ WHERE source_chat_id='-1001957768371'
+   AND source_message_id IN (18321,18323,18326,18332,18336,18340,18347,18355,18361,18362,18366,18375)
+   AND target_message_id IS NOT NULL;                            -- expect ≤ 12, one per source
+```
+
+## 6. Rollback
+
+Full-file rollback: `cp /tmp/atf-phase2-snap.db`-style pre-apply snapshot (take a fresh `.backup` immediately before apply). Row-level rollback:
+
+```sql
+UPDATE delivery_jobs SET status='FAILED', next_retry_at=0, lease_until=0, updated_at=strftime('%s','now')
+WHERE id IN (SELECT id FROM _recovery_rows)
+  AND status='PENDING' AND error_detail LIKE '%requeued-by:phase2-runbook-2026-11-06%';
+```
+
+(The original `error_detail` text is preserved by the append-only requeue marker.)
+
+## 7. Appendix A — corrected artifact-removal SQL (Phase 5, DO NOT RUN — no authorization)
+
+**CRITICAL CORRECTION vs the PR #339 ledger §5(c):** the 3 artifacts are `e2e-go-1793747456`, `e2e-go2-1793747663`, `e2e-go3-1793747953`. The old pattern `id LIKE 'e2e-go-%'` matches **only 1 of 3** (`e2e-go2-…`/`e2e-go3-…` fail the literal `-` after `e2e-go`). Corrected form (explicit ID list — no wildcards):
+
+```sql
+DELETE FROM delivery_jobs WHERE id IN (
+  'e2e-go-1793747456','e2e-go2-1793747663','e2e-go3-1793747953'
+) AND status='UNKNOWN';   -- expect changes()=3
+```
+
+Fence rows for source messages 999997-999999 are intentionally **not** touched (fence modification is outside authorization); they age out of the reconciler's in-window orphan checks and are already outside the mapping-retention asymmetry (no message_map rows exist for them).
+
+## 8. Appendix B — observed (not actionable here)
+
+16 `mtx-*` SENT rows in production DB, newest 2026-11-04 19:26 — live-fire matrix residue predating the run_matrix namespace guard (6274c96). All SENT, no failure state; report-only.
diff --git a/docs/reliability/phase5-artifact-manifest.md b/docs/reliability/phase5-artifact-manifest.md
new file mode 100644
index 00000000..ea94f0a7
--- /dev/null
+++ b/docs/reliability/phase5-artifact-manifest.md
@@ -0,0 +1,31 @@
+# Phase 5 — e2e-go Artifact Cleanup Manifest (INVESTIGATION ONLY — NOTHING DELETED)
+
+Snapshot basis: `/tmp/atf-phase2-snap.db` (2026-11-06 22:58) + live worker container inspection.
+
+**Naming correction (affects the old recovery SQL):** the historical label "3 × `e2e-go-*`" hides that the actual IDs use three different prefixes — `e2e-go-`, `e2e-go2-`, `e2e-go3-`. The prepared SQL in `docs/reliability/pr339-audit-ledger.md` §5(c) (`id LIKE 'e2e-go-%'`) would remove only artifact #1. See runbook Appendix A for the corrected explicit-ID statement.
+
+## Per-artifact manifest
+
+| # | Job ID | Created | Error class | rule_id | Src msg | Fence (delivery_attempts) | message_map | Physical staged file (SHA-256) | diagnostic_events |
+|---|---|---|---|---|---|---|---|---|---|
+| 1 | `e2e-go-1793747456` | 2026-11-03 23:10:56 | `Unavailable: dial tcp …` (gRPC conn refused) | `ea565d5c-…` (test rule) | 999999 | status=UNKNOWN, target_message_id=NULL, attempts=0, worker `a168ab7e95e1-1` | none | **absent** — `/tmp/atf-media` does not exist in atf-worker | 0 rows |
+| 2 | `e2e-go2-1793747663` | 2026-11-03 23:14:23 | same | same | 999998 | same pattern | none | **absent** | 0 rows |
+| 3 | `e2e-go3-1793747953` | 2026-11-03 23:19:13 | same | same | 999997 | same pattern | none | **absent** | 0 rows |
+
+## Assessment
+
+1. **Provably no Telegram side effect:** the RPC dial never connected (`Unavailable`, transport-level, before any handler). `attempts=0`, fence `target_message_id=NULL`, zero message_map rows.
+2. **No physical artifact exists:** the media staging directory is absent in the worker container; there is nothing to hash and nothing retaining bytes. SHA-256: n/a (artifact is the DB row itself).
+3. **No referencing records:** DLQ empty; diagnostic_events 0 rows for all three (job-id and source-message lookups); only the fences reference the synthetic source messages 999997–999999.
+4. **Forensic value:** none beyond the row itself — the failure class is fully recorded in `error_detail`, and the run_matrix seeding path that produced them was already fixed (6274c96 namespace guard refuses non-`mtx-*` ids).
+5. **Rule context:** parent rule `ea565d5c-…` is the live-fire test rule (52 SENT + 4 FEN-skip FAILED + these 3 UNKNOWN); it remains active production data — out of scope.
+
+## Recommendation per artifact
+
+| Artifact | Recommendation |
+|---|---|
+| `e2e-go-1793747456` | **Safe to remove after authorization** (corrected SQL, runbook Appendix A) |
+| `e2e-go2-1793747663` | **Safe to remove after authorization** |
+| `e2e-go3-1793747953` | **Safe to remove after authorization** |
+
+No deletion performed. Fence rows for 999997–999999 are left untouched (fence modification not authorized; they age out of in-window orphan checks naturally). Related-but-out-of-scope: 16 `mtx-*` SENT rows are separate live-fire residue — report-only.
diff --git a/docs/reliability/phase6-live-test-spec.md b/docs/reliability/phase6-live-test-spec.md
new file mode 100644
index 00000000..1492af0c
--- /dev/null
+++ b/docs/reliability/phase6-live-test-spec.md
@@ -0,0 +1,90 @@
+# Phase 6 — Controlled Telegram Live-Test Specification (PREPARED — NOT EXECUTED)
+
+**No real Telegram message is sent, forwarded, copied or re-uploaded by this task.** Execution requires separate explicit authorization. This document is the exact execution procedure for that future authorized run.
+
+## 0. Objective
+
+Prove the end-to-end `PHOTO_EXT_INVALID` recovery path against real Telegram: a media message whose copy is rejected by the target is staged with its real extension, re-uploaded via `messages.SendMedia`, delivered exactly once, and recorded correctly (fence → mapping → job).
+
+## 1. Test assets & prerequisites
+
+| Item | Requirement |
+|---|---|
+| Worker session | The production worker's USER session (`sessions` table, healthy, not flood-limited). Verify `GET /healthz` → `db_connected=true`, session ready. |
+| Source | A dedicated test channel owned by the operator (NOT `-1001957768371`; never a private user chat). Post the probe media here. |
+| Target | A second dedicated test channel where the worker's account is admin (NOT `-1001854636317`). |
+| Rule | Create rule `id=<test-rule-uuid>` source→target, `delivery_transport=USER` (default), **`fallback_enabled=true`**, no filters, no FEN, `content_mode=AUTO`, `album_mode` fixed per cell. |
+| Isolation | Use the live-fire matrix namespace discipline (`worker/test/live/run_matrix.py` clone/drop helpers, `mtx-` prefix) or an equivalent dedicated rule id — production jobs must remain untouched. |
+| Version gate | Worker must report `commit_sha` of the PR head that ships this phase's fixes (verify `docker logs atf-worker \| jq -r .commit_sha`). |
+
+## 2. Probe media
+
+| Cell | Media type | Original filename | Expected runtime `GetFileExtension` | Expected staged name |
+|---|---|---|---|---|
+| L1 | photo (JPEG) | n/a (MTProto has none) | `.jpg` (adapter photo branch) | `<jobID>.jpg` → `FileName job-<id>-media.jpg` |
+| L2 | document (real `.Mp4` uppercase name) | `clip.Mp4` | `.Mp4` (raw suffix passthrough) | normalized `.mp4` |
+| L3 | document, nameless | (no file_name) | `""` | bare `<jobID>` → `send_document` (extension-less is correct for documents) |
+| L4 | video note | n/a | `""` → kind fallback | `.mp4` → `send_video` |
+| L5 | album ×3 photos (album_mode=album) | n/a | group copy path | group `CopyMediaGroup`; no per-item re-upload on success |
+
+All probe assets are uploaded fresh by the operator into the source channel; no production content, no private messages.
+
+## 3. Expected RPC sequence per cell (L1 canonical)
+
+1. Ingestion → enqueue (`delivery_jobs` PENDING, `message_map` PENDING).
+2. Worker claim → `FenceClaim` (delivery_attempts CLAIMED).
+3. `CopyMessage` → Telegram returns `400 PHOTO_EXT_INVALID` → classified `IsMediaRejectedByTarget` → `NeedsReupload()=true`.
+4. `fallback_enabled=true` → `reuploadMedia`: `GetFileExtension` → `.jpg` → `DownloadMedia` to `/tmp/atf-media/<jobID>.jpg` → `PrepareMediaUpload` (token == staged path) → `SendMedia` → **Telegram accepts**, returns Message id.
+5. Persist: `FenceRecordSent(tmsg)` → `NoteMapping` → `MarkSent` → job SENT, mapping SENT with target id.
+6. Staged file `/tmp/atf-media/<jobID>.jpg` **removed** (this phase's cleanup fix — assert absent).
+
+## 4. Expected database transitions (per cell)
+
+- `delivery_jobs`: exactly one row, final `SENT`, `attempts=1`, `error_detail` empty (or unchanged), `worker_id` set.
+- `delivery_attempts`: one row `SENT` with `target_message_id` = accepted id; no second row.
+- `message_map`: one row `SENT`, `target_message_id` non-null, single row per source (album: one row keyed by media_group_id).
+- No `dead_letter_queue` rows; no `UNKNOWN` states at any point; no diagnostic ERROR events.
+
+## 5. Log / correlation expectations
+
+- One `correlation_id` spanning copy-failure → re-upload → sent.
+- WARN (single) for the copy rejection with `error_type=PHOTO_EXT_INVALID`; no stack noise; no media bytes in logs; no source/target titles leaked beyond ids already used by the existing redaction rules.
+
+## 6. Flood-wait & retry safety
+
+- If Telegram returns FLOOD_WAIT at any step, the run **aborts** (do not sleep-loop a manual test): record `retry_after`, wait ≥ that value + 60s manually, re-run the cell from a fresh probe message.
+- Max 1 execution per cell per 10 minutes; abort the whole run on 2 consecutive FLOOD_WAITs.
+
+## 7. Abort conditions (stop immediately, no state cleanup by hand)
+
+- Any job lands `UNKNOWN` (ambiguous outcome) — preserve as forensic evidence, resolve only through the reconciler procedure.
+- Any job lands `FAILED` with an error other than the deliberately provoked one.
+- Any write appears in production rules/queues outside the test namespace (`mtx-`/test-rule id check before AND after).
+- Duplicate target message observed for a cell.
+
+## 8. Cleanup requirements
+
+- Drop the test rule (run_matrix `drop_rule` pattern), delete probe messages from source+target channels, allow job rows to age out under the 7-day retention (do not hand-delete), remove `/tmp/atf-media` leftovers only if the cleanup fix is verified absent (which would itself be a finding).
+
+## 9. Exact commands (authorized run only)
+
+```bash
+# preflight
+ssh vps 'docker logs atf-worker --tail 1 | jq -r .commit_sha'          # must equal PR head
+ssh vps 'curl -s localhost:<worker-healthz>/healthz'                   # db_connected true
+
+# per cell: post probe media in source channel, then
+sqlite3 /var/lib/docker/volumes/autotelegramforward_atf-data/_data/atf.db \
+  "SELECT id,status,attempts,error_detail FROM delivery_jobs WHERE rule_id='<test-rule-uuid>';"
+# expect: SENT ×1 per cell, then run the §4 verification queries verbatim
+```
+
+## 10. Success criteria (all required)
+
+1. Copy rejection observed (WARN with PHOTO_EXT_INVALID) — recovery path actually exercised, not bypassed.
+2. Staged filename carried the expected extension (worker log line with staged path).
+3. Telegram accepted exactly one send per probe; target shows exactly one message per probe.
+4. Fence/mapping/job all SENT with the same target id (§4).
+5. Staged file removed post-send.
+6. Zero unrelated rows touched (before/after row-count delta limited to the test namespace).
+7. No sensitive content in logs (manual grep of the correlation_id lines).
diff --git a/worker/cmd/worker/main.go b/worker/cmd/worker/main.go
index 4dd51a17..2610594f 100644
--- a/worker/cmd/worker/main.go
+++ b/worker/cmd/worker/main.go
@@ -592,6 +592,11 @@ func runLoop(ctx context.Context, log *slog.Logger, engine *delivery.Engine, sto
 		defer sweepWG.Done()
 		ticker := time.NewTicker(15 * time.Second)
 		defer ticker.Stop()
+		// Terminal fences age out on the same 7-day window the job rows
+		// they guarded use; the hourly gate keeps the DELETE off the hot
+		// path. UNKNOWN fences are never swept — they are evidence.
+		const fenceRetention = 7 * 24 * time.Hour
+		var lastFencePurge time.Time
 		for {
 			select {
 			case <-ctx.Done():
@@ -605,6 +610,18 @@ func runLoop(ctx context.Context, log *slog.Logger, engine *delivery.Engine, sto
 				if n > 0 {
 					log.Warn("fence sweep moved expired claims to UNKNOWN", "count", n)
 				}
+				if time.Since(lastFencePurge) < time.Hour {
+					continue
+				}
+				lastFencePurge = time.Now()
+				purged, err := store.PurgeAgedFences(ctx, time.Now().Add(-fenceRetention), 500)
+				if err != nil {
+					log.Warn("fence retention purge failed", "error", err)
+					continue
+				}
+				if purged > 0 {
+					log.Info("fence retention purged terminal rows", "count", purged)
+				}
 			}
 		}
 	}()
diff --git a/worker/internal/domain/autoreply/autoreply_test.go b/worker/internal/domain/autoreply/autoreply_test.go
index 55d446f5..c13839c3 100644
--- a/worker/internal/domain/autoreply/autoreply_test.go
+++ b/worker/internal/domain/autoreply/autoreply_test.go
@@ -233,20 +233,27 @@ func TestDecideGates(t *testing.T) {
 		t.Errorf("gate 6 command failed: %v %s", allowed, reason)
 	}
 
-	// Gate 7: inactive day
+	// Gate 7: inactive day. IsActiveDay reads the account-offset day (the
+	// default offset is 0 → UTC): derive the "different day" from the same
+	// clock the model reads, or the gate flips depending on the machine's
+	// local timezone.
+	utcNow := now.UTC()
 	sDay := DefaultSettings(1)
 	sDay.Enabled = true
-	sDay.ActiveDays = []int{((int(now.Weekday()) + 6) % 7 + 1) % 7} // explicitly different day
+	sDay.ActiveDays = []int{((int(utcNow.Weekday()) + 6) % 7 + 1) % 7} // explicitly different day
 	allowed, reason, _ = Decide(sDay, nil, nil, &IncomingMessage{Text: "hi", ChatType: "private"}, 0, 0, 0, now)
 	if allowed || reason != "inactive_day" {
 		t.Errorf("gate 7 inactive_day failed: %v %s", allowed, reason)
 	}
 
-	// Gate 8: quiet hours
+	// Gate 8: quiet hours. IsQuietNow reads the hour through the account's
+	// timezone offset (0 by default → UTC), not the machine's local zone:
+	// write the window in that same clock or the test only passes on UTC
+	// hosts.
 	sQuiet := DefaultSettings(1)
 	sQuiet.Enabled = true
-	sQuiet.QuietHoursStart = now.Hour()
-	sQuiet.QuietHoursEnd = (now.Hour() + 2) % 24
+	sQuiet.QuietHoursStart = utcNow.Hour()
+	sQuiet.QuietHoursEnd = (utcNow.Hour() + 2) % 24
 	allowed, reason, _ = Decide(sQuiet, nil, nil, &IncomingMessage{Text: "hi", ChatType: "private"}, 0, 0, 0, now)
 	if allowed || reason != "quiet_hours" {
 		t.Errorf("gate 8 quiet_hours failed: %v %s", allowed, reason)
diff --git a/worker/internal/engine/delivery/branding_stage.go b/worker/internal/engine/delivery/branding_stage.go
index 9f965f02..7194f45e 100644
--- a/worker/internal/engine/delivery/branding_stage.go
+++ b/worker/internal/engine/delivery/branding_stage.go
@@ -82,6 +82,25 @@ func unsupportedFailure(ferr *telegraph.DeliveryFailure) bool {
 		(ferr.ErrorType == "BOT_METHOD_INVALID" || ferr.ErrorType == "BOT_TRANSPORT_UNAVAILABLE")
 }
 
+// botMediaRefused reports whether a bot-route media copy failed with a
+// permanent Telegram refusal that the account transport may still deliver.
+//
+// The Bot API adapter reports every 400 as TELEGRAM_PERMANENT_ERROR with a
+// BOT_API_* error type (telegram_runtime/adapters/bot_api), so the
+// copy/re-upload classes (FORWARDING_RESTRICTED, PHOTO_EXT_INVALID, …) can
+// never match NeedsReupload() on this route. The legacy Python path
+// explicitly rescued these jobs by re-dispatching them through the account
+// transport (durable_queue.py _on_restricted); without that rescue a
+// restricted source or a target-side media rejection killed the job
+// permanently even though the USER session would deliver it. The account
+// transport applies its own copy→re-upload fallback chain after the handoff.
+func botMediaRefused(ferr *telegraph.DeliveryFailure, fallbackEnabled bool) bool {
+	if ferr == nil || ferr.IsTransient() || !fallbackEnabled {
+		return false
+	}
+	return true
+}
+
 // --- branding stage 1: post the deliverable into C ------------------------- //
 
 // brandingView builds the message-side input the hop decision reads from the
diff --git a/worker/internal/engine/delivery/delivery.go b/worker/internal/engine/delivery/delivery.go
index a06fc4fc..9b63abfe 100644
--- a/worker/internal/engine/delivery/delivery.go
+++ b/worker/internal/engine/delivery/delivery.go
@@ -27,6 +27,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"os"
 	"strconv"
 	"strings"
 	"sync"
@@ -816,8 +817,8 @@ func (e *Engine) dispatch(ctx context.Context, j *job.DeliveryJob, r *rule.Forwa
 			if p.HasMedia() {
 				id, ferr := e.gateway.BotCopyMessage(callCtx, bt, target, src, srcMsg, "", p.TopicID(), buttons)
 				e.metrics.ObserveRPC("bot_copy_message", 0)
-				if ferr != nil && unsupportedFailure(ferr) {
-					e.jobLogger(j).Warn("bot copy unsupported — falling back to account transport", "reason", ferr.Message)
+				if ferr != nil && (unsupportedFailure(ferr) || botMediaRefused(ferr, p.FallbackEnabled())) {
+					e.jobLogger(j).Warn("bot copy refused the media — falling back to account transport", "reason", ferr.Message)
 					return e.gateway.CopyMessage(callCtx, sessionID, target, src, srcMsg, "", p.TopicID(), j.Payload.DropSourceButtons())
 				}
 				return id, ferr
@@ -829,8 +830,8 @@ func (e *Engine) dispatch(ctx context.Context, j *job.DeliveryJob, r *rule.Forwa
 		if p.HasMedia() {
 			id, ferr := e.gateway.BotCopyMessage(callCtx, bt, target, src, srcMsg, caption, p.TopicID(), buttons)
 			e.metrics.ObserveRPC("bot_copy_message", 0)
-			if ferr != nil && unsupportedFailure(ferr) {
-				e.jobLogger(j).Warn("bot copy unsupported — falling back to account transport", "reason", ferr.Message)
+			if ferr != nil && (unsupportedFailure(ferr) || botMediaRefused(ferr, p.FallbackEnabled())) {
+				e.jobLogger(j).Warn("bot copy refused the media — falling back to account transport", "reason", ferr.Message)
 				return e.gateway.CopyMessage(callCtx, sessionID, target, src, srcMsg, caption, p.TopicID(), j.Payload.DropSourceButtons())
 			}
 			return id, ferr
@@ -1005,26 +1006,41 @@ func (e *Engine) dispatchAlbum(ctx context.Context, j *job.DeliveryJob, albumIDs
 	if ferr == nil && len(ids) > 0 {
 		return firstNonZero(ids), nil
 	}
-	if ferr != nil && !ferr.NeedsReupload() {
-		// The group copy failed for a non-restriction reason: fall through to
-		// the per-item re-upload (the Python path does the same).
-		e.jobLogger(j).Warn("album group copy failed — falling back to re-upload", "error", ferr.Message)
+	if ferr != nil && ferr.IsAmbiguous() {
+		// The group copy's outcome is unknown: the upload may have been
+		// accepted by Telegram before the connection died. Re-sending the
+		// album item-by-item would duplicate whatever landed, so record the
+		// ambiguity and let the reconciler decide. (An ambiguous group copy
+		// used to fall straight into the per-item re-upload loop and deliver
+		// the album twice.)
+		e.jobLogger(j).Warn("album group copy outcome ambiguous — refusing to re-upload", "error", ferr.Message)
+		return 0, ferr
 	}
-	if ferr != nil && ferr.NeedsReupload() && !p.FallbackEnabled() {
+	if ferr != nil && !p.FallbackEnabled() {
+		// The operator disabled the re-upload fallback: honor the opt-out for
+		// every failure class, not only target-side rejections.
 		return 0, ferr
 	}
-	// Per-item re-upload fallback.
-	var last int64
+	if ferr != nil {
+		// The group copy failed for a non-ambiguous reason: fall through to
+		// the per-item re-upload (the Python path does the same).
+		e.jobLogger(j).Warn("album group copy failed — falling back to re-upload", "error", ferr.Message)
+	}
+	// Per-item re-upload fallback. The mapping keys on the FIRST delivered
+	// item — edit/delete sync and auto-delete act on that id (the Python
+	// path returns res[0].id; returning the last id pointed them at the
+	// wrong message).
+	var first int64
 	for _, itemID := range albumIDs {
 		id, rferr := e.reuploadMedia(ctx, j, src, itemID, caption)
 		if rferr != nil {
 			return 0, rferr
 		}
-		if id > 0 {
-			last = id
+		if id > 0 && first == 0 {
+			first = id
 		}
 	}
-	return last, nil
+	return first, nil
 }
 
 // reuploadMedia is the restricted-source bypass: download the media through
@@ -1051,6 +1067,18 @@ func (e *Engine) reuploadMedia(ctx context.Context, j *job.DeliveryJob, srcChat,
 	if _, ferr := e.gateway.DownloadMedia(ctx, sessionID, srcChat, srcMsg, dest); ferr != nil {
 		return 0, ferr
 	}
+	// The staged file is scratch: every attempt re-downloads it, nothing
+	// reads it after this call returns, and the runtime holds no reference
+	// beyond the send. Remove it whichever way the call ends — success,
+	// failure, timeout, or cancel — so the media dir does not grow with
+	// every retried re-upload (the Python path deletes its temp files the
+	// same way; the Go port dropped that and files piled up on shared-FS
+	// deployments).
+	defer func() {
+		if err := os.Remove(dest.Path); err != nil && !os.IsNotExist(err) {
+			e.jobLogger(j).Info("staged media cleanup failed", "path", dest.Path, "error", err.Error())
+		}
+	}()
 	token, ferr := e.gateway.PrepareMediaUpload(ctx, sessionID, dest, srcChat, srcMsg, p.MediaType())
 	if ferr != nil {
 		return 0, ferr
diff --git a/worker/internal/engine/delivery/recovery_gaps_test.go b/worker/internal/engine/delivery/recovery_gaps_test.go
new file mode 100644
index 00000000..5ce41d4f
--- /dev/null
+++ b/worker/internal/engine/delivery/recovery_gaps_test.go
@@ -0,0 +1,289 @@
+package delivery
+
+// Regression tests for the media-delivery recovery gaps found in the phase-2
+// audit (D1 album duplication, D2 bot-route media loss, D3 staged-file leak).
+// Each test pins the exact production failure sequence the audit constructed.
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"atf/worker/internal/domain/job"
+	"atf/worker/internal/domain/rule"
+	"atf/worker/internal/queue"
+	"atf/worker/internal/telegraph"
+	"atf/worker/internal/telegraph/fakegateway"
+)
+
+// TestAlbumAmbiguousGroupCopyNeverReuploads: a CopyMediaGroup whose outcome
+// is ambiguous (timeout after Telegram may have accepted the upload) used to
+// fall straight into the per-item re-upload loop, delivering the album twice
+// — once as the group, once as N fresh messages. The ambiguity must record
+// UNKNOWN and re-upload nothing.
+func TestAlbumAmbiguousGroupCopyNeverReuploads(t *testing.T) {
+	h := newHarness(t)
+	j := &job.DeliveryJob{
+		ID: "r1:c:21:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 21,
+		TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5,
+		Payload: *jobPayload(map[string]any{
+			"text": "album", "has_media": true, "media_type": "photo",
+			"media_group_id": "grp-21", "album_mode": "album",
+			"album_message_ids": []any{11, 12, 13},
+			"fallback_enabled":  true,
+		}),
+	}
+	h.enqueue(t, j)
+	h.gateway.CopyMediaGroupFailsWith = &telegraph.DeliveryFailure{
+		Code: "NETWORK_TIMEOUT", Message: "rpc timeout after the upload was accepted",
+	}
+
+	if err := h.engine.Process(context.Background(), j); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	assertJobStatus(t, h.store, j.ID, string(job.StatusUnknown))
+	for _, c := range h.gateway.Recorded() {
+		if c.Op == "DownloadMedia" || c.Op == "SendMedia" {
+			t.Errorf("ambiguous group copy re-uploaded media (%s) — the album would be delivered twice", c.Op)
+		}
+	}
+}
+
+// TestAlbumFallbackMapsTheFirstItem: when the group copy is refused
+// (NeedsReupload) and the fallback runs, the mapping/fence target id must be
+// the FIRST delivered item — edit/delete sync and auto-delete key on it.
+// The Go port used to return the LAST id and pointed those at the wrong
+// message (Python returns res[0].id).
+func TestAlbumFallbackMapsTheFirstItem(t *testing.T) {
+	h := newHarness(t)
+	j := &job.DeliveryJob{
+		ID: "r1:c:22:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 22,
+		TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5,
+		Payload: *jobPayload(map[string]any{
+			"text": "album", "has_media": true, "media_type": "photo",
+			"media_group_id": "grp-22", "album_mode": "album",
+			"album_message_ids": []any{11, 12, 13},
+			"fallback_enabled":  true,
+		}),
+	}
+	h.enqueue(t, j)
+	h.gateway.CopyMediaGroupFailsWith = &telegraph.DeliveryFailure{
+		Code: "TELEGRAM_PERMANENT_ERROR", ErrorType: "PHOTO_EXT_INVALID",
+		Message: "Telegram says: [400 PHOTO_EXT_INVALID]",
+	}
+	// ScriptedIDs indexes the fake's GLOBAL call log: group-copy failure,
+	// then 3 re-upload rounds of (GetFileExtension, DownloadMedia,
+	// PrepareMediaUpload, SendMedia). The SendMedia positions carry the ids
+	// 101/102/103; the first delivered item must become the mapping id.
+	h.gateway.ScriptedIDs = []int64{0, 0, 0, 0, 0, 101, 0, 0, 0, 0, 102, 0, 0, 0, 0, 103}
+
+	if err := h.engine.Process(context.Background(), j); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	assertJobStatus(t, h.store, j.ID, string(job.StatusSent))
+
+	probeKey := job.AttemptKey{RuleID: j.RuleID, SourceChatID: j.SourceChatID, SourceMessageID: j.SourceMessageID, TargetChatID: j.TargetChatID, JobKind: j.JobKind}
+	outcome, row, err := h.store.FenceClaim(context.Background(), probeKey, "probe-reader", float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("fence probe: %v", err)
+	}
+	if outcome != job.FenceAlreadySent {
+		t.Fatalf("fence after album fallback = %s, want already_sent", outcome)
+	}
+	if row == nil || row.TargetMessageID != 101 {
+		t.Fatalf("album mapping id = %+v, want 101 (the FIRST re-uploaded item)", row)
+	}
+}
+
+// TestAlbumFallbackHonorsOptOut: the re-upload opt-out applies to every
+// failure class on the group-copy path, not only target-side rejections.
+func TestAlbumFallbackHonorsOptOut(t *testing.T) {
+	h := newHarness(t)
+	j := &job.DeliveryJob{
+		ID: "r1:c:23:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 23,
+		TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5,
+		Payload: *jobPayload(map[string]any{
+			"text": "album", "has_media": true, "media_type": "photo",
+			"media_group_id": "grp-23", "album_mode": "album",
+			"album_message_ids": []any{11, 12, 13},
+			"fallback_enabled":  false,
+		}),
+	}
+	h.enqueue(t, j)
+	h.gateway.CopyMediaGroupFailsWith = &telegraph.DeliveryFailure{
+		Code: "TELEGRAM_TEMPORARY_ERROR", Message: "flood-ish refusal",
+	}
+
+	if err := h.engine.Process(context.Background(), j); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	for _, c := range h.gateway.Recorded() {
+		if c.Op == "DownloadMedia" || c.Op == "SendMedia" {
+			t.Errorf("fallback_enabled=false still re-uploaded (%s) — the operator's opt-out was ignored", c.Op)
+		}
+	}
+}
+
+// TestBotMediaRefusalFallsBackToAccount: a bot-route media copy refused by
+// Telegram with a permanent 400 (restricted source, target-side media
+// rejection — the Bot API adapter reports them as BOT_API_400) used to fail
+// the job permanently even though the account transport delivers it. The
+// legacy Python path rescued these jobs; the port must too.
+func TestBotMediaRefusalFallsBackToAccount(t *testing.T) {
+	store := queue.NewTestStore(t)
+	gw := fakegateway.New("")
+	gw.BotSendFailsWith = &telegraph.DeliveryFailure{
+		Code: "TELEGRAM_PERMANENT_ERROR", ErrorType: "BOT_API_400",
+		Message: "Bad Request: message can't be forwarded",
+	}
+	r := &fixedRules{rule: &rule.ForwardRule{
+		ID: "r-bot-media", SessionID: "sess-user", IsActive: true,
+		DeliveryTransport: "BOT", BotTokenID: "tok-1", OwnerUserID: 7,
+	}}
+
+	log := &testLogger{}
+	engine := New(store, r, gw, nil, log, &testMetrics{}, "w1", t.TempDir(), Options{
+		NumWorkers: 2, BatchSize: 5, LeaseDuration: 30 * time.Second, RPCTimeout: 5 * time.Second,
+	})
+	j := &job.DeliveryJob{
+		ID: "r-bot-media:c:31:t", RuleID: "r-bot-media", SourceChatID: "c", SourceMessageID: 31,
+		TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5,
+		Payload: *jobPayload(map[string]any{
+			"text": "caption", "has_media": true, "media_type": "photo",
+			"delivery_transport": "BOT", "bot_token_id": "tok-1", "owner_user_id": 7,
+			"session_id": "sess-user", "fallback_enabled": true,
+		}),
+	}
+	queue.InsertJob(t, store, j)
+
+	if err := engine.Process(context.Background(), j); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	assertJobStatus(t, store, j.ID, string(job.StatusSent))
+	userCopy := false
+	for _, c := range gw.Recorded() {
+		if c.Op == "CopyMessage" && c.SessionID == "sess-user" {
+			userCopy = true
+		}
+	}
+	if !userCopy {
+		t.Fatal("bot media refusal did not fall back to the account transport — the job would be lost")
+	}
+}
+
+// TestBotMediaRefusalHonorsOptOut: without the fallback opt-in the bot
+// refusal fails the job; no silent account-transport bypass.
+func TestBotMediaRefusalHonorsOptOut(t *testing.T) {
+	store := queue.NewTestStore(t)
+	gw := fakegateway.New("")
+	gw.BotSendFailsWith = &telegraph.DeliveryFailure{
+		Code: "TELEGRAM_PERMANENT_ERROR", ErrorType: "BOT_API_400",
+		Message: "Bad Request: message can't be forwarded",
+	}
+	r := &fixedRules{rule: &rule.ForwardRule{
+		ID: "r-bot-nofb", SessionID: "sess-user", IsActive: true,
+		DeliveryTransport: "BOT", BotTokenID: "tok-1", OwnerUserID: 7,
+	}}
+	log := &testLogger{}
+	engine := New(store, r, gw, nil, log, &testMetrics{}, "w1", t.TempDir(), Options{
+		NumWorkers: 2, BatchSize: 5, LeaseDuration: 30 * time.Second, RPCTimeout: 5 * time.Second,
+	})
+	j := &job.DeliveryJob{
+		ID: "r-bot-nofb:c:32:t", RuleID: "r-bot-nofb", SourceChatID: "c", SourceMessageID: 32,
+		TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5,
+		Payload: *jobPayload(map[string]any{
+			"text": "caption", "has_media": true, "media_type": "photo",
+			"delivery_transport": "BOT", "bot_token_id": "tok-1", "owner_user_id": 7,
+			"session_id": "sess-user", "fallback_enabled": false,
+		}),
+	}
+	queue.InsertJob(t, store, j)
+
+	if err := engine.Process(context.Background(), j); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	assertJobStatus(t, store, j.ID, string(job.StatusFailed))
+	for _, c := range gw.Recorded() {
+		if c.Op == "CopyMessage" {
+			t.Error("fallback_enabled=false still dispatched through the account transport")
+		}
+	}
+}
+
+// TestStagedMediaCleanedUpAfterReupload: the staged scratch file must be
+// gone whichever way the re-upload ends. The Go path used to leave every
+// staged file behind (the Python path deletes its temp files).
+func TestStagedMediaCleanedUpAfterReupload(t *testing.T) {
+	store := queue.NewTestStore(t)
+	gw := fakegateway.New("")
+	dir := t.TempDir()
+	h := &harness{store: store, gateway: gw, engine: New(store, staticRules{}, gw, nil, &testLogger{}, &testMetrics{}, "w-test", dir, Options{
+		NumWorkers: 2, BatchSize: 5, LeaseDuration: 30 * time.Second, RPCTimeout: 5 * time.Second,
+	}), metrics: &testMetrics{}, log: &testLogger{}}
+	j := &job.DeliveryJob{
+		ID: "r1:c:41:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 41,
+		TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5,
+		Payload: *jobPayload(map[string]any{
+			"text": "caption", "has_media": true, "media_type": "photo",
+			"fallback_enabled": true,
+		}),
+	}
+	h.enqueue(t, j)
+	h.gateway.CopyMediaRejected = true
+	h.gateway.DownloadWritesFile = true
+	ext := ".jpg"
+	h.gateway.ExtensionOverride = &ext
+
+	if err := h.engine.Process(context.Background(), j); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	assertJobStatus(t, h.store, j.ID, string(job.StatusSent))
+	staged := filepath.Join(dir, j.ID+ext)
+	if _, err := os.Stat(staged); !os.IsNotExist(err) {
+		t.Errorf("staged file %s survived a successful re-upload — the media dir grows with every retry", staged)
+	}
+}
+
+// TestStagedMediaCleanedUpAfterUploadFailure pins the failure side of the
+// same contract, with a Telegram upload-time rejection (a knob the fake
+// never had, which is why this path was untestable before).
+func TestStagedMediaCleanedUpAfterUploadFailure(t *testing.T) {
+	store := queue.NewTestStore(t)
+	gw := fakegateway.New("")
+	dir := t.TempDir()
+	h := &harness{store: store, gateway: gw, engine: New(store, staticRules{}, gw, nil, &testLogger{}, &testMetrics{}, "w-test", dir, Options{
+		NumWorkers: 2, BatchSize: 5, LeaseDuration: 30 * time.Second, RPCTimeout: 5 * time.Second,
+	}), metrics: &testMetrics{}, log: &testLogger{}}
+	j := &job.DeliveryJob{
+		ID: "r1:c:42:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 42,
+		TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5,
+		Payload: *jobPayload(map[string]any{
+			"text": "caption", "has_media": true, "media_type": "photo",
+			"fallback_enabled": true,
+		}),
+	}
+	h.enqueue(t, j)
+	h.gateway.CopyMediaRejected = true
+	h.gateway.DownloadWritesFile = true
+	h.gateway.SendMediaFailsWith = &telegraph.DeliveryFailure{
+		Code: "FLOOD_WAIT", Message: "Telegram says: [420 FLOOD_WAIT] (60)",
+	}
+	ext := ".jpg"
+	h.gateway.ExtensionOverride = &ext
+
+	if err := h.engine.Process(context.Background(), j); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	// A flood-waited upload is transient: the job must be retry-waiting,
+	// not permanently failed — and the staged file must still be gone.
+	got := jobStatus(t, h.store, j.ID)
+	if got != string(job.StatusRetryWait) && got != string(job.StatusPending) {
+		t.Errorf("flood-waited re-upload status = %s, want a retry state", got)
+	}
+	staged := filepath.Join(dir, j.ID+ext)
+	if _, err := os.Stat(staged); !os.IsNotExist(err) {
+		t.Errorf("staged file %s survived a failed upload", staged)
+	}
+}
diff --git a/worker/internal/engine/reconciler/reconciler.go b/worker/internal/engine/reconciler/reconciler.go
index 32c44b32..1635e178 100644
--- a/worker/internal/engine/reconciler/reconciler.go
+++ b/worker/internal/engine/reconciler/reconciler.go
@@ -64,7 +64,17 @@ func New(store Store, workerID string, grace, interval time.Duration, log Logger
 // code 12; the Python runtime's server raises NotImplementedError for a
 // method it does not register, and the Go client surfaces it as
 // `code = Unimplemented desc = Unexpected <class 'NotImplementedError'>…`.
-var provableMarkers = []string{"code = Unimplemented"}
+//
+// A dial-phase Unavailable is equally safe: when the client's transport
+// failed *while dialing* (`transport: Error while dialing: dial tcp …`),
+// no HTTP/2 stream ever existed, so no request byte reached the runtime and
+// no handler can have run. A bare `code = Unavailable` is NOT on the list —
+// a transport that dies mid-call after Telegram accepted the upload reports
+// the same code, and closing that would duplicate the delivery.
+var provableMarkers = []string{
+	"code = Unimplemented",
+	"Error while dialing",
+}
 
 // provablyUnexecuted reports whether the recorded failure proves the RPC
 // never executed a Telegram operation. Anything not on this list stays
@@ -90,8 +100,8 @@ func (r *Reconciler) RunOnce(ctx context.Context) (resolved, remaining int, err
 			remaining++
 			continue
 		}
-		reason := "reconciled: the runtime rejected the RPC (Unimplemented) before any handler ran — " +
-			"no Telegram operation was attempted, so the message cannot be in the target; original: " +
+		reason := "reconciled: the runtime rejected the RPC before any handler ran (no Telegram " +
+			"operation was attempted, so the message cannot be in the target); original: " +
 			truncate(f.LastError, 200)
 		ok, rerr := r.store.ReconcileUnknown(ctx, f.Key, r.workerID, reason)
 		if rerr != nil {
diff --git a/worker/internal/engine/reconciler/reconciler_test.go b/worker/internal/engine/reconciler/reconciler_test.go
index 0bf4ff8e..8fa71417 100644
--- a/worker/internal/engine/reconciler/reconciler_test.go
+++ b/worker/internal/engine/reconciler/reconciler_test.go
@@ -147,6 +147,13 @@ func TestProvablyUnexecutedTable(t *testing.T) {
 		{"rpc error: code = Unimplemented desc = unknown method SendMedia", true},
 		{"rpc error: code = DeadlineExceeded", false},
 		{"rpc error: code = Unavailable desc = connection error", false},
+		// Dial-phase Unavailable proves no stream existed: nothing reached a
+		// handler, so no Telegram operation can have been attempted.
+		{"rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.0.0.5:50053: connect: connection rejection\"", true},
+		// A bare Unavailable says nothing about WHERE the transport died: a
+		// runtime killed mid-upload reports the same code after Telegram
+		// accepted the send. Classifying it would duplicate the delivery.
+		{"rpc error: code = Unavailable desc = transport is closing", false},
 		{"Uncertain outcome: transport dropped mid-send", false},
 		{"", false},
 	}
diff --git a/worker/internal/queue/fence_lease_gaps_test.go b/worker/internal/queue/fence_lease_gaps_test.go
new file mode 100644
index 00000000..5fabd3b1
--- /dev/null
+++ b/worker/internal/queue/fence_lease_gaps_test.go
@@ -0,0 +1,191 @@
+package queue
+
+// Regression tests for the fence-claim lease semantics (audit round S4/S3).
+//
+// S4: every goroutine in a worker process shares one worker id, so a
+// CLAIMED fence whose worker_id matches the caller used to be handed back
+// as FenceDispatch with no lease check — a dispatch that outlived the 30s
+// job lease (a large album re-upload) was handed to a second claim and the
+// job was delivered twice. The lease now decides, for same-worker claims
+// exactly as for cross-worker ones.
+//
+// S3: a SENT fence is terminal. The SQLite store used to accept both
+// FenceRecordSent and FenceRecordOutcome unconditionally, so a late sibling
+// failure could stamp FAILED over a confirmed send and turn it back into a
+// re-dispatchable job.
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"atf/worker/internal/domain/job"
+)
+
+func TestFenceClaimSameWorkerHonorsLease(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	key := job.AttemptKey{
+		RuleID: "r1", SourceChatID: "c", SourceMessageID: 41,
+		TargetChatID: "t", JobKind: "forward",
+	}
+	const id = "r1:c:41:t"
+
+	// Claim with a LIVE lease: the same worker re-claiming must see it in
+	// flight, not get a second dispatch. The first dispatch is still
+	// running — that is what the claim means.
+	outcome, _, err := s.FenceClaim(ctx, key, "w1", float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("claim 1: %v", err)
+	}
+	if outcome != job.FenceDispatch {
+		t.Fatalf("first claim = %s, want dispatch", outcome)
+	}
+	outcome, _, err = s.FenceClaim(ctx, key, "w1", float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("same-worker claim: %v", err)
+	}
+	if outcome != job.FenceInFlight {
+		t.Fatalf("same-worker live-lease claim = %s, want in_flight (a second dispatch here delivers the job twice)", outcome)
+	}
+
+	// Once the fence's lease has expired the claim is stale: the stale
+	// sweep + reconciler own it, never a direct re-dispatch.
+	freshKey := job.AttemptKey{
+		RuleID: "r1", SourceChatID: "c", SourceMessageID: 42,
+		TargetChatID: "t", JobKind: "forward",
+	}
+	outcome, _, err = s.FenceClaim(ctx, freshKey, "w1", float64(time.Now().Add(-time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("expired claim: %v", err)
+	}
+	if outcome != job.FenceDispatch {
+		t.Fatalf("first claim of expired-lease fence = %s, want dispatch (the claim IS the lease)", outcome)
+	}
+	outcome, _, err = s.FenceClaim(ctx, freshKey, "w1", float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("second claim on expired fence: %v", err)
+	}
+	if outcome != job.FenceStale {
+		t.Fatalf("same-worker expired-lease claim = %s, want stale", outcome)
+	}
+
+	// A FAILED fence must still be re-claimable by the same worker — the
+	// replay path depends on it (the pre-read restructure must not have
+	// closed it).
+	retryKey := job.AttemptKey{
+		RuleID: "r1", SourceChatID: "c", SourceMessageID: 43,
+		TargetChatID: "t", JobKind: "forward",
+	}
+	if _, _, err := s.FenceClaim(ctx, retryKey, "w1", float64(time.Now().Add(time.Minute).Unix())); err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if err := s.FenceRecordOutcome(ctx, retryKey, "w1", "FAILED", "transient"); err != nil {
+		t.Fatalf("record outcome: %v", err)
+	}
+	outcome, row, err := s.FenceClaim(ctx, retryKey, "w1", float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("reclaim after FAILED: %v", err)
+	}
+	if outcome != job.FenceDispatch {
+		t.Fatalf("FAILED reclaim = %s, want dispatch", outcome)
+	}
+	if row == nil || row.Attempts != 1 {
+		t.Fatalf("FAILED reclaim attempts = %+v, want 1 (the counter must survive the restructure)", row)
+	}
+}
+
+func TestFenceSentIsTerminalOnSQLite(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	key := job.AttemptKey{
+		RuleID: "r1", SourceChatID: "c", SourceMessageID: 44,
+		TargetChatID: "t", JobKind: "forward",
+	}
+
+	if _, _, err := s.FenceClaim(ctx, key, "w1", float64(time.Now().Add(time.Minute).Unix())); err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if err := s.FenceRecordSent(ctx, key, "w1", 4400); err != nil {
+		t.Fatalf("record sent: %v", err)
+	}
+
+	// A sibling's late failure must NOT turn a confirmed send back into a
+	// re-dispatchable failure (S3 — the message is physically in the
+	// target; FAILED here would eventually duplicate it).
+	if err := s.FenceRecordOutcome(ctx, key, "w1", "FAILED", "late sibling failure"); err != nil {
+		t.Fatalf("record outcome over SENT: %v", err)
+	}
+	outcome, _, err := s.FenceClaim(ctx, key, "w2", float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("claim after outcome-over-sent: %v", err)
+	}
+	if outcome != job.FenceAlreadySent {
+		t.Fatalf("claim after FAILED-over-SENT = %s, want already_sent (the fence must stay terminal)", outcome)
+	}
+
+	// FenceRecordSent is likewise never a downgrade path once SENT.
+	if err := s.FenceRecordSent(ctx, key, "w2", 4401); err != nil {
+		t.Fatalf("second record sent: %v", err)
+	}
+	_, row, err := s.FenceClaim(ctx, key, "w3", float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("final claim: %v", err)
+	}
+	if row.TargetMessageID != 4400 {
+		t.Fatalf("target id after second record-sent = %d, want 4400 (first confirmed send wins)", row.TargetMessageID)
+	}
+}
+
+func TestPurgeAgedFencesKeepsUnknownAndFresh(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	db := StoreDB(t, s)
+
+	mk := func(msg int, status string, ageDays int) job.AttemptKey {
+		key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: int64(msg), TargetChatID: "t", JobKind: "forward"}
+		_, err := db.Exec(`INSERT INTO delivery_attempts (
+			id, rule_id, source_chat_id, source_message_id, target_chat_id, job_kind,
+			status, worker_id, lease_until, attempts, last_error, created_at, updated_at)
+			VALUES (?,?,?,?,?,?,?,?,0,0,'',?,?)`,
+			key.String(), key.RuleID, key.SourceChatID, key.SourceMessageID,
+			key.TargetChatID, key.JobKind, status, "w1",
+			time.Now().Unix()-int64(ageDays*24*3600), time.Now().Unix()-int64(ageDays*24*3600))
+		if err != nil {
+			t.Fatalf("seed fence %d: %v", msg, err)
+		}
+		return key
+	}
+	oldSent := mk(61, "SENT", 10)
+	oldFailed := mk(62, "FAILED", 10)
+	oldUnknown := mk(63, "UNKNOWN", 10)
+	freshSent := mk(64, "SENT", 1)
+
+	// Jobs-side guard rail: nothing here may delete unresolved evidence.
+	purged, err := s.PurgeAgedFences(ctx, time.Now().Add(-7*24*time.Hour), 500)
+	if err != nil {
+		t.Fatalf("purge: %v", err)
+	}
+	if purged != 2 {
+		t.Fatalf("purged = %d, want exactly the 2 aged terminal rows", purged)
+	}
+	for _, tc := range []struct {
+		key   job.AttemptKey
+		want  bool
+		label string
+	}{
+		{oldSent, false, "aged SENT fence"},
+		{oldFailed, false, "aged FAILED fence"},
+		{oldUnknown, true, "aged UNKNOWN fence (evidence — never swept)"},
+		{freshSent, true, "fresh SENT fence (inside retention)"},
+	} {
+		out, _, err := s.FenceClaim(ctx, tc.key, "probe", float64(time.Now().Add(time.Minute).Unix()))
+		if err != nil {
+			t.Fatalf("%s probe: %v", tc.label, err)
+		}
+		gone := out == job.FenceDispatch // absent row → fresh claim → dispatch
+		if gone == tc.want {
+			t.Errorf("%s: gone=%v, want row-present=%v (outcome %s)", tc.label, gone, tc.want, out)
+		}
+	}
+}
diff --git a/worker/internal/queue/postgres.go b/worker/internal/queue/postgres.go
index 8287f959..e4cf49f9 100644
--- a/worker/internal/queue/postgres.go
+++ b/worker/internal/queue/postgres.go
@@ -639,62 +639,88 @@ func (s *pgStore) FenceClaim(ctx context.Context, key job.AttemptKey, workerID s
 		return "", nil, fmt.Errorf("fence begin: %w", err)
 	}
 	defer tx.Rollback()
-	if _, err := tx.ExecContext(ctx, `
-		INSERT INTO delivery_attempts (
-			id, rule_id, source_chat_id, source_message_id, target_chat_id,
-			job_kind, status, worker_id, lease_until, attempts,
-			last_error, created_at, updated_at
-		) VALUES ($1,$2,$3,$4,$5,$6,'CLAIMED',$7,$8,0,'',$9,$10)
-		ON CONFLICT (id) DO UPDATE SET
-			status = 'CLAIMED',
-			worker_id = EXCLUDED.worker_id,
-			lease_until = EXCLUDED.lease_until,
-			attempts = delivery_attempts.attempts + 1,
-			last_error = '',
-			updated_at = EXCLUDED.updated_at
-		WHERE delivery_attempts.status = 'FAILED'`,
-		key.String(), key.RuleID, key.SourceChatID, key.SourceMessageID,
-		key.TargetChatID, key.JobKind, workerID, leaseUntil, now, now); err != nil {
-		return "", nil, fmt.Errorf("fence upsert: %w", err)
-	}
+	// Read (and lock) the current state BEFORE any write: a re-claim upsert
+	// stamps the caller's worker id, so a post-upsert read cannot tell a
+	// legitimate FAILED re-claim apart from a live same-worker claim — and
+	// that distinction IS the idempotency contract (a dispatch that outlives
+	// its lease must never be handed to a second claim, or the job is sent
+	// twice).
 	row := tx.QueryRowContext(ctx, `
 		SELECT status, worker_id, lease_until, target_message_id, last_error, attempts
-		FROM delivery_attempts WHERE id=$1`, key.String())
+		FROM delivery_attempts WHERE id=$1 FOR UPDATE`, key.String())
 	var (
-		a        job.DeliveryAttempt
-		targetID sql.NullInt64
+		preStatus string
+		preWorker string
+		preLease  float64
+		targetID  sql.NullInt64
+		preErr    string
+		preTries  int
 	)
-	if err := row.Scan(&a.Status, &a.WorkerID, &a.LeaseUntil, &targetID, &a.LastError, &a.Attempts); err != nil {
-		if err == sql.ErrNoRows {
-			// The upsert succeeded but the row cannot be read: treat it as
-			// unavailable rather than guessing — refuse to dispatch unprotected.
-			return "", nil, fmt.Errorf("fence row vanished after upsert for %s", key.String())
-		}
+	err = row.Scan(&preStatus, &preWorker, &preLease, &targetID, &preErr, &preTries)
+	if err != nil && err != sql.ErrNoRows {
 		return "", nil, fmt.Errorf("fence read: %w", err)
 	}
-	if err := tx.Commit(); err != nil {
-		return "", nil, fmt.Errorf("fence commit: %w", err)
+
+	var outcome job.FenceOutcome
+	a := job.DeliveryAttempt{}
+	switch {
+	case err == sql.ErrNoRows:
+		// First ever claim for this logical send.
+		if _, err := tx.ExecContext(ctx, `
+			INSERT INTO delivery_attempts (
+				id, rule_id, source_chat_id, source_message_id, target_chat_id,
+				job_kind, status, worker_id, lease_until, attempts,
+				last_error, created_at, updated_at
+			) VALUES ($1,$2,$3,$4,$5,$6,'CLAIMED',$7,$8,0,'',$9,$10)`,
+			key.String(), key.RuleID, key.SourceChatID, key.SourceMessageID,
+			key.TargetChatID, key.JobKind, workerID, leaseUntil, now, now); err != nil {
+			return "", nil, fmt.Errorf("fence insert: %w", err)
+		}
+		outcome = job.FenceDispatch
+		a.Status, a.WorkerID, a.LeaseUntil, a.Attempts = "CLAIMED", workerID, leaseUntil, 0
+	case preStatus == "SENT":
+		outcome = job.FenceAlreadySent
+		a.Status, a.WorkerID, a.LeaseUntil, a.LastError, a.Attempts =
+			preStatus, preWorker, preLease, preErr, preTries
+	case preStatus == "UNKNOWN":
+		// An earlier send's outcome was never recorded. Never dispatch
+		// blindly — the reconciler / an operator decides.
+		outcome = job.FenceUnknown
+		a.Status, a.WorkerID, a.LeaseUntil, a.LastError, a.Attempts =
+			preStatus, preWorker, preLease, preErr, preTries
+	case preStatus == "CLAIMED" || preStatus == "SENDING":
+		// CLAIMED/SENDING rows are never stolen, live or expired, same
+		// worker or not: every goroutine in this process shares one worker
+		// id, so a same-worker claim means the original dispatch is still
+		// running. Expiry is the stale sweep's + reconciler's decision, not
+		// a dispatch opportunity.
+		if preLease > 0 && preLease >= float64(now) {
+			outcome = job.FenceInFlight
+		} else {
+			outcome = job.FenceStale
+		}
+		a.Status, a.WorkerID, a.LeaseUntil, a.LastError, a.Attempts =
+			preStatus, preWorker, preLease, preErr, preTries
+	default:
+		// FAILED (the replay path): the caller re-claims it.
+		if _, err := tx.ExecContext(ctx, `
+			UPDATE delivery_attempts
+			SET status='CLAIMED', worker_id=$1, lease_until=$2, attempts=attempts+1,
+			    last_error='', updated_at=$3
+			WHERE id=$4 AND status='FAILED'`,
+			workerID, leaseUntil, now, key.String()); err != nil {
+			return "", nil, fmt.Errorf("fence reclaim: %w", err)
+		}
+		outcome = job.FenceDispatch
+		a.Status, a.WorkerID, a.LeaseUntil, a.Attempts = "CLAIMED", workerID, leaseUntil, preTries+1
 	}
 	if targetID.Valid {
 		a.TargetMessageID = targetID.Int64
 	}
-
-	switch a.Status {
-	case "SENT":
-		return job.FenceAlreadySent, &a, nil
-	case "UNKNOWN":
-		return job.FenceUnknown, &a, nil
-	case "FAILED":
-		// The UPSERT re-claimed it for this caller.
-		return job.FenceDispatch, &a, nil
-	}
-	if a.WorkerID == workerID {
-		return job.FenceDispatch, &a, nil
-	}
-	if a.LeaseUntil > 0 && a.LeaseUntil >= float64(now) {
-		return job.FenceInFlight, &a, nil
+	if err := tx.Commit(); err != nil {
+		return "", nil, fmt.Errorf("fence commit: %w", err)
 	}
-	return job.FenceStale, &a, nil
+	return outcome, &a, nil
 }
 
 // FenceRecordSent stamps a confirmed Telegram result. It is written BEFORE
@@ -739,6 +765,23 @@ func (s *pgStore) FenceExpireStale(ctx context.Context, now float64) (int, error
 	return int(n), nil
 }
 
+// PurgeAgedFences deletes terminal fences older than the retention window —
+// bounded, blind to UNKNOWN rows (unresolved evidence is never deleted by a
+// sweep), and semantically identical to the SQLite implementation.
+func (s *pgStore) PurgeAgedFences(ctx context.Context, olderThan time.Time, limit int) (int, error) {
+	res, err := s.db.ExecContext(ctx, `
+		DELETE FROM delivery_attempts WHERE id IN (
+			SELECT id FROM delivery_attempts
+			WHERE status IN ('SENT','FAILED') AND updated_at < $1
+			LIMIT $2
+		)`, olderThan.Unix(), limit)
+	if err != nil {
+		return 0, err
+	}
+	n, _ := res.RowsAffected()
+	return int(n), nil
+}
+
 // FenceResolveUnknown reopens an UNKNOWN fence after the caller has PROVED
 // (a Telegram-side read or a bounded recovery decision) that no earlier copy
 // is in the target. The reason is recorded — prefixed REOPENED, exactly as
diff --git a/worker/internal/queue/postgres_test.go b/worker/internal/queue/postgres_test.go
index 9a461817..3bd16317 100644
--- a/worker/internal/queue/postgres_test.go
+++ b/worker/internal/queue/postgres_test.go
@@ -497,7 +497,7 @@ func TestPostgresConcurrentFenceClaimSingleWinner(t *testing.T) {
 	}
 }
 
-func TestPostgresDuplicateFenceClaimSameWorkerRedispatches(t *testing.T) {
+func TestPostgresDuplicateFenceClaimSameWorkerNeverRedispatchesLiveLease(t *testing.T) {
 	s := newPGSuite(t)
 	j := pgJob("j1", 100, "-100800")
 	key := pgFenceKey(j)
@@ -510,15 +510,56 @@ func TestPostgresDuplicateFenceClaimSameWorkerRedispatches(t *testing.T) {
 	if first != job.FenceDispatch {
 		t.Fatalf("first claim = %s, want dispatch", first)
 	}
-	// CLAIMED by the SAME worker follows the holder: a crash-recovery replay
-	// by the same worker identity re-dispatches rather than blocking.
+	// Every goroutine in one worker process shares the worker id, so a
+	// same-worker CLAIMED fence means the original dispatch is still
+	// running. Handing the claim back used to double-deliver any job whose
+	// dispatch outlived the 30s job lease (a large album re-upload). The
+	// lease now decides — same worker or not. A crash-recovery replay never
+	// reaches this path: a restarted process runs under a NEW worker id and
+	// a FAILED fence re-claims explicitly.
 	again, _, err := s.store.FenceClaim(context.Background(), key, "w1",
 		float64(time.Now().Add(time.Minute).Unix()))
 	if err != nil {
 		t.Fatalf("reclaim: %v", err)
 	}
-	if again != job.FenceDispatch {
-		t.Fatalf("same-worker reclaim = %s, want dispatch (holder semantics)", again)
+	if again != job.FenceInFlight {
+		t.Fatalf("same-worker live-lease reclaim = %s, want in_flight", again)
+	}
+
+	// Expired lease → stale, never a dispatch opportunity.
+	staleKey := pgFenceKey(pgJob("j2", 101, "-100801"))
+	if _, _, err := s.store.FenceClaim(context.Background(), staleKey, "w1",
+		float64(time.Now().Add(-time.Minute).Unix())); err != nil {
+		t.Fatalf("stale claim: %v", err)
+	}
+	out, _, err := s.store.FenceClaim(context.Background(), staleKey, "w1",
+		float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("claim on expired fence: %v", err)
+	}
+	if out != job.FenceStale {
+		t.Fatalf("same-worker expired-lease claim = %s, want stale", out)
+	}
+
+	// FAILED fences stay re-claimable — the replay path.
+	failedKey := pgFenceKey(pgJob("j3", 102, "-100802"))
+	if _, _, err := s.store.FenceClaim(context.Background(), failedKey, "w1",
+		float64(time.Now().Add(time.Minute).Unix())); err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if err := s.store.FenceRecordOutcome(context.Background(), failedKey, "w1", "FAILED", "transient"); err != nil {
+		t.Fatalf("record outcome: %v", err)
+	}
+	replay, row, err := s.store.FenceClaim(context.Background(), failedKey, "w1",
+		float64(time.Now().Add(time.Minute).Unix()))
+	if err != nil {
+		t.Fatalf("reclaim after FAILED: %v", err)
+	}
+	if replay != job.FenceDispatch {
+		t.Fatalf("FAILED reclaim = %s, want dispatch", replay)
+	}
+	if row == nil || row.Attempts != 1 {
+		t.Fatalf("FAILED reclaim attempts = %+v, want 1", row)
 	}
 }
 
diff --git a/worker/internal/queue/sqlite.go b/worker/internal/queue/sqlite.go
index 1e528123..bc218f98 100644
--- a/worker/internal/queue/sqlite.go
+++ b/worker/internal/queue/sqlite.go
@@ -86,6 +86,12 @@ type Store interface {
 	FenceRecordSent(ctx context.Context, key job.AttemptKey, workerID string, targetMessageID int64) error
 	// FenceRecordOutcome stamps a FAILED or UNKNOWN outcome on the fence row.
 	FenceRecordOutcome(ctx context.Context, key job.AttemptKey, workerID, status, error string) error
+	// PurgeAgedFences deletes terminal fences (SENT/FAILED) older than the
+	// job-retention window, so the fence table cannot grow without bound
+	// the way the delivery_jobs table it guards does not. UNKNOWN fences are
+	// NEVER purged here: they are unresolved evidence.
+	PurgeAgedFences(ctx context.Context, olderThan time.Time, limit int) (int, error)
+
 	// FenceExpireStale moves a CLAIMED row whose lease passed to UNKNOWN —
 	// never back to "dispatch by default", because the send may have landed.
 	FenceExpireStale(ctx context.Context, now float64) (int, error)
@@ -719,82 +725,114 @@ func (s *sqliteStore) FenceClaim(ctx context.Context, key job.AttemptKey, worker
 		return "", nil, fmt.Errorf("fence begin: %w", err)
 	}
 	defer tx.Rollback()
-	if _, err := tx.ExecContext(ctx, `
-		INSERT INTO delivery_attempts (
-			id, rule_id, source_chat_id, source_message_id, target_chat_id,
-			job_kind, status, worker_id, lease_until, attempts,
-			last_error, created_at, updated_at
-		) VALUES (?, ?, ?, ?, ?, ?, 'CLAIMED', ?, ?, 0, '', ?, ?)
-		ON CONFLICT(id) DO UPDATE SET
-			status = 'CLAIMED',
-			worker_id = excluded.worker_id,
-			lease_until = excluded.lease_until,
-			attempts = delivery_attempts.attempts + 1,
-			last_error = '',
-			updated_at = excluded.updated_at
-		WHERE delivery_attempts.status = 'FAILED'`,
-		key.String(), key.RuleID, key.SourceChatID, key.SourceMessageID,
-		key.TargetChatID, key.JobKind, workerID, leaseUntil, now, now); err != nil {
-		return "", nil, fmt.Errorf("fence upsert: %w", err)
-	}
+	// Read the current state BEFORE any write: a re-claim upsert stamps the
+	// caller's worker id, so a post-upsert read cannot tell a legitimate
+	// FAILED re-claim apart from a live same-worker claim — and that
+	// distinction IS the idempotency contract (a dispatch that outlives its
+	// lease must never be handed to a second claim, or the job is sent
+	// twice).
 	row := tx.QueryRowContext(ctx, `
 		SELECT status, worker_id, lease_until, target_message_id, last_error, attempts
 		FROM delivery_attempts WHERE id=?`, key.String())
 	var (
-		a        job.DeliveryAttempt
-		targetID sql.NullInt64
+		preStatus string
+		preWorker string
+		preLease  float64
+		targetID  sql.NullInt64
+		preErr    string
+		preTries  int
 	)
-	if err := row.Scan(&a.Status, &a.WorkerID, &a.LeaseUntil, &targetID, &a.LastError, &a.Attempts); err != nil {
-		if err == sql.ErrNoRows {
-			// The upsert succeeded but the row cannot be read: treat it as
-			// unavailable rather than guessing — refuse to dispatch unprotected.
-			return "", nil, fmt.Errorf("fence row vanished after upsert for %s", key.String())
-		}
+	err = row.Scan(&preStatus, &preWorker, &preLease, &targetID, &preErr, &preTries)
+	if err != nil && err != sql.ErrNoRows {
 		return "", nil, fmt.Errorf("fence read: %w", err)
 	}
-	if err := tx.Commit(); err != nil {
-		return "", nil, fmt.Errorf("fence commit: %w", err)
+
+	var outcome job.FenceOutcome
+	a := job.DeliveryAttempt{}
+	switch {
+	case err == sql.ErrNoRows:
+		// First ever claim for this logical send.
+		if _, err := tx.ExecContext(ctx, `
+			INSERT INTO delivery_attempts (
+				id, rule_id, source_chat_id, source_message_id, target_chat_id,
+				job_kind, status, worker_id, lease_until, attempts,
+				last_error, created_at, updated_at
+			) VALUES (?, ?, ?, ?, ?, ?, 'CLAIMED', ?, ?, 0, '', ?, ?)`,
+			key.String(), key.RuleID, key.SourceChatID, key.SourceMessageID,
+			key.TargetChatID, key.JobKind, workerID, leaseUntil, now, now); err != nil {
+			return "", nil, fmt.Errorf("fence insert: %w", err)
+		}
+		outcome = job.FenceDispatch
+		a.Status, a.WorkerID, a.LeaseUntil, a.Attempts = "CLAIMED", workerID, leaseUntil, 0
+	case preStatus == "SENT":
+		outcome = job.FenceAlreadySent
+		a.Status, a.WorkerID, a.LeaseUntil, a.LastError, a.Attempts =
+			preStatus, preWorker, preLease, preErr, preTries
+	case preStatus == "UNKNOWN":
+		// An earlier send's outcome was never recorded. Never dispatch
+		// blindly — the reconciler / an operator decides.
+		outcome = job.FenceUnknown
+		a.Status, a.WorkerID, a.LeaseUntil, a.LastError, a.Attempts =
+			preStatus, preWorker, preLease, preErr, preTries
+	case preStatus == "CLAIMED" || preStatus == "SENDING":
+		// CLAIMED/SENDING rows are never stolen, live or expired, same
+		// worker or not: every goroutine in this process shares one worker
+		// id, so a same-worker claim means the original dispatch is still
+		// running. Expiry is the stale sweep's + reconciler's decision, not
+		// a dispatch opportunity.
+		if preLease > 0 && preLease >= float64(now) {
+			outcome = job.FenceInFlight
+		} else {
+			outcome = job.FenceStale
+		}
+		a.Status, a.WorkerID, a.LeaseUntil, a.LastError, a.Attempts =
+			preStatus, preWorker, preLease, preErr, preTries
+	default:
+		// FAILED (the replay path): the caller re-claims it.
+		if _, err := tx.ExecContext(ctx, `
+			UPDATE delivery_attempts
+			SET status='CLAIMED', worker_id=?, lease_until=?, attempts=attempts+1,
+			    last_error='', updated_at=?
+			WHERE id=? AND status='FAILED'`,
+			workerID, leaseUntil, now, key.String()); err != nil {
+			return "", nil, fmt.Errorf("fence reclaim: %w", err)
+		}
+		outcome = job.FenceDispatch
+		a.Status, a.WorkerID, a.LeaseUntil, a.Attempts = "CLAIMED", workerID, leaseUntil, preTries+1
 	}
 	if targetID.Valid {
 		a.TargetMessageID = targetID.Int64
 	}
-
-	switch a.Status {
-	case "SENT":
-		return job.FenceAlreadySent, &a, nil
-	case "UNKNOWN":
-		return job.FenceUnknown, &a, nil
-	case "FAILED":
-		// The UPSERT re-claimed it for this caller.
-		return job.FenceDispatch, &a, nil
-	}
-	if a.WorkerID == workerID {
-		return job.FenceDispatch, &a, nil
-	}
-	if a.LeaseUntil > 0 && a.LeaseUntil >= float64(now) {
-		return job.FenceInFlight, &a, nil
+	if err := tx.Commit(); err != nil {
+		return "", nil, fmt.Errorf("fence commit: %w", err)
 	}
-	return job.FenceStale, &a, nil
+	return outcome, &a, nil
 }
 
 // FenceRecordSent stamps a confirmed Telegram result. It is written BEFORE
 // the job-row commit on purpose: a crash between the Telegram response and
-// the batch commit must leave the fence saying the send landed.
+// the batch commit must leave the fence saying the send landed. A SENT fence
+// is terminal — a concurrent sibling's late failure must never overwrite it
+// (parity with the PostgreSQL store and the Python repository, which both
+// carry this guard).
 func (s *sqliteStore) FenceRecordSent(ctx context.Context, key job.AttemptKey, workerID string, targetMessageID int64) error {
 	_, err := s.db.ExecContext(ctx, `
 		UPDATE delivery_attempts
 		SET status='SENT', worker_id=?, target_message_id=?, updated_at=?
-		WHERE id=?`, workerID, targetMessageID, time.Now().Unix(), key.String())
+		WHERE id=? AND status <> 'SENT'`, workerID, targetMessageID, time.Now().Unix(), key.String())
 	return err
 }
 
 // FenceRecordOutcome stamps a FAILED outcome, releasing the fence for a
 // retry. Only call this when the Telegram failure is *unambiguous* — a
-// timeout's outcome is UNKNOWN, not FAILED.
+// timeout's outcome is UNKNOWN, not FAILED. A SENT fence is terminal: the
+// message is physically in the target, so a FAILED stamp here would turn a
+// confirmed delivery back into a re-dispatchable job (parity with the
+// PostgreSQL store and the Python repository).
 func (s *sqliteStore) FenceRecordOutcome(ctx context.Context, key job.AttemptKey, workerID, status, detail string) error {
 	_, err := s.db.ExecContext(ctx, `
 		UPDATE delivery_attempts SET status=?, worker_id=?, last_error=?, updated_at=?
-		WHERE id=?`, status, workerID, detail, time.Now().Unix(), key.String())
+		WHERE id=? AND status NOT IN ('SENT')`, status, workerID, detail, time.Now().Unix(), key.String())
 	return err
 }
 
@@ -812,6 +850,25 @@ func (s *sqliteStore) FenceExpireStale(ctx context.Context, now float64) (int, e
 	return int(n), nil
 }
 
+// PurgeAgedFences deletes terminal fences older than the retention window —
+// bounded, indexed by updated_at, and blind to UNKNOWN rows (unresolved
+// evidence is never deleted by a sweep). The 127 orphaned SENT fences the
+// production audit found (jobs purged at 7d, fences forever) are the growth
+// this stops.
+func (s *sqliteStore) PurgeAgedFences(ctx context.Context, olderThan time.Time, limit int) (int, error) {
+	res, err := s.db.ExecContext(ctx, `
+		DELETE FROM delivery_attempts WHERE id IN (
+			SELECT id FROM delivery_attempts
+			WHERE status IN ('SENT','FAILED') AND updated_at < ?
+			LIMIT ?
+		)`, olderThan.Unix(), limit)
+	if err != nil {
+		return 0, err
+	}
+	n, _ := res.RowsAffected()
+	return int(n), nil
+}
+
 // FenceResolveUnknown reopens an UNKNOWN fence after the caller has PROVED
 // (a Telegram-side read or a bounded recovery decision) that no earlier copy
 // is in the target. The reason is recorded so an operator can audit why a
diff --git a/worker/internal/telegraph/fakegateway/fake.go b/worker/internal/telegraph/fakegateway/fake.go
index b3ba8b71..2d8e108f 100644
--- a/worker/internal/telegraph/fakegateway/fake.go
+++ b/worker/internal/telegraph/fakegateway/fake.go
@@ -19,6 +19,7 @@
 package fakegateway
 
 import (
+	"os"
 	"context"
 	"fmt"
 	"sync"
@@ -42,8 +43,14 @@ type Fake struct {
 	SendFailsWith             *telegraph.DeliveryFailure
 	CopyMediaGroupFailsWith   *telegraph.DeliveryFailure
 	DownloadFailsWith         *telegraph.DeliveryFailure
+	SendMediaFailsWith        *telegraph.DeliveryFailure
 	GetFileExtensionFailsWith *telegraph.DeliveryFailure
 
+	// DownloadWritesFile makes DownloadMedia actually create the
+	// destination file, so tests can pin the staged-file lifecycle (the
+	// worker must clean the scratch file up whichever way the send ends).
+	DownloadWritesFile bool
+
 	// ExtensionOverride, when set, is what GetFileExtension answers instead
 	// of the default ".bin". Tests use it to pin the staging-name contract
 	// for both "the runtime knows the extension" and "the runtime cannot
@@ -165,6 +172,11 @@ func (f *Fake) DownloadMedia(ctx context.Context, sessionID string, chatID, mess
 	if f.DownloadFailsWith != nil {
 		return 0, f.DownloadFailsWith
 	}
+	if f.DownloadWritesFile {
+		if err := os.WriteFile(dest.Path, []byte("staged-bytes"), 0o600); err != nil {
+			return 0, &telegraph.DeliveryFailure{Code: "MEDIA_DOWNLOAD_FAILED", Message: err.Error()}
+		}
+	}
 	return 1024, nil
 }
 
@@ -258,6 +270,9 @@ func (f *Fake) CopyMediaGroup(ctx context.Context, sessionID string, targetChatI
 // SendMedia uploads a staged file as new media.
 func (f *Fake) SendMedia(ctx context.Context, sessionID string, targetChatID int64, token, caption, mediaType string, threadID int64) (int64, *telegraph.DeliveryFailure) {
 	f.record(Call{Op: "SendMedia", SessionID: sessionID, TargetID: targetChatID, Caption: caption, MediaPath: token, ThreadID: threadID})
+	if f.SendMediaFailsWith != nil {
+		return 0, f.SendMediaFailsWith
+	}
 	if f.SendFailsWith != nil {
 		return 0, f.SendFailsWith
 	}