> ## Documentation Index > Fetch the complete documentation index at: https://praison.ai/docs/llms.txt > Use this file to discover all available pages before exploring further. # Inbound Journal > Deduplicate webhook redeliveries and recover in-flight messages after a crash ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent agent = Agent(name="durable-bot", instructions="Process webhook messages exactly once.") agent.start("Handle webhook delivery with deduplication and crash recovery.") ``` **Different scope from the [Run-State Journal](/docs/features/run-state-journal).** The **inbound journal** deduplicates and recovers *webhook messages*. The **run-state journal** persists the *execution cursor* of a single run. A webhook may spawn a run; the two journals track the two halves of the pipeline. Inbound Journal tracks messages before agents process them, providing deduplication of webhook redeliveries and crash-safe replay of in-flight messages. The user sends a webhook message; the journal deduplicates redeliveries and tracks in-flight work. **On by default for gateway/bot runs** — Every bot started via `praisonai bot start`, `onboard`, or `bot.yaml` gets the journal automatically. No code needed. The journal lives at `~/.praisonai/state//ingress.sqlite`. Set `delivery.durable: false` to opt out. ## Automatic crash-replay Every gateway-managed channel calls `replay()` for you at start-up and hot-reload — no startup hook needed. Every channel started via `praisonai bot start`, `onboard`, `bot.yaml`, or on hot-reload of a single channel automatically calls `InboundJournal.replay()` for its journal. What `replay()` does: * Resets stale `claimed` rows back to `pending` so they can be reprocessed via redelivery or session-resume (`_resume_interrupted_turns()`, [PraisonAI #3379](https://github.com/MervinPraison/PraisonAI/pull/3379)). * Quarantines genuine poison entries to the inbound DLQ under the shared age-gated policy (see [Inbound DLQ](/docs/features/inbound-dlq)). * Does **not** synthesise per-adapter native message objects to self-dispatch — that duplication is deliberately left to the session-resume path. Best-effort: any failure is logged and skipped, so channel start never aborts. Reset counts are logged at `INFO` so operators can see the recovery happened. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph LR subgraph "Inbound Journal" Webhook[📨 Webhook] --> Receive[📒 receive] Receive -->|new| Claim[🔒 claim] Receive -->|duplicate| Skip[⏭️ skip] Claim --> Agent[🤖 agent.chat] Agent --> Complete[✅ complete] end classDef input fill:#8B0000,stroke:#7C90A0,color:#fff classDef process fill:#189AB4,stroke:#7C90A0,color:#fff classDef success fill:#10B981,stroke:#7C90A0,color:#fff classDef skip fill:#F59E0B,stroke:#7C90A0,color:#fff class Webhook input class Receive,Claim process class Agent,Complete success class Skip skip ``` The outbound reply path is now symmetric — see [Outbound Resilience → Automatic crash-recovery on start](/docs/features/outbound-resilience#automatic-crash-recovery-on-start). ## Default behavior (no config needed) Every bot started through `build_session_manager` (all shipped adapters) automatically gets an Inbound Journal at: ``` ~/.praisonai/state//ingress.sqlite ``` For example, a Telegram bot journals messages at `~/.praisonai/state/telegram/ingress.sqlite`. A Discord bot uses `~/.praisonai/state/discord/ingress.sqlite`. Each platform is fully isolated — no cross-platform dedup collisions. Set `PRAISONAI_HOME` to override the base directory: ``` $PRAISONAI_HOME/state//ingress.sqlite ``` ## Opt out or override path Add a `delivery:` block to any channel in your `bot.yaml` or `gateway.yaml`: ```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} channels: telegram: token: "${TELEGRAM_BOT_TOKEN}" delivery: durable: false # opt out of durable inbound delivery for this channel ``` Override the store directory: ```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} channels: telegram: token: "${TELEGRAM_BOT_TOKEN}" delivery: store: /var/lib/praisonai/telegram-state # → ingress.sqlite lives in /var/lib/praisonai/telegram-state/ ``` ## Quick Start (advanced: manual setup) Most users get the journal automatically. For custom setups outside `build_session_manager`: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonai.bots import InboundJournal, BotSessionManager journal = InboundJournal(path="~/.praisonai/state/telegram/ingress.sqlite") ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} session = BotSessionManager(platform="telegram", ingress_journal=journal) agent = Agent(name="Support Bot", instructions="Help users with their questions") ``` You don't need to call `complete()` yourself when using `BotSessionManager.chat()` — successful chats mark the journal entry as completed automatically. You only need to call `complete()` manually if you're driving `InboundJournal` directly without `BotSessionManager`. *** ## How It Works Before [PR #1980](https://github.com/MervinPraison/PraisonAI/pull/1980), the default `chat()` path claimed entries but never called `complete()`. On startup, `journal.replay()` could re-issue messages the user had already received answers to. Upgrade to a release after 2026-06-19 — replay then only covers genuinely incomplete entries. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant Platform as Telegram/Discord/Slack participant Adapter as Bot Adapter participant Journal as InboundJournal participant Agent Platform->>Adapter: webhook(message_id=42) Adapter->>Journal: receive(platform, account, chat_id, 42) alt First delivery Journal-->>Adapter: key Adapter->>Journal: claim(key) Adapter->>Agent: chat(prompt) Agent-->>Adapter: response Adapter->>Journal: complete(key) Adapter->>Platform: send reply else Redelivery — prior attempt still pending or stale-claimed Journal-->>Adapter: same key (retry) Adapter->>Agent: chat(prompt) Agent-->>Adapter: response Adapter->>Journal: complete(key) Adapter->>Platform: send reply else Redelivery — already completed or actively claimed Journal-->>Adapter: None Adapter-->>Platform: 200 OK (silent skip) end ``` | Stage | Purpose | What happens | | ------------ | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | | **receive** | Deduplication & retry | Returns a key for new messages, retries pending or stale-claimed duplicates, returns `None` only for already-completed or actively-claimed entries | | **claim** | Crash protection | Marks the entry as being processed (auto-released on stale timeout) | | **complete** | Cleanup | Marks successful processing — `BotSessionManager.chat()` calls this for you automatically | *** ## Automatic crash recovery under the gateway The gateway calls `InboundJournal.replay()` for you — at channel start-up and on every hot-reload — so a gateway killed mid-turn recovers stranded messages without you invoking `replay()` yourself. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant Op as Operator participant Gateway participant Journal as InboundJournal participant Session as _resume_interrupted_turns Op->>Gateway: bot start / hot-reload Gateway->>Journal: replay() Journal-->>Gateway: N stale claims reset to pending Gateway->>Session: reprocess exactly-once Session-->>Op: log line: reset N stranded messages ``` `replay()` resets stale `claimed` rows back to `pending` so a redelivery — or the gateway's session-level resume ([Issue #3379](https://github.com/MervinPraison/PraisonAI/pull/3379)) — can reprocess them exactly once, and quarantines genuine poison entries to the inbound DLQ. Reset counts are logged at INFO so operators can grep for recovery: ``` Inbound crash-recovery: reset N stranded journaled message(s) to pending on start (unblocked for redelivery/resume) ``` How in-flight turns resume after a gateway restart Session-level exactly-once resume path (#3379) *** ## When to use which option ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph TB Start{Need durability?} -->|Webhook redeliveries| Journal[InboundJournal] Start -->|LLM might fail| DLQ[InboundDLQ] Start -->|Both| Both[Use both — journal first, DLQ on exception] classDef question fill:#6366F1,stroke:#7C90A0,color:#fff classDef option fill:#189AB4,stroke:#7C90A0,color:#fff class Start question class Journal,DLQ,Both option ``` | Feature | **InboundJournal** | **InboundDLQ** | | ------------------ | ------------------------------------ | --------------------------- | | **When triggered** | Every inbound message | Only on agent failure | | **Solves** | Webhook redeliveries, crash recovery | Failed LLM calls | | **Performance** | Fast dedup check | No overhead until failure | | **Use together** | ✅ Journal → agent → DLQ on exception | ✅ Complete durability stack | *** ## Configuration Options | Option | Type | Default | Description | | --------------------- | ---------------------------------- | --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- | | `path` | `str \| Path` | **required** | SQLite file path. Parent dirs auto-created. `~` is expanded. | | `max_size` | `int` | `50_000` | Max entries kept. Excess evicted: completed first, then oldest pending. | | `ttl_seconds` | `int` | `2_592_000` (30 days) | Completed entries older than this are evicted. | | `claim_timeout` | `int` | `300` (5 min) | Claimed entries older than this are considered stale and replayed. | | `dead_letter_min_age` | `float` (seconds) | `21600` (6h) | Minimum wall-clock age since first receipt before an attempt-exhausted transient failure is quarantined. `0` restores pre-#3521 attempt-only behaviour. | | `dead_letter_policy` | `DeadLetterPolicyProtocol \| None` | `None` | Optional custom `praisonaiagents.gateway` policy. When supplied, overrides `max_attempts` + `dead_letter_min_age`. | ### Age-gated quarantine `replay()` and the redelivery path no longer quarantine on attempts alone. An exhausted entry is only moved to the DLQ once it is **both** attempt-exhausted **and** at least `dead_letter_min_age` (default 6h) old — unless the failure is a known-permanent error class, which short-circuits immediately. A routine LLM outage that burns the claim budget in seconds keeps being retried instead of quarantining every in-flight message. See [Durable Outbound Delivery → Age-gated dead-lettering](/docs/docs/features/durable-delivery#age-gated-dead-lettering-attempts--wall-clock) for the shared policy and decision diagram. Introduced in [PraisonAI PR #3521](https://github.com/MervinPraison/PraisonAI/pull/3521). *** ## Per-platform examples ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonai.bots import InboundJournal, BotSessionManager journal = InboundJournal(path="~/.praisonai/state/telegram/ingress.sqlite") session = BotSessionManager(platform="telegram", ingress_journal=journal) agent = Agent( name="Telegram Bot", instructions="Respond helpfully to Telegram users" ) # In your webhook handler: response = await session.chat( agent=agent, user_id=update.effective_user.id, prompt=update.message.text, message_id=str(update.message.message_id), account="my_telegram_bot" ) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonai.bots import InboundJournal, BotSessionManager journal = InboundJournal(path="~/.praisonai/state/discord/ingress.sqlite") session = BotSessionManager(platform="discord", ingress_journal=journal) agent = Agent( name="Discord Bot", instructions="Help Discord server members" ) # In your message handler: response = await session.chat( agent=agent, user_id=str(message.author.id), prompt=message.content, message_id=str(message.id), account="my_discord_bot" ) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonai.bots import InboundJournal, BotSessionManager journal = InboundJournal(path="~/.praisonai/state/slack/ingress.sqlite") session = BotSessionManager(platform="slack", ingress_journal=journal) agent = Agent( name="Slack Bot", instructions="Assist Slack workspace users" ) # In your event handler: response = await session.chat( agent=agent, user_id=event["user"], prompt=event["text"], message_id=event["ts"], account="my_slack_bot" ) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonai.bots import InboundJournal, BotSessionManager journal = InboundJournal(path="~/.praisonai/state/whatsapp/ingress.sqlite") session = BotSessionManager(platform="whatsapp", ingress_journal=journal) agent = Agent( name="WhatsApp Bot", instructions="Help WhatsApp users with their questions" ) # In your webhook handler: response = await session.chat( agent=agent, user_id=message["from"], prompt=message["text"]["body"], message_id=message["id"], account="my_whatsapp_bot" ) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonai.bots import InboundJournal, BotSessionManager journal = InboundJournal(path="~/.praisonai/state/email/ingress.sqlite") session = BotSessionManager(platform="email", ingress_journal=journal) agent = Agent( name="Email Assistant", instructions="Respond to email inquiries professionally" ) # In your email processor: response = await session.chat( agent=agent, user_id=email["from"], prompt=email["body"], message_id=email["message_id"], account="support@company.com" ) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonai.bots import InboundJournal, BotSessionManager journal = InboundJournal(path="~/.praisonai/state/agentmail/ingress.sqlite") session = BotSessionManager(platform="agentmail", ingress_journal=journal) agent = Agent( name="AgentMail Bot", instructions="Handle AgentMail messages efficiently" ) # In your AgentMail handler: response = await session.chat( agent=agent, user_id=message.sender, prompt=message.content, message_id=message.id, account="my_agentmail_account" ) ``` *** ## Common Patterns ### Pattern 1: Dedup-only (webhook redelivery protection) ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # Minimal setup for webhook deduplication journal = InboundJournal( path="~/.praisonai/state/telegram/ingress.sqlite", claim_timeout=60 # Fast timeout for quick processing ) session = BotSessionManager(platform="telegram", ingress_journal=journal) # Behaviour on duplicate webhook deliveries: # - already completed → skipped (no agent call) # - actively being handled → skipped (another worker has the claim) # - prior attempt still pending or stale-claimed → retried automatically ``` ### Pattern 2: Manual replay (only when driving `InboundJournal` outside the gateway) Gateway-managed bots get startup and hot-reload replay for free (see [Automatic crash-replay](#automatic-crash-replay)). This pattern is only for custom setups that drive `InboundJournal` directly. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # Setup with crash recovery journal = InboundJournal( path="~/.praisonai/state/telegram/ingress.sqlite", claim_timeout=600 # 10 minutes for complex agent tasks ) # On startup, replay any stale claimed entries replayed_count = journal.replay() print(f"Replayed {replayed_count} stale entries from previous crash") session = BotSessionManager(platform="discord", ingress_journal=journal) ``` After PR #1989, redelivery of a still-pending message also retries automatically — you no longer need to wait for a restart plus `replay()` to recover from a mid-flight crash if the platform redelivers the same `message_id`. `replay()` is still the right call on startup to recover any orphaned claims. Under the gateway, both mechanisms now compose: `replay()` unblocks stranded claimed rows at start-up, and redelivery + session-resume then reprocess them exactly once. ### Pattern 3: Combining with InboundDLQ for full durability stack ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonai.bots import InboundJournal, InboundDLQ, BotSessionManager # Both journal and DLQ for complete durability journal = InboundJournal(path="~/.praisonai/state/slack/ingress.sqlite") dlq = InboundDLQ(path="~/.praisonai/state/slack/inbound_dlq.sqlite") session = BotSessionManager( platform="slack", ingress_journal=journal, # Handles dedup + crash recovery dlq=dlq # Handles agent failures ) # Flow: webhook → journal.receive() → agent.chat() → dlq on exception ``` *** ## Best Practices The journal's SQLite file must survive restarts for crash recovery to work. Use an absolute path or a location that persists across deployments. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # ✅ Good: persists across restarts journal = InboundJournal(path="/var/lib/myapp/ingress.sqlite") # ❌ Bad: temporary path, lost on restart journal = InboundJournal(path="/tmp/journal.sqlite") ``` Set `claim_timeout` to be longer than your p99 `agent.chat()` latency to avoid false stale entry detection. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # If your agent takes up to 30 seconds, use a longer timeout journal = InboundJournal( path="~/.praisonai/state/telegram/ingress.sqlite", claim_timeout=900 # 15 minutes safety margin ) ``` When running under the gateway (`praisonai bot start`, `onboard`, `bot.yaml`, `bot.WebSocketGateway`), `replay()` is called automatically at channel start-up and hot-reload — you do not need to call it yourself. Reset counts are logged. See [Automatic crash recovery under the gateway](#automatic-crash-recovery-under-the-gateway). When you drive `InboundJournal` yourself (bare `BotSessionManager`, custom adapter loops), replay stale entries when your bot starts to recover from crashes. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} def startup_hook(): # Replay any messages that were claimed but not completed replayed = journal.replay() logger.info(f"Startup replay: {replayed} messages recovered") # Call this before starting your bot's event loop startup_hook() ``` The `account` parameter is part of the deduplication key. Keep it consistent for each bot instance. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # ✅ Good: stable account identifier ACCOUNT = "production_telegram_bot_v1" await session.chat(..., account=ACCOUNT) # ❌ Bad: changing account breaks deduplication await session.chat(..., account=f"bot_{random.randint(1,100)}") ``` Platforms redeliver webhooks at fixed intervals (Telegram retries within seconds, Slack waits \~3s before its first retry, etc.). Set `claim_timeout` **shorter** than your platform's redelivery window so that a crashed worker's claim becomes stale before the next redelivery arrives — that's what enables automatic recovery without a restart. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # Slack retries after 3s; 30s leaves room for normal processing # while ensuring a crashed claim is stale on the next redelivery journal = InboundJournal( path="~/.praisonai/state/slack/ingress.sqlite", claim_timeout=30, ) ``` *** ## Related Full reference for the `delivery:` channel config block — defaults, opt-out, and path override Failure-side durability when agent execution fails Outbound counterpart — persist outgoing messages with retry and idempotency Multi-channel session routing for complex bot setups The delivery-side half of "no inbound message ends without a recorded outcome" Automatic crash-replay introduced in [PraisonAI PR #3622](https://github.com/MervinPraison/PraisonAI/pull/3622) (fixes [#3621](https://github.com/MervinPraison/PraisonAI/issues/3621)).