Dagonizer
Dagonizer

Example 21: Per-Node Timeout ​

What It Is ​

Per-Node Timeout gives one node its own wall-clock budget. The Dispatcher uses it to keep slow model-backed support steps from hanging the browser runner indefinitely.

Set timeout: Timeout.ofMs(n) on a node implementation when that operation should fail fast on its own schedule. The rest of the DAG keeps the normal run-level signal unless the caller also cancels or sets deadlineMs.

How It Works ​

The engine wraps one node execution in a child signal derived from the run signal. If the timeout fires, the child signal aborts, the engine raises NodeTimeoutError, and the run records a structured interrupted point for that node. The parent run signal stays independent unless the caller also supplied a run-level deadline or abort signal.

That scope is the useful part. A timeout on classify-message means that node exceeded its budget; it does not mean every other in-flight branch must inherit a cancelled run signal.

Diagrams, Examples, and Outputs ​

DAG registration and diagram ​

Timeout is node configuration, not a new placement shape. The Dispatcher uses real engine-level node timeouts on its LLM-backed classify-message and ai-compose nodes so slow model calls are bounded in the browser runnable.

support-dispatcher timeout 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")))

Set timeout: Timeout.ofMs(n) on a node's NodeInterface definition to activate engine-level per-node timeout. When the node's execute() call does not resolve within the budget:

  1. The engine derives a child AbortController from the run's signal.
  2. The child signal is aborted after the budget expires.
  3. The engine throws NodeTimeoutError, fires onError, and marks the run failed with result.interruptedAt.reason === 'timeout'.

Key difference from run-level deadlineMs (in ExecuteOptions):

  • timeout is scoped to one node's execute() only. The parent run-level signal is not aborted; other nodes are unaffected.
  • deadlineMs aborts the entire run; timeout aborts just the node.

In the runnable Dispatcher, the timeout is attached to the node implementation. The JSON-LD placement stays ordinary; timeout behavior belongs to the registered node contract.

Run ​

bash
npm run docs:dev

What It Lets You Do ​

Per-node timeout lets applications bound one slow node without aborting the whole run-level signal. Use it when a model call, tool call, API adapter, or parser has its own wall-clock budget and should fail cleanly without cancelling unrelated work.

For product code, this keeps latency policy close to the operation that owns the risk. A support classifier can have a short budget, while a downstream operator hand-off or cleanup phase still follows its own contract.

Code Samples ​

The node snippets show the timeout on real Dispatcher nodes. The DAG snippet shows that placements remain ordinary; the timeout belongs to the registered node implementation.

ts
/**
 * ClassifyMessageNode: mode-switched triage with LLM recovery.
 *
 * Routes each inbound customer message to one of three outputs:
 *   'routine'   — AI can handle; routes to ai-compose.
 *   'escalate'  — human operator needed; routes to park-for-operator.
 *   'off-topic' — blank or unrelated; routes to decline.
 *
 * Fast paths (win regardless of `state.classificationMode`):
 *   Trolley switch (state.humanMode === true) forces escalation on every
 *   message before any classification is attempted.
 *   Empty message → off-topic immediately.
 *
 * Mode dispatch (`state.classificationMode`):
 *   `'llm'`      — runs `services.llm.classify` exclusively; `services.intent`
 *                  is never consulted.
 *   `'embedder'` — when `services.intent` is provisioned, classifies via
 *                  cosine similarity against the three intent anchors — no
 *                  LLM round-trip, so trivial messages never risk the
 *                  adapter timeout. If the embedder is confident (above its
 *                  floor), its verdict routes directly. If the embedder is
 *                  unavailable (`services.intent === null`) or unconfident
 *                  (returns `null`), the node runs the LLM path.
 *
 * LLM recovery and error handling:
 *   If the LLM call throws, the node escalates with a safety reason rather
 *   than surfacing an unhandled error — a conservative recovery that keeps
 *   customers in the flow.
 */

import { Batch, BatchItemExecutor, MonadicNode, NodeOutput } from '@studnicky/dagonizer';
import type { ItemType, NodeContextType, NodeOutputType, RoutedBatchType, SchemaObjectType } from '@studnicky/dagonizer';
import { Timeout } from '@studnicky/dagonizer/runtime';

import type { DispatcherState } from '../DispatcherState.ts';
import type { DispatcherServices } from '../services.ts';

