Contracts

What It Is

Contracts are the extension seams Dagonizer calls: nodes, runtime providers, stores, checkpoint stores, containers, handoff channels, adapters, embedders, tools, graph primitives, and registry modules.

Use this page when authoring a plugin, backend, custom node, custom store, worker container, streaming channel, or runtime integration. If Dagonizer calls into your code, the relevant shape belongs here.

How It Works

Contracts keep implementation detail outside JSON-LD. A DAG placement carries graph identity in @id, references node/DAG implementations by registered IRI, and names container roles, channels, gather strategies, or reducers. Registries and injected services bind those references to concrete code that satisfies these interfaces.

The rule is practical: implement the smallest contract Dagonizer actually needs, then register or inject it at the boundary that owns it.

Diagrams, Examples, and Outputs

Contracts appear throughout the runnable examples. The links below show default implementations and common extension points:

• Reference: Core - GatherStrategy, OutcomeReducer extension classes
• Reference: Runtime - default implementations of the runtime contracts
• Reference: Checkpoint - uses CheckpointStore
• Reference: Store - Store, BaseStore, MemoryStore, StoreError

What It Lets You Do

The contracts reference lets applications implement extension seams without importing private internals.

Adapter contracts live at the root of src/contracts/ and ship through @studnicky/dagonizer/contracts. Single source of truth: never re-exported from a sibling module.

Code Samples

The code below lists the interfaces, option objects, and record shapes application code and plugin packages compile against.

Import

import type {
  // Core dispatcher contracts
  HandoffChannelInterface,
  CheckpointStoreInterface,
  ClockProviderInterface,
  DagContainerInterface,
  EmbedderInterface,
  ErrorConstructorType,
  ExecuteOptionsType,
  GatherExecutionType,
  GatherRecordType,
  LlmAdapterInterface,
  LlmClientInterface,
  MessageChannelInterface,
  NodeInterface,
  NodeInvokerInterface,
  OutcomeRecordType,
  RegistryBundleInterface,
  RegistryModuleInterface,
  RemoteStoreInterface,
  RemoteStoreEndpointType,
  RemoteStoreLeaseType,
  RetryPolicyOptionsType,
  SchedulerProviderInterface,
  SnapshottableInterface,
  StateAccessorInterface,
  StoreInterface,
  StoreSnapshotType,
  StoreSnapshotEntryType,
  SystemInfoInterface,
} from '@studnicky/dagonizer/contracts';

// DagOutcomeType and DagTaskInterface ship through the root barrel
import type {
  DagOutcomeType,
  DagTaskInterface,
} from '@studnicky/dagonizer';

NodeInterface

interface NodeInterface<
  TState extends NodeStateInterface = NodeStateInterface,
  TOutput extends string = string,
> {
  readonly name: string;
  readonly outputs: readonly TOutput[];
  readonly outputSchema: Record<TOutput, SchemaObjectType>;
  readonly timeout: Timeout;
  execute(batch: Batch<TState>, context: NodeContextType): Promise<RoutedBatchType<TOutput, TState>>;
  destroy?(): Promise<void>;
  validate?(): ValidationResultType;
}

The contract every application node implements. Nodes are stateless; they mutate state and route to a named output. They never throw: caught errors route to 'error' or another declared error output.

outputSchema is a mandatory per-output-port JSON Schema 2020-12 record describing the state delta each port guarantees. Every declared output port in outputs must have an entry. Schemas are partial over state — they validate the fields the node writes; do not set additionalProperties: false. MonadicNode provides a passthrough default ({ type: 'object' } per port); concrete nodes should override with real schemas.

timeout is a per-node wall-clock budget expressed as a Timeout value (Timeout.ofMs(n) or Timeout.none()). When set to a non-none value, the engine derives a child AbortController from the run's signal and schedules an abort after the budget. On expiry, NodeTimeoutError is thrown and the run is marked failed. The MonadicNode base class defaults to Timeout.none(); nodes that do not extend it should omit the field (treated as Timeout.none() by the engine).

ExecuteOptionsType

interface ExecuteOptionsType {
  readonly signal?: AbortSignal;
  readonly deadlineMs?: number;
}

Dagonizer.execute and Dagonizer.resume accept this as their third argument. Signal.compose (from @studnicky/signal) folds the two fields into a single signal.

ClockProviderInterface

interface ClockProviderInterface {
  hrtime(): bigint;
}

Backend for the Clock singleton. Implement to swap time sources (typically in tests via VirtualClockProvider from @studnicky/dagonizer/testing).

SchedulerProviderInterface

interface SchedulerProviderInterface {
  after(delayMs: number, options?: AbortableOptionsType): Promise<void>;
  at(atMs: number, options?: AbortableOptionsType): Promise<void>;
  every(intervalMs: number, options?: AbortableOptionsType): AsyncIterable<void>;
  cancelAll(): void;
}

