Durable Outbound Delivery

Persist outbound bot messages with retry, idempotency, and crash-safe drain on restart

from praisonaiagents import Agent

agent = Agent(name="bot-agent", instructions="Reply on messaging channels.")
agent.start("Send the user a confirmation once the job completes.")

Durable Delivery persists every outbound bot message to a SQLite outbox so retries, restarts, and platform outages never drop a reply. For permanently dead targets — bot kicked, chat deleted — see Dead Target Registry, which suppresses doomed sends instead of queuing them.

The user expects a bot reply; durable delivery queues outbound messages, retries transient failures, and drains after restarts.

Quick Start

1. Easiest path — DurableAdapterMixin

   Add three lines to any existing adapter and every send_durable() call is crash-safe:

   import os
   from praisonaiagents import Agent
   from praisonai.bots import TelegramBot
   
   # TelegramBot already supports durable delivery via DurableAdapterMixin
   agent = Agent(name="assistant", instructions="Help users")
   bot = TelegramBot(
       token=os.getenv("TELEGRAM_BOT_TOKEN"),
       agent=agent,
   )
   
   # Replay anything queued before the last crash.
   # Also fires automatically on channel recovery — no need to wire it yourself.
   # Manual call is only needed for adapters not driven by ChannelSupervisor.
   await bot.drain_outbox()
   
   # Send with durability
   await bot.send_durable(
       channel_id="12345",
       content="Hello! I'm crash-safe.",
       idempotency_key="welcome-msg-12345",
   )

2. Configure manually with DurableDelivery

   For full control, wire up OutboundQueue and DurableDelivery directly:

   import os
   from praisonaiagents import Agent
   from praisonai.bots import OutboundQueue, DurableDelivery, TelegramBot
   
   outbox = OutboundQueue(path="~/.praisonai/state/outbox.sqlite")
   adapter = TelegramBot(
       token=os.getenv("TELEGRAM_BOT_TOKEN"),
       agent=Agent(name="assistant", instructions="Help users"),
   )
   delivery = DurableDelivery(
       outbox, adapter, platform="telegram",
       # mark_recovered=True,  # opt-in: prefix crash-recovered re-sends with a "possible duplicate" marker
   )
   
   # On startup: replay anything queued before the last crash
   succeeded, failed = await delivery.drain_pending()
   
   # Send with durability — idempotent if you reuse the key
   success = await delivery.send(
       channel_id="12345",
       content="Hello, world!",
       idempotency_key="msg-123",
   )

How It Works

Each message moves through statuses: pending → sending → sent (or failed / permanent_failure).

State Machine

The outbox tracks six statuses: pending, sending, recovered, sent, failed, and permanent_failure.

On restart, stale sending entries transition to recovered instead of pending. This preserves the information that the send was in-flight when the crash occurred.

status = 'sending'  (crash)  ──►  status = 'recovered'  (on restart)

recovered entries are retryable like pending, but when a reconciler is supplied to drain(), they are offered to it first — allowing adapters that can check the platform to confirm delivery and avoid re-sending an already-delivered message (effectively-once). Without a reconciler, recovered entries are re-sent as normal (at-least-once, unchanged behaviour).

Age-gated dead-lettering (attempts + wall-clock)

permanent_failure is now a two-condition decision, so a brief channel outage no longer silently drops deliverable messages.

An entry moves to permanent_failure when either:

• The error class is known-permanent ("credential" or "permanent_target") — short-circuits immediately regardless of age.
• Both attempts >= max_attempts and the entry has been in the queue for >= dead_letter_min_age (default 6h).

Otherwise the entry stays retryable and the next drain() reschedules it under the normal capped backoff.

The default 6-hour floor is far longer than any realistic channel incident and well under the 7-day retention TTL, so most operators need no config change — transient outages simply keep retrying until the channel recovers.

Restoring legacy behaviour — pass dead_letter_min_age=0 if you rely on attempt-count-only quarantine (e.g. poison-message unit tests):

from praisonai.bots import OutboundQueue

# Attempt-only dead-lettering (pre-#3521 behaviour)
outbox = OutboundQueue(
    path="~/.praisonai/state/outbox.sqlite",
    max_attempts=5,
    dead_letter_min_age=0,
)

Custom policy — inject any DeadLetterPolicyProtocol implementation for one-off decisions:

from praisonaiagents.gateway import AttemptAndAgeDeadLetterPolicy
from praisonai.bots import OutboundQueue

outbox = OutboundQueue(
    path="~/.praisonai/state/outbox.sqlite",
    dead_letter_policy=AttemptAndAgeDeadLetterPolicy(
        max_attempts=10,
        min_age_seconds=24 * 3600,  # allow a full day of retries
    ),
)

Introduced in PraisonAI PR #3521.

Effectively-Once Delivery

Adapters that can confirm whether a prior send actually landed can upgrade durable delivery from at-least-once to effectively-once.