export class ClassifyMessageNode extends MonadicNode<DispatcherState, 'routine' | 'escalate' | 'off-topic'> {
  readonly name = 'classify-message';
  readonly '@id' = 'urn:noocodec:node:classify-message';
  readonly outputs = ['routine', 'escalate', 'off-topic'] as const;
  override readonly timeout = Timeout.ofMs(60_000);

  readonly #services: DispatcherServices;

  constructor(services: DispatcherServices) {
    super();
    this.#services = services;
  }

  override get outputSchema(): Record<'routine' | 'escalate' | 'off-topic', SchemaObjectType> {
    return {
      'routine':   { 'type': 'object' },
      'escalate':  { 'type': 'object' },
      'off-topic': { 'type': 'object' },
    };
  }

  override async execute(
    batch: Batch<DispatcherState>,
    context: NodeContextType,
  ): Promise<RoutedBatchType<'routine' | 'escalate' | 'off-topic', DispatcherState>> {
    const acc = new Map<'routine' | 'escalate' | 'off-topic', ItemType<DispatcherState>[]>();
    const results = await BatchItemExecutor.map(batch.items(), async (item) => {
      const output = await this.routeItem(item.state, context);

      for (const error of output.errors) {
        item.state.collectError(error);
      }
      return { item, output };
    }, this.#services.execution, context.signal);

    for (const result of results) {
      const bucket = acc.get(result.output.output);
      if (bucket === undefined) {
        acc.set(result.output.output, [result.item]);
      } else {
        bucket.push(result.item);
      }
    }

    const routed = new Map<'routine' | 'escalate' | 'off-topic', Batch<DispatcherState>>();
    for (const [output, items] of acc) {
      routed.set(output, Batch.from(items));
    }
    return routed;
  }

  private async routeItem(
    state: DispatcherState,
    context: NodeContextType,
  ): Promise<NodeOutputType<'routine' | 'escalate' | 'off-topic'>> {
    // Trolley switch: force human routing regardless of content.
    if (state.humanMode) {
      state.escalationReason = 'Human mode active — all messages routed to operator';
      return NodeOutput.create('escalate');
    }

    // Empty message → off-topic without LLM.
    if (state.message.trim().length === 0) {
      return NodeOutput.create('off-topic');
    }

    if (state.classificationMode === 'llm') {
      return this.classifyViaLlm(state, context);
    }

    // Embedder mode: cosine similarity against the intent anchors, no LLM
    // round-trip, no timeout exposure — routes to the LLM path below
    // when the embedder is unavailable or unconfident.
    if (this.#services.intent !== null) {
      const result = await this.#services.intent.classify(state.message);
      if (result !== null) return this.route(state, result.intent);
    }
    return this.classifyViaLlm(state, context);
  }

  /** LLM classification with conservative escalation on error. */
  private async classifyViaLlm(
    state: DispatcherState,
    context: NodeContextType,
  ): Promise<NodeOutputType<'routine' | 'escalate' | 'off-topic'>> {
    let intent: 'routine' | 'escalate' | 'off-topic';
    try {
      intent = await this.#services.llm.classify(state.message, state.conversation, context.signal);
    } catch {
      state.escalationReason = 'LLM unavailable; escalated for safety';
      return NodeOutput.create('escalate');
    }
    return this.route(state, intent);
  }

  /** Shared routing for both the embedder and LLM classification paths. */
  private route(
    state: DispatcherState,
    intent: 'routine' | 'escalate' | 'off-topic',
  ): NodeOutputType<'routine' | 'escalate' | 'off-topic'> {
    if (intent === 'escalate') {
      state.escalationReason = 'Agent determined this message requires human review.';
      return NodeOutput.create('escalate');
    }
    if (intent === 'off-topic') return NodeOutput.create('off-topic');
    return NodeOutput.create('routine');
  }
}
ts
/**
 * AiComposeNode: LLM-powered reply composition for routine support queries.
 *
 * Calls the LLM to compose a concise, friendly response to the customer's
 * message using recent conversation history as context.
 *
 * Error handling:
 *   If the LLM call fails, a polite recovery message is set on state so
 *   the flow can continue to send-response without surfacing a raw error.
 *
 * Routes 'drafted' on every path.
 */

import { BatchItemExecutor, MonadicNode, RoutedBatch } from '@studnicky/dagonizer';
import type { Batch, NodeContextType, RoutedBatchType, SchemaObjectType } from '@studnicky/dagonizer';
import { Timeout } from '@studnicky/dagonizer/runtime';

import type { DispatcherState } from '../DispatcherState.ts';
import type { DispatcherServices } from '../services.ts';

export class AiComposeNode extends MonadicNode<DispatcherState, 'drafted'> {
  readonly name = 'ai-compose';
  readonly '@id' = 'urn:noocodec:node:ai-compose';
  readonly outputs = ['drafted'] as const;
  override readonly timeout = Timeout.ofMs(60_000);

