Dagonizer
Dagonizer

ReAct Agent Routing ​

What It Is ​

ReAct Agent Routing demultiplexes concurrent token streams through a DAG. One shared RoutingCallModelNode and one shared StreamChannel<RoutedChatStreamChunkType> serve two conversations; a routing DAG scatters over the channel and writes each chunk to the transcript keyed by routeKey.

The sink is not a passive callback map. It is a DAG source, so routing, buffering, and per-conversation transcript writes stay inside the same graph machinery used elsewhere.

How It Works ​

The shared model node writes routed chunks into a StreamChannel. The routing DAG consumes that channel as a scatter source, reads each chunk's routeKey, and appends the delta to the matching transcript. The channel is drained before the conversations begin, so concurrent producers do not deadlock on bounded buffer backpressure.

Each chunk carries its route key and source metadata. The routing DAG does not trust timing or ordering to separate conversations; it routes by payload.

Diagrams, Examples, and Outputs ​

DAG registration and diagram ​

The routing example registers the same canonical ReAct agent loop, then starts a second routing DAG that scatters over the shared StreamChannel<RoutedChatStreamChunkType>.

ReAct agent loop DAG

12 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: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"]

ReAct routed stream sink DAG

3 placements
DAG JSON-LD registered with the dispatcher
{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:react-agent-routing",
  "@type": "DAG",
  "name": "react-agent-routing",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:react-agent-routing/node/scatter-chunks"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:react-agent-routing/node/scatter-chunks",
      "@type": "ScatterNode",
      "name": "scatter-chunks",
      "body": {
        "node": "urn:noocodec:node:route-chunk"
      },
      "source": "source",
      "itemKey": "routed-chunk-item",
      "execution": {
        "mode": "item",
        "concurrency": 4
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
        "partial": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
        "all-error": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
        "empty": "urn:noocodec:dag:react-agent-routing/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
      "@type": "GatherNode",
      "name": "collect-chunks",
      "sources": {
        "urn:noocodec:dag:react-agent-routing/node/scatter-chunks": {}
      },
      "gather": {
        "strategy": "append",
        "target": "chunks"
      },
      "outputs": {
        "success": "urn:noocodec:dag:react-agent-routing/node/end",
        "error": "urn:noocodec:dag:react-agent-routing/node/end",
        "empty": "urn:noocodec:dag:react-agent-routing/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent-routing/node/end",
      "@type": "TerminalNode",
      "name": "end",
      "outcome": "completed"
    }
  ]
}
Mermaid generated from the same DAG
Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% react-agent-routing (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_react-agent-routing/node/scatter-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks[/"scatter-chunks"/]
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|all-success| urn_noocodec_dag_react-agent-routing/node/collect-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|partial| urn_noocodec_dag_react-agent-routing/node/collect-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|all-error| urn_noocodec_dag_react-agent-routing/node/collect-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|empty| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/collect-chunks{"collect-chunks"}
  urn_noocodec_dag_react-agent-routing/node/collect-chunks -->|success| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/collect-chunks -->|error| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/collect-chunks -->|empty| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/end((("end")))

The first graph is the producer of streamed model chunks. The second graph is the sink-as-DAG: every routed chunk becomes one scatter item, and RouteChunkNode demultiplexes it into the transcript store by routeKey.

RoutingCallModelNode wraps its constructor-bound { sink } in fresh RoutingStreamSink instances for the batch items it processes, stamping every pushed chunk with routeKey(state) and { dagName, nodeName }. A single registered node instance serves both concurrent conversations.

Run ​

bash
npx tsx examples/react-agent-routing.ts

Output ​

--- react-agent-routing: two concurrent conversations, one shared sink ---

Total routed chunks observed: 26

c1 transcript: "Based on the lookup, {"result":"Dagonizer is a type-safe, abortable DAG dispatcher for TypeScript."} "
c2 transcript: "Based on the lookup, {"result":"Dagonizer is a type-safe, abortable DAG dispatcher for TypeScript."} "

Route keys with recorded transcripts: ["c1","c2"]

Lesson: one shared StreamChannel, fed by one shared RoutingCallModelNode instance,
        carried BOTH conversations' chunks interleaved. A routing DAG scattering
        over that same channel classified each chunk by its stamped routeKey and
        demultiplexed it into a separate, uncontaminated transcript per conversation
        — the sink itself is a DAG that routes by payload, not a passive buffer.

ScriptedAdapter's final answer text is prompt-INSENSITIVE (a fixed tool observation), so c1 and c2's printed transcripts are legitimately identical strings — that alone is not proof of correct routing. The actual proof is at the chunk level (see the test suite): every one of the 26 routed chunks is owned by exactly one routeKey, and each conversation's recorded transcript equals the exact concatenation of only its own chunks, with zero loss and zero cross-contamination under concurrent execution on a single shared node instance and a single shared sink.

What It Lets You Do ​

ReAct agent routing lets applications demultiplex concurrent model token streams through a DAG instead of bespoke callback maps. Use it when one shared model node and one shared sink serve multiple conversations that must remain separated by route key.

Reuses AgentState and the eight agent-loop node subclasses from Example: ReAct agent memory unchanged, and adds one subclassed node (RoutingCallModelNode, overriding routeKey(state)) plus a second, small DAG that scatters over a shared StreamChannel to demultiplex two conversations running concurrently against ONE shared node instance and ONE shared sink.

Code Samples ​

The executable entry point starts the routing drain, runs two conversations concurrently, then closes the shared channel:

ts
/**
 * react-agent-routing: ONE shared `RoutingCallModelNode` and ONE shared
 * `StreamChannel<RoutedChatStreamChunkType>` sink serve TWO concurrent
 * conversations. Every chunk the node streams is self-describing
 * (`routeKey` + `source`), so a routing DAG scattering over the shared
 * channel demultiplexes the interleaved chunks back into separate
 * per-conversation transcripts — proving the sink itself can classify and
 * route by payload, not just buffer.
 *
 * DAG definitions + reusable classes: examples/dags/react-agent-routing.ts
 * (which reuses `AgentState` and the eight agent-loop node subclasses from
 * examples/dags/react-agent-memory.ts unchanged).
 *
 * Run: npx tsx examples/react-agent-routing.ts
 */

import { Dagonizer } from '@studnicky/dagonizer';
import { StreamChannel } from '@studnicky/dagonizer/channels';
import type { RoutedChatStreamChunkType } from '@studnicky/dagonizer/adapter';
import { ToolRegistry } from '@studnicky/dagonizer/tool';
import {
  agentDag,
  AgentState,
  LookupTool,
  MyAppendAssistantNode,
  MyBuildChatRequestNode,
  MyBuildToolWorksetsNode,
  MyCollectToolResultsNode,
  MyDecodeTextToolCallsNode,
  MyNormalizeResponseNode,
  MyNormalizeToolCallsNode,
  ScriptedAdapter,
} from './dags/react-agent-memory.js';
import {
  RouteChunkNode,
  RoutingCallModelNode,
  routingDag,
  RoutingState,
  TranscriptStore,
} from './dags/react-agent-routing.js';

// ---------------------------------------------------------------------------
// RoutingRunHarness: wires the shared sink, the shared node, and the routing
// DAG's concurrent drain in the order that avoids deadlock.
// ---------------------------------------------------------------------------

class RoutingRunHarness {
  private constructor() { /* static class */ }

  static async run(): Promise<{ transcripts: TranscriptStore; chunks: readonly RoutedChatStreamChunkType[] }> {
    // ONE shared sink every routed chunk from either conversation lands on.
    const channel = new StreamChannel<RoutedChatStreamChunkType>();

    // ONE shared agent dispatcher: one RoutingCallModelNode instance, wired
    // to the shared channel, plus the other seven agent-loop node subclasses.
    const llm = new ScriptedAdapter();
    const agentDispatcher = new Dagonizer<AgentState>();
    agentDispatcher.registerNode(new MyBuildChatRequestNode());
    agentDispatcher.registerNode(new RoutingCallModelNode(llm, { 'sink': channel }));
    agentDispatcher.registerNode(new MyNormalizeResponseNode());
    agentDispatcher.registerNode(new MyDecodeTextToolCallsNode());
    agentDispatcher.registerNode(new MyNormalizeToolCallsNode());
    agentDispatcher.registerNode(new MyBuildToolWorksetsNode());
    agentDispatcher.registerNode(new MyCollectToolResultsNode());
    agentDispatcher.registerNode(new MyAppendAssistantNode());

    const tools = new ToolRegistry();
    tools.register(new LookupTool());
    agentDispatcher.registerBundle(tools.bundle());
    agentDispatcher.registerDAG(agentDag);

    // A SEPARATE routing dispatcher: one RouteChunkNode writing into ONE
    // shared TranscriptStore, scattering over the same shared channel.
    const transcripts = new TranscriptStore();
    const routingDispatcher = new Dagonizer<RoutingState>();
    routingDispatcher.registerNode(new RouteChunkNode(transcripts));
    routingDispatcher.registerDAG(routingDag);

    const routingState = new RoutingState();
    routingState.source = channel;

    // LIFECYCLE ORDER (critical — avoids deadlock):
    //
    // 1. Start the routing drain FIRST, without awaiting. It begins pulling
    //    from `channel` immediately, so the (256-item default) buffer never
    //    fills — pushes from either conversation never block.
    const routingDone = routingDispatcher.execute('urn:noocodec:dag:react-agent-routing', routingState);

    // 2. Run two conversations CONCURRENTLY against the ONE shared node and
    //    ONE shared channel, each seeded with a distinct conversationId.
    const stateFor = (conversationId: string, prompt: string): AgentState => {
      const state = new AgentState();
      state.conversationId = conversationId;
      state.prompt = prompt;
      return state;
    };

    await Promise.all([
      agentDispatcher.execute('urn:noocodec:dag:react-agent', stateFor('c1', 'Tell me about dagonizer.')),
      agentDispatcher.execute('urn:noocodec:dag:react-agent', stateFor('c2', 'Tell me about dagonizer, briefly.')),
    ]);

    // 3. Only after both conversations finish pushing does the sink close —
    //    this is what ends the routing scatter's async-iterable source.
    channel.close();

    // 4. Now the routing drain can be awaited to completion.
    await routingDone;

    return { transcripts, 'chunks': routingState.chunks };
  }
}

// ---------------------------------------------------------------------------
// Run + report
// ---------------------------------------------------------------------------

process.stdout.write('--- react-agent-routing: two concurrent conversations, one shared sink ---\n\n');

const { transcripts, chunks } = await RoutingRunHarness.run();

process.stdout.write(`Total routed chunks observed: ${String(chunks.length)}\n\n`);

process.stdout.write(`c1 transcript: "${transcripts.transcript('c1')}"\n`);
process.stdout.write(`c2 transcript: "${transcripts.transcript('c2')}"\n\n`);

process.stdout.write(`Route keys with recorded transcripts: ${JSON.stringify(transcripts.keys())}\n\n`);

process.stdout.write('Lesson: one shared StreamChannel, fed by one shared RoutingCallModelNode instance,\n');
process.stdout.write('        carried BOTH conversations\' chunks interleaved. A routing DAG scattering\n');
process.stdout.write('        over that same channel classified each chunk by its stamped routeKey and\n');
process.stdout.write('        demultiplexed it into a separate, uncontaminated transcript per conversation\n');
process.stdout.write('        — the sink itself is a DAG that routes by payload, not a passive buffer.\n');

DAG definitions and reusable classes ​

The DAG module defines RoutingCallModelNode, the shared TranscriptStore, RouteChunkNode, and the routing scatter DAG:

ts
/**
 * react-agent-routing/dags: the streaming SINK is itself a DAG — one shared
 * `StreamChannel<RoutedChatStreamChunkType>` demultiplexes CONCURRENT agent
 * runs by feeding a routing DAG that scatters over the channel and classifies
 * each chunk by `routeKey` into a per-conversation transcript.
 *
 * Pure module — no side effects at import time; every export is a class, a
 * state, a node, or a pre-assembled DAG. Reuses `AgentState` and the eight
 * concrete agent-loop node subclasses from `examples/dags/react-agent-memory.ts`
 * unchanged — this module adds only what routing needs on top of them:
 *
 *   - `RoutingCallModelNode` — the SAME `call-model` node, subclassed to
 *     override `routeKey(state)` with a per-run id (`state.conversationId`).
 *     One instance of this node, wired to one shared sink, is safe to run
 *     concurrently for many conversations: `CallModelNode.execute` wraps
 *     the shared sink in a fresh `RoutingStreamSink` per execution, so every
 *     chunk it pushes downstream already carries `routeKey` + `source`.
 *   - `TranscriptStore` — a plain shared bucket (`Map<routeKey, string[]>`),
 *     constructor-injected into the routing DAG's body node, mirroring how
 *     `react-agent-memory`'s `RecordReasoningStepNode` writes into a shared
 *     `RdfStore` rather than folding through gather.
 *   - `RouteChunkNode` — the routing DAG's scatter body: reads the current
 *     `RoutedChatStreamChunkType` item (stamped by the scatter under
 *     `ROUTE_ITEM_KEY`) and appends its `delta` into `TranscriptStore` under
 *     `item.routeKey`. This is the classify-and-route step: the DAG reads a
 *     payload field and dispatches to a destination bucket by that field's
 *     value.
 *   - `routingDag` — a `ScatterNode` over `RoutingState.source` (the shared
 *     `StreamChannel`, itself an `AsyncIterable<RoutedChatStreamChunkType>`),
 *     body `route-chunk`, then `collect-chunks` append gather
 *     logging every routed chunk on `RoutingState` for inspection/assertions
 *     (source-stamp checks) without participating in the demux itself — the
 *     demux happens via `TranscriptStore`, not through gather.
 *
 * The runnable entry point (`examples/react-agent-routing.ts`) wires ONE
 * shared `StreamChannel` as the sink for ONE shared `RoutingCallModelNode`,
 * runs TWO conversations concurrently against it, and drains the routing DAG
 * over the same channel to reconstruct each conversation's transcript
 * separately — proving the sink can BE a DAG that classifies and routes by
 * payload, not just a passive buffer.
 */

import { Batch, MonadicNode, NodeOutput, NodeStateBase, RoutedBatch, Validator } from '@studnicky/dagonizer';
import type { DAGType, EntityValidatorInterface, SchemaObjectType } from '@studnicky/dagonizer';
import { RoutedChatStreamChunkSchema } from '@studnicky/dagonizer/adapter';
import type { RoutedChatStreamChunkType } from '@studnicky/dagonizer/adapter';
import { DAG_CONTEXT } from '@studnicky/dagonizer';
import { AgentState, MyCallModelNode } from './react-agent-memory.ts';

// ---------------------------------------------------------------------------
// RoutingCallModelNode: the SAME call-model node, routed per conversation
// ---------------------------------------------------------------------------

/**
 * Overrides `routeKey` to read `state.conversationId` — the per-run id each
 * concurrent conversation seeds before calling `dispatcher.execute`. Every
 * other behavior (request/response wiring) is inherited unchanged from
 * `MyCallModelNode`.
 */
export class RoutingCallModelNode extends MyCallModelNode {
  protected override routeKey(state: AgentState): string {
    return state.conversationId;
  }
}

// ---------------------------------------------------------------------------
// TranscriptStore: shared per-routeKey bucket the routing DAG writes into
// ---------------------------------------------------------------------------

/**
 * Plain shared accumulator, constructor-injected into `RouteChunkNode` —
 * mirrors `RdfStore` injection in `react-agent-memory`. Not a state field:
 * the routing DAG's scatter body runs on a per-item clone, so the
 * cross-item, cross-route accumulation lives here instead of on state.
 */
export class TranscriptStore {
  readonly #transcripts: Map<string, string[]>;

  constructor() {
    this.#transcripts = new Map();
  }

  /** Append `delta` to the transcript for `routeKey`. */
  append(routeKey: string, delta: string): void {
    const existing = this.#transcripts.get(routeKey) ?? [];
    existing.push(delta);
    this.#transcripts.set(routeKey, existing);
  }

  /** The concatenated transcript for `routeKey`; `''` when no deltas were routed. */
  transcript(routeKey: string): string {
    return (this.#transcripts.get(routeKey) ?? []).join('');
  }

  /** Every route key that received at least one delta. */
  keys(): readonly string[] {
    return [...this.#transcripts.keys()];
  }
}

// ---------------------------------------------------------------------------
// RoutingState: the routing DAG's scatter state
// ---------------------------------------------------------------------------

export class RoutingState extends NodeStateBase {
  /** The shared `StreamChannel<RoutedChatStreamChunkType>` — the sink AND the scatter source. */
  source: AsyncIterable<RoutedChatStreamChunkType> | null = null;
  /** Every routed chunk seen, in arrival order (gather 'append') — inspection/assertion log, not the demux itself. */
  chunks: RoutedChatStreamChunkType[] = [];
}

// ---------------------------------------------------------------------------
// RouteChunkNode: classify each chunk by routeKey, route it into TranscriptStore
// ---------------------------------------------------------------------------

/** Metadata key the scatter stamps each `RoutedChatStreamChunkType` item under. */
export const ROUTE_ITEM_KEY = 'routed-chunk-item';

const CHUNK_VALIDATOR: EntityValidatorInterface<RoutedChatStreamChunkType> = Validator.compile<RoutedChatStreamChunkType>(RoutedChatStreamChunkSchema);

/**
 * The routing DAG's scatter body: reads the current scattered
 * `RoutedChatStreamChunkType`, and appends its `delta` into the shared
 * `TranscriptStore` under `item.routeKey` — the classify-and-route step that
 * demultiplexes concurrent agent runs sharing one sink.
 */
export class RouteChunkNode extends MonadicNode<RoutingState, 'done'> {
  readonly name = 'route-chunk';
  readonly '@id' = 'urn:noocodec:node:route-chunk';
  readonly outputs = ['done'] as const;

  readonly #transcripts: TranscriptStore;

  constructor(transcripts: TranscriptStore) {
    super();
    this.#transcripts = transcripts;
  }

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

  override async execute(batch: Batch<RoutingState>) {
    for (const item of batch) {
      const raw = item.state.getMetadata(ROUTE_ITEM_KEY);
      if (CHUNK_VALIDATOR.is(raw)) {
        this.#transcripts.append(raw.routeKey, raw.delta);
      }
    }
    return RoutedBatch.create(NodeOutput.create('done').output, batch);
  }
}

// ---------------------------------------------------------------------------
// routingDag: scatter over the shared StreamChannel, classifying by routeKey
// ---------------------------------------------------------------------------

// #region react-routing-dag
export const routingDag: DAGType = {
  '@context':   DAG_CONTEXT,
  '@id': 'urn:noocodec:dag:react-agent-routing',
  '@type':      'DAG',
  'name':       'react-agent-routing',
  'version':    '1',
  'entrypoints': { 'main': 'urn:noocodec:dag:react-agent-routing/node/scatter-chunks' },
  'nodes': [
    {
      '@id': 'urn:noocodec:dag:react-agent-routing/node/scatter-chunks',
      '@type':       'ScatterNode',
      'name':        'scatter-chunks',
      'body':        { 'node': 'urn:noocodec:node:route-chunk' },
      'source':      'source',
      'itemKey':     ROUTE_ITEM_KEY,
      // Chunks route by their own `routeKey`; TranscriptStore.append is safe
      // under any interleaving (single-threaded event loop), so raising this
      // is a free performance choice, not a correctness requirement.
      'execution': { 'mode': 'item', 'concurrency': 4 },
      'outputs': {
        'all-success': 'urn:noocodec:dag:react-agent-routing/node/collect-chunks',
        'partial': 'urn:noocodec:dag:react-agent-routing/node/collect-chunks',
        'all-error': 'urn:noocodec:dag:react-agent-routing/node/collect-chunks',
        'empty': 'urn:noocodec:dag:react-agent-routing/node/end',
      },
    },
    {
      '@id': 'urn:noocodec:dag:react-agent-routing/node/collect-chunks',
      '@type': 'GatherNode',
      'name': 'collect-chunks',
      sources: { 'urn:noocodec:dag:react-agent-routing/node/scatter-chunks': {} },
      'gather': { 'strategy': 'append', 'target': 'chunks' },
      'outputs': {
        'success': 'urn:noocodec:dag:react-agent-routing/node/end',
        'error': 'urn:noocodec:dag:react-agent-routing/node/end',
        'empty': 'urn:noocodec:dag:react-agent-routing/node/end',
      },
    },
    {
      '@id': 'urn:noocodec:dag:react-agent-routing/node/end',
      '@type':   'TerminalNode',
      'name':    'end',
      'outcome': 'completed',
    },
  ],
};
// #endregion react-routing-dag

Details for Nerds ​

  • One node instance, many concurrent runs. CallModelNode.execute wraps the shared sink per execution so route metadata is stamped per conversation.
  • The sink is itself a DAG. RoutedChatStreamChunkType is a JSON Schema-derived entity like any other, so the same StreamChannel passed to CallModelNode as { sink } is ALSO passed as a ScatterNode's source. The scatter's body (RouteChunkNode) classifies each item by its own routeKey field and routes it into a per-conversation destination (TranscriptStore.append) — this is the same DagStreamProducer → StreamChannel → outer-scatter idiom used everywhere else in the docs, fed by a live concurrently-written sink instead of a single producer.
  • Deadlock-free lifecycle ordering. The routing DAG's drain is started (not awaited) BEFORE the two conversations run, so the channel's bounded buffer is always being drained and push never blocks. The channel is closed only after BOTH conversations finish pushing, which is what ends the routing scatter's async-iterable source; only then is the routing drain awaited.
  • Behaviorally verified demultiplexing. examples/tests/react-agent-routing.test.ts asserts every routed chunk is owned by exactly one conversation, that TranscriptStore.keys() is exactly {c1, c2}, that every chunk's source is stamped {dagName, nodeName}, and that each conversation's recorded transcript equals the exact concatenation of only its own routed chunks.

Topology ​

channel: StreamChannel<RoutedChatStreamChunkType>       ← shared sink AND shared scatter source
  → RoutingCallModelNode (ONE instance, { sink: channel })
      conversation 'c1' → adapter.chatStream(...) → RoutingStreamSink → channel.push({routeKey:'c1', delta, source})
      conversation 'c2' → adapter.chatStream(...) → RoutingStreamSink → channel.push({routeKey:'c2', delta, source})
  → routing DAG ('react-agent-routing'): ScatterNode ('scatter-chunks', concurrency 4)
      → body: RouteChunkNode ('route-chunk')
          → TranscriptStore.append(item.routeKey, item.delta)

Lifecycle, in order:

  1. routingDispatcher.execute('urn:noocodec:dag:react-agent-routing', routingState) starts (not awaited) — the routing scatter begins pulling from channel immediately.
  2. Promise.all([agentDispatcher.execute('urn:noocodec:dag:react-agent', c1State), agentDispatcher.execute('urn:noocodec:dag:react-agent', c2State)]) runs both conversations concurrently against the one shared node and channel.
  3. channel.close() — only after both conversations finish pushing.
  4. await routingDone — the routing drain completes once the channel is closed and empty.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky