fix(delivery): PHOTO_EXT_INVALID root cause — re-upload staging now carries the real file extension (+ fallback extended to target-side media rejection)

.github/workflows/ci.yml

@@ -60,7 +60,9 @@ jobs:
         # big locale-scan files (80s+ each) still land on separate workers.
         env:
           ATF_TEST_REDIS_ADDR: 127.0.0.1:6379
-        run: python -m pytest -q -n auto --dist loadfile core/tests scripts/tests
+        # -rs: every skipped test prints its skip reason in the log — a
+        # skipped suite must be inspectable, not just countable.
+        run: python -m pytest -q -rs -n auto --dist loadfile core/tests scripts/tests
 
       - name: Telegram UI hardcoded-string guard
         # New visible strings in Telegram UI handlers, callbacks, and button

core/application/delivery_reconciler.py

@@ -214,11 +214,20 @@ async def _check_orphan_mappings(self, owner_user_id: Optional[int]) -> List[Dic
         job was deleted while its mapping survived (or vice versa): the copy
         may or may not be in the destination, and nothing currently drives
         it to a conclusion.
+
+        Retention asymmetry is NOT a mismatch: terminal delivery jobs purge
+        after 7 days while mappings keep 30, so a mapping older than the
+        job-retention window whose job is gone is an expected outcome of
+        cleanup, not an anomaly — flagging it every sweep produced 58
+        identical production warnings for rows nothing can act on. Only
+        mappings inside the job-retention window are findings.
         """
         scope, params = await self._scope(owner_user_id)
         db = self._db()
         if db is None:
             return []
+        job_retention_seconds = 60 * 60 * 24 * 7  # mirrors the delivery_jobs cleanup window
+        cutoff = int(time.time()) - job_retention_seconds
         rows = await self._query(
             db,
             "SELECT m.rule_id, m.source_chat_id, m.source_message_id, m.target_chat_id, "
@@ -227,8 +236,8 @@ async def _check_orphan_mappings(self, owner_user_id: Optional[int]) -> List[Dic
             "ON j.rule_id = m.rule_id AND j.source_chat_id = m.source_chat_id "
             "AND j.source_message_id = m.source_message_id "
             "AND j.target_chat_id = m.target_chat_id "
-            "WHERE j.id IS NULL" + scope.replace("rule_id", "m.rule_id"),
-            params,
+            "WHERE j.id IS NULL AND m.updated_at >= ?" + scope.replace("rule_id", "m.rule_id"),
+            (cutoff,) + tuple(params),
         )
         findings = []
         for r in rows or ():

core/domain/entities.py

@@ -1840,20 +1840,38 @@ def watermark_text(self, val: str) -> None:
             self.metadata = {}
         self.metadata["watermark_text"] = str(val or "")
 
+    # Paid-addon keys the catalogue clients write, mapped onto the engine's
+    # canonical gate keys. The bot catalogue rows (pro_bot_ui _ADDON_ROWS)
+    # and the web dashboard's add-ons panel predate the engine key
+    # vocabulary and still record the short spellings; a paid add-on whose
+    # entitlement was recorded by ANY client must fire — silently never
+    # firing is the worst outcome for a feature a user paid for and enabled.
+    ADDON_KEY_ALIASES = {
+        "keyword_reply": ("kw_reply",),
+        "auto_delete_message": ("auto_del",),
+        "affiliate_kit": ("affiliate",),
+    }
+
     def is_addon_enabled(self, key: str) -> bool:
         """Server-side paid-addon entitlement check.
 
         A client-supplied `paid_addons` flag is NEVER trusted on its own: the
         addon must be marked enabled in the persisted rule metadata. This is
         the single enforcement point consulted by the pipeline before any
-        premium transformation runs.
+        premium transformation runs. Canonical keys and the catalogue
+        clients' recorded aliases are both honored (see ADDON_KEY_ALIASES).
         """
         if not isinstance(self.metadata, dict):
             return False
         addons = self.metadata.get("paid_addons", {})
         if not isinstance(addons, dict):
             return False
-        return bool(addons.get(key, False))
+        if bool(addons.get(key, False)):
+            return True
+        for alias in self.ADDON_KEY_ALIASES.get(key, ()):
+            if bool(addons.get(alias, False)):
+                return True
+        return False
 
     @property
     def link_replacement(self) -> str:

core/infrastructure/persistence/postgres_repositories.py

@@ -1368,6 +1368,38 @@ async def replay_dead_letter(
                         payload, max(1, max_attempts), now, now,
                     ),
                 )
+                # Restore the placeholder mapping row the enqueue path
+                # guarantees (parity with the SQLite replay): mapping rows are
+                # purged after 30 days while dead-letter entries persist, so a
+                # rebuilt forward job must not stay mapping-less — the
+                # worker's NoteMapping only UPDATEs and would silently drop
+                # the confirmed target id. The never-regress guard keeps a
+                # mapping row that still says SENT exactly as it is.
+                try:
+                    payload_data = json.loads(payload) if isinstance(payload, str) else {}
+                except Exception:
+                    payload_data = {}
+                if not (payload_data or {}).get("is_reply"):
+                    # Replies share the mapping key with the primary forward;
+                    # writing here would refresh another deliverable's row.
+                    await _execute_db(
+                        self._db,
+                        "INSERT INTO message_map ("
+                        " id, rule_id, source_chat_id, source_message_id, target_chat_id,"
+                        " target_message_id, media_group_id, delivery_status, created_at, updated_at"
+                        ") VALUES (%s, %s, %s, %s, %s, NULL, %s, 'PENDING', %s, %s)"
+                        " ON CONFLICT (rule_id, source_chat_id, source_message_id, target_chat_id) DO UPDATE SET"
+                        " delivery_status = CASE"
+                        " WHEN message_map.delivery_status IN ('SENT') THEN message_map.delivery_status"
+                        " ELSE EXCLUDED.delivery_status END,"
+                        " updated_at = EXCLUDED.updated_at",
+                        (
+                            f"map:{rule_id}:{src_chat}:{src_msg}:{tgt_chat}",
+                            rule_id, src_chat, src_msg, tgt_chat,
+                            str((payload_data or {}).get("media_group_id") or ""),
+                            now, now,
+                        ),
+                    )
             await _execute_db(self._db, "DELETE FROM dead_letter_queue WHERE id=%s", (str(entry_id),))
         except Exception as exc:
             logger.warning("replay of dead letter %s failed: %s", entry_id, exc)
@@ -1744,6 +1776,18 @@ async def resolve_gap_for_source(self, source_chat_id: str, message_id: int) ->
         count = getattr(res, "rowcount", None)
         return (count > 0) if count is not None else True
 
+    async def cleanup(self, retention_seconds: int = 60 * 60 * 24 * 30) -> int:
+        """Bounded retention for terminal gap rows (parity with SQLite)."""
+        cutoff = int(time.time()) - int(retention_seconds)
+        res = await _execute_db(
+            self._db,
+            "DELETE FROM ingestion_gaps"
+            " WHERE state IN ('RESOLVED','FAILED') AND updated_at < %s",
+            (cutoff,),
+        )
+        count = getattr(res, "rowcount", None)
+        return int(count) if count is not None else 0
+
     async def pending_gaps(self, session_id: str, source_chat_id: str,
                            limit: int = 100) -> List[dict]:
         rows = await _fetch_all_db(

core/infrastructure/persistence/sqlite_repositories.py

@@ -2497,6 +2497,43 @@ def _unit(conn) -> bool:
                     # The unique key answered after all: a row for this
                     # delivery already exists (someone replayed it first).
                     return False
+                # Restore the placeholder mapping row the enqueue path
+                # guarantees. Cleanup purges message_map rows after 30 days
+                # while dead-letter entries persist, so a replay of a purged
+                # row would otherwise rebuild a forward job whose mapping row
+                # never returns — and the worker's NoteMapping (UPDATE-only)
+                # would silently drop the confirmed target id, leaving
+                # edit/delete sync blind to this delivery. The upsert is the
+                # enqueue path's own never-regress shape: a mapping row that
+                # still says SENT is never flipped back.
+                try:
+                    payload_data = SqliteDatabase.loads(payload, {})
+                except Exception:
+                    payload_data = {}
+                if not payload_data.get("is_reply"):
+                    # Replies share the mapping key with the primary forward
+                    # (durable_queue._write_mapping_and_job); writing here
+                    # would refresh a row that describes another deliverable.
+                    conn.execute(
+                        """
+                        INSERT INTO message_map (
+                            id, rule_id, source_chat_id, source_message_id, target_chat_id,
+                            target_message_id, media_group_id, delivery_status, created_at, updated_at
+                        ) VALUES (?, ?, ?, ?, ?, NULL, ?, 'PENDING', ?, ?)
+                        ON CONFLICT(rule_id, source_chat_id, source_message_id, target_chat_id) DO UPDATE SET
+                            delivery_status = CASE
+                                WHEN message_map.delivery_status IN ('SENT') THEN message_map.delivery_status
+                                ELSE excluded.delivery_status
+                            END,
+                            updated_at = excluded.updated_at
+                        """,
+                        (
+                            f"map:{rule_id}:{src_chat}:{src_msg}:{tgt_chat}",
+                            rule_id, src_chat, src_msg, tgt_chat,
+                            str(payload_data.get("media_group_id") or ""),
+                            now, now,
+                        ),
+                    )
                 conn.execute("DELETE FROM dead_letter_queue WHERE id=?", (str(entry_id),))
                 return True
 
@@ -2959,6 +2996,25 @@ async def pending_gaps(self, session_id: str, source_chat_id: str,
         )
         return [dict(r) for r in rows or []]
 
+    async def cleanup(self, retention_seconds: int = 60 * 60 * 24 * 30) -> int:
+        """Bounded retention for terminal gap rows.
+
+        ``FAILED`` rows are deliberate loss records ("visibly, never
+        silently") — but terminal records are not exempt from bounded
+        storage, and without a sweep the table grows monotonically with
+        every permanently-absent id. Terminal rows older than the window go,
+        on the same 30-day retention the message map applies to delivery
+        confirmations. Open rows (``PENDING``/``RETRYING``) are live
+        recovery work and are never touched.
+        """
+        cutoff = _now() - int(retention_seconds)
+        cur = self._db.execute(
+            "DELETE FROM ingestion_gaps"
+            " WHERE state IN ('RESOLVED','FAILED') AND updated_at < ?",
+            (cutoff,),
+        )
+        return cur.rowcount
+
     async def open_gaps(self, session_id: str, source_chat_id: str,
                         below: int = 0, limit: int = 100) -> List[dict]:
         """Still-open recovery work for one channel.

core/infrastructure/telegram/channel_polling.py

@@ -325,6 +325,7 @@ async def stop(self) -> None:
             self._task = None
 
     async def _run(self) -> None:
+        last_gap_retention = 0.0
         while not self._stop.is_set():
             try:
                 if await self._may_sweep():
@@ -335,6 +336,23 @@ async def _run(self) -> None:
                 # are still lost, just with a row saying so.
                 if self._gaps is not None:
                     await self._recover_gaps()
+                    # Gap retention: terminal (FAILED) rows older than the
+                    # 30-day window go, on the same hourly cadence the
+                    # durable queue's retention sweep uses. Without it the
+                    # table grows monotonically with every permanently-absent
+                    # id. One indexed DELETE per hour; a failure only delays
+                    # the next sweep.
+                    now_ts = time.time()
+                    if now_ts - last_gap_retention >= 3600:
+                        last_gap_retention = now_ts
+                        try:
+                            purged = await self._gaps.cleanup()
+                            if purged:
+                                logger.info(
+                                    "Gap retention purged %d terminal row(s).", purged
+                                )
+                        except Exception as exc:
+                            logger.warning("poll: gap retention sweep failed: %s", exc)
             except asyncio.CancelledError:
                 raise
             except Exception as exc:
@@ -397,8 +415,8 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic
         gap_id = str(gap.get("id") or "")
         mid = int(gap.get("source_message_id") or 0)
         try:
-            from pyrogram.raw.functions.updates import GetChannelDifference
-            from pyrogram.raw.types import ChannelMessagesFilter, MessageRange
+            from pyrogram.raw.functions.channels import GetMessages as ChannelGetMessages
+            from pyrogram.raw.types import InputMessageID, MessageEmpty
 
             peer = None
             try:
@@ -407,22 +425,24 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic
             except Exception as exc:
                 raise PollFailure(f"cannot resolve peer: {exc}", transient=_is_transient(exc)) from exc
 
-            full_pts = 0
-            try:
-                from pyrogram.raw.functions.channels import GetFullChannel
-                full = await client.invoke(GetFullChannel(channel=peer))
-                full_pts = getattr(getattr(full, "full_chat", None), "pts", 0) or 0
-            except Exception:
-                pass  # pts is only a hint for the difference call
-
-            msg_filter = ChannelMessagesFilter(ranges=[MessageRange(min_id=mid, max_id=mid)])
-            diff = await client.invoke(
-                GetChannelDifference(channel=peer, filter=msg_filter, pts=full_pts, limit=1)
+            # Fetch the id directly with channels.getMessages. The earlier
+            # GetChannelDifference approach needed the channel's pts from
+            # GetFullChannel — which Telegram flood-limits (275 rate-limit
+            # waits over one production weekend), and when it failed the
+            # code silently fell back to pts=0, where Telegram rejects the
+            # difference call with PERSISTENT_TIMESTAMP_EMPTY (450
+            # occurrences in the same window): every gap's retry budget was
+            # burned on a request that could never succeed, and the posts
+            # were lost. getMessages answers "does this id exist?" with one
+            # call, no pts, and no flood-sensitive full-channel round-trip.
+            res = await client.invoke(
+                ChannelGetMessages(channel=peer, id=[InputMessageID(id=mid)])
             )
-            raw_messages = getattr(diff, "new_messages", None) or []
+            users = {u.id: u for u in (getattr(res, "users", None) or [])}
+            chats = {c.id: c for c in (getattr(res, "chats", None) or [])}
+            raw_messages = [m for m in (getattr(res, "messages", None) or [])
+                            if not isinstance(m, MessageEmpty)]
             if raw_messages:
-                users = {u.id: u for u in (getattr(diff, "users", None) or [])}
-                chats = {c.id: c for c in (getattr(diff, "chats", None) or [])}
                 for raw in raw_messages:
                     logger.info(
                         "[POLL GAP] recovered post id=%s from %s — dispatching through the pipeline",
@@ -448,19 +468,21 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic
                 exhausted=exhausted,
             )
             if exhausted:
-                logger.error(
-                    "[POLL GAP] source %s id=%s could not be recovered after %d attempt(s) — "
-                    "marked FAILED (the post is most likely deleted; no delivery is owed)",
-                    chat_id, mid, GAP_MAX_ATTEMPTS,
-                )
+                # The structured event below is the ONE authoritative record
+                # of this exhaustion (it carries the reason code, the
+                # diagnostic row, and the Log Monitor entry). A second
+                # logger.error of the same fact tripled every terminal row in
+                # the stores (473 + 463 + 463 production rows for one event
+                # class) without adding a single field.
                 self._emit("error", "INGESTION_GAP_EXHAUSTED", (
-                    f"channel {chat_id}: id={mid} not returned after {GAP_MAX_ATTEMPTS} "
-                    f"attempt(s) — marked FAILED (deleted or service post; no delivery owed)"
+                    f"[POLL GAP] source {chat_id} id={mid} could not be recovered after "
+                    f"{GAP_MAX_ATTEMPTS} attempt(s) — marked FAILED (the post is most "
+                    f"likely deleted; no delivery is owed)"
                 ), {
                     "chat_id": str(chat_id), "session_id": str(session_id),
                     "source_message_id": mid, "attempts": GAP_MAX_ATTEMPTS,
                     "raw_error": "ID_NOT_RETURNED",
-                    "error_detail": "GetChannelDifference returned no message at this id",
+                    "error_detail": "channels.GetMessages returned no message at this id",
                     "retryable": False,
                 })
             return 1
@@ -483,7 +505,7 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic
                 # a paraphrase of it.
                 from .errors import extract as _extract_error
                 try:
-                    event = _extract_error(exc, method="updates.GetChannelDifference",
+                    event = _extract_error(exc, method="channels.GetMessages",
                                            operation="gap_recovery", chat_id=chat_id,
                                            session_id=session_id, attempt=attempts)
                     raw = str(getattr(event, "error_type", "") or "")