Channels

A channel is a typed reactive state slot. Nodes in the graph subscribe to channels; a node fires when any of its subscribed channels advances. Channels carry the per-turn state the runtime flows through, and the concurrent-write semantics are defined per kind — not handwaved.

Channels are one of three concepts in the execution-graph cluster — graph, nodes, channels. See Architecture → the execution-graph cluster for the cluster framing.

The six kinds correspond to the LangGraph channel taxonomy. If you've used LangGraph, the shapes will look familiar; the substrate-level invariants (deterministic reducers, sync-only verdicts) come from @pleach/core. See Checkpointing for how checkpoint() / restore() make per-channel time-travel work, and Architecture for how channels compose with the lattice and seams.

import {
  LastValue,
  BinaryOperatorAggregate,
  Topic,
  EphemeralValue,
  NamedBarrier,
  DataChannel,
  appendReducer,
  messagesReducer,
  unionReducer,
  Overwrite,
  REMOVE_ALL_MESSAGES,
} from "@pleach/core";

Picking a channel

The pick is structural — switching channels mid-flight is rare because each kind encodes a different invariant the graph depends on.

LastValue<T>

Keeps the most recent value. Concurrent writes in the same step throw — there's no ambiguity about what the value "should be."

const intent = new LastValue<string>("intent", "unknown");

intent.update("lookup");
intent.get(); // → "lookup"

update takes the scalar value directly — the constructor is new LastValue<T>(name, initialValue), and get() returns the current value.

Use for inputs the graph treats as singular: the resolved provider, the current model id, the active intent.

When two nodes legitimately want to write the same LastValue in the same step, that's a bug in the graph topology — fan-in through an aggregate instead. Two writes to the same LastValue within one step surface as a write conflict on the channel — the runtime emits a write-conflict breadcrumb naming the offending channel. There's no silent "last one wins" behavior.

BinaryOperatorAggregate<T>

Reduces concurrent writes via a deterministic operator. The reducer must be commutative and associative — that's how concurrent writes produce the same result regardless of arrival order.

import {
  BinaryOperatorAggregate,
  appendReducer,
  messagesReducer,
  unionReducer,
} from "@pleach/core";

const messages = new BinaryOperatorAggregate<Message[]>(
  "messages",
  [],
  messagesReducer,
);

Built-in reducers

Build your own when the built-ins don't fit — just keep it commutative + associative. The substrate doesn't validate this; violating the property silently breaks replay determinism. A worked case: a reducer that subtracts ((a, b) => a - b) compiles fine, runs fine, and passes a single live turn — but replaying the same event slice (whether through your own diff harness today or the planned @pleach/eval SKU when it ships) produces different final state when tool.completed events arrive in a different order, and the diff flags the channel as the source of divergence. The fix is to swap to an order-independent shape (max, set union, keyed merge) before the next replay run.

The REMOVE_ALL_MESSAGES sentinel

messagesReducer recognizes a REMOVE_ALL_MESSAGES constant — a write that includes an item whose id equals this sentinel clears the accumulator (the reducer returns []). Used by context-compaction nodes to clear a long history, then write the summary on the next update.

import { REMOVE_ALL_MESSAGES } from "@pleach/core";

// clear the accumulator
messages.update([{ id: REMOVE_ALL_MESSAGES }]);
// then append the summary (every message needs an `id`)
messages.update([{ id: "summary-1", role: "system", content: summary }]);

Topic<T>

Append-only list. Optional clearOnStep flag — when true, the channel resets at the start of every step (good for "things emitted this step"); when false, accumulates across the whole turn.

const artifacts = new Topic<ArtifactRef>("artifacts", { clearOnStep: false });

artifacts.update([artifactA, artifactB]);
artifacts.get(); // → [artifactA, artifactB]

The difference from BinaryOperatorAggregate + appendReducer is intent: a Topic is "many writers contribute items," not "two writers each propose a state." The runtime's tooling treats them differently (Topic updates are atomic per item; BinaryOperatorAggregate updates are atomic per state).

