FailureType & Step¶

Core data types used throughout the triage API.

FailureType¶

FailureType ¶

Bases: Enum

Semantic classification of why an agent step or plan failed.

Each type maps to a distinct recovery strategy. The classifier (rules-based or LLM) assigns one of these to every detected failure.

The 9 members and their stable string values:

Member	String value	Default recovery intent
`WRONG_TOOL_CALLED`	`wrong_tool_called`	Retry with correct manifest
`CONSTRAINT_IGNORED`	`constraint_ignored`	Replan with constraint reminder
`LOOP_DETECTED`	`loop_detected`	Replan or rollback
`PLAN_INCOMPLETE`	`plan_incomplete`	Resume from subgoal
`SCHEMA_MISMATCH`	`schema_mismatch`	Retry with schema hint
`CONTEXT_OVERFLOW`	`context_overflow`	Replan with compressed context
`EXTERNAL_FAULT`	`external_fault`	Backoff and retry
`TIMEOUT`	`timeout`	Backoff and retry
`UNKNOWN`	`unknown`	Escalate

String values are the stable public identifiers used in logs and serialized state.

Step¶

Step `dataclass` ¶

A single recorded step in an agent's execution trajectory.

metadata is caller-supplied and unenforced — strategies, hooks, and RulesClassifier may inspect it, but nothing in triage populates it automatically. Two keys have a documented convention:

metadata["http_status"] (int): the real HTTP status code from the exception that produced this step (e.g. an anthropic/openai APIStatusError.status_code, an httpx.HTTPStatusError's response.status_code, or Ollama's ResponseError.status_code).
metadata["json_rpc_code"] (int): the JSON-RPC 2.0 error code from an MCP tool-call failure (e.g. an McpError's error.code).

RulesClassifier checks both when present, in addition to its message-text patterns — see triage/classifier/rules.py's module docstring and docs/concepts/classifiers.md's "Structured error codes" section. Populating them is the caller's responsibility: extract the code from the real exception object and pass it via record_step(Step(..., metadata={"http_status": 429})).

agent_id is caller-supplied and optional — which agent produced this step, for multi-agent systems where a single Trajectory interleaves steps from more than one agent. None (the default) means either a single-agent system or an agent identity the caller didn't track; existing single-agent callers need no changes. RulesClassifier's loop detection does not currently key on it — see docs/concepts/multi-agent-failures.md for why that turned out to already be correct rather than a gap to close. triage.observability.otel_ingest.trajectory_from_spans() populates it from an OTel span's gen_ai.agent.id/gen_ai.agent.name attribute when present.

idempotent defaults to False. Mark True only for steps that are genuinely safe to replay — read-only tool calls and pure computations. Steps that send email, write to a database, or charge payment methods must stay False.

When Agent(strict_idempotency=True) is set, a RETRY action is blocked if the trajectory contains any idempotent=False step.

FailureContext¶

FailureContext `dataclass` ¶

Everything a recovery strategy needs to know about a failure.

Passed to every strategy callable and raised inside TriageEscalationError and TriageAbortError.

failed_step `property` ¶

failed_step: Step | None

The step at the critical failure index.

steps_after_failure `property` ¶

steps_after_failure: list[Step]

All steps that executed after the critical failure.

attempt_history¶

A list of (FailureType, action_kind) tuples from all prior recovery attempts in the current run() call. Use it to detect repeated failures and escalate intelligently:

prior_retries = sum(1 for _, kind in ctx.attempt_history if kind == "retry")
if prior_retries >= 2:
    return RecoveryAction.ESCALATE("Too many retries.")

TriageContext¶

TriageContext `dataclass` ¶

Structured recovery context injected into agents as _triage_context.

Replaces the scattered _triage_hint, _triage_subgoal, and _triage_state kwargs with a single typed object. The individual kwargs are still injected for backward compatibility.

Usage inside a wrapped agent::

async def my_agent(task: str, *, record_step, update_state, **kwargs) -> Any:
    ctx: TriageContext | None = kwargs.get("_triage_context")
    if ctx:
        print(ctx.failure_type, ctx.hint, ctx.attempt_number)

Injected as _triage_context on every recovery attempt:

async def my_agent(task: str, *, record_step, **kwargs) -> Any:
    tc: TriageContext | None = kwargs.get("_triage_context")
    if tc:
        print(f"Recovering from {tc.failure_type.value}, attempt {tc.attempt_number}")
        if tc.hint:
            ...  # pass hint into the LLM prompt
        if tc.state:
            data = tc.state.get("data")  # restored from checkpoint