LangGraph Integration

This document is a top-down walkthrough of the Dragon backend for LangGraph for developers who will maintain and extend it. It starts with the big picture — what the integration does and how the code is organized — then traces a single node invocation through the system end-to-end, and finally drills into each component in the order a request touches it.

For the Python API reference, see LangGraph Integration.

1 — What This Integration Does

The Dragon LangGraph backend lets an existing LangGraph StateGraph run across the nodes of an HPC cluster without changing the graph definition or the node functions. The user writes ordinary LangGraph code; the only difference is that each node function executes inside a persistent Dragon AgentHost process rather than in the local interpreter — a sync (def) node on the host’s thread pool, an async (async def) node on the host’s event loop. Agents co-located on the same host share memory, LLM clients, and other heavyweight resources.

The design splits cleanly into a control plane, an execution plane, and a data plane:

• Control plane (owned by this integration). Node dispatch, result transport, completion signaling, backpressure, and result routing. Agent results travel inline as small cloudpickle envelopes over shared Dragon Channels. This is how a task gets there and how its result comes back.

• Execution plane (shared — integration provides the host, user provides the compute). Where the node function actually runs: a persistent Dragon AgentHost process, its thread pool, agent co-location and resource sharing, Policy placement, and any Process / ProcessGroup / Batch the agent itself launches — and supervises and aborts mid-flight. This is where the science actually runs, and how it scales out and is steered.

• Data plane (owned by the user). Bulk scientific data (tensors, arrays, datasets) belongs in a user-managed DDict. Agent functions store artifacts there and return only a lightweight handle/key through LangGraph state. This integration never creates or manages a DDict, and never moves bulk bytes through its transport. This is where large data lives and moves.

In one line: control = signaling, execution = compute, data = storage.

LangGraph keeps full ownership of topology, routing, and checkpointing; Dragon owns process placement, IPC, and completion signaling. The two layers meet at exactly one seam — a fn(state) -> dict callable per node.

Why Dragon on HPC

Agentic AI and HPC science are converging. The graph that decides what to do next — which simulation to launch, which region to refine, which candidate to keep — increasingly sits next to petascale compute and the multi-gigabyte artifacts it produces. The hard part is no longer “call an LLM”; it is keeping the agent layer next to the data, so a decision does not cost a network round-trip of every result. Dragon addresses this with the in-situ agent — the deciding graph runs in place, inside the allocation, next to the data and compute it reasons about.

Today this means integrating several separate systems. Plain LangGraph runs every node in the driver process (threads for invoke, asyncio tasks for ainvoke): it is bounded by one node’s cores and one GIL, with no way to place work near data. The usual workaround stitches together a workflow layer (such as Parsl), a central store (such as Redis) so agents exchange results by key, and a separate inference server for the LLMs — three or four systems to operate, and results still pass by value through the store, so every GB-scale artifact is serialized and copied over the network, often several times.

In-situ agents run inside the allocation, next to the data. The graph, the agents, the simulations, and the data all live in the same Dragon runtime spanning every node you were allocated. Backing each node with a Dragon AgentHost process, the same graph now:

• Spans the allocation. launch_host(..., policy=...) pins each agent host to a specific node, NUMA zone, or GPU set (through a Policy). Because you decide where each host lands, an agent can run on the same node where its data already lives.

• Runs agents in real parallel, and lets them drive HPC jobs. Each node function runs inside a host process out on the allocation, rather than on the driver’s single interpreter. A step that fans out to many agents therefore executes concurrently across many nodes, instead of being serialized by one Python GIL. An agent can also launch its own Dragon ProcessGroup or Batch job of hundreds of ranks, supervise it, and abort it early if the partial results look wrong. In other words, a single decision node in the graph can steer a full-scale simulation.

• Keeps bulk data node-local. When an agent produces a large result, it writes that result to the node-local DDict manager, which lives in shared memory on the same node, so the write needs no serialization and no network copy. The agent then returns only a small handle (a key) through LangGraph state. The orchestrator passes handles between nodes and never moves the gigabytes themselves.

• Moves data over the HPC fabric, not TCP. If a later agent has to read data that lives on another node, Dragon transfers it over the cluster interconnect (RDMA/HSTA) rather than the slower kernel TCP stack.

• Shares heavyweight resources. Agents that sit in the same host share one address space, so expensive objects such as LLM clients, model weights, and database connections are loaded once and reused by every agent on that host.

