Dagonizer

What It Is

Dagonizer<TState> is the dispatcher. It owns the node registry, DAG registry, plugin registration boundary, lifecycle hooks, execution entrypoints, resume entrypoints, and read accessors.

Use this page when integrating the dispatcher directly into an application host. The key distinction is simple: the DAG document describes what should run; the dispatcher owns the registered implementations and moves state through the routed graph.

How It Works

Register nodes before DAGs. Register plugins before parent DAGs that embed plugin-provided DAG IRIs. Register state factories when embedded DAGs need child state that is not just a clone of the parent state.

execute() and resume() return lazy Execution<TState> objects. Nothing runs until the application awaits the result or iterates events. Validation happens at registration time so dangling node/DAG references, missing embedded DAGs, invalid placement-IRI routes, and contract mismatches fail before a run starts.

Diagrams, Examples, and Outputs

The dispatcher is visible in every runnable demo: it registers the same JSON-LD DAGs that the docs render as diagrams, then executes those registered names.

• Reference: Execution - what execute and resume return
• Reference: Contracts - NodeInterface, ExecuteOptionsType
• Reference: Core - GatherStrategies, OutcomeReducers
• Reference: Lifecycle
• The Archivist - browser runner registering a large conversational DAG
• The Cartographer - browser runner registering plugin-defined and embedded data-pipeline DAGs

What It Lets You Do

The Dagonizer reference lets applications register nodes, DAGs, bundles, and plugins, then execute or resume registered graphs. It is the API to reach for when a CLI, browser page, worker, serverless handler, or long-running service needs to host a DAG.

@studnicky/dagonizer root export.

Code Samples

The code below covers constructor options, registration, DAG document loading, execution, resume, teardown, lifecycle hooks, bundles, and reserved progress metadata.

Import

import { Dagonizer, NodeStateBase } from '@studnicky/dagonizer';

Class: Dagonizer<TState>

The DAG dispatcher. Holds node and DAG registries, validates configurations at registration time, and runs the node-graph iterator.

const dispatcher = new Dagonizer<MyState>();

TState must satisfy NodeStateInterface. In practice, always extend NodeStateBase.

Constructor

// constructor(options?: DagonizerOptionsType)
declare const options: DagonizerOptionsType;
const d = new Dagonizer(options);

options.accessor swaps the path resolver for scatter source reads, state-mapping input copies, and gather writes. Defaults to DottedPathAccessor.

DagonizerOptionsType

declare const _opts: DagonizerOptionsType;
// accessor?: StateAccessorInterface
// containers?: Readonly<Record<string, DagContainerInterface>>
// channels?: Readonly<Record<string, HandoffChannelInterface>>
// registryVersion?: string
// validateOutputs?: boolean
export {};

Field	Type	Description
accessor	StateAccessorInterface	Path resolver for scatter source reads, gather writes, and state-mapping copies. Defaults to DottedPathAccessor.
containers	Readonly<Record<string, DagContainerInterface>>	Named container backends keyed by logical role name. On a non-empty registry, a placement that declares a role this map does not bind throws DAGError at registerDAG time. Defaults to an empty registry, where declared roles are inert and DAG bodies run in-process.
channels	Readonly<Record<string, HandoffChannelInterface>>	Named egress channels keyed by terminal placement name. When a non-embedded flow reaches a named terminal, the dispatcher builds a DAGHandoff envelope and calls channel.publish(handoff). Unbound terminals do not publish.
registryVersion	string	Registry version string included in every DAGHandoff envelope for receiver version-handshake validation. Defaults to '0'.
validateOutputs	boolean	When true, validates each node output against the node's declared outputSchema for that port after execution. On mismatch the item is re-routed to 'error'. Default false — zero overhead in production. Enable in dev/test to catch contract violations early.

---

registerNode(node)

declare const node: NodeInterface<MyState, string>;
const d = new Dagonizer<MyState>();
d.registerNode(node);

Registers a node in the dispatcher's node registry. If the node defines an optional validate() method, it is called immediately and throws DAGError if it returns { valid: false }.

Nodes are stored widened to NodeInterface<NodeStateInterface, string>. TState is widened to NodeStateInterface so heterogeneous child-node states (whose concrete class may differ from TState) are stored without casts. Narrowing TOutput to wide string is sound covariantly.

---

registerBundle(bundle)

declare const bundle: DispatcherBundleType<MyState>;
const d = new Dagonizer<MyState>();
d.registerBundle(bundle);

Register every node, then every DAG, in the supplied bundle. Order is fixed: nodes first so the semantic-pass DAG validator can resolve every node reference. Throws as soon as any individual registration throws (validation failure, duplicate name, etc.); registrations that ran before the failing one remain installed.

declare const _b: DispatcherBundleType<NodeStateInterface>;
// readonly nodes: readonly NodeInterface<TState, string>[]
// readonly dags:  readonly DAGType[]
export {};

Both arrays are required. Either may be empty (a node-only bundle uses dags: []; a DAG-only bundle uses nodes: []).

//
// IRI identity: DAGBuilder embeds the canonical DAG_CONTEXT in every built
// DAG's `@context` field. The archivist uses explicit placement IRIs via
// DAGIdentity.placementId(dagIri, placementIdentifier) so route targets,
// gather sources, and entrypoint wiring all stay on canonical IRIs while the
// placement `name` field remains display-only.
//
// @id values on each node placement follow the urn:noocodec:dag:<dagName>/node/<placementName>
// convention produced by DAGIdentity.placementId(), e.g.:
//   urn:noocodec:dag:the-archivist/node/recall-context
//
// A plugin shipping nodes under its own namespace would declare a prefix in
// the bundle's `context` field — e.g. { context: { archivist: 'https://archivist.example.com/' } }
// — and use prefixed names like 'archivist:recallContext' to prevent collisions
// with other plugins that might register a node named 'recallContext'.
// See docs/guide/iri-identity.md for the full expansion rule set.
const BOOK_SEARCH_TOOL_DAGS = [
  'urn:noocodec:tool:web_search_books',
  'urn:noocodec:tool:google_books_search',
  'urn:noocodec:tool:subject_search',
  'urn:noocodec:tool:wikipedia_summary',
] as const;

const ARCHIVIST_DAG_IRI = 'urn:noocodec:dag:the-archivist';
const BOOK_SEARCH_SCATTER_DAG_IRI = 'urn:noocodec:dag:book-search-scatter';
const COMPOSE_RETRY_LOOP_DAG_IRI = 'urn:noocodec:dag:compose-retry-loop';
const placement = (placementIdentifier: string): string => DAGIdentity.placementId(ARCHIVIST_DAG_IRI, placementIdentifier);
const display = <T extends string>(name: T): { name: T } => ({ name });

const nodes = {
  'preRunSetup': new PlaceholderNode<ArchivistState, 'ready'>('urn:noocodec:node:pre-run-setup', ['ready']),
  'parkForInput': new PlaceholderNode<ArchivistState, 'parked' | 'resumed'>('urn:noocodec:node:park-for-input', ['parked', 'resumed']),
  'recallContext': new PlaceholderNode<ArchivistState, 'recalled'>('urn:noocodec:node:recall-context', ['recalled']),
  'classifyIntent': new PlaceholderNode<ArchivistState, 'lookup-author' | 'find-reviews' | 'describe-book' | 'recommend-similar' | 'recall-memories' | 'on-topic' | 'recommend-top-rated' | 'off-topic' | 'retry' | 'salvage'>('urn:noocodec:node:classify-intent', ['lookup-author', 'find-reviews', 'describe-book', 'recommend-similar', 'recall-memories', 'on-topic', 'recommend-top-rated', 'off-topic', 'retry', 'salvage']),
  'classifyIntentSalvage': new PlaceholderNode<ArchivistState, 'done'>('urn:noocodec:node:classify-intent-salvage', ['done']),
  'extractQuery': new PlaceholderNode<ArchivistState, 'success' | 'retry' | 'salvage'>('urn:noocodec:node:extract-query', ['success', 'retry', 'salvage']),
  'extractQuerySalvage': new PlaceholderNode<ArchivistState, 'done'>('urn:noocodec:node:extract-query-salvage', ['done']),
  'decideTools': new PlaceholderNode<ArchivistState, 'tools' | 'no-tools' | 'retry' | 'salvage'>('urn:noocodec:node:decide-tools', ['tools', 'no-tools', 'retry', 'salvage']),
  'decideToolsSalvage': new PlaceholderNode<ArchivistState, 'done'>('urn:noocodec:node:decide-tools-salvage', ['done']),
  'buildBookWorksets': new PlaceholderNode<ArchivistState, 'ready'>('urn:noocodec:node:build-book-worksets', ['ready']),
  'rankByRating': new PlaceholderNode<ArchivistState, 'ranked'>('urn:noocodec:node:rank-by-rating', ['ranked']),
  'pickBestMatch': new PlaceholderNode<ArchivistState, 'picked'>('urn:noocodec:node:pick-best-match', ['picked']),
  'mergeCandidates': new PlaceholderNode<ArchivistState, 'ranked' | 'empty'>('urn:noocodec:node:merge-candidates', ['ranked', 'empty']),
  'recordFindings': new PlaceholderNode<ArchivistState, 'recorded'>('urn:noocodec:node:record-findings', ['recorded']),
  'hasCitationsGate': new PlaceholderNode<ArchivistState, 'pass' | 'fail'>('urn:noocodec:node:has-citations-gate', ['pass', 'fail']),
  'groupByYear': new PlaceholderNode<ArchivistState, 'ordered'>('urn:noocodec:node:group-by-year', ['ordered']),
  'recallPastVisits': new PlaceholderNode<ArchivistState, 'recalled'>('urn:noocodec:node:recall-past-visits', ['recalled']),
  'recommendSimilar': new PlaceholderNode<ArchivistState, 'seeded' | 'empty'>('urn:noocodec:node:recommend-similar', ['seeded', 'empty']),
  'recallMemories': new PlaceholderNode<ArchivistState, 'recalled'>('urn:noocodec:node:recall-memories', ['recalled']),
  'composeMemoryResponse': new PlaceholderNode<ArchivistState, 'drafted' | 'retry' | 'salvage'>('urn:noocodec:node:compose-memory-response', ['drafted', 'retry', 'salvage']),
  'composeMemoryResponseSalvage': new PlaceholderNode<ArchivistState, 'done'>('urn:noocodec:node:compose-memory-salvage', ['done']),
  'respondToVisitor': new PlaceholderNode<ArchivistState, 'success'>('urn:noocodec:node:respond-to-visitor', ['success']),
  'declineOffTopic': new PlaceholderNode<ArchivistState, 'success'>('urn:noocodec:node:decline-off-topic', ['success']),
  'composeEmptyResponse': new PlaceholderNode<ArchivistState, 'drafted' | 'retry' | 'salvage'>('urn:noocodec:node:compose-empty', ['drafted', 'retry', 'salvage']),
  'composeEmptyResponseSalvage': new PlaceholderNode<ArchivistState, 'done'>('urn:noocodec:node:compose-empty-salvage', ['done']),
} as const;

