Inbound Journal

Deduplicate webhook redeliveries and recover in-flight messages after a crash

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.

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.

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:

1. Create journal that survives restarts

   from praisonaiagents import Agent
   from praisonai.bots import InboundJournal, BotSessionManager
   
   journal = InboundJournal(path="~/.praisonai/state/telegram/ingress.sqlite")

2. Enable on your bot

   session = BotSessionManager(platform="telegram", ingress_journal=journal)
   
   agent = Agent(name="Support Bot", instructions="Help users with their questions")

How It Works

Defer completion until reply is delivered (opt-in)

By default chat() settles the inbound entry the instant the agent finishes — before your adapter sends the reply. A crash in that window loses the user’s answer and marks the inbound complete, so replay() never re-drives it (Issue #5152).

Set defer_journal_completion=True to keep the entry pending until your adapter confirms the reply was durably delivered.

Enable it on the manager:

session = BotSessionManager(
    platform="telegram",
    ingress_journal=journal,
    defer_journal_completion=True,
)

Then your adapter owns settlement: after session.chat() returns, capture the deferred key from session._last_journal_key, send the reply, and only on durable delivery call session.complete_last_journal_entry(journal_key). If the process dies between chat() returning and delivery, the entry stays pending, so replay() on the next boot (or platform redelivery) resurfaces it.

When to opt in

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)

When to use which option

Configuration Options

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)

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

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

Pattern 4: Defer inbound completion until reply is delivered (opt-in, safest)

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,
    defer_journal_completion=True,  # opt in
)

agent = Agent(name="Support Bot", instructions="Help users with their questions")

# In your adapter's webhook handler:
reply = 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",
)

# Capture the deferred key BEFORE sending (single-turn slot).
journal_key = session._last_journal_key

# Deliver the reply durably (retry, DLQ-park, etc.).
await telegram.send_message(chat_id=..., text=reply)

# Settle the inbound ONLY after delivery succeeded.
session.complete_last_journal_entry(journal_key)

Best Practices

session = BotSessionManager(
    platform="slack",
    ingress_journal=journal,
    defer_journal_completion=True,
)

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

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

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

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

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

# 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