  readonly #services: DispatcherServices;

  constructor(services: DispatcherServices) {
    super();
    this.#services = services;
  }

  override get outputSchema(): Record<'drafted', SchemaObjectType> {
    return { 'drafted': { 'type': 'object' } };
  }

  override async execute(
    batch: Batch<DispatcherState>,
    context: NodeContextType,
  ): Promise<RoutedBatchType<'drafted', DispatcherState>> {
    await BatchItemExecutor.map(batch.items(), async (item) => {
      try {
        item.state.response = await this.#services.llm.compose(item.state.message, item.state.conversation, context.signal);
      } catch {
        item.state.response = 'I apologize — I had trouble composing a reply. Please try again or ask for a human agent.';
      }
    }, this.#services.execution, context.signal);
    return RoutedBatch.create('drafted', batch);
  }
}
ts

const setup           = new PlaceholderNode<DispatcherState, 'ready'>('urn:noocodec:node:dispatcher-setup', ['ready']);
const classifyMessage = new PlaceholderNode<DispatcherState, 'routine' | 'escalate' | 'off-topic'>('urn:noocodec:node:classify-message', ['routine', 'escalate', 'off-topic']);
const aiCompose       = new PlaceholderNode<DispatcherState, 'drafted'>('urn:noocodec:node:ai-compose', ['drafted']);
const parkForOperator = new PlaceholderNode<DispatcherState, 'parked' | 'ready'>('urn:noocodec:node:park-for-operator', ['parked', 'ready']);
const sendResponse    = new PlaceholderNode<DispatcherState, 'sent'>('urn:noocodec:node:send-response', ['sent']);
const decline         = new PlaceholderNode<DispatcherState, 'declined'>('urn:noocodec:node:decline', ['declined']);

const supportDispatcherDagIri = 'urn:noocodec:dag:support-dispatcher' as const;
const placement = (placementIdentifier: string): string =>
  DAGIdentity.placementId(supportDispatcherDagIri, placementIdentifier);

export const supportDispatcherDAG: DAGType = new DAGBuilder(supportDispatcherDagIri, '1')
  // Pre-phase: stamps runId before the entrypoint runs.
  .phase(placement('setup'), 'pre', setup)

  // Entrypoint: classify the inbound message.
  .node(placement('classify-message'), classifyMessage, {
    'routine':   placement('ai-compose'),
    'escalate':  placement('park-for-operator'),
    'off-topic': placement('decline'),
  })

  // Routine branch: AI composes a canned reply -> send -> done.
  .node(placement('ai-compose'), aiCompose, {
    'drafted': placement('send-response'),
  })

  // Escalation branch: HITL suspension point.
  .node(placement('park-for-operator'), parkForOperator, {
    'parked': placement('end'),
    'ready':  placement('send-response'),
  })

  // Shared convergence: both routine and escalated paths flow through send-response.
  .node(placement('send-response'), sendResponse, {
    'sent': placement('end'),
  })

  // Off-topic branch: decline and close.
  .node(placement('decline'), decline, {
    'declined': placement('end'),
  })

  .terminal(placement('end'), { 'outcome': 'completed' })

  .build();

Details for Nerds ​

  • timeout on NodeInterface. Set override readonly timeout = Timeout.ofMs(n) on the node class to activate the per-node deadline. Import Timeout from @studnicky/dagonizer/runtime. The engine arms the timer before calling execute() and cancels it when execute() resolves normally.
  • NodeTimeoutError. Thrown by the engine (not the node) when the deadline fires. Carries nodeName and the timeout budget for diagnostic reporting.
  • Child AbortController. The node's context.signal is the child signal derived from the run signal. Aborting the child does not abort the parent run or any other running node.
  • result.interruptedAt.reason. After a per-node timeout, result.interruptedAt.reason === 'timeout' and result.interruptedAt.nodeName identifies the timed-out node. The run lifecycle is failed.
  • Contrast with deadlineMs. Run-level deadlineMs aborts context.signal (the run signal), which propagates to all running nodes. Per-node timeout aborts only the node's child signal.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky