Inbound Journal - PraisonAI

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. 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/<platform>/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).
Quarantines genuine poison entries to the inbound DLQ under the shared age-gated policy (see 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.

The outbound reply path is now symmetric — see 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/<platform>/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/<platform>/ingress.sqlite

Opt out or override path

Add a delivery: block to any channel in your bot.yaml or gateway.yaml:

channels:
  telegram:
    token: "${TELEGRAM_BOT_TOKEN}"
    delivery:
      durable: false   # opt out of durable inbound delivery for this channel

Override the store directory:

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:

Create journal that survives restarts

from praisonaiagents import Agent
from praisonai.bots import InboundJournal, BotSessionManager

journal = InboundJournal(path="~/.praisonai/state/telegram/ingress.sqlite")

Enable on your bot

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

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. replay() resets stale claimed rows back to pending so a redelivery — or the gateway’s session-level resume (Issue #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)

Gateway Restart Continuation

How in-flight turns resume after a gateway restart

Gateway Session Continuity

Session-level exactly-once resume path (#3379)

When to use which 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 for the shared policy and decision diagram. Introduced in PraisonAI PR #3521.

Per-platform examples

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

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

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

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

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

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)

# 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). This pattern is only for custom setups that drive InboundJournal directly.

# 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

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

Use the same path across restarts

The journal’s SQLite file must survive restarts for crash recovery to work. Use an absolute path or a location that persists across deployments.

# ✅ Good: persists across restarts
journal = InboundJournal(path="/var/lib/myapp/ingress.sqlite")

# ❌ Bad: temporary path, lost on restart
journal = InboundJournal(path="/tmp/journal.sqlite")

Tune claim_timeout to match your agent latency

Set claim_timeout to be longer than your p99 agent.chat() latency to avoid false stale entry detection.

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

Under the gateway: replay() is automatic

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.

Driving InboundJournal directly: call replay() in your startup hook

When you drive InboundJournal yourself (bare BotSessionManager, custom adapter loops), replay stale entries when your bot starts to recover from crashes.

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

Keep account stable per bot instance

The account parameter is part of the deduplication key. Keep it consistent for each bot instance.

# ✅ 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)}")

Set claim_timeout below the platform's redelivery window

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.

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

Delivery Config

Full reference for the delivery: channel config block — defaults, opt-out, and path override

Inbound DLQ

Failure-side durability when agent execution fails

Durable Outbound Delivery

Outbound counterpart — persist outgoing messages with retry and idempotency

Bot Routing

Multi-channel session routing for complex bot setups

Visible-Outcome Guarantee

The delivery-side half of “no inbound message ends without a recorded outcome”

Automatic crash-replay introduced in PraisonAI PR #3622 (fixes #3621).

​Automatic crash-replay

​Default behavior (no config needed)

​Opt out or override path

​Quick Start (advanced: manual setup)

​How It Works

​Automatic crash recovery under the gateway

Gateway Restart Continuation

Gateway Session Continuity

​When to use which option

​Configuration Options

​Age-gated quarantine

​Per-platform examples

​Common Patterns

​Pattern 1: Dedup-only (webhook redelivery protection)

​Pattern 2: Manual replay (only when driving InboundJournal outside the gateway)

​Pattern 3: Combining with InboundDLQ for full durability stack

​Best Practices

​Related

Delivery Config

Inbound DLQ

Durable Outbound Delivery

Bot Routing

Visible-Outcome Guarantee

Automatic crash-replay

Default behavior (no config needed)

Opt out or override path

Quick Start (advanced: manual setup)

How It Works

Automatic crash recovery under the gateway

When to use which option

Configuration Options

Age-gated quarantine

Per-platform examples

Common Patterns

Pattern 1: Dedup-only (webhook redelivery protection)

Pattern 2: Manual replay (only when driving `InboundJournal` outside the gateway)

Pattern 3: Combining with InboundDLQ for full durability stack

Best Practices

Related