FailurePolicy

FailurePolicy is a dataclass that maps each FailureType to a strategy callable. It is the primary configuration object passed to Agent.

FailurePolicy

FailurePolicy dataclass

Maps each FailureType to a recovery strategy callable.

Usage::

from triage import FailurePolicy
from triage.strategies.retry import retry_with_tool_manifest, backoff_and_retry
from triage.strategies.replan import replan, resume_from_subgoal
from triage.strategies.rollback import rollback_to_checkpoint

policy = FailurePolicy(
    WRONG_TOOL_CALLED  = retry_with_tool_manifest(max_attempts=3),
    CONSTRAINT_IGNORED = replan(hint="re-read the constraints carefully"),
    LOOP_DETECTED      = replan(max_replans=2),
    PLAN_INCOMPLETE    = resume_from_subgoal(),
    EXTERNAL_FAULT     = backoff_and_retry(max_attempts=5),
    default            = FailurePolicy.escalate_by_default(),
)

Any FailureType not explicitly listed falls through to default.

__init__

__init__(WRONG_TOOL_CALLED: StrategyFn | None = None, CONSTRAINT_IGNORED: StrategyFn | None = None, LOOP_DETECTED: StrategyFn | None = None, PLAN_INCOMPLETE: StrategyFn | None = None, SCHEMA_MISMATCH: StrategyFn | None = None, CONTEXT_OVERFLOW: StrategyFn | None = None, EXTERNAL_FAULT: StrategyFn | None = None, TIMEOUT: StrategyFn | None = None, UNKNOWN: StrategyFn | None = None, default: StrategyFn | None = None) -> None

escalate_by_default staticmethod

escalate_by_default() -> StrategyFn

Default strategy: always escalate to human.

abort_by_default staticmethod

abort_by_default() -> StrategyFn

Default strategy: hard abort on any unhandled failure.

chain staticmethod

chain(primary: StrategyFn, fallback: StrategyFn, after_kinds: tuple[str, ...] = ('escalate',)) -> StrategyFn

Return a strategy that tries primary and falls through to fallback when primary returns an action whose kind is in after_kinds.

The default after_kinds=("escalate",) means: use primary normally, but if primary decides to escalate, try fallback first instead.

Example — replan first, rollback if replan has already been tried::

from triage.strategies.replan import replan
from triage.strategies.rollback import rollback_to_checkpoint

policy = FailurePolicy(
    LOOP_DETECTED=FailurePolicy.chain(
        replan(hint="Try a different approach."),
        rollback_to_checkpoint(),
        after_kinds=("escalate",),
    ),
)

Example — retry up to 2 times, then replan::

policy = FailurePolicy(
    EXTERNAL_FAULT=FailurePolicy.chain(
        backoff_and_retry(max_attempts=2),
        replan(hint="External service is down, try a different approach."),
        after_kinds=("escalate",),
    ),
)

sequence staticmethod

sequence(*strategies: StrategyFn) -> StrategyFn

Return a strategy that steps through strategies in order across successive failures of the same type.

On the first failure, strategies[0] is called. On the second failure of the same type, strategies[1], and so on. Once all strategies are exhausted, returns RecoveryAction.ESCALATE.

The current position is derived from ctx.attempt_history — the number of prior attempts that share ctx.failure_type — so no external state is needed and the sequence is safe to share across concurrent runs.

Example — replan once, then rollback, then escalate::

from triage.strategies.replan import replan
from triage.strategies.rollback import rollback_to_checkpoint

policy = FailurePolicy(
    LOOP_DETECTED=FailurePolicy.sequence(
        replan(hint="Try a different approach."),
        rollback_to_checkpoint(),
    ),
)

Example — retry twice, then replan, then escalate::

policy = FailurePolicy(
    EXTERNAL_FAULT=FailurePolicy.sequence(
        backoff_and_retry(),
        backoff_and_retry(),
        replan(hint="External service may be down."),
    ),
)

from_yaml classmethod

from_yaml(path: str, *, strategy_registry: dict[str, Any] | None = None) -> FailurePolicy

Load a FailurePolicy from a TOML or YAML config file.

File format is determined by extension: - .toml — parsed with stdlib tomllib (Python 3.11+) - .yaml / .yml — parsed with pyyaml (install triage-agent[yaml])

TOML example (policy.toml)::

[EXTERNAL_FAULT]
strategy = "backoff_and_retry"
max_attempts = 5

[TIMEOUT]
strategy = "backoff_and_retry"
max_attempts = 3

[WRONG_TOOL_CALLED]
strategy = "retry_with_tool_manifest"
max_attempts = 2

default = "escalate"

YAML example (policy.yaml)::

EXTERNAL_FAULT:
  strategy: backoff_and_retry
  max_attempts: 5
default: escalate

Built-in strategy names: backoff_and_retry, retry_with_tool_manifest, replan, resume_from_subgoal, rollback_to_checkpoint, escalate, abort.

Parameters:

Name	Type	Description	Default
path	str	Path to the config file.	required
strategy_registry	dict[str, Any] | None	Optional dict mapping name → callable (or factory). When provided, merged with the built-in registry (custom names take precedence).	None

---

Conceptual notes

Declaration

Pass strategy factories as keyword arguments matching FailureType member names. default is a catch-all for any type without an explicit entry:

from triage.policy import FailurePolicy
from triage.strategies.retry import backoff_and_retry, retry_with_tool_manifest
from triage.strategies.replan import replan, resume_from_subgoal

policy = FailurePolicy(
    WRONG_TOOL_CALLED=retry_with_tool_manifest(max_attempts=3),
    SCHEMA_MISMATCH=retry_with_tool_manifest(max_attempts=2),
    EXTERNAL_FAULT=backoff_and_retry(max_attempts=5),
    LOOP_DETECTED=replan(hint="Try a different approach."),
    CONSTRAINT_IGNORED=replan(hint="Re-read the constraints carefully."),
    PLAN_INCOMPLETE=resume_from_subgoal(),
    default=FailurePolicy.escalate_by_default(),
)

Custom strategy callables

Any async def that takes a FailureContext and returns a RecoveryAction is a valid strategy:

async def smart_external_fault(ctx: FailureContext) -> RecoveryAction:
    external_faults = sum(1 for ft, _ in ctx.attempt_history if ft == FailureType.EXTERNAL_FAULT)
    if external_faults >= 3:
        return RecoveryAction.ESCALATE(message="Service unavailable after 3 retries.")
    return RecoveryAction.RETRY(delay=2.0 ** len(ctx.attempt_history))

Default actions

# Escalate to human on any unhandled failure
default = FailurePolicy.escalate_by_default()

# Hard stop on any unhandled failure
default = FailurePolicy.abort_by_default()

YAML / TOML loading

Policies can be loaded from a .yaml or .toml file (requires pyyaml for YAML):

policy = FailurePolicy.from_yaml("policy.yaml")

Built-in strategy names: backoff_and_retry, retry_with_tool_manifest, replan, resume_from_subgoal, rollback_to_checkpoint, escalate, abort. Pass strategy_registry={"my_fn": my_fn} for custom strategies.