Durable agent runs

Persist execution, identify sessions and runs, resume work, and use run handles for results and progress.

This is an advanced topic. Start with Agents, Tools, and Streaming. Add durable execution when an application needs to recover unfinished work or persist approval waits across process restarts.

For example, a sales agent writes a report, saves the tool result, and then its Python process crashes before delivering the answer. A replacement process can restore the run, reuse that committed result, and finish the answer. The PostgreSQL walkthrough demonstrates this with a report file and a tool invocation count that stays at one.

When to use durable execution

Need	API	What is retained
Run an agent in your application process	`Agent.run()` or `Agent.run_stream()`	Conversation state while that instance lives
Explicitly save and reload a conversation	`Agent.save_session()` / `Agent.load_session()`	A conversation snapshot, configuration, and supported extension state
Recover interrupted execution or a persisted approval	`AgentRuntime`	Execution boundaries, messages, tool outcomes, approvals, events, and artifact references

An ordinary agent does not need a runtime or registry. A conversation snapshot does not resume an in-flight tool call. Choose AgentRuntime when those execution boundaries matter to your workflow.

What each object does

Object	Responsibility
`Agent`	Defines the model, prompt, tools, permissions, middleware, and extensions
`PostgresBackend`	Stores durable state in your PostgreSQL database
`AgentRuntime`	Drives the agent, persists progress, and exposes start, resume, approval, and control operations
`RunHandle`	Gives your application access to one submitted or resumed run's progress and result
`RunResult`	Describes the run's output, status, usage, errors, and pending operations

For PostgreSQL, pass the configured agent directly: AgentRuntime(agent, backend=backend). No AgentRegistry or AgentRef is required for this pattern.

On a single host, SQLiteBackend is the default store. Every backend is an async context manager that initialises on entry and closes on exit, and await SQLiteBackend.connect(path) constructs and initialises one in a single call when a context manager does not fit:

from harnessx import Agent, AgentRuntime, SQLiteBackend


async def local_run(prompt: str):
    async with SQLiteBackend("runtime.sqlite3") as backend:
        async with AgentRuntime(Agent(), backend=backend) as runtime:
            return await runtime.run(prompt)

The runtime offers the same three verbs as Agent: run() waits for the RunResult, run_stream() yields typed events, and stream_text() yields the answer's text and raises RunError if the run does not complete. The 0.3 names execute(), execute_stream(), and get_status() remain as deprecated aliases until 0.5.

Start a PostgreSQL run

Install the PostgreSQL extra and provide a connection string for an existing database. The connection setup covers the JSON config, password placeholder, TLS settings, and connection check.

These functions illustrate the SDK API with a live model. They use Anthropic by default and require its API key. Call them from your async application with await; the separate crash demonstration uses a scripted provider and needs no model key.

import json
from pathlib import Path

from harnessx import Agent, AgentConfig, AgentRuntime, PostgresBackend, RunResult


async def start_run(dsn: str, session_file: Path, prompt: str) -> RunResult:
    async with PostgresBackend(dsn) as backend:  # opens the pool and migrates on entry
        agent = Agent(config=AgentConfig(system_prompt="You are helpful."))
        async with agent, AgentRuntime(agent, backend=backend) as runtime:
            session_file.parent.mkdir(parents=True, exist_ok=True)
            with session_file.open("x", encoding="utf-8") as file:
                json.dump({"session_id": runtime.session_id}, file)

            handle = await runtime.submit(prompt, request_id="first-request")
            print("Session:", handle.session_id)
            print("Run:", handle.run_id)
            return await handle.result()

For example, call await start_run(dsn, Path(".agent_sessions/my-chat.json"), "Hello"). Use an unused session-file path. The function saves the ID before submitting work and refuses to overwrite an existing file. In a service, store that ID alongside the application's chat or workflow record instead.

Entering async with AgentRuntime(...) calls start() automatically. submit() starts driving a run and returns its handle. The example waits for the result before closing the runtime. If you only need a result, await runtime.run(prompt, request_id=...) combines submission and waiting; runtime.run_stream(...) and runtime.stream_text(...) are the streaming forms.

Sessions and run IDs

Identifier	Meaning	Application use
`session_id`	One durable conversation or workflow session	Save when starting; pass to `resume()` after a restart
`run_id`	One submitted request within that session	Associate output and progress with a particular execution
`request_id`	Application-supplied submission deduplication key	Reuse for the same request in the same session
Event `cursor`	A position in one run's stored event stream	Reconnect after the last processed event

