Authoring DAGs

What It Is

Dagonizer has one workflow artifact: a schema-validated JSON-LD DAG document. You can create it with DAGBuilder, load it from serialized JSON-LD, or receive it from a plugin bundle. After that, it is the same object: the dispatcher registers it by expanded DAG IRI, the visualizers render it, and validators check it.

That makes authoring a topology decision, not a framework fork. Choose the most convenient source form for your application, then let everything converge on the canonical DAG document.

How It Works

Authoring produces one object: a schema-valid DAG. DAGBuilder is the typed factory for that object; JSON loading is the ingest path for the same object; plugin registration installs the same object into the registry. The dispatcher does not care which authoring path produced it.

The DAG type is the API. A DAG is a JSON-LD 1.1 document with @context, @id, @type, and a nodes array of placement objects. The DAG @id is the identity the registry expands and stores. Each placement @id is the routing target inside the graph. The name fields stay useful for diagrams, logs, and watchers in the deep, but they are not the source of execution identity.

                ┌──────────────────────────────────────┐
                │    DAG (JSON-LD canonical)           │   The single API
                │    @context / @id / @type / nodes    │   stable across versions
                │    DAGSchema-validated               │   dispatcher-consumed
                └──────────────────────────────────────┘
                          ▲
                          │
              ┌───────────┴────────┐
              │     DAGBuilder     │
              │                    │
              │  Code factory for  │
              │  JSON-LD DAG       │
              │  documents         │
              └────────────────────┘

Diagrams, Examples, and Outputs

Example 02 builds a small chat DAG in TypeScript and registers the resulting JSON-LD object. The code and diagram below are generated from the same runnable source file:

export const dag = new DAGBuilder(chatDAGIri, '1')
  // First .node() call → entrypoint is set to 'classify' automatically.
  .node(placement(chatDAGIri, 'classify'), new ClassifyNode(), {
    "on_topic": placement(chatDAGIri, 'respond'),
    "off_topic": placement(chatDAGIri, 'respond'),
  })
  // routes for 'respond' must cover exactly { success }, no more, no less.
  .node(placement(chatDAGIri, 'respond'), new RespondNode(), { "success": placement(chatDAGIri, 'end') })
  .terminal(placement(chatDAGIri, 'end'))
  .build();  // materialises the canonical JSON-LD DAG document

Example 02 builder DAG

3 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:chat",
  "@type": "DAG",
  "name": "dag:chat",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:chat/node/classify"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:chat/node/classify",
      "@type": "SingleNode",
      "name": "dag:chat/node/classify",
      "node": "urn:noocodec:node:classify",
      "outputs": {
        "on_topic": "urn:noocodec:dag:chat/node/respond",
        "off_topic": "urn:noocodec:dag:chat/node/respond"
      }
    },
    {
      "@id": "urn:noocodec:dag:chat/node/respond",
      "@type": "SingleNode",
      "name": "dag:chat/node/respond",
      "node": "urn:noocodec:node:respond",
      "outputs": {
        "success": "urn:noocodec:dag:chat/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:chat/node/end",
      "@type": "TerminalNode",
      "name": "dag:chat/node/end",
      "outcome": "completed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:chat (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_chat/node/classify
  urn_noocodec_dag_chat/node/classify["dag:chat/node/classify"]
  urn_noocodec_dag_chat/node/classify -->|on_topic| urn_noocodec_dag_chat/node/respond
  urn_noocodec_dag_chat/node/classify -->|off_topic| urn_noocodec_dag_chat/node/respond
  urn_noocodec_dag_chat/node/respond["dag:chat/node/respond"]
  urn_noocodec_dag_chat/node/respond -->|success| urn_noocodec_dag_chat/node/end
  urn_noocodec_dag_chat/node/end((("dag:chat/node/end")))

Use the runnable pages for execution output:

• Example 02: DAGBuilder shows the builder source, JSON-LD DAG, Mermaid view, and run output.
• Example 03: Tool Schemas loads a DAG from a JSON-LD string and round-trips it through validation.
• The Archivist is a full browser demo whose parent DAG and embedded DAGs are authored with DAGBuilder.
• The Cartographer shows DAG authoring across worker-bound scatter, embedded compliance checks, and pipeline delivery flows.

What It Lets You Do

Use when

Use this guide when deciding whether to author a DAG in TypeScript with DAGBuilder, load a serialized JSON-LD document, or package a child flow for embedding or plugin registration. The application decision is not "builder versus JSON-LD runtime"; both produce the same canonical DAG document.

Code Samples

Error-routing contract

Nodes never throw past the node boundary. An error condition is a flow decision: the node returns the failed items on an 'error' routed sub-batch and the DAG routes that output to a recovery node or an error terminal. The engine does not intercept throws and reroute them.

This means every node that can fail must:

1. Declare 'error' (or a domain-specific name like 'salvage') as one of its output ports.
2. Return a routed sub-batch on 'error' when the failure condition is met.
3. Have that output wired to a downstream placement in the DAG.

The Cartographer demonstrates this contract in runnable code. route-redaction does not throw when redaction is unnecessary; it routes either to needs-redaction or skip-redaction and lets the DAG decide which path runs:

export class RouteRedactionNode extends MonadicNode<CartographerState, 'needs-redaction' | 'skip-redaction'> {
  readonly '@id' = 'urn:noocodec:node:route-redaction';
  readonly 'name' = 'route-redaction';
  readonly 'outputs' = ['needs-redaction', 'skip-redaction'] as const;

  override get outputSchema(): Record<'needs-redaction' | 'skip-redaction', SchemaObjectType> {
    return {
      'needs-redaction': { 'type': 'object' },
      'skip-redaction':  { 'type': 'object' },
    };
  }

  override async execute(
    batch: Batch<CartographerState>,
    _context: NodeContextType,
  ): Promise<RoutedBatchType<'needs-redaction' | 'skip-redaction', CartographerState>> {
    const acc = new Map<'needs-redaction' | 'skip-redaction', ItemType<CartographerState>[]>();

    for (const item of batch) {
      const result = this.routeItem(item.state);
      for (const error of result.errors) {
        item.state.collectError(error);
      }
      const bucket = acc.get(result.output);
      if (bucket === undefined) {
        acc.set(result.output, [item]);
      } else {
        bucket.push(item);
      }
    }

    const routed = new Map<'needs-redaction' | 'skip-redaction', Batch<CartographerState>>();
    for (const [output, items] of acc) {
      routed.set(output, Batch.from(items));
    }
    return routed;
  }

  private routeItem(state: CartographerState): NodeOutputType<'needs-redaction' | 'skip-redaction'> {
    const ev = state.currentEvent;
    const hasPii =
      state.canonical.pii === true ||
      ev.recipientName.length > 0 ||
      ev.recipientEmail.length > 0;
    const alreadyHandled = state.canonical.consentHandled === true;

    const consentStatus = Consent.statusFor(ev.shipmentId, ev.marketingConsent);
    const juris = state.geoContext.jurisdiction;
    const lightRegime = juris === 'baseline' || juris === 'international-waters';
    // Light regime + valid consent imposes no redaction obligation.
    const notRequired = lightRegime && consentStatus === 'valid';

    const skip = !hasPii || alreadyHandled || notRequired;

    if (skip) {
      state.routing = { ...state.routing, 'redactionSkipped': true, 'redactionRun': false };
      // Set a minimal no-op GdprResult: redaction NOT applied, precise coords
      // retained. Marketing analytics eligibility still tracks valid consent.
      state.gdprResult = {
        ...state.gdprResult,
        'consentStatus':              consentStatus,
        'lawfulBasis':                state.raw.lawfulBasis,
        'jurisdiction':               state.geoContext.jurisdiction,
        'redactionApplied':           false,
        'coordsCoarsened':            false,
        'marketingAnalyticsEligible': consentStatus === 'valid',
      };
      return NodeOutput.create('skip-redaction');
    }
    state.routing = { ...state.routing, 'redactionRun': true, 'redactionSkipped': false };
    return NodeOutput.create('needs-redaction');
  }
}

export const routeRedaction = new RouteRedactionNode();

The reusable gdpr-compliance sub-DAG makes terminal outcome part of topology: compliant is a completed terminal and violation is a failed terminal. Parent placements route the embedded DAG's success and error outputs explicitly:

import { consentGate, classifyPii, redactPii } from '../nodes/gdprNodes.ts';
import { CARTOGRAPHER_IRIS } from '../cartographerIds.ts';
import type { CartographerState } from '../CartographerState.ts';

import type { DAGType, DispatcherBundleType } from '@studnicky/dagonizer';
import { DAGBuilder } from '@studnicky/dagonizer';

const GDPR_COMPLIANCE_DAG_IRI = CARTOGRAPHER_IRIS.dag.gdprCompliance;

export const gdprComplianceDAG: DAGType = new DAGBuilder(GDPR_COMPLIANCE_DAG_IRI, '1.0')

  // ── 1. consent-gate ──────────────────────────────────────────────────────
  // Resolves the consent status from marketingConsent + simulated expiry.
  // Always routes 'classify' (both consented and non-consented proceed;
  // the consent status drives redaction rules downstream).
  .node(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'consent-gate'), consentGate, {
    'classify': CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'classify-pii'),
  })

  // ── 2. classify-pii ──────────────────────────────────────────────────────
  // Records which fields are personal/sensitive; no routing decision yet.
  .node(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'classify-pii'), classifyPii, {
    'redact': CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'redact-pii'),
  })

  // ── 3. redact-pii ────────────────────────────────────────────────────────
  // Applies GdprRedactor.redact. Routes to 'compliant' (ok) or 'violation'.
  .node(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'redact-pii'), redactPii, {
    'ok':        CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'compliant'),
    'violation': CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'violation'),
  })

  // ── Terminals ─────────────────────────────────────────────────────────────
  .terminal(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'compliant'), { outcome: 'completed' })
  .terminal(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'violation'), { outcome: 'failed' })

  .build();

