AgentVisorCheckpointer

A LangGraph-compatible checkpointer that persists state to Temporal workflows.

Usage

from agentvisor.langgraph import AgentVisorCheckpointer
from langgraph.graph import StateGraph

# Build your graph
builder = StateGraph(MyState)
builder.add_node("process", process_node)
builder.set_entry_point("process")

# Compile with checkpointer
checkpointer = AgentVisorCheckpointer()
graph = builder.compile(checkpointer=checkpointer)

# Use normally - checkpointing happens automatically
result = graph.invoke({"messages": [...]})

How It Works

The checkpointer implements LangGraph's BaseCheckpointSaver interface:

1. After each super-step, LangGraph calls put() with the current state
2. The checkpointer sends state to the host
3. Host stores state in the Temporal workflow
4. On resume, get_tuple() retrieves the checkpoint

Thread ID

The checkpointer uses AGENTVISOR_THREAD_ID environment variable, which is set automatically by the runtime.

import os

# Thread ID is automatically available
thread_id = os.environ.get("AGENTVISOR_THREAD_ID")

get_tuple(), list(), put(), and put_writes() all accept a config whose configurable.thread_id overrides the environment default — but since the host binds thread_id authoritatively per connection, a value that disagrees with AGENTVISOR_THREAD_ID can never reach another thread's data. Rather than silently serving this thread's data under the requested label, all four raise ValueError on a mismatch, the same way delete_thread() does.

Configuration

The checkpointer has no required configuration. It connects to the host socket automatically.

# Default configuration - works out of the box
checkpointer = AgentVisorCheckpointer()

Interface Methods

put()

Save a checkpoint:

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    ...

Called automatically by LangGraph after each super-step.

get_tuple()

Retrieve a checkpoint:

def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    ...

Called when resuming a graph.

list()

List checkpoints for a thread:

def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    ...

Returns checkpoints for the thread, newest first, bounded by the host's temporal.max_checkpoint_history setting (default 1 — raise it to retain more than the latest checkpoint). before excludes checkpoints at or after the given configuration's checkpoint_id. filter is applied client-side as an exact-match on each key/value against checkpoint metadata. When both filter and limit are given, the host is paged (using each page's oldest checkpoint as the next cursor) until limit matching checkpoints have been returned or the host has none left — limit bounds matches, not raw host-side pages, matching BaseCheckpointSaver.list()'s contract.

put_writes()

Persists intermediate (mid-superstep) writes — the output of each completed task within a superstep, before the superstep as a whole finishes:

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
    task_path: str = "",
) -> None:
    ...

Called once per completed task. This is what lets LangGraph skip re-executing an already-completed task after a crash mid-superstep (e.g. AgentVisor's Continue-As-New interrupting an active run): on resume, LangGraph's own _reapply_writes_to_succeeded_nodes restores each task's persisted write instead of re-running it — including __interrupt__/__resume__ writes for a task that errors and later resumes. Without this, every task in the interrupted superstep would replay from scratch, which is unsafe for non-idempotent side effects (e.g. a tool call that isn't safe to repeat).

Each write's position (idx) follows LangGraph's own upsert rule: regular channels are first-write-wins (safe under SDK retry), while the special __error__/__scheduled__/__interrupt__/__resume__ channels are last-write-wins. Pending writes are always scoped to the current head checkpoint — once a new checkpoint is stored, prior pending writes are no longer reachable.

delete_thread()

Clears this thread's own checkpoint store — all checkpoints and pending writes for the calling run — along with its Values/Messages read-model, so GET /threads/{thread_id} stops serving pre-delete data once graph.get_state() reports empty. Run history (Runs, e.g. listRuns) is untouched. It does not delete the thread itself: a thread is a first-class, policy-adjudicated Temporal workflow with its own lifecycle, deleted only through the authenticated AgentVisor thread API (DELETE /threads/{thread_id}), which performs a policy check before terminating the workflow.