export const archivistDAG: DAGType = new DAGBuilder(ARCHIVIST_DAG_IRI, '6.0', display('the-archivist'))

      // ── pre-phase: setup ─────────────────────────────────────────────────────
      // Stamps state.runId and clears any stale draft before the main loop starts.
      // PhaseNode placement: runs before the entrypoint node; errors abort the run.
      // No routing: phase placements are out-of-band and never set the entrypoint.
      .phase(placement('setup'), 'pre', nodes.preRunSetup, display('setup'))

      // ── 0. park-for-input (HITL gate) ────────────────────────────────────────
      // First added → auto-entrypoint. Parks the flow when state.query is empty,
      // waiting for the human to supply input via the browser HITL banner. On
      // resume, `state.query` is set by the caller before `dispatcher.resume()`;
      // the node then routes `'resumed'` and proceeds to `recall-context`.
      // The `'parked'` output routes to the engine park machinery (null wiring).
      .node(placement('park-for-input'), nodes.parkForInput, {
        'parked':  placement('park-for-input'),
        'resumed': placement('recall-context'),
      }, display('park-for-input'))

      // ── 1. recall-context ────────────────────────────────────────────────────
      // Runs before classifyIntent so the classifier can benefit from
      // prior-session continuity hints.
      .node(placement('recall-context'), nodes.recallContext, {
        'recalled': placement('classify-intent'),
      }, display('recall-context'))

      // ── 1. classify-intent ───────────────────────────────────────────────────
      // Wide output union routes to seven branches. EmbeddedDAG placements and inline
      // branches share the same shared terminal: compose-loop and compose-empty.
      // recall-memories routes directly to memory-recall → compose-memory-recall
      // → memory-respond (no search needed; the memory store is the source).
      .node(placement('classify-intent'), nodes.classifyIntent, {
        'lookup-author':        placement('author-search'),
        'find-reviews':         placement('reviews-extract'),
        'describe-book':        placement('describe-extract'),
        'recommend-similar':    placement('recommend-similar'),
        'recall-memories':      placement('memory-recall'),
        'on-topic':             placement('on-topic-search'),
        'recommend-top-rated':  placement('recommend-extract'),
        'off-topic':            placement('decline-off-topic'),
        // Own timeout / classifier failure → retry budget decides. 'retry' loops
        // back; 'salvage' defaults to the broadest on-topic search via a node.
        'retry':                placement('classify-intent'),
        'salvage':              placement('classify-intent-salvage'),
      }, display('classify-intent'))
      .node(placement('classify-intent-salvage'), nodes.classifyIntentSalvage, {
        'done': placement('on-topic-search'),
      }, display('classify-intent-salvage'))

      // ── on-topic branch ──────────────────────────────────────────────────────
      // EmbeddedDAGNode: book-search-scatter handles extract-query, decide-tools,
      // all four scouts, rank-candidates, merge, record, gate, and recall.
      // One packaged cluster; first of three placements of the same sub-DAG.
      // gather.map copies the fields the sub-DAG writes back to the parent state
      // so compose-loop and group-by-year can read them.
      .embed(placement('on-topic-search'), BOOK_SEARCH_SCATTER_DAG_IRI, {
        'success': placement('compose-loop'),
        'error':   placement('compose-empty'),
      }, {
        'name': 'on-topic-search',
        'outputs': {
          'terms':         'terms',
          'toolPlan':      'toolPlan',
          'candidates':    'candidates',
          'shortlist':     'shortlist',
          'priorContext':  'priorContext',
          'failureCause':  'failureCause',
        },
      })

      // ── lookup-author branch ─────────────────────────────────────────────────
      // EmbeddedDAGNode: same book-search-scatter cluster, second placement.
      // After success, group-by-year sorts results chronologically before the
      // compose loop; author surveys read better in publication-timeline order.
      .embed(placement('author-search'), BOOK_SEARCH_SCATTER_DAG_IRI, {
        'success': placement('group-by-year'),
        'error':   placement('compose-empty'),
      }, {
        'name': 'author-search',
        'outputs': {
          'terms':         'terms',
          'toolPlan':      'toolPlan',
          'candidates':    'candidates',
          'shortlist':     'shortlist',
          'priorContext':  'priorContext',
          'failureCause':  'failureCause',
        },
      })
      // group-by-year is author-branch-specific: sorts shortlist chronologically.
      .node(placement('group-by-year'), nodes.groupByYear, {
        'ordered': placement('compose-loop'),
      }, display('group-by-year'))

      // ── find-reviews branch ───────────────────────────────────────────────────
      // Inlined. Uses rankByRating (deterministic, rating-weighted) in place of
      // rankCandidates (LLM-driven). The Google Books scout carries notes.rating /
      // notes.ratingsCount; rankByRating weights those for reviews-style output.
      .node(placement('reviews-extract'), nodes.extractQuery, {
        'success': placement('reviews-decide-tools'),
        'retry':   placement('reviews-extract'),
        'salvage': placement('reviews-extract-salvage'),
      }, display('reviews-extract'))
      .node(placement('reviews-extract-salvage'), nodes.extractQuerySalvage, {
        'done': placement('reviews-decide-tools'),
      }, display('reviews-extract-salvage'))
      .node(placement('reviews-decide-tools'), nodes.decideTools, {
        'tools':    placement('reviews-build-worksets'),
        'no-tools': placement('reviews-build-worksets'),
        'retry':    placement('reviews-decide-tools'),
        'salvage':  placement('reviews-decide-tools-salvage'),
      }, display('reviews-decide-tools'))
      .node(placement('reviews-decide-tools-salvage'), nodes.decideToolsSalvage, {
        'done': placement('reviews-build-worksets'),
      }, display('reviews-decide-tools-salvage'))
      // Build scatter worksets: converts toolPlan into bookWorksets items so the
      // scatter can dispatch to each declared tool DAG IRI.
      .node(placement('reviews-build-worksets'), nodes.buildBookWorksets, {
        'ready': placement('reviews-scatter'),
      }, display('reviews-build-worksets'))
      // Tool-registry scatter: each bookWorksets item carries its own tool DAG
  // IRI. The following GatherNode reads each clone's output via
      // accessor (no cast) and folds CandidateType[] into parent candidates.
      .scatter(placement('reviews-scatter'), 'bookWorksets', { 'dag': { 'from': 'item', 'path': 'dagIri', 'candidates': BOOK_SEARCH_TOOL_DAGS } }, {
        'success': placement('reviews-gather'),
        'error': placement('reviews-gather'),
        'empty':   placement('reviews-rank'),
      }, {
        'name': 'reviews-scatter',
        'execution': { 'mode': 'item', 'concurrency': 4 },
        'reducer': 'any-success',
      })
  .gather(placement('reviews-gather'), { [placement('reviews-scatter')]: {} }, { 'strategy': 'tool-candidate-merge' }, {
    'success': placement('reviews-rank'),
    'error': placement('reviews-rank'),
    'empty': placement('reviews-rank'),
  }, display('reviews-gather'))
      .node(placement('reviews-rank'),    nodes.rankByRating,     { 'ranked': placement('reviews-merge') }, display('reviews-rank'))
      .node(placement('reviews-merge'),   nodes.mergeCandidates,  { 'ranked': placement('reviews-record'), 'empty': placement('compose-empty') }, display('reviews-merge'))
      .node(placement('reviews-record'),  nodes.recordFindings,   { 'recorded': placement('reviews-gate') }, display('reviews-record'))
      .node(placement('reviews-gate'),    nodes.hasCitationsGate, { 'pass': placement('reviews-recall'), 'fail': placement('compose-empty') }, display('reviews-gate'))
      .node(placement('reviews-recall'),  nodes.recallPastVisits, { 'recalled': placement('compose-loop') }, display('reviews-recall'))

      // ── recommend-top-rated branch ───────────────────────────────────────────
      // Inlined, structural sibling of find-reviews. Reuses rankByRating
      // (deterministic, rating-weighted) instead of rankCandidates (LLM-driven)
      // because a vague "good book / good story" request carries no topic for
      // relevance ranking — rating is the only signal that makes sense.
      .node(placement('recommend-extract'), nodes.extractQuery, {
        'success': placement('recommend-decide-tools'),
        'retry':   placement('recommend-extract'),
        'salvage': placement('recommend-extract-salvage'),
      }, display('recommend-extract'))
      .node(placement('recommend-extract-salvage'), nodes.extractQuerySalvage, {
        'done': placement('recommend-decide-tools'),
      }, display('recommend-extract-salvage'))
      .node(placement('recommend-decide-tools'), nodes.decideTools, {
        'tools':    placement('recommend-build-worksets'),
        'no-tools': placement('recommend-build-worksets'),
        'retry':    placement('recommend-decide-tools'),
        'salvage':  placement('recommend-decide-tools-salvage'),
      }, display('recommend-decide-tools'))
      .node(placement('recommend-decide-tools-salvage'), nodes.decideToolsSalvage, {
        'done': placement('recommend-build-worksets'),
      }, display('recommend-decide-tools-salvage'))
      // Build scatter worksets: converts toolPlan into bookWorksets items so the
      // scatter can dispatch to each declared tool DAG IRI.
      .node(placement('recommend-build-worksets'), nodes.buildBookWorksets, {
        'ready': placement('recommend-scatter'),
      }, display('recommend-build-worksets'))
      // Tool-registry scatter: each bookWorksets item carries its own tool DAG
  // IRI. The following GatherNode reads each clone's output via
      // accessor (no cast) and folds CandidateType[] into parent candidates.
      .scatter(placement('recommend-scatter'), 'bookWorksets', { 'dag': { 'from': 'item', 'path': 'dagIri', 'candidates': BOOK_SEARCH_TOOL_DAGS } }, {
        'success': placement('recommend-gather'),
        'error': placement('recommend-gather'),
        'empty':   placement('recommend-rank'),
      }, {
        'name': 'recommend-scatter',
        'execution': { 'mode': 'item', 'concurrency': 4 },
        'reducer': 'any-success',
      })
  .gather(placement('recommend-gather'), { [placement('recommend-scatter')]: {} }, { 'strategy': 'tool-candidate-merge' }, {
    'success': placement('recommend-rank'),
    'error': placement('recommend-rank'),
    'empty': placement('recommend-rank'),
  }, display('recommend-gather'))
      .node(placement('recommend-rank'),   nodes.rankByRating,     { 'ranked': placement('recommend-merge') }, display('recommend-rank'))
      .node(placement('recommend-merge'),  nodes.mergeCandidates,  { 'ranked': placement('recommend-record'), 'empty': placement('compose-empty') }, display('recommend-merge'))
      .node(placement('recommend-record'), nodes.recordFindings,   { 'recorded': placement('recommend-gate') }, display('recommend-record'))
      .node(placement('recommend-gate'),   nodes.hasCitationsGate, { 'pass': placement('recommend-recall'), 'fail': placement('compose-empty') }, display('recommend-gate'))
      .node(placement('recommend-recall'), nodes.recallPastVisits, { 'recalled': placement('compose-loop') }, display('recommend-recall'))

      // ── describe-book branch ─────────────────────────────────────────────────
      // Inlined. Uses pickBestMatch to narrow multi-hit results to the top-3
      // title-similar candidates before merge. Ensures the composer receives the
      // specific book the visitor named, not arbitrary top-5 hits.
      .node(placement('describe-extract'),      nodes.extractQuery,     { 'success': placement('describe-decide-tools'), 'retry': placement('describe-extract'), 'salvage': placement('describe-extract-salvage') }, display('describe-extract'))
      .node(placement('describe-extract-salvage'), nodes.extractQuerySalvage, { 'done': placement('describe-decide-tools') }, display('describe-extract-salvage'))
      .node(placement('describe-decide-tools'), nodes.decideTools,      { 'tools': placement('describe-build-worksets'), 'no-tools': placement('describe-build-worksets'), 'retry': placement('describe-decide-tools'), 'salvage': placement('describe-decide-tools-salvage') }, display('describe-decide-tools'))
      .node(placement('describe-decide-tools-salvage'), nodes.decideToolsSalvage, { 'done': placement('describe-build-worksets') }, display('describe-decide-tools-salvage'))
      // Build scatter worksets before dispatch.
      .node(placement('describe-build-worksets'), nodes.buildBookWorksets, {
        'ready': placement('describe-scatter'),
      }, display('describe-build-worksets'))
      // Tool-registry scatter: DagReference resolves body DAG from each item's dagName.
      // any-success reducer: 'success' → describe-pick, 'error' → compose-empty.
      // 'error' fires when all tool scouts return empty.
      .scatter(placement('describe-scatter'), 'bookWorksets', { 'dag': { 'from': 'item', 'path': 'dagIri', 'candidates': BOOK_SEARCH_TOOL_DAGS } }, {
        'success': placement('describe-gather'),
        'error':   placement('compose-empty'),
        'empty':   placement('compose-empty'),
      }, {
        'name': 'describe-scatter',
        'execution': { 'mode': 'item', 'concurrency': 4 },
        'reducer': 'any-success',
      })
  .gather(placement('describe-gather'), { [placement('describe-scatter')]: {} }, { 'strategy': 'tool-candidate-merge' }, {
    'success': placement('describe-pick'),
    'error': placement('compose-empty'),
    'empty': placement('compose-empty'),
  }, display('describe-gather'))
      .node(placement('describe-pick'),   nodes.pickBestMatch,    { 'picked': placement('describe-merge') }, display('describe-pick'))
      .node(placement('describe-merge'),  nodes.mergeCandidates,  { 'ranked': placement('describe-record'), 'empty': placement('compose-empty') }, display('describe-merge'))
      .node(placement('describe-record'), nodes.recordFindings,   { 'recorded': placement('describe-gate') }, display('describe-record'))
      .node(placement('describe-gate'),   nodes.hasCitationsGate, { 'pass': placement('describe-recall'), 'fail': placement('compose-empty') }, display('describe-gate'))
      .node(placement('describe-recall'), nodes.recallPastVisits, { 'recalled': placement('compose-loop') }, display('describe-recall'))

      // ── recommend-similar branch ─────────────────────────────────────────────
      // recommendSimilar seeds state.terms from prior-run shortlist memory.
      // 'seeded' routes to the book-search-scatter sub-DAG; third placement of
      // the same packaged cluster. 'empty' routes to the compose-empty terminal.
      .node(placement('recommend-similar'), nodes.recommendSimilar, {
        'seeded': placement('similar-search'),
        'empty':  placement('compose-empty'),
      }, display('recommend-similar'))

      // EmbeddedDAGNode: same book-search-scatter, third and final placement.
      .embed(placement('similar-search'), BOOK_SEARCH_SCATTER_DAG_IRI, {
        'success': placement('compose-loop'),
        'error':   placement('compose-empty'),
      }, {
        'name': 'similar-search',
        'outputs': {
          'terms':         'terms',
          'toolPlan':      'toolPlan',
          'candidates':    'candidates',
          'shortlist':     'shortlist',
          'priorContext':  'priorContext',
          'failureCause':  'failureCause',
        },
      })

      // ── compose-loop: shared compose/validate sub-DAG ──────────────────────────
      // All branches that successfully find candidates converge here.
      // composeResponse → validateResponse (retry loop, bounded by the retry budget on state (retriesFor('compose'))).
      // One sub-DAG definition serves all four convergent branches.
      // stateMapping.outputs copies the compose loop's writes back to the parent.
      //
      // Convergence policy: 'success' routes to the shared respond-to-visitor terminal
      // at the parent level; the sub-DAG produces state.draft and exits cleanly;
      // exactly ONE respond-to-visitor fires per run regardless of branch count.
      // 'error' (retry budget exhausted) falls through to compose-empty so the
      // visitor always receives an in-character response rather than a silent drop.
      .embed(placement('compose-loop'), COMPOSE_RETRY_LOOP_DAG_IRI, {
        'success': placement('respond-to-visitor'),
        'error':   placement('compose-empty'),
      }, {
        'name': 'compose-loop',
        'outputs': {
          'draft':         'draft',
          'approvalState': 'approvalState',
        },
      })

      // ── respond-to-visitor: single shared happy-path terminal ───────────────
      // Every branch that successfully composes a response converges here.
      // compose-loop (success) and both memory + empty-result paths all route
      // through this one placement. Convergence policy: exactly ONE respond-to-visitor
      // fires per run with the full converged state.draft in context. Success routes
      // to the canonical `end` TerminalNode rather than a bare null end-of-flow.
      .node(placement('respond-to-visitor'), nodes.respondToVisitor, { 'success': placement('end') }, display('respond-to-visitor'))

      // ── recall-memories branch ───────────────────────────────────────────────
      // No search needed; the memory store is queried directly.
      // recallMemories → composeMemoryResponse → respond-to-visitor (shared terminal).
      .node(placement('memory-recall'),          nodes.recallMemories,       { 'recalled': placement('compose-memory-recall') }, display('memory-recall'))
      .node(placement('compose-memory-recall'),  nodes.composeMemoryResponse, {
        'drafted': placement('respond-to-visitor'),
        'retry':   placement('compose-memory-recall'),
        'salvage': placement('compose-memory-salvage'),
      }, display('compose-memory-recall'))
      .node(placement('compose-memory-salvage'), nodes.composeMemoryResponseSalvage, { 'done': placement('respond-to-visitor') }, display('compose-memory-salvage'))

      // ── Terminal nodes ───────────────────────────────────────────────────────
      .node(placement('decline-off-topic'), nodes.declineOffTopic, { 'success': placement('end') }, display('decline-off-topic'))
      .node(placement('compose-empty'),     nodes.composeEmptyResponse,  {
        'drafted': placement('respond-to-visitor'),
        'retry':   placement('compose-empty'),
        'salvage': placement('compose-empty-salvage'),
      }, display('compose-empty'))
      .node(placement('compose-empty-salvage'), nodes.composeEmptyResponseSalvage, { 'done': placement('respond-to-visitor') }, display('compose-empty-salvage'))

      // Canonical end-of-flow: every completed path (a composed answer or an
      // off-topic decline) routes to this one `TerminalNode(completed)` instead of
      // a bare `null` route. The flow ends explicitly, not by absence of a route.
      .terminal(placement('end'), { outcome: 'completed', name: 'end' })

      .build();