• Runs inference on the same runtime. LLM serving can run inside the allocation through the Dragon inference service (on-cluster vLLM), reached over Dragon Queues and Channels. There is no separate inference server to stand up, secure, and route traffic to.

• Keeps the coordinator thin. Only small result envelopes travel back through the completion path, so the single coordinator node stays lightly loaded even when a step fans out to many agents at once.

The result is one runtime instead of a stack of separate systems — workflow execution, node-local data, and LLM inference all run on Dragon. And your graph does not change: the same StateGraph runs on top; only the substrate underneath does.

Note

What the runtime provides vs. what this integration adds. Running work as real processes across the allocation, launching and supervising a ProcessGroup/Batch, node-local zero-copy data, data-aware placement, and RDMA transport are all properties of the Dragon runtime (ProcessGroup + a node Policy + DDict) — you get them with or without LangGraph. This integration adds a zero-rewrite path onto that substrate: keep your existing StateGraph — topology, conditional edges, checkpointing, state — and run it on Dragon, with the single-process orchestrator no longer executing every agent itself. The execution-plane benefit — true cross-node parallelism instead of one GIL, and agents that can drive full HPC jobs — applies to any multi-agent graph; the data-plane benefit is largest for data-heavy science, while for small JSON/text results a central store works just as well.

Architecture at a glance

One coordinator process (the driver running your graph) talks to one or more persistent AgentHost processes spread across the allocation. A single host can host many agents — each task message carries a node_name, and the host looks it up in its agents dict (agents[node_name]) to route the task to the right agent function. The coordinator dispatches a task by putting a message on a host’s input Queue; the host resolves the target agent by node_name and runs that agent function (async on its event loop, sync on a thread pool), then sends a small result envelope back over a shared completion Channel, where a single watcher thread routes it to the waiting Future. The agent function runs on the execution plane — and may itself launch supervised Dragon compute. Bulk data never travels the control path — it lives in a user-managed DDict on the compute nodes.

In the diagram below, the LangGraph Pregel engine is LangGraph’s own, built-in node scheduler — the thing graph.invoke() drives to decide which nodes run and to merge their outputs back into state. It is unchanged by this integration; DragonExecutor and DragonWatcher are the two pieces this integration adds around it.

Green arrows are the control plane (this integration); the orange arrow is the execution plane (the agent launching and steering real compute); dotted blue arrows are the data plane (the user’s DDict). The rest of this guide follows these arrows in order.

2 — Code Organization

All source lives under src/dragon/ai/langgraph/:

Module	Responsibility
__init__.py	Public surface. Exports DragonExecutor and get_llm().
executor.py	DragonExecutor — lifecycle manager. Spawns hosts, registers one DragonAgentNode per agent, and returns LangGraph-compatible callables.
_llm.py	get_llm — per-host, per-pipeline accessor for a Dragon Inference DragonQueueLLMProxy. Builds the (loop-bound) proxy lazily inside the host and caches it by inference queue.
_host.py	agent_host_entry — the function that runs inside each Dragon process. Listens for task messages and runs each agent function on a shared event loop (sync nodes offloaded to a thread pool).
_node.py	DragonAgentNode — the per-node callable (sync __call__ and async acall) that LangGraph invokes; delegates to the watcher.
_watcher.py	DragonWatcher — the blocking-recv result router. Owns the shared completion Channel(s) and resolves task futures.
_constants.py	Completion-envelope field names and status codes shared by host and watcher so the two sides never drift.

3 — End-to-End: One Node Invocation

When LangGraph executes a node backed by this integration, a task flows from the driver to a host and a result envelope flows back. The sequence below shows the full round trip:

Before reading the trace, it helps to know what the watcher is. The watcher is the coordinator-side component that collects results coming back from the hosts — it does not dispatch jobs (jobs are sent over each host’s input Queue, steps 3–4 below). Its whole job is the return path: receive each result envelope, deserialize it, and hand the result to the call that is waiting for it.

To match a returning result to its caller, the watcher keeps two pieces of bookkeeping per task, both populated at dispatch time (before the job is sent):

• a Future in a pending table keyed by task_id — the Future is the standard Python placeholder the caller blocks on, and task_id is a unique id minted per task so the returning envelope can find it;

