nodestep.state

State schemas, the Reducer type, history and the state store protocol.

The stores, InMemoryStateStore and FilesystemStateStore, are in nodestep.state.integrations and exported by nodestep. The reducers, such as add and add_messages, are in nodestep.core. See State and Persistence and time travel.

State schemas

BaseState

Base class for pydantic graph state schemas, with pydantic's default config.

StateT

StateT = TypeVar('StateT')

The state type a StateSchema validates and builds.

Reducer

Reducer: Callable[[Any, Any], Any]

A reducer: takes the current value and the update and returns the new value.

StateSchema

StateSchema(schema_type, fields, kind)

Fields and reducers of a state type.

Validators and reducers may run more than once per write, so keep them free of side effects. See Validation for how values are validated and stored.

Attributes

schema_type	Any	The state type: a TypedDict, a pydantic model or dict.
fields	dict[str, FieldDescriptor]	The fields by name; empty for dict.
kind	str	"typeddict", "pydantic" or "dict".

Methods

from_type(schema)

class method Build a schema from a TypedDict, a pydantic model or dict.

from_type(schema) -> StateSchema[Any]

dict means untyped state: any key, no reducers, JSON values only. A field's reducer is the one callable in its Annotated metadata, as in Annotated[list[str], add]; the rest of the metadata is kept for validation. Class variables and private attributes of a pydantic schema are not fields.

Parameters

schema

Any

A TypedDict, a pydantic BaseModel subclass, or dict.

Raises

GraphConfigError

If schema is anything else (None, Any and dict subclasses included), declares no fields or has a type that cannot be resolved; if a field has more than one reducer, a reducer that does not take exactly two positional parameters without defaults (current, update), a reducer inside Optional or Union, or add_messages on a type that is not a list of chat messages; or if a pydantic schema sets extra='allow', uses validator or serializer decorators, has a default_factory that takes the validated data, or has a field with exclude=True or exclude_if.

is_dynamic()

Whether the schema is untyped dict state, which accepts any key.

is_dynamic() -> bool

to_declared(state)

Convert a state dict to the declared type, without validating it again.

to_declared(state) -> Any

to_dict(value)

Convert a state value to a dict.

to_dict(value) -> dict[str, Any]

Parameters

value

Any

defaults()

Default values of the fields that have one.

defaults() -> dict[str, Any]

merge_input(current, value, ...)

Merge the input of a run into a thread's state through the reducers.

merge_input(current, value, *, writer='the input') -> tuple[dict[str, Any], dict[str, Any]]

The input is validated and merged like a node's update, on new and existing threads alike; a new thread starts from the schema defaults. An instance of the pydantic schema is the full state: it replaces every field, bypassing the reducers, and its values are validated again only if the schema sets revalidate_instances. Partial input is a dict.

Parameters

current	dict[str, Any] | None	The thread's state, or None for a new thread.
value	Any	The input; None writes nothing.
writer	str	Who wrote the input, as named in error messages.

Returns

tuple[dict, dict]

The new state, and the validated update that was merged.

Raises

StateUpdateError	If the input is neither a dict nor an instance of the pydantic schema, has an unknown key, an invalid value, a value its reducer rejects or a secret, or leaves a required field of a new thread without a value.
StateStoreError	If an input value, or a default a new thread keeps, does not survive being stored as JSON through its field's type.

validate_update(current, update, ...)

Validate an update against the state it will be merged into.

validate_update(current, update, *, writer) -> dict[str, Any] | None

A value of a replace field, or in Replace, is validated as the field's value. A value for another reducer is validated with update_annotation and merged into current to check that it applies; the merged value is validated too when the field has its own constraints or validators, or a custom reducer. Chat messages without an id get one, and the result holds copies of the values.

Parameters

current	dict[str, Any]	The state the update will be merged into.
update	dict[str, Any] | None
writer	str	Who wrote the update, for error messages, e.g. "node 'plan'".

Returns

dict or None

The validated, copied update, ready to merge and store.

Raises

StateUpdateError	If a key is not a state field; a value is invalid, rejected by its reducer, holds a secret or cannot be copied; a value is a dict whose only key is "__nodestep_replace__"; or a RemoveMessage is inside Replace or sent to a field whose reducer is not add_messages.
StateStoreError	If a value does not survive being stored as JSON through its type.

validate_payload(payload, ...)

Validate the values of a Send payload as field values.

validate_payload(payload, *, writer) -> dict[str, Any]

A payload bypasses the reducers, so each value is validated as its field's value. The result holds copies of the values.

Parameters

payload	dict[str, Any]	Values keyed by state field.
writer	str	Who sent the payload, for error messages.

Raises

StateUpdateError	If a value is not valid for its field, holds a secret or cannot be copied.
StateStoreError	If a value does not survive being stored as JSON through its type.

dump_state(values)

Convert a state, or part of one such as a Send payload, to JSON data.

dump_state(values) -> dict[str, Any]

restore_state and restore_payload read the result back.

Parameters

values

Mapping[str, Any]

Validated values keyed by state field.

dump_update(update)

Convert a validated update to JSON data, as it is stored in history.

dump_update(update) -> dict[str, Any] | None

Replace(value) becomes {"__nodestep_replace__": value} and RemoveMessage(id) becomes {"__nodestep_remove__": id}. restore_update reads the result back.

Parameters

update

Mapping[str, Any] | None

An update from validate_update.

restore_state(stored)

Read a stored state, such as a checkpoint, back into the field types.

restore_state(stored) -> dict[str, Any]

Fields missing from the stored state, such as fields added to the schema later, get their defaults; no other default factory is called.

Parameters

stored

Mapping[str, Any]

The state as it was stored.

Raises

StateStoreError	If the stored state has a key that is not a state field.
ResumeError	If a stored value is not valid for its field.

restore_payload(stored)

Read a stored Send payload back into the field types.

restore_payload(stored) -> dict[str, Any]

Parameters

stored

Mapping[str, Any]

The payload as it was stored.

Raises

StateStoreError	If the payload has a key that is not a state field.
ResumeError	If a stored value is not valid for its field.

restore_update(update)

Read an update from a stored event back as a validated update.

restore_update(update) -> dict[str, Any]

Reverses dump_update, turning the stored forms back into Replace and RemoveMessage. The result is ready for merge_updates.

Parameters

update

Mapping[str, Any]

The update as it was stored.

Raises

StateStoreError	If the update has a key that is not a state field.
ResumeError	If a stored value is not valid for its field.

FieldDescriptor

FieldDescriptor(name, annotation, reducer=replace, update_annotation=None, config=None)

A state field: its validation types and its reducer.

Parameters

name	str	Field name.
annotation	Any	The field's type with its constraints and validators, without the reducer; every value of the field is validated with it.
reducer	Reducer	Merges an update into the field; replace by default.
update_annotation	Any	Type an update is validated with before a reducer other than replace merges it: annotation without the field's own constraints and validators, which apply to the merged value. None means annotation.
config	ConfigDict | None	Pydantic config of the state schema, applied to both types.

Attributes

adapter	TypeAdapter[Any]	TypeAdapter for annotation, built on first use.
update_adapter	TypeAdapter[Any]	TypeAdapter for update_annotation, built on first use.

Merging updates

merge_updates

merge_updates(schema, current, update) -> tuple[dict[str, Any], set[str]]

Apply an update to a state through the field reducers.

The update comes from StateSchema.validate_update or StateSchema.restore_update. Values of replace fields and Replace(value) are set as they are; for another reducer the merged value is validated when the field has its own constraints or validators, or a custom reducer.

Parameters

schema	StateSchema[Any]
current	dict[str, Any]
update	dict[str, Any] | None

Returns

tuple[dict, set[str]]

The new state and the written field names.

Raises

StateUpdateError

If the update contains an unknown field, a reducer rejects a value, or a merged value is not valid for its field.

merge_parallel_updates

merge_parallel_updates(schema, current, updates) -> tuple[dict[str, Any], set[str]]

Apply the updates of one superstep in node-name order.

Each update is merged as in merge_updates.

Parameters

schema	StateSchema[Any]
current	dict[str, Any]
updates	list[tuple[str, dict[str, Any] | None]]	Updates keyed by node name.

Raises

InvalidUpdateError	If different nodes write the same field and one of the writes has no merging reducer or is a Replace.
StateUpdateError	If a reducer rejects a value or a merged value is not valid for its field; the error names every node that wrote the field.

snapshot_state

snapshot_state(state) -> dict[str, Any]

Return a deep copy of a state dict.

Thread history

StateStore

Protocol for storing thread history.

Methods

async append_event(event)

Store an event as given.

async append_event(event) -> HistoryEvent

Parameters

event

HistoryEvent

async append_next_event(event)

Store an event with the next sequence number of its thread.

async append_next_event(event) -> HistoryEvent

Parameters

event

HistoryEvent

Returns

HistoryEvent

The stored event with its sequence number.

async save_checkpoint(checkpoint)

Store a state snapshot.

async save_checkpoint(checkpoint) -> CheckpointRecord

Parameters

checkpoint

CheckpointRecord

async get_history(thread_id, branch_id)

Return the events and checkpoints of a thread branch.

async get_history(thread_id, branch_id='main') -> History

Parameters

thread_id	str
branch_id	str

async latest_checkpoint(thread_id, branch_id, at)

Return the latest checkpoint, optionally at or before an event.

async latest_checkpoint(thread_id, branch_id='main', at=None) -> CheckpointRecord | None

Parameters

thread_id	str
branch_id	str
at	str | None	Event id.

async fork(thread_id, ...)

Record a new branch starting at an event.

async fork(thread_id, *, from_, branch_id='main', name=None) -> BranchRecord

Parameters

thread_id	str
from_	str	Event id to branch from.
branch_id	str	Branch to fork.
name	str | None	Id of the new branch; a random id when omitted.

Raises

StateStoreError

If the thread already has a branch with that id, including "main".

async list_branches(thread_id)

Return the branches forked from a thread, oldest first.

async list_branches(thread_id) -> list[BranchRecord]

Parameters

thread_id

str

Returns

list[BranchRecord]

Every recorded fork of the thread; "main" is not listed.

History

History(*, thread_id, branch_id, events=[], checkpoints=[])

Events and checkpoints of one thread branch.

Attributes

thread_id	str
branch_id	str
events	list[HistoryEvent]
checkpoints	list[CheckpointRecord]

HistoryEvent

HistoryEvent(*, id=str(uuid4()), thread_id, branch_id='main', sequence, type, node=None, parent_id=None, data_json=None, error_type=None, message=None, created_at=time.time())

One recorded event of a thread.

Attributes

id	str
thread_id	str
branch_id	str
sequence	int	Position on its branch, counted from 1; append_next_event sets it.
type	str	What happened, such as "run_started", "superstep_completed", "interrupted" or "error".
node	str | None	Node the event belongs to.
parent_id	str | None	Not set by nodestep.
data_json	str | None	Details of the event as JSON, such as the step and the updates.
error_type	str | None	Exception class name, on "error" and "superstep_failed" events.
message	str | None	The exception message, with error_type.
created_at	float	Unix time in seconds.

CheckpointRecord

CheckpointRecord(*, id=str(uuid4()), thread_id, branch_id='main', event_id, sequence, state_json, active_json='[]', status='running', created_at=time.time())

Full state snapshot at a point in a thread's history.

Attributes

id	str
thread_id	str
branch_id	str
event_id	str	Id of the event the snapshot was taken at.
sequence	int	sequence of that event.
state_json	str	The state as JSON, written by StateSchema.dump_state.
active_json	str	JSON list of the tasks still to run at that point.
status	str	"running", "completed", or "forked" for the first snapshot of a branch.
created_at	float	Unix time in seconds.

BranchRecord

BranchRecord(*, id=str(uuid4()), thread_id, parent_branch_id='main', from_event_id, created_at=time.time())

A history branch forked from an event.

Attributes

id	str	Id of the branch, the name given to fork or a random id.
thread_id	str
parent_branch_id	str	Branch it was forked from.
from_event_id	str	Event of that branch it starts at.
created_at	float	Unix time in seconds.

StateSnapshot

StateSnapshot(*, value, sequence)

A thread's state (value) as of the history event numbered sequence.

Attributes

value	StateT
sequence	int

Models and JSON

NodestepModel

Base pydantic model of nodestep's own models; unknown fields raise.

to_json_value

to_json_value(value, *, fallback=None) -> Any

Convert a value to JSON-compatible data, encoded by its runtime type.

Models are dumped as by model_dump(mode="json", by_alias=True), secrets are masked, and tuples and sets become lists. State is stored through the field types instead (StateSchema.dump_state).

Parameters

value	Any
fallback	Callable[[Any], Any] | None	Called with each value the encoder does not know; its result is encoded instead.

Raises

PydanticSerializationError

If a value cannot be encoded and no fallback is given.

dump_json_object

dump_json_object(value, *, sort_keys=True) -> str | None

Serialize a dict to JSON.

Parameters

value	dict[str, Any] | None
sort_keys	bool	Sort the keys; True by default.

load_json_object

load_json_object(value) -> dict[str, Any] | None

Parse a JSON object.

Parameters

value

str | None

Raises

ValueError

If the JSON is not an object.