---

registerDAG(dag, stateFactory?)

declare const dag: DAGType;
declare const stateFactory: ChildStateFactoryType;
const d = new Dagonizer<MyState>();
d.registerDAG(dag);
// With an explicit child-state factory:
d.registerDAG(dag, stateFactory);

Registers a DAG after a semantic validation pass, followed by an optional contract check. The optional stateFactory argument overrides the default child-state constructor thunk for embedded-DAG and scatter executions within this DAG; when omitted, ChildStateFactory.cloneParent is stored.

1. Semantic pass. Verifies every labeled entrypoint targets a placement IRI, every output route targets a placement IRI, node and DAG references resolve against the registry view, recursive DAG-reference components have a reachable terminal exit, and every registered node output has a routing entry in the placement's outputs map.

After the semantic pass, a data-flow check runs for each placement whose backing node declares required and produced state paths. Dangling reads (a non-entrypoint node requires a path no upstream node produces) and dead writes (a node produces a path no downstream node requires) both throw DAGError. This check is skipped for placements without a contract.

Schema validation is handled at the JSON ingest boundary (DAGDocument.load); registerDAG does not repeat the structural pass because DAGType already guarantees schema conformance.

Throws DAGError with a multi-line message listing all failures.

---

getDAG(iri)

Returns DAG | undefined for the exact registered DAG IRI. undefined when the DAG IRI has not been registered.

getNode(iri)

Returns NodeInterface<NodeStateInterface, string> | undefined for the exact registered node IRI. undefined when the node IRI has not been registered.

listDAGs()

Snapshot of every registered DAG. The returned array is a fresh shallow copy; mutating it does not affect the registry.

listNodes()

Snapshot of every registered node. The returned array is a fresh shallow copy; mutating it does not affect the registry.

All four read accessors in context:

// Read accessors: getDAG, getNode, listDAGs, listNodes.
// getDAG/getNode use exact registry keys; list accessors return fresh shallow copies.
const registeredDag  = dispatcher.getDAG('urn:noocodec:dag:chat'); // DAG | undefined
const registeredNode = dispatcher.getNode('classify');  // NodeInterface<...> | undefined
const allDags        = dispatcher.listDAGs();            // readonly DAG[]
const allNodes       = dispatcher.listNodes();           // readonly NodeInterface<...>[]
void registeredDag; void registeredNode; void allDags; void allNodes;

---

DAGDocument.load(json, options?)

// static load(json: string, options?: DAGDocumentLoadOptionsType): DAGType
declare const rawJsonString: string;
const dag = DAGDocument.load(rawJsonString);

Parse a JSON string and validate against DAGSchema. The single permitted ingest boundary where unknown enters the package. Throws ValidationError for malformed JSON or schema-noncompliant input.

declare const rawJsonString: string;
const dag = DAGDocument.load(rawJsonString);
const dispatcher = new Dagonizer<MyState>();
dispatcher.registerDAG(dag);

options.overrides — a Partial<DAGType> merged into the decoded document before schema validation. Use this to inject runtime configuration (e.g. concurrency limits from an environment config) without mutating the source JSON.

declare const rawJsonString: string;
declare const concurrency: number;
const dag = DAGDocument.load(rawJsonString, {
  overrides: {
    nodes: JSON.parse(rawJsonString).nodes.map((n: { '@id': string }) =>
      n['@id'] === 'urn:noocodec:dag:my-dag/node/scatter'
        ? { ...n, execution: { mode: 'item', concurrency } }
        : n
    ),
  },
});

---

DAGDocument.serialize(dag)

// static serialize(dag: DAGType): string
declare const dag: DAGType;
const json: string = DAGDocument.serialize(dag);

Serialize a DAG to pretty JSON (2-space indent). Does not re-validate.

---

execute(dagIri, initialState, options?)

const dispatcher = new Dagonizer<MyState>();
declare const state: MyState;
const execution: Execution<MyState> = dispatcher.execute('urn:noocodec:dag:my-flow', state);

Returns an Execution<TState> starting at the DAG's entrypoint. The first argument is the registered DAG IRI. The execution is lazy: the generator does not run until the caller awaits or iterates.

// ── Dispatcher ───────────────────────────────────────────────────────────
// ObservedDag: generic Dagonizer subclass wiring every lifecycle hook to an
// injected logger. The driver passes the top-level `logger` so both streams
// share one console sink; `dispatcher.logger` reads it back for convenience.
const dispatcher = new ObservedDag<ArchivistState>(logger);

// ── Tool registry (molecular pattern) ────────────────────────────────────
// Register each book-search tool as an embeddable tool DAG IRI.
// ToolRegistry.bundle() returns the synthesized nodes + DAGs so the
// dispatcher resolves `urn:noocodec:tool:*` IRIs at scatter time.
// Register BEFORE bookSearchScatterDAG so the embedded-DAG references
// from the scatter body are resolvable when the parent DAG is validated.
const toolRegistry = new ToolRegistry();
toolRegistry.register(new OpenLibrarySearchTool());
toolRegistry.register(new GoogleBooksTool());
toolRegistry.register(new SubjectSearchTool());
toolRegistry.register(new WikipediaSummaryTool());
dispatcher.registerBundle(toolRegistry.bundle());

// ── Bundle registration (molecular pattern) ──────────────────────────────
// Each bundle packages its nodes + DAG. Embedded-DAG bundles register first
// so the parent's semantic validator can resolve embedded references by name.
// Construct every services-injected node exactly once; the shared set is
// passed to all three registrations so duplicate registrations refer to identical
// instances and the registrar accepts them.
const nodes = ArchivistNodes.build(services);
dispatcher.registerBundle({ 'nodes': nodes.bookSearchScatterNodes, 'dags': [bookSearchScatterDAG] });
dispatcher.registerBundle({ 'nodes': nodes.composeRetryLoopNodes, 'dags': [composeRetryLoopDAG] });
dispatcher.registerBundle({ 'nodes': nodes.parentNodes, 'dags': [archivistDAG] });

// ── Demo run via ArchivistRunner + OnceTrigger ────────────────────────────
// ArchivistRunner encapsulates the canonical register→seed→execute→project
// loop. The dispatcher is the already-configured ObservedDag whose lifecycle
// hooks log every node boundary without any manual iteration here.
//
// Query source: first CLI argument, or the bundled demo question when absent.
// Override:   npx tsx examples/the-archivist/runArchivist.ts "your question"
const DEMO_QUERY = "I'm looking for a book about a strange house and a library";
const visitorQuery = process.argv[2] ?? DEMO_QUERY;

const runnerOptions: DagRunnerOptionsType<ArchivistState> = { 'dispatcher': dispatcher };
const archivistRunner = new ArchivistRunner(runnerOptions);

const onceTrigger = new OnceTrigger<ArchivistInput, ArchivistState, ArchivistResult>(
  'the-archivist',
  { 'query': visitorQuery },
);

// DAGError (code NODE_TIMEOUT) fires when the dispatcher's per-node deadline elapses.
// DAGError (code EXECUTION_ERROR) wraps a node throw that was not a timeout.
// LlmError wraps adapter-level failures (rate limit, bad credentials, etc.).
// Distinguish by `.code` (Dagonizer's own error taxonomy is one class) so
// callers can log or retry at the right granularity.
try {
  await onceTrigger.attach(archivistRunner);
} catch (err) {
  if (err instanceof DAGError && err.code === 'NODE_TIMEOUT') {
    logger.warn(`node timed out: ${err.message}`);
    throw err;
  }
  if (err instanceof DAGError && err.code === 'EXECUTION_ERROR') {
    logger.warn(`execution failed: ${err.message}`);
    throw err;
  }
  if (err instanceof LlmError) {
    logger.warn(`llm error [${err.classification.reason}]: ${err.message}`);
    throw err;
  }
  throw err;
}

const result = onceTrigger.result;
if (result === null) throw new Error('OnceTrigger resolved with null result');

logger.result(`intent=${result.state.intent}`);
logger.result(`shortlist=${String(result.state.shortlist.length)}`);
logger.result(`draft=${result.state.draft}`);
logger.result(`lifecycle=${result.state.lifecycle.variant}`);
logger.result(`triples=${String(services.memory.size)} written`);

ExecuteOptionsType has two fields: signal?: AbortSignal and deadlineMs?: number.

---

resume(dagIri, state, fromStage, options?)

const dispatcher = new Dagonizer<MyState>();
declare const state: MyState;
const execution: Execution<MyState> = dispatcher.resume(
  'my-flow',
  state,
  'urn:noocodec:dag:my-flow/node/node-b',
);

Identical to execute() but begins at fromStage instead of the DAG's entrypoint. fromStage is a placement IRI. The caller is responsible for rehydrating state (typically via Checkpoint.load(raw).restoreState(CheckpointRestoreAdapter.wrap(fn))) before calling.

if (cancelResult.cursor !== null) {
  const store = new MemoryCheckpointStore();
  const ckpt = await Checkpoint.capture(ARCHIVIST_DAG_IRI, cancelResult, { 'stores': { 'memory': services.memory } });
  await ckpt.persist(store, `archivist:${cancelVisitor.query}`);

  const recalled = await Checkpoint.recall(store, `archivist:${cancelVisitor.query}`);
  if (recalled !== null) {
    const freshMemory = new MemoryStore();
    await recalled.restoreStores({ 'memory': freshMemory });
    const { dagName, state, cursor } = recalled.restoreState(
      CheckpointRestoreAdapter.wrap((snap) => ArchivistState.restore(snap)),
    );
    const resumeResult = await dispatcher.resume(dagName, state, cursor);
    logger.result(`resumed draft=${resumeResult.state.draft}`);
    logger.result(`resumed lifecycle=${resumeResult.state.lifecycle.variant}`);
    logger.result(`resumed memory triples=${String(freshMemory.size)}`);
  }
} else {
  logger.result('cancellation-run completed before cursor; no checkpoint needed');
}

