Dagonizer
Dagonizer

Distribution and Cloud ​

What It Is ​

Distribution is how a Dagonizer host moves work across threads, processes, workers, or other hosts without changing the DAG document. The graph still names placements and routes; deployment code binds logical container roles and handoff channels to concrete infrastructure.

There are two scales: in-fleet containment for worker-style isolation, and cross-host handoff for envelope-driven resume on another host.

How It Works ​

Distribution is placement-level. A scatter or embedded DAG declares a logical container role, and the host binds that role to a DagContainerInterface. Cross-host hand-off serializes state plus cursor into a DAGHandoff envelope and lets another host resume the registered DAG.

Dagonizer has two distribution scales. They solve different problems and compose independently.

ScaleMechanismCompute ownership
In-fleet containmentDagContainerInterface backends (threads, child processes)The dispatcher spawns and owns the isolates
Cross-host hand-offDAGHandoff envelope over a HandoffChannelInterfaceThe envelope travels to a separate host; that host runs the next DAG

The DAG is always the unit of distribution. A single node never travels to a container or a remote host.

Diagrams, Examples, and Outputs ​

The Cartographer worker example shows a parent DAG delegating scatter body work to a worker-bound sub-DAG. Both diagrams below are generated from the runnable worker example:

Cartographer worker parent DAG

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

stream-event worker body DAG

9 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:stream-event",
  "@type": "DAG",
  "name": "dag:stream-event",
  "version": "1.0",
  "entrypoints": {
    "main": "urn:noocodec:dag:stream-event/node/decode-payload"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:stream-event/node/decode-payload",
      "@type": "SingleNode",
      "name": "dag:stream-event/node/decode-payload",
      "node": "urn:noocodec:node:decode-payload",
      "outputs": {
        "decoded": "urn:noocodec:dag:stream-event/node/route-event-type-variant",
        "invalid": "urn:noocodec:dag:stream-event/node/rejected"
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/route-event-type-variant",
      "@type": "SingleNode",
      "name": "dag:stream-event/node/route-event-type-variant",
      "node": "urn:noocodec:node:route-event-type-variant",
      "outputs": {
        "position-ping": "urn:noocodec:dag:stream-event/node/pipeline-position-ping",
        "sensor-reading": "urn:noocodec:dag:stream-event/node/pipeline-sensor-reading",
        "customs-event": "urn:noocodec:dag:stream-event/node/pipeline-customs-event",
        "facility-scan": "urn:noocodec:dag:stream-event/node/pipeline-facility-scan",
        "delivery-confirmation": "urn:noocodec:dag:stream-event/node/pipeline-delivery-confirmation"
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-position-ping",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-position-ping",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-position-ping",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "legKm": "legKm",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-sensor-reading",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-sensor-reading",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-sensor-reading",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "coldChainBreach": "coldChainBreach",
          "legKm": "legKm",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-customs-event",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-customs-event",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-customs-event",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "customsDwellHours": "customsDwellHours",
          "legKm": "legKm",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-facility-scan",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-facility-scan",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-facility-scan",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "pricedOrder": "pricedOrder",
          "shippingQuote": "shippingQuote",
          "deliveryEstimate": "deliveryEstimate",
          "legKm": "legKm",
          "gdprResult": "gdprResult",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-delivery-confirmation",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-delivery-confirmation",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-delivery-confirmation",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "legKm": "legKm",
          "gdprResult": "gdprResult",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/done",
      "@type": "TerminalNode",
      "name": "dag:stream-event/node/done",
      "outcome": "completed"
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/rejected",
      "@type": "TerminalNode",
      "name": "dag:stream-event/node/rejected",
      "outcome": "failed"
    }
  ]
}
Mermaid generated from the same DAG
Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:stream-event (v1.0)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_stream-event/node/decode-payload
  urn_noocodec_dag_stream-event/node/decode-payload["dag:stream-event/node/decode-payload"]
  urn_noocodec_dag_stream-event/node/decode-payload -->|decoded| urn_noocodec_dag_stream-event/node/route-event-type-variant
  urn_noocodec_dag_stream-event/node/decode-payload -->|invalid| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/route-event-type-variant["dag:stream-event/node/route-event-type-variant"]
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|position-ping| urn_noocodec_dag_stream-event/node/pipeline-position-ping
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|sensor-reading| urn_noocodec_dag_stream-event/node/pipeline-sensor-reading
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|customs-event| urn_noocodec_dag_stream-event/node/pipeline-customs-event
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|facility-scan| urn_noocodec_dag_stream-event/node/pipeline-facility-scan
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|delivery-confirmation| urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation
  urn_noocodec_dag_stream-event/node/pipeline-position-ping[["dag:stream-event/node/pipeline-position-ping"]]
  urn_noocodec_dag_stream-event/node/pipeline-position-ping -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-position-ping -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-sensor-reading[["dag:stream-event/node/pipeline-sensor-reading"]]
  urn_noocodec_dag_stream-event/node/pipeline-sensor-reading -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-sensor-reading -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-customs-event[["dag:stream-event/node/pipeline-customs-event"]]
  urn_noocodec_dag_stream-event/node/pipeline-customs-event -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-customs-event -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-facility-scan[["dag:stream-event/node/pipeline-facility-scan"]]
  urn_noocodec_dag_stream-event/node/pipeline-facility-scan -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-facility-scan -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation[["dag:stream-event/node/pipeline-delivery-confirmation"]]
  urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/done((("dag:stream-event/node/done")))
  urn_noocodec_dag_stream-event/node/rejected>"dag:stream-event/node/rejected"]

