feat(gateway): persist outbound delivery outcomes

cron/scheduler.py

@@ -12,6 +12,7 @@ import asyncio
 import atexit
 import concurrent.futures
 import contextvars
+import hashlib
 import json
 import logging
 import os
@@ -1401,6 +1402,20 @@ def _is_channel_dm_topic(
     return is_channel
 
 
+def _cron_delivery_id(job: dict, platform: str, chat_id: str, payload: str) -> str:
+    """Stable identity for one cron run's text fan-out target."""
+    identity = "\0".join(
+        (
+            str(job.get("id", "unknown")),
+            str(job.get("last_run_at") or job.get("next_run_at") or "unscheduled"),
+            platform,
+            chat_id,
+            hashlib.sha256(payload.encode("utf-8")).hexdigest(),
+        )
+    )
+    return "cron-" + hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32]
+
+
 def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]:
     """
     Deliver job output to the configured target(s) (origin chat, specific platform, etc.).
@@ -1697,6 +1712,12 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                         thread_id=route_thread_id,
                         is_explicit=True,
                     )
+                    # A stable per-run/per-target identity makes cron fan-out
+                    # restart-aware without conflating distinct destinations.
+                    route_metadata["delivery_id"] = _cron_delivery_id(
+                        job, platform_name, str(chat_id), text_to_send,
+                    )
+                    route_metadata["delivery_origin"] = f"cron:{job['id']}"
                     # Pass thread routing via the target (not a bare metadata
                     # "thread_id"): the router only applies its Telegram DM-topic
                     # detection when "thread_id"/"message_thread_id" are absent

gateway/delivery.py

@@ -11,6 +11,7 @@ Routes messages to the appropriate destination based on:
 import logging
 import os
 import re
+import uuid
 from pathlib import Path
 from datetime import datetime
 from dataclasses import dataclass
@@ -57,6 +58,7 @@ def _is_silence_narration(content: Optional[str]) -> bool:
 from .config import Platform, GatewayConfig
 from .session import SessionSource
 from .dead_targets import DeadTargetRegistry
+from .delivery_outbox import DeliveryOutbox, UnknownDeliveryError
 
 
 def looks_like_telegram_private_chat_id(chat_id: Optional[str]) -> bool:
@@ -242,6 +244,9 @@ class DeliveryRouter:
         self.adapters = adapters or {}
         self.output_dir = get_hermes_home() / "cron" / "output"
         self.dead_targets = dead_targets or DeadTargetRegistry()
+        self.outbox = DeliveryOutbox(
+            get_hermes_home() / "gateway" / "delivery_outbox.sqlite3"
+        )
     
     async def deliver(
         self,
@@ -399,6 +404,25 @@ class DeliveryRouter:
         
         if not target.chat_id:
             raise ValueError(f"No chat ID for {target.platform.value} delivery")
+
+        # Cron supplies a stable id so a restart addresses the same record.
+        # Other callers retain one-shot behavior via a generated id.
+        metadata = dict(metadata or {})
+        delivery_id = str(metadata.get("delivery_id") or uuid.uuid4())
+        origin = str(metadata.get("delivery_origin") or metadata.get("job_id") or "gateway")
+        record = self.outbox.create(
+            delivery_id, origin, target.to_string(), content,
+        )
+        if record.state == "confirmed":
+            return {
+                "success": True,
+                "deduplicated": True,
+                "provider_receipt": record.provider_receipt,
+            }
+        if record.state in {"dispatched", "unknown"}:
+            raise UnknownDeliveryError(
+                f"delivery {delivery_id} may already have been sent and is not retried automatically"
+            )
         
         # Guard: handle oversized cron output.
         #
@@ -465,6 +489,8 @@ class DeliveryRouter:
                 target.chat_id,
                 content[:40],
             )
+            receipt = {"filtered": "silence_narration", "delivered": False}
+            self.outbox.mark_confirmed(delivery_id, receipt)
             return {
                 "success": True,
                 "filtered": "silence_narration",
@@ -524,32 +550,60 @@ class DeliveryRouter:
                 send_metadata["telegram_dm_topic_reply_fallback"] = True
             elif "thread_id" not in send_metadata and "message_thread_id" not in send_metadata and not has_explicit_direct_topic:
                 send_metadata["thread_id"] = target_thread_id
-        result = await adapter.send(target.chat_id, content, metadata=send_metadata or None)
-        if _send_result_failed(result):
-            if (
-                is_named_telegram_private_topic
-                and named_telegram_private_topic_name
-                and _is_thread_not_found_delivery_error(result)
-            ):
-                ensure_dm_topic = getattr(adapter, "ensure_dm_topic", None)
-                if ensure_dm_topic is None:
-                    raise RuntimeError(
-                        "Telegram adapter cannot refresh named private DM topics"
-                    )
-                refreshed_thread_id = await ensure_dm_topic(
-                    target.chat_id,
-                    named_telegram_private_topic_name,
-                    force_create=True,
-                )
-                if not refreshed_thread_id:
-                    raise RuntimeError(
-                        f"Failed to refresh Telegram private DM topic '{named_telegram_private_topic_name}'"
-                    )
-                send_metadata["thread_id"] = str(refreshed_thread_id)
-                send_metadata["telegram_dm_topic_created_for_send"] = True
-                result = await adapter.send(target.chat_id, content, metadata=send_metadata or None)
+        self.outbox.mark_dispatched(delivery_id)
+        try:
+            result = await adapter.send(target.chat_id, content, metadata=send_metadata or None)
             if _send_result_failed(result):
-                raise RuntimeError(_send_result_error(result) or f"{target.platform.value} delivery failed")
+                if (
+                    is_named_telegram_private_topic
+                    and named_telegram_private_topic_name
+                    and _is_thread_not_found_delivery_error(result)
+                ):
+                    ensure_dm_topic = getattr(adapter, "ensure_dm_topic", None)
+                    if ensure_dm_topic is None:
+                        raise RuntimeError(
+                            "Telegram adapter cannot refresh named private DM topics"
+                        )
+                    refreshed_thread_id = await ensure_dm_topic(
+                        target.chat_id,
+                        named_telegram_private_topic_name,
+                        force_create=True,
+                    )
+                    if not refreshed_thread_id:
+                        raise RuntimeError(
+                            f"Failed to refresh Telegram private DM topic '{named_telegram_private_topic_name}'"
+                        )
+                    send_metadata["thread_id"] = str(refreshed_thread_id)
+                    send_metadata["telegram_dm_topic_created_for_send"] = True
+                    result = await adapter.send(target.chat_id, content, metadata=send_metadata or None)
+                if _send_result_failed(result):
+                    error = _send_result_error(result) or f"{target.platform.value} delivery failed"
+                    self.outbox.mark_failed(delivery_id, error)
+                    raise RuntimeError(error)
+        except BaseException as exc:
+            # Cancellation means the non-idempotent send may already have reached
+            # the provider. Never convert that uncertainty into an automatic retry.
+            if isinstance(exc, (KeyboardInterrupt, SystemExit)):
+                raise
+            current = self.outbox.get(delivery_id)
+            if current and current.state == "dispatched":
+                if isinstance(exc, Exception):
+                    self.outbox.mark_failed(delivery_id, str(exc))
+                else:
+                    self.outbox.mark_unknown(delivery_id, type(exc).__name__)
+            raise
+
+        receipt = {
+            "message_id": (
+                result.get("message_id") if isinstance(result, dict)
+                else getattr(result, "message_id", None)
+            ),
+            "raw_response": (
+                result.get("raw_response") if isinstance(result, dict)
+                else getattr(result, "raw_response", None)
+            ),
+        }
+        self.outbox.mark_confirmed(delivery_id, receipt)
         return result
 
 

gateway/delivery_outbox.py

@@ -0,0 +1,177 @@
+"""Profile-local durable state for outbound platform deliveries."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+from hermes_constants import get_hermes_home
+
+
+_STATES = frozenset({"pending", "dispatched", "confirmed", "unknown", "failed"})
+
+
+class UnknownDeliveryError(RuntimeError):
+    """Raised when a possibly-sent delivery cannot safely be retried."""
+
+
+@dataclass(frozen=True)
+class DeliveryRecord:
+    delivery_id: str
+    origin: str
+    destination: str
+    payload_hash: str
+    state: str
+    attempt: int
+    created_at: str
+    updated_at: str
+    dispatched_at: Optional[str]
+    confirmed_at: Optional[str]
+    provider_receipt: Optional[dict[str, Any]]
+    last_error: Optional[str]
+
+
+class DeliveryOutbox:
+    """Small SQLite outbox recording certainty around non-idempotent sends."""
+
+    def __init__(self, path: Optional[Path] = None):
+        self.path = Path(path) if path else get_hermes_home() / "gateway" / "delivery_outbox.sqlite3"
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._conn = sqlite3.connect(self.path, timeout=5, check_same_thread=False)
+        self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA busy_timeout=5000")
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS deliveries (
+                delivery_id TEXT PRIMARY KEY,
+                origin TEXT NOT NULL,
+                destination TEXT NOT NULL,
+                payload_hash TEXT NOT NULL,
+                state TEXT NOT NULL CHECK(state IN ('pending','dispatched','confirmed','unknown','failed')),
+                attempt INTEGER NOT NULL DEFAULT 0,
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL,
+                dispatched_at TEXT,
+                confirmed_at TEXT,
+                provider_receipt TEXT,
+                last_error TEXT
+            )
+            """
+        )
+        self._conn.commit()
+        self.reconcile_restart()
+
+    @staticmethod
+    def _now() -> str:
+        return datetime.now(timezone.utc).isoformat()
+
+    @staticmethod
+    def hash_payload(payload: str) -> str:
+        return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def _record(row: sqlite3.Row) -> DeliveryRecord:
+        values = dict(row)
+        receipt = values.pop("provider_receipt")
+        values["provider_receipt"] = json.loads(receipt) if receipt else None
+        return DeliveryRecord(**values)
+
+    def close(self) -> None:
+        self._conn.close()
+
+    def reconcile_restart(self) -> int:
+        """Make crash-interrupted sends explicitly uncertain, never retryable."""
+        now = self._now()
+        cursor = self._conn.execute(
+            "UPDATE deliveries SET state='unknown', updated_at=? WHERE state='dispatched'",
+            (now,),
+        )
+        self._conn.commit()
+        return cursor.rowcount
+
+    def get(self, delivery_id: str) -> Optional[DeliveryRecord]:
+        row = self._conn.execute(
+            "SELECT * FROM deliveries WHERE delivery_id=?", (delivery_id,)
+        ).fetchone()
+        return self._record(row) if row else None
+
+    def create(
+        self,
+        delivery_id: str,
+        origin: str,
+        destination: str,
+        payload: str,
+    ) -> DeliveryRecord:
+        payload_hash = self.hash_payload(payload)
+        existing = self.get(delivery_id)
+        if existing:
+            identity = (origin, destination, payload_hash)
+            if identity != (existing.origin, existing.destination, existing.payload_hash):
+                raise ValueError(f"delivery id {delivery_id!r} already identifies a different delivery")
+            return existing
+        now = self._now()
+        self._conn.execute(
+            "INSERT INTO deliveries "
+            "(delivery_id, origin, destination, payload_hash, state, attempt, created_at, updated_at) "
+            "VALUES (?, ?, ?, ?, 'pending', 0, ?, ?)",
+            (delivery_id, origin, destination, payload_hash, now, now),
+        )
+        self._conn.commit()
+        return self.get(delivery_id)  # type: ignore[return-value]
+
+    def _transition(
+        self,
+        delivery_id: str,
+        state: str,
+        *,
+        receipt: Optional[dict[str, Any]] = None,
+        error: Optional[str] = None,
+    ) -> DeliveryRecord:
+        if state not in _STATES:
+            raise ValueError(f"invalid delivery state: {state}")
+        now = self._now()
+        dispatched_at = now if state == "dispatched" else None
+        confirmed_at = now if state == "confirmed" else None
+        increment = 1 if state == "dispatched" else 0
+        self._conn.execute(
+            "UPDATE deliveries SET state=?, attempt=attempt+?, updated_at=?, "
+            "dispatched_at=COALESCE(?, dispatched_at), confirmed_at=COALESCE(?, confirmed_at), "
+            "provider_receipt=COALESCE(?, provider_receipt), last_error=? WHERE delivery_id=?",
+            (
+                state,
+                increment,
+                now,
+                dispatched_at,
+                confirmed_at,
+                json.dumps(receipt, sort_keys=True, default=str) if receipt is not None else None,
+                error,
+                delivery_id,
+            ),
+        )
+        self._conn.commit()
+        record = self.get(delivery_id)
+        if record is None:
+            raise KeyError(delivery_id)
+        return record
+
+    def mark_dispatched(self, delivery_id: str) -> DeliveryRecord:
+        return self._transition(delivery_id, "dispatched")
+
+    def mark_confirmed(self, delivery_id: str, receipt: Optional[dict[str, Any]] = None) -> DeliveryRecord:
+        return self._transition(delivery_id, "confirmed", receipt=receipt)
+
+    def mark_unknown(self, delivery_id: str, error: Optional[str] = None) -> DeliveryRecord:
+        return self._transition(delivery_id, "unknown", error=error)
+
+    def mark_failed(self, delivery_id: str, error: str) -> DeliveryRecord:
+        return self._transition(delivery_id, "failed", error=error)
+
+    def retryable(self, delivery_id: str) -> bool:
+        record = self.get(delivery_id)
+        return bool(record and record.state in {"pending", "failed"})

tests/gateway/test_delivery_outbox.py

@@ -0,0 +1,175 @@
+"""Real-SQLite coverage for the profile-local outbound delivery outbox."""
+
+import asyncio
+import hashlib
+import sqlite3
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform
+from gateway.delivery import DeliveryRouter, DeliveryTarget
+from gateway.delivery_outbox import DeliveryOutbox, UnknownDeliveryError
+from gateway.platforms.base import SendResult
+from cron.scheduler import _cron_delivery_id
+
+
+class ReceiptAdapter:
+    def __init__(self):
+        self.calls = []
+
+    async def send(self, chat_id, content, metadata=None):
+        self.calls.append((chat_id, content, metadata))
+        return SendResult(
+            success=True,
+            message_id="provider-42",
+            raw_response={"request_id": "req-7"},
+        )
+
+
+class BlockingAdapter:
+    def __init__(self):
+        self.started = asyncio.Event()
+        self.release = asyncio.Event()
+        self.calls = 0
+
+    async def send(self, chat_id, content, metadata=None):
+        self.calls += 1
+        self.started.set()
+        await self.release.wait()
+        return SendResult(success=True, message_id="late-1")
+
+
+def test_cron_fanout_delivery_id_is_stable_per_run_target():
+    job = {"id": "job-1", "last_run_at": "2026-07-09T12:00:00+00:00"}
+
+    first = _cron_delivery_id(job, "telegram", "123", "hello")
+
+    assert first == _cron_delivery_id(job, "telegram", "123", "hello")
+    assert first != _cron_delivery_id(job, "telegram", "456", "hello")
+    assert first != _cron_delivery_id(job, "telegram", "123", "changed")
+
+
+def test_outbox_is_profile_local_real_sqlite(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    outbox = DeliveryOutbox()
+    row = outbox.create(
+        delivery_id="d-1",
+        origin="cron:job-1",
+        destination="telegram:123",
+        payload="hello",
+    )
+
+    assert outbox.path == tmp_path / "gateway" / "delivery_outbox.sqlite3"
+    assert row.state == "pending"
+    assert row.attempt == 0
+    assert row.payload_hash == hashlib.sha256(b"hello").hexdigest()
+    with sqlite3.connect(outbox.path) as conn:
+        stored = conn.execute(
+            "SELECT delivery_id, origin, destination, payload_hash, state, attempt "
+            "FROM deliveries"
+        ).fetchone()
+    assert stored == ("d-1", "cron:job-1", "telegram:123", row.payload_hash, "pending", 0)
+
+
+def test_restart_reconciliation_marks_dispatched_unknown_but_keeps_pending(tmp_path):
+    path = tmp_path / "outbox.sqlite3"
+    first = DeliveryOutbox(path)
+    first.create("sent-maybe", "cron:j", "telegram:1", "payload")
+    first.mark_dispatched("sent-maybe")
+    first.create("not-started", "cron:j", "telegram:2", "payload")
+    first.close()
+
+    restarted = DeliveryOutbox(path)
+
+    assert restarted.get("sent-maybe").state == "unknown"
+    assert restarted.get("not-started").state == "pending"
+    assert restarted.retryable("sent-maybe") is False
+    assert restarted.retryable("not-started") is True
+
+
+@pytest.mark.asyncio
+async def test_delivery_chokepoint_persists_confirmation_and_provider_receipt(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    adapter = ReceiptAdapter()
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
+    target = DeliveryTarget.parse("telegram:123")
+
+    result = await router._deliver_to_platform(
+        target,
+        "hello",
+        metadata={"delivery_id": "cron-run-1", "delivery_origin": "cron:job-1"},
+    )
+
+    row = router.outbox.get("cron-run-1")
+    assert result.message_id == "provider-42"
+    assert row.state == "confirmed"
+    assert row.attempt == 1
+    assert row.provider_receipt == {
+        "message_id": "provider-42",
+        "raw_response": {"request_id": "req-7"},
+    }
+    assert adapter.calls[0][2]["delivery_id"] == "cron-run-1"
+
+
+@pytest.mark.asyncio
+async def test_confirmed_delivery_id_is_not_sent_twice_after_restart(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    target = DeliveryTarget.parse("telegram:123")
+    first_adapter = ReceiptAdapter()
+    first = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: first_adapter})
+    metadata = {"delivery_id": "stable-id", "delivery_origin": "cron:job-1"}
+    await first._deliver_to_platform(target, "hello", metadata=metadata)
+    first.outbox.close()
+
+    restarted_adapter = ReceiptAdapter()
+    restarted = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: restarted_adapter})
+    result = await restarted._deliver_to_platform(target, "hello", metadata=metadata)
+
+    assert result["success"] is True
+    assert result["deduplicated"] is True
+    assert result["provider_receipt"]["message_id"] == "provider-42"
+    assert restarted_adapter.calls == []
+
+
+@pytest.mark.asyncio
+async def test_cancelled_in_flight_send_becomes_unknown_and_is_not_blindly_retried(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    adapter = BlockingAdapter()
+    target = DeliveryTarget.parse("telegram:123")
+    metadata = {"delivery_id": "maybe-sent", "delivery_origin": "cron:job-1"}
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
+
+    task = asyncio.create_task(router._deliver_to_platform(target, "hello", metadata=metadata))
+    await adapter.started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    assert router.outbox.get("maybe-sent").state == "unknown"
+
+    restarted = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
+    with pytest.raises(UnknownDeliveryError, match="not retried automatically"):
+        await restarted._deliver_to_platform(target, "hello", metadata=metadata)
+    assert adapter.calls == 1
+
+
+def test_failed_delivery_can_be_retried_with_incremented_attempt(tmp_path):
+    outbox = DeliveryOutbox(tmp_path / "outbox.sqlite3")
+    outbox.create("d", "cron:j", "telegram:1", "payload")
+    outbox.mark_dispatched("d")
+    outbox.mark_failed("d", "network refused")
+
+    assert outbox.retryable("d") is True
+    outbox.mark_dispatched("d")
+    assert outbox.get("d").attempt == 2
+    assert outbox.get("d").state == "dispatched"
+
+
+def test_delivery_id_rejects_changed_payload_or_destination(tmp_path):
+    outbox = DeliveryOutbox(tmp_path / "outbox.sqlite3")
+    outbox.create("same", "cron:j", "telegram:1", "one")
+
+    with pytest.raises(ValueError, match="different delivery"):
+        outbox.create("same", "cron:j", "telegram:1", "two")
+    with pytest.raises(ValueError, match="different delivery"):
+        outbox.create("same", "cron:j", "telegram:2", "one")