A session can contain multiple sequential runs. Resuming an unfinished run preserves its run ID; submitting another message creates another run. resume() takes a session ID, loads that session's latest persisted run, and resumes it if eligible. It does not select an arbitrary session or look up a user automatically.

Request deduplication is scoped to a session. It does not make an external tool effect idempotent. Your application must also retain the session ID; repeating start() creates a new session.

Resume after a restart

Create a fresh agent with matching configuration and compatible tools and extensions. Connect to the same PostgreSQL database and schema. Read the saved session ID and explicitly resume it:

import json
from pathlib import Path

from harnessx import Agent, AgentConfig, AgentRuntime, PostgresBackend, RunResult


async def resume_run(dsn: str, session_file: Path) -> RunResult | None:
    saved = json.loads(session_file.read_text(encoding="utf-8"))
    async with PostgresBackend(dsn) as backend:
        agent = Agent(config=AgentConfig(system_prompt="You are helpful."))
        async with agent:
            runtime = AgentRuntime(agent, backend=backend)
            try:
                handle = await runtime.resume(saved["session_id"])
                if handle is None:
                    print("No run was resumed.")
                    return None
                print("Resuming run:", handle.run_id)
                return await handle.result()
            finally:
                await runtime.aclose()  # alias of stop()

Call await resume_run(dsn, Path(".agent_sessions/my-chat.json")). If the earlier function finished normally, there is no unfinished run to drive. Use the crash demonstration below to exercise recovery.

Do not enter the runtime context manager before calling resume(). That context manager starts a new session. The outer async with agent only manages the agent's resources and is safe for this recovery pattern.

resume() returns a handle for a resumed run, or None when the session has no run or its latest run is completed/cancelled. It can raise for a missing session, incompatible definition/configuration, or an active worker lease. None does not mean the session could not be found.

On a successfully bound runtime, await runtime.status() returns a dictionary with session_id, state, and run, the latest RunResult or None for a session without a run. This lets an application read a completed result even when resume() returns no handle.

How the CLI chooses a session

The PostgreSQL demonstration's crash command saves the generated ID in .agent_sessions/postgres-demo/session.json:

{
  "session_id": "the-generated-session-id",
  "schema": "harness_x"
}

Its resume command reads that file and calls runtime.resume(saved["session_id"]). The SDK always receives an explicit session ID. The file is a pointer to the database state; it contains neither the password nor the saved execution itself.

From the repository checkout, after configuring PostgreSQL:

.venv/bin/uv run python -m examples.postgres_runtime check --config examples/postgres.config.json
.venv/bin/uv run python -m examples.postgres_runtime crash --config examples/postgres.config.json
.venv/bin/uv run python -m examples.postgres_runtime status --config examples/postgres.config.json
# Wait at least 30 seconds after the crash for the previous worker's lease to expire.
.venv/bin/uv run python -m examples.postgres_runtime resume --config examples/postgres.config.json

check prepares/validates the schema and creates no agent session. crash intentionally exits with code 42 after committing the report tool's result. resume should report completed with Tool invocations: 1. The separate chat command runs a live-model example; it does not create this crash demo's session file.

If you pass --workspace, use the same directory on every command; the demo reads session.json from that workspace. Choose a fresh workspace for another demonstration. The local files and PostgreSQL must remain available for this particular example.

Use a RunHandle

A handle provides access to an execution. It is not the agent's answer, and it is not a running agent to serialize into your database. Store the relevant IDs instead.

Handle API	What it provides
`handle.session_id`	The session this execution belongs to
`handle.run_id`	This execution's ID
`await handle.result()`	Wait for a `RunResult` from the current execution
`handle.events(after=cursor)`	Read stored events and follow new ones
`handle.stream(after=cursor)`	An async-context-managed event stream with its own `result()` method

Both await runtime.submit(...) and a successful await runtime.resume(...) return a handle. run() returns a RunResult directly. Use the new handle returned when resuming after an approval or pause; an earlier local handle can retain its earlier execution result. handle.backend is deprecated; runtimes own their backend.

Waiting for a result does not guarantee completed. A run can stop at an approval wait, pause, cancellation, or failure. Inspect its status before treating the output as a successful answer. Infrastructure failures can also raise exceptions; normal application exception handling still applies.

Understand a RunResult

Field	Meaning
`session_id`, `run_id`	Identify the conversation and execution
`output`	The answer accumulated for this run; check status before using it as final output
`status`	`running`, `completed`, `paused`, `awaiting_input`, `cancelled`, or `failed`
`error`	Failure details containing `type` and `message`, or `None`
`pending`	A list of `PendingTool` objects: approvals and uncertain tool operations, read through `execution_key`, `call.name`, `call.input`, and `status`
`stop_reason`	The recorded reason execution stopped
`usage`	Recorded token usage
`usage_incomplete`	Whether interrupted model work leaves usage accounting incomplete

