Strategies¶

Built-in strategy factory functions. Each returns a StrategyFn = Callable[[FailureContext], Awaitable[RecoveryAction]].

retry_with_tool_manifest¶

retry_with_tool_manifest ¶

retry_with_tool_manifest(max_attempts: int = 3) -> StrategyFn

Retry with a hint to use only tools in the current manifest.

Escalates after max_attempts retries rather than deferring entirely to Agent(max_recovery_attempts=...).

backoff_and_retry¶

backoff_and_retry ¶

backoff_and_retry(max_attempts: int = 5) -> StrategyFn

Retry with exponential backoff delay. agent.py calls anyio.sleep(delay).

Escalates after max_attempts retries rather than deferring entirely to Agent(max_recovery_attempts=...).

replan¶

replan ¶

replan(hint: str | None = None, max_replans: int = 3) -> StrategyFn

Abort the current plan and generate a new one.

Escalates after max_replans replan attempts rather than deferring entirely to Agent(max_recovery_attempts=...).

resume_from_subgoal¶

resume_from_subgoal ¶

resume_from_subgoal() -> StrategyFn

Continue execution from the first incomplete sub-goal.

Requires ctx.metadata["incomplete_subgoal"] to be set by the agent before raising. If the key is missing, escalates rather than silently performing a no-op retry.

rollback_to_checkpoint¶

rollback_to_checkpoint ¶

rollback_to_checkpoint(checkpoint_id: str | None = None) -> StrategyFn

Restore state to a named checkpoint (or latest if not specified).

circuit_breaker¶

circuit_breaker ¶

circuit_breaker(breaker: CircuitBreaker, inner: StrategyFn, *, open_action: str = 'escalate') -> StrategyFn

Wrap inner with a circuit breaker guard.

Parameters:

Name	Type	Description	Default
`breaker`	`CircuitBreaker`	Shared `CircuitBreaker` instance. Must be the same object across all agents and runs that should share the failure-rate window.	required
`inner`	`StrategyFn`	The strategy to call when the breaker is CLOSED or HALF_OPEN.	required
`open_action`	`str`	What to do when the breaker is OPEN. `"escalate"` (default) raises `TriageEscalationError`; `"abort"` raises `TriageAbortError`.	`'escalate'`

Conceptual notes¶

Strategies declare intent; Agent executes it¶

A strategy receives a FailureContext and returns a RecoveryAction. It must not call the wrapped agent, restore checkpoints, or sleep. Agent executes the action.

Custom strategies¶

from triage.taxonomy import FailureContext, FailureType
from triage.policy import RecoveryAction

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

Sequencing strategies¶

Use FailurePolicy.sequence() to step through strategies across successive failures of the same type without managing state yourself:

from triage.policy import FailurePolicy
from triage.strategies.retry import backoff_and_retry
from triage.strategies.replan import replan

policy = FailurePolicy(
    EXTERNAL_FAULT=FailurePolicy.sequence(
        backoff_and_retry(max_attempts=2),
        replan(hint="External service may be down — try a different approach."),
    ),
)