SchedulerProviderInterface is the backend contract; implement it to swap in a custom scheduler. Scheduler.current() returns the active SchedulerProviderInterface. Production uses RealTimeScheduler; tests install VirtualScheduler from @studnicky/dagonizer/testing.

StateAccessorInterface

interface StateAccessorInterface {
  get(state: object, path: string): unknown;
  set(state: object, path: string, value: unknown): void;
}

Path resolver used for scatter source reads, state-mapping input copies, and gather writes. Default implementation: DottedPathAccessor in runtime/. Pass a custom implementation via new Dagonizer({ accessor }).

SnapshottableInterface

interface SnapshottableInterface {
  snapshot(): Promise<StoreSnapshotType>;
  restore(snapshot: StoreSnapshotType): Promise<void>;
}

The capability checkpointing depends on. Checkpoint.capture(dag, result, { stores }) and ckpt.restoreStores(map) take Record<string, SnapshottableInterface>, so a non-KV backing (RDF triple store, vector index) can ride along in a checkpoint without implementing the key-value surface. StoreInterface extends SnapshottableInterface. The StoreSnapshotType / StoreSnapshotEntryType envelopes live with it. See Store for the envelope shape and BaseStore.

CheckpointStoreInterface

interface CheckpointStoreInterface {
  save(key: string, json: string): Promise<void>;
  load(key: string): Promise<string | null>;
  delete(key: string): Promise<void>;
}

Persistence backend for checkpoints. ckpt.persist(store, key) and Checkpoint.recall(store, key) compose the codec with the store. Reference impl: MemoryCheckpointStore. See persistence for a Postgres example.

EmbedderInterface

interface EmbedderInterface {
  readonly id: string;
  readonly displayName: string;
  readonly dimensions: number;
  embed(text: string): Promise<readonly number[]>;
  embedBatch(texts: readonly string[]): Promise<readonly (readonly number[])[]>;
  probe(): Promise<boolean>;
  connect(): Promise<void>;
  disconnect(): Promise<void>;
}

Produces a fixed-dimensionality vector for a text input. Plugins implement this (typically by extending BaseEmbedder from @studnicky/dagonizer/adapter) to swap embedding backends. The adapter cascade pattern applies: register multiple EmbedderInterfaces, probe at runtime, pick the first available.

Member	Description
id	Provider identifier ('ollama', 'gemini-api', etc.)
displayName	Human-readable label for logs and UI
dimensions	Output vector dimensionality. Applications verify match against pre-computed corpus embeddings
embed(text)	Embed a single text, returning a number[] of length dimensions. Throws LlmError on failure
embedBatch(texts)	Batch convenience. Default in BaseEmbedder calls embed() in series
probe()	Quick availability check. Must not throw; returns false so a cascade can route around the embedder
connect() / disconnect()	Per-session lifecycle hooks

RetryPolicyOptionsType / ErrorConstructorType

type ErrorConstructorType = new (...args: never[]) => Error;

interface RetryPolicyOptionsType {
  readonly maxAttempts?: number;
  readonly strategy?: BackoffStrategyType;
  readonly baseDelay?: number;
  readonly maxDelay?: number;
  readonly multiplier?: number;
  readonly jitterFactor?: number;
  readonly retryOn?: readonly ErrorConstructorType[];
  readonly abortOn?: readonly ErrorConstructorType[];
}

Construction options for RetryPolicy. retryOn and abortOn are checked via instanceof. Supply error classes, not error names.

Store / StoreSnapshotType / StoreSnapshotEntryType

The store contracts ship through @studnicky/dagonizer/contracts alongside the other adapter interfaces. Full documentation (concurrency contract, BaseStore authoring guide, StoreErrorClassification taxonomy) lives in Reference: Store.

import type { StoreInterface, StoreSnapshotType, StoreSnapshotEntryType } from '@studnicky/dagonizer/contracts';

See Shared state for the decision matrix and usage patterns.

RemoteStore / RemoteStoreEndpointType / RemoteStoreLeaseType

Extension of Store for network-backed or replicated store plugins. Implements the same Store surface plus endpoint, acquireLease, releaseLease, and health for distributed coordination.

import type { RemoteStoreInterface, RemoteStoreEndpointType, RemoteStoreLeaseType } from '@studnicky/dagonizer/contracts';

See Reference: Store for the full interface and Shared state for the authoring guide.

DagContainerInterface

interface DagContainerInterface {
  runDag(task: DagTaskInterface, options?: { readonly relay?: ObserverRelayInterface }): Promise<DagOutcomeType>;
  destroy?(): Promise<void>;
}