from harnessx import RunResult


def present_result(result: RunResult) -> None:
    print("Run:", result.run_id)
    print("Status:", result.status.value)
    if result.ok:
        print("Answer:", result.output)
    elif result.failed:
        print("Error:", result.error["message"])
    for pending in result.pending:
        print(pending.status, pending.execution_key, pending.call.name)

result.ok, result.failed, and result.needs_input test the three statuses an application usually branches on; result.raise_for_status() returns the result or raises RunFailed, RunAwaitingInput, RunTruncated, or RunCancelled, each carrying the result. PendingTool also exposes status ("approval" or "uncertain"), attempt, policy, concurrent, and timeout_seconds. Dictionary access (pending["execution_key"]) still works but is deprecated.

running normally appears when inspecting an active run. paused means execution settled at a pause boundary. awaiting_input requires an approval or reconciliation decision. Investigate failed runs before resuming. A cancelled run is not automatically resumed and can still have uncertain effects to inspect.

Stream progress and reconnect

For a UI, consume the handle's events. This helper takes two async callbacks supplied by your application: send(event) renders or forwards a typed event, and save_cursor(run_id, cursor) remembers the last processed position.

from harnessx import RunHandle, RunResult


async def forward_progress(
    handle: RunHandle, send, save_cursor, *, after: str = "0"
) -> RunResult:
    async for event in handle.events(after=after):
        await send(event)
        if event.cursor:
            await save_cursor(handle.run_id, event.cursor)
    return await handle.result()

On reconnect, provide the saved cursor for that same run. With no cursor, SQL handles start from "0" and include previously stored events. A replayed tool event describes past execution; receiving it does not invoke the tool again.

Events include text_delta, text_complete, tool_call_start, tool_result, approval_required, recovery_required, and run_result. Render text deltas as provisional. Handle attempt_reset by discarding the interrupted attempt's provisional text, and use processed/final output to settle the display. See Streaming for response-middleware behavior.

Delivery and saving your UI cursor are separate operations. Reconnection can repeat an event or result summary; deduplicate stored events by event ID and render the run's final state idempotently. Some synthesized terminal summaries have no cursor, which is why the helper checks before saving one.

Disconnecting an event subscriber does not cancel a durable run. Keep the worker and runtime alive for execution to continue. A browser reconnect should normally reattach to the existing run's observation stream, not ask another worker to resume an actively leased session.

Approvals and uncertain tools

Durable ASK permissions produce persisted pending requests: the run returns needs_input and result.pending lists PendingTool objects with status == "approval". Show the tool and arguments, obtain the application's decision, and record it. With resume=True the run continues and the finished RunResult is returned:

from harnessx import AgentRuntime, RunResult


async def answer_approval(
    runtime: AgentRuntime, result: RunResult, allow: bool
) -> RunResult:
    pending = result.pending[0]
    print("Approve?", pending.call.name, pending.call.input)
    if allow:
        return await runtime.approve(pending, resume=True)
    return await runtime.decline(pending, resume=True)

approve(target, *, allow=True, resume=False) and decline(target, *, resume=False) 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". Call these on the runtime that owns the session; after a process restart, first bind it with resume(session_id) and inspect the resulting pending operations. An execution key the run does not have raises UnknownExecutionKey; a decision that does not fit the tool's state raises ResolutionError. Workers do not read terminal input automatically.

An uncertain pending operation is different from an approval request: the tool might already have produced an external effect. Investigate that system, then choose exactly one resolution:

Resolution	API	Meaning
Record a verified result	`await runtime.resolve_tool(pending, result=ToolResult(pending.call.id, verified_output))`	Supply an outcome established by reconciliation
Accept another invocation	`await runtime.resolve_tool(pending, retry=True)`	Explicitly accept possible repetition
Abort	`await runtime.resolve_tool(pending, abort=True)`	Stop the run; this does not undo an external effect

resolve_tool() accepts the PendingTool or its execution key. After supplying a result or accepting a retry, resume the session to continue. See Approvals & recovery for execution keys and the manual, safe, and idempotent replay policies.

Pause, cancel, and cleanup

Controls belong to AgentRuntime, while the handle observes a particular run:

Runtime API	Behavior
`await runtime.pause()`	Stops new steps and lets active tools settle
`await runtime.cancel()`	Requests cooperative cancellation; threads or remote effects may outlive it
`await runtime.stop()` or `await runtime.aclose()`	Pauses active local work, closes the agent the runtime executes, and closes a backend the runtime created
`await runtime.status()`	Returns `session_id`, `state`, and `run`, the latest `RunResult` or `None`