What It Lets You Do ​

Use when ​

Use distribution when one DAG needs to run across threads, worker processes, or separate hosts without changing the canonical JSON-LD graph. Choose containment for in-fleet isolation; choose hand-off when state and cursor must cross a transport boundary.

Code Samples ​

The snippets below show container role binding, registry-module loading, and cross-host handoff envelopes.

Details for Nerds ​

In-fleet containment ​

An EmbeddedDAGNode or ScatterNode (dag body) placement declares a logical container role:

ts
export const dag: DAGType = {
  '@context':  DAG_CONTEXT,
  '@id': 'urn:noocodec:dag:square-all',
  '@type':     'DAG',
  "name":        'square-all',
  "version":     '1',
  "entrypoints": { "main": 'urn:noocodec:dag:square-all/node/square-all' },
  "nodes": [
    {
      '@id': 'urn:noocodec:dag:square-all/node/square-all',
      '@type':      'ScatterNode',
      "name":         'square-all',
      "body":         { "dag": 'urn:noocodec:dag:square-item' },   // scatter body: run this sub-DAG per item
      "source":       'tasks',                     // state field holding the source array
      "itemKey":      'task',                      // metadata key each item is written under
      "execution": { "mode": "item", "concurrency": 2 },                           // run up to 2 items concurrently
      "container": 'cpu',                       // route each item through the worker container
      "outputs": {
        'all-success': 'urn:noocodec:dag:square-all/node/collect-results',
        "partial": 'urn:noocodec:dag:square-all/node/collect-results',
        'all-error': 'urn:noocodec:dag:square-all/node/collect-results',
        "empty":       'urn:noocodec:dag:square-all/node/end',
      },
    },
    {
      '@id': 'urn:noocodec:dag:square-all/node/collect-results',
      '@type': 'GatherNode',
      "name": 'collect-results',
      sources: { "urn:noocodec:dag:square-all/node/square-all": {} },
      "gather": {
        "strategy": GatherStrategyNames.APPEND,
        "field": 'lastResult',
        "target": 'results',
      },
      "outputs": { "success": 'urn:noocodec:dag:square-all/node/end', "error": 'urn:noocodec:dag:square-all/node/end', "empty": 'urn:noocodec:dag:square-all/node/end' },
    },
    {
      '@id': 'urn:noocodec:dag:square-all/node/end',
      '@type':   'TerminalNode',
      "name":    'end',
      "outcome": 'completed',
    },
  ],
};

