Persistence

BeamWeaver persists graph state through explicit checkpoint adapters. When a graph or agent is compiled with a checkpointer, BeamWeaver saves checkpoint records as the run advances through graph steps. Those records make human-in-the-loop flows, conversational memory, state inspection, time travel, forking, and fault recovery possible.

Persistence is split into two contracts:

• Checkpointers save graph execution state for one thread.

• Memory stores save namespaced application data across threads.

Why Use Persistence

Persistence is required or useful for:

Use a memory store when data should outlive one thread: user preferences, account facts, extracted profile data, durable feature flags, or cross-session application state.

Threads

A thread is the stable identifier for one checkpointed execution history. BeamWeaver reads it from the LangGraph-compatible configurable config:

config = %{"configurable" => %{"thread_id" => "support-thread-1"}}

Use a new thread_id for a separate conversation or workflow run. Reuse the same thread_id when the next invocation should see the same checkpointed state.

context: is separate. It carries per-run values for tools and middleware, such as user_id, request metadata, or permissions. Do not use context: as a persistence key.

Checkpoints

A checkpoint is a saved snapshot of graph execution. BeamWeaver checkpoints at execution boundaries and stores enough metadata to restore the thread, inspect history, and resume interrupted or failed runs.

This graph produces checkpointed state for a simple two-step flow:

alias BeamWeaver.Checkpoint.ETS, as: CheckpointETS
alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

checkpointer = CheckpointETS.new()

graph =
  Graph.new(name: "CheckpointExample")
  |> Graph.add_reducer(:bar, fn existing, update -> existing ++ List.wrap(update) end)
  |> Graph.add_node(:node_a, fn _state -> %{foo: "a", bar: ["a"]} end)
  |> Graph.add_node(:node_b, fn _state -> %{foo: "b", bar: ["b"]} end)
  |> Graph.add_edge(:node_a, :node_b)
  |> Graph.add_edge(Graph.start(), :node_a)
  |> Graph.add_edge(:node_b, Graph.end_node())
  |> Graph.compile!(checkpointer: checkpointer)

config = %{"configurable" => %{"thread_id" => "thread-1"}}

{:ok, state} =
  Compiled.invoke(graph, %{foo: "", bar: []}, config: config)

state.bar

The :bar reducer accumulates values across nodes. Without a reducer, later writes overwrite earlier values for the same key.

Snapshot Fields

Compiled.get_state/2 returns the latest checkpointed snapshot for a thread:

{:ok, snapshot} = Compiled.get_state(graph, config)

Common snapshot fields include:

State History

State history is ordered newest first:

history =
  Compiled.get_state_history(graph, config, limit: 20)

latest = List.first(history)

Use history to find checkpoints by node, step, interrupt, or metadata:

before_node_b =
  Enum.find(history, fn snapshot ->
    snapshot.next == ["node_b"]
  end)

manual_updates =
  Enum.filter(history, fn snapshot ->
    snapshot.metadata["source"] == "update"
  end)

interrupted =
  Enum.find(history, fn snapshot ->
    snapshot.interrupts != []
  end)

The lower-level checkpoint facade can return typed records for adapter and exporter code:

records =
  BeamWeaver.Checkpoint.list_records(checkpointer, config, limit: 20)

Update State

Compiled.update_state/4 writes a new checkpoint. It does not mutate an older checkpoint in place:

{:ok, updated_config} =
  Compiled.update_state(graph, config, %{foo: "manual"})

{:ok, snapshot} =
  Compiled.get_state(graph, updated_config)

Updates go through the same channel merge logic as node outputs. If a key has a reducer, the update is accumulated; otherwise it overwrites the current value.

When BeamWeaver can infer the node that should be treated as the source of the update, it does so. If multiple continuation nodes are possible, pass :as_node:

Compiled.update_state(graph, before_node_b.config, %{bar: ["fork"]},
  as_node: :node_a
)

Compiled.bulk_update_state/4 creates a sequence of checkpoint updates with super-step merge semantics:

{:ok, final_config} =
  Compiled.bulk_update_state(graph, config, [
    [%{bar: ["manual-a"]}, %{bar: ["manual-b"]}],
    [%{foo: "after-bulk"}]
  ])

Replay And Fork

A snapshot's config contains the checkpoint_id needed to revisit that point in a thread.

Use a prior snapshot config to inspect that checkpoint:

{:ok, old_snapshot} =
  Compiled.get_state(graph, before_node_b.config)

Use update_state/4 against a prior checkpoint to fork from there:

{:ok, fork_config} =
  Compiled.update_state(graph, before_node_b.config, %{foo: "forked"},
    as_node: :node_a
  )

Then invoke or inspect from the returned config. Nodes after the fork point can run again and may call models or external APIs again. Keep side effects idempotent or guard them with human review and application-level idempotency keys.

Pending Writes

When a super-step fans out to multiple nodes and one node fails, BeamWeaver can persist successful sibling writes as pending writes. On resume, successful sibling work does not need to run again.

For example, if :ok succeeds and :fail fails in the same fan-out step:

{:error, _error} = Compiled.invoke(graph, %{}, config: config)
{:ok, snapshot} = Compiled.get_state(graph, config)

snapshot.values
snapshot.pending_writes
snapshot.next

The latest state applies pending writes so the application sees recoverable state. Fetching the raw checkpoint by snapshot.config shows the checkpoint before pending writes were applied.

Interrupts

Human-in-the-loop flows require a checkpointer because a paused run must survive between the interrupt and the resume call:

checkpointer = BeamWeaver.Checkpoint.ETS.new()
config = %{"configurable" => %{"thread_id" => "review-thread-1"}}

graph =
  workflow
  |> BeamWeaver.Graph.compile!(checkpointer: checkpointer)

{:interrupted, interrupt} =
  Compiled.invoke(graph, %{draft: "Approve this?"}, config: config)

{:ok, final_state} =
  Compiled.resume(graph, %{interrupt.id => "approved"}, config: config)

See Human-In-The-Loop for the agent middleware and graph interrupt patterns.

Memory Store

Checkpointers persist execution state for one thread. Stores persist application memory across threads:

alias BeamWeaver.Memory

store = Memory.ETS.new()
namespace = ["users", "user-123", "memories"]

{:ok, _item} =
  Memory.put(store, namespace, "food", %{
    "memory" => "User likes pizza"
  })

items =
  Memory.search(store, ["users", "user-123"],
    query: "food preferences",
    limit: 3
  )

Compile a graph with both:

graph =
  workflow
  |> Graph.compile!(
    checkpointer: BeamWeaver.Checkpoint.ETS.new(),
    store: Memory.ETS.new()
  )

Agents accept the same concepts through module DSL or BeamWeaver.Agent.build/1:

defmodule MyApp.MemoryAgent do
  use BeamWeaver.Agent

  model BeamWeaver.Models.init_chat_model!("openai:gpt-5.4")
  checkpointer BeamWeaver.Checkpoint.ETS.new()
  store BeamWeaver.Memory.ETS.new()
end

Tools can access the store through injected :store arguments or runtime.store. See Long-Term Memory for full examples.

Checkpointer Adapters

BeamWeaver ships two checkpoint adapters:

Database setup is explicit:

defmodule MyApp.Repo.Migrations.CreateBeamWeaverCheckpoints do
  use Ecto.Migration

  def up do
    BeamWeaver.Migrations.up(adapters: [:checkpoint])
  end

  def down do
    BeamWeaver.Migrations.down(adapters: [:checkpoint], version: 1)
  end
end

BeamWeaver.Migrations selects PostgreSQL or SQLite from the application Repo. The checkpoint schema and BeamWeaver.Checkpoint.Ecto API are the same for both databases. Only migration DDL and transaction serialization are adapter-specific. SQLite applications add the optional adapter themselves:

{:ecto_sqlite3, "~> 0.24"}

BeamWeaver declares ecto_sqlite3 as optional. Applications using PostgreSQL are not forced to install it; SQLite applications include it explicitly in their own dependency list.

