Resume or fork an agent run

Save every step of an Agentor run to disk so a new process can pick the job up by run id, finish it after a crash, or branch it into a new run with fork().

Long jobs get interrupted. A machine reboots, a deploy rolls, a container is evicted — and your agent was halfway through, having already paid for the tokens and called half its tools.

Give the agent a store — somewhere to write down what it did — and each step is saved as it happens. A different process, knowing nothing but the run’s id, can pick the job up and finish it, or branch it into an independent run.

Durable runs, and forking a run into a new one, are part of Agentor’s 0.1.0 engine — the current stable release. Plain pip install agentor gets you both.

Save a run

Pass a store when you build the agent. FileStore keeps one file per run in the directory you name:

agent.py

from agentor import Agentor, function_tool
from agentor.engine.store import FileStore


@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a city.

    Args:
        city: The city to look up.
    """
    return f"The weather in {city} is sunny and 22C."


agent = Agentor(
    name="Weather Agent",
    model="gpt-4o-mini",
    instructions="You are a helpful weather assistant.",
    tools=[get_weather],
    store=FileStore("runs"),
)

result = agent.run("What is the weather in London?")

print(result.run_id)        # cabc9046ac5040298aa362c0d6d2c5c5
print(result.final_output)  # The weather in London is sunny with a temperature of 22°C.

Every run now gets an id, and runs/<run_id>.jsonl holds the whole story. Keep the id somewhere you can find it again — a database row, a queue message, a log line.

Without store=, result.run_id is None and nothing is written. Runs work exactly as before; they are just not recoverable.

Resume after a crash

resume() takes a run id and continues from the last thing that was saved:

resume.py

agent.resume("cabc9046ac5040298aa362c0d6d2c5c5")

The agent object has to be built the same way — same tools, same instructions — but it does not have to be the same process. This is the whole point: the process that started the run can be gone.

Two things make resume() safe to call whenever you are unsure:

Finished runs are returned, not re-run

If the run already completed, you get its result back without another model call and without any tool running twice.

Half-finished tool calls are rewound

If the process died between asking for tools and recording every result, the agent goes back to the request that produced them and decides again.

Use await agent.aresume(run_id) from async code.

A worked example

Start a run and kill the process partway through — after the weather tool answered but before the agent wrote its reply:

$ python agent.py
RUN_ID cabc9046ac5040298aa362c0d6d2c5c5
event: run_start
event: generation
event: tool_call get_weather
event: tool_result get_weather
Killed

Four lines made it to disk. Now a separate interpreter, given only that id:

$ python resume.py cabc9046ac5040298aa362c0d6d2c5c5
status: completed
final_output: The weather in London is sunny with a temperature of 22°C.

The tool did not run a second time. The saved result was enough.

Fork a run

resume() continues a run in place, and deliberately refuses to re-run one that already finished. fork() takes a different approach: it copies a run’s whole history — every step, including the model’s reasoning where the provider returned it — into a new run id, then continues from there. The original run is never touched, whether it is still in progress, finished, or crashed.

That makes fork the way to branch a conversation: try a different instruction from the same starting point without losing the original, or pick up a completed run and keep going, which resume() will not do for you.

fork.py

from agentor import Agentor
from agentor.engine.store import FileStore

agent = Agentor(name="Assistant", model="gpt-5-mini", store=FileStore("runs"))

result = agent.run("Draft a launch plan")

# Branch off with a new instruction — result.run_id is completely untouched.
variant = agent.fork(result.run_id, "Make it punchier")
print(variant.run_id)         # a new id, different from result.run_id
print(variant.final_output)

Use await agent.afork(...) from async code.

As with resume(), the agent doing the forking does not have to be the one that made the original run — but if the original run has no system message, the forking agent’s instructions fill in on the fork.

Fork without new input

Leave out the second argument and fork() just copies the run under a new id:

• If the parent was still unfinished — interrupted mid-run — the fork picks up where it left off and finishes, exactly like resume() would.
• If the parent had already finished, the fork is an identical copy of it. No model call is made.

snapshot = agent.fork(result.run_id)   # a full copy of result, under a new run_id

Choosing the new run’s id

Pass fork_id to name the new run yourself instead of getting a generated one:

agent.fork(result.run_id, "Make it punchier", fork_id="launch-plan-punchy")

fork() raises ValueError if that id is already taken.

max_turns and tracing behave the same as on run() — they apply only to the fork’s continuation, never to the run being copied.

What gets written

The store records the same events the agent emits while it runs, one JSON object per line:

runs/cabc9046ac5040298aa362c0d6d2c5c5.jsonl

{"type": "run_start", "agent": "Weather Agent", "model": "gpt-4o-mini", "messages": [...]}
{"type": "generation", "turn": 1, "usage": {...}, "calls": [...]}
{"type": "tool_call", "name": "get_weather", "call_id": "call_8dGMCmO...", "turn": 1}
{"type": "tool_result", "name": "get_weather", "result": "The weather in London is sunny and 22C.", "turn": 1}
{"type": "run_end", "status": "completed", "text": "...", "usage": {...}}

Each generation records the exact messages that were sent to the model — reasoning included, when the provider returns any — so rebuilding the conversation is a replay rather than a guess. FileStore flushes and fsyncs after every line, because a crash is the case it exists for.

Forking adds one more line to the copy: a fork event naming the run it came from. It is provenance only — replaying the log or checking whether it is complete works exactly as it does for a run that was never forked.

You can read the same events back off any finished result:

for event in result.events:
    if event.type == "tool_result":
        print(event.name, "->", event.result)

Choose a store

FileStore

One append-only file per run, on local disk. The right default for a worker, a cron job, or anything with a volume attached.

from agentor.engine.store import FileStore

store = FileStore("runs")        # directory is created if missing
store.list_runs()                # ['cabc9046ac50...', ...]
store.load("cabc9046ac50...")    # list[Event]

MemoryStore

Keeps runs in a dictionary for the life of the process. Useful in tests, and when you want result.run_id and event history without touching the filesystem.

from agentor.engine.store import MemoryStore

agent = Agentor(name="Agent", store=MemoryStore())

A MemoryStore cannot survive the crash it would be recovering from. Use FileStore if you actually need resume.

Your own

A store is any object with these three methods, so a Postgres or S3-backed store is a small class:

from agentor.engine.events import Event


class MyStore:
    def append(self, run_id: str, event: Event) -> None:
        """Save one event. Called once per step."""

    def load(self, run_id: str) -> list[Event]:
        """Return every event for a run, in order."""
        return []

    def list_runs(self) -> list[str]:
        """Return every known run id."""
        return []

Use Event.to_json() to serialise and Event.from_dict() to read back. append is called on a worker thread, so blocking IO inside it will not stall the event loop.

Add a fourth method, fork(src_run_id, dst_run_id), to copy a run in one bulk write instead of the slower event-by-event copy fork() uses when it is missing. Raise ValueError if dst_run_id already exists. FileStore and MemoryStore both implement it.

Streamed runs are saved too

stream_chat() persists a run whenever the agent has a store, so a stream that a client abandons is still recoverable:

agent = Agentor(name="Agent", tools=[get_weather], store=FileStore("runs"))

async for event in agent.stream_chat("What is the weather in Tokyo?"):
    print(event)

Limits to know about

Two workers must not resume the same unfinished run

The bundled stores are single-process and take no lock. If two workers both see a run as incomplete, both will continue it, and any tool with a side effect runs twice. Coordinate outside Agentor — a queue with visibility timeouts, a row lock, a lease — if more than one worker can recover the same run.

Forking a run still in progress copies whatever has been saved so far

Like resume, fork takes no lock. Fork a run that is actively being written to and you get a copy of whatever prefix of its log had reached the store at that instant — not necessarily its very latest step.

Pausing for human approval is not built in

There is no way to stop a run, ask a person to approve a tool call, and continue. The event log makes it possible to build, but Agentor does not ship it yet.

Saving is best-effort during a live run

If a write to the store fails, the error is logged and the run carries on rather than dying. You get your answer; you may not be able to resume that run.

Nothing prunes old runs

FileStore writes files and never deletes them. Rotate or expire the directory yourself.

Next steps

Tracing

The same events power Celesto traces. Watch a run instead of reading its log.

Structured output

Get a typed object back from a run instead of a block of text.