fix(delivery): one dispatch per job — fence lease, album ambiguity, bot-route media rescue, staged-file hygiene

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:

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

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),
                     ),
                 )

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
 # --------------------------------------------------------------------------- #

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).