Conversational Agents

What It Is

Conversational applications are still workflows. A user message enters state, nodes classify intent or fill slots, the DAG either replies, parks for a human, dispatches tools, or loops through another model call. Dagonizer keeps those turn decisions visible as JSON-LD topology instead of hiding them in callback stacks.

This guide covers three conversation shapes that appear in the runnable examples: request/response turns, human-in-the-loop parking, and the reusable agent loop.

How It Works

Conversational flows keep serializable domain progress on state and put ephemeral IO handles behind metadata, stores, triggers, or host code. A turn either completes with a response, parks with a cursor, or streams through a producer/channel surface while the DAG remains the explicit control-flow graph.

Interactive DAGs — those that prompt users, wait for responses, escalate to humans, or stream results — present a challenge: the input/output cannot flow through serializable state.params when it is ephemeral (a live socket, a request/response pair, an SSE stream, a queue item). This guide documents two proven patterns for threading non-serializable IO through a DAG.

Diagrams, Examples, and Outputs

The runnable examples show two complementary conversational shapes. The ReAct loop is the reusable agent graph; the Dispatcher support flow shows park-and-correlate handoff:

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:react-agent",
  "@type": "DAG",
  "name": "react-agent",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:react-agent/node/build-request"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:react-agent/node/build-request",
      "@type": "SingleNode",
      "name": "build-request",
      "node": "urn:noocodec:node:build-request",
      "outputs": {
        "ready": "urn:noocodec:dag:react-agent/node/call-model",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/call-model",
      "@type": "SingleNode",
      "name": "call-model",
      "node": "urn:noocodec:node:call-model",
      "outputs": {
        "text": "urn:noocodec:dag:react-agent/node/normalize-response",
        "tools": "urn:noocodec:dag:react-agent/node/normalize-response",
        "mixed": "urn:noocodec:dag:react-agent/node/normalize-response",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/normalize-response",
      "@type": "SingleNode",
      "name": "normalize-response",
      "node": "urn:noocodec:node:normalize-response",
      "outputs": {
        "text": "urn:noocodec:dag:react-agent/node/append-assistant",
        "tools": "urn:noocodec:dag:react-agent/node/decode-tools",
        "mixed": "urn:noocodec:dag:react-agent/node/decode-tools",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/append-assistant",
      "@type": "SingleNode",
      "name": "append-assistant",
      "node": "urn:noocodec:node:append-assistant",
      "outputs": {
        "done": "urn:noocodec:dag:react-agent/node/end-done",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/decode-tools",
      "@type": "SingleNode",
      "name": "decode-tools",
      "node": "urn:noocodec:node:decode-tools",
      "outputs": {
        "decoded": "urn:noocodec:dag:react-agent/node/normalize-tools",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/normalize-tools",
      "@type": "SingleNode",
      "name": "normalize-tools",
      "node": "urn:noocodec:node:normalize-tools",
      "outputs": {
        "valid": "urn:noocodec:dag:react-agent/node/worksets",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/worksets",
      "@type": "SingleNode",
      "name": "worksets",
      "node": "urn:noocodec:node:build-worksets",
      "outputs": {
        "ready": "urn:noocodec:dag:react-agent/node/dispatch-tools",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/dispatch-tools",
      "@type": "ScatterNode",
      "name": "dispatch-tools",
      "source": "safeWorkset",
      "body": {
        "dag": {
          "@type": "DagReference",
          "from": "item",
          "path": "dagIri",
          "candidates": [
            "urn:noocodec:tool:lookup"
          ]
        }
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:react-agent/node/join-tool-results",
        "partial": "urn:noocodec:dag:react-agent/node/join-tool-results",
        "all-error": "urn:noocodec:dag:react-agent/node/join-tool-results",
        "empty": "urn:noocodec:dag:react-agent/node/join-tool-results"
      },
      "itemKey": "currentItem",
      "reducer": "aggregate"
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/join-tool-results",
      "@type": "GatherNode",
      "name": "join-tool-results",
      "sources": {
        "urn:noocodec:dag:react-agent/node/dispatch-tools": {}
      },
      "gather": {
        "strategy": "map",
        "mapping": {
          "output": "toolOutputs"
        }
      },
      "outputs": {
        "success": "urn:noocodec:dag:react-agent/node/collect-results",
        "error": "urn:noocodec:dag:react-agent/node/end-error",
        "empty": "urn:noocodec:dag:react-agent/node/collect-results"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/collect-results",
      "@type": "SingleNode",
      "name": "collect-results",
      "node": "urn:noocodec:node:collect-results",
      "outputs": {
        "done": "urn:noocodec:dag:react-agent/node/build-request",
        "empty": "urn:noocodec:dag:react-agent/node/build-request",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/end-done",
      "@type": "TerminalNode",
      "name": "end-done",
      "outcome": "completed"
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/end-error",
      "@type": "TerminalNode",
      "name": "end-error",
      "outcome": "failed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% react-agent (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_react-agent/node/build-request
  urn_noocodec_dag_react-agent/node/build-request["build-request"]
  urn_noocodec_dag_react-agent/node/build-request -->|ready| urn_noocodec_dag_react-agent/node/call-model
  urn_noocodec_dag_react-agent/node/build-request -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/call-model["call-model"]
  urn_noocodec_dag_react-agent/node/call-model -->|text| urn_noocodec_dag_react-agent/node/normalize-response
  urn_noocodec_dag_react-agent/node/call-model -->|tools| urn_noocodec_dag_react-agent/node/normalize-response
  urn_noocodec_dag_react-agent/node/call-model -->|mixed| urn_noocodec_dag_react-agent/node/normalize-response
  urn_noocodec_dag_react-agent/node/call-model -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-response["normalize-response"]
  urn_noocodec_dag_react-agent/node/normalize-response -->|text| urn_noocodec_dag_react-agent/node/append-assistant
  urn_noocodec_dag_react-agent/node/normalize-response -->|tools| urn_noocodec_dag_react-agent/node/decode-tools
  urn_noocodec_dag_react-agent/node/normalize-response -->|mixed| urn_noocodec_dag_react-agent/node/decode-tools
  urn_noocodec_dag_react-agent/node/normalize-response -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-response -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/append-assistant["append-assistant"]
  urn_noocodec_dag_react-agent/node/append-assistant -->|done| urn_noocodec_dag_react-agent/node/end-done
  urn_noocodec_dag_react-agent/node/append-assistant -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/decode-tools["decode-tools"]
  urn_noocodec_dag_react-agent/node/decode-tools -->|decoded| urn_noocodec_dag_react-agent/node/normalize-tools
  urn_noocodec_dag_react-agent/node/decode-tools -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/decode-tools -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-tools["normalize-tools"]
  urn_noocodec_dag_react-agent/node/normalize-tools -->|valid| urn_noocodec_dag_react-agent/node/worksets
  urn_noocodec_dag_react-agent/node/normalize-tools -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-tools -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/worksets["worksets"]
  urn_noocodec_dag_react-agent/node/worksets -->|ready| urn_noocodec_dag_react-agent/node/dispatch-tools
  urn_noocodec_dag_react-agent/node/worksets -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/worksets -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/dispatch-tools[/"dispatch-tools"/]
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|all-success| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|partial| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|all-error| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|empty| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/join-tool-results{"join-tool-results"}
  urn_noocodec_dag_react-agent/node/join-tool-results -->|success| urn_noocodec_dag_react-agent/node/collect-results
  urn_noocodec_dag_react-agent/node/join-tool-results -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/join-tool-results -->|empty| urn_noocodec_dag_react-agent/node/collect-results
  urn_noocodec_dag_react-agent/node/collect-results["collect-results"]
  urn_noocodec_dag_react-agent/node/collect-results -->|done| urn_noocodec_dag_react-agent/node/build-request
  urn_noocodec_dag_react-agent/node/collect-results -->|empty| urn_noocodec_dag_react-agent/node/build-request
  urn_noocodec_dag_react-agent/node/collect-results -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/end-done((("end-done")))
  urn_noocodec_dag_react-agent/node/end-error>"end-error"]

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")))

State & metadata - use state.metadata as the inter-node IO bus
Checkpoint & resume - persist and reload state across process boundaries
Dependency injection - inject IO adapters via node constructors
Example 29: Agent DAG - working example of the 8-node agent loop with stub LLM
The Dispatcher - runnable support handoff and operator response flow
ReAct Agent Memory - runnable trace streaming and provenance recall

What It Lets You Do

Use when

Use this guide when a DAG interacts with people, live transports, or turn-based agent loops. It covers slot filling, human escalation, request/response boundaries, streaming, and the canonical agent topology.

Code Samples

The sections below describe the implementation shapes behind the runnable examples. Prefer the linked demos for copy/paste starting points; use the sketches here to understand the design tradeoffs.

Details for Nerds

Overview

Both approaches use the same core idea: the state metadata bus (state.setMetadata(key, value) to write; state.getter.string/number/...(key) for typed cast-free reads, or state.getMetadata(key) for the raw unknown you narrow yourself) to pass messages in and out of nodes. The differences are:

Aspect	Request/response turn termination	Park-and-correlate handoff
When to use	Conversational slot-filling, sequential dialogs, request/response cycles	HITL escalations, approvals, notifications, async hand-offs
Pause mechanism	Node returns a specific output → routes to a terminal → HTTP turn ends	DAG completes normally; admin action triggers SSE push to waiting client
Resume	Next HTTP request → new DAG execution with reloaded state	Out-of-band event (admin decision, callback) → client receives push
Dependency injection	Constructor args	Constructor args or request-scoped wrapper passed at construction
Complexity	Lower; familiar request/response model	Higher; requires event bus, queue, or webhook framework

Choose turn-termination for conversational flows (the resume-generator use case). Choose out-of-band for HITL workflows where humans think in parallel to the DAG.

Pattern 1: Request/response turn termination

How it works

A conversational DAG runs per HTTP request. The LLM (or a state machine) updates the conversation state. If more information is needed, the node returns a specific output that routes to a terminal placement IRI, ending the turn. The caller reads state.getMetadata('assistantMessage') and sends it to the user. The user's next message starts a fresh DAG execution with the persisted state object.

Example: slot-filling trip planner

typescript

// TripState.ts — domain state
import { NodeStateBase } from '@studnicky/dagonizer';

export class TripState extends NodeStateBase {
  conversationId: string = '';
  preferences: Partial<{ budget: number; duration: string }> = {};
  handoff: HandoffState = { variant: 'empty' };
  // ... other fields
}

// HandoffState.ts — conversation machine
export type HandoffState =
  | { variant: 'empty' }
  | { variant: 'awaitingSlots'; nextQuestion: string }
  | { variant: 'ready'; normalizedPreferences: Record<string, unknown> };

// HandoffMachine.ts — reducer
export class HandoffMachine {
  static transition(
    state: HandoffState,
    event: { type: 'firstTurnExtracted' | 'slotsUpdated' | 'reset' },
    context: { extractedPreferences: Record<string, unknown> }
  ): HandoffState {
    if (state.variant === 'empty' && event.type === 'firstTurnExtracted') {
      const missing = Object.keys(context.extractedPreferences).length < 2;
      if (missing) {
        return { variant: 'awaitingSlots', nextQuestion: 'How long do you want to travel?' };
      }
      return { variant: 'ready', normalizedPreferences: context.extractedPreferences };
    }
    // ... more transitions
    return state;
  }
}

// IntakeNode.ts — the interactive node
import { Batch, MonadicNode, RoutedBatch } from '@studnicky/dagonizer';
import type { ItemType, RoutedBatchType, SchemaObjectType } from '@studnicky/dagonizer';
import type { NodeContextType } from '@studnicky/dagonizer/entities';

export class IntakeNode extends MonadicNode<TripState, 'incomplete' | 'success'> {
  readonly name = 'intake';
  readonly outputs: readonly ('incomplete' | 'success')[] = ['incomplete', 'success'];

  constructor(private readonly llm: LLMAdapter) {
    super();
  }

  override get outputSchema(): Record<'incomplete' | 'success', SchemaObjectType> {
    return MonadicNode.permissiveSchema(this.outputs);
  }

  async execute(batch: Batch<TripState>, context: NodeContextType): Promise<RoutedBatchType<'incomplete' | 'success', TripState>> {
    // Nodes implement execute over a Batch; this node owns its per-item routing locally.
    const incomplete: ItemType<TripState>[] = [];
    const success: ItemType<TripState>[] = [];

    for (const item of batch) {
      const state = item.state;

      // Read inbound message from orchestrator (typed, cast-free)
      const userMessage = state.getter.string('userMessage');

      // Extract preferences with LLM or prompt
      const { extracted } = await this.llm.extract({
        message: userMessage,
        schema: PreferenceSchema,
        guidance: 'Extract travel preferences: budget, duration, destination.',
      });

      // Update state and check conversation state machine
      state.preferences = { ...state.preferences, ...extracted };
      state.handoff = HandoffMachine.transition(state.handoff, { type: 'slotsUpdated' }, {
        extractedPreferences: state.preferences,
      });

      // Write outbound message
      if (state.handoff.variant === 'awaitingSlots') {
        state.setMetadata('assistantMessage', {
          role: 'assistant',
          content: state.handoff.nextQuestion,
        });
        incomplete.push(item);
        continue;
      }

      // All slots filled; continue to specialist nodes
      state.setMetadata('assistantMessage', {
        role: 'assistant',
        content: 'Got it! Finding options for you...',
      });
      success.push(item);
    }

    return RoutedBatch.create([
      ['incomplete', Batch.from(incomplete)],
      ['success', Batch.from(success)],
    ]);
  }
}

Wiring the orchestrator

typescript

// Orchestrator.ts
import { DAG_CONTEXT, Dagonizer } from '@studnicky/dagonizer';
import type { DAGType } from '@studnicky/dagonizer';

export class TripOrchestrator {
  private dispatcher: Dagonizer<TripState>;
  private stateStore: StateStore;

  assemble() {
    // Wire nodes with constructor injection
    const intake = new IntakeNode(this.llm);
    const retrieve = new RetrievalNode(this.searchEngine);
    const propose = new ProposalNode(this.templateEngine);

    const tripPlannerDag: DAGType = {
      '@context': DAG_CONTEXT,
      '@id': 'urn:noocodec:trip-planner:dag',
      '@type': 'DAG',
      'name': 'trip-planner',
      'version': '1.0',
      'entrypoints': { 'main': 'intake' },
      'nodes': [
        {
          '@id': 'urn:noocodec:trip-planner:dag/intake',
          '@type': 'SingleNode',
          'name': 'intake',
          'node': 'intake',
          'outputs': {
            'success': 'retrieve',
            'incomplete': 'end-incomplete',
            'error': 'end-error',
          },
        },
        {
          '@id': 'urn:noocodec:trip-planner:dag/retrieve',
          '@type': 'SingleNode',
          'name': 'retrieve',
          'node': 'retrieve',
          'outputs': {
            'success': 'propose',
            'error': 'end-error',
          },
        },
        {
          '@id': 'urn:noocodec:trip-planner:dag/propose',
          '@type': 'SingleNode',
          'name': 'propose',
          'node': 'propose',
          'outputs': {
            'success': 'end-success',
            'error': 'end-error',
          },
        },
        {
          '@id': 'urn:noocodec:trip-planner:dag/end-incomplete',
          '@type': 'TerminalNode',
          'name': 'end-incomplete',
          'outcome': 'completed',
        },
        {
          '@id': 'urn:noocodec:trip-planner:dag/end-success',
          '@type': 'TerminalNode',
          'name': 'end-success',
          'outcome': 'completed',
        },
        {
          '@id': 'urn:noocodec:trip-planner:dag/end-error',
          '@type': 'TerminalNode',
          'name': 'end-error',
          'outcome': 'failed',
        },
      ],
    };

    this.dispatcher = new Dagonizer<TripState>();
    this.dispatcher.registerNode(intake);
    this.dispatcher.registerNode(retrieve);
    this.dispatcher.registerNode(propose);
    this.dispatcher.registerDAG(tripPlannerDag);
  }

  async runTurn(conversationId: string, userMessage: string): Promise<string> {
    // Load persisted state or create new
    let state = await this.stateStore.load(conversationId);
    if (!state) {
      state = new TripState();
      state.conversationId = conversationId;
    }

    // Seed metadata for this turn
    state.setMetadata('userMessage', userMessage);

    // Execute the DAG
    await this.dispatcher.execute('urn:noocodec:dag:trip-planner', state);

    // Read response from metadata (raw unknown → narrow to your message shape)
    const raw = state.getMetadata('assistantMessage');
    const reply = (raw !== null && typeof raw === 'object' && 'content' in raw) ? raw : null;

    // Persist state for next turn
    await this.stateStore.save(conversationId, state);

    return reply?.content ?? 'Something went wrong.';
  }
}

Key properties

No checkpoint/resume machinery: Each turn is a fresh DAG execution. State is persisted in a store (database, file system, etc.), not via Checkpoint.capture().
Output routing is the pause trigger: Returning { output: 'incomplete' } routes to the 'incomplete' terminal, which causes the top-level DAG to complete. The caller sees the completion and can return control to the user.
State metadata is the message bus: userMessage flows in, assistantMessage flows out. No other mechanism needed.
Simple dependency injection: Constructor args wire dependencies at instantiation time. The dispatcher is built once per orchestrator lifetime; nodes hold their dependencies as private fields.
Familiar HTTP semantics: POST /chat → runs DAG → reads assistantMessage → returns HTTP 200. GET /chat?id=X → loads state → GET /chat → next turn. No long-polling, no WebSocket complexity.

Pattern 2: Park-and-correlate handoff

How it works

The DAG completes a turn normally, synchronously. A separate system (an event bus, queue, or webhook) holds HITL items (escalations, approvals, reviews). Humans act on the queue via a separate admin interface. When they decide, an event fires. A browser SSE stream, a message queue subscription, or a webhook handler receives the event and notifies the waiting client.

Example: refund escalation queue

typescript

// RefundAgentServices.ts
export interface RefundAgentServices {
  readonly llm: LLMAdapter;
  readonly escalations: EscalationQueue;  // <-- HITL queue
  readonly ledger: RefundLedger;
  readonly store: Store;  // for auth state, config
  readonly eventBus: EventBus;  // for push notifications
}

// EscalationQueue.ts — the HITL queue
export interface Escalation {
  id: string;
  decision: 'refund' | 'deny' | 'escalate';
  customerId: string;
  conversationId: string;
  recommendation: string;
  createdAt: number;
}

export class EscalationQueue {
  private queue: Map<string, Escalation> = new Map();

  enqueue(
    decision: string,
    recommendation: string,
    customerId: string,
    conversationId: string,
    context: object
  ): Escalation {
    const item = {
      id: nanoid(),
      decision,
      customerId,
      conversationId,
      recommendation,
      createdAt: Date.now(),
    };
    this.queue.set(item.id, item);
    return item;
  }

  remove(id: string): void {
    this.queue.delete(id);
  }

  items(): Escalation[] {
    return Array.from(this.queue.values());
  }
}

// EventBus.ts — in-process pub/sub
export class EventBus {
  private listeners: Map<string, Set<(data: JsonValue) => void>> = new Map();

  subscribe(topic: string, listener: (data: JsonValue) => void): () => void {
    if (!this.listeners.has(topic)) {
      this.listeners.set(topic, new Set());
    }
    this.listeners.get(topic)!.add(listener);
    return () => this.listeners.get(topic)?.delete(listener);
  }

  publish(topic: string, data: JsonValue): void {
    this.listeners.get(topic)?.forEach(listener => listener(data));
  }
}

// EscalateForDecisionNode.ts — the HITL node
import { Batch, MonadicNode, RoutedBatch } from '@studnicky/dagonizer';
import type { SchemaObjectType } from '@studnicky/dagonizer';
import type { NodeContextType } from '@studnicky/dagonizer/entities';

export class EscalateForDecisionNode
  extends MonadicNode<RefundAgentState, 'queued'>
{
  private readonly escalations: EscalationQueue;
  private readonly eventBus: EventBus;

  constructor(escalations: EscalationQueue, eventBus: EventBus) {
    this.escalations = escalations;
    this.eventBus = eventBus;
  }

  readonly name = 'escalate-for-decision';
  readonly outputs: readonly 'queued'[] = ['queued'];

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

  async execute(
    batch: Batch<RefundAgentState>,
    context: NodeContextType
  ): Promise<RoutedBatchType<'queued', RefundAgentState>> {
    if (context.signal.aborted) return RoutedBatch.create();
    const { escalations, eventBus } = this;
    for (const row of batch) {
      const state = row.state;

      // Enqueue the decision for a human
      const item = escalations.enqueue(
        'review',
        `Customer requested refund for order ${state.orderId}. Policy allows if < 30 days.`,
        state.customerId,
        state.conversationId,
        { orderDate: state.orderDate }
      );

      // Publish to admin SSE stream
      eventBus.publish('escalations', { variant: 'queued', item });

      // Also publish to the customer's SSE stream (tell them to wait)
      eventBus.publish(`conversation:${state.customerId}`, {
        variant: 'pending',
        message: 'Your request is under review. You will be notified shortly.',
      });

      state.escalation = item;
    }
    return RoutedBatch.create('queued', batch);
  }
}

Wiring the orchestrator

typescript

// ConversationEngine.ts
export class ConversationEngine {
  constructor(private eventBus: EventBus, private escalations: EscalationQueue) {}

  async runTurn(
    conversationId: string,
    customerId: string,
    userMessage: string
  ): Promise<Resolution> {
    // Dispatcher is constructed once; nodes hold their deps via constructor injection.
    const dispatcher = new Dagonizer<RefundAgentState>();

    // Load state
    let state = await this.stateStore.load(conversationId);
    if (!state) {
      state = new RefundAgentState();
      state.customerId = customerId;
      state.conversationId = conversationId;
    }

    state.setMetadata('userMessage', userMessage);

    // Execute the DAG
    const result = await dispatcher.execute('urn:noocodec:dag:refund-turn', state);

    // Read output (raw unknown → narrow to your message shape)
    const raw = state.getMetadata('assistantMessage');
    const reply = (raw !== null && typeof raw === 'object' && 'content' in raw) ? raw : null;

    await this.stateStore.save(conversationId, state);

    return {
      reply: reply?.content ?? 'Something went wrong.',
      resolution: state.resolution, // 'approved', 'denied', or 'pending'
    };
  }
}

// Admin endpoint: approve escalation
export async function handleEscalationApprove(escalationId: string, decision: 'approve' | 'deny') {
  const item = escalations.queue.get(escalationId);
  if (!item) return { error: 'not found' };

  // Resolve in ledger
  if (decision === 'approve') {
    ledger.spend(item.customerId, 100);  // spend refund token
    const reply = 'Your refund has been approved!';

    // Notify customer via EventBus → SSE stream
    eventBus.publish(`conversation:${item.customerId}`, {
      variant: 'resolved',
      decision: 'approved',
      reply,
    });
  } else {
    eventBus.publish(`conversation:${item.customerId}`, {
      variant: 'resolved',
      decision: 'denied',
      reply: 'Unfortunately, your request does not qualify.',
    });
  }

  // Remove from queue
  escalations.remove(escalationId);

  // Notify admin stream
  eventBus.publish('escalations', { variant: 'resolved', itemId: escalationId, decision });

  return { ok: true };
}

// Customer SSE endpoint
export function makeSseStream(eventBus: EventBus, customerId: string) {
  return async function* sseGenerator() {
    const unsubscribe = eventBus.subscribe(`conversation:${customerId}`, (event) => {
      yield `data: ${JSON.stringify(event)}\n\n`;
    });

    try {
      // Keep stream alive
      while (true) {
        await new Promise(resolve => setTimeout(resolve, 30000));
        yield `: heartbeat\n\n`;
      }
    } finally {
      unsubscribe();
    }
  };
}

Key properties

DAG completes normally: No special terminal routing. The DAG runs to completion; the caller reads assistantMessage and returns.
Asynchronous handoff: Humans work from a queue. The customer sees "your request is under review" immediately.
Event-driven resolution: Admin action publishes an event; the customer's SSE stream receives a push. No polling.
Per-turn variation via constructor args: When a node needs per-request context (auth, session, ledger), pass it in at construction time as a constructor argument. Nodes are registered per dispatcher instance; rebuild the dispatcher or use a factory that produces nodes with the appropriate per-turn state.
Out-of-band state machine: The escalation queue is separate from the DAG state. It can outlive an HTTP request.

Choosing a pattern

Your scenario	Pattern
"I want a conversational bot that asks questions until it has all the info."	Turn-termination. Simpler, familiar.
"I want an LLM agent to make decisions, but escalate to a human when uncertain."	Out-of-band. The DAG completes; humans review async.
"I need streaming responses (Claude streaming tokens to the browser)."	Turn-termination with a `CallModelNode` subclass constructed with a `{ sink }` option — every `LlmAdapterInterface` implements `chatStream(request, sink)`. See ReAct agent: live token streaming.
"I need multi-turn undo / branching conversations."	Either; add branches to the state machine.
"I need the same DAG to run from a CLI, an API, a queue worker, etc."	Out-of-band is more portable; turn-termination ties you to HTTP request/response. Both work with proper abstraction.

Dependency injection

Constructor injection is the dependency injection model. Nodes receive dependencies through their constructors and hold them as private fields.

typescript

class IntakeNode extends MonadicNode<TripState, 'incomplete' | 'success'> {
  private readonly llm: LLMAdapter;
  readonly name = 'intake';
  readonly outputs: readonly ('incomplete' | 'success')[] = ['incomplete', 'success'];

  constructor(llm: LLMAdapter) {
    super();
    this.llm = llm;
  }
  // define execute(batch, context) — uses this.llm
}

const intake = new IntakeNode(this.llm);
dispatcher.registerNode(intake);

When a dependency varies per execution (for example, a per-request ledger keyed by customerId), construct the node with the appropriate instance before registering it, or restructure the dependency to be execution-agnostic (read the customerId from state inside the node, and pass a factory or a repository rather than a pre-built per-user object).

Nodes are testable in isolation: instantiate with a stub or fake, call execute(Batch.of(state), context) directly, and assert on the result without wiring a dispatcher.

Common questions

Q: How do I stream responses (e.g., Claude streaming tokens)?

A: Every LlmAdapterInterface implements chatStream(request, sink): Promise<ChatResponseType> alongside the buffered chat(request). Construct your CallModelNode subclass with a { sink: StreamSinkInterface<ChatStreamChunkType> } option; the node forwards it to adapter.chatStream(...) and pushes one { delta } chunk per token/fragment as the provider emits it. The sink is a pure observation channel — the assembled ChatResponseType is still written to state exactly as the buffered path writes it, so downstream nodes are unaffected. See ReAct agent: live token streaming and the react-agent-memory example for a complete working setup.

Q: Can I checkpoint a turn and resume it later?

A: Yes; see checkpoint. It's heavier than turn-termination, but necessary if you need to pause inside a node and resume from that exact point in a different process. Most conversational workflows don't need this.

Q: What if a node needs to ask the user a question mid-execution?

A: Return 'needs_more_info' and route that output to a terminal. The caller reads state.metadata('userQuestion') and prompts. The next turn restarts the DAG with the answer in state.metadata('userAnswer').

Q: How do I handle multi-turn context / conversation history?

A: Store it in state (not metadata). The orchestrator loads the persisted state; it already has the history. Each node appends to the history as needed.

Q: Can I mix turn-termination and out-of-band in the same DAG?

A: Yes. A node can enqueue an escalation (out-of-band) and continue. Or it can return 'escalated' and route that to a terminal (turn-termination). Neither pattern is exclusive.

Migration and adoption

To adopt these patterns:

Start from the request/response state-machine shape if you need slot filling. Keep the handoff state serializable and make the missing-slot question explicit in state metadata.
Use the state.metadata bus as documented here. It is already part of NodeStateBase in the framework.
Use constructor injection for all node dependencies. Pass the dependency at new NodeClass(dep) and hold it as a private field.
Test nodes in isolation with mock state and mocked services. The patterns are testable without a framework.
Plan for persistence: You own the state store. Use a database, file system, Redis, or whatever fits your deployment.

Reusable agent loops use DAGBuilder for topology, the eight agent node bases from @studnicky/dagonizer/patterns for state-specific behavior, and AgentTraceProducer for live reasoning traces.

Agent loop

The patterns above describe how to structure a single DAG turn. When the agent needs to call tools and loop back to the model with the results, the turn contains the inner loop itself: build request → call model → inspect variant → dispatch tools → collect results → loop.

The canonical 8-node topology

build-request
  └─ ready ──► call-model
                └─ text|tools|mixed ──► normalize-response
                     ├─ text  ──► append-assistant ──► end-done (completed)
                     └─ tools|mixed ──► decode-tools
                                          └─ decoded ──► normalize-tools
                                               └─ valid ──► worksets
                                                    └─ ready ──► dispatch-tools
                                                         (scatter: DagReference item.dagIri)
                                                         └─ collect-results ──► build-request

Every LLM-with-tools agent repeats this structure. Author it as an explicit JSON-LD DAGType so the verified topology remains visible while callers subclass the node bases.

Agent DAG as JSON-LD

import { Dagonizer, NodeStateBase } from '@studnicky/dagonizer';
import { DAGBuilder } from '@studnicky/dagonizer/builder';
import {
  BuildChatRequestNode,
  CallModelNode,
  NormalizeResponseNode,
  DecodeTextToolCallsNode,
  NormalizeToolCallsNode,
  BuildToolWorksetsNode,
  CollectToolResultsNode,
  AppendAssistantNode,
} from '@studnicky/dagonizer/patterns';

const nodes = {
  chatRequest:         new MyBuildChatRequestNode(),
  callModel:           new MyCallModelNode(llm),
  normalizeResponse:   new MyNormalizeResponseNode(),
  decodeTextToolCalls: new MyDecodeTextToolCallsNode(),
  normalizeToolCalls:  new MyNormalizeToolCallsNode(),
  toolWorksets:        new MyBuildToolWorksetsNode(),
  collectToolResults:  new MyCollectToolResultsNode(),
  appendAssistant:     new MyAppendAssistantNode(),
};

const dispatcher = new Dagonizer<AgentState>();
// Register all 8 nodes, tool bundle, and the authored DAG.
dispatcher.registerNode(nodes.chatRequest);
dispatcher.registerNode(nodes.callModel);
// … register remaining nodes …
dispatcher.registerBundle(toolRegistry.bundle()); // urn:noocodec:tool:<name> DAGs

const dagIri = 'workflow:my-agent';
const p = (placement: string) => `${dagIri}/node/${placement}`;

const agentDag = new DAGBuilder(dagIri, '1', { name: 'my-agent' })
  .node(p('build-request'), nodes.chatRequest, {
    ready: p('call-model'),
    error: p('end-error'),
  }, { name: 'build-request' })
  .node(p('call-model'), nodes.callModel, {
    text: p('normalize-response'),
    tools: p('normalize-response'),
    mixed: p('normalize-response'),
    error: p('end-error'),
  }, { name: 'call-model' })
  // Continue with normalize, decode, worksets, scatter, collect, terminals.
  .build();
dispatcher.registerDAG(agentDag);

Subclassing each base node

Each abstract base node separates framework concerns (error wrapping, routing) from domain concerns (state reads and writes). Implement only the abstract template methods for your state shape:

Node	Template methods
`BuildChatRequestNode`	`buildRequest(state, ctx): ChatRequestType`
`CallModelNode`	`getRequest(state, ctx)`, `storeResponse(state, response, ctx)`
`NormalizeResponseNode`	`getResponse(state, ctx): ChatResponseType \| null`
`DecodeTextToolCallsNode`	`getText(state, ctx)`, `storeToolCalls(state, calls, ctx)`
`NormalizeToolCallsNode`	`getToolCalls(state, ctx)`, `writeNormalized(state, calls, ctx)`
`BuildToolWorksetsNode`	`getToolCalls`, `classifyCall`, `writeSafeWorkset`, `writeExclusiveWorkset`
`CollectToolResultsNode`	`getGatheredResults(state, ctx)`, `writeResult(state, results, ctx)`
`AppendAssistantNode`	`getResponse(state, ctx)`, `append(state, response, ctx)`

CallModelNode receives the LlmAdapterInterface via its constructor — dependency injection at the node level:

class MyCallModelNode extends CallModelNode<AgentState> {
  readonly name = 'call-model';
  constructor(llm: LlmAdapterInterface) { super(llm); }

  protected getRequest(state: AgentState, _ctx: NodeContextType): ChatRequestType {
    if (state.chatRequest === null) throw new Error('chatRequest not set');
    return state.chatRequest;
  }

  protected storeResponse(state: AgentState, response: ChatResponseType, _ctx: NodeContextType): void {
    state.chatResponse = response;
  }
}

Tool dispatch via DagReference

BuildToolWorksetsNode stamps each scatter item with dagIri: 'urn:noocodec:tool:' + encodeURIComponent(call.name). The field value is a tool DAG IRI/CURIE. The scatter placement uses { dag: { from: 'item', path: 'dagIri', candidates: [...] } } so the engine resolves the body DAG from each item at runtime and validates it against the registered tool candidates. Register tool DAGs with toolRegistry.bundle():

import { ToolRegistry } from '@studnicky/dagonizer/tool';

const tools = new ToolRegistry();
tools.register(new MyCalculatorTool());
tools.register(new MySearchTool());

dispatcher.registerBundle(tools.bundle());
// Registers tool:calculator and tool:search as embeddable DAGs.

State & metadata - use state.metadata as the inter-node IO bus
Checkpoint & resume - persist and reload state across process boundaries
Dependency injection - inject IO adapters via node constructors
Example 29: Agent DAG - working example of the 8-node agent loop with stub LLM
Chat Event Orchestration - run one registered agent DAG per inbound event or request turn
ReAct agent: streaming + provenance recall - the 8-node loop as ReAct, trace streaming, live token deltas, provenance recall
Lifecycle phases - understand when a DAG completes vs. when it pauses
ReAct Agent Memory - trace streaming, live token deltas, and graph provenance recall
ReAct Agent Routing - concurrent stream routing by routeKey

Conversational Agents

What It Is

How It Works

Diagrams, Examples, and Outputs

ReAct agent loop DAG

support-dispatcher conversation DAG

What It Lets You Do

Use when

Code Samples

Details for Nerds

Overview

Pattern 1: Request/response turn termination

How it works

Example: slot-filling trip planner

Wiring the orchestrator

Key properties

Pattern 2: Park-and-correlate handoff

How it works

Example: refund escalation queue

Wiring the orchestrator

Key properties

Choosing a pattern

Dependency injection

Common questions

Migration and adoption

Agent loop

The canonical 8-node topology

Agent DAG as JSON-LD

Subclassing each base node

Tool dispatch via DagReference

Conversational Agents ​

What It Is ​

How It Works ​

Diagrams, Examples, and Outputs ​

What It Lets You Do ​

Use when ​

Code Samples ​

Details for Nerds ​

Overview ​

Pattern 1: Request/response turn termination ​

How it works ​

Example: slot-filling trip planner ​

Wiring the orchestrator ​

Key properties ​

Pattern 2: Park-and-correlate handoff ​

How it works ​

Example: refund escalation queue ​

Wiring the orchestrator ​

Key properties ​

Choosing a pattern ​

Dependency injection ​

Common questions ​

Migration and adoption ​

Agent loop ​

The canonical 8-node topology ​

Agent DAG as JSON-LD ​

Subclassing each base node ​

Tool dispatch via DagReference ​

Related Concepts ​

Conversational Agents

What It Is

How It Works

Diagrams, Examples, and Outputs

What It Lets You Do

Use when

Code Samples

Details for Nerds

Overview

Pattern 1: Request/response turn termination

How it works

Example: slot-filling trip planner

Wiring the orchestrator

Key properties

Pattern 2: Park-and-correlate handoff

How it works

Example: refund escalation queue

Wiring the orchestrator

Key properties

Choosing a pattern

Dependency injection

Common questions

Migration and adoption

Agent loop

The canonical 8-node topology

Agent DAG as JSON-LD

Subclassing each base node

Tool dispatch via DagReference

Related Concepts