Adapter contract for running an embedded DAG or DAG-body scatter in an isolate (worker thread, forked child, spawned process, Web Worker, or remote host). Bound to the dispatcher via DagonizerOptionsType.containers keyed by logical role name. On a dispatcher with a non-empty containers registry, a declared-but-unbound role throws DAGError at registerDAG time. A pure in-process dispatcher (empty containers) treats declared roles as inert and runs every body in-process.

runDag must preserve the child DAG boundary: the task carries the selected DAG IRI, placement path, state snapshot, timeout, and execution context; the outcome returns terminal output, terminal state snapshot, collected errors, and intermediates. Transport failures, host crashes, and serialization errors are returned as collected errors in DagOutcomeType.errors with recoverable: false.

destroy() is optional. Implement it to release pool resources when the dispatcher shuts down.

HandoffChannelInterface

interface HandoffChannelInterface {
  publish(handoff: DAGHandoffType): Promise<void>;
  destroy?(): Promise<void>;
}

Adapter contract for publishing completed-DAG hand-off envelopes to a downstream transport (queue, message bus, or loopback store). Bound via DagonizerOptionsType.channels keyed by terminal placement name. Implementations must not throw out of the dispatcher; any internal transport error is the implementation's responsibility. InMemoryChannel in @studnicky/dagonizer/channels is the reference implementation.

MessageChannelInterface

interface MessageChannelInterface {
  send(message: BridgeMessageType): void;
  onMessage(handler: (message: BridgeMessageType) => void): void;
  close(): void;
}

Duplex channel contract between a parent dispatcher and a DagHost. send is fire-and-forget (does not throw). onMessage registers the inbound handler (replaces any previous handler). close severs both directions; outstanding send calls are silently dropped. Implementations include LoopbackChannel (in-memory, for testing), MessagePortChannel (worker threads), IpcChannel (child process), and NdjsonChannel (stdio, polyglot hosts).

RegistryModuleInterface / RegistryBundleInterface

interface RegistryBundleInterface {
  readonly bundle:          DispatcherBundleType<NodeStateInterface>;
  readonly registryVersion: string;
  readonly restoreState:    CheckpointRestoreAdapterInterface<NodeStateInterface>;
  destroy?():               Promise<void>;
}

interface RegistryModuleInterface {
  instantiate(servicesConfig: JsonObjectType): Promise<RegistryBundleInterface>;
}

RegistryModuleInterface is the default export shape of a registry module loaded by DagHost via dynamic import. instantiate receives the opaque servicesConfig JSON from the init message and returns a fully initialised RegistryBundleInterface.

RegistryBundleInterface bundles the node+DAG registry (bundle), the semantic version for the init ↔ ready handshake (registryVersion), and the state restore factory (restoreState). Node instances are constructed inside the registry module (with their constructor-injected dependencies); the constructed instances cross no isolate boundary — each isolate builds its own graph via its registry module.

DagOutcomeType

interface DagOutcomeType {
  readonly terminalOutput: string;
  readonly errors:         readonly NodeErrorWireType[];
  readonly stateSnapshot:  JsonObjectType | null;
  readonly intermediates:  readonly ExecutorIntermediateType[];
}

Result returned by DagContainerInterface.runDag() after a child DAG completes in an isolate. terminalOutput is the routing output the child resolved to. stateSnapshot is the terminal child state snapshot (null when the container cannot produce one, e.g. transport failure); the parent calls cloneState.applySnapshot(stateSnapshot) when non-null. intermediates are per-node results forwarded to the parent execution stream.

DagTaskInterface

interface DagTaskInterface {
  dagName:        string;
  placementPath:  string[];
  correlationId:  string;
  timeout:        Timeout;
  state:          NodeStateInterface;
  context:        NodeContextType;
  toRequest(): ExecutionRequestType;
}

Engine-side descriptor of a contained DAG execution. Carries a live seeded child clone (state, typed at the NodeStateInterface contract because the engine is heterogeneous-state) for the in-process path. Isolating containers call toRequest() to snapshot the clone into a wire-safe ExecutionRequest. correlationId is a dispatcher-monotonic id (no randomness). timeout is a Timeout; Timeout.none() means no per-task budget applies.

SystemInfoInterface

interface SystemInfoInterface {
  recommendedWorkerCount(config: RecommendedWorkerCountConfigType): number;
}

Host-environment probe for pool sizing recommendations. Implementations are environment-specific (Node os.availableParallelism() + os.totalmem(); Web navigator.hardwareConcurrency). The recommended count follows the quadrascope formula: clamp(parallelism − mainThreadReservation, minimumWorkerCount, maximumWorkers), optionally further clamped by memoryPerWorkerBytes.

GatherExecutionType / GatherRecordType / OutcomeRecordType

