Healing API

healing

Healing primitives — watchdogs, recovery actions, model fallback.

The cognitive substrate's "immune system." Watchdogs subscribe to the :class:EventBus, detect stall / loop / regression patterns, and feed :class:RecoveryAction directives through a :class:WatchdogHook into the existing :class:HookDecision pipeline.

See :doc:/concepts/healing for the full picture; this module hosts:

• :class:HealingConfig — frozen dataclass.
• :class:Watchdog Protocol + :class:Detection Pydantic model.
• :class:StallDetector / :class:LoopDetector / :class:RegressionDetector — concrete watchdogs.
• :class:RecoveryAction discriminated union + :class:WatchdogHook.
• :func:resolve_model_with_fallback + :class:FallbackModel.
• :class:HealingRunner — wires everything to a :class:Workbench's bus.

HealingConfig dataclass

HealingConfig(
    stall_timeout_s: float = 60.0,
    loop_threshold_k: int = 3,
    loop_window_n: int = 10,
    regression_window_n: int = 5,
    regression_drop_threshold: float = 0.2,
    poll_interval_s: float = 1.0,
    fallback_models: tuple[str, ...] = tuple(),
    enable_stall: bool = True,
    enable_loop: bool = True,
    enable_regression: bool = False,
)

Knobs for healing primitives. Pass explicitly to :class:HealingRunner.

stall_timeout_s class-attribute instance-attribute

stall_timeout_s: float = 60.0

Detector fires when an open tool call exceeds this duration.

loop_threshold_k class-attribute instance-attribute

loop_threshold_k: int = 3

Same (tool_name, args_hash) more than k times in the sliding window triggers :class:LoopDetector.

loop_window_n class-attribute instance-attribute

loop_window_n: int = 10

Sliding-window size for :class:LoopDetector — number of recent tool.before events tracked.

regression_window_n class-attribute instance-attribute

regression_window_n: int = 5

Number of recent metacognition.confidence values :class:RegressionDetector averages over each half (head vs tail).

regression_drop_threshold class-attribute instance-attribute

regression_drop_threshold: float = 0.2

Confidence drop (head_mean − tail_mean) required to trigger :class:RegressionDetector.

poll_interval_s class-attribute instance-attribute

poll_interval_s: float = 1.0

How often :class:HealingRunner polls watchdogs.

fallback_models class-attribute instance-attribute

fallback_models: tuple[str, ...] = field(
    default_factory=tuple
)

Ordered list of provider:model_id strings for :func:resolve_model_with_fallback. Empty disables auto-fallback.

enable_regression class-attribute instance-attribute

enable_regression: bool = False

Off by default — needs metacognition.confidence events flowing on the bus.

FallbackModel

FallbackModel(
    models: list[Model],
    *,
    bus: EventBus | None = None,
    transient_predicate: Callable[[BaseException], bool]
    | None = None,
)

pydantic-AI :class:Model wrapping a chain of fallback models.

On transient failure during :meth:request or :meth:request_stream, advances _idx to the next model. The advance is sticky across requests in the same instance — once the primary failed, subsequent calls go straight to the fallback. Non-transient failures propagate.

Source code in orqest/healing/fallback.py

def __init__(
    self,
    models: list[Model],
    *,
    bus: EventBus | None = None,
    transient_predicate: Callable[[BaseException], bool] | None = None,
) -> None:
    super().__init__()
    if not models:
        raise ValueError("FallbackModel requires at least one underlying model")
    self._models = list(models)
    self._bus = bus
    self._is_transient = transient_predicate or _default_transient_predicate
    self._idx = 0

active_model property

active_model: Model

The currently-active underlying model.

request_stream async

request_stream(
    messages: list[ModelMessage],
    model_settings: ModelSettings | None,
    model_request_parameters: ModelRequestParameters,
    run_context: Any = None,
) -> AsyncIterator[Any]

Streaming variant. Mid-stream errors propagate (we may have already yielded chunks). Only pre-stream failures fall back.

Source code in orqest/healing/fallback.py

@asynccontextmanager
async def request_stream(
    self,
    messages: list[ModelMessage],
    model_settings: ModelSettings | None,
    model_request_parameters: ModelRequestParameters,
    run_context: Any = None,
) -> AsyncIterator[Any]:
    """Streaming variant. Mid-stream errors propagate (we may have
    already yielded chunks). Only pre-stream failures fall back."""
    last_exc: BaseException | None = None
    for i in range(self._idx, len(self._models)):
        try:
            async with self._models[i].request_stream(
                messages, model_settings, model_request_parameters, run_context
            ) as stream:
                yield stream
            return
        except Exception as exc:
            last_exc = exc
            if not self._is_transient(exc):
                raise
            await self._emit_fallback(i, i + 1, exc)
            self._idx = i + 1
    await self._emit_chain_exhausted(last_exc)
    raise RuntimeError(
        f"All fallback models exhausted (stream); last error: {last_exc}"
    ) from last_exc

