# Channels (/docs/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](/docs/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](/docs/checkpointing) for how `checkpoint()` / `restore()` make per-channel time-travel work, and [Architecture](/docs/architecture) for how channels compose with the lattice and seams. ```typescript import { LastValue, BinaryOperatorAggregate, Topic, EphemeralValue, NamedBarrier, DataChannel, appendReducer, messagesReducer, unionReducer, Overwrite, REMOVE_ALL_MESSAGES, } from "@pleach/core"; ``` ## Picking a channel [#picking-a-channel] | Channel | Concurrent writes | Persists across steps | Use for | | ---------------------------- | -------------------- | -------------------------- | -------------------------------------------------- | | `LastValue` | **Throws** | yes | Scalars: provider config, model id, current intent | | `BinaryOperatorAggregate` | Reduced via operator | yes | Accumulators: messages, plans, retrieved docs | | `Topic` | Appended | configurable | Per-step lists: artifacts, pending writes | | `EphemeralValue` | Last-write-wins | **no** (cleared each step) | Transient: current chunk, in-progress hint | | `NamedBarrier` | Tracked by name | until released | Synchronization: wait for a named set | | `DataChannel` | LRU-evicted | yes (bounded) | Cached lookups, retrieval results | The pick is structural — switching channels mid-flight is rare because each kind encodes a different invariant the graph depends on. ## `LastValue` [#lastvaluet] Keeps the most recent value. Concurrent writes in the same step throw — there's no ambiguity about what the value "should be." ```typescript const intent = new LastValue("intent", "unknown"); intent.update("lookup"); intent.get(); // → "lookup" ``` `update` takes the scalar value directly — the constructor is `new LastValue(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` [#binaryoperatoraggregatet] 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. ```typescript import { BinaryOperatorAggregate, appendReducer, messagesReducer, unionReducer, } from "@pleach/core"; const messages = new BinaryOperatorAggregate( "messages", [], messagesReducer, ); ``` ### Built-in reducers [#built-in-reducers] | Reducer | Shape | Behavior | | ------------------ | --------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | | `appendReducer` | `(a: T[], b: T[]) => T[]` | Concatenates; order = arrival order | | `messagesReducer` | `(a: T[], b: T[]) => T[]` | Append with `id`-based dedup (the `T extends { id: string }` bound is what makes dedup possible); honors `REMOVE_ALL_MESSAGES` sentinel | | `unionReducer` | `(a: Set, b: Set) => Set` | Set union | 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 [#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. ```typescript 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` [#topict] 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. ```typescript const artifacts = new Topic("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` [#ephemeralvaluet] 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. ```typescript const currentChunk = new EphemeralValue("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` [#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." ```typescript 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` [#datachannelt] 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. ```typescript 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: ```typescript new DataChannel({ hasS3Fallback, // whether evicted entries can be re-fetched from S3 refetch, // (ref) => Promise 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: ```typescript 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 [#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. ```typescript const intent = new LastValue("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 [#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](/docs/event-log) for the durability + ordering contract on the source side. ## The `Channel` interface [#the-channelt-interface] All six kinds implement the same contract — building a custom channel is one interface away: ```typescript import type { Channel, ChannelUpdate, ChannelType } from "@pleach/core"; interface Channel { readonly name: string; readonly type: ChannelType; get(): T; update(value: ChannelUpdate): void; reset(): void; readonly version: number; } ``` `update` takes a single value (pass an `Overwrite` 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 [#overwrite-and-concurrent-write-disambiguation] Some channels accept an `Overwrite`-flagged update that bypasses the reducer: ```typescript 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 [#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` must report its state honestly. Both are [determinism contracts](/docs/determinism) — break one and replay diverges silently rather than failing a gate. The channel is one row in the [extension map](/docs/extending). ## Where to go next [#where-to-go-next]