Leaving the runtime context manager calls stop(). Returning a handle from inside that context does not leave an independent background worker running after the context exits. Keep the runtime open for as long as its driver is needed.

The runtime closes its agent. It closes the backend only if it created that backend; close an explicitly supplied backend yourself. The outer agent contexts also cover failures before an agent is bound to the runtime. Repeated close calls are safe. Create a fresh agent after its resources have been closed.

PostgreSQL stores state and coordinates leases; it does not restart Python processes or schedule recovery workers. Your application or supervisor restarts the worker and calls resume(). A hard crash leaves a lease that must expire before another worker claims the session. The default lease lasts 30 seconds with renewal approximately every 10 seconds.

Agent, AgentRef, and AgentRegistry

Use the direct-agent pattern for straightforward PostgreSQL applications. The application already knows how to construct its agent.

AgentRef contains a name and version, such as assistant@v1. In the current implementation it does not carry a registry or factory. AgentRegistry maps those identifiers to process-local factories that construct fresh agents.

Backend and binding	How construction works
SQLite / PostgreSQL with `Agent`	Pass the configured instance directly to `AgentRuntime`
SQLite / PostgreSQL with a named `AgentRef`	Supply `registry=` for a custom registry, or register with the default process-local registry
Current Temporal backend	Submit an `AgentRef`; configure its factory in the registry used by the Temporal worker

The current Temporal path can execute on a different process or machine. Its worker uses the identifier to construct an agent with locally available code, tools, and credentials. A Python instance in the submitting process does not supply those capabilities to the remote worker. Temporal's worker registry is configured through TemporalBackend; it need not also be passed to the submitting AgentRuntime.

This is the current HarnessX binding contract, not a universal requirement that every Temporal integration expose a public registry. AgentRegistry and AgentRef remain supported advanced APIs. PostgreSQL durability does not require them.

For direct agents, the runtime checks configuration compatibility; your application remains responsible for the appropriate tool implementations and extensions. For named definitions, retain the same compatible name/version when resuming. A version label does not prove code compatibility or serialize a factory into storage.

What survives a failure

The engine records dispatch intent before invoking a tool and a raw result before result middleware. Recovery depends on what actually committed:

Last saved state	Meaning on recovery
Prepared call, before dispatch is recorded	The runtime can dispatch the call
Dispatch recorded, no saved result	The external outcome can be unknown; a `manual` tool waits for reconciliation
Raw result committed	Reuse it and finish processing without invoking the tool again
Run completed	Read the saved result; there is no run to resume

SQL execution-boundary updates and their events commit transactionally. Every persisted phase boundary also emits HookEvent.CHECKPOINT on the agent's hooks with session_id, run_id, status, and phase. PostgreSQL fencing rejects state writes from workers that lose their leases. It cannot stop an already-running remote request.

External effects and runtime storage are not one transaction. The SDK does not provide universal rollback or exactly-once external effects. An idempotent tool requires its external adapter/service to enforce the execution key. A safe tool must actually tolerate repetition. Result middleware may run again and should tolerate replay.

Interrupted live-model requests may be reissued, return different content, and incur additional charges. Arbitrary local files are not automatically copied into PostgreSQL. Recovery on another machine requires compatible code and shared access to the workflow's resources.

Setup and troubleshooting

Question or symptom	What to do
How does resume choose a session?	Your application supplies the saved `session_id`; the CLI reads it from its workspace's `session.json`
Why did resume return `None`?	The session has no run, or its latest run is completed/cancelled; inspect status on the bound runtime
Why is status still `running` after a crash?	It is the last persisted status, not proof that a worker is alive
Why is the session busy?	A worker owns the lease, or the crashed worker's lease has not expired
PostgreSQL extra is installed but cannot be imported	Use the interpreter you installed into: `.venv/bin/python`; Homebrew/system Python can use different packages
Activation cannot find Python after moving the checkout	The activation script may reference the old path; use the explicit interpreter and repair or recreate the local environment
Password contains `@` or `/`	Put the raw password in the example config's password field and retain `{password}` in its URL; encoding is automatic
DNS, connectivity, or schema-permission failure	Check host, network access, database, credentials, TLS, and role permissions separately from package installation

Documentation and local recovery checks use controlled model responses and SQLite stand-ins where no PostgreSQL test service is configured. Service-backed PostgreSQL and Temporal qualification are separate requirements. Follow the PostgreSQL verification instructions before making deployment guarantees.