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

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.

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

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.

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: Crash recovery + replay loop on startup

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

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
)

Call journal.replay() in your bot's startup hook

Always 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

​Default behavior (no config needed)

​Opt out or override path

​Quick Start (advanced: manual setup)

​How It Works

​When to use which option

​Configuration Options

​Per-platform examples

​Common Patterns

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

​Pattern 2: Crash recovery + replay loop on startup

​Pattern 3: Combining with InboundDLQ for full durability stack

​Best Practices

​Related

Delivery Config

Inbound DLQ

Durable Outbound Delivery

Bot Routing

Default behavior (no config needed)

Opt out or override path

Quick Start (advanced: manual setup)

How It Works

When to use which option

Configuration Options

Per-platform examples

Common Patterns

Pattern 1: Dedup-only (webhook redelivery protection)

Pattern 2: Crash recovery + replay loop on startup

Pattern 3: Combining with InboundDLQ for full durability stack

Best Practices

Related