The deployment binds the logical role to a backend at dispatcher construction:

ts
const dispatcher = new Dagonizer<WorkState>({
  // Bind the 'cpu' container role. The ScatterNode's `container: 'cpu'`
  // tells the engine to route each item's sub-DAG through this container
  // instead of running it in-process.
  "containers": { "cpu": container },
});
dispatcher.registerNode(new SquareWorkerNode());
dispatcher.registerDAG(workerDag);
dispatcher.registerDAG(dag);

When the dispatcher reaches the enrich placement, it delegates the sub-DAG to the cpu backend. The child state crosses as a JSON snapshot; the worker runs the sub-DAG to completion; the terminal snapshot is applied in place. The DAG document and node code are identical in both paths. An unbound role runs in-process and fires contractWarning so the DAG still runs and the condition is visible.

Available Node.js backends ​

All four backends are in @studnicky/dagonizer-executor-node:

BackendTransportWhen to use
WorkerThreadContainerMessagePort (shared memory)CPU-bound work; fastest startup
ForkContainerIPC pipeIsolated V8 heap per request; Node stdlib available
ClusterContainerIPC pipeReuse a pre-forked cluster worker pool
SpawnContainerNDJSON over stdioPolyglot isolates; any executable that speaks the wire protocol

The browser package @studnicky/dagonizer-executor-web ships WebWorkerContainer over postMessage.

Pool sizing ​

NodeSystemInfo.recommendedWorkerCount(config) returns a cgroup-aware default based on os.availableParallelism() and available memory. Spread RecommendedWorkerCountConfigDefault and override only the fields you want to set:

ts
const sysInfo  = new NodeSystemInfo();
const poolSize = sysInfo.recommendedWorkerCount({
  ...RecommendedWorkerCountConfigDefault,
  maximumWorkers: 8,
});
process.stdout.write(`Pool sizing: recommendedWorkerCount = ${String(poolSize)} (capped at 8)\n`);

The registry module contract ​

Cross-process containers (fork, cluster, spawn, worker) dynamic-import a registry module inside the isolate to reconstruct the DAG bundle and services without crossing the boundary. The module's default export implements RegistryModuleInterface:

ts
import type { RegistryBundleInterface, RegistryModuleInterface } from '@studnicky/dagonizer/contracts';
import { CheckpointRestoreAdapter } from '@studnicky/dagonizer/checkpoint';
import type { JsonObjectType } from '@studnicky/dagonizer/entities';

import { dag, SquareWorkerNode, workerDag, WorkState } from './12-workers.js';

const registry: RegistryModuleInterface = {
  async instantiate(_servicesConfig: JsonObjectType): Promise<RegistryBundleInterface> {
    return {
      "bundle": {
        "nodes": [new SquareWorkerNode()],
        "dags":  [workerDag, dag],
      },
      "registryVersion": '1.0.0',
      "restoreState":    CheckpointRestoreAdapter.wrap((snapshot: JsonObjectType) => WorkState.restore(snapshot)),
    };
  },
};

export default registry;

registryVersion must match the string the parent passed to WorkerThreadContainer (or the equivalent backend). The DagHost inside the worker rejects an init message with a mismatched version before accepting any execute requests.