Checkpoint schema version 2 adds commit_order, backfills existing rows by their stored insertion order, and makes (thread_id, checkpoint_ns, commit_order) unique. Applications already on checkpoint schema version 1 must add a normal application migration that calls up/1 again:

defmodule MyApp.Repo.Migrations.OrderBeamWeaverCheckpoints do
  use Ecto.Migration

  def up do
    BeamWeaver.Migrations.up(adapters: [:checkpoint])
  end

  def down do
    BeamWeaver.Migrations.down(adapters: [:checkpoint], version: 2)
  end
end

The upgrade preserves checkpoint IDs and uses the persisted order for latest, history, pruning, and fork selection. Run BeamWeaver.Migrations.verify_migrated!(repo: MyApp.Repo) before starting an Ecto checkpointer to fail clearly when the application migration is missing.

Then compile with the adapter:

checkpointer = BeamWeaver.Checkpoint.Ecto.new(repo: MyApp.Repo)
graph = Graph.compile!(workflow, checkpointer: checkpointer)

BeamWeaver.Checkpoint.Ecto.new/1 also supports shallow?: true, which keeps only the latest checkpoint for a thread namespace. Use it when you want retention-limited resumability and do not need full state history or time travel.

Checkpointer Interface

Adapters implement BeamWeaver.Checkpoint.Saver. The core callbacks mirror LangGraph's checkpointer contract while using Elixir return shapes:

Public facade functions such as BeamWeaver.Checkpoint.get_tuple/2, list_records/3, copy_thread/3, delete_thread/2, and prune/3 emit telemetry and normalize adapter output.

Use BeamWeaver.Checkpoint.fetch_tuple/2 and list_result/3 when storage errors must be handled explicitly. They also work with existing savers by wrapping get_tuple/2 and list/3; exceptions become typed checkpoint errors. The compatibility get_tuple/2 and list/3 facades raise on storage failure instead of representing it as a missing checkpoint or empty history.

put_checkpoint_with_writes/7 is atomic by contract. If a saver does not implement that callback, the facade returns {:error, %BeamWeaver.Core.Error{type: :atomic_checkpoint_write_unsupported}} before writing anything.

put_many/3 never falls back to sequential writes. ETS publishes the prepared batch together, while BeamWeaver.Checkpoint.Ecto uses one database transaction on PostgreSQL or SQLite. Both adapters persist a per-thread, per-namespace commit order, so explicit legacy checkpoint IDs remain identities rather than being misused as timestamps. fork_at/4 copies only the selected checkpoint's bounded ancestor lineage and fails when a parent is missing or the configured count or byte limit is exceeded.

Each put_many/3 entry is a map with :config and an explicit-ID :checkpoint; optional fields are :metadata, :versions, :writes, and :write_opts. A batch accepts at most 1,000 checkpoints, 10,000 pending writes, and 64 MiB of Erlang terms. fork_at/4 defaults to 1,000 checkpoints and accepts :max_checkpoints and :max_bytes within hard ceilings of 10,000 checkpoints and 64 MiB.

Storage Optimization

BeamWeaver includes BeamWeaver.Graph.Channels.DeltaChannel, used by message channels and available for append-heavy state. A delta channel checkpoints as missing or as periodic snapshots and can replay write history from the checkpointer.

Use it when a channel grows over many turns and replay cost is acceptable:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Channels.DeltaChannel

messages =
  Graph.channel(
    {DeltaChannel, fn existing, updates -> existing ++ List.wrap(updates) end},
    key: :messages,
    initial: []
  )

For durable retention, combine channel choice with adapter retention:

BeamWeaver.Checkpoint.prune(checkpointer, ["thread-1"])
BeamWeaver.Checkpoint.prune(checkpointer, ["thread-1"], strategy: :delete)

Serialization And Encryption

BeamWeaver uses safe, allowlisted JSON serialization at adapter boundaries where serialization is required. The serialization layer can also encrypt payloads with BeamWeaver.Serialization.Encrypted when an adapter accepts a serialization option:

serialization: [
  codec: BeamWeaver.Serialization.Encrypted,
  encryption_key: :crypto.strong_rand_bytes(32)
]