5Graph Execution Engine

Nodes fire when inputs arrive, not in a fixed order. The GraphExecutor supports cycles natively, enables real-time streaming, and opens the door for future parallelism.

Design docs: Graph Executor · Loop Control · State Containers

1. Overview

The GraphExecutor is the primary execution engine. It replaces a linear DAG executor with a dataflow model where data drives execution order. This naturally supports:

• DAG workflows — single forward pass
• Cyclic agent loops — iterOut feeds back to iterIn; run-start values wire into iterIn's init side (ADR-dataflow-008)
• Mixed patterns — both in the same graph

Ready queue: [entry₁, entry₂, ...]
         │
         ▼
    Pop node → fire → propagate outputs → enqueue ready targets
         │                                       │
         └───────────────────────────────────────┘
              (until queue empty or terminated)

2. Core Concepts

3. Execution Algorithm

3.1 Initialization

# 1. Merge hooks from 3 sources: global → graph → node
merged = merge_hooks(global_hooks, graph.hooks, extract_node_hooks(graph.nodes))

# 2. Flatten composites (expand subgraphs into flat graph)
graph, flatten_map = flatten_graph(graph)

# 3. Build state containers + access-grant index
containers = build_containers(graph.containers)
for ag in graph.access_grants:
    access_grant_index[ag.node_id].add(ag.container_id)

# 4. Build node instances + adjacency map
for node_def in graph.nodes:
    nodes[node_def.id] = NodeInstance(id=..., type=..., config=...)
for edge in graph.edges:
    adjacency[edge.source].append(edge)

# 5. Entry discovery — three structural conditions:
#      type != "iterIn"  AND  no incoming edges  AND  no required input ports
#    validate_graph_connectivity() (graph_def.py) rejects required-but-unwired
#    ports at load time, so this rule never silently drops a well-formed entry node.
for node in nodes.values():
    if node.type == "iterIn": continue
    if node.id in incoming_targets: continue
    if _get_required_ports_for_node(node): continue
    ready_queue.append(node.id)

# 6. Fire run_start signal (lifetime="run" states reset) + GraphStart hook
broadcast_signal("run_start", {...})
await hook_runner.run_hooks("GraphStart", ...)

Required ports are resolved per-instance: _get_required_ports_for_node merges class-level get_required_inputs(type) with anything _resolve_ports(config) marks optional=False, so configurable-port nodes (ADR-dataflow-003) like iterIn/iterOut/llmCall participate in both entry discovery and readiness.

3.2 Dataflow Loop

while ready_queue and total_firings < max_firings:
    await pause.wait()
    if stop.is_set(): break

    node_id = ready_queue.pop(0)
    node = nodes[node_id]

    # iterIn: emit step_start before firing so lifetime="step" states see a
    # fresh slate for the upcoming iteration
    if node.type == "iterIn":
        broadcast_signal("step_start", {"step": step_counter + 1})

    result = await _fire_node(node, session)   # runs Pre/PostNodeExecute hooks
    total_firings += 1

    # iterOut: per-scope step counter + checkpoint + nav_step + transfer to paired iterIn
    # (snippet simplified to single-scope; see §10 — actual code resolves _io_scope_id from
    #  node.config["pairedWith"], advances _io_scope.step_counter, and only mirrors to
    #  self.step_counter / checkpoints on the outermost scope)
    if node.type == "iterOut":
        step_counter += 1
        broadcast_signal("step_end", {"step": step_counter})
        if containers:
            _checkpoints[step_counter] = {cid: c.checkpoint() for cid, c in containers.items()}
        await _broadcast_step(session)
        if step_counter >= step_budget: break
        # ADR-dataflow-006 auto-prefix: iterOut output key X declared in
        # iterIn.config.loop_ports routes to paired.pending_inputs["iterout_X"]
        for key, val in result.items():
            if key in paired_loop_ports:
                _route_value_to_port(paired, f"iterout_{key}", val)
        ready_queue.append(paired.id)

    node.pending_inputs = {}  # consumed

    # Propagate outputs to downstream — LIST[T] coercion applied at this seam
    for edge in adjacency[node_id]:
        target = nodes[edge.target]
        _route_value_to_port(target, edge.targetHandle, result[edge.sourceHandle])
        if _is_ready(target) and target.id not in ready_queue:
            ready_queue.append(target.id)

3.3 Readiness Check

A node is ready when:

• Has required ports: all required ports (class-level ∪ instance-resolved) have data in pending_inputs
• All ports optional: fires when ANY data arrives — or, for iterIn, when init_port_cache has any values even if no loop inputs have arrived yet

def _is_ready(node):
    required = _get_required_ports_for_node(node)   # class + _resolve_ports(config)
    if required:
        return all(p in node.pending_inputs for p in required)
    if node.type == "iterIn" and node.init_port_cache:
        return True
    return len(node.pending_inputs) > 0

3.4 Hooks Around Each Firing

_fire_node wraps instance.forward(fire_inputs, ctx) with two user-extensible hooks:

Graph-level hooks (GraphStart / GraphComplete / GraphError) fire at run entry and both exit paths. All hooks are fail-open — hook errors never break execution.

4. Two-Pivot Iteration (ADR-dataflow-008)

Cyclic agent loops use two pivot node types — the loop-carry transfer goes through the executor, not through a canvas wire:

The pairedWith config field on iterOut names the paired iterIn id. iterIn's unified port surface is synthesised at load (_synthesize_iterin_ports) from two writer namespaces:

• initPorts (authored on iterIn) — run-start values written by ordinary canvas edges into init_<name> handles; persist=true slots re-emit every iteration, persist=false slots are one-shot (iter 0 only)
• paired iterOut.ports — written each iteration by the executor transfer into iterout_<name> slots

Handle ids on iterIn are always-prefixed so the two writers never collide: init_X for init values, iterout_X for loop-carry values. All slots live in NodeInstance.port_slots; at fire time the populated slots become the fire inputs.

An entry node is queued iff type != "iterIn" AND no incoming edges AND no required input ports. Required-but-unwired ports are rejected at the API boundary by validate_graph_connectivity() (see graph_def.py), so the executor never silently fails to fire an entry node.

5. List-Typed Ports (ADR-dataflow-005)

Any wire type T can be wrapped as LIST[T] on a consumer port. Producers stay scalar; the executor applies coercion at a single port-binding seam (_route_value_to_port):

• Consumer port T (non-list): write-through (overwrite).
• Consumer port LIST[T]:
  • Incoming scalar → append to the existing list (starting a length-1 list if empty).
  • Incoming list → extend.
  • Fan-in order follows edge declaration order (adjacency iteration is stable).

This unblocks multi-image LLM calls, multi-LLM debate (LIST[TEXT]), search-operator population fan-in, and the two-pivot handoff without forcing list shape on single-item producers.

6. State Management

Each node has two kinds of state:

6.1 Node State (_NodeStateProxy)

Persistent across firings. Handlers read/write attributes via ctx:

async def forward(self, inputs, ctx):
    history = ctx.history or []
    ctx.history = history + [inputs["observation"]]
    ctx.step            # current step counter
    ctx._executor       # back-ref — proxy nodes use this for per-runner URL overrides (ADR-eval-002)

_NodeStateProxy redirects attribute access to the node's state dict, which persists across iterOut→iterIn cycles.

6.2 Config (self.config)

Per-node config from graph JSON — set once at graph-build time, immutable during execution.

7. State Containers & Signals

State containers (ADR-dataflow-002, ADR-dataflow-004) provide visible shared state across nodes. Nodes obtain access via access grants (formerly called "state edges") in graph.access_grants — grants carry no data and do not trigger firing.

7.1 Access

Containers are injected into the ctx proxy only when the node holds an access grant:

async def forward(self, inputs, ctx):
    container = ctx.containers["exploration_state"]   # granted container
    container.write("visited_rooms", room_id)

    # graph_state is a regular ContainerDef — no auto-inject after ADR-dataflow-004;
    # the node must also hold an access grant to it.
    if ctx.graph_state is not None:
        ctx.graph_state.write("step_count", ctx.step)

7.2 Reducers

Each state entry has a reducer that defines how writes accumulate:

Per-iteration clearing is declared separately on the state's lifetime field (forever / step / episode / run / custom); the removed ephemeral reducer is now expressed as lastWrite + lifetime="step".

7.3 Signal Bus

broadcast_signal(name, payload) fans a signal out to every live container; each BaseState with that name in its reset_on list clears to initial_value. Canonical signals:

7.4 Checkpoints

After the step_end signal fans out, the executor snapshots every container into _checkpoints[step_counter] (sliding window, max 100). restore_step(n) rewinds the containers, resets the step counter to n, and prunes future checkpoints. REST: GET /run/checkpoints, POST /run/restore/{step}.

8. Worker-Pool Overrides (ADR-eval-002)

Each GraphExecutor carries two optional per-runner maps so worker-pool batch eval (BatchEvalRunner + EnvWorkerPool) can route unqualified lookups to its own tagged env subprocess instead of the global singleton:

Both maps are empty on canvas Play and single-worker eval — bit-identical behaviour.

9. Termination

Four ways execution ends (there is no termination node — removed 2026-06-11; the halt signal is the stop BOOL input on the loop's iterOut, checked once per iteration at the boundary):

1. iterOut.stop — a truthy stop value ends the scope at the iteration boundary; the root scope's stop ends the run. Termination triggers the final-side emission: the terminal iteration's values go out once on the iterOut's final_* handles, feeding the after-loop verdict stage (evaluate → graphOut).
2. Step budget — step_counter >= step_budget (default 1000, DEFAULT_STEP_BUDGET in config.py; the executor's step_budget=500 attribute is only a pre-run placeholder that run() overwrites with the resolved value); also emits the final side.
3. Max firings — safety limit step_budget × node_count × 3 (prevents infinite non-iterating loops); final side emitted best-effort from the last completed iteration.
4. Stop event — the frontend Stop button / API sets the runner's stop_event.

On every exit path — clean or error — run_end fires, GraphComplete or GraphError hooks run, and the execution logger flushes.

10. Multi-Scope Execution (ADR-dataflow-007 + ADR-executor-003)

The two-pivot model in §4 describes one (iterIn, iterOut) pair. A single flat graph can contain N coexisting pairs — each defines a scope with its own iteration cadence. The simplified single-scope snippet in §3.2 expands to per-scope state inside the executor: scope_state[scope_id] carries a _ScopeState with its own step_counter, termination flag, and pending-input bookkeeping.

10.1 Scope forest

analyze_scopes() (agent_loop/scope_analysis.py) runs before the main loop. It walks every iterIn and uses the pairedWith graph (iterOut→iterIn) plus BFS over the body region to determine which other pivots are interior to each scope. The result is a parent / child tree keyed by the iterIn id; the root is the outermost scope whose stop signal (or budget exhaust) ends the whole run.

10.2 Per-scope step counter + backward compat

When iterOut fires, the executor resolves _io_scope_id = node.config["pairedWith"] or self._outermost_scope_id and advances _io_scope.step_counter. For backward compatibility with single-scope graphs, the outermost scope's counter is also mirrored to self.step_counter — code paths that read the engine-wide counter (checkpoints, nav_step broadcast, step-budget check) see unchanged behaviour. Inner scopes get their own counter only.

10.3 Checkpoint + nav_step only on root-scope iter

Container snapshots into _checkpoints[step] and the consolidated nav_step WebSocket broadcast both fire only when _io_scope is the outermost scope. Inner scopes don't checkpoint and don't emit per-iteration nav frames — that keeps a single-outer-scope graph bit-identical to the pre-multi-scope behaviour, and avoids checkpoint storms when a fast inner loop iterates many times per outer step.

10.4 Scope re-entry + graphOut latch

• Scope re-entry: when an outer scope's iterOut re-queues its paired iterIn and that brings an inner scope's iter_in back onto the ready queue, the executor resets the inner _ScopeState (terminated = False, step_counter = 0). This makes inner scopes function-call-shaped: each outer iteration re-runs the inner loop from scratch.
• graphOut latch: when a graphOut node lives inside a non-root scope, it buffers its incoming value into state["latched_value"] (flushed by _propagate_graphout_latches) instead of propagating immediately. On the owning scope's termination, the buffered value is flushed to outer-scope downstream — preserves the "scope returns its last value" semantics across iteration boundaries.

10.5 Stop is per-scope

In a multi-scope graph, each scope's halt signal lives on its own iterOut's stop input — structurally bound to the right loop. An inner scope's stop ends only that scope (its final side emits, the outer loop continues); only the root scope's stop ends the run.

11. Key Files

Status

All items fully implemented.