• a slot taken from a semaphore — the backpressure budget; each outstanding task holds one slot, so the slot count caps how many tasks are in flight at once.

For throughput the watcher can be split into num_shards independent shards — parallel copies of {completion Channel + recv thread + pending table + semaphore}, each handling a disjoint slice of tasks chosen by hash(task_id) % num_shards. num_shards defaults to 1, so on a first read there is simply one lane.

Two diagrams cover the two halves of a task’s life. First, dispatch — when a request comes in, DragonAgentNode (the LangGraph node) calls submit_task on the watcher; the watcher mints a task_id, hashes it to pick a shard, reserves a slot from that shard (waiting if the shard is at its concurrency cap — see §4.4 for how the sync and async paths wait differently), records the Future in that shard’s pending table, and only then hands the job to the host — tagging the message with that shard’s completion Channel so the reply comes back to the right lane:

Fig. 40 Dispatch — a request comes in and is forwarded to a host.

Then, return — when the host’s result arrives, that same shard’s recv thread receives it, deserializes it, matches it to the Future by task_id, resolves it, and frees the slot:

Fig. 41 Return — a completed task’s result is resolved.

So the pending table and semaphore are populated on dispatch and drained on return, and the job itself never travels through the shard — only the small result envelope does. With that in mind, here is the full round trip; §4.4 covers sharding and backpressure in more depth.

Step by step:

1. Dispatch. DragonAgentNode calls submit_task (sync) or submit_task_async (async) on the watcher.

2. Register before dispatch. The watcher picks a shard by hash(task_id) % num_shards, acquires a backpressure slot from that shard’s semaphore, and registers a _PendingTask (carrying the Future) in the shard’s pending table — before the message leaves for the host. This is the core ordering invariant: a completion envelope can never arrive before the watcher knows which Future it resolves.

3. To the host. The task message {task_id, node_name, state, serialized_done_queue} is put on the host’s input Queue, which cloudpickles the whole message internally — so state rides as a raw object (no manual serialization). serialized_done_queue is the serialized descriptor of the shard’s shared completion Queue, which the host attaches to in order to reply.

4. Execute. The host runs the node as a concurrent task — await fn(state) on its event loop for an async node, or offloaded to its thread pool for a sync node. Co-located agents reuse the host’s shared LLM clients and weights.

5. Reply inline. The task puts a small envelope on the completion Queue, whose EnvelopePickler frames and serializes it. On success the return dict rides inline; on failure the exception does.

6. Route and resolve. The shard’s recv thread wakes from get, decodes the envelope, looks up the Future by task_id, resolves it (sync: set_result; async: loop.call_soon_threadsafe), and releases the slot.

7. Return. DragonAgentNode hands the result back and the LangGraph Pregel engine merges the returned dict into graph state (its apply_writes step) before starting the next superstep — Pregel’s next scheduling round.

4 — Component Deep-Dive

4.1 — DragonExecutor (coordinator)

DragonExecutor is the user-facing lifecycle manager and the owner of the shared result transport. It owns the single DragonWatcher that resolves all task futures. Its responsibilities:

• launch_host(agents, policy=...) — spawns one persistent dragon.native.process.Process running agent_host_entry, performs a handshake to receive the host’s input Queue handle, and registers one DragonAgentNode per agent. Call it multiple times (with different Policy objects) to place agents on different nodes.

• node(name) — returns a LangGraph RunnableCallable wrapping the agent’s __call__ (sync) and acall (async). This is what feeds StateGraph.add_node.

• shutdown(graceful=...) — sends a None sentinel to each host, joins/kills the processes, and stops the watcher. Also runs via the context-manager __exit__.

Backpressure sizing. max_concurrent_tasks caps how many tasks may be in flight at once. It is a single dial with a fixed default (64): raise it for more parallelism and pipelining, lower it to bound coordinator memory or to protect a rate-limited downstream. Everything else — per-shard semaphore slots, completion-channel capacity, and each host’s input-queue size — is derived from this one number, so you normally set only this. The completion Channels are sized to the cap when the watcher is built (in the constructor) and are not resized afterward.

The executor transports only agent results. It deliberately does not own any DDict — bulk data is the user’s responsibility.

4.2 — AgentHost (worker process)

agent_host_entry (_host.py) is the function executed inside each Dragon process. One host runs per cluster node (or hardware resource group) and multiplexes many agent functions on a single asyncio event loop (mirroring the Dragon agent framework’s SubAgent.listen). Because the agents share one address space, they share LLM clients, model weights, and DB connections — loaded once, reused by all.

Node functions may be sync or async, and each runs on the lane that suits it:

• A plain def node is offloaded to a bounded ThreadPoolExecutor (max_threads), so its blocking work is isolated on its own thread.

• An async def node runs as a task directly on the shared loop, so many I/O-bound async agents interleave at ~zero threads each. An existing LangGraph graph with async nodes (or async LLM/tool clients) runs unchanged, and a loop-bound resource (e.g. an async client) created once is reused across calls.

The blocking Dragon IPC — the input Queue.get and the completion-Channel sends — runs on a small separate pool so a burst of long sync nodes can never starve task intake or result delivery.

The host is created by launch_host() through a short handshake: the host allocates its own input Queue and ships the serialized handle back to the coordinator, which blocks until it arrives before registering any node.

Lifecycle:

1. Create a Dragon Queue and send its serialized handle back to the coordinator over the handshake reply_queue.

2. Run one asyncio loop: receive each task message and dispatch it as a concurrent task — await fn(state) for an async node, or offload fn(state) to the thread pool for a sync node.

3. Each task serializes a completion envelope and puts it inline on the task’s done-Queue (the shared completion Queue handed in per task).

4. Exit cleanly on the None sentinel, draining in-flight tasks first.

If a task names an unknown agent, the host still sends an error envelope so the caller never blocks forever.

Note

The async-loop contract (and its blast radius). Because async nodes share one loop, the standard asyncio rule applies: an async def node must not block the loop — a synchronous blocking call, or a long CPU-bound stretch, without await stalls every other agent co-located on that host until it returns. The blast radius is that one host process; other hosts are separate processes with their own loops. This is the normal responsibility of any async code — wrap blocking work in await asyncio.to_thread(...). You do not need to know about co-location to write a correct async node: following the standard rule is sufficient, and co-location only changes who is affected if the rule is broken. Plain def nodes are never subject to this — they run on their own thread.

4.3 — DragonAgentNode (the seam)

DragonAgentNode (_node.py) is the fn(state) -> dict callable LangGraph sees. It holds no per-task state; it just forwards to the watcher:

• __call__(state) — synchronous path used by graph.invoke. LangGraph runs it inside a BackgroundExecutor thread that parks on future.result().

• acall(state) — asynchronous path used by graph.ainvoke. It awaits an asyncio.Future, suspending the coroutine on the event loop instead of parking a thread. N concurrent agents in a superstep therefore cost zero extra threads — only the event loop plus the watcher thread(s).

4.4 — DragonWatcher (result router)

DragonWatcher (_watcher.py) is the heart of the control plane. Its dispatch and return paths — and the four parts of a shard (completion Channel, recv thread, pending table, slots semaphore) — are diagrammed in §3; this section covers the design rationale.

Why one shared Channel, not a per-task channel. A Dragon Channel is a true multi-producer / single-consumer durable queue: every message sent is reliably enqueued and later received, whether or not the consumer happens to be parked in recv at that instant. Carrying the result as a queued message (rather than an edge-triggered event, as a ChannelSet bcast_wait would) means a completion can never be lost to a missed wakeup — so no task is ever left hanging. The watcher is still fully event-driven: recv_bytes(timeout=None) parks the thread in the kernel (zero CPU when idle) and wakes the instant a message arrives. No polling.

Sharding (``num_shards``). The per-message route step (recv_bytes → cloudpickle.loads → resolve Future) is serial on one thread. Setting num_shards > 1 creates that many independent lanes (_Shard — its own Channel, recv thread, semaphore, and pending table); tasks are partitioned by hash(task_id) % num_shards, so each shard owns a disjoint slice and no lock is ever contended across shards. Raise it only when a single watcher thread becomes the bottleneck (very high completion rate of small results).

Backpressure. max_concurrent_tasks caps simultaneous in-flight tasks via a BoundedSemaphore (split evenly across shards). It has a fixed default (64); pass a value to raise or lower it (see §4.1). When the cap is hit, a further dispatch waits for a task to complete — the sync path (submit_task) blocks its calling BackgroundExecutor thread, while the async path (submit_task_async) awaits the slot without blocking the event loop (the blocking wait is offloaded to a worker thread), so graph.ainvoke stays responsive under backpressure. Each shard’s Channel is sized to per_shard_ring + headroom, so the channel can never overflow even if every in-flight task completes at once. Total tasks over the executor’s lifetime are unlimited — only peak concurrency matters.

