Dagonizer
Dagonizer

Example 20: Streaming Execution ​

What It Is ​

Streaming Execution lets an application observe node completions while the final DAG result is still pending. Dagonizer.execute() returns an Execution<TState> that can be awaited like a promise or consumed with for await.

The Cartographer runner uses this to update the live graph pane as each placement completes, then awaits the same execution for the final state. One call, one run, two consumption styles.

How It Works ​

execute() returns an Execution<TState> wrapper around one internal run. for await consumes node-completion stages from that run, while await resolves the cached final result from the same run. The caller can stream first, await later, or only await when no progress UI is needed.

This is a caller API, not a different DAG shape. The JSON-LD graph, registry, routes, cancellation behavior, and final lifecycle are the same whether the caller streams progress or simply awaits completion.

Diagrams, Examples, and Outputs ​

DAG registration and diagram ​

The application chooses streaming execution to observe each node result as it completes. The Cartographer uses this in the live DAG pane: dispatcher.execute() returns an awaitable async iterable, and the runner iterates stages to light up graph nodes while the final result is still pending.

Cartographer streaming execution DAG

6 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:cartographer",
  "@type": "DAG",
  "name": "dag:cartographer",
  "version": "1.0",
  "entrypoints": {
    "position-ping": "urn:noocodec:dag:cartographer/node/intake-gather",
    "facility-scan": "urn:noocodec:dag:cartographer/node/intake-gather",
    "sensor-reading": "urn:noocodec:dag:cartographer/node/intake-gather",
    "customs-event": "urn:noocodec:dag:cartographer/node/intake-gather",
    "delivery-confirmation": "urn:noocodec:dag:cartographer/node/intake-gather"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:cartographer/node/intake-gather",
      "@type": "GatherNode",
      "name": "dag:cartographer/node/intake-gather",
      "sources": {
        "urn:noocodec:dag:cartographer/entrypoint/position-ping": {},
        "urn:noocodec:dag:cartographer/entrypoint/facility-scan": {},
        "urn:noocodec:dag:cartographer/entrypoint/sensor-reading": {},
        "urn:noocodec:dag:cartographer/entrypoint/customs-event": {},
        "urn:noocodec:dag:cartographer/entrypoint/delivery-confirmation": {}
      },
      "gather": {
        "strategy": "source-intake"
      },
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/process-stream",
        "error": "urn:noocodec:dag:cartographer/node/process-stream",
        "empty": "urn:noocodec:dag:cartographer/node/failed"
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/process-stream",
      "@type": "ScatterNode",
      "name": "dag:cartographer/node/process-stream",
      "source": "sources",
      "body": {
        "dag": "urn:noocodec:dag:stream-event"
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:cartographer/node/fold-insights",
        "partial": "urn:noocodec:dag:cartographer/node/fold-insights",
        "all-error": "urn:noocodec:dag:cartographer/node/fold-insights",
        "empty": "urn:noocodec:dag:cartographer/node/summarize-insights"
      },
      "itemKey": "source-payload",
      "reducer": "aggregate",
      "container": "cpu",
      "execution": {
        "mode": "reservoir",
        "concurrency": 16,
        "reservoir": {
          "keyField": "eventType",
          "capacity": 1000
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/fold-insights",
      "@type": "GatherNode",
      "name": "dag:cartographer/node/fold-insights",
      "sources": {
        "urn:noocodec:dag:cartographer/node/process-stream": {}
      },
      "gather": {
        "strategy": "insights-fold"
      },
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/summarize-insights",
        "error": "urn:noocodec:dag:cartographer/node/summarize-insights",
        "empty": "urn:noocodec:dag:cartographer/node/summarize-insights"
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/summarize-insights",
      "@type": "EmbeddedDAGNode",
      "name": "dag:cartographer/node/summarize-insights",
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/done",
        "error": "urn:noocodec:dag:cartographer/node/failed"
      },
      "dag": "urn:noocodec:dag:insights-summary",
      "container": "io"
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/done",
      "@type": "TerminalNode",
      "name": "dag:cartographer/node/done",
      "outcome": "completed"
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/failed",
      "@type": "TerminalNode",
      "name": "dag:cartographer/node/failed",
      "outcome": "failed"
    }
  ]
}
Mermaid generated from the same DAG
Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:cartographer (v1.0)
  entry_position-ping(["position-ping"])
  entry_position-ping --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_facility-scan(["facility-scan"])
  entry_facility-scan --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_sensor-reading(["sensor-reading"])
  entry_sensor-reading --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_customs-event(["customs-event"])
  entry_customs-event --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_delivery-confirmation(["delivery-confirmation"])
  entry_delivery-confirmation --> urn_noocodec_dag_cartographer/node/intake-gather
  urn_noocodec_dag_cartographer/node/intake-gather{"dag_cartographer/node/intake-gather"}
  urn_noocodec_dag_cartographer/node/intake-gather -->|success| urn_noocodec_dag_cartographer/node/process-stream
  urn_noocodec_dag_cartographer/node/intake-gather -->|error| urn_noocodec_dag_cartographer/node/process-stream
  urn_noocodec_dag_cartographer/node/intake-gather -->|empty| urn_noocodec_dag_cartographer/node/failed
  urn_noocodec_dag_cartographer/node/process-stream[/"dag:cartographer/node/process-stream ▣ eventType ×1000"/]
  urn_noocodec_dag_cartographer/node/process-stream -->|all-success| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|partial| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|all-error| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|empty| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/fold-insights{"dag_cartographer/node/fold-insights"}
  urn_noocodec_dag_cartographer/node/fold-insights -->|success| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/fold-insights -->|error| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/fold-insights -->|empty| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/summarize-insights[["dag:cartographer/node/summarize-insights"]]
  urn_noocodec_dag_cartographer/node/summarize-insights -->|success| urn_noocodec_dag_cartographer/node/done
  urn_noocodec_dag_cartographer/node/summarize-insights -->|error| urn_noocodec_dag_cartographer/node/failed
  urn_noocodec_dag_cartographer/node/done((("dag:cartographer/node/done")))
  urn_noocodec_dag_cartographer/node/failed>"dag:cartographer/node/failed"]
  classDef contained-cpu fill:#b45309,stroke:#d97706,color:#eef3f7
  class urn_noocodec_dag_cartographer/node/process-stream contained-cpu
  classDef contained-io fill:#7c3aed,stroke:#8b5cf6,color:#eef3f7
  class urn_noocodec_dag_cartographer/node/summarize-insights contained-io
  classDef reservoir fill:#1e3a5f,stroke:#3b82f6,color:#bfdbfe
  class urn_noocodec_dag_cartographer/node/process-stream reservoir

Dagonizer.execute() returns an Execution<TState> that is both:

Run ​

bash
npm run docs:dev

What It Lets You Do ​

Streaming execution lets applications observe DAG progress while the final result is still pending. Use it for live graph panes, progress bars, logs, server-sent events, and long-running browser or CLI flows where waiting for the final ExecutionResult hides useful intermediate state.

It pairs naturally with cancellation: a UI can show progress, keep a cancel button active, and still receive one final lifecycle result when the run completes or stops.

Code Samples ​

The runner snippet shows the for await loop that feeds the Cartographer graph UI. The DAG snippet is included to make the point explicit: streaming execution changes how the caller observes a run, not how the graph is authored.

vue
const execution = dispatcher.execute('cartographer', state, { 'signal': activeAbortController.signal });
for await (const stage of execution) {
  // Each yielded stage lights up a node via the observer hooks above.
  // Consume silently; the observer drives the UI.
  void stage;
}
await execution;
ts
/** Default reservoir capacity for the process-stream scatter in the workers DAG. */
export const DEFAULT_RESERVOIR_CAPACITY = 1000;

/**
 * CartographerWorkersDag: static factory for the cartographer-workers DAG and
 * its associated dispatcher bundle. Consumers call CartographerWorkersDag.build()
 * (DAG only) or CartographerWorkersDag.bundle() (full dispatcher bundle) with an
 * optional reservoir capacity override.
 *
 * DAG topology — identical to cartographerDAG with containerized boundaries:
 *   - container: 'cpu' so each stream-event body runs inside a
 *     WorkerThreadContainer/WebWorkerContainer rather than in-process.
 *   - container: 'io' so the final summary runs through the same embedded-DAG
 *     interface used by plugins and nested flows.
 *   - reservoir.capacity is parameterised; callers pass their UI-controlled
 *     batch size rather than relying on the compile-time default.
 *
 *   5 data-type entrypoints → gather('intake-gather', source-intake)
 *     → scatter('process-stream', 'sources', { dag: 'stream-event' },
 *               concurrency: 16, container: 'cpu', reservoir: { capacity })
 *     → gather('fold-insights', strategy: insights-fold)
 *     → embed('summarize-insights', 'insights-summary', container: 'io')
 *     → done
 */
export class CartographerWorkersDag {
  private constructor() { /* static-only */ }

  /**
   * Build the cartographer-workers DAG with the given reservoir capacity.
   * CLI, smoke tests, and dag-validate consumers use cartographerWorkersDAG
   * (the pre-built constant); the browser demo calls this with a UI-controlled value.
   */
  static build(capacity: number = DEFAULT_RESERVOIR_CAPACITY): DAGType {
    return new DAGBuilder(CARTOGRAPHER_DAG_IRI, '1.0')

      .gather(
        CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        CARTOGRAPHER_IRIS.intakeSources(CARTOGRAPHER_DAG_IRI),
        { 'strategy': 'source-intake' },
        {
          'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
          'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
          'empty':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'),
        },
      )

      .scatter(
        CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
        'sources',
        { 'dag': STREAM_EVENT_DAG_IRI },
        {
          'all-success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
          'partial':     CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
          'all-error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
          'empty':       CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
        },
        {
          'itemKey':     'source-payload',
          'container':   'cpu',
          'execution': { 'mode': 'reservoir', 'concurrency': 16, 'reservoir': { 'keyField': 'eventType', 'capacity': capacity } },
        },
      )
      .gather(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'), {
        [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream')]: {},
      }, { 'strategy': 'insights-fold' }, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
        'empty':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
      })

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'), INSIGHTS_SUMMARY_DAG_IRI, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'),
      }, {
        'container': 'io',
      })

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

      .entrypoints({
        'position-ping':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'facility-scan':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'sensor-reading':         CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'customs-event':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'delivery-confirmation':  CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
      })

      .build();
  }

  /**
   * Build the workers bundle with a configurable reservoir capacity. The returned
   * bundle is identical to cartographerWorkersBundle except that its cartographer
   * DAG is built with CartographerWorkersDag.build(capacity) so the process-stream
   * scatter uses the caller-supplied batch size.
   *
   * Used by the browser demo to wire UI-controlled knobs into each run() without
   * mutating the shared default-capacity constants.
   */
  static bundle(
    capacity: number = DEFAULT_RESERVOIR_CAPACITY,
  ): DispatcherBundleType<CartographerState> {
    return {
      'nodes': [
        ...cartographerWorkerRuntimeBundle.nodes,
      ],
      'dags': [
        ...cartographerWorkerRuntimeBundle.dags,
        CartographerWorkersDag.build(capacity),
      ],
    };
  }
}

/**
 * cartographerWorkersDAG: pre-built workers DAG at DEFAULT_RESERVOIR_CAPACITY.
 * CLI, smoke tests, and dag-validate consumers use this constant; the browser
 * demo uses CartographerWorkersDag.build(capacity) with a UI-controlled value.
 */
export const cartographerWorkersDAG: DAGType = CartographerWorkersDag.build();

Details for Nerds ​

  • Execution<TState> dual interface. execute() returns an object that satisfies both Promise<ExecutionResultType<TState>> and AsyncIterable<NodeResultType<NodeStateInterface>>. No separate streaming method needed.
  • Awaitable — await dispatcher.execute(...) waits for the final summary (ExecutionResultType<TState>).
  • AsyncIterable — for await (const stage of dispatcher.execute(...)) yields a NodeResultType<NodeStateInterface> for each node as it completes. The base type is used because embedded-child nodes may carry different concrete state shapes.

The two consumption modes share a single internal generator. Iterating and then awaiting returns the cached final result; the flow body runs exactly once.

  • Per-node NodeResultType<NodeStateInterface>. Each yielded value carries nodeName, output, state (typed as the base NodeStateInterface so embedded-child nodes with different concrete states are covered), and the node's own lifecycle snapshot. Downcast stage.state locally when you need concrete state access.
  • Single pass. The internal generator runs once. Awaiting after iteration returns the same resolved value; the flow does not re-execute.
  • Compose with cancellation. Pass signal in the execute options to cancel the stream mid-flight; any in-flight node resolves or throws, and the async iterator drains cleanly.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky