Recovery¶
Create and retain a StrategyHinter when recovery attempts should escalate
across calls. The convenience functions are stateless when no hinter is passed.
Each hinter retains state for at most 1,024 tasks by default; hosts can select a
different positive bound and should call reset_attempts() when a task closes.
sakshi.recovery
¶
Operational failure-to-recovery hinting for agent loops.
Classifies runtime error strings into operational FailureType values and
emits typed RecoveryAction hints. Distinct from — and complementary to —
the cognitive TRAP failure taxonomy in sakshi.interpret.trap; see
sakshi.recovery.hinting for the full relationship.
RECOVERY_STRATEGIES: Mapping[FailureType, tuple[RecoveryStrategy, ...]] = MappingProxyType(_DEFAULT_STRATEGIES)
module-attribute
¶
FailureType
¶
Bases: StrEnum
Categories of recoverable operational failures.
TIMEOUT = 'timeout'
class-attribute
instance-attribute
¶
EMPTY_RESULTS = 'empty_results'
class-attribute
instance-attribute
¶
PARSE_ERROR = 'parse_error'
class-attribute
instance-attribute
¶
RATE_LIMIT = 'rate_limit'
class-attribute
instance-attribute
¶
CONNECTION_ERROR = 'connection_error'
class-attribute
instance-attribute
¶
TOOL_ERROR = 'tool_error'
class-attribute
instance-attribute
¶
MODEL_ERROR = 'model_error'
class-attribute
instance-attribute
¶
BRIDGE_FAILURE = 'bridge_failure'
class-attribute
instance-attribute
¶
RecoveryAction
¶
Bases: StrEnum
Recovery actions a host agent loop can consider.
RETRY_SAME = 'retry_same'
class-attribute
instance-attribute
¶
RETRY_PROMOTED = 'retry_promoted'
class-attribute
instance-attribute
¶
BROADEN_QUERY = 'broaden_query'
class-attribute
instance-attribute
¶
FALLBACK_TOOL = 'fallback_tool'
class-attribute
instance-attribute
¶
INCREASE_BUDGET = 'increase_budget'
class-attribute
instance-attribute
¶
REDUCE_SCOPE = 'reduce_scope'
class-attribute
instance-attribute
¶
LOCAL_ONLY = 'local_only'
class-attribute
instance-attribute
¶
SKIP_AND_LOG = 'skip_and_log'
class-attribute
instance-attribute
¶
ESCALATE = 'escalate'
class-attribute
instance-attribute
¶
RecoveryStrategy
dataclass
¶
A bounded recovery strategy for one operational failure type.
failure_type: FailureType
instance-attribute
¶
action: RecoveryAction
instance-attribute
¶
hint_template: str
instance-attribute
¶
priority: int = 0
class-attribute
instance-attribute
¶
max_attempts: int = 2
class-attribute
instance-attribute
¶
cooldown_seconds: float = 1.0
class-attribute
instance-attribute
¶
metadata: Mapping[str, Any] = field(default_factory=dict)
class-attribute
instance-attribute
¶
StrategyHinter
¶
Generate recovery hints with host-owned strategy and attempt state.
register_strategy(strategy: RecoveryStrategy) -> None
¶
Register a strategy on this hinter instance.
get_strategies(failure_type: FailureType) -> list[RecoveryStrategy]
¶
Return this instance's strategies in priority order.
classify_error(error_msg: str) -> FailureType
¶
Classify an error message into a FailureType.
get_hint(error_or_type: str | FailureType, context: Mapping[str, Any] | None = None, task_id: str = 'default') -> str
¶
Return the next recovery hint for a task and failure.
reset_attempts(task_id: str = 'default') -> None
¶
Reset all attempt counts associated with one host task.
get_strategy_action(failure_type: FailureType, task_id: str = 'default') -> RecoveryAction
¶
Return the next action without consuming an attempt.
get_strategies_for_failure(failure_type: FailureType, strategies: Mapping[FailureType, Sequence[RecoveryStrategy]] | None = None) -> list[RecoveryStrategy]
¶
Return strategies for a failure type, highest priority first.
hint_for_empty_results(context: Mapping[str, Any] | None = None, task_id: str = 'default', *, hinter: StrategyHinter | None = None) -> str
¶
Return an empty-results hint, stateless unless hinter is supplied.
hint_for_parse_error(context: Mapping[str, Any] | None = None, task_id: str = 'default', *, hinter: StrategyHinter | None = None) -> str
¶
Return a parse-error hint, stateless unless hinter is supplied.
hint_for_timeout(context: Mapping[str, Any] | None = None, task_id: str = 'default', *, hinter: StrategyHinter | None = None) -> str
¶
Return a timeout recovery hint, stateless unless hinter is supplied.
wrap_with_resilience(observation: Any, task_id: str = 'default', *, hinter: StrategyHinter | None = None) -> Any
¶
Append a recovery hint when a tool observation looks like a failure.
Without hinter, each call is intentionally stateless. Pass a host-owned
StrategyHinter when repeated observations should advance through the
configured strategy budgets.