Sizing cheat-sheet. You tune exactly one number — max_concurrent_tasks — and everything else is derived from it, so no number is ever a silent second bottleneck:

Number	What it is / how it’s set
max_concurrent_tasks (the knob)	Max tasks in flight (dispatched but not yet completed) at once. Fixed default 64; raise for more parallelism, lower to use less memory. Everything below follows from it.
Per-shard slots	max_concurrent_tasks / num_shards — the in-flight cap split evenly across each shard’s backpressure semaphore.
Completion Channel capacity	per-shard slots + headroom — big enough to hold a full burst of completions; it can never overflow because at most that many tasks are in flight.
Host input Queue size	max_concurrent_tasks + headroom (set by the executor, per host) — so the coordinator can always enqueue a dispatched task without put() blocking on queue capacity (worst case: every in-flight task routes to one host). Backpressure stays solely on the semaphore.
num_shards	Number of parallel receive/decode lanes. Raise only if one watcher thread can’t keep up with the completion rate (default 1).
max_threads (per host)	Threads for sync (def) node bodies in each host’s pool (default max(len(agents) * 4, 8)). Async nodes run on the host’s shared event loop and are not bounded by this.

In one line: set ``max_concurrent_tasks``; the semaphore, the completion channels, and the host input queues all size themselves from it.

Choosing ``max_concurrent_tasks`` and ``max_threads`` together. Both have sensible defaults and both are yours to override — but they bound different things:

• max_concurrent_tasks (coordinator) bounds tasks in flight across the whole system. max_threads (per host) bounds how many sync (def) node bodies run at once inside one host (they run on a thread pool). Async (async def) nodes do not use that pool — they run on the host’s shared event loop, bounded only by max_concurrent_tasks.

• Async (I/O-bound) agents: the cap is the real limiter — raise it freely for more concurrency; max_threads does not gate them.

• Sync agents: admitting far more than a host’s max_threads (cap ≫ per-host sync threads) just queues them — more coordinator/host memory and latency, no extra throughput (a host runs at most max_threads sync bodies at a time). Sizing the cap near the sync-thread capacity keeps workers busy without a backlog.

Neither knob changes the other: the cap never spawns host threads, and raising max_threads never widens the cap. Set them deliberately — you own the behavior of the combination you pick.

Return-message collection (fan-in). This is where results from every host in the allocation converge. Regardless of how many AgentHost processes are running or how wide a superstep fans out, every completion is funneled back through the same small set of shard channels and resolved by the shard’s single recv thread — multiple hosts (the producers) send envelopes into the sharded completion channels, and each shard’s recv thread (the consumer) drains its channel and matches each envelope to a waiting Future by task_id:

Reading the diagram: the number of consumer threads is fixed at num_shards — not the number of hosts or the width of a superstep — so the coordinator’s footprint stays constant no matter how wide the allocation fans out. In effect the shards are a fixed-width fan-in: the entire allocation’s results are collected through num_shards lanes and demultiplexed to the correct Futures purely by table lookup.

4.5 — Completion envelope (_constants.py)

The host and watcher agree on a tiny cloudpickle’d dict per completion:

{
  "task_id": str,                       # routes to the pending Future
  "status":  "done" | "error",
  "payload": <result_dict | exception>,
}

The field-name constants live in _constants.py so the two sides can never drift. On success the agent’s return value rides inline; on failure the exception is carried inline (falling back to a RuntimeError with the text if the exception object cannot be pickled).

5 — Data-Plane Scaling: Central Store vs Node-Local DDict

The three planes were defined in §1. The data plane is the one that decides whether a data-heavy campaign scales: bulk artifacts live in a user-managed DDict and only a lightweight handle/key crosses LangGraph state, so the watcher stays a pure signal router and the coordinator never materializes gigabytes of intermediate data. It is worth seeing why that matters, by contrast with the usual central-store approach.

Before: the central-store funnel

Without Dragon, a LangGraph orchestrator on one node glues itself to heavy compute elsewhere through the filesystem or a central store (Redis, shared FS). Even with the “handle-in-state” pattern (store the payload under a key, pass the key), the bulk still lives in one server: every producer writes it over the network and every consumer reads it back over the network — the consumer can never be moved to the data. That server’s RAM and NIC are the ceiling.

After: node-local DDict + data-aware placement

With Dragon the agent layer runs inside the allocation. A producer writes its large result to the node-local DDict manager (zero-copy, no network) and returns only a key through LangGraph state. The consumer is placed on the node where the data lives, so its read is local; Dragon uses the HPC interconnect (RDMA/HSTA) only when a read must cross nodes. The orchestrator only ever sees handles.

Note

Locality and placement are properties of the Dragon runtime (ProcessGroup + a node Policy + DDict), available with or without LangGraph. This integration’s job is to let an existing LangGraph graph reach that substrate without a rewrite — it does not, by itself, own locality. The data-plane win is real only for data-heavy results; for small JSON/text payloads a central store is perfectly adequate.

6 — Sync vs Async Concurrency

What is the same. invoke and ainvoke share the entire pipeline — the same dispatch, the same watcher, the same backpressure semaphore, and, most importantly, the same host-side execution. Whichever you call, a node runs on the host the same way: a sync (def) node on the host’s thread pool, an async (async def) node on the host’s event loop. The driver does not change where or how your node runs, and both produce the identical result. Your node’s sync/async-ness is therefore independent of the invoke/ainvoke choice.

What differs — only how the coordinator waits. The single difference is on the caller side: how the coordinator waits for each node’s Future.

Path	graph.invoke (sync)	graph.ainvoke (async)
Node entry	DragonAgentNode.__call__	DragonAgentNode.acall
Future type	concurrent.futures.Future	asyncio.Future bound to the running loop
Caller waits by	parking a LangGraph BackgroundExecutor thread on result()	suspending the coroutine on the event loop
Coordinator threads for N concurrent nodes	up to N (one parked thread per in-flight node)	0 extra (only the loop + watcher thread)

The watcher resolves async futures via loop.call_soon_threadsafe from its recv thread, which is why ainvoke can fan out to many concurrent agents without spawning a thread per node.

The same split applies to backpressure. When in-flight tasks reach max_concurrent_tasks, the wait for a free slot follows the caller’s concurrency model: invoke blocks its BackgroundExecutor thread, while ainvoke awaits the slot — the blocking wait is offloaded to a worker thread so the event loop keeps turning and resolving completions (which is exactly what frees slots), instead of freezing. In both cases the single BoundedSemaphore per shard remains the one budget, so the completion-channel-can’t-overflow guarantee is unchanged.

When to use which — this matters as you scale concurrent agents. Because the host runs your nodes identically either way, the choice is purely about the coordinator’s cost of waiting, which scales with how many nodes are in flight at once (a superstep’s fan-out width). Know what happens under the hood so you pick responsibly:

• ainvoke is a functional superset — anything invoke does it can do (wrap it: asyncio.run(graph.ainvoke(...))). Its only requirement is that you call it from an event loop. It adds no capability over invoke; it only changes how the coordinator waits.

• Chain or modest fan-out (one, or a handful of, nodes in flight at a time): invoke parks that many BackgroundExecutor threads on future.result() — idle threads (memory + a scheduler slot, no CPU), so the overhead is negligible. And invoke is simpler: a plain blocking call, no event loop, linear stack traces. Prefer ``invoke`` here when your driver is synchronous.

• Wide fan-out (tens/hundreds/thousands of nodes concurrent in one superstep): invoke needs one coordinator thread per in-flight node and can saturate LangGraph’s sync executor pool (serializing work beyond the pool size), while ainvoke awaits every node on one event loop at ~zero threads each. Prefer ``ainvoke`` when you scale concurrency — it is the difference between N parked threads (and a possible pool bottleneck) and none.

Rule of thumb: pick the driver by your caller’s shape and fan-out width, not by your node bodies. Synchronous driver + narrow graph → invoke; async application or wide fan-out → ainvoke. Either way, max_concurrent_tasks still caps total in-flight work regardless of the driver (see §4.4).

7 — Dragon Primitives Used

The integration is built on a small set of Dragon primitives. Understanding which primitives are used and where is key to understanding the architecture.

Primitive	Module	Role in LangGraph Integration
Queue	dragon.native.queue	Host input queues and the startup handshake reply queue. One Queue per AgentHost process, shared by all agents on that host. The primary dispatch mechanism (coordinator → host).
Channel	dragon.channels	Shared completion channel (one per shard), created through Global Services and attached from its serialized descriptor — which is what makes it reachable from hosts on remote nodes. All hosts enqueue result envelopes; the watcher’s recv thread drains it. Durable MPSC queue — messages are never lost.
dragon.globalservices.channel	dragon.globalservices	create / release_refcnt / destroy for the completion channel(s). GS registration is what lets the transport agent route a remote host’s send into the channel.
MemoryPool	dragon.managed_memory	Only attach_default() — the watcher creates no pool of its own; the completion channel(s) live in the runtime’s default pool. The pool is shared infrastructure, so it is detached (never destroyed) at teardown.
ChannelSendH / ChannelRecvH	dragon.channels	Send/receive handles for the completion channel. One send handle per agent thread (opened per task), one recv handle per watcher shard (persistent).
Process	dragon.native.process	AgentHost process lifecycle. One Process per launch_host() call.
Policy	dragon.infrastructure.policy	Optional placement control — pin a host to a specific cluster node, NUMA zone, or GPU set. Passed via launch_host(policy=...).

Dragon import concentration. Only 3 of 6 source files contain dragon.* imports:

File	Dragon primitives
executor.py	Process (spawn), Queue (handshake reply)
_host.py	Queue (create input queue), ChannelSendH (send envelope)
_watcher.py	globalservices.channel (create/release/destroy the completion channel), MemoryPool.attach_default() (pool for that channel), Channel.attach (attach the descriptor), ChannelRecvH (recv loop), ChannelSendH (stop sentinel)

The remaining files (__init__.py, _node.py, _constants.py) are pure Python with no Dragon imports.

8 — Error Handling

The integration handles errors at every boundary:

Scenario	Handling
Agent function raises	Exception is packed into an error envelope (status="error", payload=exception), sent inline on the completion Channel. The watcher resolves the future with set_exception(). LangGraph sees a normal Python exception.
Unpicklable exception	The Queue’s EnvelopePickler substitutes a RuntimeError carrying the original text, so dump never raises and the envelope always reaches the wire under its own task_id.
Unknown node_name	Host logs the error and sends a ValueError error envelope so the caller never blocks forever.
Task timeout	Sync: future.result(timeout=...) raises TimeoutError. Async: asyncio.wait_for raises TimeoutError. The backpressure slot is released. A late-arriving completion for an unknown task_id is silently discarded by the watcher.
Host process crashes	No envelope arrives. If event_timeout is set, the caller’s future.result(timeout=...) or asyncio.wait_for raises TimeoutError. The backpressure slot is released.
Dispatch failure	If input_queue.put() raises, the watcher unwinds the registration: pops the pending task, releases the slot, and fails the Future immediately.
Coordinator crashes	Use a persistent LangGraph checkpointer (SqliteSaver, PostgresSaver) so a new coordinator can resume from the last checkpoint.

All error paths guarantee that exactly one of these happens for every dispatched task:

1. The Future is resolved with a result, or

2. The Future is resolved with an exception, or

3. The caller’s timeout fires.

No task is ever left hanging indefinitely (provided event_timeout is set or the host process remains alive).

9 — Lifecycle & Memory Management

The lifecycle of every resource managed by this integration:

DragonExecutor()
  ├─ Attaches to the default MemoryPool (once, not per shard)
  ├─ Creates ONE shared completion Channel per shard
  │    via Global Services, in that default pool
  └─ Starts ONE watcher recv thread (per shard)

executor.launch_host(agents={...}, policy=...)
  ├─ Creates Dragon Process (persistent)
  ├─ Process creates input Queue, sends handle back (handshake)
  └─ Executor creates DragonAgentNode per agent name

[... graph runs, tasks dispatched ...]

Per task (HOT PATH — nothing created/destroyed):
  ├─ Acquire backpressure slot (semaphore)
  ├─ Register Future in pending table
  ├─ Queue.put(msg) → host
  ├─ fn(state) → result (async: on the host loop; sync: on the thread pool)
  ├─ ChannelSendH.send_bytes(envelope) → watcher
  ├─ recv_bytes() → decode → resolve Future
  └─ Release backpressure slot