---

destroy()

const dispatcher = new Dagonizer<MyState>();
await dispatcher.destroy();

Calls the optional destroy() method on every registered node, then clears all registries. Use to clean up connection pools or other resources held by nodes.

---

Observability hooks

Seven protected no-op methods. Subclass Dagonizer and override to attach metrics, logging, or tracing.

class ObservableDagonizer extends Dagonizer<MyState> {
  protected override onFlowStart(dagIri: string, state: MyState): void {
    console.log('start', dagIri);
  }
  protected override onFlowEnd(dagIri: string, state: MyState, result: ExecutionResultType<MyState>): void {
    console.log('end', dagIri, result.terminalOutcome);
  }
  protected override onNodeStart(nodeName: string, state: NodeStateInterface, placementPath: readonly string[]): void {}
  protected override onNodeEnd(nodeName: string, output: string | null, state: NodeStateInterface, placementPath: readonly string[]): void {}
  protected override onError(nodeName: string, error: Error, state: NodeStateInterface, placementPath: readonly string[]): void {}
  protected override onPhaseEnter(dagIri: string, phase: 'pre' | 'post', placementName: string, state: NodeStateInterface, placementPath: readonly string[]): void {}
  protected override onPhaseExit(dagIri: string, phase: 'pre' | 'post', placementName: string, state: NodeStateInterface, placementPath: readonly string[]): void {}
}

Hook	Fires
onFlowStart	After state.markRunning(), before the first node
onFlowEnd	After the final node (all paths: normal, cancelled, failed)
onNodeStart	Before node.execute() for each node entry point
onNodeEnd	After each node resolves, before the result is yielded; output is string | null (null = no route emitted)
onError	When the signal fires or a node throws
onPhaseEnter	Before a pre or post phase placement runs; signature (dagIri, phase: 'pre'|'post', placementName, state, placementPath)
onPhaseExit	After a pre or post phase placement completes (success or collected error); same signature as onPhaseEnter

placementPath is the ordered array of parent embedded-DAG placement labels leading to the current node. Top-level nodes receive []; a node inside an EmbeddedDAGNode labelled 'search' receives ['search']. Graph identity still comes from placement @id; the path is observability context for watchers and worker relays.

See Observability for usage examples. Contract misalignment — a dangling read or a dead write — throws a DAGError at registerDAG/build time rather than surfacing a warning.

---

Interface: DispatcherBundleType

declare const bundle: DispatcherBundleType<NodeStateInterface>;
const _nodes: readonly NodeInterface<NodeStateInterface, string>[] = bundle.nodes;
const _dags: readonly DAGType[] = bundle.dags;

A coherent unit of nodes and DAGs registered together. Plugin packages and feature modules export a DispatcherBundleType so applications register the whole unit in one call.

---

Const: SCATTER_PROGRESS_KEY

// SCATTER_PROGRESS_KEY === '__dagonizer_scatter_progress__'
const key = SCATTER_PROGRESS_KEY; // type: "__dagonizer_scatter_progress__"
export {};

Reserved metadata key used by the scatter executor to persist per-item resume bookkeeping. Application nodes must not write to this key. The stored value is a StoredScatterProgress map keyed by the scatter placement @id.

// Illustrative local shapes (the actual StoredScatterProgress is a discriminated union):
interface ScatterItemResult {
  readonly index:          number;
  readonly output:         string;
  readonly mappingValues?: Readonly<Record<string, unknown>>;
  readonly fieldValue?:    unknown;
}
interface ScatterProgress {
  readonly placementIri:     string;
  readonly completedIndices: readonly number[];
  readonly itemResults:      readonly ScatterItemResult[];
}
type StoredScatterProgress = Readonly<Record<string, ScatterProgress>>;

export {};

---

Const: WORKSET_PROGRESS_KEY

// WORKSET_PROGRESS_KEY === '__dagonizer_workset_progress__'
const key = WORKSET_PROGRESS_KEY; // type: "__dagonizer_workset_progress__"
export {};

Reserved metadata key used by the work-set scheduler to persist the in-flight work set on interruption. Application nodes must not write to this key. The stored value is a WorkSetProgress blob serialised by WorkSetCheckpoint.write and read back by WorkSetCheckpoint.read. Absent for size-1 canonical runs where the cursor model handles state directly.

---

Details for Nerds

Dagonizer intentionally keeps assembly explicit. JSON-LD carries DAG IRIs, placement IRIs, routes, contexts, state mappings, phases, scatter bodies, gather barriers, and embedded DAG references. Registries bind registered references to node implementations, DAG documents, child-state factories, containers, and channels. Visualization is generated from the DAG document, not from the live dispatcher.

That separation is why a DAG can be built with DAGBuilder, loaded from JSON-LD, packaged by a plugin, rendered as Mermaid, and executed by the same dispatcher without conversion.

Related Concepts

• Reference: Execution - what execute and resume return
• Reference: Contracts - NodeInterface, ExecuteOptionsType
• Reference: Core - GatherStrategies, OutcomeReducers
• Reference: Lifecycle
• DAGBuilder - author DAG documents before registering them
• Dependency Injection - pass services into node constructors before registration
• Observability - lifecycle hooks and structured run events