Graph Executor

This page is in two parts. Part I teaches the concepts with diagrams. Part II prints the engine as one continuous annotated listing — colored region bands segment run() so you can see at a glance which block does what — keyed to agentcanvas/backend/app/agent_loop/graph_executor.py.

1. The drawn graph and its contract

A canvas graph is nodes joined by typed wires. Its execution semantics rest on exactly two things: a firing rule that decides when any node runs (§1.1), and one recognised structure — the loop pivot pair — that expresses iteration, halting, and the episode verdict (§1.2). Everything else on the canvas is ordinary dataflow. This section states the contract; §2–§5 describe the algorithm that honours it.

1.1 The firing rule

Three consequences follow, and they shape everything below:

• Order is derived, not authored. There is no step list; whichever node is ready runs. A branch that never receives data never runs, so conditional paths need no special construct.
• Inputs are consumed. A fire empties the ports it read; the node fires again only when fresh values arrive. One delivery is one firing opportunity, not a standing subscription.
• No drawn cycles. The graph is acyclic as drawn — a wired cycle would give the data no valid order. Iteration is expressed instead by the pivot structure of §1.2 (ADR-executor-001).

1.2 Anatomy of an agent loop

Plain dataflow ends when data stops flowing — enough for a pipeline, not for an agent that must observe, act, and observe again. Iteration is expressed by the one structure the engine recognises: a paired IterIn / IterOut, the loop pivots. The smallest agent loop:

Reading the figure in run order:

1. Run start. The entry node fires once — it needs no inputs — and its init_* edges hand the run-invariants (instruction, episode config) to the IterIn's init side.
2. Each iteration. The IterIn emits the loop-carry bundle; the body — here predict and env_step, ordinary nodes with no special treatment — computes; the IterOut collects the next carry. The engine hands the carry back to the IterIn by pairing: the top edge is not a wire you draw, and the drawn graph stays acyclic.
3. Halting. Any BOOL wired into the IterOut's stop input — here the env's done — ends the loop after the current iteration. Left unwired, the loop runs to its step budget.
4. Termination. The IterOut's final_* handles mirror its carry ports (plus a constant final_stop). They are real wires, but deliver exactly once — the terminal iteration's values.
5. Verdict. Everything downstream of final_* — here evaluate and graphOut — is the after-loop band: it runs once, after the loop, and may not take inputs from the loop body.

The drawing therefore has three places, and a run three times — the map that View 2 walks in order:

2. Where the engine sits

One class owns everything this page describes — GraphExecutor, entered through a single method, run(graph, session, …). The layers above it (HTTP endpoint, LoopRunner) own lifecycle — pause, stop, resume — and never look inside the graph; the layer below (the node handlers) computes and never schedules. The executor is the only place that decides what fires when.

A fresh executor is built per run, so no state leaks between episodes. For parallel batch eval it exposes two per-worker overrides — get_env_panel(name) and get_server_url(nodeset) — so each eval worker talks to its own env subprocess (ADR-eval-002); outside batch eval they pass through.

Inside run(), every run has the same three-phase shape — the outline of the rest of View 2, and the engine-side mirror of §1.2's three places: pre-loop builds live state and primes the queue (§3) · in-loop fires until a halt (§4) · after-loop delivers the verdict and finalises (§5).

3. Pre-loop — build & discover entries

Before a single node fires, run() does a deterministic, once-per-run setup in two movements. BUILD turns the static graph JSON into live, mutable runtime state; ENTRY DISCOVERY finds the entry nodes — the nodes that can fire with no inputs — and primes the queue with them. Nothing here loops — it is pure preparation, identical on every run.

3.1 BUILD — compile the graph into live state