DragonExecutor.shutdown() / __exit__()
  ├─ Send None sentinel to each host's input Queue
  ├─ Host stops accepting work, drains in-flight tasks, closes its pools
  ├─ proc.join(timeout=10), then proc.kill()
  └─ watcher.stop()
       ├─ Send stop sentinel envelope to each shard's channel
       ├─ Join each shard's recv thread
       ├─ Close recv handles
       ├─ Per shard: release_refcnt(cuid) → destroy(cuid) (Global Services)
       └─ Detach the default MemoryPool (never destroyed — shared
          runtime infrastructure)

Memory guarantee: This layer never leaks resources. The shared completion Channels are created once at startup through Global Services and released and destroyed the same way on executor exit; the default MemoryPool they live in is only attached and detached, never destroyed, because it is shared runtime infrastructure. No Dragon objects are created or destroyed on the hot path (per task).

User-managed DDict: Any DDict you create is yours to manage. The executor neither creates nor destroys it. Free it (del keys / ddict.destroy()) when your experiment finishes.

10 — Multi-Node Deployment

Each launch_host() call spawns one Dragon Process. Use a Policy to place each host on a specific cluster node:

Fig. 42 Multi-node deployment — where each component runs.

All communication uses the Dragon primitives catalogued in §7: Queue for task dispatch and the startup handshake, one shared Channel per shard for completions, Process per launch_host() call, Policy for placement, plus the user-owned DDict and any ProcessGroup an agent launches.

11 — Usage Patterns

Pattern A — Single host, all agents together (simplest)

with DragonExecutor() as executor:
    executor.launch_host(agents={"a": fn_a, "b": fn_b, "c": fn_c})

All agents share one process, one thread pool, one set of resources.

Pattern B — Multi-node, explicit placement (batch style)

with DragonExecutor() as executor:
    executor.launch_hosts(
        hosts=[
            {"researcher": fn},
            {"analyzer": fn, "writer": fn},
        ],
        policies=[policy_node0, policy_node1],
    )

One host per node. Each entry in policies is applied to the corresponding host. Agents on the same node share resources; agents on different nodes are isolated.

Pattern C — GPU isolation (one host per GPU)

with DragonExecutor() as executor:
    executor.launch_hosts(
        hosts=[
            {"gpu_agent_0": fn},
            {"gpu_agent_1": fn},
        ],
        policies=[gpu0_policy, gpu1_policy],
    )

Pattern D — With Dragon Inference (on-cluster LLM)

Call get_llm() inside an async node, passing the inference pipeline’s queue handle. The proxy is built lazily in the host process and cached per pipeline, so it binds to the host’s event loop and is reused across calls. Do not build the proxy on the coordinator and pass it in — it is loop-bound and per-process.

from dragon.ai.langgraph import DragonExecutor, get_llm

input_queue = Queue()                       # the inference pipeline's ingress
Process(target=start_inference, args=(input_queue,),
        policy=gpu_policy).start()

async def researcher(state):                # async node → runs on the host loop
    resp = await get_llm(input_queue).chat([...])   # built once, in the host
    return {"research": resp}

with DragonExecutor() as executor:
    executor.launch_host(agents={"researcher": researcher})
    graph.ainvoke(...)                      # async driver

Two models on one host is just two queues — each get_llm(q) returns that pipeline’s own cached proxy:

async def a(state): return {"x": await get_llm(llama_q).chat([...])}
async def b(state): return {"y": await get_llm(mistral_q).chat([...])}

(If your inference service exposes an OpenAI-compatible HTTP endpoint, you can instead use a normal ChatOpenAI(base_url=...) created once at module scope — lowest friction, at the cost of HTTP transport instead of Dragon Channels.)

Pattern E — Agent dispatches HPC compute; bulk stays in user’s DDict

from dragon.data.ddict import DDict
from dragon.ai.langgraph import get_llm

sim_data = DDict(managers_per_node=1, total_mem=10_000_000_000)

async def simulation_agent(state):
    plan = await get_llm(input_queue).chat([...])
    pg = ProcessGroup(nproc=256, policy=compute_policy)
    pg.start(target=run_sim, args=(plan, sim_data))
    pg.join()
    return {"result_key": "sim_output_42"}  # handle only

with DragonExecutor() as executor:
    executor.launch_host(agents={"simulate": simulation_agent})