thread_id must match this process's own AGENTVISOR_THREAD_ID — passing any other thread's ID raises ValueError rather than silently doing nothing or reaching into another thread's state:

def delete_thread(self, thread_id: str) -> None:
    ...

import os

# Clears this run's own checkpoints
checkpointer.delete_thread(os.environ["AGENTVISOR_THREAD_ID"])

# Raises ValueError - cross-thread deletion is not supported here
checkpointer.delete_thread("some-other-threads-id")

Async Variants

aget_tuple(), alist(), aput(), aput_writes(), and adelete_thread() are also implemented (each delegates to its sync counterpart via a thread pool), so the checkpointer works with async LangGraph execution.

Direct Checkpoint API

For custom checkpointing outside LangGraph:

from agentvisor import checkpoint

# Save state
checkpoint_id = checkpoint.save("my-key", {"counter": 42})

# Load state by checkpoint ID
data = checkpoint.load(checkpoint_id)

Checkpoint Data

Checkpoints include:

Field	Description
id	Unique checkpoint ID
ts	Timestamp
channel_values	Current state values
channel_versions	Version tracking for channels
versions_seen	Node version tracking
pending_sends	Pending channel writes

Resume Detection

The runtime detects when a thread has an interrupted checkpoint and automatically resumes:

# Run 1: hits interrupt
result = graph.invoke({"messages": [{"role": "user", "content": "hello"}]})
# Returns with status=interrupted

# Run 2: automatically resumes from checkpoint
result = graph.invoke({"messages": [{"role": "user", "content": "continue"}]})

Limitations

• Checkpoint state uses LangGraph's own serde.dumps_typed()/loads_typed() and is not restricted to JSON — LangChain message objects (AIMessage, ToolMessage, etc.) round-trip correctly. Checkpoint metadata is sent as a protobuf Struct and must be JSON-serializable.
• Protobuf Struct's NumberValue is a double regardless of source type, so every metadata number round-trips as a Python float — an int like 2 comes back as 2.0. The SDK recovers LangGraph's own step counter back to int (LangGraph requires it as one), but no other metadata key is corrected: a user-supplied whole-valued float (e.g. {"score": 2.0}) round-trips as a float, and a user-supplied int (e.g. {"count": 2}) round-trips as a float (2.0), not the original int.
• Each checkpoint write is a single Temporal payload, so it is bound by Temporal's own per-payload ("blob size") limit — 2 MiB by default, both on Temporal Cloud (fixed) and self-hosted Temporal (per-namespace configurable). This is independent of — and tighter than — AgentVisor's own AGENTVISOR_GUEST_CHECKPOINT_MAX_SIZE guard (default 50MB): a checkpoint under that guard can still exceed Temporal's real ceiling and fail the underlying Temporal call. The same ceiling binds get_tuple()/list() reads, not just writes — but with External Checkpoint Storage enabled, a checkpoint or read above threshold moves to the configured backend on both the write and the read side, so this ceiling stops being a hard cap. See Checkpoint Sizing and Limits for the full picture, including how it also governs pending writes and Continue-As-New.
• Older checkpoints are pruned based on configuration (temporal.max_checkpoint_history, default 1); pending writes not attached to the current head checkpoint are dropped whenever a new checkpoint is stored.
• Checkpoints are namespace-partitioned: each LangGraph checkpoint_ns (the empty string for the root graph, a derived value per subgraph invocation) resolves its own independent head checkpoint, so a subgraph's get_tuple()/ put()/put_writes() calls never read or overwrite its parent's (or a sibling subgraph's) checkpoint state. A namespace is reclaimed automatically once the root checkpoint that was current when it was first touched ages out of root's own retention window, so namespace count is bounded rather than growing for the life of the thread; the Continue-As-New payload budget remains the backstop for whatever reaping hasn't (yet) reclaimed. See Subgraphs and Namespace Growth for the full growth characteristics and what happens once that budget is exceeded.