/**
 * event-pipeline-typed: the live enrichment scatter body for process-events.
 *
 * Reads the scattered CanonicalEventVariant from metadata key 'canonical-event'
 * and routes it to one of five per-type embedded DAGs via route-event-type-variant.
 * Each per-type DAG starts with parse-variant (which also reads from metadata),
 * embeds geo-pipeline for geo resolution, runs type-specific enrichment nodes,
 * and converges on aggregate-event → done.
 *
 *   route-event-type-variant
 *     ├─position-ping──────────► pipeline-position-ping (embedded)
 *     ├─sensor-reading─────────► pipeline-sensor-reading (embedded)
 *     ├─customs-event──────────► pipeline-customs-event (embedded)
 *     ├─facility-scan──────────► pipeline-facility-scan (embedded)
 *     └─delivery-confirmation──► pipeline-delivery-confirmation (embedded)
 *   Each per-type DAG:
 *     parse-variant → geo-pipeline → canonicalize-core → [type-specific] → aggregate-event → done
 *
 * Metadata propagation: the scatter sets 'canonical-event' on each clone's
 * metadata. NodeStateBase.clone() copies _metadata, so metadata propagates
 * to embedded child clones. Both route-event-type-variant and parse-variant
 * read 'canonical-event' from metadata.
 */
