Dead-Target Registry

Self-healing registry that short-circuits permanently-unreachable channels and re-probes them automatically

from praisonaiagents import Agent

agent = Agent(name="gateway-agent", instructions="Route outbound bot messages safely.")
agent.start("Send the daily digest to the Telegram group.")

The dead-target registry stops your gateway from wasting requests on channels where the bot was kicked, the group was deleted, or a 403/404 is permanent — and automatically self-heals when a target recovers.

The user triggers an outbound send; the registry skips dead targets and re-probes them when recovery is due.

Quick Start

1. Enable the Registry

   Attach a registry to suppress known-dead targets:

   from praisonaiagents import Agent
   from praisonai.bots import TelegramBot
   from praisonai.bots import DeadTargetRegistry, DeliveryRouter
   
   agent = Agent(
       name="Broadcast Agent",
       instructions="Deliver scheduled updates to subscribers.",
   )
   
   registry = DeadTargetRegistry()
   
   bot = TelegramBot(
       token="YOUR_TOKEN",
       agent=agent,
       dead_target_registry=registry,
   )
   await bot.start()

2. Persist to a Custom Path

   By default the registry persists to ~/.praisonai/state/dead_targets.json. Override it:

   from pathlib import Path
   from praisonai.bots import DeadTargetRegistry
   
   registry = DeadTargetRegistry(
       persist_path=Path("/var/lib/praisonai/dead_targets.json"),
       ttl_seconds=7 * 86400,      # forget after 7 days
       max_size=5000,               # cap at 5 000 dead entries
       reprobe_seconds=1800,        # re-probe every 30 minutes
   )

3. Share Across Multiple Workers

   For a horizontally-scaled gateway, pass a shared DeliveryControlStore so every worker sees the same dead-target set. Marking a channel dead in worker A immediately suppresses sends from worker B; a successful send from any worker self-heals across all of them.

   from pathlib import Path
   from praisonai.bots import DeadTargetRegistry, DeliveryControlStore
   
   store = DeliveryControlStore(Path("~/.praisonai/state/delivery_control.sqlite"))
   
   registry = DeadTargetRegistry(
       store=store,
       ttl_seconds=7 * 86400,
       max_size=5000,
       reprobe_seconds=1800,
   )

4. Inspect and Clear Manually

   from praisonai.bots import DeadTargetRegistry
   
   registry = DeadTargetRegistry()
   
   dead = registry.list_dead()
   for target in dead:
       print(f"{target.platform}:{target.channel_id} — {target.reason}")
   
   registry.clear("telegram", "-1001234567890")
   print(f"Registry size: {registry.size()}")

How It Works

Self-heal timeline:

Configuration Options

DeadTargetRegistry constructor

from praisonai.bots import DeadTargetRegistry

registry = DeadTargetRegistry(
    persist_path="~/.praisonai/state/dead_targets.json",
    max_size=10_000,
    ttl_seconds=2_592_000,
    reprobe_seconds=3600,
)

Public methods

DeadTarget is a frozen dataclass: platform: str, channel_id: str, reason: str, ts: float (Unix epoch).

What Counts as Permanent

is_permanent_target_failure(err, platform) classifies an exception before DeliveryRouter calls mark_dead.

Permanent — target marked dead

Not permanent — stays on the retry path

Adapter Return Conventions

A channel adapter’s send_message() can signal delivery outcome three ways:

Common Patterns

Long-running broadcast gateway

A gateway pushing periodic updates to many channels suppresses dead ones so they don’t burn rate-limit budget on every cycle.

import asyncio
from praisonai.bots import BotOS, DeliveryRouter, DeadTargetRegistry

registry = DeadTargetRegistry()
botos    = BotOS(...)
router   = DeliveryRouter(botos, dead_targets=registry)

async def broadcast(channels, message):
    for platform, channel_id in channels:
        await router.deliver(f"{platform}:{channel_id}", message)

Dead channels are skipped silently; recovered channels re-enter service within one reprobe cycle (~1 h by default).

Operator dashboard via list_dead()

Expose suppressed channels in a health or admin endpoint so operators can see what has been suppressed and why.

from praisonai.bots import DeadTargetRegistry

registry = DeadTargetRegistry()

@app.get("/admin/dead-targets")
async def dead_targets():
    return [
        {"platform": t.platform, "channel_id": t.channel_id,
         "reason": t.reason, "since": t.ts}
        for t in registry.list_dead()
    ]

Custom suppression for one chat

Un-suppress a channel manually after confirming the bot has been re-added.

registry.clear("telegram", "-1001234567890")

Best Practices

registry = DeadTargetRegistry(
    persist_path="/data/praisonai/state/dead_targets.json"
)

registry = DeadTargetRegistry(reprobe_seconds=300)

print(f"Suppressed channels: {registry.size()}")
for t in registry.list_dead():
    print(f"  {t.platform}:{t.channel_id} — {t.reason}")

Related