1. Declare the capability

   Set reconciles_unknown_send=True on PlatformCapabilities:

   from praisonaiagents.bots import PlatformCapabilities
   
   class MyAdapter(DurableAdapterMixin):
       platform_capabilities = PlatformCapabilities(
           reconciles_unknown_send=True,
       )

2. Implement was_delivered

   Add an async method that checks the platform for a prior send:

   async def was_delivered(self, idempotency_key: str) -> bool:
       status = await my_provider.get_message_status(idempotency_key)
       return status == "sent"

3. Call drain_pending() at startup

   DurableDelivery auto-wires the reconciler from the adapter’s capability — no extra configuration required:

   from praisonaiagents.bots import PlatformCapabilities
   from praisonai.bots._adapter_base import DurableAdapterMixin
   
   
   class AcmeAdapter(DurableAdapterMixin):
       platform_capabilities = PlatformCapabilities(
           reconciles_unknown_send=True,
       )
   
       async def was_delivered(self, idempotency_key: str) -> bool:
           status = await acme_client.messages.lookup(idempotency_key)
           return status.delivered
   
       async def send_message(self, channel_id: str, content: str) -> None:
           await acme_client.messages.send(channel_id, content)
   
   
   # DurableDelivery picks up the reconciler automatically.
   succeeded, failed = await adapter.drain_outbox()

Choosing the Right Primitive

Configuration Options

OutboundQueue

SQLite-backed outbox. All parameters after path are keyword-only.

OutboundQueue.drain() signature

async def drain(
    self,
    sender: Callable[[str, Dict[str, Any]], Awaitable[bool]],
    limit: Optional[int] = None,
    *,
    reconciler: Optional[Callable[[OutboundEntry], Awaitable[bool]]] = None,
) -> Tuple[int, int]:

from praisonai.bots import OutboundQueue

outbox = OutboundQueue(
    path="~/.praisonai/state/outbox.sqlite",
    max_size=10_000,
    ttl_seconds=3 * 86400,  # 3 days
    max_attempts=3,
)

OutboundQueue.enqueue() signature

enqueue() accepts an optional keyword-only lane_key for per-conversation grouping. It defaults to target, so messages to the same chat share a lane under ordering="strict".

async def enqueue(
    self,
    idempotency_key: str,
    target: str,
    payload: Dict[str, Any],
    metadata: Optional[Dict[str, Any]] = None,
    *,
    lane_key: Optional[str] = None,   # defaults to target
) -> str:
    ...

OutboundQueue.status_for() signature

status_for() returns the current status of the entry for an idempotency key.

def status_for(self, idempotency_key: str) -> Optional[str]:
    ...

Returns — the current entry status ("pending", "sending", "recovered", "sent", "failed", or "permanent_failure"), or None if no entry exists for that key.

Use it to tell an already-delivered duplicate (status "sent") apart from a genuine delivery failure after a drain — so a deduplicated re-fire is reported as a suppressed success rather than a spurious failure.

from praisonai.bots import OutboundQueue

outbox = OutboundQueue(path="~/.praisonai/state/gateway_outbox.sqlite")
await outbox.enqueue(
    idempotency_key="daily-brief-2026-07-20",
    target="telegram:-100123",
    payload={"text": "Morning brief ready."},
)
await outbox.drain(sender)

if outbox.status_for("daily-brief-2026-07-20") == "sent":
    print("Delivered (or suppressed as a duplicate of a prior send).")

DurableDelivery

Wraps an OutboundQueue and an adapter to provide a simple .send() / .drain_pending() API.

DurableAdapterMixin.setup_durable_delivery()

Call once in your adapter’s __init__ to wire up the outbox.

deliver_with_retry()

Bounded retry without persistence — on a recoverable failure, the delay is the server-mandated wait (server_retry_after(err)) when present, otherwise compute_backoff(policy, attempt).

from praisonai.bots._delivery import deliver_with_retry
from praisonai.bots._rate_limit import RateLimiter
from praisonai.bots._resilience import BackoffPolicy

limiter = RateLimiter.for_platform("telegram")
ok, err = await deliver_with_retry(
    send_fn, channel_id="telegram-chat-12345",
    rate_limiter=limiter,
    policy=BackoffPolicy(initial_ms=2000, max_ms=30000, max_attempts=5),
    platform="telegram",
)

server_retry_after()

Extracts a server-mandated wait (seconds) from an error or response. Used internally by deliver_with_retry, ConnectionMonitor.record_error, and OutboundQueue.drain to honour explicit throttle signals over the policy backoff.

from praisonai.bots._resilience import server_retry_after

# Telegram RetryAfter
wait = server_retry_after(err)          # -> 30.0

# HTTP Retry-After header (seconds or HTTP-date)
wait = server_retry_after(http_err)     # -> 5.0 or seconds-until-date

# No hint present
wait = server_retry_after(generic_err)  # -> None

Returns None when no hint is present — callers fall back to the policy backoff.

deliver_chunked()

Splits a long message at paragraph boundaries and sends each chunk separately. Returns the number of chunks sent.

