Unified Degraded-Capability Registry

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

This is a purely additive surface — there is no gateway.yaml knob and no CLI flag. The channel-only channels map in health() is unchanged, so existing monitors keep working. The registry, its dataclass, its protocol, and the closed vocabularies are all public exports from praisonaiagents.gateway.

DegradedCapabilityProtocol (write contract: mark/clear/list_degraded) is unchanged. Registries that also implement the point read find(...) satisfy the extended DegradedCapabilityLookupProtocol. The module-level assert_owner_available() guard degrades gracefully on legacy registries by scanning list_degraded() — so upgrading the SDK never breaks an external registry written against the original contract.

Quick Start

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"}
]

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)

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.

Field	Type	Default	Description
`owner_kind`	`str`	required	One of `channel`, `provider`, `capability`, `route`, `gateway`. Anything else raises `ValueError` at construction.
`owner_id`	`str`	required	Stable identity for the owner (e.g. `telegram:main`, `openai`, `mcp:notion`).
`state`	`str`	required	Either `cold` (no last-known-good) or `stale` (reusing last-known-good). Anything else raises `ValueError`.
`reason`	`str`	required	Redacted, operator-safe explanation. Must never leak token/secret material.
`retry_hint`	`str`	`""`	The exact next action, e.g. `praisonai gateway doctor --fix`. Every `retry_hint` MUST name a command that exists — `praisonai gateway doctor --fix` is the canonical example (see Gateway CLI › Auto-repair).

Closed Vocabularies

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

Constant	Values
`OWNER_KINDS`	`("channel", "provider", "capability", "route", "gateway")`
`DEGRADED_STATES`	`("cold", "stale")`

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

route:redis-pubsub is the built-in owner recorded by the gateway HA fan-out adapter (RedisPubSubAdapter) whenever the Redis pub/sub connection is dropped — the first shipped route owner and a good exemplar for the route owner_kind. See Real-Time Push Notifications → HA & cross-instance delivery.

Concrete owners shipped with the gateway

`owner_kind`	`owner_id`	Reason surfaced
`channel`	`<platform>`	Channel credential unavailable or supervision marked it degraded.
`provider`	`<llm>`	LLM provider auth rejected or unreachable.
`capability`	`mcp:<name>`	MCP capability failed to resolve a secret.
`route`	`redis-pubsub`	Cross-instance HA transport (Redis pub/sub) is disconnected; reconnecting with bounded backoff.
`gateway`	`fleet`	Fleet-level crash-loop breaker tripped.

Registry Methods

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

Method	Signature	Behaviour
`mark`	`mark(owner: DegradedOwner) -> None`	Record (or update) a degraded owner. Last write wins per `(owner_kind, owner_id)` key. Thread-safe.
`clear`	`clear(owner_kind: str, owner_id: str) -> None`	Remove on recovery. Idempotent — clearing a missing key is a no-op.
`list_degraded`	`list_degraded() -> List[DegradedOwner]`	Stable, sorted-by-`(owner_kind, owner_id)` snapshot.
`find`	`find(owner_kind: str, owner_id: str) -> Optional[DegradedOwner]`	Point read: the record if degraded, else `None`. Thread-safe.
`assert_owner_available`	`assert_owner_available(owner_kind: str, owner_id: str) -> None`	Fail-closed guard: raises `OwnerUnavailable` if degraded, no-op if healthy.
`to_list`	`to_list() -> List[Dict[str, str]]`	Same as `list_degraded()` rendered as plain dicts for JSON.

`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 channel-only channels map is unchanged — this is a purely additive surface, safe for existing monitors.
degraded_owners is present only when the list is non-empty.
The aggregation is defensive — health() never raises even if a registry read fails.

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

Field	Value
`owner_kind`	`gateway`
`owner_id`	`fleet`
`state`	`stale` — the fleet is still serving from last-known-good while restarts are held
`reason`	`"channel crash-loop: <N>/<M> 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.

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. Three call shapes cover every dispatch path.

Call	Signature	Behaviour
`reg.find`	`find(owner_kind: str, owner_id: str) -> Optional[DegradedOwner]`	Point read. Returns the record if degraded, else `None`. Thread-safe.
`reg.assert_owner_available`	`assert_owner_available(owner_kind: str, owner_id: str) -> None`	Raises `OwnerUnavailable` if degraded. No-op if healthy.
`praisonaiagents.gateway.assert_owner_available`	`assert_owner_available(registry: Optional[DegradedCapabilityProtocol], owner_kind: str, owner_id: str) -> None`	Same guard for a possibly-absent registry. No-op when `registry is None`. Falls back to `list_degraded()` on legacy registries that lack `find`/the guard.

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

Attribute	Type	Notes
`owner`	`DegradedOwner`	Full redacted record
`owner_kind`	`str`	Convenience mirror of `owner.owner_kind`
`owner_id`	`str`	Convenience mirror of `owner.owner_id`
`state`	`str`	`cold` or `stale`
`reason`	`str`	Redacted, operator-safe
`retry_hint`	`str`	Exact next action
`to_dict()`	method	Same shape as `DegradedOwner.to_dict()` — safe to serialise to a JSON response

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

Use the closed vocabularies

Import OWNER_KINDS and DEGRADED_STATES from praisonaiagents.gateway and validate against them before constructing a DegradedOwner. The frozen dataclass raises ValueError on an unknown owner_kind or state, so checking first turns a runtime crash into a clean guard.

Redact secrets in reason

The registry is operator-facing — a token or secret in reason surfaces on every dashboard, log line, and status probe. Say what failed, not what the value was: "auth rejected (401)", not the key itself.

Set an actionable retry_hint

praisonai gateway doctor --fix is the sanctioned default — the command exists and actually repairs a weak/missing gateway auth token, then re-validates (see Gateway CLI › Auto-repair). Every retry_hint in this registry MUST name a command that exists. Prefer a specific one that names the exact env var, secret file, or rotation command — e.g. re-set OPENAI_API_KEY — so the operator’s next step is unambiguous.

Clear on recovery

mark() alone leaves the entry until the process restarts. Pair every mark() at the failure point with a clear() on the success path so health() reflects reality instead of a stale failure.

Guard before you dispatch, not after

Call assert_owner_available() immediately before the work targeting that owner. A guard after the fact converts a silent failure into a loud one but wastes the round-trip; a guard before it keeps the failure typed and free.

Degraded Channel Isolation

The channel-specific boot-time isolation contract this composes over.

Gateway Channel Supervision

The runtime supervisor that produces the stale channel entries.

Gateway CLI

gateway doctor / gateway status — the operator surfaces that render degraded_owners.

Gateway Secret References

The source of credential unavailable reasons.

Weak / Placeholder Secret Guard

The most common producer of the secret unresolved records this guard then fails-closed against.

​Quick Start

​How It Works

​Fields Reference

​Closed Vocabularies

​Concrete owners shipped with the gateway

​Registry Methods

​cold vs stale

​The health() Surface

​The gateway / fleet owner

​Fail-Closed Guard

​Common Patterns

​Best Practices

​Related

Degraded Channel Isolation

Gateway Channel Supervision

Gateway CLI

Gateway Secret References

Weak / Placeholder Secret Guard

Quick Start

How It Works

Fields Reference

Closed Vocabularies

Concrete owners shipped with the gateway

Registry Methods

`cold` vs `stale`

The `health()` Surface

The `gateway` / `fleet` owner

Fail-Closed Guard

Common Patterns

Best Practices

Related