Store

Guide: Save and resume.

Store

Bases: ABC

Keeps States outside the process. Used by Agent(store=...), which saves as the run goes.

A subclass implements write and read (and list, delete if it can), or their async versions, or both. The Agent calls write from one thread at a time for a given State, outside the State lock. FileStore is the implementation that ships with alpineagents.

A State is saved in one store only. The Agent compares stores with ==, so define __eq__ if two objects can stand for the same storage.

Raises:

Type	Description
TypeError	When creating an instance of a subclass that implements neither write nor awrite, or neither read nor aread.

Example

class RedisStore(Store):
    async def awrite(self, state_id, entries, snapshot, *, create=False): ...
    async def aread(self, state_id): ...

write

write(state_id: str, entries: Sequence[dict[str, Any]], snapshot: dict[str, Any] | None, *, create: bool = False) -> None

Adds history entries and replaces the snapshot of one State.

Parameters:

Name	Type	Description	Default
state_id	str	The State id.	required
entries	Sequence[dict[str, Any]]	New history entries in seq order. Skip those whose seq you already have. They continue right after what you have (no gaps).	required
snapshot	dict[str, Any] | None	The new snapshot, or None to keep the current one. Write it after the entries.	required
create	bool	The first write of a new State. If state_id already exists, write nothing and raise ValueError. Either everything is written or nothing is.	False

Raises:

Type	Description
ValueError	create is true and the id exists.
LookupError	create is false and the id does not exist (deleted while in use).

awrite async

awrite(state_id: str, entries: Sequence[dict[str, Any]], snapshot: dict[str, Any] | None, *, create: bool = False) -> None

The async version of write. By default it runs write on a worker thread.

read

read(state_id: str) -> Record | None

Everything saved for one State, or None if the id does not exist.

aread async

aread(state_id: str) -> Record | None

The async version of read. By default it runs read on a worker thread.

delete

delete(state_id: str) -> None

Deletes one State. Does nothing if the id does not exist.

adelete async

adelete(state_id: str) -> None

The async version of delete. By default it runs delete on a worker thread.

list

list() -> list[SavedState]

Every saved State, most recently updated first.

alist async

alist() -> list[SavedState]

The async version of list. By default it runs list on a worker thread.

load

load(state_id: str) -> State

Rebuilds a saved State, ready to continue with agent.run(state).

Tool calls whose results were never saved (the process stopped while they ran) get a result telling the model that they may or may not have run. The first run warns with ResumeWarning if its Agent differs from the one that saved the State.

Raises:

Type	Description
LookupError	No State with this id.
ValueError	The saved State is damaged, or was saved by a newer alpineagents.

Example

state = store.load("bug-hunt-1")
agent.run(state)

aload async

aload(state_id: str) -> State

The async version of load.

FileStore

FileStore(path: str | PathLike[str])

Bases: Store

Saves each State in a folder of its own: {path}/{id}/log.jsonl (history, one entry per line) and snapshot.json.

Files are readable only by their owner, and each write is flushed to disk (fsync) before it returns. One process writes a State at a time: two processes running the same State id are not supported.

Parameters:

Name	Type	Description	Default
path	str | PathLike[str]	The folder to keep States in. Created on the first write.	required

Example

store = FileStore(".agent-runs")
agent = Agent(model="claude-sonnet-5", store=store)
agent.run(State("Find the bug", id="bug-hunt-1"))

SavedState dataclass

SavedState(id: str, task: str, created_at: datetime, updated_at: datetime, turn: int, stopped_by: str | None, finished: bool)

One saved State, as Store.list shows it.

id instance-attribute

id: str

The State id. Continue it with store.load(id).

task instance-attribute

task: str

The task the State was created with.

created_at instance-attribute

created_at: datetime

When the State was created (UTC).

updated_at instance-attribute

updated_at: datetime

When the last saved step was recorded (UTC).

turn instance-attribute

turn: int

Turns so far.

stopped_by instance-attribute

stopped_by: str | None

Why its last run stopped ("finish", "limit", an until name), or None if it ended with an exception or was still running.

finished instance-attribute

finished: bool

Whether finish() was called. A finished State cannot run again.

from_snapshot classmethod

from_snapshot(state_id: str, snapshot: Mapping[str, Any]) -> SavedState

Builds one from a snapshot dict, for Store implementations.

Record dataclass

Record(entries: list[dict[str, Any]], snapshot: dict[str, Any] | None)

What a store has for one State id (Store.read).

entries instance-attribute

entries: list[dict[str, Any]]

History entries in seq order, starting at 0.

snapshot instance-attribute

snapshot: dict[str, Any] | None

The latest snapshot, or None if none was written yet.