channels

Hydrate channels from a checkpoint, reconstructing any DeltaChannel whose value is absent from channel_values by replaying ancestor writes.

For most channels (and for delta channels with a DeltaSnapshot or a migrated plain value in channel_values), emptyChannels is sufficient and no saver access is required. When a delta channel is absent from channel_values, an ancestor walk via saver.getDeltaChannelHistory finds the nearest seed and accumulates the writes between it and the target. All delta channels needing replay are batched into a single saver call.

Return the set of DeltaChannel names that should snapshot now.

A channel snapshots when EITHER its accumulated update count reaches snapshotFrequency OR the total supersteps since its last snapshot reaches DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT. Pure predicate — no mutation.

Synthetic task id for exit-mode DeltaChannel writes.

Embeds the superstep in the first UUID group so ORDER BY task_id, idx preserves chronological order while remaining a valid RFC UUID (required by Postgres checkpoint_writes.task_id uuid columns).

Structural check for a DeltaChannel without importing it (avoids an import cycle: delta.ts imports base.ts).

Stores the last value received, but only made available after finish(). Once made available, clears the value.

Stores the result of applying a binary operator to the current value and each new value.

Reducer channel that stores only a sentinel in checkpoint blobs and reconstructs state by replaying ancestor writes through the reducer.

DeltaChannel avoids re-serializing the full accumulated value at every step. Instead of writing the value into channel_values, the channel is omitted entirely and its state is reconstructed on read by walking the ancestor chain and replaying the per-step writes through the reducer (see BaseCheckpointSaver.getDeltaChannelHistory).

Snapshot cadence is driven by two counters: a per-channel update count and the total supersteps since the last snapshot. A full DeltaSnapshot blob is written when EITHER the update count reaches snapshotFrequency OR the supersteps count reaches the system-wide DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT bound (default 5000), bounding replay depth even for channels that stop receiving writes.

Stores the last value received, assumes that if multiple values are received, they are all equal.

Note: Unlike 'LastValue' if multiple nodes write to this channel in a single step, the values will be continuously overwritten.

A channel that switches between two states

• in the "priming" state it can't be read from.
  • if it receives a WaitForNames update, it switches to the "waiting" state.
• in the "waiting" state it collects named values until all are received.
  • once all named values are received, it can be read once, and it switches back to the "priming" state.

Stores the value received in the step immediately preceding, clears after.

A channel that waits until all named values are received before making the value available.

This ensures that if node N and node M both write to channel C, the value of C will not be updated until N and M have completed updating.

A configurable PubSub Topic.

A batch reducer for use with DeltaChannel.

Receives the current accumulated value and a batch of writes in one call, returning the new accumulated value: reducer(state, [write1, write2, ...]) -> newState.

Reducers must be deterministic and batching-invariant (associative across folds): applying two consecutive write batches separately must produce the same state as applying their concatenation once:

reducer(reducer(state, xs), ys) === reducer(state, xs.concat(ys))

This lets LangGraph replay checkpointed writes in larger batches than they were originally produced without changing reconstructed state. If your reducer is not associative, use BinaryOperatorAggregate instead — DeltaChannel is not a drop-in replacement for every reducer.