A node's dependencies never cross the isolation boundary — the isolate's registry module constructs each node with its dependencies (derived from the init message's servicesConfig) inside the isolate. If an isolate requires a network connection, it opens it locally; the parent does not proxy requests.

Trust boundary

The registryModule path passed to a container backend is dynamically imported inside the isolate with full module privileges. Pass only operator-controlled paths. Accepting a registryModule value from untrusted input (user data, external queue messages) creates a remote code execution vector.

Cross-host hand-off ​

When a top-level DAG completes at a terminal placement bound to a HandoffChannelInterface, the dispatcher publishes a DAGHandoff envelope to that channel. The envelope carries:

  • dagName — the DAG IRI/CURIE string that just completed
  • terminalName — the terminal placement label for observability
  • terminalOutput — the routing output string
  • stateSnapshot — by-value terminal state (or stateSnapshotRef for size-limited transports)
  • registryVersion — version the receiving host uses for its handshake
  • correlationId — monotonic dispatcher-assigned identifier for deduplication
  • placementPath — nesting path for instrumentation

Subclass InMemoryChannel and override the protected onPublished hook to chain a downstream DAG. InMemoryChannel carries no constructor options; passing a callback object is not supported — extension via subclass is the only mechanism. The 11-handoff example does exactly this: the override restores the envelope state and runs DAG B on a second dispatcher.

ts
class HandoffChannel extends InMemoryChannel {
  // The most recent DAG B result, written by the override after each publish.
  lastResultState: PipelineState | null = null;

  protected override async onPublished(handoff: DAGHandoffType): Promise<void> {
    // Restore the state snapshot carried by the envelope.
    // DAGHandoff is a discriminated union (stateSnapshot vs stateSnapshotRef);
    // narrow to the by-value branch before calling restore.
    if (!('stateSnapshot' in handoff)) return;
    const snapshot: unknown = handoff.stateSnapshot;
    if (!JsonObject.is(snapshot)) return;
    const continuationState = PipelineState.restore(snapshot);

    // Execute DAG B on the restored state.
    const result = await dispatcherB.execute('urn:noocodec:dag:pipeline-b', continuationState);
    this.lastResultState = result.state;
  }
}

const channel = new HandoffChannel();

The dispatcher binds the channel to a terminal name; reaching that terminal publishes a DAGHandoff envelope to it:

ts
const dispatcherA = new Dagonizer<PipelineState>({
  // Bind the "handoff" terminal to the channel.
  // When DAG A reaches the terminal named "handoff" the engine publishes
  // a DAGHandoff envelope to this channel.
  "channels": { "handoff": channel },
});
dispatcherA.registerNode(new CollectANode());
dispatcherA.registerNode(new CollectBNode());
dispatcherA.registerNode(new CollectCNode());
dispatcherA.registerDAG(dagA);

A terminal not listed in channels follows today's behavior — the run completes, no envelope is published. Embedded and contained child DAGs never publish; only the top-level host does. Run the full producer → channel → downstream-host chain with npx tsx examples/11-handoff.ts.

A publish failure collects a HANDOFF_PUBLISH_FAILED error on the run's state; the returned ExecutionResult and terminalOutcome are unchanged.

Serverless function handler pattern ​

A serverless function receives a DAGHandoff envelope, restores state, runs the DAG to completion, and lets the bound egress channels publish the next envelope. The function itself requires no Dagonizer-specific runtime; it is a plain Dagonizer instance.

The egress channel implements HandoffChannelInterface. The channel below backs the contract with a real in-process queue — a complete, runnable implementation. Production replaces the array push with an SQS / Pub/Sub / RabbitMQ SDK call; the method signature is identical:

ts
export class InMemoryQueueChannel implements HandoffChannelInterface {
  readonly #queue: DAGHandoffType[];

  constructor(queue: DAGHandoffType[]) {
    this.#queue = queue;
  }

  async publish(handoff: DAGHandoffType): Promise<void> {
    // Production: await sqsClient.send(new SendMessageCommand({ ... }));
    this.#queue.push(handoff);
  }
}

The handler is envelope-in / envelope-out: verify the version, restore state, build a per-invocation dispatcher with egress channels bound to terminal names, execute:

ts
export class ServerlessHandler {
  static async handle(
    envelope: DAGHandoffType,
    egress: HandoffChannelInterface,
  ): Promise<OrderState> {
    // 1. Verify version before executing.
    if (envelope.registryVersion !== REGISTRY_VERSION) {
      throw new Error(`Version mismatch: expected ${REGISTRY_VERSION}, got ${envelope.registryVersion}`);
    }

    // 2. Restore state. DAGHandoff is a oneOf: stateSnapshot (by-value) or
    //    stateSnapshotRef (by-reference URI). Narrow before restore.
    if (!('stateSnapshot' in envelope)) {
      throw new Error('stateSnapshotRef envelopes require fetching the URI before restore');
    }
    const snapshot: unknown = envelope.stateSnapshot;
    if (!JsonObject.is(snapshot)) {
      throw new Error('stateSnapshot is not a JSON object');
    }
    const state = OrderState.restore(snapshot);

    // 3. Build a per-invocation dispatcher with egress channels bound to terminal
    //    names. The channel publishes the next envelope after the terminal.
    const dispatcher = new Dagonizer<OrderState>({ channels: { done: egress } });
    dispatcher.registerNode(new SettleNode());
    dispatcher.registerDAG(settleDag);

    // 4. Execute. The dispatcher is constructed, used, and discarded per call.
    const result = await dispatcher.execute(SETTLE_DAG_IRI, state);
    return result.state;
  }
}

This pattern composes with any function-as-a-service platform (AWS Lambda, Cloud Run, Cloudflare Worker). The serverless-handler example drives a complete envelope through the handler and observes the downstream envelope land in the queue; run it with npx tsx examples/serverless-handler.ts. The dispatcher is constructed, used, and discarded per invocation. There is no persistent scheduler, no long-lived worker, and no agent protocol — Dagonizer is the in-function runtime.

External orchestrators (Step Functions, Cloud Workflows, etc.) ​

When the deployment uses an external state machine (AWS Step Functions, Google Cloud Workflows, Azure Durable Functions), each state maps to one function invocation. The function is envelope-in / envelope-out:

[State machine]
   │
   ├─ InvokeA → handler(envelopeA) → channel.publish(envelopeB)
   ├─ InvokeB → handler(envelopeB) → channel.publish(envelopeC)
   └─ InvokeC → handler(envelopeC) → done

Dagonizer never compiles a DAG to ASL or workflow YAML. Dagonizer never invokes or manages state machine instances. The state machine is authored by the deployment; Dagonizer is the in-function executor that processes each step's envelope and publishes the next. The boundary is explicit: the orchestrator owns transitions; Dagonizer owns node execution within each step.

registryVersion handshake ​

The registryVersion field on DAGHandoff carries the same guarantee as the bridge protocol used by containers: a receiving host that detects a version mismatch should reject the envelope before executing. This prevents a host running a stale bundle from applying state from a newer bundle's run.

Recommended pattern:

  • Publish registryVersion from a single constant in the bundle (e.g. package.json version or a content hash of the DAG registry).
  • Validate on ingress before calling AppState.restore.
  • On mismatch: route to a DLQ or a manual-review queue, not a silent drop.

Idempotent node authoring ​

Exactly-once delivery is out of scope for Dagonizer's channel contract. At-least-once is the documented delivery guarantee for most queue transports. Node authors are responsible for writing idempotent side effects:

  • Upsert records rather than insert-and-fail on duplicate.
  • Check for prior completion in state before calling an external API.
  • Use correlationId from the envelope as an idempotency key when the downstream API supports one.

Dagonizer guarantees envelope fidelity (a stateSnapshot round-trip is a fixed point) and correlationId uniqueness within a dispatcher instance. Delivery semantics above that are a property of the channel transport and the deployment.

Trust boundaries ​

stateSnapshotRef URI dereference. When an envelope carries stateSnapshotRef instead of stateSnapshot, the receiver must fetch the snapshot from the referenced URI. The receiver owns SSRF and allowlist responsibility: validate that the URI resolves to an operator-controlled storage backend (S3 bucket, GCS object, internal blob store) before fetching. Dagonizer does not fetch stateSnapshotRef values; the fetch and the allowlist check are deployment code.

Incoming state from workers. State-snapshot keys that arrive from a worker or a remote host are untrusted-shaped. If your restoreData override reads state keys from a snapshot, treat incoming keys defensively — validate schema, coerce types, and apply defaults before trusting the values. The engine does not validate the shape of individual state fields; the JsonObjectType constraint only guarantees the top-level is an object.

Further reading ​

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky