Dagonizer
Dagonizer

Observability ​

What It Is ​

Observability is how a host application projects DAG execution into logs, traces, metrics, progress streams, and UI state without changing the DAG document. The graph remains business flow; observers watch execution boundaries.

You can subclass Dagonizer and override protected on* hooks, or attach DispatcherObserverType records through the observers option. Both surfaces observe flow, node, phase, and error boundaries.

How It Works ​

Subclass Dagonizer and override protected on* hooks. The dispatcher calls those hooks around flow, phase, node, and error boundaries. Hook implementations can write to an event bus, tracer, logger, metrics sink, or browser state without changing the DAG document.

Protected on* hooks on Dagonizer fire at every execution boundary. Subclass the dispatcher and override whichever hooks you need. Class extension is the only extension mechanism; the dispatcher exposes no callback API.

Diagrams, Examples, and Outputs ​

The Dispatcher demo is the compact runnable for observing routing, handoff, and operational control. The graph below is the same support dispatcher DAG rendered on the example pages:

support-dispatcher observable DAG

7 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:support-dispatcher",
  "@type": "DAG",
  "name": "dag:support-dispatcher",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:support-dispatcher/node/classify-message"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/setup",
      "@type": "PhaseNode",
      "name": "dag:support-dispatcher/node/setup",
      "node": "urn:noocodec:node:dispatcher-setup",
      "phase": "pre"
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/classify-message",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/classify-message",
      "node": "urn:noocodec:node:classify-message",
      "outputs": {
        "routine": "urn:noocodec:dag:support-dispatcher/node/ai-compose",
        "escalate": "urn:noocodec:dag:support-dispatcher/node/park-for-operator",
        "off-topic": "urn:noocodec:dag:support-dispatcher/node/decline"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/ai-compose",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/ai-compose",
      "node": "urn:noocodec:node:ai-compose",
      "outputs": {
        "drafted": "urn:noocodec:dag:support-dispatcher/node/send-response"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/park-for-operator",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/park-for-operator",
      "node": "urn:noocodec:node:park-for-operator",
      "outputs": {
        "parked": "urn:noocodec:dag:support-dispatcher/node/end",
        "ready": "urn:noocodec:dag:support-dispatcher/node/send-response"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/send-response",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/send-response",
      "node": "urn:noocodec:node:send-response",
      "outputs": {
        "sent": "urn:noocodec:dag:support-dispatcher/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/decline",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/decline",
      "node": "urn:noocodec:node:decline",
      "outputs": {
        "declined": "urn:noocodec:dag:support-dispatcher/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/end",
      "@type": "TerminalNode",
      "name": "dag:support-dispatcher/node/end",
      "outcome": "completed"
    }
  ]
}
Mermaid generated from the same DAG
Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:support-dispatcher (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_support-dispatcher/node/classify-message
  urn_noocodec_dag_support-dispatcher/node/setup(["dag:support-dispatcher/node/setup (pre)"])
  urn_noocodec_dag_support-dispatcher/node/classify-message["dag:support-dispatcher/node/classify-message"]
  urn_noocodec_dag_support-dispatcher/node/classify-message -->|routine| urn_noocodec_dag_support-dispatcher/node/ai-compose
  urn_noocodec_dag_support-dispatcher/node/classify-message -->|escalate| urn_noocodec_dag_support-dispatcher/node/park-for-operator
  urn_noocodec_dag_support-dispatcher/node/classify-message -->|off-topic| urn_noocodec_dag_support-dispatcher/node/decline
  urn_noocodec_dag_support-dispatcher/node/ai-compose["dag:support-dispatcher/node/ai-compose"]
  urn_noocodec_dag_support-dispatcher/node/ai-compose -->|drafted| urn_noocodec_dag_support-dispatcher/node/send-response
  urn_noocodec_dag_support-dispatcher/node/park-for-operator["dag:support-dispatcher/node/park-for-operator"]
  urn_noocodec_dag_support-dispatcher/node/park-for-operator -->|parked| urn_noocodec_dag_support-dispatcher/node/end
  urn_noocodec_dag_support-dispatcher/node/park-for-operator -->|ready| urn_noocodec_dag_support-dispatcher/node/send-response
  urn_noocodec_dag_support-dispatcher/node/send-response["dag:support-dispatcher/node/send-response"]
  urn_noocodec_dag_support-dispatcher/node/send-response -->|sent| urn_noocodec_dag_support-dispatcher/node/end
  urn_noocodec_dag_support-dispatcher/node/decline["dag:support-dispatcher/node/decline"]
  urn_noocodec_dag_support-dispatcher/node/decline -->|declined| urn_noocodec_dag_support-dispatcher/node/end
  urn_noocodec_dag_support-dispatcher/node/end((("dag:support-dispatcher/node/end")))

What It Lets You Do ​

Use when ​

Use observability hooks when a host needs traces, metrics, progress events, audit logs, or UI updates from a DAG run. The DAG should stay about business flow; observers project execution state into product and operations surfaces.

Code Samples ​

The snippets below show the hook surface, subclass pattern, observer mux, EventBus publishing, SSE streaming, and timing integration.

API surface ​

SymbolSourceRole
Dagonizer.onFlowStart@studnicky/dagonizerFires after state.markRunning(), before the first node
Dagonizer.onFlowEnd@studnicky/dagonizerFires after the last node (including aborted or failed paths)
Dagonizer.onNodeStart@studnicky/dagonizerFires before each node.execute()
Dagonizer.onNodeEnd@studnicky/dagonizerFires after each node resolves, before yield
Dagonizer.onError@studnicky/dagonizerFires when a signal fires or a node throws
Dagonizer.onPhaseEnter@studnicky/dagonizerFires before a pre/post phase placement runs
Dagonizer.onPhaseExit@studnicky/dagonizerFires after a pre/post phase placement completes

Hook contracts ​

HookWhen calledArguments
onFlowStartAfter state.markRunning(), before the first nodedagName, state
onFlowEndAfter the last node (including aborted or failed paths)dagName, state, result
onNodeStartBefore node.execute() for each node entrynodeName, state, placementPath
onNodeEndAfter each node resolves, before yieldnodeName, output: string | null, state, placementPath
onErrorWhen a signal fires or a node throwsnodeName, error, state, placementPath
onPhaseEnterBefore a pre/post phase placement runsdagName, phase, placementName, state, placementPath
onPhaseExitAfter a pre/post phase placement completesdagName, phase, placementName, state, placementPath

onFlowEnd is always called, even when the flow fails or is cancelled. onError may fire before onFlowEnd in the same execution.

Hook argument names follow the public API (dagName, nodeName, placementName) because those values are optimized for logs, spans, and UI labels. They are observability labels, not routing keys. Use the DAG document @id and placement @id when you need execution identity.

For scatter and embedded-DAG nodes, onNodeStart and onNodeEnd fire once for the group entry (the containing scatter or embedded-dag placement), not once per constituent clone or inner node.

placementPath ​

placementPath is a required readonly string[] argument on onNodeStart, onNodeEnd, and onError. It is the ordered list of parent embedded-DAG placement labels that led to the current node:

  • Top-level node: []
  • Node inside an EmbeddedDAGNode placement named on-topic-search: ['on-topic-search']
  • Doubly-nested: ['on-topic-search', 'inner-placement']

Use it to disambiguate same-labeled inner placements across multiple embedded-DAG instances. The full qualified observability label of the current node is [...placementPath, nodeName].join('/').

Details for Nerds ​

Subclass hooks ​

ts
/**
 * ObservedDag re-export from `@studnicky/dagonizer`.
 *
 * The class has been promoted to the framework. This file preserves
 * the import path so existing imports within the-archivist resolve
 * without updating every file.
 */

export { ObservedDag } from '@studnicky/dagonizer';

The Cartographer uses the same subclassing surface in a deterministic ETL pipeline. Its dispatcher logs flow boundaries, node outputs, errors, phase entries, phase exits, and embedded placement paths:

ts
import { Dagonizer } from '@studnicky/dagonizer';
import type { DagonizerOptionsType, ExecutionResultType } from '@studnicky/dagonizer';

import { CartographerState } from './CartographerState.ts';
import { ConsoleLogger } from './logger/ConsoleLogger.ts';

const COMPONENT = 'ObservedCartographer';

export class ObservedCartographer extends Dagonizer<CartographerState> {
  readonly #logger = new ConsoleLogger();

  constructor(options: DagonizerOptionsType) {
    super(options);
  }

  /** The example's own logger; the CLI routes report display through `result`. */
  get logger(): ConsoleLogger {
    return this.#logger;
  }

  /**
   * Fires before the entrypoint node runs. Logs DAG entry plus the run's
   * configured event count so a reader can correlate scale with throughput.
   */
  protected override onFlowStart(dagName: string, state: CartographerState): void {
    this.#logger.info(COMPONENT, 'onFlowStart', `dag=${dagName} events=${String(state.eventCount)}`);
  }

  /**
   * Fires after the execution loop drains (terminal node, error, or abort).
   * Logs the outcome, executed-node count, and resolved lifecycle variant.
   */
  protected override onFlowEnd(
    dagName: string,
    state: CartographerState,
    result: ExecutionResultType<CartographerState>,
  ): void {
    const executed = result.executedNodes.length;
    const outcome  = result.terminalOutcome ?? result.interruptedAt?.reason ?? 'none';
    this.#logger.info(
      COMPONENT,
      'onFlowEnd',
      `dag=${dagName} outcome=${outcome} executed=${String(executed)} lifecycle=${state.lifecycle.variant}`,
    );
  }

  /**
   * Fires before each node's `execute()` call. `placementPath` is the
   * ordered list of parent embedded-DAG placement names (empty for
   * top-level nodes). Fires for in-process AND worker / contained nodes.
   */
  protected override onNodeStart(
    nodeName: string,
    _state: CartographerState,
    placementPath: readonly string[],
  ): void {
    const path = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    this.#logger.trace(COMPONENT, 'onNodeStart', `${path}${nodeName}`);
  }

  /**
   * Fires after a node completes successfully. `output` is the routing tag
   * the node returned, or `null` for terminal placements.
   */
  protected override onNodeEnd(
    nodeName: string,
    output: string | null,
    _state: CartographerState,
    placementPath: readonly string[],
  ): void {
    const path   = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    const outTag = output ?? '(terminal)';
    this.#logger.debug(COMPONENT, 'onNodeEnd', `${path}${nodeName} -> ${outTag}`);
  }

  /**
   * Fires when the dispatcher catches an error from a node or from the
   * abort / timeout machinery. Routed to `error` (stderr).
   */
  protected override onError(
    nodeName: string,
    error: Error,
    _state: CartographerState,
    placementPath: readonly string[],
  ): void {
    const path = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    this.#logger.error(
      COMPONENT,
      'onError',
      `${path}${nodeName} threw ${error.constructor.name}: ${error.message}`,
    );
  }

  /**
   * Fires before a `pre` or `post` phase placement runs.
   */
  protected override onPhaseEnter(
    dagName: string,
    phase: 'pre' | 'post',
    placementName: string,
    _state: CartographerState,
    _placementPath: readonly string[],
  ): void {
    this.#logger.trace(COMPONENT, 'onPhaseEnter', `dag=${dagName} phase=${phase} placement=${placementName}`);
  }

  /**
   * Fires after a `pre` or `post` phase placement completes.
   */
  protected override onPhaseExit(
    dagName: string,
    phase: 'pre' | 'post',
    placementName: string,
    _state: CartographerState,
    _placementPath: readonly string[],
  ): void {
    this.#logger.trace(COMPONENT, 'onPhaseExit', `dag=${dagName} phase=${phase} placement=${placementName}`);
  }
}

All seven default to no-ops. Override only the hooks you need. Multi-observer composition (logger plus tracer plus metrics) is a subclass concern: write it into the subclass body.

Subclass observer ​

ts
class TracingDispatcher extends Dagonizer<PipelineState> {
  readonly #prefix: string;
  readonly #lines: string[] = [];

  constructor(prefix: string) {
    super();
    this.#prefix = prefix;
  }

  get lines(): readonly string[] { return this.#lines; }

  protected override onFlowStart(dagName: string, _state: PipelineState): void {
    this.#lines.push(`${this.#prefix} flowStart  dag=${dagName}`);
  }

  protected override onFlowEnd(
    dagName: string,
    _state: PipelineState,
    result: ExecutionResultType<PipelineState>,
  ): void {
    this.#lines.push(
      `${this.#prefix} flowEnd    dag=${dagName} outcome=${result.terminalOutcome ?? result.interruptedAt?.reason ?? 'none'} nodes=${String(result.executedNodes.length)}`,
    );
  }

  protected override onNodeStart(
    nodeName: string,
    _state: PipelineState,
    placementPath: readonly string[],
  ): void {
    const path = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    this.#lines.push(`${this.#prefix} nodeStart  ${path}${nodeName}`);
  }

  protected override onNodeEnd(
    nodeName: string,
    output: string | null,
    _state: PipelineState,
    placementPath: readonly string[],
  ): void {
    const path   = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    const outTag = output ?? '(terminal)';
    this.#lines.push(`${this.#prefix} nodeEnd    ${path}${nodeName} → ${outTag}`);
  }

  protected override onError(
    nodeName: string,
    error: Error,
    _state: PipelineState,
    placementPath: readonly string[],
  ): void {
    const path = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    this.#lines.push(`${this.#prefix} error      ${path}${nodeName} ${error.message}`);
  }
}

OpenTelemetry integration ​

OpenTelemetry spans map directly onto the onFlowStart / onFlowEnd and onNodeStart / onNodeEnd pairs. The pattern is identical to the subclass observer above:

  • onFlowStart → tracer.startSpan('flow.<dagName>'), stash in a Map.
  • onNodeStart → tracer.startSpan('node.<nodeName>'), stash keyed by node name.
  • onNodeEnd → retrieve the span, call span.setAttribute('output', ...), then span.end().
  • onError → retrieve the span, call span.recordException(error) and span.setStatus({ code: SpanStatusCode.ERROR }).
  • onFlowEnd → end the flow span and clear the map.

Wire @opentelemetry/api in through the constructor as a Tracer instance. The subclass holds the Map<string, Span> as a private field; nothing leaks to Dagonizer's public surface.

observers option: mux without subclassing ​

Per-turn-rebuilt dispatchers (serverless handlers, per-request factories) cannot use subclassing because the dispatcher is constructed fresh each turn. The observers option accepts a ReadonlyArray<DispatcherObserverType> — each record's callbacks are muxed into the corresponding lifecycle hook in array order, after any subclass override.

ts
import { Dagonizer } from '@studnicky/dagonizer';
import type { DispatcherObserverType } from '@studnicky/dagonizer';

const logObserver: DispatcherObserverType = {
  onFlowStart: (dagName) => console.log('[flow] start', dagName),
  onFlowEnd:   (dagName, _, res) => console.log('[flow] end', dagName, res.terminalOutcome),
  onNodeEnd:   (name, output) => console.log('[node]', name, '->', output),
};

const metrics = { nodeStart: 0, nodeEnd: 0 };
const metricsObserver: DispatcherObserverType = {
  onNodeStart: () => { metrics.nodeStart++; },
  onNodeEnd:   () => { metrics.nodeEnd++; },
};

const dispatcher = new Dagonizer<MyState>({
  observers: [logObserver, metricsObserver],
});

The DispatcherObserverType record mirrors the seven protected hooks. Every callback is optional — include only the hooks you need. Observers fire after the subclass override (if any), so both mechanisms are composable: a subclass can still call super.onFlowStart(...) and the muxed observers fire after it.

See also the full example: npx tsx examples/33-plugin.ts (the #observers-option region).

Multi-observer composition ​

When one caller owns the dispatcher, the subclass pattern is sufficient. For multiple observers (logger plus tracer plus metrics), accept each as a constructor parameter and dispatch to all inside the relevant hook overrides:

ts
class ComposedDispatcher extends Dagonizer<PipelineState> {
  readonly #logA: (msg: string) => void;
  readonly #logB: (msg: string) => void;

  constructor(logA: (msg: string) => void, logB: (msg: string) => void) {
    super();
    this.#logA = logA;
    this.#logB = logB;
  }

  protected override onNodeStart(
    nodeName: string,
    _state: PipelineState,
    placementPath: readonly string[],
  ): void {
    const path = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    this.#logA(`nodeStart  ${path}${nodeName}`);
    this.#logB(`nodeStart  ${path}${nodeName}`);
  }

  protected override onNodeEnd(
    nodeName: string,
    output: string | null,
    _state: PipelineState,
    placementPath: readonly string[],
  ): void {
    const path   = placementPath.length > 0 ? `[${placementPath.join('/')}] ` : '';
    const outTag = output ?? '(terminal)';
    this.#logA(`nodeEnd    ${path}${nodeName} → ${outTag}`);
    this.#logB(`nodeEnd    ${path}${nodeName} → ${outTag}`);
  }
}

EventBus: decoupled multi-subscriber observability ​

The subclass pattern works when one caller owns the dispatcher. When multiple orthogonal subscribers need to observe the same run — a console logger, an SSE endpoint, and a metrics counter — the EventBus from @studnicky/dagonizer/progress decouples the hook implementation from each subscriber.

Instead of dispatching to every observer inside each hook body, the subclass publishes one structured event per hook, and each observer subscribes independently. Adding a new observer does not require touching the dispatcher.

ts
import { Dagonizer } from '@studnicky/dagonizer';
import { EventBus } from '@studnicky/dagonizer/progress';
import type { BusEventEnvelopeType } from '@studnicky/dagonizer/progress';

type LifecyclePayloadType =
  | { event: 'flowStart'; dagName: string }
  | { event: 'flowEnd';   dagName: string; outcome: string }
  | { event: 'nodeStart'; nodeName: string; path: string }
  | { event: 'nodeEnd';   nodeName: string; output: string | null; path: string }
  | { event: 'nodeError'; nodeName: string; message: string };

class ObservingDispatcher extends Dagonizer<MyState> {
  readonly #bus: EventBus;
  readonly #topic: string;

  constructor(bus: EventBus, topic: string) {
    super();
    this.#bus = bus;
    this.#topic = topic;
  }

  protected override onFlowStart(dagName: string): void {
    this.#bus.publish(this.#topic, { event: 'flowStart', dagName });
  }

  protected override onFlowEnd(dagName: string, _state: MyState, result): void {
    const outcome = result.terminalOutcome ?? result.interruptedAt?.reason ?? 'none';
    this.#bus.publish(this.#topic, { event: 'flowEnd', dagName, outcome });
  }

  protected override onNodeStart(nodeName: string, _state: MyState, path: readonly string[]): void {
    this.#bus.publish(this.#topic, { event: 'nodeStart', nodeName, path: path.join('/') });
  }

  protected override onNodeEnd(nodeName: string, output: string | null, _state: MyState, path: readonly string[]): void {
    this.#bus.publish(this.#topic, { event: 'nodeEnd', nodeName, output, path: path.join('/') });
  }

  protected override onError(nodeName: string, error: Error, _state: MyState, path: readonly string[]): void {
    this.#bus.publish(this.#topic, { event: 'nodeError', nodeName, message: error.message });
  }
}

Wire subscribers before executing:

ts
const bus = new EventBus();

// Subscriber A: console logger
bus.subscribe('lifecycle', (e: BusEventEnvelopeType) => {
  const p = e.payload as LifecyclePayloadType;
  console.log(`[log] ${p.event}`);
});

// Subscriber B: metrics counter
const metrics = { nodeStart: 0, nodeEnd: 0 };
bus.subscribe('lifecycle', (e: BusEventEnvelopeType) => {
  const p = e.payload as LifecyclePayloadType;
  if (p.event === 'nodeStart') metrics.nodeStart++;
  if (p.event === 'nodeEnd')   metrics.nodeEnd++;
});

// Subscriber C: SSE endpoint (ReadableStream piped as response body)
import { SseStream } from '@studnicky/dagonizer/progress';
const stream = SseStream.of(bus, ['lifecycle'], { heartbeatMs: 15_000 });
// return new Response(stream.readable, { headers: { 'Content-Type': 'text/event-stream' } });

const dispatcher = new ObservingDispatcher(bus, 'lifecycle');
// register nodes + DAG ...
await dispatcher.execute('urn:noocodec:dag:my-dag', state);

bus.dispose(); // unsubscribes all subscribers

EventBus delivery is synchronous. Every subscriber fires inline before publish returns, in subscription order. A throwing subscriber is caught and ignored so that sibling subscribers still receive the event.

SseStream heartbeats. The default heartbeat interval is 15 000 ms (a : heartbeat\n\n SSE comment frame, invisible to EventSource listeners). Set heartbeatMs: 0 to disable in tests. The heartbeat timer is cleared when the subscriber cancels the stream.

See Example 30: EventBus and SseStream for a complete runnable demonstration.

BusObserver ​

BusObserver is a pre-built DispatcherObserverType implementation that publishes each lifecycle event as a typed DagLifecycleEventType payload to a named bus topic. Pass it in the observers option instead of writing the publishing subclass yourself.

ts
import { Dagonizer } from '@studnicky/dagonizer';
import { EventBus, BusObserver, SseStream } from '@studnicky/dagonizer/progress';
import type { DagLifecycleEventType } from '@studnicky/dagonizer/progress';

const bus = new EventBus();
const dispatcher = new Dagonizer<MyState>({
  observers: [new BusObserver(bus, 'pipeline-events')],
});

// Multiple independent subscribers on the same topic
bus.subscribe('pipeline-events', (e) => {
  const p = e.payload as DagLifecycleEventType;
  logger.info(p.event);
});
bus.subscribe('pipeline-events', (e) => {
  const p = e.payload as DagLifecycleEventType;
  metrics.record(p);
});

// SSE stream for browser clients
const stream = SseStream.of(bus, ['pipeline-events'], { heartbeatMs: 15_000 });
// return new Response(stream.readable, { headers: { 'Content-Type': 'text/event-stream' } });

// register nodes + DAG, then run:
await dispatcher.execute('urn:noocodec:dag:my-dag', new MyState());
bus.dispose();

DagLifecycleEventType ​

A discriminated union where the event field is the discriminant:

eventAdditional fields
'flowStart'dagName
'flowEnd'dagName, outcome
'nodeStart'nodeName, placementPath
'nodeEnd'nodeName, output, placementPath
'nodeError'nodeName, error, placementPath
'phaseEnter'dagName, phase, placementName
'phaseExit'dagName, phase, placementName

State is never included in the payload. State is a mutable reference; publishing it to multiple independent subscribers would allow one subscriber's read to race another subscriber's write.

outcome on 'flowEnd' is result.terminalOutcome ?? result.interruptedAt?.reason ?? 'none'.

Combining BusObserver with subclassing ​

BusObserver is composable with subclass hooks. If you also need the ObservedDag console logger, pass both:

ts
const dispatcher = new ObservedDag<MyState>(logger, {
  observers: [new BusObserver(bus, 'pipeline-events')],
});

The subclass hook fires first, then the observer array fires in order. No changes to either the subclass or BusObserver are needed.

Timing ​

ObservedDag accepts an optional substrate Timing sink:

ts
import { Timing } from '@studnicky/timing';

const timing = Timing.create({ maxEvents: 100 });
const dispatcher = new ObservedDag<MyState>(logger, {
  timing,
  observers: [new BusObserver(bus, 'pipeline-events')],
});

The observer records dag.flow.start, dag.flow.complete, dag.node.start, dag.node.complete, dag.node.error, dag.phase.start, and dag.phase.complete. Read timing.getEvents() when emitting the final request or trace log context.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky