operonx.core

Engine, op decorators, graph composition, state markers, and middleware. This page is the primary public surface — everything you need to build and run a workflow without touching providers or telemetry.

Engine

Operon

Operon(
    graph: Union[GraphOp, Callable[..., GraphOp]],
    *,
    params: Optional[Dict[str, Any]] = None,
    trace: Optional[Union[str, Consumer, List[Union[str, Consumer]]]] = None,
    errors: str = "record",
    max_concurrency: Optional[int] = None,
    journal: Any = None,
    durability: str = "async",
    on_resume: str = "restart",
    carry: Sequence[str] = (),
)

Workflow execution engine.

Operon takes a GraphOp and provides execution capabilities: - Builds and validates the graph structure - Creates state schema for data flow - Executes workflows with fresh state per run - Integrates with tracers for observability

Attributes:

Name	Type	Description
graph		The GraphOp to execute
name		Workflow name (from graph)
schema	StateSchema	State schema for the workflow

Example

# Define graph
with GraphOp(name="chatbot") as graph:
    llm = LLMOp(name="llm", resource="gpt-4o", inputs={"prompt": ...})
    START >> llm >> END

# Create engine (builds automatically)
engine = Operon(graph)

# Run multiple times with fresh state
result = await engine.run(inputs={"query": "Hello!"})
print(result["response"])      # workflow output
print(result["$state"])        # MemoryState for debugging

# Or use callable syntax
result = await engine({"query": "Goodbye!"})

Initialize Operon engine with a GraphOp or a graph factory.

Pure orchestrator — does not load .env or resources.yaml. Call :func:operonx.bootstrap (or :meth:ResourceHub.from_yaml directly) before constructing the engine if your graph uses provider ops. Pure-compute graphs need no setup.

Parameters:

Name	Type	Description	Default
graph	Union[GraphOp, Callable[..., GraphOp]]	A GraphOp workflow, or a callable that returns one. When a callable is passed, it is invoked with **params immediately — call :func:operonx.bootstrap first if the factory needs the hub.	required
params	Optional[Dict[str, Any]]	Keyword arguments passed to the graph factory. Ignored when graph is already a GraphOp. Defaults to {}.	None
trace	Optional[Union[str, Consumer, List[Union[str, Consumer]]]]	V3 tracing. Accepts a ResourceHub key (str), a :class:Consumer instance, or a list of either. Each consumer gets handle.trace at the end of every run and writes its own view (disk, Langfuse, report, …); a live one (the ClickHouse and SQL stores) also gets the run as it starts and each execution as it lands, and lists the run as running meanwhile. Failures are caught + logged per-consumer so one bad backend never affects the call. Requires :func:operonx.bootstrap when using string keys. "local" is the built-in local consumer: inside a project (an operonx.toml at or above the working directory) it writes to <project>/.operonx/runs. "project" is the project's own [tracing] sinks, chosen as its services' and jobs' are. Unset, nothing is traced. A run started inside an op of a running engine is part of that run and calls no consumer. Examples:: trace="trace_local:default" trace=CallbotLocalConsumer(config={"root": "/tmp/x"}) trace=["trace_langfuse:edupia", MyDebugConsumer()]	None
errors	str	What a run does when an op fails. "record" (the default) records it — "$errors", handle.errors — and carries on, so one failing op does not end a live session. "raise" ends the run at the first failure no error edge handles: the ops still running are cancelled and run(), result(), collect(), iteration and stream() raise :class:OpFailed. For jobs, tests and batch scripts, which want the failure rather than a result with a key missing.	'record'
max_concurrency	Optional[int]	The most ops of one run that may be running at once, counted across every nested graph. A graph's own concurrency= caps that graph only, so nested graphs multiply (2 subgraphs at concurrency=2 each run 4 ops); this is the one cap for the run. Counts the ops that run as tasks (async, bound="cpu"); a plain def op runs inline and a subgraph holds no slot of its own. None (the default): no shared cap.	None
journal	Any	Makes runs durable: a :class:~operonx.durable.Journal (MemoryJournal, SqliteJournal) records each execution's yields and end with the cell writes they made, and :meth:resume continues a run that stopped — after a crash, on any worker opening the same journal — without running again what had ended. None (the default): runs are not recorded, at the cost of one is None test per execution. See docs/RUNTIME_R3_PLAN.md.	None
durability	str	With a journal, when steps are written: "async" (the default; a background writer, in order — a crash loses at most the unwritten tail, which runs again), "sync" (each step committed before the scheduler sees its event) or "exit" (all at the end of the run).	'async'
carry	Sequence[str]	Declared cells (PARENT.declare) a thread keeps between its runs: a run started with thread_id=T begins with them as T's last run left them, and saves them when it ends. Needs a journal=, where threads are kept.	()
on_resume	str	An execution that had yielded and not ended when the run stopped: "restart" runs it again and checks its first yields against the journal; "fail" refuses.	'restart'

Raises:

Type	Description
RuntimeError	If a provider op needs the hub but none has been installed. The message points at operonx.bootstrap().
TypeError	If a trace= item is neither a str nor a Consumer instance.

Source code in operonx/core/engine.py

def __init__(
    self,
    graph: Union[GraphOp, Callable[..., GraphOp]],
    *,
    params: Optional[Dict[str, Any]] = None,
    trace: Optional[Union[str, "Consumer", List[Union[str, "Consumer"]]]] = None,
    errors: str = "record",
    max_concurrency: Optional[int] = None,
    journal: Any = None,
    durability: str = "async",
    on_resume: str = "restart",
    carry: Sequence[str] = (),
):
    """Initialize Operon engine with a GraphOp or a graph factory.

    Pure orchestrator — does **not** load ``.env`` or ``resources.yaml``.
    Call :func:`operonx.bootstrap` (or :meth:`ResourceHub.from_yaml` directly)
    before constructing the engine if your graph uses provider ops.
    Pure-compute graphs need no setup.

    Args:
        graph: A GraphOp workflow, or a callable that returns one.
               When a callable is passed, it is invoked with ``**params``
               immediately — call :func:`operonx.bootstrap` first if the
               factory needs the hub.
        params: Keyword arguments passed to the graph factory. Ignored
                when *graph* is already a GraphOp. Defaults to ``{}``.
        trace: V3 tracing. Accepts a ResourceHub key (str),
               a :class:`Consumer` instance, or a list of either.
               Each consumer gets ``handle.trace`` at the end of
               every run and writes its own view (disk, Langfuse,
               report, …); a live one (the ClickHouse and SQL stores)
               also gets the run as it starts and each execution as it
               lands, and lists the run as ``running`` meanwhile.
               Failures are caught + logged per-consumer so one bad
               backend never affects the call. Requires
               :func:`operonx.bootstrap` when using string keys.
               ``"local"`` is the built-in local consumer: inside a
               project (an ``operonx.toml`` at or above the working
               directory) it writes to ``<project>/.operonx/runs``.
               ``"project"`` is the project's own ``[tracing]`` sinks,
               chosen as its services' and jobs' are. Unset, nothing
               is traced. A run started inside an op of a running
               engine is part of that run and calls no consumer.
               Examples::

                   trace="trace_local:default"
                   trace=CallbotLocalConsumer(config={"root": "/tmp/x"})
                   trace=["trace_langfuse:edupia", MyDebugConsumer()]

        errors: What a run does when an op fails. ``"record"`` (the
               default) records it — ``"$errors"``, ``handle.errors`` —
               and carries on, so one failing op does not end a live
               session. ``"raise"`` ends the run at the first failure no
               error edge handles: the ops still running are cancelled
               and ``run()``, ``result()``, ``collect()``, iteration and
               ``stream()`` raise :class:`OpFailed`. For jobs, tests and
               batch scripts, which want the failure rather than a
               result with a key missing.
        max_concurrency: The most ops of one run that may be running at
               once, counted across every nested graph. A graph's own
               ``concurrency=`` caps that graph only, so nested graphs
               multiply (2 subgraphs at ``concurrency=2`` each run 4
               ops); this is the one cap for the run. Counts the ops
               that run as tasks (async, ``bound="cpu"``); a plain
               ``def`` op runs inline and a subgraph holds no slot of
               its own. ``None`` (the default): no shared cap.
        journal: Makes runs durable: a :class:`~operonx.durable.Journal`
               (``MemoryJournal``, ``SqliteJournal``) records each
               execution's yields and end with the cell writes they made,
               and :meth:`resume` continues a run that stopped — after a
               crash, on any worker opening the same journal — without
               running again what had ended. ``None`` (the default): runs
               are not recorded, at the cost of one ``is None`` test per
               execution. See docs/RUNTIME_R3_PLAN.md.
        durability: With a journal, when steps are written: ``"async"``
               (the default; a background writer, in order — a crash
               loses at most the unwritten tail, which runs again),
               ``"sync"`` (each step committed before the scheduler sees
               its event) or ``"exit"`` (all at the end of the run).
        carry: Declared cells (``PARENT.declare``) a thread keeps between
               its runs: a run started with ``thread_id=T`` begins with
               them as T's last run left them, and saves them when it
               ends. Needs a ``journal=``, where threads are kept.
        on_resume: An execution that had yielded and not ended when the
               run stopped: ``"restart"`` runs it again and checks its
               first yields against the journal; ``"fail"`` refuses.

    Raises:
        RuntimeError: If a provider op needs the hub but none has been
            installed. The message points at ``operonx.bootstrap()``.
        TypeError: If a ``trace=`` item is neither a str nor a
            Consumer instance.
    """
    if errors not in ERRORS_MODES:
        raise ValueError(f"errors= must be 'record' or 'raise', got {errors!r}")
    if max_concurrency is not None and (
        isinstance(max_concurrency, bool)
        or not isinstance(max_concurrency, int)
        or max_concurrency < 1
    ):
        raise ValueError(
            f"max_concurrency= takes an int >= 1 (ops running at once), got {max_concurrency!r}"
        )
    if callable(graph) and not isinstance(graph, GraphOp):
        graph = graph(**(params or {}))

    self._errors = errors
    self._max_concurrency = max_concurrency
    if journal is not None:
        from operonx.durable import DURABILITY, ON_RESUME, Journal

        if not isinstance(journal, Journal):
            raise TypeError(
                f"journal= takes a Journal (MemoryJournal, SqliteJournal), not {journal!r}"
            )
        if durability not in DURABILITY:
            raise ValueError(
                f"durability= is one of {', '.join(DURABILITY)}, got {durability!r}"
            )
        if on_resume not in ON_RESUME:
            raise ValueError(f"on_resume= is one of {', '.join(ON_RESUME)}, got {on_resume!r}")
    self._journal = journal
    self._durability = durability
    self._on_resume = on_resume
    self._fingerprint: Optional[str] = None
    self.graph = graph
    self.name = graph.name
    self._carry = tuple(carry)
    if self._carry:
        if journal is None:
            raise ValueError(
                "carry= keeps cells between a thread's runs in the journal: pass journal="
            )
        declared = set(getattr(graph, "_shared_vars", None) or {})
        unknown = [name for name in self._carry if name not in declared]
        if unknown:
            raise ValueError(
                f"carry={list(self._carry)}: {unknown} not declared by graph {graph.name!r} "
                f"(PARENT.declare(...)); it declares {sorted(declared) or 'none'}"
            )
    self._trace_consumers = self._resolve_trace_consumers(trace)
    #: The runtime inputs a caller must pass, when whoever compiled the
    #: engine knows them (a served graph: its unbound parameters). The
    #: serve layer checks a door's RunRequest against it; None = unchecked.
    self.inputs_expected = None
    # set by the serve layer (`compile_graph`): the @graph's own defaults
    self.inputs_defaults: Dict[str, Any] = {}
    # whether the graph has an ingress door (`operonx.app.doors.has_doors`)
    self._doors: Optional[bool] = None

    # Build graph and create schema immediately
    self.graph.build()
    self._schema = StateSchema(self.graph)

    # Eagerly init backends if a hub is already configured
    self._warmup_ops()

    LOGGER.debug(
        "Operon engine initialized for workflow [highlight]%s[/highlight]",
        self.name,
    )

Attributes

trace_consumers property

trace_consumers: List[Any]

The consumers every run of this engine feeds, resolved from trace= — for a caller that runs another graph beside this one and wants it traced to the same places (an eval's judges).

schema property

schema: StateSchema

Access the workflow state schema.

fingerprint property

fingerprint: str

The compiled graph's structural fingerprint (what :meth:resume checks): its ops, their code and their wiring.

Methods:

start

start(
    inputs: Dict[str, Any],
    *,
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
    request_id: Optional[str] = None,
    trace_id: Optional[str] = None,
    scratch: Optional[Dict[str, Any]] = None,
    checkpointer=None,
    context: Any = None,
    run_id: Optional[str] = None,
    thread_id: Optional[str] = None,
) -> ExecutionHandle

Start workflow execution and return a streaming handle immediately.

Does not block — the graph runs in the background. Use the handle to stream frames, await specific outputs, or collect the final result.

V3 trace consumers (declared via Operon(..., trace=...)) fire automatically once the scheduler completes — no explicit finalize needed.

Parameters:

Name	Type	Description	Default
inputs	Dict[str, Any]	Input data for the workflow	required
user_id	Optional[str]	Optional user identifier (auto-generated if not provided)	None
session_id	Optional[str]	Optional session identifier (auto-generated if not provided)	None
request_id	Optional[str]	Optional request identifier (auto-generated if not provided)	None
scratch	Optional[Dict[str, Any]]	Optional initial values for per-call scratch space. Applied synchronously before the scheduler task is created — race-free. Equivalent to writing handle.scratch[k] = v before the first await after start(), but guaranteed to be visible to entry ops.	None
context	Any	Any object the run's ops may read as run_context().context — a tenant, a feature flag set, a typed dataclass of what the caller knows. Information only: stored as given, never read by operonx.	None
run_id	Optional[str]	The run's id — its trace id, and with a journal= the key :meth:resume continues it by. Same as trace_id.	None
thread_id	Optional[str]	The thread the run belongs to (session_id names the same thing). With carry=, the run starts with the carried cells as the thread's last run left them.	None

The run's run_context().run_id is its trace id (trace_id, else request_id), and thread_id is session_id as given (None when not given).

Returns:

Type	Description
ExecutionHandle	ExecutionHandle — async-iterable, supports await handle["op","var"]
ExecutionHandle	and await handle.collect()

Source code in operonx/core/engine.py

def start(
    self,
    inputs: Dict[str, Any],
    *,
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
    request_id: Optional[str] = None,
    trace_id: Optional[str] = None,
    scratch: Optional[Dict[str, Any]] = None,
    checkpointer=None,
    context: Any = None,
    run_id: Optional[str] = None,
    thread_id: Optional[str] = None,
) -> "ExecutionHandle":
    """Start workflow execution and return a streaming handle immediately.

    Does not block — the graph runs in the background. Use the handle to
    stream frames, await specific outputs, or collect the final result.

    V3 trace consumers (declared via ``Operon(..., trace=...)``) fire
    automatically once the scheduler completes — no explicit finalize
    needed.

    Args:
        inputs: Input data for the workflow
        user_id: Optional user identifier (auto-generated if not provided)
        session_id: Optional session identifier (auto-generated if not provided)
        request_id: Optional request identifier (auto-generated if not provided)
        scratch: Optional initial values for per-call scratch space. Applied
            synchronously before the scheduler task is created — race-free.
            Equivalent to writing ``handle.scratch[k] = v`` before the first
            ``await`` after ``start()``, but guaranteed to be visible to
            entry ops.
        context: Any object the run's ops may read as
            ``run_context().context`` — a tenant, a feature flag set, a
            typed dataclass of what the caller knows. Information only:
            stored as given, never read by operonx.
        run_id: The run's id — its trace id, and with a ``journal=`` the
            key :meth:`resume` continues it by. Same as ``trace_id``.
        thread_id: The thread the run belongs to (``session_id`` names
            the same thing). With ``carry=``, the run starts with the
            carried cells as the thread's last run left them.

    The run's ``run_context().run_id`` is its trace id (``trace_id``,
    else ``request_id``), and ``thread_id`` is ``session_id`` as given
    (``None`` when not given).

    Returns:
        ExecutionHandle — async-iterable, supports ``await handle["op","var"]``
        and ``await handle.collect()``
    """
    if run_id is not None and trace_id is not None and run_id != trace_id:
        raise ValueError(f"run_id={run_id!r} and trace_id={trace_id!r} name one id; give one")
    trace_id = trace_id or run_id
    if thread_id is not None and session_id is not None and thread_id != session_id:
        raise ValueError(
            f"thread_id={thread_id!r} and session_id={session_id!r} name one thread; give one"
        )
    thread_id = thread_id or session_id
    session_id = session_id or thread_id
    user_id = user_id or str(uuid.uuid4())
    session_id = session_id or str(uuid.uuid4())
    request_id = request_id or str(uuid.uuid4())

    state = self._schema.create_state(
        inputs=inputs,
        user_id=user_id,
        session_id=session_id,
        request_id=request_id,
    )
    carried: Dict[str, Any] = {}
    if self._carry and thread_id is not None:
        saved = self._journal.load_thread(thread_id)
        carried = {name: saved[name] for name in self._carry if name in saved}
        self._seed(state, carried)
    if self._journal is not None:
        from operonx.durable import RunHeader

        recorder = self._recorder(trace_id or request_id, state)
        self._journal.open_run(
            RunHeader(
                run_id=recorder.run_id,
                fingerprint=self.fingerprint,
                inputs=dict(inputs),
                thread_id=thread_id,
                carried=carried,
            )
        )
    return self._launch(
        state,
        request_id=request_id,
        user_id=user_id,
        session_id=session_id,
        trace_id=trace_id,
        thread_id=thread_id,
        scratch=scratch,
        checkpointer=checkpointer,
        context=context,
        inputs=inputs,
    )

resume async

resume(
    run_id: str,
    *,
    answers: Optional[Dict[str, Any]] = None,
    allow_graph_change: bool = False,
    checkpointer=None,
    context: Any = None,
) -> ExecutionHandle

Continue the journalled run run_id and return its handle::

handle = await engine.resume("order-42")
out = await handle.result()

Every cell is put back as the journal last had it; an execution that ended is not run again — its yields are replayed, so what came after it is dispatched as before — and the rest runs. The result is the one the run would have had. Any process with the same graph and a journal holding the run can resume it.

A run parked on interrupts (its result's "$interrupted") takes their responses as answers={interrupt_id: value}; one left unanswered parks the run again. A drained run resumes as it is.

Raises:

Type	Description
RuntimeError	The engine has no journal=.
JournalError	The journal has no such run, the graph changed since the run started (allow_graph_change=True resumes anyway, at the caller's risk), an answer names no question the run asked, or the graph reads through a door.

Source code in operonx/core/engine.py

async def resume(
    self,
    run_id: str,
    *,
    answers: Optional[Dict[str, Any]] = None,
    allow_graph_change: bool = False,
    checkpointer=None,
    context: Any = None,
) -> "ExecutionHandle":
    """Continue the journalled run *run_id* and return its handle::

        handle = await engine.resume("order-42")
        out = await handle.result()

    Every cell is put back as the journal last had it; an execution that
    ended is not run again — its yields are replayed, so what came after
    it is dispatched as before — and the rest runs. The result is the one
    the run would have had. Any process with the same graph and a
    journal holding the run can resume it.

    A run parked on interrupts (its result's ``"$interrupted"``) takes
    their responses as ``answers={interrupt_id: value}``; one left
    unanswered parks the run again. A drained run resumes as it is.

    Raises:
        RuntimeError: The engine has no ``journal=``.
        JournalError: The journal has no such run, the graph changed
            since the run started (``allow_graph_change=True`` resumes
            anyway, at the caller's risk), an answer names no question
            the run asked, or the graph reads through a door.
    """
    from operonx.durable import JournalError

    if self._journal is None:
        raise RuntimeError(
            f"resume({run_id!r}): this engine has no journal=; build it with the journal "
            "the run was recorded in"
        )
    door = _ingress_of(self.graph)
    if door is not None:
        raise JournalError(
            f"resume({run_id!r}): {door} is a door (ingress) — its items come from a live "
            "connection, which the journal does not hold. A served graph's runs are "
            "journalled for audit, not resumed"
        )
    header, steps = await asyncio.to_thread(self._journal.read, run_id)
    if header.fingerprint != self.fingerprint and not allow_graph_change:
        raise JournalError(
            f"run {run_id!r} ran graph {header.fingerprint} and this engine's graph is "
            f"{self.fingerprint}: an op, its code or its wiring changed, so the journal's "
            "steps may name other work. Resume with the graph it ran, or pass "
            "allow_graph_change=True"
        )
    request_id = str(uuid.uuid4())
    session_id = header.thread_id or str(uuid.uuid4())
    user_id = str(uuid.uuid4())
    state = self._schema.create_state(
        inputs=header.inputs,
        user_id=user_id,
        session_id=session_id,
        request_id=request_id,
    )
    self._seed(state, header.carried)  # what the thread gave it, then its own writes
    recorder = self._recorder(run_id, state)
    recorder.restore(state, steps, lenient=allow_graph_change)
    asked = set(recorder.questions.values())
    for interrupt_id in answers or {}:
        if interrupt_id not in asked:
            raise JournalError(
                f"resume({run_id!r}): the run has no question {interrupt_id!r}; it asked "
                f"{sorted(asked) or 'none'}"
            )
    recorder.answers = dict(answers or {})
    await asyncio.to_thread(self._journal.set_status, run_id, "running")
    return self._launch(
        state,
        request_id=request_id,
        user_id=user_id,
        session_id=session_id,
        trace_id=run_id,
        thread_id=header.thread_id,
        scratch=None,
        checkpointer=checkpointer,
        context=context,
    )

runs

runs(status: Optional[str] = None) -> List[Any]

The journalled runs (their headers), optionally only status ("running": stopped or still going, the ones to resume).

Source code in operonx/core/engine.py

def runs(self, status: Optional[str] = None) -> List[Any]:
    """The journalled runs (their headers), optionally only *status*
    (``"running"``: stopped or still going, the ones to resume)."""
    if self._journal is None:
        raise RuntimeError("runs(): this engine has no journal=")
    return self._journal.runs(status)

run async

run(
    inputs: Dict[str, Any],
    *,
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
    request_id: Optional[str] = None,
    trace_id: Optional[str] = None,
    scratch: Optional[Dict[str, Any]] = None,
    checkpointer=None,
    context: Any = None,
    run_id: Optional[str] = None,
    thread_id: Optional[str] = None,
) -> Dict[str, Any]

Execute the workflow with given inputs.

Each call creates a fresh state, so the same engine can be used for multiple independent executions. Equivalent to::

handle = engine.start(inputs, ...)
result = await handle.collect(unwrap=True)

V3 trace consumers (declared via Operon(..., trace=...)) fire automatically inside start() when the scheduler completes.

Parameters:

Name	Type	Description	Default
inputs	Dict[str, Any]	Input data for the workflow	required
user_id	Optional[str]	Optional user identifier (auto-generated if not provided)	None
session_id	Optional[str]	Optional session identifier (auto-generated if not provided)	None
request_id	Optional[str]	Optional request identifier (auto-generated if not provided)	None
scratch	Optional[Dict[str, Any]]	Optional initial values for per-call scratch space.	None
context	Any	What the run's ops read as run_context().context (see :meth:start).	None
run_id	Optional[str]	The run's id (see :meth:start).	None
thread_id	Optional[str]	The run's thread (see :meth:start).	None

Returns:

Name	Type	Description
	Dict[str, Any]	Dictionary containing workflow outputs plus "$state" key
	Dict[str, Any]	with the MemoryState for debugging/tracing access, plus
	Dict[str, Any]	"$errors" — {op_full_name: error_text} — when at least one
	Dict[str, Any]	op raised. An op that raises does not raise out of the run;
	Dict[str, Any]	its outputs, and those of every op after it, are missing —
	Dict[str, Any]	unless the engine was built with errors="raise", which
raises	Dict[str, Any]	class:OpFailed instead. A BaseException
	Dict[str, Any]	(ObserveBudgetExceeded, a misdirected Interrupt) does
	Dict[str, Any]	raise.

Source code in operonx/core/engine.py

async def run(
    self,
    inputs: Dict[str, Any],
    *,
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
    request_id: Optional[str] = None,
    trace_id: Optional[str] = None,
    scratch: Optional[Dict[str, Any]] = None,
    checkpointer=None,
    context: Any = None,
    run_id: Optional[str] = None,
    thread_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Execute the workflow with given inputs.

    Each call creates a fresh state, so the same engine can be
    used for multiple independent executions.  Equivalent to::

        handle = engine.start(inputs, ...)
        result = await handle.collect(unwrap=True)

    V3 trace consumers (declared via ``Operon(..., trace=...)``)
    fire automatically inside ``start()`` when the scheduler
    completes.

    Args:
        inputs: Input data for the workflow
        user_id: Optional user identifier (auto-generated if not provided)
        session_id: Optional session identifier (auto-generated if not provided)
        request_id: Optional request identifier (auto-generated if not provided)
        scratch: Optional initial values for per-call scratch space.
        context: What the run's ops read as ``run_context().context``
            (see :meth:`start`).
        run_id: The run's id (see :meth:`start`).
        thread_id: The run's thread (see :meth:`start`).

    Returns:
        Dictionary containing workflow outputs plus "$state" key
        with the MemoryState for debugging/tracing access, plus
        "$errors" — ``{op_full_name: error_text}`` — when at least one
        op raised. An op that raises does not raise out of the run;
        its outputs, and those of every op after it, are missing —
        unless the engine was built with ``errors="raise"``, which
        raises :class:`OpFailed` instead. A ``BaseException``
        (``ObserveBudgetExceeded``, a misdirected ``Interrupt``) does
        raise.
    """
    handle = self.start(
        inputs,
        user_id=user_id,
        session_id=session_id,
        request_id=request_id,
        trace_id=trace_id,
        scratch=scratch,
        checkpointer=checkpointer,
        context=context,
        run_id=run_id,
        thread_id=thread_id,
    )

    try:
        result = await handle.collect(unwrap=True)
    except BaseException:
        # `asyncio.wait_for(engine.run(...), t)` cancels this await,
        # not the run behind it: the graph kept going after the caller
        # had been told it timed out, and the op after the timeout
        # still ran. The run belongs to this call, so it ends with it.
        handle.cancel()
        raise
    result["$state"] = handle.state

    return result

invoke async

invoke(inputs: Dict[str, Any], **kwargs) -> Dict[str, Any]

LangGraph-familiar alias for :meth:run. Same signature.

Source code in operonx/core/engine.py

async def invoke(self, inputs: Dict[str, Any], **kwargs) -> Dict[str, Any]:
    """LangGraph-familiar alias for :meth:`run`. Same signature."""
    return await self.run(inputs, **kwargs)

stream async

stream(
    inputs: Dict[str, Any],
    *,
    mode: Union[str, List[str], tuple] = "updates",
    channels: Optional[List[str]] = None,
    checkpointer: Optional[Any] = None,
    **kwargs: Any,
) -> asyncio.AsyncGenerator[Any, None]

LangGraph-familiar streaming iterator.

Parameters:

Name	Type	Description	Default
inputs	Dict[str, Any]	workflow inputs (same as run/invoke)	required
mode	Union[str, List[str], tuple]	one mode, or a list of them. A string yields that mode's chunks; a list yields (mode, chunk) pairs from one run, in the order they arrive. The modes: "updates" — yields {op_name: {var: value, ...}} per op completion (matches LangGraph's stream_mode="updates"). Covers every op, including generators in the middle of the graph, and delivers each write as it lands. When an :class:~operonx.InterruptOp suspends, also yields its :class:~operonx.checkpoint.InterruptEvent, in order with the updates, unless "interrupts" is also streamed; answer with event.resume(value). "interrupts" — yields only those InterruptEvent objects. "values" — yields the full state snapshot per step; requires a checkpointer (auto-created in-memory if omitted) "frames" — yields (op, ctx, data) for ops that write a graph output (PARENT- or END-bound). An op feeding only a downstream consumer emits nothing here, whatever it yields — use "updates" to watch those. "custom" — yields :class:~operonx.checkpoint.CustomEvent emitted by any :class:~operonx.EmitOp, optionally filtered by channels=[...] "tasks" — yields :class:~operonx.TaskStarted, :class:~operonx.TaskFinished and :class:~operonx.TaskFailed: one start and one end per op invocation (a generator's end comes after its last item) and per child() execution, each with its attempt.	'updates'
channels	Optional[List[str]]	for mode="custom" only — restrict to these channel names	None
checkpointer	Optional[Any]	for mode="values"; auto-created InMemory if None	None
**kwargs	Any	forwarded to start() (user_id, session_id, etc.)	{}

Yields:

Type	Description
AsyncGenerator[Any, None]	mode-specific chunks, or (mode, chunk) for a list of modes.

Raises:

Type	Description
ValueError	an unknown mode, one given twice, or an empty list — before anything runs.
BaseException	whatever run() raises — a fatal error such as ObserveBudgetExceeded — in every mode, after the chunks that landed before it. An op that raises is not fatal: the stream ends normally, as run() returns normally.

Source code in operonx/core/engine.py

async def stream(
    self,
    inputs: Dict[str, Any],
    *,
    mode: Union[str, List[str], tuple] = "updates",
    channels: Optional[List[str]] = None,
    checkpointer: Optional[Any] = None,
    **kwargs: Any,
) -> "asyncio.AsyncGenerator[Any, None]":
    """LangGraph-familiar streaming iterator.

    Args:
        inputs: workflow inputs (same as ``run``/``invoke``)
        mode: one mode, or a list of them. A string yields that mode's
            chunks; a list yields ``(mode, chunk)`` pairs from one run,
            in the order they arrive. The modes:

            - ``"updates"`` — yields ``{op_name: {var: value, ...}}`` per op
              completion (matches LangGraph's ``stream_mode="updates"``).
              Covers **every** op, including generators in the middle of
              the graph, and delivers each write as it lands. When an
              :class:`~operonx.InterruptOp` suspends, also yields its
              :class:`~operonx.checkpoint.InterruptEvent`, in order with
              the updates, unless ``"interrupts"`` is also streamed;
              answer with ``event.resume(value)``.
            - ``"interrupts"`` — yields only those ``InterruptEvent`` objects.
            - ``"values"`` — yields the full state snapshot per step;
              requires a checkpointer (auto-created in-memory if omitted)
            - ``"frames"`` — yields ``(op, ctx, data)`` for ops that
              write a graph **output** (PARENT- or END-bound). An op
              feeding only a downstream consumer emits nothing here,
              whatever it yields — use ``"updates"`` to watch those.
            - ``"custom"`` — yields :class:`~operonx.checkpoint.CustomEvent`
              emitted by any :class:`~operonx.EmitOp`, optionally filtered
              by ``channels=[...]``
            - ``"tasks"`` — yields :class:`~operonx.TaskStarted`,
              :class:`~operonx.TaskFinished` and :class:`~operonx.TaskFailed`:
              one start and one end per op invocation (a generator's end
              comes after its last item) and per ``child()`` execution,
              each with its attempt.
        channels: for ``mode="custom"`` only — restrict to these channel names
        checkpointer: for ``mode="values"``; auto-created InMemory if None
        **kwargs: forwarded to ``start()`` (user_id, session_id, etc.)

    Yields:
        mode-specific chunks, or ``(mode, chunk)`` for a list of modes.

    Raises:
        ValueError: an unknown mode, one given twice, or an empty list —
            before anything runs.
        BaseException: whatever ``run()`` raises — a fatal error such as
            ``ObserveBudgetExceeded`` — in every mode, after the chunks
            that landed before it. An op that raises is not fatal: the
            stream ends normally, as ``run()`` returns normally.
    """
    paired = not isinstance(mode, str)
    modes = tuple(mode) if paired else (mode,)
    unknown = [m for m in modes if m not in STREAM_MODES]
    if unknown or not modes or len(set(modes)) != len(modes):
        if not modes:
            problem = "takes at least one mode"
        elif unknown:
            problem = f"has no mode {unknown[0]!r}"
        else:
            problem = "names a mode more than once"
        raise ValueError(
            f"engine.stream(mode={mode!r}) {problem} — valid modes: "
            + ", ".join(repr(m) for m in STREAM_MODES)
            + "; pass a list for several at once"
        )
    async for name, chunk in self._stream_modes(inputs, modes, channels, checkpointer, kwargs):
        yield (name, chunk) if paired else chunk

serve

serve(
    *,
    path: str = "/",
    host: str = "0.0.0.0",
    port: int = 8000,
    websocket: bool = False,
    **kwargs: Any,
) -> None

Serve this workflow over HTTP or a WebSocket.

The one-endpoint convenience form. It builds the same ServeSpec an operonx.toml would and serves that, so a project outgrowing it moves to the manifest without changing how anything runs.

Parameters:

Name	Type	Description	Default
path	str	URL path for the endpoint (default: "/").	'/'
host	str	Bind address.	'0.0.0.0'
port	int	Bind port.	8000
websocket	bool	Serve a WebSocket endpoint instead of HTTP.	False
**kwargs	Any	Extra arguments forwarded to uvicorn.run().	{}

Source code in operonx/core/engine.py

def serve(
    self,
    *,
    path: str = "/",
    host: str = "0.0.0.0",
    port: int = 8000,
    websocket: bool = False,
    **kwargs: Any,
) -> None:
    """Serve this workflow over HTTP or a WebSocket.

    The one-endpoint convenience form. It builds the same ``ServeSpec``
    an ``operonx.toml`` would and serves that, so a project outgrowing
    it moves to the manifest without changing how anything runs.

    Args:
        path: URL path for the endpoint (default: "/").
        host: Bind address.
        port: Bind port.
        websocket: Serve a WebSocket endpoint instead of HTTP.
        **kwargs: Extra arguments forwarded to ``uvicorn.run()``.
    """
    # This used to delegate to `operonx.serve.OperonApp` — a package
    # that is not installed, not a dependency and not in this
    # repository, so every call raised ImportError. Meanwhile
    # `operonx.toml` had been declaring the same endpoint by hand for
    # the studio's benefit. The two halves are joined now: this is the
    # one-endpoint convenience form, and it builds the same ServeSpec
    # the manifest would have produced.
    from operonx.app.manifest import ServeSpec
    from operonx.app.serve.app import build_app

    try:
        import uvicorn
    except ImportError:
        raise ImportError(
            'engine.serve() needs the serve extra: pip install "operonx[serve]"'
        ) from None

    spec = ServeSpec(
        name=self.name or "serve",
        kind="websocket" if websocket else "http",
        graph="",  # the engine is passed directly
        path=path,
        host=host,
        port=port,
        session="per_connection" if websocket else "per_request",
        max_inflight=kwargs.pop("max_inflight", 4096) if websocket else None,
    )
    app = build_app((spec,), engines={spec.name: self})
    uvicorn.run(app, host=host, port=port, **kwargs)

batch async

batch(
    inputs_list: List[Dict[str, Any]], *, concurrency: int = 10, **kwargs: Any
) -> List[Dict[str, Any]]

Run the workflow concurrently on multiple inputs.

Parameters:

Name	Type	Description	Default
inputs_list	List[Dict[str, Any]]	List of input dicts to process.	required
concurrency	int	Max concurrent executions (default: 10).	10
**kwargs	Any	Extra arguments forwarded to run().	{}

Returns:

Type	Description
List[Dict[str, Any]]	List of result dicts in the same order as inputs.

Source code in operonx/core/engine.py

async def batch(
    self,
    inputs_list: List[Dict[str, Any]],
    *,
    concurrency: int = 10,
    **kwargs: Any,
) -> List[Dict[str, Any]]:
    """Run the workflow concurrently on multiple inputs.

    Args:
        inputs_list: List of input dicts to process.
        concurrency: Max concurrent executions (default: 10).
        **kwargs: Extra arguments forwarded to ``run()``.

    Returns:
        List of result dicts in the same order as inputs.
    """
    sem = asyncio.Semaphore(concurrency)

    async def _run(inp: Dict[str, Any]) -> Dict[str, Any]:
        async with sem:
            return await self.run(inp, **kwargs)

    return list(await asyncio.gather(*[_run(inp) for inp in inputs_list]))

cli

cli() -> None

Interactive CLI mode — read JSON from stdin, print result to stdout.

Source code in operonx/core/engine.py

def cli(self) -> None:
    """Interactive CLI mode — read JSON from stdin, print result to stdout."""
    inputs = json.load(sys.stdin)
    result = asyncio.run(self.run(inputs))
    # Filter internal keys for clean output
    output = {k: v for k, v in result.items() if not k.startswith("$")}
    json.dump(output, sys.stdout, indent=2)
    sys.stdout.write("\n")

input_schema

input_schema() -> Dict[str, Any]

Return JSON Schema describing the workflow's expected inputs.

Source code in operonx/core/engine.py

def input_schema(self) -> Dict[str, Any]:
    """Return JSON Schema describing the workflow's expected inputs."""
    return self._params_to_schema(self.graph.inputs or {}, f"{self.name}_input")

output_schema

output_schema() -> Dict[str, Any]

Return JSON Schema describing the workflow's outputs.

Source code in operonx/core/engine.py

def output_schema(self) -> Dict[str, Any]:
    """Return JSON Schema describing the workflow's outputs."""
    return self._params_to_schema(self.graph.outputs or {}, f"{self.name}_output")

show

show() -> None

Display workflow structure for debugging.

Source code in operonx/core/engine.py

def show(self) -> None:
    """Display workflow structure for debugging."""
    print(f"\n=== Operon Engine: {self.name} ===")
    self.graph.show()
    print()
    self._schema.show()

Decorators

The two decorators that turn ordinary Python into Operonx ops:

op

op(
    func: Optional[Callable] = None,
    *,
    bound: Optional[str] = None,
    cache: Optional[Any] = None,
    delay: float = 0,
    exclude: Optional[Any] = None,
    include: Optional[Any] = None,
    observe_max: Optional[int] = None,
    transient: bool = False,
    show_keys: Optional[Any] = None,
    door: Optional[str] = None,
    retry: Optional[Retry] = None,
    timeout: Optional[Timeout] = None,
) -> Any

Decorator that turns a plain function into a FuncOp factory.

Can be used bare or with keyword arguments::

@op
def double(x: int):
    return {"result": x * 2}

@op(bound="cpu")
def heavy_compute(data: list):
    return {"result": process(data)}

@op(bound="io")
async def call_api(url: str):
    return {"data": await fetch(url)}

# Phase 2 observability filter (Checkpointer + Tracer both respect these):
@op(exclude=["tokens"])              # skip these vars in both observers
@op(include=[])                      # silence entire op (allowlist of nothing)
@op(exclude={"trace": ["pii"]})      # per-observer split (dict form)
@op(observe_max=10_000)              # circuit breaker: raise if op emits > N events

Parameters:

Name	Type	Description	Default
bound	Optional[str]	Execution bound hint for the scheduler. None (default) auto-detects: async → "io", sync → "sync". "sync" — inline dispatch, no asyncio task (fastest). "io" — asyncio task, for network/disk I/O. "cpu" — asyncio.to_thread(), for heavy compute (C extensions release GIL).	None
exclude	Optional[Any]	Vars to hide from observers. list[str] applies to both trace and checkpoint; {"trace": [...], "checkpoint": [...]} splits per observer. Mutually exclusive with include=.	None
include	Optional[Any]	Vars to expose to observers (allowlist). Same shapes as exclude. include=[] silences the op entirely.	None
transient	bool	This op's per-item outputs are stored, delivered, then evicted when the consuming context finishes, instead of being kept for the life of the run. For high-rate streams — a socket reader, an audio chunker — where retaining every item is the difference between a flat run and one that grows without bound. Transience propagates along pull refs: a cell that caches a pull from a transient port is itself transient, so declaring it once on the producer stops the whole chain from retaining. Incompatible with Ref.collect(), which buffers until EOF by definition.	False
observe_max	Optional[int]	Per-op circuit breaker. If the op emits more than this many events in a single run, :class:ObserveBudgetExceeded is raised so runaway generators (e.g. streaming frame sources) fail loudly instead of quietly bloating the checkpointer.	None
show_keys	Optional[Any]	The output key (or keys) that stand for this op — what a viewer prints for it when it has room for one line instead of a port list. @op(show_keys="text") declares it; my_op(..., show_keys="text") at the call site overrides. Unset, the viewer picks from the dataflow.	None
door	Optional[str]	"ingress" or "egress" — this op is where a served graph meets its caller: it reads the session (current_session()) or writes to it. @op(door="egress") on a project's own door op is all a service and the studio need to find it; the built-in ingress() / egress() declare it the same way.	None
retry	Optional[Retry]	Retry(...) — run the op again when it fails with an error worth another try (by default a timeout, a connection error, an HTTP 429 or 5xx), after a growing pause. A generator is retried only before its first yield.	None
timeout	Optional[Timeout]	Timeout(run=..., idle=...) — a deadline per attempt; an op past it fails with TimeoutError. A plain def op needs bound="cpu" for it. my_op(x=..., retry=..., timeout=...) overrides either for one use of the op.	None

Source code in operonx/core/ops/transform/func_op.py

def op(
    func: Optional[Callable] = None,
    *,
    bound: Optional[str] = None,
    cache: Optional[Any] = None,
    delay: float = 0,
    exclude: Optional[Any] = None,
    include: Optional[Any] = None,
    observe_max: Optional[int] = None,
    transient: bool = False,
    show_keys: Optional[Any] = None,
    door: Optional[str] = None,
    retry: Optional[Retry] = None,
    timeout: Optional[Timeout] = None,
) -> Any:
    """Decorator that turns a plain function into a FuncOp factory.

    Can be used bare or with keyword arguments::

        @op
        def double(x: int):
            return {"result": x * 2}

        @op(bound="cpu")
        def heavy_compute(data: list):
            return {"result": process(data)}

        @op(bound="io")
        async def call_api(url: str):
            return {"data": await fetch(url)}

        # Phase 2 observability filter (Checkpointer + Tracer both respect these):
        @op(exclude=["tokens"])              # skip these vars in both observers
        @op(include=[])                      # silence entire op (allowlist of nothing)
        @op(exclude={"trace": ["pii"]})      # per-observer split (dict form)
        @op(observe_max=10_000)              # circuit breaker: raise if op emits > N events

    Args:
        bound: Execution bound hint for the scheduler.
            ``None`` (default) auto-detects: async → ``"io"``, sync → ``"sync"``.
            ``"sync"``  — inline dispatch, no asyncio task (fastest).
            ``"io"``    — asyncio task, for network/disk I/O.
            ``"cpu"``   — asyncio.to_thread(), for heavy compute (C extensions release GIL).
        exclude: Vars to hide from observers. ``list[str]`` applies to both trace and
            checkpoint; ``{"trace": [...], "checkpoint": [...]}`` splits per observer.
            Mutually exclusive with ``include=``.
        include: Vars to expose to observers (allowlist). Same shapes as ``exclude``.
            ``include=[]`` silences the op entirely.
        transient: This op's per-item outputs are stored, delivered, then
            evicted when the consuming context finishes, instead of being kept
            for the life of the run. For high-rate streams — a socket reader,
            an audio chunker — where retaining every item is the difference
            between a flat run and one that grows without bound.

            Transience propagates along pull refs: a cell that caches a pull
            from a transient port is itself transient, so declaring it once on
            the producer stops the whole chain from retaining. Incompatible
            with ``Ref.collect()``, which buffers until EOF by definition.
        observe_max: Per-op circuit breaker. If the op emits more than this many
            events in a single run, :class:`ObserveBudgetExceeded` is raised so
            runaway generators (e.g. streaming frame sources) fail loudly instead
            of quietly bloating the checkpointer.
        show_keys: The output key (or keys) that stand for this op — what a viewer
            prints for it when it has room for one line instead of a port
            list. ``@op(show_keys="text")`` declares it; ``my_op(..., show_keys="text")``
            at the call site overrides. Unset, the viewer picks from the
            dataflow.
        door: ``"ingress"`` or ``"egress"`` — this op is where a served graph
            meets its caller: it reads the session (``current_session()``)
            or writes to it. ``@op(door="egress")`` on a project's own door
            op is all a service and the studio need to find it; the built-in
            ``ingress()`` / ``egress()`` declare it the same way.
        retry: ``Retry(...)`` — run the op again when it fails with an error
            worth another try (by default a timeout, a connection error, an
            HTTP 429 or 5xx), after a growing pause. A generator is retried
            only before its first yield.
        timeout: ``Timeout(run=..., idle=...)`` — a deadline per attempt; an
            op past it fails with ``TimeoutError``. A plain ``def`` op needs
            ``bound="cpu"`` for it. ``my_op(x=..., retry=..., timeout=...)``
            overrides either for one use of the op.
    """
    if door not in (None, "ingress", "egress"):
        raise ValueError(f"@op(door=...) must be 'ingress' or 'egress', got {door!r}")
    # Checked here so a bad value fails where the op is declared.
    op_policy(retry, timeout)
    # Validate observability config at decoration time so the raise
    # surfaces where the op is declared, not where it runs.
    from operonx.core.ops.base import _normalise_observability

    _decl_exclude, _decl_include = _normalise_observability(exclude, include)

    def decorator(fn):
        module = fn.__module__ or ""
        if module in ("__main__", "") or "." not in module:
            fn._func_name = fn.__name__
        else:
            fn._func_name = f"{module}.{fn.__name__}"
        if bound is not None:
            fn._op_bound = bound
        if cache is not None:
            fn._op_cache = cache
        sig = inspect.signature(fn)
        # a keyword the function takes is its input, never an op setting
        own = frozenset(
            name
            for name, p in sig.parameters.items()
            if p.kind not in (p.VAR_KEYWORD, p.VAR_POSITIONAL)
        )

        @wraps(fn)
        def wrapper(**kwargs):
            return _build(kwargs, {})

        def _build(kwargs, settings):
            _refuse_inside_op_body(fn)
            mappings, init_kwargs = split_shorthand_kwargs(kwargs, {"return_keys"}, own=own)
            init_kwargs.update(settings)
            op_bound = init_kwargs.pop("bound", bound)
            op_delay = init_kwargs.pop("delay", delay)
            # Per-call kwargs override the @op-decoration filter, if provided.
            call_exclude = init_kwargs.pop("exclude", None)
            call_include = init_kwargs.pop("include", None)
            call_observe_max = init_kwargs.pop("observe_max", observe_max)
            op_transient = init_kwargs.pop("transient", transient)
            op_show_keys = init_kwargs.pop("show_keys", show_keys)
            op_door = init_kwargs.pop("door", door)
            op_retry = init_kwargs.pop("retry", retry)
            op_timeout = init_kwargs.pop("timeout", timeout)
            eff_exclude = call_exclude if call_exclude is not None else exclude
            eff_include = call_include if call_include is not None else include
            return FuncOp(
                code_fn=fn,
                bound=op_bound,
                delay=op_delay,
                exclude=eff_exclude,
                include=eff_include,
                observe_max=call_observe_max,
                transient=op_transient,
                show_keys=op_show_keys,
                door=op_door,
                retry=op_retry,
                timeout=op_timeout,
                _mappings=mappings or None,
                **init_kwargs,
            )

        register_skip(wrapper)
        register_skip(_build)
        wrapper.__wrapped__ = fn
        wrapper.configure = configurer(_build, fn.__name__)
        return wrapper

    if func is not None:
        # @op without parentheses
        return decorator(func)
    # @op(bound="cpu") with parentheses
    return decorator

graph

GraphOp — container op that manages a graph of child ops.

Classes

GraphOp

GraphOp(
    concurrency: int = 64,
    auto_soft: bool = True,
    strict_dag: bool = False,
    **kwargs,
)

Bases: BaseOp

Container op that holds and executes a directed graph of child ops.

Lifecycle::

1. DEFINE        with GraphOp(name="wf") as g:
                     a = double(x=PARENT["x"])
                     b = add(a=a["result"], b=PARENT["y"])
                     START >> a >> b >> END
                 Ops auto-register via context manager. Edges via >> operator.
                 Inputs/outputs auto-discovered from PARENT refs.

2. BUILD         g.build()  (or auto on first run)
                 _setup_schema    scan PARENT refs → graph inputs/outputs
                 _setup_endpoints find entry/exit ops from topology
                 _build()         adj list + ready counts + stream ready counts
                 validate         branch targets, cycles, reachability, refs

3. EXECUTE       g.run(state, context_id)  — async generator
                 → run_task_scheduler()  drives ops via Frame/EOF events
                 → yields (ctx, outputs) per batch or per stream frame
                 → loop iteration handled inside scheduler EOF handler

4. EXPORT        serialize()  the graph as a config dict
                 validate()   graph structure validation
                 show()       debug display

Source code in operonx/core/ops/graph/graph_op.py

def __init__(
    self,
    concurrency: int = 64,
    auto_soft: bool = True,
    strict_dag: bool = False,
    **kwargs,
):
    super().__init__(**kwargs)
    if self._policy is not None:
        self._check_graph_policy(*self._policy)
    self._token = None
    self._is_building = True
    self._ops: Dict[str, BaseOp] = {}
    self._edges = {}
    self.entries = []
    self.exits = []
    self.prevs = defaultdict(list)
    self.nexts = defaultdict(list)
    self.concurrency = concurrency
    self._loop_config = None
    self._shared_vars = {}  # {var_name: initial_value} — set by PARENT.shared() or PARENT.declare()
    self._reducer_vars = {}  # {var_name: reducer_fn} — set by PARENT.declare(reducers=...)
    # Declared vars whose writers may race: PARENT.declare(..., allow_race=True).
    self._race_vars: set = set()
    self._adj = {}  # {op_name: [Link(dst, soft), ...]}
    # {op_name: ((handler, vars), ...)} — error edges, `op.on_error(h)`:
    # followed only when op fails; `vars` are the handler's unwired
    # `error` / `op` / `inputs` parameters the failure is written into.
    self._err_adj: Dict[str, tuple] = {}
    self._initial_ready = {}  # {op_name: ready_count}
    self._stream_initial_ready = {}  # {gen_name: {op_name: ready_count_for_stream_ctx}}
    self._scheduler = None  # set by build()
    self._out_vars: Dict[
        str, dict
    ] = {}  # {op_name: {src_var: dest_var}} — vars mapped to PARENT output
    # {output_var: [ops that write it]} — set by _setup_schema; tells an
    # output its ops wrote from one that only holds the same-named input.
    self._output_writers: Dict[str, List[BaseOp]] = {}
    self._auto_soft = auto_soft  # auto-soften branch-merge edges at build time
    # Phase 3: opt-out for the Level-2 cycle→loop rewrite. When True, back-edges
    # remain as-is and hit the classic validate() warning path.
    self._strict_dag = strict_dag
    # Phase 3: hidden loops created by the cycle rewrite carry these markers.
    self._synthetic = False
    self._loop_mode = None  # None (classic), "synthetic" (rewritten hidden loop)
    self._back_edge_sources: set = set()  # audit only; termination consults _back_edges
    self._back_edges: list = []  # List[(u_name, v_name)] for termination per back-edge
    # List[(u_name, dst)]: an edge from loop-body op u to dst outside the
    # loop. The scheduler routes the loop op to dst, once, when the loop
    # exits — and only if u took that edge in the final iteration.
    self._exit_edges: list = []
    self._rewritten_from = None  # audit dict populated by rewrite_cycles_to_loops

Methods:

get_current_graph staticmethod

get_current_graph() -> Optional[GraphOp]

Return the current graph from context.

Source code in operonx/core/ops/graph/graph_op.py

@staticmethod
def get_current_graph() -> Optional["GraphOp"]:
    """Return the current graph from context."""
    try:
        return _current_graph.get()
    except LookupError:
        return None

add_op

add_op(op: BaseOp) -> BaseOp

Add an op to the graph.

Source code in operonx/core/ops/graph/graph_op.py

def add_op(self, op: BaseOp) -> BaseOp:
    """Add an op to the graph."""
    if not self._is_building:
        raise RuntimeError("Cannot add op after graph has been built")

    if getattr(op, "_is_operonx_builder", False):
        name = getattr(op, "_name", None) or type(op).__name__
        LOGGER.error(
            "%s '%s' is not built. Call .build() or .else_() before adding to graph.",
            type(op).__name__,
            name,
        )
        raise TypeError(
            f"{type(op).__name__} '{name}' is not built. "
            f"Call .build() or .else_() to create the op."
        )

    if op in [START, END]:
        return op

    if op.name in self._ops:
        LOGGER.warning(
            "Graph [highlight]%s[/highlight]: op [highlight]%s[/highlight] already exists and will be overwritten",
            self.name,
            op.name,
        )

    self._ops[op.name] = op

    if hasattr(op, "start") and op.start:
        if op.name not in self.entries:
            self.entries.append(op.name)

    if hasattr(op, "end") and op.end:
        if op.name not in self.exits:
            self.exits.append(op.name)

    return op

add_edge

add_edge(
    source: str,
    target: str,
    type: EdgeType = "normal",
    soft: bool = False,
    hard: bool = False,
)

Add an edge between two ops.

Parameters:

Name	Type	Description	Default
source	str	Source op name.	required
target	str	Target op name.	required
type	EdgeType	Edge type (normal, lookback, condition).	'normal'
soft	bool	If True, edge does not count toward ready_count. Used for branch outputs when only one branch executes.	False
hard	bool	If True, opt this edge out of auto-softening at build time. Use for the rare case where a branch-descended pred must still be waited on as a hard dependency.	False

Source code in operonx/core/ops/graph/graph_op.py

def add_edge(
    self,
    source: str,
    target: str,
    type: EdgeType = "normal",
    soft: bool = False,
    hard: bool = False,
):
    """Add an edge between two ops.

    Args:
        source: Source op name.
        target: Target op name.
        type: Edge type (normal, lookback, condition).
        soft: If True, edge does not count toward ready_count.
              Used for branch outputs when only one branch executes.
        hard: If True, opt this edge out of auto-softening at build time.
              Use for the rare case where a branch-descended pred must
              still be waited on as a hard dependency.
    """
    if not self._is_building:
        raise RuntimeError("Cannot add edge after graph has been built!")

    if source == START.name:
        if target not in self._ops:
            raise ValueError(f"Target op '{target}' not found")

        target_node = self._ops[target]
        target_node.start = True

        if target not in self.entries:
            self.entries.append(target)

        return

    if target == END.name:
        if source not in self._ops:
            raise ValueError(f"Source op '{source}' not found")

        source_node = self._ops[source]
        source_node.end = True

        if source not in self.exits:
            self.exits.append(source)

        return

    if target == PARENT.name:
        return

    if source not in self._ops:
        raise ValueError(f"Source op '{source}' not found")
    if target not in self._ops:
        raise ValueError(f"Target op '{target}' not found")

    new_edge = EdgeConfig(
        from_node=source, to_node=target, type=type, soft=soft, pinned_hard=hard
    )
    if (source, target) not in self._edges:
        self._edges[source, target] = new_edge
        self.nexts[source].append(target)
        self.prevs[target].append(source)

build

build()

Build graph: cycle-rewrite → children → schema → endpoints → validation → topology.

The cycle-rewrite pass (Phase 3 Level-2) runs FIRST so that any synthetic hidden GraphOp.loop children it creates are then built alongside the user's own children. Ordering matters — validate() and auto_soft assume a DAG and would produce wrong results on cyclic input.

Source code in operonx/core/ops/graph/graph_op.py

def build(self):
    """Build graph: cycle-rewrite → children → schema → endpoints → validation → topology.

    The cycle-rewrite pass (Phase 3 Level-2) runs FIRST so that any
    synthetic hidden ``GraphOp.loop`` children it creates are then built
    alongside the user's own children. Ordering matters — validate() and
    auto_soft assume a DAG and would produce wrong results on cyclic input.
    """
    # Phase 3: rewrite user-authored back-edges into hidden loop nodes
    # BEFORE building children (so the hidden loop children get built too).
    from operonx.core.ops.graph.cycle_rewrite import check_loop_caps, rewrite_cycles_to_loops

    rewrite_cycles_to_loops(self)
    check_loop_caps(self)

    for child in self._ops.values():
        if hasattr(child, "build"):
            child.build()

    self._setup_schema()
    self._setup_endpoints()

    result = self.validate()
    result.raise_if_errors()

    self._auto_soften_edges()

    self._build()

    self._scheduler = Scheduler(self)
    self._is_building = False
    self._cache_full_names()

    # Auto-detect bound from children, with user override.
    # If user explicitly set bound on the graph, respect it.
    # Otherwise: all children sync → graph is sync (inline); any io/cpu → task.
    # A policy — the graph's own timeout, a child's retry — needs a task:
    # inline, nothing could interrupt it or wait between attempts.
    if self.bound is None:
        if self._policy is None and all(
            getattr(op, "bound", None) == "sync" and getattr(op, "_policy", None) is None
            for op in self._ops.values()
        ):
            self.bound = "sync"
        else:
            self.bound = "io"

store_result

store_result(
    state: MemoryState, result: Dict[str, Any], context_id: str
) -> None

Store the graph's outputs — except its declared cells.

A declared cell already holds the graph's value: every write inside the graph went into it. Storing the graph's output there again hands the cell's reducer its own accumulated value as a fresh delta — an add_messages cell came out with every message twice.

Source code in operonx/core/ops/graph/graph_op.py

def store_result(self, state: "MemoryState", result: Dict[str, Any], context_id: str) -> None:
    """Store the graph's outputs — except its declared cells.

    A declared cell already holds the graph's value: every write inside
    the graph went into it. Storing the graph's output there again hands
    the cell's reducer its own accumulated value as a fresh delta — an
    ``add_messages`` cell came out with every message twice.
    """
    if self._shared_vars and result:
        result = {k: v for k, v in result.items() if k not in self._shared_vars}
    super().store_result(state, result, context_id)

run async

run(
    state: MemoryState, context_id: Optional[tuple] = None
) -> AsyncGenerator[Tuple[tuple, Dict[str, Any]], None]

Execute graph: get inputs → schedule ops → loop if needed → store results.

Source code in operonx/core/ops/graph/graph_op.py

async def run(
    self,
    state: "MemoryState",
    context_id: Optional[tuple] = None,
) -> AsyncGenerator[Tuple[tuple, Dict[str, Any]], None]:
    """Execute graph: get inputs → schedule ops → loop if needed → store results."""

    # `BaseOp.run` honours this and `GraphOp` overrides `run`, so without
    # the same check here `enabled=False` silenced a subgraph's *output*
    # while its children still ran — every op inside it, LLM calls and
    # all. A consumer switching a stage off to save the spend kept
    # paying for it and had no way to tell.
    if context_id is None:
        context_id = DEFAULT_CONTEXT

    if not self.enabled:
        # Same contract as `BaseOp.run`: completed, produced nothing.
        # Returning without a yield leaves every successor waiting
        # forever — see the note there.
        yield context_id, {}
        return

    request_id = state.request_id
    start_time = datetime.now(timezone.utc)
    perf_start = perf_counter()
    _inputs = {}
    _outputs = {}
    error_msg = None

    try:
        _inputs = self.get_inputs(state, context_id=context_id)
        self._seed_declared_inputs(state, context_id)

        if self._is_building:
            self.build()

        if self._policy is None:
            _outputs, stream_ctxs, _interrupted = await self._scheduler.run(state, context_id)
        else:
            # The deadline cancels the nested scheduler, whose `finally`
            # cancels the op tasks it started; the TimeoutError is then
            # this graph's failure, recorded below like any other.
            run_s = self._policy.timeout.run
            async with _Deadline(
                asyncio.get_running_loop().time() + run_s,
                f"{self.full_name} ran past Timeout(run={run_s:g}) and was cancelled",
            ):
                _outputs, stream_ctxs, _interrupted = await self._scheduler.run(
                    state, context_id
                )

        # A graph with a generator hands its parent one output per stream
        # context. A run in which no stream ran — a branch went around it —
        # hands on its one output at this context, like a graph without
        # one, provided its last ops wrote it. A generator that yielded
        # nothing wrote no output: it still hands on nothing.
        _streamed = bool(stream_ctxs) or (
            any(op.is_gen for op in self._ops.values())
            and all(v is None for v in _outputs.values())
        )
        if _interrupted:
            # The invocation was cancelled from inside. `_outputs` is
            # whatever the cells happened to hold — all-`None` when the
            # cancelled op never wrote — and yielding it hands the parent
            # a result indistinguishable from a successful null answer.
            # The streaming branch below already skips all-`None` items;
            # this is the batch equivalent.
            _outputs = {}
        elif not _streamed:
            if state._op_errors:
                self._drop_unwritten(state, context_id, _outputs)
            failed = (
                self._failed_descendants(state, context_id)
                if all(v is None for v in _outputs.values())
                else ()
            )
            if failed:
                # An op under this graph raised and nothing was written:
                # this run failed. Yielding the all-`None` outputs ran
                # the next op on them, where flat the failing op's
                # successor never runs — and a door answered `200 null`.
                # Not yielding is the flat rule; the streaming branch
                # below already skips all-`None` items.
                _outputs = {}
                if not getattr(self, "_synthetic", False):
                    # A synthetic loop is not an op its author wrote;
                    # the graph around it reports the failure.
                    error_msg = self._subgraph_error(failed)
                    state.record_op_error(
                        self.full_name, error_msg, context_id, handled=bool(self._error_routes)
                    )
                    if self._error_routes:
                        yield (
                            context_id,
                            Failure(self.name, context_id, error_msg, dict(_inputs)),
                        )
            else:
                self.store_result(state, _outputs, context_id)
                yield context_id, _outputs
        else:
            for sctx in stream_ctxs:
                item = self.get_outputs(state, context_id=sctx)
                if state._op_errors:
                    self._drop_unwritten(state, sctx, item)
                if any(v is not None for v in item.values()):
                    self.store_result(state, item, sctx)
                    yield sctx, item

    except Exception:
        import sys

        error_msg = (
            traceback.format_exc()
            if LOGGER.isEnabledFor(40)
            else f"{type(sys.exc_info()[1]).__name__}: {sys.exc_info()[1]}"
        )
        LOGGER.error(
            "[title]\\[%s][/title] Error in op [highlight]%s[/highlight]:\n%s",
            request_id,
            self.name,
            error_msg.rstrip(),
        )
        # A child's failure is caught in the child's own `BaseOp.run`;
        # this is the subgraph failing around its children. Same record.
        state.record_op_error(self.full_name, sys.exc_info()[1], context_id)
        if self._error_routes:
            failed = sys.exc_info()[1]
            yield (
                context_id,
                Failure(
                    self.name, context_id, f"{type(failed).__name__}: {failed}", dict(_inputs)
                ),
            )
        else:
            run_policy = state._run_policy
            if run_policy is not None and run_policy.fail_fast:
                raise fail_fast(state, self.full_name, sys.exc_info()[1])

    finally:
        end_time = datetime.now(timezone.utc)
        duration_ms = (perf_counter() - perf_start) * 1000
        self._log(request_id, context_id, _inputs, _outputs, duration_ms)
        self._store_metrics(
            state,
            context_id,
            start_time=start_time,
            end_time=end_time,
            duration_ms=duration_ms,
        )
        if error_msg is not None:
            state[self.full_name, "error", context_id] = error_msg

serialize

serialize() -> dict

The whole graph as a config dict: ops, edges, entries and exits.

A graph's canonical description; an eval's fingerprint hashes it. It was written for the dropped Rust runtime, and some keys are still that runtime's (initial_ready_count, compiled_adj, stream_initial_ready): kept so the output, and every hash taken of it, stays the same.

Source code in operonx/core/ops/graph/graph_op.py

def serialize(self) -> dict:
    """The whole graph as a config dict: ops, edges, entries and exits.

    A graph's canonical description; an eval's fingerprint hashes it.
    It was written for the dropped Rust runtime, and some keys are
    still that runtime's (``initial_ready_count``, ``compiled_adj``,
    ``stream_initial_ready``): kept so the output, and every hash
    taken of it, stays the same.
    """
    # Phase 3: synthetic hidden loops can't serialize as classic
    # loop_config because their termination is scheduler-side (back-edge
    # activation) with no equivalent until-expression. Emitting a
    # classic loop_config would silently tell external consumers "iter
    # to max_iterations with no exit condition" (HAZARD from Phase 3
    # review). Refuse loudly instead — callers must recompile from
    # source for now.
    if getattr(self, "_loop_mode", None) == "synthetic":
        raise NotImplementedError(
            f"GraphOp '{self.name}' is a synthetic loop from the Phase 3 "
            "cycle-rewrite pass and does not yet have a serialization "
            "format. Serialize the pre-rewrite graph or use "
            "@graph(strict_dag=True) to opt out."
        )

    # Also refuse to serialize an outer graph that contains any synthetic
    # loop descendant — the missing sub-config would poison consumers.
    def _contains_synthetic(node):
        for child in node._ops.values():
            if getattr(child, "_synthetic", False):
                return True
            if hasattr(child, "_ops") and _contains_synthetic(child):
                return True
        return False

    if _contains_synthetic(self):
        raise NotImplementedError(
            f"GraphOp '{self.name}' contains a synthetic hidden loop "
            "(Phase 3 cycle-rewrite output). Serialize the pre-rewrite "
            "graph or use @graph(strict_dag=True) on the affected subgraph."
        )

    base = super().serialize()
    base.update(
        {
            "ops": {name: op.serialize() for name, op in self._ops.items()},
            # An error edge says so; no other edge changes, and neither
            # does any fingerprint hashed from a graph without one.
            "edges": [
                {"from": src, "to": dst, "soft": edge.soft}
                | ({"kind": "error"} if edge.type == "error" else {})
                for (src, dst), edge in self._edges.items()
            ],
            "entries": list(self.entries),
            "exits": list(self.exits),
            "initial_ready_count": dict(self._initial_ready),
            "compiled_adj": {
                op: [[link.dst, link.soft] for link in links] for op, links in self._adj.items()
            },
            "stream_initial_ready": self._stream_initial_ready,
            # Only synthetic loops carry a loop config, and they refuse
            # to serialize above, so this is always None today.
            "loop_config": {"max_iterations": self._loop_config.max_iterations}
            if self._loop_config
            else None,
            "max_stream_concurrent": self.concurrency,
        }
    )
    return base

validate

validate() -> ValidationResult

Run all validations and return result.

Source code in operonx/core/ops/graph/graph_op.py

def validate(self) -> ValidationResult:
    """Run all validations and return result."""
    # Collect ancestor GraphOp names so _validate_refs allows moved-SCC
    # ops to keep their outer PARENT refs after Phase 3 rewrite.
    ancestor_names: set = set()
    cur = self.parent
    while cur is not None and hasattr(cur, "_ops"):
        if getattr(cur, "name", None):
            ancestor_names.add(cur.name)
        cur = getattr(cur, "parent", None)

    # Collect descendant op names inside synthetic hidden loops so
    # _validate_refs allows outer ops to reference moved SCC ops (BUG 3).
    descendant_names: set = set()

    def _walk(node):
        for child in node._ops.values():
            if getattr(child, "_synthetic", False):
                for grand_name in child._ops:
                    descendant_names.add(grand_name)
                _walk(child)

    _walk(self)

    # Collect sibling op names — ops living in any ancestor's _ops (not
    # including ancestor graphs themselves). A BranchOp moved into a
    # synthetic hidden loop may reference siblings that stayed at the
    # outer level (BUG 2 / E3 multi-exit-via-branch): the branch's
    # __branch_target__ is re-routed through the loop's outgoing edges
    # at the outer level.
    sibling_names: set = set()
    cur = self.parent
    while cur is not None and hasattr(cur, "_ops"):
        for name in cur._ops:
            if name != self.name:
                sibling_names.add(name)
        cur = getattr(cur, "parent", None)

    return validate_graph(
        self.name,
        self._ops,
        self._edges,
        self.prevs,
        self.nexts,
        self.entries,
        self.exits,
        ancestor_names=ancestor_names,
        descendant_names=descendant_names,
        sibling_names=sibling_names,
        graph=self,
        arm_sig=self._arm_signatures(),
    )

show

show(indent=0)

Display graph structure (debug).

Source code in operonx/core/ops/graph/graph_op.py

def show(self, indent=0):
    """Display graph structure (debug)."""
    prefix = "  " * indent
    LOGGER.debug("%sGraph: %s", prefix, self.name)
    LOGGER.debug("%sOps: %s", prefix, list(self._ops.keys()))
    LOGGER.debug("%sEdges:", prefix)
    for edge in self._edges.values():
        soft_marker = " (soft)" if edge.soft else ""
        LOGGER.debug(
            "%s  %s -> %s: %s%s", prefix, edge.from_node, edge.to_node, edge.type, soft_marker
        )
    LOGGER.debug("%sReady count: %s", prefix, dict(self._initial_ready))

    for child in self._ops.values():
        if isinstance(child, GraphOp):
            child.show(indent + 1)

GraphValidationError

GraphValidationError(result: ValidationResult)

Bases: Exception

Exception raised when graph validation fails.

Source code in operonx/core/ops/graph/validation.py

def __init__(self, result: ValidationResult):
    self.result = result
    details = "\n".join(f"- {issue.message}" for issue in result.errors)
    super().__init__(
        f"Graph '{result.graph_name}' validation failed with "
        f"{len(result.errors)} error(s):\n{details}"
    )

ValidationIssue dataclass

ValidationIssue(
    level: ValidationLevel,
    category: str,
    message: str,
    op_name: Optional[str] = None,
    target_name: Optional[str] = None,
    available_nodes: List[str] = list(),
    suggestions: List[str] = list(),
)

A single validation issue found in the graph.

ValidationLevel

Bases: Enum

Severity level for validation issues.

ValidationResult dataclass

ValidationResult(graph_name: str, issues: List[ValidationIssue] = list())

Result of graph validation.

Methods:

raise_if_errors

raise_if_errors()

Raise exception if there are any errors.

Source code in operonx/core/ops/graph/validation.py

def raise_if_errors(self):
    """Raise exception if there are any errors."""
    if self.has_errors:
        LOGGER.error(
            "Graph [highlight]%s[/highlight] validation found %d error(s):",
            self.graph_name,
            len(self.errors),
        )
        for issue in self.errors:
            LOGGER.error(
                "  [%s] %s: %s | Location: %s -> '%s' | Available nodes: %s",
                issue.level.value.upper(),
                issue.category,
                issue.message,
                issue.op_name,
                issue.target_name,
                issue.available_nodes,
            )
        raise GraphValidationError(self)

Inside a run

What an op body can read about its run, the steps it records, and the events engine.stream(mode="tasks") yields:

run_context

run_context() -> Optional[RunContext]

The :class:RunContext of the op body calling it, or None outside one (module level, a plain function call).

Source code in operonx/core/runtime.py

def run_context() -> Optional[RunContext]:
    """The :class:`RunContext` of the op body calling it, or ``None``
    outside one (module level, a plain function call)."""
    frame = _current_frame.get()
    if frame is None:
        return None
    run = frame.run
    return RunContext(
        run_id=run.run_id if run is not None else None,
        thread_id=run.thread_id if run is not None else None,
        op_path=frame.full_name,
        ctx=frame.ctx,
        attempt=frame.attempt,
        deadline=frame.deadline,
        context=run.context if run is not None else None,
    )

RunContext dataclass

RunContext(
    run_id: Optional[str],
    thread_id: Optional[str],
    op_path: str,
    ctx: Tuple[str, ...],
    attempt: int = 1,
    deadline: Optional[float] = None,
    context: Any = None,
)

The run and the invocation an op body is executing in.

Attributes:

Name	Type	Description
run_id	Optional[str]	The run's trace id (start(trace_id=), else its request_id): what every trace consumer files the run under. None for an op driven outside an engine run.
thread_id	Optional[str]	The session_id the caller passed to start(), the conversation this run belongs to; None when none was passed.
op_path	str	The execution's full name ("graph.sub.op").
ctx	Tuple[str, ...]	The execution's ctx tuple (("main", "[2]")).
attempt	int	The attempt running, 1-based (@op(retry=)).
deadline	Optional[float]	When the attempt's Timeout(run=) fires, on the time.monotonic() clock (the event loop's); None without one. :attr:remaining is the time left.
context	Any	The object passed as start(..., context=) (or run()/stream()), as given; None when none was.

Attributes

idempotency_key property

idempotency_key: Optional[str]

:func:invocation_key of this execution — the same on every attempt, different in every other run, op or ctx. Pass it to an external API that deduplicates (a payment, an email). None without a run_id.

remaining property

remaining: Optional[float]

Seconds until :attr:deadline (negative once it passed), or None.

child

child(
    name: str,
    inputs: Optional[Dict[str, Any]] = None,
    *,
    op_type: str = "",
    current: bool = True,
) -> "_ChildScope"

Record the block as a child execution of the op running it::

@op
async def agent(messages: list) -> dict:
    async with child("model", inputs={"messages": messages}, op_type="llm") as call:
        reply = await model(messages)
        call.outputs = reply
        call.attrs["gen_ai.operation.name"] = "chat"
    ...

Parameters:

Name	Type	Description	Default
name	str	What a trace viewer prints (model, lookup_order). No ., [, ] or #: it becomes the ctx segment name[n] and the last part of the record's full name.	required
inputs	Optional[Dict[str, Any]]	What the step was given, recorded as its inputs.	None
op_type	str	The step's kind (llm, tool, turn), so a consumer can type it (Langfuse makes llm a generation).	''
current	bool	Whether the code inside the block runs as the child: its own child() blocks nest under it and :func:run_context describes it. False records the step without that, for a block held open across an async generator's yield (a streamed model call): the consumer's code runs between the yields in the same context, and would otherwise run inside the step — and stay there if the generator is abandoned unclosed.	True

The record's ctx is its parent's plus "<name>[<n>]" — the n-th child of that name under that parent — and its full name the parent's plus ".<name>", so the parent is derived, never stored. The parent is the op's record; for a generator, the record of the yield being produced when the block opens; inside another child, that child. The op's @op(exclude=/include=) trace filter applies to the child's inputs and outputs.

An exception inside the block is recorded as error and re-raised; a cancellation as cancelled. Inside the block, :func:run_context describes the child, so each step has its own idempotency_key. Outside a traced run it records nothing.

Async only: a plain def op cannot use it, and a bound="cpu" body runs in a thread, where records could interleave.

Source code in operonx/core/runtime.py

def child(
    name: str,
    inputs: Optional[Dict[str, Any]] = None,
    *,
    op_type: str = "",
    current: bool = True,
) -> "_ChildScope":
    """Record the block as a child execution of the op running it::

        @op
        async def agent(messages: list) -> dict:
            async with child("model", inputs={"messages": messages}, op_type="llm") as call:
                reply = await model(messages)
                call.outputs = reply
                call.attrs["gen_ai.operation.name"] = "chat"
            ...

    Args:
        name: What a trace viewer prints (``model``, ``lookup_order``). No
            ``.``, ``[``, ``]`` or ``#``: it becomes the ctx segment
            ``name[n]`` and the last part of the record's full name.
        inputs: What the step was given, recorded as its inputs.
        op_type: The step's kind (``llm``, ``tool``, ``turn``), so a
            consumer can type it (Langfuse makes ``llm`` a generation).
        current: Whether the code inside the block runs *as* the child:
            its own ``child()`` blocks nest under it and :func:`run_context`
            describes it. ``False`` records the step without that, for a
            block held open across an async generator's ``yield`` (a
            streamed model call): the consumer's code runs between the
            yields in the same context, and would otherwise run inside the
            step — and stay there if the generator is abandoned unclosed.

    The record's ctx is its parent's plus ``"<name>[<n>]"`` — the n-th
    child of that name under that parent — and its full name the parent's
    plus ``".<name>"``, so the parent is derived, never stored. The parent
    is the op's record; for a generator, the record of the yield being
    produced when the block opens; inside another child, that child. The
    op's ``@op(exclude=/include=)`` trace filter applies to the child's
    inputs and outputs.

    An exception inside the block is recorded as ``error`` and re-raised;
    a cancellation as ``cancelled``. Inside the block, :func:`run_context`
    describes the child, so each step has its own ``idempotency_key``.
    Outside a traced run it records nothing.

    Async only: a plain ``def`` op cannot use it, and a ``bound="cpu"``
    body runs in a thread, where records could interleave.
    """
    if not isinstance(name, str) or not _CHILD_NAME.match(name):
        raise ValueError(
            f"child name {name!r} is not usable: it becomes the ctx segment "
            f"'<name>[n]' and the last part of a full name, so it must be a non-empty "
            "string with no '.', '[', ']' or '#'. Use a plain name such as 'model'."
        )
    return _ChildScope(name, inputs, op_type, current)

ChildExecution

ChildExecution(name: str)

The handle :func:child yields. Set :attr:outputs (a dict; any other value is recorded as {"_": value}), :attr:attrs (semantic attributes such as gen_ai.operation.name) and :attr:redact before the block ends.

redact is (values: dict) -> dict: every exporter (the run stores, the Local and Langfuse consumers) writes the record's inputs and outputs through it; the record in memory keeps them as recorded. For a credential a step saw that must not leave the process, scrubbed where the trace is written instead of on the run's event loop.

Source code in operonx/core/runtime.py

def __init__(self, name: str) -> None:
    self.name = name
    self.outputs: Any = {}
    self.attrs: Dict[str, Any] = {}
    self.redact: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None

TaskStarted dataclass

TaskStarted(op: str, ctx: Tuple[str, ...], attempt: int = 1)

An op invocation, or a child execution, began an attempt.

op is its full name (a child's is its op's plus .<name>), ctx its ctx tuple, attempt 1-based (@op(retry=)).

TaskFinished dataclass

TaskFinished(
    op: str, ctx: Tuple[str, ...], attempt: int = 1, duration_ms: float = 0.0
)

It ended without an error — a generator once, after its last item.

TaskFailed dataclass

TaskFailed(
    op: str,
    ctx: Tuple[str, ...],
    attempt: int = 1,
    duration_ms: float = 0.0,
    error: str = "",
    cancelled: bool = False,
    retrying: bool = False,
)

It ended with an error (error is "TypeName: message"), or was cancelled (cancelled, no error). retrying means retry= runs another attempt, which starts with its own :class:TaskStarted.

Op types

The base classes that compose into a workflow. Most users only touch GraphOp directly (via with GraphOp(...) as g:) — the others are constructed by decorators or factory helpers.

GraphOp

GraphOp(
    concurrency: int = 64,
    auto_soft: bool = True,
    strict_dag: bool = False,
    **kwargs,
)

Bases: BaseOp

Container op that holds and executes a directed graph of child ops.

Lifecycle::

1. DEFINE        with GraphOp(name="wf") as g:
                     a = double(x=PARENT["x"])
                     b = add(a=a["result"], b=PARENT["y"])
                     START >> a >> b >> END
                 Ops auto-register via context manager. Edges via >> operator.
                 Inputs/outputs auto-discovered from PARENT refs.

2. BUILD         g.build()  (or auto on first run)
                 _setup_schema    scan PARENT refs → graph inputs/outputs
                 _setup_endpoints find entry/exit ops from topology
                 _build()         adj list + ready counts + stream ready counts
                 validate         branch targets, cycles, reachability, refs

3. EXECUTE       g.run(state, context_id)  — async generator
                 → run_task_scheduler()  drives ops via Frame/EOF events
                 → yields (ctx, outputs) per batch or per stream frame
                 → loop iteration handled inside scheduler EOF handler

4. EXPORT        serialize()  the graph as a config dict
                 validate()   graph structure validation
                 show()       debug display

Source code in operonx/core/ops/graph/graph_op.py

def __init__(
    self,
    concurrency: int = 64,
    auto_soft: bool = True,
    strict_dag: bool = False,
    **kwargs,
):
    super().__init__(**kwargs)
    if self._policy is not None:
        self._check_graph_policy(*self._policy)
    self._token = None
    self._is_building = True
    self._ops: Dict[str, BaseOp] = {}
    self._edges = {}
    self.entries = []
    self.exits = []
    self.prevs = defaultdict(list)
    self.nexts = defaultdict(list)
    self.concurrency = concurrency
    self._loop_config = None
    self._shared_vars = {}  # {var_name: initial_value} — set by PARENT.shared() or PARENT.declare()
    self._reducer_vars = {}  # {var_name: reducer_fn} — set by PARENT.declare(reducers=...)
    # Declared vars whose writers may race: PARENT.declare(..., allow_race=True).
    self._race_vars: set = set()
    self._adj = {}  # {op_name: [Link(dst, soft), ...]}
    # {op_name: ((handler, vars), ...)} — error edges, `op.on_error(h)`:
    # followed only when op fails; `vars` are the handler's unwired
    # `error` / `op` / `inputs` parameters the failure is written into.
    self._err_adj: Dict[str, tuple] = {}
    self._initial_ready = {}  # {op_name: ready_count}
    self._stream_initial_ready = {}  # {gen_name: {op_name: ready_count_for_stream_ctx}}
    self._scheduler = None  # set by build()
    self._out_vars: Dict[
        str, dict
    ] = {}  # {op_name: {src_var: dest_var}} — vars mapped to PARENT output
    # {output_var: [ops that write it]} — set by _setup_schema; tells an
    # output its ops wrote from one that only holds the same-named input.
    self._output_writers: Dict[str, List[BaseOp]] = {}
    self._auto_soft = auto_soft  # auto-soften branch-merge edges at build time
    # Phase 3: opt-out for the Level-2 cycle→loop rewrite. When True, back-edges
    # remain as-is and hit the classic validate() warning path.
    self._strict_dag = strict_dag
    # Phase 3: hidden loops created by the cycle rewrite carry these markers.
    self._synthetic = False
    self._loop_mode = None  # None (classic), "synthetic" (rewritten hidden loop)
    self._back_edge_sources: set = set()  # audit only; termination consults _back_edges
    self._back_edges: list = []  # List[(u_name, v_name)] for termination per back-edge
    # List[(u_name, dst)]: an edge from loop-body op u to dst outside the
    # loop. The scheduler routes the loop op to dst, once, when the loop
    # exits — and only if u took that edge in the final iteration.
    self._exit_edges: list = []
    self._rewritten_from = None  # audit dict populated by rewrite_cycles_to_loops

Methods:

get_current_graph staticmethod

get_current_graph() -> Optional[GraphOp]

Return the current graph from context.

Source code in operonx/core/ops/graph/graph_op.py

@staticmethod
def get_current_graph() -> Optional["GraphOp"]:
    """Return the current graph from context."""
    try:
        return _current_graph.get()
    except LookupError:
        return None

add_op

add_op(op: BaseOp) -> BaseOp

Add an op to the graph.

Source code in operonx/core/ops/graph/graph_op.py

def add_op(self, op: BaseOp) -> BaseOp:
    """Add an op to the graph."""
    if not self._is_building:
        raise RuntimeError("Cannot add op after graph has been built")

    if getattr(op, "_is_operonx_builder", False):
        name = getattr(op, "_name", None) or type(op).__name__
        LOGGER.error(
            "%s '%s' is not built. Call .build() or .else_() before adding to graph.",
            type(op).__name__,
            name,
        )
        raise TypeError(
            f"{type(op).__name__} '{name}' is not built. "
            f"Call .build() or .else_() to create the op."
        )

    if op in [START, END]:
        return op

    if op.name in self._ops:
        LOGGER.warning(
            "Graph [highlight]%s[/highlight]: op [highlight]%s[/highlight] already exists and will be overwritten",
            self.name,
            op.name,
        )

    self._ops[op.name] = op

    if hasattr(op, "start") and op.start:
        if op.name not in self.entries:
            self.entries.append(op.name)

    if hasattr(op, "end") and op.end:
        if op.name not in self.exits:
            self.exits.append(op.name)

    return op

add_edge

add_edge(
    source: str,
    target: str,
    type: EdgeType = "normal",
    soft: bool = False,
    hard: bool = False,
)

Add an edge between two ops.

Parameters:

Name	Type	Description	Default
source	str	Source op name.	required
target	str	Target op name.	required
type	EdgeType	Edge type (normal, lookback, condition).	'normal'
soft	bool	If True, edge does not count toward ready_count. Used for branch outputs when only one branch executes.	False
hard	bool	If True, opt this edge out of auto-softening at build time. Use for the rare case where a branch-descended pred must still be waited on as a hard dependency.	False

Source code in operonx/core/ops/graph/graph_op.py

def add_edge(
    self,
    source: str,
    target: str,
    type: EdgeType = "normal",
    soft: bool = False,
    hard: bool = False,
):
    """Add an edge between two ops.

    Args:
        source: Source op name.
        target: Target op name.
        type: Edge type (normal, lookback, condition).
        soft: If True, edge does not count toward ready_count.
              Used for branch outputs when only one branch executes.
        hard: If True, opt this edge out of auto-softening at build time.
              Use for the rare case where a branch-descended pred must
              still be waited on as a hard dependency.
    """
    if not self._is_building:
        raise RuntimeError("Cannot add edge after graph has been built!")

    if source == START.name:
        if target not in self._ops:
            raise ValueError(f"Target op '{target}' not found")

        target_node = self._ops[target]
        target_node.start = True

        if target not in self.entries:
            self.entries.append(target)

        return

    if target == END.name:
        if source not in self._ops:
            raise ValueError(f"Source op '{source}' not found")

        source_node = self._ops[source]
        source_node.end = True

        if source not in self.exits:
            self.exits.append(source)

        return

    if target == PARENT.name:
        return

    if source not in self._ops:
        raise ValueError(f"Source op '{source}' not found")
    if target not in self._ops:
        raise ValueError(f"Target op '{target}' not found")

    new_edge = EdgeConfig(
        from_node=source, to_node=target, type=type, soft=soft, pinned_hard=hard
    )
    if (source, target) not in self._edges:
        self._edges[source, target] = new_edge
        self.nexts[source].append(target)
        self.prevs[target].append(source)

build

build()

Build graph: cycle-rewrite → children → schema → endpoints → validation → topology.

The cycle-rewrite pass (Phase 3 Level-2) runs FIRST so that any synthetic hidden GraphOp.loop children it creates are then built alongside the user's own children. Ordering matters — validate() and auto_soft assume a DAG and would produce wrong results on cyclic input.

Source code in operonx/core/ops/graph/graph_op.py

def build(self):
    """Build graph: cycle-rewrite → children → schema → endpoints → validation → topology.

    The cycle-rewrite pass (Phase 3 Level-2) runs FIRST so that any
    synthetic hidden ``GraphOp.loop`` children it creates are then built
    alongside the user's own children. Ordering matters — validate() and
    auto_soft assume a DAG and would produce wrong results on cyclic input.
    """
    # Phase 3: rewrite user-authored back-edges into hidden loop nodes
    # BEFORE building children (so the hidden loop children get built too).
    from operonx.core.ops.graph.cycle_rewrite import check_loop_caps, rewrite_cycles_to_loops

    rewrite_cycles_to_loops(self)
    check_loop_caps(self)

    for child in self._ops.values():
        if hasattr(child, "build"):
            child.build()

    self._setup_schema()
    self._setup_endpoints()

    result = self.validate()
    result.raise_if_errors()

    self._auto_soften_edges()

    self._build()

    self._scheduler = Scheduler(self)
    self._is_building = False
    self._cache_full_names()

    # Auto-detect bound from children, with user override.
    # If user explicitly set bound on the graph, respect it.
    # Otherwise: all children sync → graph is sync (inline); any io/cpu → task.
    # A policy — the graph's own timeout, a child's retry — needs a task:
    # inline, nothing could interrupt it or wait between attempts.
    if self.bound is None:
        if self._policy is None and all(
            getattr(op, "bound", None) == "sync" and getattr(op, "_policy", None) is None
            for op in self._ops.values()
        ):
            self.bound = "sync"
        else:
            self.bound = "io"

store_result

store_result(
    state: MemoryState, result: Dict[str, Any], context_id: str
) -> None

Store the graph's outputs — except its declared cells.

A declared cell already holds the graph's value: every write inside the graph went into it. Storing the graph's output there again hands the cell's reducer its own accumulated value as a fresh delta — an add_messages cell came out with every message twice.

Source code in operonx/core/ops/graph/graph_op.py

def store_result(self, state: "MemoryState", result: Dict[str, Any], context_id: str) -> None:
    """Store the graph's outputs — except its declared cells.

    A declared cell already holds the graph's value: every write inside
    the graph went into it. Storing the graph's output there again hands
    the cell's reducer its own accumulated value as a fresh delta — an
    ``add_messages`` cell came out with every message twice.
    """
    if self._shared_vars and result:
        result = {k: v for k, v in result.items() if k not in self._shared_vars}
    super().store_result(state, result, context_id)

run async

run(
    state: MemoryState, context_id: Optional[tuple] = None
) -> AsyncGenerator[Tuple[tuple, Dict[str, Any]], None]

Execute graph: get inputs → schedule ops → loop if needed → store results.

Source code in operonx/core/ops/graph/graph_op.py

async def run(
    self,
    state: "MemoryState",
    context_id: Optional[tuple] = None,
) -> AsyncGenerator[Tuple[tuple, Dict[str, Any]], None]:
    """Execute graph: get inputs → schedule ops → loop if needed → store results."""

    # `BaseOp.run` honours this and `GraphOp` overrides `run`, so without
    # the same check here `enabled=False` silenced a subgraph's *output*
    # while its children still ran — every op inside it, LLM calls and
    # all. A consumer switching a stage off to save the spend kept
    # paying for it and had no way to tell.
    if context_id is None:
        context_id = DEFAULT_CONTEXT

    if not self.enabled:
        # Same contract as `BaseOp.run`: completed, produced nothing.
        # Returning without a yield leaves every successor waiting
        # forever — see the note there.
        yield context_id, {}
        return

    request_id = state.request_id
    start_time = datetime.now(timezone.utc)
    perf_start = perf_counter()
    _inputs = {}
    _outputs = {}
    error_msg = None

    try:
        _inputs = self.get_inputs(state, context_id=context_id)
        self._seed_declared_inputs(state, context_id)

        if self._is_building:
            self.build()

        if self._policy is None:
            _outputs, stream_ctxs, _interrupted = await self._scheduler.run(state, context_id)
        else:
            # The deadline cancels the nested scheduler, whose `finally`
            # cancels the op tasks it started; the TimeoutError is then
            # this graph's failure, recorded below like any other.
            run_s = self._policy.timeout.run
            async with _Deadline(
                asyncio.get_running_loop().time() + run_s,
                f"{self.full_name} ran past Timeout(run={run_s:g}) and was cancelled",
            ):
                _outputs, stream_ctxs, _interrupted = await self._scheduler.run(
                    state, context_id
                )

        # A graph with a generator hands its parent one output per stream
        # context. A run in which no stream ran — a branch went around it —
        # hands on its one output at this context, like a graph without
        # one, provided its last ops wrote it. A generator that yielded
        # nothing wrote no output: it still hands on nothing.
        _streamed = bool(stream_ctxs) or (
            any(op.is_gen for op in self._ops.values())
            and all(v is None for v in _outputs.values())
        )
        if _interrupted:
            # The invocation was cancelled from inside. `_outputs` is
            # whatever the cells happened to hold — all-`None` when the
            # cancelled op never wrote — and yielding it hands the parent
            # a result indistinguishable from a successful null answer.
            # The streaming branch below already skips all-`None` items;
            # this is the batch equivalent.
            _outputs = {}
        elif not _streamed:
            if state._op_errors:
                self._drop_unwritten(state, context_id, _outputs)
            failed = (
                self._failed_descendants(state, context_id)
                if all(v is None for v in _outputs.values())
                else ()
            )
            if failed:
                # An op under this graph raised and nothing was written:
                # this run failed. Yielding the all-`None` outputs ran
                # the next op on them, where flat the failing op's
                # successor never runs — and a door answered `200 null`.
                # Not yielding is the flat rule; the streaming branch
                # below already skips all-`None` items.
                _outputs = {}
                if not getattr(self, "_synthetic", False):
                    # A synthetic loop is not an op its author wrote;
                    # the graph around it reports the failure.
                    error_msg = self._subgraph_error(failed)
                    state.record_op_error(
                        self.full_name, error_msg, context_id, handled=bool(self._error_routes)
                    )
                    if self._error_routes:
                        yield (
                            context_id,
                            Failure(self.name, context_id, error_msg, dict(_inputs)),
                        )
            else:
                self.store_result(state, _outputs, context_id)
                yield context_id, _outputs
        else:
            for sctx in stream_ctxs:
                item = self.get_outputs(state, context_id=sctx)
                if state._op_errors:
                    self._drop_unwritten(state, sctx, item)
                if any(v is not None for v in item.values()):
                    self.store_result(state, item, sctx)
                    yield sctx, item

    except Exception:
        import sys

        error_msg = (
            traceback.format_exc()
            if LOGGER.isEnabledFor(40)
            else f"{type(sys.exc_info()[1]).__name__}: {sys.exc_info()[1]}"
        )
        LOGGER.error(
            "[title]\\[%s][/title] Error in op [highlight]%s[/highlight]:\n%s",
            request_id,
            self.name,
            error_msg.rstrip(),
        )
        # A child's failure is caught in the child's own `BaseOp.run`;
        # this is the subgraph failing around its children. Same record.
        state.record_op_error(self.full_name, sys.exc_info()[1], context_id)
        if self._error_routes:
            failed = sys.exc_info()[1]
            yield (
                context_id,
                Failure(
                    self.name, context_id, f"{type(failed).__name__}: {failed}", dict(_inputs)
                ),
            )
        else:
            run_policy = state._run_policy
            if run_policy is not None and run_policy.fail_fast:
                raise fail_fast(state, self.full_name, sys.exc_info()[1])

    finally:
        end_time = datetime.now(timezone.utc)
        duration_ms = (perf_counter() - perf_start) * 1000
        self._log(request_id, context_id, _inputs, _outputs, duration_ms)
        self._store_metrics(
            state,
            context_id,
            start_time=start_time,
            end_time=end_time,
            duration_ms=duration_ms,
        )
        if error_msg is not None:
            state[self.full_name, "error", context_id] = error_msg

serialize

serialize() -> dict

The whole graph as a config dict: ops, edges, entries and exits.

A graph's canonical description; an eval's fingerprint hashes it. It was written for the dropped Rust runtime, and some keys are still that runtime's (initial_ready_count, compiled_adj, stream_initial_ready): kept so the output, and every hash taken of it, stays the same.

Source code in operonx/core/ops/graph/graph_op.py

def serialize(self) -> dict:
    """The whole graph as a config dict: ops, edges, entries and exits.

    A graph's canonical description; an eval's fingerprint hashes it.
    It was written for the dropped Rust runtime, and some keys are
    still that runtime's (``initial_ready_count``, ``compiled_adj``,
    ``stream_initial_ready``): kept so the output, and every hash
    taken of it, stays the same.
    """
    # Phase 3: synthetic hidden loops can't serialize as classic
    # loop_config because their termination is scheduler-side (back-edge
    # activation) with no equivalent until-expression. Emitting a
    # classic loop_config would silently tell external consumers "iter
    # to max_iterations with no exit condition" (HAZARD from Phase 3
    # review). Refuse loudly instead — callers must recompile from
    # source for now.
    if getattr(self, "_loop_mode", None) == "synthetic":
        raise NotImplementedError(
            f"GraphOp '{self.name}' is a synthetic loop from the Phase 3 "
            "cycle-rewrite pass and does not yet have a serialization "
            "format. Serialize the pre-rewrite graph or use "
            "@graph(strict_dag=True) to opt out."
        )

    # Also refuse to serialize an outer graph that contains any synthetic
    # loop descendant — the missing sub-config would poison consumers.
    def _contains_synthetic(node):
        for child in node._ops.values():
            if getattr(child, "_synthetic", False):
                return True
            if hasattr(child, "_ops") and _contains_synthetic(child):
                return True
        return False

    if _contains_synthetic(self):
        raise NotImplementedError(
            f"GraphOp '{self.name}' contains a synthetic hidden loop "
            "(Phase 3 cycle-rewrite output). Serialize the pre-rewrite "
            "graph or use @graph(strict_dag=True) on the affected subgraph."
        )

    base = super().serialize()
    base.update(
        {
            "ops": {name: op.serialize() for name, op in self._ops.items()},
            # An error edge says so; no other edge changes, and neither
            # does any fingerprint hashed from a graph without one.
            "edges": [
                {"from": src, "to": dst, "soft": edge.soft}
                | ({"kind": "error"} if edge.type == "error" else {})
                for (src, dst), edge in self._edges.items()
            ],
            "entries": list(self.entries),
            "exits": list(self.exits),
            "initial_ready_count": dict(self._initial_ready),
            "compiled_adj": {
                op: [[link.dst, link.soft] for link in links] for op, links in self._adj.items()
            },
            "stream_initial_ready": self._stream_initial_ready,
            # Only synthetic loops carry a loop config, and they refuse
            # to serialize above, so this is always None today.
            "loop_config": {"max_iterations": self._loop_config.max_iterations}
            if self._loop_config
            else None,
            "max_stream_concurrent": self.concurrency,
        }
    )
    return base

validate

validate() -> ValidationResult

Run all validations and return result.

Source code in operonx/core/ops/graph/graph_op.py

def validate(self) -> ValidationResult:
    """Run all validations and return result."""
    # Collect ancestor GraphOp names so _validate_refs allows moved-SCC
    # ops to keep their outer PARENT refs after Phase 3 rewrite.
    ancestor_names: set = set()
    cur = self.parent
    while cur is not None and hasattr(cur, "_ops"):
        if getattr(cur, "name", None):
            ancestor_names.add(cur.name)
        cur = getattr(cur, "parent", None)

    # Collect descendant op names inside synthetic hidden loops so
    # _validate_refs allows outer ops to reference moved SCC ops (BUG 3).
    descendant_names: set = set()

    def _walk(node):
        for child in node._ops.values():
            if getattr(child, "_synthetic", False):
                for grand_name in child._ops:
                    descendant_names.add(grand_name)
                _walk(child)

    _walk(self)

    # Collect sibling op names — ops living in any ancestor's _ops (not
    # including ancestor graphs themselves). A BranchOp moved into a
    # synthetic hidden loop may reference siblings that stayed at the
    # outer level (BUG 2 / E3 multi-exit-via-branch): the branch's
    # __branch_target__ is re-routed through the loop's outgoing edges
    # at the outer level.
    sibling_names: set = set()
    cur = self.parent
    while cur is not None and hasattr(cur, "_ops"):
        for name in cur._ops:
            if name != self.name:
                sibling_names.add(name)
        cur = getattr(cur, "parent", None)

    return validate_graph(
        self.name,
        self._ops,
        self._edges,
        self.prevs,
        self.nexts,
        self.entries,
        self.exits,
        ancestor_names=ancestor_names,
        descendant_names=descendant_names,
        sibling_names=sibling_names,
        graph=self,
        arm_sig=self._arm_signatures(),
    )

show

show(indent=0)

Display graph structure (debug).

Source code in operonx/core/ops/graph/graph_op.py

def show(self, indent=0):
    """Display graph structure (debug)."""
    prefix = "  " * indent
    LOGGER.debug("%sGraph: %s", prefix, self.name)
    LOGGER.debug("%sOps: %s", prefix, list(self._ops.keys()))
    LOGGER.debug("%sEdges:", prefix)
    for edge in self._edges.values():
        soft_marker = " (soft)" if edge.soft else ""
        LOGGER.debug(
            "%s  %s -> %s: %s%s", prefix, edge.from_node, edge.to_node, edge.type, soft_marker
        )
    LOGGER.debug("%sReady count: %s", prefix, dict(self._initial_ready))

    for child in self._ops.values():
        if isinstance(child, GraphOp):
            child.show(indent + 1)

BranchOp

BranchOp(
    cases: Optional[List[Tuple[Ref, str]]] = None,
    candidates: Optional[List[str]] = None,
    default: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    max_iterations: Optional[int] = None,
    **kwargs,
)

Bases: BaseOp

Op that evaluates conditions and routes execution to different targets.

Conditions are Ref objects with comparison operators. The first matching condition determines the target. An optional anchor input overrides all conditions. Use soft edges (>>~) to connect branch targets to a merge op.

Inputs

anchor (str, optional): Hard-coded target name that overrides conditions. (any): Variables referenced in condition Refs (auto-extracted).

Outputs

target (str): Name of the selected target op. matched (str): Description of which condition matched.

Example::

router = if_(PARENT["score"] >= 90, "excellent").else_("fail")
START >> router >> ~excellent >> merge >> END
router >> ~fail >> merge

Source code in operonx/core/ops/flow/branch_op.py

def __init__(
    self,
    cases: Optional[List[Tuple[Ref, str]]] = None,
    candidates: Optional[List[str]] = None,
    default: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    max_iterations: Optional[int] = None,
    **kwargs,
):
    # Parse inputs/outputs from cases
    parsed_inputs, parsed_outputs, input_keys = self._parse_cases(cases or [])

    # Call super().__init__ without inputs/outputs
    super().__init__(**kwargs)

    # Merge parsed schema with user-provided
    self._init_io(parsed_inputs, parsed_outputs, inputs, outputs)

    self.default = default.name if isinstance(default, BaseOp) else default
    self.given_candidates = candidates
    self.cases = cases or []
    self._case_descriptions = [ref.describe() for ref, _ in self.cases]
    #: input name -> the ``(source, var)`` key its value has when a
    #: condition runs. See `_evaluate_conditions`.
    self._input_keys: Dict[str, Tuple[Any, str]] = input_keys
    #: Predicate ops this branch tests — populated by `Branch._build()`
    #: when a condition is an op rather than a Ref. Empty for the
    #: classic `if_(op["field"], ...)` form.
    self.condition_ops: List[BaseOp] = []
    #: The iteration cap of the loop this branch closes, when one of
    #: its arms is the loop's back-edge. None keeps the default cap;
    #: the cycle rewrite reads it (see `cycle_rewrite.loop_cap`).
    self.max_iterations = _checked_max_iterations(max_iterations)

    self._set_core(self._create_core_function())

Attributes

candidates property

candidates: List[str]

List of possible target op names.

specific_metadata property

specific_metadata: Dict[str, Any]

Return subclass-specific metadata.

Methods:

get_target

get_target(
    state: MemoryState, context_id: Optional[str] = None
) -> Optional[str]

Get the routed target from state.

Source code in operonx/core/ops/flow/branch_op.py

def get_target(self, state: "MemoryState", context_id: Optional[str] = None) -> Optional[str]:
    """Get the routed target from state."""
    return state[self.full_name, "target", context_id]

serialize

serialize() -> dict

Serialize branch op with its conditions.

Source code in operonx/core/ops/flow/branch_op.py

def serialize(self) -> dict:
    """Serialize branch op with its conditions."""
    base = super().serialize()
    base.update(
        {
            "cases": [
                {"condition": ref.serialize(), "target": target} for ref, target in self.cases
            ],
            "default": self.default,
            "candidates": self.given_candidates,
        }
    )
    return base

FuncOp

FuncOp(
    code_fn: Optional[Callable] = None,
    return_keys: Optional[List[str]] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    _mappings: Dict[str, Any] = None,
    **kwargs,
)

Bases: BaseOp

Op that executes a Python function.

Inputs and outputs are auto-extracted from the function's signature and return-statement AST. Both sync and async functions are supported. Prefer the @op decorator over instantiating FuncOp directly.

Inputs

Auto-parsed from the function's parameter list.

Outputs

Auto-parsed from return {"key": ...} via AST, or from explicit return_keys.

Example::

@op
def add(a: int, b: int):
    return {"sum": a + b}

with GraphOp(name="main") as graph:
    result = add(a=PARENT["x"], b=PARENT["y"])
    START >> result >> END

Source code in operonx/core/ops/transform/func_op.py

def __init__(
    self,
    code_fn: Optional[Callable] = None,
    return_keys: Optional[List[str]] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    _mappings: Dict[str, Any] = None,
    **kwargs,
):
    # Parse inputs/outputs từ function signature/AST
    parsed_inputs, parsed_outputs, static_outputs = self._parse_function(code_fn, return_keys)

    # The function's own name is the fallback when auto-naming cannot
    # trust what it read off the calling line — an op built inline
    # inside `if_(...)` has no assignment to read.
    kwargs.setdefault("name_hint", getattr(code_fn, "__name__", None))

    # Split _mappings into inputs/outputs using parsed schema
    if _mappings:
        for key, value in _mappings.items():
            if key in parsed_inputs:
                if inputs is None:
                    inputs = {}
                inputs[key] = value
            elif key in parsed_outputs:
                if outputs is None:
                    outputs = {}
                outputs[key] = value
            else:
                raise TypeError(
                    f"'{key}' is not a known input or output of {code_fn.__name__}(). "
                    f"Inputs: {set(parsed_inputs)}, Outputs: {set(parsed_outputs)}"
                )

    # Resolve cache from @op(cache=...) if not set in kwargs
    if "cache" not in kwargs:
        fn_cache = getattr(code_fn, "_op_cache", None)
        if fn_cache is not None:
            kwargs["cache"] = fn_cache

    # Gọi super().__init__ không truyền inputs/outputs
    super().__init__(**kwargs)

    # Merge parsed schema with user-provided (handles {"*": PARENT} wildcard)
    self._init_io(parsed_inputs, parsed_outputs, inputs, outputs)

    self.code_fn = code_fn
    self._static_outputs = static_outputs
    self._set_core(code_fn)

    self.source = _shape_of(code_fn).source if code_fn else ""

    # Set description từ docstring nếu chưa có
    if not self.description and code_fn and code_fn.__doc__:
        self.description = code_fn.__doc__.strip().split("\n")[0]

Attributes

specific_metadata property

specific_metadata: Dict[str, Any]

Trả về metadata riêng của subclass.

Note (1.0.0): the standalone ParserOp was removed. Text parsing lives inline in LLMOp(fields=..., parser=..., validators=...); pure text helpers (no LLM) are in operonx.providers.parsing.

Branch helpers

Branch

Branch(name: Optional[str] = None, **kwargs)

Fluent builder for creating a BranchOp.

Two usage flavors — both supported, both idiomatic:

Inline form (recommended for common if/else) — pass op instances as targets. The Branch auto-wires branch >> target edges at build time so you never write them yourself, and the branch can drop right into a >> chain::

START >> source >> if_(source["kind"] == "audio", asr).else_(skip_stt)
asr >> denoise >> picker
skip_stt >> picker

Auto-name resolves to the LHS if there is one (stt_route = if_(...)); inline, to a per-graph counter, route_1, route_2.

Named form (for forward refs or a shared branch node) — pass op names as strings. No auto-wiring; you wire branch >> target yourself as before::

router = (if_(PARENT["score"] >= 90, "excellent")
          .if_(PARENT["score"] >= 70, "good")
          .else_("fail"))
# ...
router >> excellent >> merge
router >> good      >> merge
router >> fail      >> merge

Mixed is allowed — string targets skip auto-wiring, op-instance targets get auto-wired.

Initialise the builder.

Parameters:

Name	Type	Description	Default
name	Optional[str]	Op name. If None, auto-inferred from the variable name.	None

Source code in operonx/core/ops/flow/branch_op.py

def __init__(self, name: Optional[str] = None, **kwargs):
    """Initialise the builder.

    Args:
        name: Op name. If None, auto-inferred from the variable name.
    """
    self._name = name
    # cases stores the ORIGINAL target (op instance or string), not just
    # the name, so ``_build()`` can auto-wire op-instance targets.
    self._cases: List[Tuple[Ref, Any]] = []
    self._default: Any = None
    self._inputs: Dict[str, Any] = {}
    self._kwargs = kwargs

Methods:

if_

if_(condition: Union[Ref, BaseOp], target: Union[str, BaseOp]) -> Branch

Add a condition–target case.

Parameters:

Name	Type	Description	Default
condition	Union[Ref, BaseOp]	Ref with comparison (e.g., PARENT["score"] >= 90).	required
target	Union[str, BaseOp]	Target op instance (enables auto-wiring) or op name string.	required

Returns:

Type	Description
Branch	self for chaining.

Source code in operonx/core/ops/flow/branch_op.py

def if_(self, condition: Union[Ref, BaseOp], target: Union[str, BaseOp]) -> "Branch":
    """Add a condition–target case.

    Args:
        condition: Ref with comparison (e.g., ``PARENT["score"] >= 90``).
        target: Target op instance (enables auto-wiring) or op name string.

    Returns:
        self for chaining.
    """
    self._cases.append((condition, target))
    return self

else_

else_(target: Union[str, BaseOp]) -> BranchOp

Set default target and build the BranchOp.

Parameters:

Name	Type	Description	Default
target	Union[str, BaseOp]	Fallback op instance or name string.	required

Returns:

Type	Description
BranchOp	The constructed BranchOp.

Source code in operonx/core/ops/flow/branch_op.py

@register_skip
def else_(self, target: Union[str, BaseOp]) -> "BranchOp":
    """Set default target and build the BranchOp.

    Args:
        target: Fallback op instance or name string.

    Returns:
        The constructed BranchOp.
    """
    self._default = target
    return self._build()

build

build() -> BranchOp

Build the BranchOp without a default target.

Returns:

Type	Description
BranchOp	The constructed BranchOp.

Source code in operonx/core/ops/flow/branch_op.py

@register_skip
def build(self) -> "BranchOp":
    """Build the BranchOp without a default target.

    Returns:
        The constructed BranchOp.
    """
    return self._build()

if_

if_(
    condition: Union[Ref, BaseOp],
    target: Union[str, BaseOp],
    *,
    max_iterations: Optional[int] = None,
) -> Branch

Start a branch declaration with the first condition.

Parameters:

Name	Type	Description	Default
condition	Union[Ref, BaseOp]	What to test — a Ref comparison or a single-output op.	required
target	Union[str, BaseOp]	The op (or op name) to route to when it holds.	required
max_iterations	Optional[int]	When one of this branch's arms loops back to an earlier op, the most iterations that loop may run (default 1000). A loop that reaches it stops, records LoopLimitExceeded in $errors, and runs nothing after it. Refused at build on a branch that closes no loop.	None

Example (inline, auto-wired)::

START >> source >> if_(cond, asr).else_(skip_stt)

Example (named, string targets, wire manually)::

router = if_(PARENT["score"] >= 90, "excellent").else_("fail")
router >> excellent >> merge
router >> fail      >> merge

Example (a loop capped at 50 iterations)::

START >> s >> if_(s["done"] == True, END, max_iterations=50).else_(s)

Source code in operonx/core/ops/flow/branch_op.py

def if_(
    condition: Union[Ref, BaseOp],
    target: Union[str, BaseOp],
    *,
    max_iterations: Optional[int] = None,
) -> Branch:
    """Start a branch declaration with the first condition.

    Args:
        condition: What to test — a Ref comparison or a single-output op.
        target: The op (or op name) to route to when it holds.
        max_iterations: When one of this branch's arms loops back to an
            earlier op, the most iterations that loop may run (default
            1000). A loop that reaches it stops, records
            ``LoopLimitExceeded`` in ``$errors``, and runs nothing after it.
            Refused at build on a branch that closes no loop.

    Example (inline, auto-wired)::

        START >> source >> if_(cond, asr).else_(skip_stt)

    Example (named, string targets, wire manually)::

        router = if_(PARENT["score"] >= 90, "excellent").else_("fail")
        router >> excellent >> merge
        router >> fail      >> merge

    Example (a loop capped at 50 iterations)::

        START >> s >> if_(s["done"] == True, END, max_iterations=50).else_(s)
    """
    if max_iterations is None:
        return Branch().if_(condition, target)
    return Branch(max_iterations=_checked_max_iterations(max_iterations)).if_(condition, target)

State markers

Constants used inside with GraphOp(...) blocks to wire edges and references. None of these are real instances you'd construct — they're sentinels the graph builder recognises.

Marker	Meaning
START	Entry node. Every graph's first hard edge goes from START.
END	Exit node. op >> END auto-forwards op's outputs as the graph result.
PARENT	Reference root for inputs from engine.run(inputs={...}) or the parent graph in nested contexts. Used as PARENT["key"].
PENDING	Sentinel returned by ops that absorb input without producing output.

Top-level convenience

bootstrap

bootstrap(
    *, resources: Optional[Union[str, Path]] = None, env: bool = True
) -> Optional[ResourceHub]

One-line setup for .env and :class:ResourceHub.

• When env is True (default), load ./.env from CWD using python-dotenv (non-override; existing env wins). The path is recorded in BOOTSTRAP_ENV_PATHS for later diagnostic messages.
• When resources is a path, install the hub via :meth:ResourceHub.from_yaml.
• When resources is None, call :meth:ResourceHub.auto — which checks ./resources.yaml and warns on miss.
• Idempotent: if a hub is already installed, return it unchanged.

Returns the installed hub, or None if no resources.yaml was found and none was provided. Pure-compute graphs that don't need a hub can ignore the return value.

Source code in operonx/__init__.py

def bootstrap(
    *,
    resources: Optional[Union[str, Path]] = None,
    env: bool = True,
) -> Optional[ResourceHub]:
    """One-line setup for ``.env`` and :class:`ResourceHub`.

    - When ``env`` is ``True`` (default), load ``./.env`` from CWD using
      ``python-dotenv`` (non-override; existing env wins). The path is
      recorded in ``BOOTSTRAP_ENV_PATHS`` for later diagnostic messages.
    - When ``resources`` is a path, install the hub via
      :meth:`ResourceHub.from_yaml`.
    - When ``resources`` is ``None``, call :meth:`ResourceHub.auto` —
      which checks ``./resources.yaml`` and warns on miss.
    - Idempotent: if a hub is already installed, return it unchanged.

    Returns the installed hub, or ``None`` if no ``resources.yaml`` was
    found and none was provided. Pure-compute graphs that don't need a
    hub can ignore the return value.
    """
    if env:
        _load_env_into_bootstrap()

    if ResourceHub._instance is not None:
        return ResourceHub._instance

    if resources is not None:
        hub = ResourceHub.from_yaml(resources)
        ResourceHub.set_instance(hub)
        return hub

    return ResourceHub.auto()

Provider-neutral types

The v0.7 LLMOp converter layer will translate provider-specific types to/from these at the provider boundary.

ChatMessage

Bases: TypedDict

A single message in a chat conversation.

Provider-neutral shape; backends translate to / from this at the LLMOp boundary.

Required fields

role: One of "system", "user", "assistant", "tool". content: The message body. str for plain text; providers may accept richer structured shapes (tool calls, multi-modal parts) via opt-in fields below.

Optional fields

name: Speaker identifier (for tool replies and named system prompts). tool_call_id: When role == "tool", the id of the tool call this message responds to. tool_calls: When role == "assistant", the list of tool calls the model is requesting. Shape is provider-specific; converter layers normalise this in v0.7.

ChatRole module-attribute

ChatRole = Literal['system', 'user', 'assistant', 'tool']