EphemeralValue<T>

Last-write-wins within a step, cleared at the start of the next step. Use for transient per-step state — a current streaming chunk, an in-progress planner hint, a draft fragment.

const currentChunk = new EphemeralValue<string>("currentChunk");

currentChunk.update("Hello");
currentChunk.get(); // → "Hello"
// next step starts:
currentChunk.get(); // → undefined

The "cleared each step" behavior is what makes ephemeral channels safe to skim from outside the graph — there's no historical state to leak.

NamedBarrier

Tracks a named set; releases when every required name has been triggered. Use for synchronization — "wait for every tool in this batch to finish before proceeding to synthesize."

const toolBarrier = new NamedBarrier("toolBatch", ["tool1", "tool2", "tool3"]);

toolBarrier.trigger("tool1");
toolBarrier.trigger("tool2");
toolBarrier.get();      // → false
toolBarrier.pending();  // → ["tool3"]
toolBarrier.trigger("tool3");
toolBarrier.get();      // → true

The constructor takes the barrier name plus the array of required names; get() returns true only once every name has been triggered, trigger(name) records one arrival, and pending() lists the names still outstanding.

The graph scheduler treats a NamedBarrier as a blocker on subscribing nodes until released. Once released, the barrier emits a single advance event and downstream nodes fire. A barrier that never receives one of its required names keeps the synthesizer parked indefinitely — the tool.failed from one of the gating tools doesn't auto-trigger the name, so a graph that wires three required names against three tool calls needs an explicit compensating trigger on the failure path, or the turn stalls until the parent AbortSignal fires.

DataChannel<T>

Bounded cache with memory-pressure eviction. Use for offloaded results keyed by a ref — retrieved docs, large tool outputs, anything where re-computing is expensive but the working set is bounded.

const retrieval = new DataChannel();

retrieval.upsert("doc:abc123", {
  ref: "doc:abc123",
  toolName: "search",
  summary: { description: "12 results" },
  sizeBytes: 4096,
  createdAt: Date.now(),
  superstep: 0,
});

retrieval.getRef("doc:abc123");   // → DataRef (summary the LLM reads)
retrieval.getEntry("doc:abc123"); // → full DataEntry
retrieval.get();                  // → whole channel state (no key arg)

DataChannel is the only channel kind with a separate ref-keyed read/write API (upsert(ref, entry) to write, getRef(ref) / getEntry(ref) to read a single entry) alongside the get() / update pair on the others — get() takes no argument and returns the whole channel state. The graph scheduler doesn't trigger nodes on individual upsert calls; it triggers on the channel's version, which advances per write.

The constructor takes only an options object — there's no positional name. The optional knobs the recovery layer wires through a closure:

new DataChannel({
  hasS3Fallback,    // whether evicted entries can be re-fetched from S3
  refetch,          // (ref) => Promise<RefetchOutcome>
  isGuest,          // skip shadow refetch on guest sessions
  maxMemoryBytes,   // override the eviction threshold
})

refetch is the closure-injected recovery primitive: every LRU eviction fires a background refetch(ref) and emits the outcome through the [UXParity:phase-d-refetch-shadow] probe. The promise is detached (never awaited) so eviction stays on the hot path. Host runtimes that own a recovery strategy (recordGarbledOutput is the parallel example for the recovery node) supply the closure at runtime construction; pure-substrate consumers leave it undefined and the channel falls back to graph-canonical recovery without a manifest round-trip. isGuest short-circuits the shadow refetch when no manifest exists to recover against.

Eviction in action — once cumulative sizeBytes crosses the maxMemoryBytes threshold, the oldest entries (by insertion order) are evicted to make room:

const corpus = new DataChannel({ maxMemoryBytes: 8192 });

const mk = (ref: string, sizeBytes: number) => ({
  ref,
  toolName: "search",
  summary: { description: ref },
  sizeBytes,
  createdAt: Date.now(),
  superstep: 0,
});

corpus.upsert("doc-abc0", mk("doc-abc0", 4096));
corpus.getEntry("doc-abc0");   // → DataEntry
corpus.upsert("doc-abc1", mk("doc-abc1", 4096));
corpus.upsert("doc-abc2", mk("doc-abc2", 4096)); // pushes total over 8192
corpus.getEntry("doc-abc0");   // → null (evicted)
corpus.getEntry("doc-abc2");   // → DataEntry

Composing channels in one turn

A retrieval turn for a knowledge-base assistant uses three kinds together — LastValue for the resolved intent, Topic for the docs the retriever emits, NamedBarrier to gate the synthesizer until every sub-query returns.

const intent = new LastValue<string>("intent", "unknown");
const retrieved = new Topic<{ id: string; score: number }>(
  "retrieval", { clearOnStep: false },
);
const subQueriesDone = new NamedBarrier("subQueries", ["q1", "q2", "q3"]);

intent.update("search_corpus");
retrieved.update([
  { id: "doc-abc123", score: 0.91 },
  { id: "doc-abc124", score: 0.88 },
]);
subQueriesDone.trigger("q1");
subQueriesDone.trigger("q2");
subQueriesDone.trigger("q3");
subQueriesDone.get();   // → true; synthesizer fires

Sequence-number ordering for projected channels

Channels derived from event-log projections (via runtime.events.fold(projection)) inherit a hard ordering invariant from the underlying table: rows on harness_event_log are ordered by (chat_id, sequence_number). The reducer sees events in that order, every replay, every cold-start hydration.

This is what lets fold-based projections reach byte-identical state with snapshot writes — the projection is a deterministic function of an ordered event stream, and sequence_number is the shared total order across producers. See Event log for the durability + ordering contract on the source side.

The Channel<T> interface

All six kinds implement the same contract — building a custom channel is one interface away:

import type { Channel, ChannelUpdate, ChannelType } from "@pleach/core";

interface Channel<T> {
  readonly name: string;
  readonly type: ChannelType;
  get(): T;
  update(value: ChannelUpdate<T>): void;
  reset(): void;
  readonly version: number;
}

update takes a single value (pass an Overwrite<T> to bypass the reducer), get() returns the current value, reset() returns the channel to its initial state, and version is a monotonically increasing getter the scheduler reads to decide which nodes fire. The checkpointer snapshots every channel's get() value and rebuilds via update/reset on rollback — that's what makes per-channel time-travel work. Custom channels MUST report their state honestly; a channel that lies about its value breaks time-travel.

Overwrite and concurrent-write disambiguation

Some channels accept an Overwrite-flagged update that bypasses the reducer:

import { Overwrite } from "@pleach/core";

messages.update(new Overwrite([summary]));
// messages is now exactly [summary], regardless of reducer

Use this sparingly — it's the escape hatch for context compaction and explicit resets, not a general-purpose "skip the reducer" tool. Every Overwrite is a determinism risk if two nodes race to do it. Two Overwrite writes to the same channel in the same step collapse to the last-arriving value, but arrival order across nodes is scheduler-defined — meaning a replay (the planned @pleach/eval SKU, or your own diff harness) can pick the other write and produce a different final state. Keep Overwrite writes single-sourced: the context compactor is one node, the explicit-reset node is one node, and they never run in the same step.

What CI checks when you add a channel

A channel has no gate of its own — it reaches the lattice through a node's subscribes / writes metadata, so the node gate (audit:graph-stages) covers it. There is no contributeChannels hook; you add a channel by adding the node that reads or writes it.

The one rule the substrate does not enforce is the reducer property: a BinaryOperatorAggregate reducer must stay commutative and associative, and a custom Channel<T> must report its state honestly. Both are determinism contracts — break one and replay diverges silently rather than failing a gate. The channel is one row in the extension map.

Where to go next