Seven steps, strictly in order — plain top-to-bottom code, no concurrency. Each step turns one part of the authored GraphDefinition into something the loop can read or mutate. (Run-start args need no build step of their own: they arrive over ordinary canvas edges into the iterIn's init_* handles — ADR-dataflow-008.)

3.2 ENTRY DISCOVERY — find the entry nodes, prime the ready-queue

An entry node is discovered automatically — no author marking — as any node satisfying all three structural conditions; every entry node is queued at run-start:

• not an iterIn — IterIn is the loop's entry; it fires only when its init_* edges deliver (iter 0) or via the IterOut transfer (iter 1+), never via entry discovery;
• no incoming edges — nothing upstream will ever fill it;
• no required input ports — it can fire on nothing.

validate_graph_connectivity (in graph_def.py) rejects required-but-unwired ports at load time, so the third condition only ever excludes well-formed leaf consumers — it never silently drops a real entry node. With the queue primed, three things arm the loop:

• run_start signal — state containers with lifetime="run" reset to a clean baseline;
• the GraphStart hook fires;
• max_firings = step_budget × nodes × 3 is computed — a runaway-loop backstop set far above any real run.

4. In-loop — the firing loop

With the queue primed, the engine starts turning. This chapter walks one turn from the inside out — the loop skeleton (§4.1), the guards before each fire (§4.2), the IterOut boundary that closes every iteration (§4.3), the propagation path every other node takes (§4.4) — then reassembles the pieces into the whole-run flowchart (§4.5) and a concrete trace (§4.6).

4.1 One iteration & the loop skeleton

Keep §1.2's graph in mind — entry node, two pivots, a two-node body; this chapter is what the engine does with it. The main loop is deliberately tiny. Each turn:

1. Gate. Run while the queue holds a node and total firings stay under the runaway cap (while ready_queue and total_firings < max_firings); honour pause; a user Stop breaks out.
2. Pop. Take the front of the FIFO ready-queue.
3. Guard, then fire. Two structural guards (§4.2), then _fire_node — the per-node fire path (graph_executor.py).
4. Fork. An IterOut runs the once-per-iteration boundary (§4.3); every other node takes the ordinary propagation path (§4.4).

Nothing else is special: an agent's body is plain dataflow firing, and IterOut is the only node type the engine schedules differently.

4.2 The guards before a fire

A popped node is not fired blindly — two cheap structural guards run first, in order; each either skips the node or readies state for the fire:

1. Scope barrier. A node belonging to an inner scope that has already terminated is frozen (pending_inputs cleared, fire skipped) until that scope re-enters — otherwise in-flight inner-body nodes fire one extra time after termination and leak per-iteration values through cross-scope wires. Exempt: iterIn / graphIn, the cascade that reopens the scope.
2. IterIn step_start. An iterIn broadcasts step_start (per scope) so lifetime="step" state resets to a clean slate for the iteration about to run; a re-fired iterIn whose inner scope was terminated resets that scope's terminated flag + counter — the multi-scope re-entry path.

Then the node fires. A StopExecution raised by the fire halts the run immediately; any other exception is published to the ErrorBus and the node's result becomes {"error": …} so downstream nodes can still react (in AGENTCANVAS_STRICT_ERRORS mode it re-raises and fails fast instead).

(Until 2026-06-11 there were four guards. The termination-defer and post-loop-skip guards were deleted with the termination node and config.post_loop — the stop signal and the after-loop stage now live on the iterOut pivot itself, so there is nothing left to defer or skip.)

4.3 The IterOut boundary — four phases

IterOut fires exactly once per iteration, so the engine hangs every per-iteration chore on it. The order of the four phases is load-bearing — in particular Settle runs before Decide:

What each phase does, and the subtle rule inside it:

• ① Record. Advance the per-scope step counter (the root scope's counter also mirrors into self.step_counter + session._current_step); broadcast step_end; checkpoint all state containers (root scope only); flush the execution log; broadcast the consolidated nav_step (root only). IterOut's own outgoing edges are not routed per iteration — they are final-side handles, emitted once at termination (③).
• ② Settle. Drain this iteration's leftover sinks (bounded at 64, restricted to the iterOut's own scope, skipping only iterIn). This is where in-loop viewer / telemetry sinks scheduled in this iteration's wave still emit — including on the terminal step. One exception to the scope filter (2026-07-04): a root boundary also drains graph-scope nodes — dead-end sinks are never on a path to any iterOut, so scope analysis leaves them in the graph scope, and without the exception they would miss the terminal step entirely (the run exits with them still queued). Mid-loop this only fires them earlier than the main loop would have; inner boundaries keep the strict filter.
• ③ Decide. Read the just-fired iterOut's own stop input: truthy means this scope is done. A root-scope stop halts the run; an inner one ends only that loop. Then the per-scope step_budget exhaust check. Either way, termination triggers the final-side emission: the terminal iteration's collected values go out once on the final_* handles (root → seeds the after-loop stage, §5; inner → feeds outer-body consumers and the graphOut latch flush). The stop signal sits on the scope's own iterOut, so it is structurally bound to the right loop and read exactly once per iteration.
• ④ Hand off. Sleep step_delay_ms (root only), then write each IterOut output X into the paired IterIn's iterout_X slot and re-enqueue the IterIn — arming the next iteration. (stop is not loop-carried, and the final-side handles never fire here — only at termination.)

4.4 Ordinary nodes & propagation

Every non-IterOut fire ends the same way, in four moves:

1. Consume. The node's pending_inputs are cleared — inputs are used up by the fire.
2. Special cases. An iterIn additionally drops its persist=false slots, so one-shot values emit once and are gone. An inner-scope graphOut buffers its value into a latch (state["latched_value"]) instead of propagating outward — the latch is flushed to outer-scope consumers only when its scope terminates.
3. Route. For every other node the engine walks its outgoing edges and writes each output onto the target's input port (_route_value_to_port). This is also where LIST[T] coercion happens (ADR-027): a scalar producer feeding a list consumer is auto-wrapped, and fan-in is concatenated in edge order.
4. Enqueue. Any target whose required inputs are now all present joins the ready-queue.

(Run-start args need no special branch: the iterIn's init_* handles are fed by ordinary canvas edges — ADR-dataflow-008 two-sided iterIn.)

4.5 The whole run, end to end

Zoom out from one iteration to the whole of run(). The flowchart below stitches §3 and §4 into one picture, with the ready-queue at its centre — ENTRY DISCOVERY primes it, every turn pops it, and both return paths feed it again. Read it top-down:

1. Set up once. BUILD compiles the graph JSON into live state (§3.1); ENTRY DISCOVERY finds the no-input entry nodes and primes the ready-queue (§3.2).
2. Turn the loop. While the queue has a ready node — and neither the global firing cap nor a user Stop has hit — pop the front one and fire it: guards first (§4.2), then _fire_node.
3. Fork on what fired. An ordinary node takes the fast path: route its outputs, enqueue whatever became ready (§4.4), turn again — this is almost every fire. An IterOut instead runs the once-per-iteration boundary (§4.3): Record, Settle, Decide, Hand off. Unless Decide halts, Hand off arms the paired IterIn and the loop turns again — that is one full iteration.
4. Exit. Two doors out. Decide halts — the stop signal or the step budget — and the terminal iteration's values leave exactly once on the final_* handles. Or the main loop runs dry — queue drained, firing cap, user Stop — and the fallback reconstructs the final-side values best-effort. Every door lands in the same after-loop + finalise stage (§5); §7 tables every exit, including the StopExecution escape hatch.

That is the entire control flow — and the engine-view counterpart of §1.2: there the boxes were the nodes of one drawn graph; here they are the engine's actions on any graph. Part II §6 prints this same shape as code — the colored region bands there are exactly these boxes.

4.6 A concrete trace

One episode of the §1.2 graph, turn by turn. Watch the ready-queue column — it is the hub of the §4.5 flowchart; entry discovery put entry in it (and only entry: IterIn is never an entry node, §3.2). Each turn pops the front, fires it, and the effect column says what that made ready next.

Note what never happens: evaluate appears in the queue exactly once, at turn 8 — during the loop its inputs simply never arrive (§5.1), so the verdict cannot read anything but the terminal iteration.

5. After-loop — the verdict stage & finalise

However the loop exits — the stop signal fired, the step budget hit, the queue drained, or a node crashed — control lands in one finalise stage. It has four movements: the after-loop pass (§5.1), the node-error conviction gate and clean finalisation (§5.2), and the error path (§5.3).

5.1 The after-loop pass

The after-loop band is every node wired downstream of the root iterOut's final_* handles — the verdict chain, typically evaluate → graphOut, possibly with adjudicator steps in between. Two facts keep it dormant while the loop runs: the final side never emits mid-loop, and the validator forbids edges from the loop body into the band. Band nodes therefore cannot become ready before termination — there is nothing to skip, defer, or flag.

When the final-side values arrive, _after_loop_pass drains whatever they made ready: fire a ready node, route its outputs, enqueue the consumers that became ready in turn. A node whose upstream band node hasn't fired yet is re-queued behind it, so a chain resolves in dependency order without a topological sort; the queue is bounded (max(nodes×4, 64)), so a never-satisfiable node cannot spin forever.

Two rules keep the pass honest:

• Graph-scope only. The loop is over, so re-firing a loop-body node — or any iterIn / iterOut, or any node inside an author (loop) scope — is meaningless and barred. Only graph-scope nodes (the band and its sinks) fire.
• graphOut is harvested, not fired. A graph-scope graphOut the chain reaches is routed-to and re-snapshotted into state["_last_inputs"] — the channel the metric harvest reads — but never itself fired; inner-scope graphOuts are skipped entirely (their latch flush is scope-bound).

Membership is topological, not declared: there is no flag to set (the former config.post_loop was removed 2026-06-11) — a node is after-loop because it hangs off the final side. The input contract is engine-guaranteed: the band receives exactly the terminal iteration's values ("verdict inputs ride the pivot" — anything the verdict needs must be an iterOut port, consumed via final_<name>). If the loop ends without a boundary emission (drained queue, max_firings, user stop, or a crash), _emit_final_fallback reconstructs the last completed iteration's values from the paired iterIn's iterout_* slots and emits best-effort before the drain.

5.2 Clean finalise

The conviction gate comes first (2026-07-04). At fire time the executor records every node failure — a raised exception, or an {"error": ...}-shaped result such as a server-mode proxy surfacing an HTTP 500 — into a per-run node_errors ledger (nodes that declare a legitimate error output port are exempt; returned errors also emit a NODE_RESULT_ERROR bus event). If the ledger is non-empty here, a NodeErrorAggregate naming the culprits is raised and control transfers to the error path (§5.3). The placement is deliberate: the verdict stage (§5.1) has already run, so final-side metrics are collected either way — but the run can no longer masquerade as completed while its downstream silently starved (the old status="completed" / step_count=0 eval pathology).

On a clean exit (empty ledger), in order:

1. Harvest metrics — from any node's state["metrics"]; a value stored on session._metrics overrides node values.
2. Broadcast run_end — lifetime="run" state containers clear.
3. Broadcast nav_complete — final step and metrics, to the frontend.
4. Fire the GraphComplete hook.
5. Flush the execution log. Session status becomes done.

5.3 The error path

If the loop body raises — or the conviction gate fired (§5.2) — the finalise stage still runs, and runs the verdict first:

1. Fallback emission + after-loop pass, best-effort. The ordering is deliberate: even after a crash the verdict evaluate can often still emit fresh metrics from the env's real final step, so the harvest reports the true result rather than a stale pre-crash value. The pass is tolerant — a band node that raises because the env is already torn down is logged and skipped (and both stages are idempotent, so re-entry from the conviction gate is a no-op).
2. Publish the exception to the ErrorBus — GRAPH_CRASH for a real crash, NODE_ERRORS for a conviction; session status becomes error, broadcast to the frontend as a nav_status error.
3. run_end still fires, carrying the error — run-lifetime state resets even on a crash.
4. Fire the GraphError hook, then the final log flush.

6. The run() listing — synced from source

The full method with its real comments: the annotations you read here are the source file's comments, so the listing and the engine cannot disagree about what a block does. The blue band headers are the file's own # ─── banner comments — the same phase vocabulary as Part I — with the matching section linked on each band, and the line numbers are real. After editing run(), refresh with python3 docs/_lib/_sync_run_listing.py; the sync stamp below names the commit it was last generated against.

Synced from graph_executor.py:274–1080 at c7ba99d by docs/_lib/_sync_run_listing.py — the band headers are the file's own banner comments; re-run the script after editing the source.

run() — signature & docstring ── graph_executor.py:274–300
274      async def run(
275          self,
276          graph: GraphDefinition,
277          session: Any,
278          execution_id: str | None = None,
279          step_delay_ms: int = 200,
280          stop_event: asyncio.Event | None = None,
281          pause_event: asyncio.Event | None = None,
282          global_hooks: list[Any] | None = None,
283          step_budget_override: int | None = None,
284      ) -> None:
285          """Main graph execution loop.
286
287          ``step_budget_override`` (when set) wins over the graph's authored
288          ``step_budget``. Used by the eval batch resolver chain to push an
289          env-supplied per-episode value (e.g. HM-EQA's
290          ``int(sqrt(scene_size) * 3)``) without mutating the shared graph
291          object across worker coroutines.
292
293          Structure (the ``# ───`` banners below mark the phases):
294          pre-loop (BUILD steps 1-7, then ENTRY DISCOVERY) → in-loop (the
295          firing loop; the iterOut boundary closes each iteration in four
296          phases: record → settle → decide → hand off) → after-loop (the
297          verdict stage) → finalise. Narrated walkthrough with diagrams:
298          docs/pages/developer-guide/design-docs/graph-executor.html (Part I).
299          """
300
pre-loop · BUILD — compile the static graph into live state ── graph_executor.py:301–421 · ↔ §3.1
302
303          # BUILD 1/7 — merge hooks from 3 sources: global → graph → node (R3)
304          node_hooks = extract_node_hooks(graph.nodes)
305          merged = merge_hooks(global_hooks or [], graph.hooks, node_hooks)
306          self._hook_runner = HookRunner(merged)
307
308          # BUILD 2/7 — flatten composite nodes before execution (ADR-dataflow-001)
309          from .flatten import flatten_graph
310
311          graph, self._flatten_map = flatten_graph(graph)
312
313          # BUILD 3/7 — state containers + access grants from the graph definition
314          if graph.containers:
315              from .state_containers import build_containers
316
317              self.containers = build_containers(graph.containers)
318              for ag in graph.access_grants:
319                  self._access_grant_index.setdefault(ag.node_id, set()).add(ag.container_id)
320              log.info(
321                  "State containers: %d containers, %d access grants",
322                  len(self.containers),
323                  len(graph.access_grants),
324              )
325              # If a container with the well-known id "graph_state" exists,
326              # set up the convenience binding.  Access is still gated by
327              # explicit access grants — no auto-inject.
328              if "graph_state" in self.containers:
329                  self._graph_state_id = "graph_state"
330
331          # BUILD 4/7 — resolve the step budget. Order: explicit override →
332          # graph authored value → framework default. The eval batch runner
333          # computes the override via its resolver chain (env-dynamic → API
334          # override).
335          from ..config import DEFAULT_STEP_BUDGET as _DEFAULT_STEP_BUDGET
336
337          if step_budget_override is not None:
338              self.step_budget = step_budget_override
339          elif graph.step_budget is not None:
340              self.step_budget = graph.step_budget
341          else:
342              self.step_budget = _DEFAULT_STEP_BUDGET
343          self.edges = [e.to_dict() for e in graph.edges]
344          node_defs = graph.nodes
345
346          if not node_defs:
347              log.error("Empty graph — nothing to execute")
348              return
349
350          # BUILD 5/7 — node instances: the static definition becomes
351          # per-run mutable state (pending_inputs buffer + persistent state)
352          for nd in node_defs:
353              self.nodes[nd.id] = NodeInstance(
354                  id=nd.id,
355                  type=nd.type,
356                  config=nd.config,
357                  label=nd.label,
358              )
359
360          # BUILD 6/7 — per-scope execution state. Always includes a
361          # synthetic GRAPH_SCOPE_ID entry; for single-scope graphs there is
362          # also one entry keyed by the outermost author scope's iter_in id.
363          # Multi-scope graphs add one entry per author scope. The root
364          # scope's stop signal halts the entire run; a non-root scope's stop
365          # only ends that inner loop and lets the outer one continue.
366          self.scope_forest, scope_errors = analyze_scopes(graph)
367          if scope_errors:
368              for err in scope_errors:
369                  log.warning("Scope analysis: %s", err)
370          # Always create the synthetic graph-scope entry — holds graph-scope
371          # nodes (run-start entry nodes, after-loop sinks).
372          self.scope_state[GRAPH_SCOPE_ID] = _ScopeState(
373              scope_id=GRAPH_SCOPE_ID,
374              parent_scope_id=None,
375              step_budget=self.step_budget,  # graph-level cap also applies here
376              member_node_ids=set(self.scope_forest.graph_scope_node_ids),
377          )
378          # Per-author-scope state; resolve step_budget with fallback chain
379          # (scope's own iter_in.step_budget → parent scope's step_budget →
380          # graph step_budget).
381          for scope_id, info in self.scope_forest.scopes.items():
382              resolved_budget = info.step_budget
383              if resolved_budget is None:
384                  # Walk up parent chain looking for a budget; fall back to graph
385                  p = info.parent_scope_id
386                  while p is not None and resolved_budget is None:
387                      p_info = self.scope_forest.scopes.get(p)
388                      if p_info is None:
389                          break
390                      resolved_budget = p_info.step_budget
391                      p = p_info.parent_scope_id
392              if resolved_budget is None:
393                  resolved_budget = self.step_budget
394              self.scope_state[scope_id] = _ScopeState(
395                  scope_id=scope_id,
396                  parent_scope_id=info.parent_scope_id,
397                  step_budget=resolved_budget,
398                  member_node_ids=set(info.member_node_ids),
399                  graphout_node_ids=list(info.graphout_node_ids),
400                  iter_out_id=info.iter_out_id,
401              )
402          # The "outermost" scope id — used for backward compat: self.step_counter
403          # and self.terminated mirror this scope's state. For 0-scope graphs
404          # this stays GRAPH_SCOPE_ID; for single-scope graphs it's the sole
405          # author scope's id; for multi-scope graphs it's the (single) root.
406          if len(self.scope_forest.root_scope_ids) == 1:
407              self._outermost_scope_id = self.scope_forest.root_scope_ids[0]
408          elif len(self.scope_forest.root_scope_ids) > 1:
409              # Multiple peer roots — pick the first deterministically; legacy
410              # self.step_counter mirrors only its counter (multi-root graphs
411              # are an advanced shape; consumers should read scope_state directly).
412              self._outermost_scope_id = self.scope_forest.root_scope_ids[0]
413          else:
414              self._outermost_scope_id = GRAPH_SCOPE_ID
415
416          # BUILD 7/7 — adjacency: outgoing edges indexed by source id, so
417          # routing a fired node's outputs is a single lookup in the loop
418          for edge in self.edges:
419              src = edge.get("source", "")
420              self.adjacency.setdefault(src, []).append(edge)
421
pre-loop · ENTRY DISCOVERY — prime the ready-queue, arm the loop ── graph_executor.py:422–492 · ↔ §3.2
423
424          # A node is an entry node if: explicitly marked OR (has no required
425          # inputs AND has no incoming edges). Nodes with incoming edges to
426          # optional ports should wait for data even if no required ports exist.
427          incoming_targets = set()
428          for edge in self.edges:
429              tgt = edge.get("target", "")
430              if tgt:
431                  incoming_targets.add(tgt)
432
433          # Entry discovery: queue at run-start iff all three structural
434          # conditions hold:
435          #   1. type != "iterIn" (iterIn fires from init edges / iterOut transfer)
436          #   2. no incoming edges
437          #   3. no required input ports
438          # validate_graph_connectivity (graph_def.py) rejects required-but-
439          # unwired ports at load time, so condition 3 normally only excludes
440          # well-formed leaf consumers — never silently drops an entry node.
441          # System Log: ready-time stamps for queue_wait_ms (set on enqueue,
442          # popped on fire in _fire_node). Run-scoped — reset on each run().
443          self._t_ready: dict[str, float] = {}
444          ready_queue: list[str] = []
445          for node in self.nodes.values():
446              if node.type == "iterIn":
447                  continue
448              if node.id in incoming_targets:
449                  continue
450              if self._get_required_ports_for_node(node):
451                  continue
452              self._enqueue(ready_queue, node.id)
453
454          log.info(
455              "GraphExecutor: %d nodes, %d edges, entry_nodes=%s, step_budget=%d",
456              len(self.nodes),
457              len(self.edges),
458              [self.nodes[nid].label or nid for nid in ready_queue],
459              self.step_budget,
460          )
461
462          stop = stop_event or asyncio.Event()
463          pause = pause_event or asyncio.Event()
464          if not pause.is_set():
465              pause.set()
466
467          session._status = "running"
468          # Node failures recorded during this run — convicts the run at the
469          # finalise stage (see NodeErrorAggregate).
470          self.node_errors: list[dict] = []
471          _suppress = getattr(getattr(session, "principles", None), "suppress_nav_events", False)
472          if not _suppress:
473              await broadcast(session._ws("nav_status", {"status": "running", "step": 0}))
474
475          # run_start signal fires before any node fires, so state
476          # containers with lifetime="run" reset to a clean baseline.
477          self.broadcast_signal("run_start", {"graph_name": graph.name})
478
479          # GraphStart hook
480          if self._hook_runner.has_hooks():
481              await self._hook_runner.run_hooks(
482                  "GraphStart",
483                  payload={
484                      "graph_name": graph.name,
485                      "node_count": len(self.nodes),
486                  },
487              )
488
489          # Safety: max total firings to prevent infinite loops
490          max_firings = self.step_budget * len(self.nodes) * 3
491          total_firings = 0
492
in-loop · the firing loop — pop, guard, fire, route ── graph_executor.py:493–672 · ↔ §4.1–§4.2
494          try:
495              while ready_queue and total_firings < max_firings:
496                  # Pause/stop check
497                  await pause.wait()
498                  if stop.is_set():
499                      break
500
501                  node_id = ready_queue.pop(0)
502                  node = self.nodes.get(node_id)
503                  if node is None:
504                      continue
505
506                  # Guard 1 — scope barrier. Once an inner scope is marked
507                  # terminated (by the iterOut boundary's decide check at the
508                  # end of the final inner iter), forbid any further fire of nodes belonging
509                  # to that scope until the scope is re-entered. Without this
510                  # guard, in-flight inner-body nodes that were enqueued
511                  # before the scope stopped (e.g. ``move_to_pose`` already
512                  # running, or ``episode_info`` triggered by it) fire one
513                  # extra time AFTER the stop and leak per-iter outputs through any
514                  # cross-scope wires (post-flatten direct edges from inner
515                  # producers to outer consumers, OR direct graphOut→outer
516                  # wires when graphOut is preserved), satisfying outer
517                  # iterOut's required gates and causing outer to cycle
518                  # without inner actually re-entering.
519                  #
520                  # Exemption set — scope-entry cascade nodes that MUST be
521                  # allowed to fire so the next outer iter can reopen this
522                  # scope: ``iterIn`` (existing re-entry path below clears
523                  # the terminated flag; its init slots re-capture
524                  # outer-supplied scope-entry args), ``graphIn`` (ferries
525                  # outer→inner values across the boundary into iterIn's
526                  # init side). Body nodes and iterOut stay barred — they
527                  # have nothing useful to do between scope terminate and
528                  # scope re-entry.
529                  _n_scope_id = self._scope_of(node_id)
530                  _n_scope = (
531                      self.scope_state.get(_n_scope_id)
532                      if _n_scope_id and _n_scope_id != GRAPH_SCOPE_ID
533                      else None
534                  )
535                  if (
536                      _n_scope is not None
537                      and _n_scope.terminated
538                      and _n_scope.parent_scope_id is not None
539                      and node.type not in ("iterIn", "graphIn")
540                  ):
541                      node.pending_inputs = {}
542                      continue
543
544                  # Guard 2 — iterIn step_start. iterIn marks the start of each
545                  # iteration — emit step_start before the node fires so
546                  # lifetime="step" states see a fresh slate for the upcoming
547                  # iteration.
548                  # Multi-scope re-entry: an inner scope's iter_in firing
549                  # while its scope is marked terminated indicates outer just
550                  # entered a new outer iter and is re-invoking inner. Reset
551                  # inner scope's terminated flag and step_counter — inner
552                  # gets a fresh loop. (Root-scope iter_in re-firing after
553                  # termination would only happen if main loop didn't break;
554                  # we still allow the reset for symmetry, though in practice
555                  # root termination breaks the loop above.)
556                  if node.type == "iterIn":
557                      _ii_scope = self.scope_state.get(node.id)
558                      if _ii_scope is not None and _ii_scope.terminated:
559                          if _ii_scope.parent_scope_id is not None:
560                              # Inner scope re-entry from outer iter — reset
561                              log.info(
562                                  "Scope re-entry: %s (parent=%s) — resetting terminated/counter",
563                                  node.id,
564                                  _ii_scope.parent_scope_id,
565                              )
566                              _ii_scope.terminated = False
567                              _ii_scope.step_counter = 0
568                              # Re-arm the final side: an inner scope emits
569                              # final_* once per termination, i.e. once per
570                              # outer iteration.
571                              self._final_emitted.discard(node.id)
572                              # Reset per-fire state on body nodes that hold
573                              # transient counters (only those tracked via
574                              # state['_scoped_reset']=True)? — out of scope
575                              # for v1; user tests must use cumulative counters.
576                          else:
577                              # Root scope: should not happen (main loop breaks
578                              # on root termination). Defensive: skip the fire.
579                              node.pending_inputs = {}
580                              continue
581                      # step_start signal — add scope_id (additive, single-scope
582                      # readers using just `step` keep working).
583                      _next_step = (
584                          _ii_scope.step_counter + 1
585                          if _ii_scope is not None
586                          else self.step_counter + 1
587                      )
588                      self.broadcast_signal(
589                          "step_start",
590                          {"step": _next_step, "scope_id": node.id},
591                      )
592
593                  # Fire the node
594                  _error_from_exception = False
595                  try:
596                      result = await self._fire_node(node, session)
597                  except StopExecution as e:
598                      log.info(
599                          "StopExecution from %s: %s at step %d", node.id, e.reason, self.step_counter
600                      )
601                      self.terminated = True
602                      break
603                  except Exception as e:
604                      origin = self._flatten_map.trace(node.id) if self._flatten_map else node.id
605                      # Surface to user via Report tab; result still gets {"error": ...}
606                      # so downstream nodes can react to the failure.
607                      get_bus().from_exception(
608                          e,
609                          source="node",
610                          code="NODE_EXEC_FAIL",
611                          scope={
612                              "node_id": node.id,
613                              "node_type": node.type,
614                              "origin": origin,
615                              "step": self.step_counter,
616                              "execution_id": getattr(session, "_execution_id", None),
617                          },
618                          title=f"Node {node.id} ({node.type}) failed",
619                      )
620                      # Strict mode (AGENTCANVAS_STRICT_ERRORS=1): re-raise after
621                      # bus emit so the outer execution loop fails fast instead of
622                      # quietly continuing with result={"error": ...}. Designed for
623                      # smoke / eval runs where any node failure should be visible.
624                      if os.environ.get("AGENTCANVAS_STRICT_ERRORS", "").lower() in (
625                          "1",
626                          "true",
627                          "yes",
628                          "on",
629                      ):
630                          raise
631                      result = {"error": str(e)}
632                      _error_from_exception = True
633
634                  # Error-shaped result — the node reported failure by returning
635                  # {"error": ...} instead of its declared ports (server-mode
636                  # proxies surface HTTP failures this way; the exception path
637                  # above converts to the same shape). Routing would silently
638                  # drop it: downstream starves and the run drains away as if
639                  # completed. Record it for the end-of-run conviction, and put
640                  # returned errors on the bus (the exception path already
641                  # emitted NODE_EXEC_FAIL). Nodes that legitimately declare an
642                  # ``error`` output port (e.g. env_libero tools) are exempt.
643                  if (
644                      isinstance(result, dict)
645                      and "error" in result
646                      and not self._declares_error_output(node)
647                  ):
648                      self.node_errors.append(
649                          {
650                              "node_id": node.id,
651                              "node_type": node.type,
652                              "step": self.step_counter,
653                              "error": str(result["error"]),
654                          }
655                      )
656                      if not _error_from_exception:
657                          get_bus().emit(
658                              severity="error",
659                              source="node",
660                              code="NODE_RESULT_ERROR",
661                              title=f"Node {node.id} ({node.type}) returned an error result",
662                              message=str(result["error"]),
663                              scope={
664                                  "node_id": node.id,
665                                  "node_type": node.type,
666                                  "step": self.step_counter,
667                                  "execution_id": getattr(session, "_execution_id", None),
668                              },
669                          )
670
671                  total_firings += 1
672
iterOut boundary — fires once per iteration; four phases — record → settle → decide → hand off ── graph_executor.py:673–681 · ↔ §4.3
675                  if node.type == "iterOut":
676                      # Resolve which scope this iterOut belongs to. The scope
677                      # is keyed by the paired iterIn's id (== scope_id).
678                      _io_scope_id = node.config.get("pairedWith", "") or self._outermost_scope_id
679                      _io_scope = self.scope_state.get(_io_scope_id)
680                      _is_root_scope = _io_scope_id == self._outermost_scope_id
681
boundary phase 1/4 · record — counters, checkpoint, log, nav_step ── graph_executor.py:682–735 · ↔ §4.3 ①
683                      # Advance per-scope counter. For backward compat,
684                      # self.step_counter mirrors the OUTERMOST scope's counter
685                      # (single-scope graphs unchanged).
686                      if _io_scope is not None:
687                          _io_scope.step_counter += 1
688                          _scope_step = _io_scope.step_counter
689                      else:
690                          _scope_step = self.step_counter + 1
691                      if _is_root_scope:
692                          self.step_counter += 1
693                          session._current_step = self.step_counter
694
695                      _parent_scope_id = _io_scope.parent_scope_id if _io_scope is not None else None
696                      log.info(
697                          "Iter: scope=%s step=%d (parent=%s, root=%s)",
698                          _io_scope_id,
699                          _scope_step,
700                          _parent_scope_id,
701                          _is_root_scope,
702                      )
703
704                      self.broadcast_signal(
705                          "step_end",
706                          {"step": _scope_step, "scope_id": _io_scope_id},
707                      )
708
709                      # Checkpoint all containers AFTER iter boundary
710                      # Semantics: snapshot = state ready for step N+1 to begin
711                      # Only checkpoint on root-scope iterations (single-scope
712                      # graphs unchanged; nested inner scopes don't checkpoint)
713                      if self.containers and _is_root_scope:
714                          self._checkpoints[self.step_counter] = {
715                              cid: c.checkpoint() for cid, c in self.containers.items()
716                          }
717                          if len(self._checkpoints) > self.max_checkpoints:
718                              oldest = min(self._checkpoints)
719                              del self._checkpoints[oldest]
720
721                      # Flush execution log entries to JSONL at iteration boundary
722                      if self._logger:
723                          self._logger.flush()
724
725                      # Broadcast consolidated nav_step from all output viewer data
726                      # Only on root-scope iter (matches pre-refactor: one
727                      # nav_step per outer iteration).
728                      if _is_root_scope:
729                          await self._broadcast_step(session)
730
731                      # Edges FROM iterOut are final-side only (``final_*``
732                      # handles) — they emit once at scope termination via
733                      # ``_emit_final_side``, never per-iteration, so iterOut's
734                      # own outputs are NOT propagated to adjacency here.
735
boundary phase 2/4 · settle — drain this iter's leftover sinks ── graph_executor.py:736–820 · ↔ §4.3 ②
737                      # Settle BEFORE the stop check, so in-loop viewer /
738                      # telemetry sinks scheduled in this iter's wave still
739                      # emit on the terminal step.
740                      # iterIn is excluded — it advances to next iter and is
741                      # correctly fired only via the pairedWith handoff after
742                      # the stop check.
743                      # Multi-scope: settle drain restricted to nodes belonging
744                      # to the same scope as the just-fired iterOut. Outer-scope
745                      # / peer-scope nodes are not drained by inner iter_out.
746                      _max_settle = 64  # safety cap against pathological queues
747                      _settle_n = 0
748                      while ready_queue and _settle_n < _max_settle:
749                          # Skip iterIn — next iter, handled by the pairedWith
750                          # handoff after the stop check.
751                          _next_idx = None
752                          for _i, _nid in enumerate(ready_queue):
753                              _n = self.nodes.get(_nid)
754                              if _n is None or _n.type == "iterIn":
755                                  continue
756                              # Multi-scope: only drain nodes in the same scope
757                              # as the iterOut that triggered this settle pass.
758                              # Single-scope graphs: every body node is in the
759                              # outermost scope, so this matches today's behaviour.
760                              # Exception — ROOT boundaries also drain graph-scope
761                              # nodes: dead-end sinks are never on a path to any
762                              # iterOut, so scope analysis leaves them in the
763                              # graph scope; without this they miss the terminal
764                              # step entirely (the run exits with them still
765                              # queued). Mid-loop this only fires them earlier
766                              # than the main loop would have. Inner boundaries
767                              # keep the strict filter — the run continues and
768                              # the main loop drains them.
769                              _n_scope = self._scope_of(_nid)
770                              if (
771                                  _io_scope is not None
772                                  and _n_scope != _io_scope_id
773                                  and not (_is_root_scope and _n_scope == GRAPH_SCOPE_ID)
774                              ):
775                                  continue
776                              _next_idx = _i
777                              break
778                          if _next_idx is None:
779                              break  # nothing to settle in this scope
780                          _nid = ready_queue.pop(_next_idx)
781                          _n = self.nodes[_nid]
782                          try:
783                              _r = await self._fire_node(_n, session)
784                          except Exception:
785                              log.exception("settle: error firing %s", _nid)
786                              _n.pending_inputs = {}
787                              _settle_n += 1
788                              continue
789                          _n.pending_inputs = {}
790                          # Multi-scope: graphOut nodes inside a non-graph
791                          # scope BUFFER value into state["latched_value"]
792                          # instead of propagating outward. Mirrors the same
793                          # suppression in the standard propagation block;
794                          # without it, every inner iter would prematurely
795                          # propagate graphOut to outer downstream.
796                          if _n.type == "graphOut" and self._scope_of(_nid) != GRAPH_SCOPE_ID:
797                              if isinstance(_r, dict):
798                                  _n.state["latched_value"] = _r.get("value")
799                              _settle_n += 1
800                              continue
801                          # Mirror the propagation block below (not extracted
802                          # to keep main-loop changes minimal).
803                          if isinstance(_r, dict):
804                              for _edge in self.adjacency.get(_nid, []):
805                                  _tgt_id = _edge.get("target", "")
806                                  if not _tgt_id:
807                                      continue
808                                  _tgt = self.nodes.get(_tgt_id)
809                                  if _tgt is None:
810                                      continue
811                                  _sh = _edge.get("sourceHandle", "default")
812                                  _th = _edge.get("targetHandle", _sh)
813                                  if _sh in _r:
814                                      self._route_value_to_port(_tgt, _th, _r[_sh])
815                                  elif _sh == "default":
816                                      self._route_value_to_port(_tgt, _th, _r)
817                                  if self._is_ready(_tgt):
818                                      self._enqueue(ready_queue, _tgt_id)
819                          _settle_n += 1
820
boundary phase 3/4 · decide — stop input, then budget check ── graph_executor.py:821–887 · ↔ §4.3 ③
822                      # Read the just-fired iterOut's own ``stop`` input. The
823                      # stop signal is structurally bound to its scope (it sits
824                      # on the scope's own iterOut), and this check runs exactly
825                      # once per iteration, after the settle drain and before
826                      # the handoff. Unwired stop = budget-only loop.
827                      # Termination (stop or budget) emits the terminal
828                      # iteration's values exactly once on the final_* handles.
829                      _stop = isinstance(result, dict) and bool(result.get("stop"))
830                      if _stop:
831                          log.info(
832                              "Stop: iterOut %s stop=True at scope=%s step=%d",
833                              node_id,
834                              _io_scope_id or "(graph)",
835                              _scope_step,
836                          )
837                          if _io_scope is not None:
838                              _io_scope.terminated = True
839                          # Final side: emit the terminal iteration's values
840                          # exactly once on the final_* handles.
841                          self._emit_final_side(
842                              node,
843                              result,
844                              _io_scope_id,
845                              ready_queue=None if _is_root_scope else ready_queue,
846                          )
847                          if _is_root_scope:
848                              self.terminated = True
849                              break
850
851                      # Per-scope step_budget exhaust check.
852                      _scope_budget = (
853                          _io_scope.step_budget if _io_scope is not None else self.step_budget
854                      )
855                      if _scope_step >= _scope_budget:
856                          log.info(
857                              "Step budget (%d) exhausted for scope=%s",
858                              _scope_budget,
859                              _io_scope_id or "(graph)",
860                          )
861                          if _io_scope is not None:
862                              _io_scope.terminated = True
863                          self._emit_final_side(
864                              node,
865                              result if isinstance(result, dict) else {},
866                              _io_scope_id,
867                              ready_queue=None if _is_root_scope else ready_queue,
868                          )
869                          if _is_root_scope:
870                              self.terminated = True
871                              break
872                          # Inner-scope budget exhaust: don't break root loop;
873                          # propagate inner graphOut latches and let outer continue
874                          self._propagate_graphout_latches(_io_scope, ready_queue)
875                          # Skip the iterIn re-queue below (scope is done)
876                          continue
877
878                      if step_delay_ms > 0 and _is_root_scope:
879                          await asyncio.sleep(step_delay_ms / 1000)
880
881                      # If this scope was just stopped (inner-scope stop above,
882                      # without breaking the root loop), propagate graphOut
883                      # latch + skip the iterIn re-queue.
884                      if _io_scope is not None and _io_scope.terminated:
885                          self._propagate_graphout_latches(_io_scope, ready_queue)
886                          continue
887
boundary phase 4/4 · hand off — carry → paired iterIn, next iteration ── graph_executor.py:888–909 · ↔ §4.3 ④
889                      # iterIn slots are prefixed with "iterout_" for
890                      # iterOut writes (always-prefix synthesis). Each iterOut
891                      # output key ``X`` maps to iterIn slot ``iterout_<X>``.
892                      # ``stop`` is not loop-carried; final_* handles never fire here.
893                      paired_id = node.config.get("pairedWith", "")
894                      paired = self.nodes.get(paired_id) if paired_id else None
895                      if paired and paired.type == "iterIn":
896                          slot_names = _iterin_port_names(paired)
897                          for key, val in result.items():
898                              slot_key = f"iterout_{key}"
899                              if slot_key in slot_names:
900                                  self._write_iterin_slot(paired, slot_key, val)
901                          self._enqueue(ready_queue, paired.id)
902
903                      # Standard propagation (line ~750 below) is skipped for
904                      # iterOut — adjacency was already propagated above (before
905                      # the settle drain) so latch buffers see the final iter
906                      # value. Clear pending_inputs and continue.
907                      node.pending_inputs = {}
908                      continue
909
ordinary path — consume · special cases · route · enqueue ── graph_executor.py:910–974 · ↔ §4.4
911                  # Clear fired node's inputs (consumed)
912                  node.pending_inputs = {}
913
914                  # iterIn: after fire, clear slots for ports with persist=False
915                  # so they don't re-emit next iteration. persist=True slots
916                  # stay populated until the next write by a matching writer
917                  # (run-start init edge, or iterOut at iter boundary).
918                  # Ordering — fire → clear → re-enqueue — persist=False
919                  # semantics: emit once on the step of the write, then gone.
920                  if node.type == "iterIn":
921                      persist_map = _iterin_persist_map(node)
922                      for slot_name, keep in persist_map.items():
923                          if not keep:
924                              node.port_slots.pop(slot_name, None)
925
926                  # Multi-scope: graphOut nodes inside a non-graph scope BUFFER
927                  # their value into ``state["latched_value"]`` instead of
928                  # propagating out immediately. ``_propagate_graphout_latches``
929                  # flushes the latch to outer-scope downstream when the
930                  # owning scope terminates. graphOuts in the graph scope (eval
931                  # graph metric harvest) propagate normally as before.
932                  if node.type == "graphOut":
933                      _po_scope = self._scope_of(node_id)
934                      if _po_scope != GRAPH_SCOPE_ID and isinstance(result, dict):
935                          node.state["latched_value"] = result.get("value")
936                          if total_firings <= 5:
937                              log.info(
938                                  "After %s: queue=%s",
939                                  node.label or node.id,
940                                  [self.nodes[nid].label or nid for nid in ready_queue],
941                              )
942                          continue  # skip standard propagation block below
943
944                  # Propagate outputs to downstream nodes
945                  for edge in self.adjacency.get(node_id, []):
946                      target_id = edge.get("target", "")
947                      target = self.nodes.get(target_id)
948                      if target is None:
949                          continue
950
951                      src_handle = edge.get("sourceHandle", "default")
952                      tgt_handle = edge.get("targetHandle", src_handle)
953
954                      # Route specific port value — the _route_value_to_port
955                      # helper applies LIST[T] coercion (ADR-027) so scalar
956                      # producers feeding a LIST[T] consumer are auto-wrapped
957                      # and fan-in concatenates in edge declaration order.
958                      if isinstance(result, dict) and src_handle in result:
959                          self._route_value_to_port(target, tgt_handle, result[src_handle])
960                      elif isinstance(result, dict) and src_handle == "default":
961                          self._route_value_to_port(target, tgt_handle, result)
962
963                      # Check if target is now ready to fire
964                      is_ready = self._is_ready(target)
965                      if is_ready:
966                          self._enqueue(ready_queue, target_id)
967
968                  if total_firings <= 5:
969                      log.info(
970                          "After %s: queue=%s",
971                          node.label or node.id,
972                          [self.nodes[nid].label or nid for nid in ready_queue],
973                      )
974
after-loop · verdict stage — fallback emission + band drain ── graph_executor.py:975–984 · ↔ §5.1
976              # If the loop ended without a boundary final-side emission
977              # (drained queue, max_firings, user stop), reconstruct the last
978              # completed iteration's values from the paired iterIn's slots
979              # and emit best-effort; then drain the after-loop band — the
980              # nodes fed by the root iterOut's final_* handles (evaluate,
981              # graphOut chains). This is the run-end verdict stage.
982              self._emit_final_fallback()
983              await self._after_loop_pass(session)
984
conviction — surface node failures at end of run ── graph_executor.py:985–998
986              # Runs AFTER the verdict stage so final-side metrics are already
987              # collected; the raise routes through the error path below and
988              # the run finishes with status="error" instead of masquerading
989              # as completed (both stages are idempotent on the re-entry).
990              if self.node_errors:
991                  _n_err = len(self.node_errors)
992                  _head = "; ".join(
993                      f"{e['node_id']}@step{e['step']}: {e['error']}" for e in self.node_errors[:3]
994                  )
995                  raise NodeErrorAggregate(
996                      f"{_n_err} node error(s) during run: {_head}" + ("; ..." if _n_err > 3 else "")
997                  )
998
finalise — clean exit ── graph_executor.py:999–1038 · ↔ §5.2
1000             session._status = "done"
1001             metrics = None
1002             # Find metrics from any node that stored them (generic — any step node)
1003             for node in self.nodes.values():
1004                 if node.state.get("metrics"):
1005                     metrics = node.state["metrics"]
1006             if session._metrics:
1007                 metrics = session._metrics
1008
1009             # run_end signal — fires after the loop finishes cleanly.
1010             # lifetime="run" states clear here.
1011             self.broadcast_signal(
1012                 "run_end",
1013                 {"step": self.step_counter, "terminated": self.terminated},
1014             )
1015
1016             if not _suppress:
1017                 await broadcast(
1018                     session._ws(
1019                         "nav_complete",
1020                         {"step": self.step_counter, "metrics": metrics},
1021                     )
1022                 )
1023
1024             # GraphComplete hook
1025             if self._hook_runner.has_hooks():
1026                 await self._hook_runner.run_hooks(
1027                     "GraphComplete",
1028                     payload={
1029                         "step": self.step_counter,
1030                         "terminated": self.terminated,
1031                     },
1032                 )
1033
1034             # Final flush of any remaining log entries
1035             if self._logger:
1036                 self._logger.flush()
1037
1038         except Exception as e:
finalise — error path (verdict first, then error reporting) ── graph_executor.py:1039–1080 · ↔ §5.3
1040             # Best-effort after-loop stage first: a final-side evaluate can
1041             # still emit fresh metrics even when the main loop crashed, so
1042             # eval harvest reports the actual final env step count instead
1043             # of a stale pre-crash value. Tolerant of cascading failures
1044             # (the env may be torn down already).
1045             try:
1046                 self._emit_final_fallback()
1047                 await self._after_loop_pass(session)
1048             except Exception as _fpe:
1049                 log.warning("after_loop_pass on error path raised: %s", _fpe)
1050             get_bus().from_exception(
1051                 e,
1052                 source="graph",
1053                 code="NODE_ERRORS" if isinstance(e, NodeErrorAggregate) else "GRAPH_CRASH",
1054                 scope={
1055                     "step": self.step_counter,
1056                     "execution_id": getattr(session, "_execution_id", None),
1057                 },
1058                 title=(
1059                     f"Run finished with node errors at step {self.step_counter}"
1060                     if isinstance(e, NodeErrorAggregate)
1061                     else f"Graph execution crashed at step {self.step_counter}"
1062                 ),
1063             )
1064             session._status = "error"
1065             # run_end also fires on the error path so lifetime="run"
1066             # states reset even when the loop crashes.
1067             self.broadcast_signal(
1068                 "run_end",
1069                 {"step": self.step_counter, "error": str(e)},
1070             )
1071             if not _suppress:
1072                 await broadcast(session._ws("nav_status", {"status": "error", "error": str(e)}))
1073
1074             # GraphError hook — fires on unhandled exceptions
1075             if self._hook_runner.has_hooks():
1076                 await self._hook_runner.run_hooks("GraphError", payload={"error": str(e)})
1077
1078             # Flush log entries even on error
1079             if self._logger:
1080                 self._logger.flush()

7. Ways a run ends

8. Execution properties

9. File reference