These contracts ship through @studnicky/dagonizer/contracts for use by custom gather strategy and outcome reducer implementations. See Reference: Core for the full authoring guide.

import type { GatherExecutionType, GatherRecordType, OutcomeRecordType } from '@studnicky/dagonizer/contracts';

GatherRecordType<TState> carries producer results into a GatherNode: source, index, item, output, terminalOutcome, result, and cloneState. GatherExecutionType<TState> is the invocation context handed to GatherStrategy.finalize: it provides records, the live parent state, the accessor, and invoker (a NodeInvoker; used by the custom strategy via invoker.invokeNode(nodeIri)). OutcomeRecordType is the per-clone summary handed to OutcomeReducer.reduce: index, output, and terminalOutcome.

LlmAdapterInterface / LlmClientInterface

interface LlmAdapterInterface {
  readonly id:           string;
  readonly displayName:  string;
  readonly capabilities: AdapterCapabilitiesType;
  chat(request: ChatRequestType): Promise<ChatResponseType>;
  chatStream(request: ChatRequestType, sink: StreamSinkInterface<ChatStreamChunkType>): Promise<ChatResponseType>;
  connect(): Promise<void>;
  disconnect(): Promise<void>;
  probe(): Promise<boolean>;
}

interface LlmClientInterface {
  chat(request: ChatRequestType): Promise<ChatResponseType>;
}

LlmAdapterInterface is the transport contract every LLM provider adapter implements. Provider packages extend BaseAdapter from @studnicky/dagonizer/adapter to inherit retry and error classification. BaseAdapterOptionsType also carries cross-cutting options every adapter accepts: systemPrompt, a default directive the base injects as the leading system message of any request that carries none (never overriding an explicit system turn), timeoutMs (default 60_000), a per-request deadline, optional circuitBreaker / tokenBucket guards, and optional timing for substrate adapter.chat.* / adapter.chatStream.* events. An expired deadline surfaces as a TIMEOUT classification so a cascade falls through instead of hanging. LlmClientInterface is the minimal chat surface pattern bases accept — any LlmAdapterInterface satisfies it. Pattern bases that need capability metadata (e.g. tool-call support) accept the full LlmAdapterInterface directly.

chat() is the buffered call: it resolves once with the complete ChatResponseType. chatStream(request, sink) additionally pushes incremental ChatStreamChunkType ({ delta }) values to sink as the response is generated, while still resolving with the same fully-assembled ChatResponseType — the sink is a pure observation channel, not an alternate return path. BaseAdapter's default performChatStream is buffered: it calls chat() internally and pushes exactly one chunk carrying the full response text, so every adapter satisfies the streaming contract even without a streaming backend. Anthropic, the Gemini API adapter, and the Ollama, Groq, Cerebras, Mistral, and OpenRouter adapters override performChatStream to push real per-token deltas parsed from a server-sent-events response body. gemini-nano streams via the in-browser LanguageModel session's promptStreaming() async iterable; web-llm streams via the @mlc-ai/web-llm engine's own stream. Tool-bearing requests (request.tools.length > 0) use the buffered default because partial tool-call JSON is unsafe to parse mid-stream. chatStream is single-attempt and bounded by the same abort+timeout deadline (timeoutMs). sink.push() delivery is best-effort: a rejecting sink never fails the call. See Adapters for the full per-provider streaming reference and ReAct agent: live token streaming for a working CallModelNode + sink example.

NodeInvokerInterface

interface NodeInvokerInterface {
  invokeNode(nodeIri: string): Promise<void>;
}

Typed contract for dispatching a registered node back through the engine. Lives on GatherExecutionType.invoker; used exclusively by custom gather strategies to invoke the registered node IRI in GatherConfig.customNode. Custom strategies access it via execution.invoker.invokeNode(nodeIri).

Details for Nerds

Contracts are intentionally narrow. A node contract does not know how the dispatcher stores registries. A store contract does not know how checkpoints serialize. A channel contract does not know which host receives the handoff. Each seam receives only the methods Dagonizer needs to call.

That keeps plugin packages portable: implement the contract, register or inject the implementation, and let JSON-LD continue describing topology by IRI.

Related Concepts

• Reference: Core - GatherStrategy, OutcomeReducer extension classes
• Reference: Runtime - default implementations of the runtime contracts
• Reference: Checkpoint - uses CheckpointStore
• Reference: Store - Store, BaseStore, MemoryStore, StoreError
• Reference: Adapters - LlmAdapterInterface implementations, buffered vs. streaming, cascades
• Cancellation - ExecuteOptionsType.signal behavior
• Dependency Injection - pass contract implementations through constructors
• State Accessors - StateAccessorInterface
• Persistence - checkpoint-store implementations
• Shared State - store contracts and remote-store behavior
• Observability - lifecycle and progress contracts