Skip to content

Execution flow

This page walks one full call to Operon(graph).run(...) end-to-end.

Sequence — a 3-op linear graph

sequenceDiagram
    autonumber
    participant U as User
    participant G as GraphOp
    participant E as Operon
    participant S as Scheduler
    participant A as op_a
    participant B as op_b
    participant C as op_c
    participant T as Tracer

    rect rgb(237, 231, 246)
    Note over U,G: Construction (untimed)
    U->>G: with GraphOp(...) as g:
    U->>G: a = op_a(...); b = op_b(...); c = op_c(...)
    U->>G: START >> a >> b >> c >> END
    G->>G: __exit__: build() → resolve refs, freeze schema
    end

    rect rgb(224, 242, 241)
    Note over U,T: Init
    U->>E: engine = Operon(graph, tracer=t)
    E->>E: eager warmup (resource resolution)
    end

    rect rgb(255, 243, 224)
    Note over U,T: Run
    U->>E: await engine.run(inputs={"x": 5})
    E->>S: seed state, START → ready queue
    S->>A: dispatch (inputs from PARENT)
    A->>T: span_start(a)
    A-->>S: Frame(outputs)
    S->>S: write a-outputs to state
    A->>T: span_end(a)
    S->>B: dispatch (reads a["..."])
    B-->>S: Frame
    S->>C: dispatch (reads b["..."])
    C-->>S: Frame, EOF
    S->>S: collect + auto-forward via >> END
    E-->>U: result
    end

Three observations:

  • Construction is build-time, not run-time. Reference resolution happens once at __exit__; the engine just executes against a frozen schema.
  • Eager warmup at init. Resource lookups, #[op] registry checks, and any expensive schema validation are paid before engine.run is ever called. Run-time errors are caller-data errors.
  • The scheduler is the only thing that holds state across ops. Ops themselves are stateless functions — if you need shared state, write it to PARENT or pipe through siblings.

Phase 1 — Graph construction

with GraphOp(name="g") as graph: enters a build context. Op constructors register themselves with the active graph; >> edges are recorded as a list of (src, dst) pairs.

When the context exits, GraphOp.__exit__ calls build(), which:

  1. Resolves every PARENT["k"] and op["k"] reference against op schemas.
  2. Validates that every edge endpoint exists.
  3. Computes auto-forwarding for >> END (the last op's outputs become the graph's outputs).
  4. Freezes the schema — runtime no longer needs to inspect Python AST.

If any reference is unresolved, build raises before the engine ever sees the graph.

Phase 2 — Engine init

engine = Operon(graph)

The engine:

  1. Stores graph and optional tracer.
  2. Eager warmup — walks the graph, calls each op's warmup() hook. Provider ops resolve their resource="..." against the ResourceHub here. If the hub is missing or the key isn't registered, you get a fix-pointing error at Operon(graph), not at first run.

Operon.__init__ does not load .env or resources.yaml. Resource setup is the caller's responsibility — see Resource hub.

Phase 3 — Run

result = await engine.run(inputs={"x": 5})

The scheduler:

  1. Seeds the root state with inputs (resolves all PARENT["k"] refs).
  2. Adds START to the ready queue.
  3. Loops: pop a ready op, run it, write its outputs into state, propagate to downstream ops, mark them ready when all hard-edge predecessors have completed.
  4. When END is reached (all outgoing paths completed), returns the forwarded result.

Branch ops emit frames on soft edges only when their condition selects that branch. build() softens branch-merge edges automatically (auto_soft=True), because forgetting to do it by hand was a silent run-time deadlock rather than a build error. That derivation is tracked separately from an authored ~: the edge record carries auto_soft for what the pass decided and pinned_hard for what the author pinned, so a soft edge you wrote for trigger control — "fire on whichever predecessor lands first" — is never confused with one derived from a branch. Generator ops yield once per item; downstream ops run once per yield (streaming default — see Streaming).

Contexts

Every op invocation runs at a context — a tuple of path segments identifying which invocation it is:

Context Meaning
("main",) the root run
("main", "[2]") the branch handling a generator's third yield
("main", "[2]", "__collect__") a collect() consumer under that branch
("main", "g.__loop_0__#1") iteration 1 of a synthesized loop

Two properties the rest of the system rests on:

  • Containment is a tuple-prefix test. ("main", "[2]") is inside ("main",). This is why an empty prefix matches every context, and why cancellation targets have to be explicit (below).
  • Loop iterations are siblings, not nesting. Iteration 1 is ("main", "g.__loop_0__#1"), not a child of iteration 0. A stale iteration is therefore never "inside" the current one, which is what makes termination decidable.

Cancellation

An op cancels in-flight work by returning or yielding an Interrupt. The scheduler drops queued frames at the target context, cancels its in-flight tasks, and clears the bookkeeping.

from operonx.core import Interrupt

@op
def guard(score: float):
    if score < 0:
        return Interrupt(reason="negative score")   # cancels this branch
    return {"ok": score}

ctx_to_cancel defaults to Interrupt.SELF, which the scheduler resolves to the emitter's own context — the yield's context for a generator, not the whole op. Cancelling everything is destructive and has to be written out:

