Unified Degraded-Capability Registry

One consolidated list of every degraded owner in the gateway — channel, provider, capability, route, or gateway itself — surfaced through health()

When any part of the gateway degrades — a channel, an LLM provider, an MCP capability, a route, or the gateway itself — it shows up once in a single, redacted degraded_owners list with an exact next action.

Before this registry the gateway recorded degradation in several disconnected places but surfaced only degraded channels through health(). Provider credential failures, unresolved SecretRef capabilities, and route-level failures were classified but never exposed — a partial degraded state where some failures stayed invisible until a message silently went nowhere. A single process-local registry now collects every degraded owner at the boundary that owns it, so health() lists them all with a consistent, redacted shape and an actionable next step. The registry now covers both sides of the contract — the write side (record degradation where it happens) and the read side (guard a dispatch path before it targets a degraded owner).

Quick Start

1. See it in health()

   The 90% path — read the surface, no code to write. Any degraded owner (channel, provider, capability, route, or gateway) appears in one place.

   from praisonaiagents import Agent
   from praisonai_bot.gateway import BotGateway   # or the wrapper your setup uses
   
   agent = Agent(name="assistant", instructions="Be helpful.")
   gateway = BotGateway(config="gateway.yaml")
   
   # Any degraded owner — channel / provider / capability / route / gateway —
   # now shows up in one place under health()["degraded_owners"].
   print(gateway.health().get("degraded_owners", []))

   Example output when a Slack token is unavailable at boot and an MCP capability failed to resolve a secret:

   [
     {"owner_kind": "capability", "owner_id": "mcp:notion",
      "state": "cold", "reason": "secret unresolved",
      "retry_hint": "praisonai gateway doctor --fix"},
     {"owner_kind": "channel", "owner_id": "slack",
      "state": "cold", "reason": "credential unavailable",
      "retry_hint": "praisonai gateway doctor --fix"}
   ]

2. Record your own degraded owner

   Advanced — for integrations building on the gateway. Mark degradation at the boundary that owns it, then clear on recovery.

   from praisonaiagents.gateway import DegradedCapabilityRegistry, DegradedOwner
   
   registry = DegradedCapabilityRegistry()
   
   # Mark degradation at the boundary that owns it
   registry.mark(DegradedOwner(
       owner_kind="provider",         # channel | provider | capability | route | gateway
       owner_id="openai",             # stable identity
       state="cold",                  # cold (no last-known-good) | stale (reusing LKG)
       reason="auth rejected (401)",  # redacted, operator-safe
       retry_hint="re-set OPENAI_API_KEY",
   ))
   
   # Clear on recovery
   registry.clear("provider", "openai")
   
   # Read a stable, sorted snapshot
   for owner in registry.list_degraded():
       print(owner.owner_kind, owner.owner_id, owner.reason)

3. Guard a dispatch path (fail-closed)

   Call the guard before an agent targets an owner so a recorded degradation short-circuits into a typed, redacted outcome instead of a silent failure.

   from praisonaiagents import Agent
   from praisonaiagents.gateway import (
       DegradedCapabilityRegistry, DegradedOwner,
       OwnerUnavailable, assert_owner_available,
   )
   
   registry = DegradedCapabilityRegistry()
   
   # Somewhere else marked this provider as degraded:
   registry.mark(DegradedOwner(
       owner_kind="provider", owner_id="openai", state="cold",
       reason="auth rejected (401)",
       retry_hint="re-set OPENAI_API_KEY then: praisonai gateway doctor --fix",
   ))
   
   agent = Agent(name="assistant", instructions="Be helpful.", llm="openai/gpt-4o")
   
   try:
       assert_owner_available(registry, "provider", "openai")   # fail-closed guard
       response = agent.start("Summarise today's incidents.")
   except OwnerUnavailable as err:
       # Typed, redacted outcome — safe to render straight to an operator surface.
       print(err.to_dict())

How It Works

Each owner writes into the registry at its own boundary; health() reads a sorted snapshot on demand.

The registry is process-local and thread-safe, keyed on (owner_kind, owner_id) so a repeated failure for the same owner updates the existing entry instead of stacking duplicates.

Fields Reference

DegradedOwner is a frozen dataclass with five fields — retry_hint is the only one with a default.

Closed Vocabularies

Two tuples define the only valid values — import them instead of hard-coding string literals.

Both are exported from praisonaiagents.gateway, so callers can validate against them before constructing a DegradedOwner.

Concrete owners shipped with the gateway

Registry Methods

DegradedCapabilityRegistry is the default in-process implementation of the DegradedCapabilityProtocol.

cold vs stale

Pick the state by whether a last-known-good value exists.

Use cold when the owner never resolved a working value (a token that was never set). Use stale when it is still serving from a last-known-good value while degraded.

The health() Surface

degraded_owners sits alongside the existing channels map and lists every degraded owner in one redacted shape.

{
  "status": "healthy",
  "channels": { "telegram": {"running": true}, "slack": {"status": "degraded"} },
  "degraded_owners": [
    {
      "owner_kind": "channel",
      "owner_id": "slack",
      "state": "cold",
      "reason": "credential unavailable",
      "retry_hint": "praisonai gateway doctor --fix"
    },
    {
      "owner_kind": "provider",
      "owner_id": "openai",
      "state": "cold",
      "reason": "auth rejected (401)",
      "retry_hint": "re-set OPENAI_API_KEY"
    }
  ]
}

The gateway / fleet owner

When the fleet-level crash-loop breaker trips, the health monitor records exactly one gateway owner — owner_id = "fleet" — so a fleet-wide reconnect storm shows up as a single, actionable fact instead of one entry per channel.

{
  "owner_kind": "gateway",
  "owner_id": "fleet",
  "state": "stale",
  "reason": "channel crash-loop: 2/3 channels failing",
  "retry_hint": "praisonai gateway doctor"
}

/

channels failing” (or “channel crash-loop” when total == 0) | | retry_hint | praisonai gateway doctor` |

It auto-clears on the next monitor sweep once the storm subsides — no external health() / gateway status poll is needed. Wait one breaker_cooldown_s after fixing the systemic cause and the entry disappears.

The gateway / durability:* owners

When a durable store — the SQLite webhook-idempotency store or the configured session store — genuinely cannot initialise, the boundary that owns it falls back to in-memory to keep serving and records a gateway owner with owner_id = "durability:idempotency" or owner_id = "durability:session".

{
  "owner_kind": "gateway",
  "owner_id": "durability:idempotency",
  "state": "cold",
  "reason": "durable idempotency store unavailable (running in-memory)",
  "retry_hint": "praisonai gateway doctor --fix"
}

• Emitted by the praisonai-bot wrapper (praisonai_bot.bots._session) at the boundary that owns the state.
• Recorded only when a durable store genuinely cannot initialise and the wrapper has to run in-memory to keep serving inbound deliveries.
• retry_hint is praisonai gateway doctor --fix.
• Auto-cleared on the next successful build of the durable store — a config hot-reload that fixes the store path clears the entry with no operator step.
• Reasons are redacted — the raw store path stays in the log, never in health() / gateway status.

See Gateway State Durability for the full contract.

Fail-Closed Guard

Before dispatching work for an owner, call the guard so a recorded degradation short-circuits into a typed outcome instead of a silent failure.

Pick the right call for how you hold the registry.

graph TB
    Q1{Do you hold the<br/>registry directly?} -->|Yes, non-optional| M[reg.assert_owner_available kind, id]
    Q1 -->|Registry may be None| F[assert_owner_available reg, kind, id<br/>module-level, no-op if None]
    Q1 -->|Just want the record| P[reg.find kind, id → Optional DegradedOwner]

    classDef q fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef call fill:#189AB4,stroke:#7C90A0,color:#fff

    class Q1 q
    class M,F,P call

Three call shapes cover every dispatch path.

OwnerUnavailable carries only redacted, operator-safe fields — never token or secret material.

The message reads {owner_kind} {owner_id!r} unavailable ({state}): {reason} — {retry_hint}, and to_dict() returns the same redacted shape as DegradedOwner:

{"owner_kind": "provider", "owner_id": "openai", "state": "cold",
 "reason": "auth rejected (401)",
 "retry_hint": "re-set OPENAI_API_KEY then: praisonai gateway doctor --fix"}

Common Patterns

Read the surface once and act on it — for alerting or for CI.

Poll for degradation from an ops script — alert on any channel (matches existing on-call playbooks) plus any provider (new).

degraded = gateway.health().get("degraded_owners", [])
for owner in degraded:
    if owner["owner_kind"] in ("channel", "provider"):
        alert(f"{owner['owner_kind']} {owner['owner_id']} degraded: {owner['reason']}")

Fail a CI check on any degraded owner — same call, exit non-zero when the list is non-empty.

import sys

degraded = gateway.health().get("degraded_owners", [])
if degraded:
    for owner in degraded:
        print(owner["owner_kind"], owner["owner_id"], owner["retry_hint"])
    sys.exit(1)

Turn OwnerUnavailable into a chat reply — catch the guard’s exception and hand the operator the exact next action instead of a hang or a 500.

from praisonaiagents.gateway import OwnerUnavailable, assert_owner_available

def handle_chat(registry, message):
    try:
        assert_owner_available(registry, "provider", "openai")
        return agent.start(message)
    except OwnerUnavailable as err:
        return err.to_dict()["retry_hint"]

Best Practices

Related