from praisonai.bots._chunk import chunk_message

# chunk_message is the underlying splitter
chunks = chunk_message(long_text, max_length=4096, preserve_fences=True)
for chunk in chunks:
    await adapter.send_message(channel_id, chunk)

BackoffPolicy

Controls retry timing for both deliver_with_retry and OutboundQueue.drain.

Idempotency & When the Outbox Drains

Idempotency Keys

Every message has an idempotency_key — a UUID generated automatically if you omit it. Reusing the same key for the same logical message prevents double-sends across retries.

# Webhook-triggered reply: derive key from inbound message ID
inbound_id = webhook_payload["message_id"]
await delivery.send(
    channel_id=chat_id,
    content=reply_text,
    idempotency_key=f"reply-{inbound_id}",
)

If the webhook is redelivered and send() is called again with the same key, the outbox skips the enqueue (SQLite UNIQUE constraint) and marks the existing row sent.

When the Outbox Drains

Held replies go out on three triggers, so operators can reason about when a queued message is actually re-attempted:

1. Adapter startup — replays anything queued before the last crash (drain_pending() / drain_outbox()).
2. Channel recovery (new — Issue #4043) — when ChannelSupervisor observes a channel come back after a recoverable outage (network blip, transient API error, stale socket), it fires drain_outbox() in the background so held replies go out promptly. Applications do not have to call anything.
3. Lazy on next inbound turn — the next chat() turn also opportunistically drains.

Call drain_pending() once at adapter startup to replay anything that was queued before the last crash:

# On adapter startup
succeeded, failed = await delivery.drain_pending()
# Log output: "Drained outbox: 3 sent, 0 failed"

# Or with the mixin:
succeeded, failed = await adapter.drain_outbox()

The drain replays oldest messages first and skips messages that have exceeded max_attempts. The retry gate is max(compute_backoff(policy, attempts + 1), server_retry_after(stored_err)) — the platform’s mandated wait survives across process restarts.

Recovery-triggered re-drain (Issue #4043)

Why did my held reply just go out on its own? When a channel drops on a transient outage, the outbox deliberately holds deliverable replies instead of dropping them. Before, those replies sat undelivered until the next inbound chat() turn or a process restart. Now ChannelSupervisor re-drains that platform’s outbox the moment the channel reconnects — see Channel supervision → Recovery-triggered outbox re-drain.

Log lines from the merged code:

# success (only when something moved):
Channel '<name>' recovered: re-drained outbox (<S> sent, <F> failed)

# failure (swallowed — retried on the next drain):
Channel '<name>' outbox re-drain on recovery failed (will retry on next drain): <exc>

Introduced in PraisonAI PR #4046 (fixes #4043).

Best Practices

delivery = DurableDelivery(outbox, adapter, platform="telegram")

# ✅ Good: stable key tied to the inbound event
await delivery.send(
    channel_id=chat_id,
    content=reply,
    idempotency_key=f"reply-{inbound_message_id}",
)

# ❌ Bad: new UUID every call — no deduplication across retries
import uuid
await delivery.send(channel_id=chat_id, content=reply, idempotency_key=str(uuid.uuid4()))

# ✅ Good: persistent path
outbox = OutboundQueue(path="~/.praisonai/state/outbox.sqlite")

# ❌ Bad: lost on restart or container rebuild
outbox = OutboundQueue(path="/tmp/outbox.sqlite")

# ✅ Good: single drain at startup
async def on_start():
    succeeded, failed = await delivery.drain_pending()

# ❌ Bad: two drainers in separate tasks
asyncio.create_task(delivery.drain_pending())
asyncio.create_task(delivery.drain_pending())

Proactive Path

Direct BotOS.deliver(...) shares the reply-path rate limiter but is fire-and-forget. When the gateway itself performs the scheduled/proactive send, it runs through a durable outbox instead.

Per-Adapter Delivery Guarantee

Only adapters that declare reconciles_unknown_send=True and implement was_delivered reach effectively-once; the rest stay at-least-once by design.

² AgentMailBot.send_message(..., idempotency_key=…) forwards the outbox key as an AgentMail Idempotency-Key header (1–256 chars, 24h TTL). The outbox key survives crashes, so a re-send of the same outbox entry is deduplicated at the AgentMail provider — even without a was_delivered reconciler. See Duplicate protection on retries.

Slack is the first built-in adapter to implement the reconciliation seam end-to-end:

• SlackBot declares reconciles_unknown_send=True on its platform_capabilities.
• send_message(target, text, ..., idempotency_key=None) stamps the key into the Slack message metadata with event_type="praisonai_outbound".
• was_delivered(target, idempotency_key) reads back recent history (conversations.history, or conversations.replies for a threaded send) and matches the key in metadata.event_payload. It returns False on any lookup failure, falling back to at-least-once.

Before: sending → (restart) → recovered → re-dispatched blindly → duplicate
After:  sending → (restart) → recovered → reconciler confirms → marked sent (no duplicate)

Related