Approvals & recovery

Part of Advanced topics. Start with Durable agent runs for the runtime API, session identity, run handles, and lifecycle.

Recovery must distinguish a recorded result from a tool whose external effect is uncertain. Persisted state cannot by itself prove whether an external system accepted a request.

Declare the replay contract

Policy	Interrupted operation	Application responsibility
manual (default)	Wait for explicit reconciliation	Verify the outcome, accept a retry, or abort
safe	May be invoked again	The operation must be safe to repeat
idempotent	May be invoked again with its execution key	The external adapter/service must enforce idempotency

The runtime persists dispatch intent before invoking a tool, and persists its raw result before result middleware. A recorded result can be reused. A dispatch record without a result leaves an ambiguity window. replay_policy accepts the ReplayPolicy enum (ReplayPolicy.IDEMPOTENT) or the plain strings; the definition stores the string.

Use an execution key

from harnessx import PermissionLevel, current_tool_context

def install_orders(agent, orders):
    @agent.tools.register(
        permission=PermissionLevel.ASK,
        replay_policy="idempotent",
    )
    async def create_order(customer_id: str) -> str:
        """Create an order for an approved customer."""
        key = current_tool_context().execution_key
        return await orders.create(customer_id, idempotency_key=key)

Here orders is your application adapter. The service must reject or deduplicate repeated writes with that key. Merely accepting a parameter called idempotency_key is insufficient.

Persist and answer approvals

In AgentRuntime, ASK tools yield persisted pending requests: the run returns needs_input, and result.pending lists PendingTool objects with status == "approval". Present the exact tool and arguments to the user, then record the decision. With resume=True the run continues and the finished RunResult is returned.

async def answer_approval(runtime, result, allow):
    pending = result.pending[0]
    print(pending.call.name, pending.call.input)
    if allow:
        return await runtime.approve(pending, resume=True)
    return await runtime.decline(pending, resume=True)

approve() and decline() accept a PendingTool or its execution key. A run with several pending tools takes one decision per tool without resume, then a single await runtime.resume(session_id). A declined tool receives the result "Permission denied". Workers never implicitly read stdin. A web or terminal application supplies the approval interaction.

Resolve an uncertain outcome

A PendingTool with status == "uncertain" might already have produced an external effect. resolve_tool() accepts the pending entry or its execution key:

from harnessx import ToolResult

async def record_verified_result(runtime, pending, outcome):
    await runtime.resolve_tool(
        pending,
        result=ToolResult(pending.call.id, outcome),
    )

Alternatively, use retry=True to explicitly accept possible repetition, or abort=True to stop the operation. Choose exactly one resolution, then resume the session. Inspect the external system before supplying a verified outcome. A key the run does not have raises UnknownExecutionKey; a decision that does not fit the tool's state raises ResolutionError.

Pause, cancellation, and failures

Pause stops new steps and lets active tools settle. Cancellation is cooperative; Python threads or remote requests may outlive it. Neither operation rolls back completed effects.

Interrupted model calls can be invoked again with a new attempt. The result may differ and unknown charges can mark usage incomplete. Storage commit failure is not proof that an external tool failed.

Try the approval workflow

From the SDK checkout:

uv run python -m examples.runtime_approvals

This scripted example exercises SQLite persistence without model credentials. See Flight recorder to capture the boundaries needed for incident inspection.