export const eventPipelineTypedDAG: DAGType = new DAGBuilder(EVENT_PIPELINE_TYPED_DAG_IRI, '1.0')

  // 1. route-event-type-variant: read eventType from 'canonical-event' metadata
  //    and dispatch to the corresponding per-type sub-DAG.
  .node(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'route-event-type-variant'), routeEventType, {
    'position-ping':         CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-position-ping'),
    'sensor-reading':        CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-sensor-reading'),
    'customs-event':         CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-customs-event'),
    'facility-scan':         CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-facility-scan'),
    'delivery-confirmation': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-delivery-confirmation'),
  })

  // 2a. pipeline-position-ping: geo + leg measurement.
  .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-position-ping'), CARTOGRAPHER_IRIS.dag.pipelinePositionPing, {
    'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
    'error':   CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
  }, {
    'outputs': {
      'canonicalVariant': 'canonicalVariant',
      'raw':              'raw',
      'normalized':       'normalized',
      'currentEvent':     'currentEvent',
      'geoContext':       'geoContext',
      'resolvedGeo':      'resolvedGeo',
      'legKm':            'legKm',
      'routing':          'routing',
      'enriched':         'enriched',
      'capturedErrors':   'capturedErrors',
    },
  })

  // 2b. pipeline-sensor-reading: geo + cold-chain + leg measurement.
  .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-sensor-reading'), CARTOGRAPHER_IRIS.dag.pipelineSensorReading, {
    'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
    'error':   CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
  }, {
    'outputs': {
      'canonicalVariant': 'canonicalVariant',
      'raw':              'raw',
      'normalized':       'normalized',
      'currentEvent':     'currentEvent',
      'geoContext':       'geoContext',
      'resolvedGeo':      'resolvedGeo',
      'coldChainBreach':  'coldChainBreach',
      'legKm':            'legKm',
      'routing':          'routing',
      'enriched':         'enriched',
      'capturedErrors':   'capturedErrors',
    },
  })

  // 2c. pipeline-customs-event: geo + customs-dwell + leg measurement.
  .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-customs-event'), CARTOGRAPHER_IRIS.dag.pipelineCustomsEvent, {
    'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
    'error':   CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
  }, {
    'outputs': {
      'canonicalVariant':  'canonicalVariant',
      'raw':               'raw',
      'normalized':        'normalized',
      'currentEvent':      'currentEvent',
      'geoContext':        'geoContext',
      'resolvedGeo':       'resolvedGeo',
      'customsDwellHours': 'customsDwellHours',
      'legKm':             'legKm',
      'routing':           'routing',
      'enriched':          'enriched',
      'capturedErrors':   'capturedErrors',
    },
  })

  // 2d. pipeline-facility-scan: geo + facility canonicalization + order enrichment
  //     + GDPR-gated redaction.
  .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-facility-scan'), CARTOGRAPHER_IRIS.dag.pipelineFacilityScan, {
    'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
    'error':   CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
  }, {
    'outputs': {
      'canonicalVariant':  'canonicalVariant',
      'raw':               'raw',
      'normalized':        'normalized',
      'currentEvent':      'currentEvent',
      'geoContext':        'geoContext',
      'resolvedGeo':       'resolvedGeo',
      'pricedOrder':       'pricedOrder',
      'shippingQuote':     'shippingQuote',
      'deliveryEstimate':  'deliveryEstimate',
      'legKm':             'legKm',
      'gdprResult':        'gdprResult',
      'routing':           'routing',
      'enriched':          'enriched',
      'capturedErrors':   'capturedErrors',
    },
  })

  // 2e. pipeline-delivery-confirmation: geo + recipient canonicalization +
  //     delivery confirmation + GDPR-gated redaction.
  .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-delivery-confirmation'), CARTOGRAPHER_IRIS.dag.pipelineDeliveryConfirmation, {
    'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
    'error':   CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
  }, {
    'outputs': {
      'canonicalVariant': 'canonicalVariant',
      'raw':              'raw',
      'normalized':       'normalized',
      'currentEvent':     'currentEvent',
      'geoContext':       'geoContext',
      'resolvedGeo':      'resolvedGeo',
      'legKm':            'legKm',
      'gdprResult':       'gdprResult',
      'routing':          'routing',
      'enriched':         'enriched',
      'capturedErrors':   'capturedErrors',
    },
  })

  .terminal(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),     { outcome: 'completed' })
  .terminal(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'), { outcome: 'failed' })

  .build();

If a node truly throws (an unexpected bug, not a handled error condition), the exception propagates as an engine-level failure and the lifecycle transitions to failed. This is distinct from routing to an 'error' port, which is a deliberate flow decision the DAG topology controls.

Details for Nerds

DAGBuilder Is The Code Factory

DAGBuilder is the factory for DAG documents in TypeScript. ETL pipelines, transformation chains, agent loops, embedded DAGs, scatter bodies, tool DAGs, plugin DAGs, dynamic DAG references, and fixed sequences use the same fluent surface.

The mental model: first this, then that, then if X go here else go there. TypeScript narrows the route map at each .node() call from the node's TOutput union, so misspelled routes are compile errors before the DAG ever runs.

The builder example registers two nodes and chains them:

export const dag = new DAGBuilder(chatDAGIri, '1')
  // First .node() call → entrypoint is set to 'classify' automatically.
  .node(placement(chatDAGIri, 'classify'), new ClassifyNode(), {
    "on_topic": placement(chatDAGIri, 'respond'),
    "off_topic": placement(chatDAGIri, 'respond'),
  })
  // routes for 'respond' must cover exactly { success }, no more, no less.
  .node(placement(chatDAGIri, 'respond'), new RespondNode(), { "success": placement(chatDAGIri, 'end') })
  .terminal(placement(chatDAGIri, 'end'))
  .build();  // materialises the canonical JSON-LD DAG document

Use DAGBuilder because:

• A new stage is one more .node() link in the chain.
• Routes stay on the page next to the node reference.
• The TypeScript compiler verifies every output is wired.
• .build() returns the canonical JSON-LD DAG document the dispatcher consumes.

JSON-LD Documents

// DAGDocument.load() throws DAGError (code VALIDATION_ERROR) if JSON is malformed or schema fails.
export const dag = DAGDocument.load(dagJson);

Serialized DAGs are JSON-LD documents. Load them with DAGDocument.load(json) at process boundaries and persist them with DAGDocument.serialize(dag). That is transport and storage for the same DAG object, not a second framework abstraction.

Node implementations sit beside authoring

Authoring decides topology; node implementations carry the work. NodeInterface<TState, TOutput> is the contract every node satisfies. The classify-intent node from the Archivist demo declares a seven-value TOutput union and routes via switch:

import type { ArchivistState } from '../ArchivistState.ts';
import type { ArchivistServices, ClassifiedIntent } from '../services.ts';

import { Batch, MonadicNode, NodeOutput, ReasoningStep, RoutedBatch } from '@studnicky/dagonizer';
import type { ItemType, NodeContextType, SchemaObjectType } from '@studnicky/dagonizer';
import { Signal } from '@studnicky/signal';

type IntentOutput =
  | 'lookup-author'
  | 'find-reviews'
  | 'describe-book'
  | 'recommend-similar'
  | 'recall-memories'
  | 'on-topic'
  | 'recommend-top-rated'
  | 'off-topic'
  | 'retry'
  | 'salvage';

/** Per-node timeout: generous for Gemini Nano's constrained-output path (20-60 s typical). */
const NODE_TIMEOUT_MS = 30_000;

/** Total attempts (initial + retries) before routing to salvage. */
const RETRY_BUDGET = 2;

export class ClassifyIntentNode extends MonadicNode<ArchivistState, IntentOutput> {
  private readonly services: ArchivistServices;
  readonly name = 'classify-intent';
  readonly '@id' = 'urn:noocodec:node:classify-intent';
  constructor(services: ArchivistServices) {
    super();
    this.services = services;
  }
  readonly outputs = ['lookup-author', 'find-reviews', 'describe-book', 'recommend-similar', 'recall-memories', 'on-topic', 'recommend-top-rated', 'off-topic', 'retry', 'salvage'] as const;
  override get outputSchema(): Record<'lookup-author' | 'find-reviews' | 'describe-book' | 'recommend-similar' | 'recall-memories' | 'on-topic' | 'recommend-top-rated' | 'off-topic' | 'retry' | 'salvage', SchemaObjectType> {
    return {
      'lookup-author':       { 'type': 'object' },
      'find-reviews':        { 'type': 'object' },
      'describe-book':       { 'type': 'object' },
      'recommend-similar':   { 'type': 'object' },
      'recall-memories':     { 'type': 'object' },
      'on-topic':            { 'type': 'object' },
      'recommend-top-rated': { 'type': 'object' },
      'off-topic':           { 'type': 'object' },
      'retry':               { 'type': 'object' },
      'salvage':             { 'type': 'object' },
    };
  }

  override async execute(batch: Batch<ArchivistState>, context: NodeContextType) {
    const buckets = new Map<IntentOutput, ItemType<ArchivistState>[]>();
    for (const output of this.outputs) buckets.set(output, []);

    for (const item of batch) {
      const { state } = item;
      const summary = state.recalledContext.summary.length > 0
        ? state.recalledContext.summary
        : undefined;
      const conversation = state.conversation.length > 0 ? state.conversation : undefined;

      const signal = Signal.compose({
        'deadlineMs': this.services.nodeTimeouts[context.nodeName] ?? NODE_TIMEOUT_MS,
        'signal':     context.signal,
      });

      try {
        const intent = await this.services.llm.classifyIntent(state.query, summary, conversation, signal);
        // Guard: empty or unrecognised intent is a classification failure; the
        // retry/salvage flow decides the path.
        if (intent.length === 0) {
          if (state.withinRetryBudget(context.nodeName, RETRY_BUDGET)) {
            const result = NodeOutput.create('retry');
            for (const error of result.errors) state.collectError(error);
            buckets.get(result.output)?.push(item);
          } else {
            state.clearAttempts(context.nodeName);
            const result = NodeOutput.create('salvage');
            for (const error of result.errors) state.collectError(error);
            buckets.get(result.output)?.push(item);
          }
          continue;
        }
        state.intent = intent;
        state.reasoning = [...state.reasoning, ReasoningStep.create({ 'kind': 'thought', 'text': `classified intent as '${intent}'` })];
        state.clearAttempts(context.nodeName);
        // Map every ClassifiedIntent variant to its node output port.
        // 'search', 'describe' are general on-topic intents that route through
        // the main pipeline (extract-query -> decide-tools -> ...). 'recommend' is
        // a vague "good book / good story" ask: it routes through the dedicated
        // rating-ranked branch instead of the LLM-relevance-ranked one.
        const intentDispatch: Record<ClassifiedIntent, IntentOutput> = {
          'off-topic':         'off-topic',
          'lookup-author':     'lookup-author',
          'find-reviews':      'find-reviews',
          'describe-book':     'describe-book',
          'recommend-similar': 'recommend-similar',
          'recall-memories':   'recall-memories',
          'search':            'on-topic',
          'describe':          'on-topic',
          'recommend':         'recommend-top-rated',
        };
        const result = NodeOutput.create(intentDispatch[intent]);
        for (const error of result.errors) state.collectError(error);
        buckets.get(result.output)?.push(item);
      } catch (err) {
        // External cancellation / run deadline propagates unchanged.
        if (context.signal.aborted) throw err;
        // Node-local timeout or LLM failure -> retry budget decides the flow. The
        // classifier never fabricates an intent it didn't receive.
        if (state.withinRetryBudget(context.nodeName, RETRY_BUDGET)) {
          const result = NodeOutput.create('retry');
          for (const error of result.errors) state.collectError(error);
          buckets.get(result.output)?.push(item);
        } else {
          state.clearAttempts(context.nodeName);
          const result = NodeOutput.create('salvage');
          for (const error of result.errors) state.collectError(error);
          buckets.get(result.output)?.push(item);
        }
      }
    }

    const routes: Array<readonly [IntentOutput, Batch<ArchivistState>]> = [];
    for (const output of this.outputs) {
      const items = buckets.get(output) ?? [];
      if (items.length > 0) routes.push([output, Batch.from(items)]);
    }
    return RoutedBatch.create(routes);
  }
}

The same classifyIntent implementation is registered with the dispatcher and referenced from a placement in the DAG. The placement @id decides where the implementation sits in topology; name is the label humans see while the watcher lanterns are lit.

Capability Matrix

DAGBuilder emits every placement shape the schema allows.

Capability	DAGBuilder
SingleNode placement	yes via .node()
ScatterNode placement	yes via .scatter()
GatherNode placement	yes via .gather()
Gather strategy (map / append / partition / custom / collect / discard)	yes via .gather(..., gatherConfig, ...)
Outcome reducer (aggregate / all-success / any-success / custom)	yes via options.reducer
Scatter body variant (node, literal dag, or dynamic DagReference)	yes via body argument
EmbeddedDAGNode placement	yes via .embed()
TerminalNode placement	yes via .terminal()
inputs (parent -> clone seed)	yes via options.inputs
Multi-port routing	yes via routes map
Compile-time route narrowing	yes from NodeInterface TOutput
Runtime-conditional topology	yes by conditionally adding placements before .build()
Recursive / trampoline flows	yes via DagReference over registered DAG IRIs

A node can still invoke the dispatcher directly when a host deliberately owns that trampoline, but the DAG-native composition surface is DagReference: the graph names its candidate DAG IRIs and the engine resolves the selected child at the invocation point.

Terminal placements

Every DAG branch must end at an explicit TerminalNode placement IRI. Declare one with .terminal(placementIri, options?):

export const dag1 = new DAGBuilder(completedDagIri, '1')
  .node(placement(completedDagIri, 'step-a'), new StepANode(), {
    'ok': placement(completedDagIri, 'end'),
  })
  .terminal(placement(completedDagIri, 'end'))  // outcome defaults to 'completed'
  .build();

.terminal(name, options?) emits a TerminalNode placement. When the engine reaches it, the flow ends with the declared outcome ('completed' by default). To mark a branch as failed, pass { outcome: 'failed' }:

export const dag2 = new DAGBuilder(explicitTerminalsDagIri, '1')
  .node(placement(explicitTerminalsDagIri, 'check'), new CheckNode(), {
    'pass': placement(explicitTerminalsDagIri, 'end-ok'),
    'fail': placement(explicitTerminalsDagIri, 'end-fail'),
  })
  .terminal(placement(explicitTerminalsDagIri, 'end-ok'))
  .terminal(placement(explicitTerminalsDagIri, 'end-fail'), { outcome: 'failed' })
  .build();

Terminals appear as discrete nodes in the visualisation. Use descriptive display names (end-ok, response-sent, workflow-failed) when the endpoint label carries meaning; route to the terminal placement IRI.

An EmbeddedDAGNode placement targets terminal placement IRIs directly. This is the idiomatic way to surface a child DAG's error as a failed lifecycle in the parent:

// Child DAG literal: routes 'done' to a TerminalNode (well-formed).
export const childDAG: DAGType = {
  '@context':  DAG_CONTEXT,
  '@id': childTerminalsDagIri,
  '@type':     'DAG',
  "name":      'child-for-terminals',
  "version":   '1',
  "entrypoints": { "main": placement(childTerminalsDagIri, 'child-work') },
  "nodes": [
    {
      '@id': 'urn:noocodec:dag:child-for-terminals/node/child-work',
      '@type':  'SingleNode',
      "name":   'child-work',
      "node":   'urn:noocodec:node:child-work',
      "outputs": { "done": placement(childTerminalsDagIri, 'child-end') },
    },
    {
      '@id': 'urn:noocodec:dag:child-for-terminals/node/child-end',
      '@type':   'TerminalNode',
      "name":    'child-end',
      "outcome": 'completed',
    },
  ],
};

export const dag4 = new DAGBuilder(embeddedTerminalsDagIri, '1')
  .embed<GateState, GateState>(placement(embeddedTerminalsDagIri, 'run'), childTerminalsDagIri, {
    'success': placement(embeddedTerminalsDagIri, 'end-ok'),
    'error':   placement(embeddedTerminalsDagIri, 'end-fail'),
  }, {
    // Seed the child's shouldPass from parent state before the child DAG runs.
    'inputs': { 'shouldPass': 'shouldPass' },
  })
  .terminal(placement(embeddedTerminalsDagIri, 'end-ok'))
  .terminal(placement(embeddedTerminalsDagIri, 'end-fail'), { outcome: 'failed' })
  .build();

See DAGBuilder, .terminal() and Example 09: Terminal Nodes for runnable examples.

Loading JSON-LD

Use DAGDocument.load(jsonString) to validate a serialized DAG at the ingest boundary; the engine refuses anything that does not match DAGSchema. See JSON-LD export and import for details.

Related Concepts

• DAGBuilder - chainable authoring API for deterministic workflows
• JSON-LD export and import - DAGDocument.serialize and DAGDocument.load
• Concepts - the DAG type itself and its placement vocabulary
• Example 02: DAGBuilder
• Reference, Dagonizer
• Reference, Entities