Interrupt(ctx_to_cancel=Interrupt.ALL, reason="fatal")

Three things worth knowing:

  • The emitter survives its own sweep. It finishes normally, so its EOF and cleanup run as usual.
  • A cancelled invocation produces no result. A subgraph whose root was swept yields nothing rather than the cells as they stand — otherwise the parent receives an all-None dict indistinguishable from a successful null answer.
  • The cancellation is reported. A ("__interrupt__", ctx, …) record reaches handle.interrupts even when the sweep happened inside a nested subgraph. It is a record, not an output: raw iteration of the handle yields it, but run(), collect() and result() leave it out.

Interrupt.SELF is a sentinel, not a tuple. If a code path ever fails to resolve it, the containment test raises TypeError rather than silently matching every context — see Failure modes §8.

Phase 4 — Tracing

If a tracer was passed to Operon(graph, tracer=...), every op start/end is recorded with timing, inputs, outputs, and the parent op span. Tracers are pluggable; see operonx.telemetry.

Failure points

Phase What can fail Surfaced as
Construction Bad PARENT/op ref BuildError at with exit
Construction A read of another op's output that no edge orders; two unordered writers of a reducer-less declared cell GraphValidationError at build
Engine init Missing resource Branch-(1)…(5) error from ResourceHub.get
Run Op raises — any Exception, OpError subclasses such as ParserError included Not raised (the default, errors="record"). Reported in the result as "$errors" and on the handle as handle.errors — see below. With Operon(g, errors="raise"): OpFailed, the run cancelled
Run Circuit breaker — ObserveBudgetExceeded (a BaseException) Raised to the caller
Run Misdirected Interrupt — InterruptTargetError Raised to the caller
Run The framework failing around an op, not the op body Raised to the caller

"Raised to the caller" means the same exception from run(), collect(), result(), async iteration of the handle, and every stream() mode — "updates", "values" and "custom" after the chunks that landed before it.

An op that raises

The run does not raise. One failing op must not end a run that is serving someone — a live call keeps going when one turn's op fails — so the failure is reported instead of propagated. An op can first try again (@op(retry=Retry(...))) or hand its failure to a handler op (op.on_error(handler)); a job, a test or a batch script that wants the failure itself builds its engine with errors="raise" (below).

The op's own BaseOp.run catches the exception; it is the only handler that does. It logs it with the traceback, writes the text to the op's error cell, records a failed trace node, and records it on the run. No frame is emitted, so the op's outputs are missing and every op waiting on them never runs; the rest of the graph carries on.

What the caller sees:

out = await engine.run(inputs={"x": "not a number"})
out["$errors"]   # {"engine.parse": {"type": "ValueError", "count": 1, "first_ctx": "main",
                 #                   "message": "Traceback ...\nValueError: invalid literal ..."}}
"n" in out       # False: the failed op's outputs are simply missing
  • "$errors" is {op_name_in_state: record}: the key is "<graph>.<op>" (a nested op by its full path), the record {type, message, count, first_ctx}. message is the first failure's traceback trimmed to the user's frames — operonx's own BaseOp.run / _exec_core frames are the same for every failure and are dropped; an exception raised by operonx itself is its last line alone. count is how many times the op failed: an op that fails on several stream items is one entry with count > 1. first_ctx is where the first failure ran, so the failed trace node is f"{op}#{first_ctx}". The full traceback, operonx frames included, stays in the trace node and the per-context error cell.
  • A structured LLMOp (fields=) that fails returns error instead of raising, so its node is ok; it records itself in "$errors" all the same (type "ParserError", or the exception's class under on_failure="error"). Otherwise only the next step's PromptError on the None fields would show.
  • The trace carries the same records (trace.errors), and trace.status is "error" when a node failed or a record exists; the local consumer writes both into meta.json (status, errors).
  • An op raising inside a subgraph stops the ops after the subgraph, as it would flat: a subgraph run in which an op under it raised and none of its outputs were written yields nothing. "$errors" then has an entry for the subgraph too, "<graph>.<sub>", naming the op that raised (type "SubgraphError", message "SubgraphError: 'g.sub.parse' raised, …"). A subgraph that wrote some of its outputs still yields them.
  • It is present only when an op failed, so a clean run's keys are unchanged. collect() and result() carry it the same way; handle.errors is the same dict ({} when none), readable while the run is still going.
  • Tracebacks stay server-side. The serve layer answers a failed HTTP run with 500 {"error": "the graph produced no output", "endpoint": …, "trace_id": …} (the id is also the x-operonx-trace-id header) and never sends "$errors"; a job marks the item failed and writes nothing to its sink.

errors="raise"

Operon(g, errors="raise") ends the run at the first op failure that no error edge handles. The failure is recorded first, exactly as above, so handle.errors and the trace still show it. Then BaseOp.run raises an internal BaseException carrier: it passes the except Exception of every enclosing subgraph — which would otherwise record it as the subgraph's own failure — and reaches each scheduler's fatal path, whose finally cancels the ops still running. The engine hands the caller OpFailed(op, error), with the op's exception as __cause__, from run(), result(), collect(), iteration and every stream() mode. A loop that reaches its cap raises it too.