LoopDetector

LoopDetector(*, threshold_k: int = 3, window_n: int = 10)

Detects an agent looping on the same tool call.

Maintains a deque of the last window_n (tool_name, args_hash) pairs. When the current pair appears more than threshold_k times in the window, raises a Detection. Each fire suppresses subsequent detections until a different pair appears (so the watchdog doesn't spam the same loop).

Source code in orqest/healing/loop.py

def __init__(self, *, threshold_k: int = 3, window_n: int = 10) -> None:
    if threshold_k < 1:
        raise ValueError("threshold_k must be >= 1")
    if window_n < threshold_k:
        raise ValueError("window_n must be >= threshold_k")
    self._k = threshold_k
    self._n = window_n
    self._window: collections.deque[tuple[str, str]] = collections.deque(
        maxlen=window_n
    )
    self._latest: Detection | None = None
    self._last_fired_pair: tuple[str, str] | None = None
    self._subscribed: bool = False

AbortRun

Abort the compound flow.

EscalateToUser

Stop autonomous execution and ask the user a question.

WatchdogHook

WatchdogHook(
    watchdogs,
    *,
    policy: Callable[[Detection], RecoveryAction]
    | None = None,
    bus: EventBus | None = None,
)

ToolHook that consults registered watchdogs and converts pending Detections into :class:HookDecision directives.

Runs the policy on the first watchdog whose :meth:signal returns a Detection in registration order. Subsequent watchdogs are still polled so they advance their internal state, but their detections are dropped (no second decision fires per call).

Returns None from :meth:after_tool and :meth:on_error — decisions only fire on :meth:before_tool (the natural seam where we can replace/skip the next call).

Source code in orqest/healing/recovery.py

def __init__(
    self,
    watchdogs,
    *,
    policy: Callable[[Detection], RecoveryAction] | None = None,
    bus: EventBus | None = None,
) -> None:
    self._watchdogs = list(watchdogs)
    self._policy = policy or default_policy
    self._bus = bus

RegressionDetector

RegressionDetector(
    *, window_n: int = 5, drop_threshold: float = 0.2
)

Confidence-trend watchdog.

Maintains a sliding window of the last window_n confidence values. Fires when the head-half mean exceeds the tail-half mean by at least drop_threshold — i.e. the agent is becoming less certain over time.

Cumulative-mean diff over half-windows is robust against single- sample noise without needing full statistics. Window size of 4-6 typically suffices.

Source code in orqest/healing/regression.py

def __init__(self, *, window_n: int = 5, drop_threshold: float = 0.2) -> None:
    if window_n < 2:
        raise ValueError("window_n must be >= 2")
    if not 0.0 <= drop_threshold <= 1.0:
        raise ValueError("drop_threshold must be in [0, 1]")
    self._n = window_n
    self._drop = drop_threshold
    self._scores: collections.deque[float] = collections.deque(maxlen=window_n)
    self._latest: Detection | None = None
    self._fired_for_window: bool = False
    self._subscribed: bool = False

HealingRunner

HealingRunner(
    config: HealingConfig,
    *,
    bus: EventBus,
    api_key: ApiKeyArg | None = None,
    watchdogs: list[Watchdog] | None = None,
)

Owns the healing lifecycle for a single :class:EventBus.

Constructs watchdogs from :class:HealingConfig, subscribes them to the bus, and runs a poll loop that:

• calls each watchdog's :meth:signal periodically (so detectors that compute on demand, like :class:StallDetector, fire);
• emits healing.detection events for each Detection produced.

The :class:WatchdogHook is exposed as runner.hook for the consumer to register on a :class:HookRunner.

The fallback :class:Model (if configured) is exposed as runner.model; pass it to :class:BaseAgent.

Source code in orqest/healing/runner.py

def __init__(
    self,
    config: HealingConfig,
    *,
    bus: EventBus,
    api_key: ApiKeyArg | None = None,
    watchdogs: list[Watchdog] | None = None,
) -> None:
    self._config = config
    self._bus = bus
    self._watchdogs: list[Watchdog] = list(watchdogs) if watchdogs else []
    if not self._watchdogs:
        if config.enable_stall:
            self._watchdogs.append(
                StallDetector(timeout_s=config.stall_timeout_s)
            )
        if config.enable_loop:
            self._watchdogs.append(
                LoopDetector(
                    threshold_k=config.loop_threshold_k,
                    window_n=config.loop_window_n,
                )
            )
        if config.enable_regression:
            self._watchdogs.append(
                RegressionDetector(
                    window_n=config.regression_window_n,
                    drop_threshold=config.regression_drop_threshold,
                )
            )
    for wd in self._watchdogs:
        wd.subscribe(self._bus)

    self.hook = WatchdogHook(self._watchdogs, bus=self._bus)
    self._poll_task: asyncio.Task[None] | None = None

    self._fallback_model: Model | None = None
    if config.fallback_models and api_key is not None:
        try:
            self._fallback_model = resolve_model_with_fallback(
                list(config.fallback_models),
                api_key=api_key,
                bus=self._bus,
            )
        except Exception as exc:
            logger.warning(
                "Could not configure fallback model chain {m!r}: {e}",
                m=config.fallback_models,
                e=exc,
            )

model property

model: Model | None

The fallback-aware :class:Model, or None if not configured.

watchdogs property

watchdogs: list[Watchdog]

The active watchdog list (for inspection / tests).

StallDetector

StallDetector(*, timeout_s: float = 60.0)

Tracks open tool calls; raises :class:Detection when one exceeds timeout_s.

Subscribes to tool.before to mark a call as open and tool.after / tool.error to close it. :meth:signal is polled by :class:HealingRunner and returns at most one Detection per open call (subsequent polls for the same open call are suppressed).

Source code in orqest/healing/stall.py

def __init__(self, *, timeout_s: float = 60.0) -> None:
    self._timeout_s = timeout_s
    self._open_calls: dict[str, datetime] = {}
    self._fired: set[str] = set()
    self._subscribed: bool = False

Detection

A signal raised by a :class:Watchdog when its condition fires.

Watchdog

A subsystem that observes the EventBus and raises Detections.

Implementations subscribe via :meth:subscribe and either:

• raise inline from event handlers, caching the latest Detection for the next :meth:signal poll (LoopDetector, RegressionDetector); or
• compute on demand at poll time (StallDetector).

Subscriptions must be idempotent — calling :meth:subscribe twice must not cause duplicate handler dispatch.

resolve_model_with_fallback

resolve_model_with_fallback(
    models: list[str],
    *,
    api_key: ApiKeyArg,
    bus: EventBus | None = None,
    transient_predicate: Callable[[BaseException], bool]
    | None = None,
) -> Model

Resolve a chain of provider:model_id strings into a single :class:FallbackModel.

Resolution failures (unknown provider, missing SDK, missing per- provider key) are logged at DEBUG and skipped. The chain remains valid as long as one entry resolves; raises ValueError if none do.

Parameters:

Name	Type	Description	Default
models	list[str]	Ordered list — first is primary.	required
api_key	ApiKeyArg	Single key (used for every provider) or per-provider map.	required
bus	EventBus | None	Optional :class:EventBus. Emits healing.model_fallback on each chain advance.	None
transient_predicate	Callable[[BaseException], bool] | None	Override the default classifier.	None

Source code in orqest/healing/fallback.py

def resolve_model_with_fallback(
    models: list[str],
    *,
    api_key: ApiKeyArg,
    bus: EventBus | None = None,
    transient_predicate: Callable[[BaseException], bool] | None = None,
) -> Model:
    """Resolve a chain of ``provider:model_id`` strings into a single
    :class:`FallbackModel`.

    Resolution failures (unknown provider, missing SDK, missing per-
    provider key) are logged at DEBUG and skipped. The chain remains
    valid as long as *one* entry resolves; raises ``ValueError`` if
    none do.

    Args:
        models: Ordered list — first is primary.
        api_key: Single key (used for every provider) or per-provider map.
        bus: Optional :class:`EventBus`. Emits ``healing.model_fallback``
            on each chain advance.
        transient_predicate: Override the default classifier.
    """
    resolved: list[Model] = []
    for spec in models:
        provider = spec.split(":", 1)[0]
        key = api_key if isinstance(api_key, str) else api_key.get(provider, "")
        if not key:
            logger.debug("No API key for provider {p}; skipping {s}", p=provider, s=spec)
            continue
        try:
            resolved.append(resolve_model(spec, api_key=key))
        except Exception as exc:
            logger.debug("resolve_model({s}) failed: {e}; skipping", s=spec, e=exc)
    if not resolved:
        raise ValueError(
            f"No model in {models!r} could be resolved. "
            "Check provider names and api_key map."
        )
    return FallbackModel(
        resolved, bus=bus, transient_predicate=transient_predicate
    )

default_policy

default_policy(detection: Detection) -> RecoveryAction

Map a Detection to a sensible default RecoveryAction.

Consumers can override by passing a custom policy to :class:WatchdogHook. The default is conservative: stalls/loops abort by default; regressions also abort.

Source code in orqest/healing/recovery.py

def default_policy(detection: Detection) -> RecoveryAction:
    """Map a Detection to a sensible default RecoveryAction.

    Consumers can override by passing a custom policy to
    :class:`WatchdogHook`. The default is conservative: stalls/loops
    abort by default; regressions also abort.
    """
    if detection.detector == "stall":
        return AbortRun(reason=f"stall: {detection.summary}")
    if detection.detector == "loop":
        return AbortRun(reason=f"loop: {detection.summary}")
    if detection.detector == "regression":
        return AbortRun(reason=f"regression: {detection.summary}")
    return AbortRun(reason=f"unknown detector: {detection.detector}")