operonx.core.ops

Op base classes, flow primitives, and graph composition.

ops

Core op types and markers for the Operon workflow engine.

• BaseOp — base class for all workflow ops
• DummyOp — placeholder for START/END markers
• GraphOp — container managing a sub-graph with parallel execution
• BranchOp — conditional routing with precompiled conditions
• FuncOp — wraps a Python function (supports generators for streaming)

Text→structured-output parsing is done inline by LLMOp(fields=..., parser=..., validators=...) — the standalone ParserOp was removed in 1.0.0. Its pure functions live in operonx.providers.parsing for callers that need text parsing without an LLM call.

Markers: START, END, PARENT, PENDING. Decorators: op, graph, if_.

Attributes

OpType module-attribute

OpType = Literal[
    "data",
    "llm",
    "embedding",
    "rerank",
    "vector-search",
    "vector-upsert",
    "vector-delete",
    "doc-fetch",
    "branch",
    "interrupt",
    "emit",
    "code",
    "lambda",
    "prompt",
    "doc-processor",
    "graph",
    "default",
    "dummy",
    "tool-executor",
    "mcp",
    "agent",
]

Các loại node được hỗ trợ trong workflow graph.

Removed in 1.2.0: for / while / stream — superseded in 1.0.0 by back-edge loops, generator ops and Ref.parallel(). parser — ParserOp went away in 1.0.0; parsing lives inside LLMOp(fields=...). milvus / mongo / s3 — named backends, not semantics, and never had ops behind them. Storage is reached through vector-search and doc-fetch. onnx / triton — assigned by the ops deleted in 1.2.0 (they were never in this Literal, which is the drift this cleanup ends).

Added after 1.14.0: vector-upsert / vector-delete — VectorUpsertOp / VectorDeleteOp, the write half of vector-search. agent — operonx_agents.AgentOp (Agent.as_op()): a whole agent run, its turns, model and tool calls child executions.

Added in 1.2.0: interrupt / emit — set by InterruptOp / EmitOp since 1.0.0 but missing here, so the Literal disagreed with the code.

Classes

EOF dataclass

EOF(op: str, ctx: tuple)

Marker that an op's async generator has exhausted.

Created by Scheduler._pump() after op.run() stops yielding (i.e. the underlying function returned or its generator was exhausted). User code never yields EOF — it is emitted implicitly when the op finishes.

Frame dataclass

Frame(op: str, ctx: tuple, result: dict)

One result yielded by an op during execution.

Created by Scheduler._pump() for every (ctx, result) tuple that op.run() yields. User code never constructs Frame directly — just return or yield from an @op function.

Interrupt dataclass

Interrupt(
    op: str = "",
    ctx: tuple = (),
    ctx_to_cancel: Any = SELF_CTX,
    reason: str = "",
)

In-band scheduler cancellation event.

Returned/yielded by user op bodies to cancel queued frames + in-flight tasks at ctx_to_cancel (and its descendants). op and ctx record the emitter for tracing; ctx_to_cancel is the target — typically the prior turn's ctx, stored in SCRATCH when long-running work began.

ctx_to_cancel defaults to :data:Interrupt.SELF, which the scheduler resolves to the emitter's own context. Cancelling the whole run is spelled Interrupt.ALL and has to be asked for::

Interrupt(reason="bad input")                    # this branch
Interrupt(ctx_to_cancel=Interrupt.ALL, ...)      # everything

The empty tuple used to be the default, and it is a prefix of every context — so omitting the argument discarded the entire run and returned cleanly, with no error anywhere for the caller to notice.

The scheduler

1. Drops Frame/EOF items at ctx_to_cancel from the queue.
2. Cancels in-flight _pump tasks at ctx_to_cancel and descendants (skipping the emitter to avoid self-cancel).
3. Clears bookkeeping (ready/seq_active/seq_origins/collect_bufs).
4. Forwards a synthetic ("__interrupt__", emitter_ctx, {...}) tuple to output_queue so ExecutionHandle consumers see it.

Best-effort: data already pushed to consumer-owned queues (e.g. a user-supplied asyncio.Queue) is NOT drained — consumer must handle that itself (see plan §4.6a).

BaseOp

BaseOp(
    id: str = None,
    name: str = None,
    description: str = "",
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    sources: List[str] = None,
    targets: List[str] = None,
    stream: bool = False,
    start: bool = False,
    end: bool = False,
    contain_generation: bool = False,
    verbose: bool = True,
    enabled: bool = True,
    bound: Optional[str] = None,
    cache: Union[bool, str, None] = None,
    delay: float = 0,
    exclude=None,
    include=None,
    observe_max: Optional[int] = None,
    transient: bool = False,
    show_keys: Union[str, Sequence[str], None] = None,
    name_hint: Optional[str] = None,
    door: Optional[str] = None,
    retry=None,
    timeout=None,
)

Bases: ABC

Base class for all ops in a workflow.

An op is the fundamental processing unit. Each op declares typed inputs and outputs (via Param), and implements a core() method that contains the execution logic. Ops are wired together inside a GraphOp using edge operators.

Sections (read top-to-bottom)::

1. INIT            __init__, __slots__, param helpers
2. EDGE OPERATORS  >>, >>~, >, <, [], ~ — wiring ops in a graph
3. EXECUTE         run(), get_inputs/outputs, store_result, _exec_core
4. OBSERVABILITY   _log(), _store_metrics()
5. SERIALIZATION   serialize(), metadata — config dict & tracing

Example::

from operonx.core import GraphOp, op, START, END, PARENT

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

with GraphOp(name="main") as graph:
    d = double(x=PARENT["x"])
    START >> d >> END

Source code in operonx/core/ops/base.py

def __init__(
    self,
    id: str = None,
    name: str = None,
    description: str = "",
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    sources: List[str] = None,
    targets: List[str] = None,
    stream: bool = False,
    start: bool = False,
    end: bool = False,
    contain_generation: bool = False,
    verbose: bool = True,
    enabled: bool = True,
    bound: Optional[str] = None,
    cache: Union[bool, str, None] = None,
    delay: float = 0,
    exclude=None,
    include=None,
    observe_max: Optional[int] = None,
    transient: bool = False,
    show_keys: Union[str, Sequence[str], None] = None,
    name_hint: Optional[str] = None,
    door: Optional[str] = None,
    retry=None,
    timeout=None,
):
    if bound not in self._VALID_BOUNDS:
        raise ValueError(
            f"bound must be 'sync', 'io', 'cpu', or None (auto-detect), got {bound!r}"
        )
    self.bound = bound  # resolved to concrete value by _set_core()
    self._input_cache = None  # built on first get_inputs() call
    self._metrics_idx = None  # (schema, st_idx, et_idx, dur_idx)
    self._error_idx = None  # (schema, err_idx)
    self.cache = cache
    self._cache_scope = None  # (full_name, scope, store id), on the first cached call
    self._static_outputs = None  # unknown unless a subclass knows better
    self.delay = delay
    # retry=Retry(...) / timeout=Timeout(...); checked against the op's
    # core in _set_core, once its bound is known.
    self._policy = op_policy(retry, timeout)
    self._error_routes = False
    self.transient = transient
    self._transient_vars = None  # stamped post-compile by StateSchema
    # Show keys: `show_keys="text"` or `show_keys=["a", "b"]` on the instance,
    # else the class default. Not validated against outputs — a
    # viewer that finds no such output simply has nothing to print.
    self.show_keys = _normalise_show(
        type(self).show_keys_default if show_keys is None else show_keys
    )
    door = type(self).door_default if door is None else door
    if door not in self._VALID_DOORS:
        raise ValueError(f"door must be 'ingress', 'egress' or None, got {door!r}")
    self.door = door

    # Phase 2: observability filter. Normalised to dict form:
    #   {"*": [vars]}                  — apply to both observers
    #   {"trace": [...], "checkpoint": [...]}  — per-observer
    # Mutual exclusion enforced here at op construction.
    self._obs_exclude, self._obs_include = _normalise_observability(exclude, include)
    self.observe_max = observe_max
    self.id = id or uuid.uuid4().hex
    if name is None:
        name = _safe_auto_name(name_hint)
    self.name = name or unique_name()
    #: The function's own name, kept so a caller that discovers the
    #: auto-detected name belonged to someone else can fall back to it.
    self._name_hint = name_hint
    self._full_name = None  # Cached at build time by GraphOp.build()
    self.description = description

    self.stream = stream
    self.start = start
    self.end = end
    self.verbose = verbose
    self.enabled = enabled

    self.sources: List[str] = sources or []
    self.targets: List[str] = targets or []

    self.core: Optional[Callable] = None
    self.is_gen: bool = False
    self.contain_generation = contain_generation
    # Đăng ký vào graph cha
    self.parent = get_current()
    # Use getattr to avoid hasattr's double lookup
    add_op = getattr(self.parent, "add_op", None)
    if add_op is not None:
        add_op(self)

    # Validate op name
    if self.name and not _is_valid_op_name(self.name):
        raise ValueError(
            f"Op name '{self.name}' may only contain alphanumeric characters, underscores, and hyphens"
        )

    # Normalize inputs/outputs to Dict[str, Param]
    self.inputs: Dict[str, Param] = self._normalize_params(inputs)
    self.outputs: Dict[str, Param] = self._normalize_params(outputs)

    # Error if there are overlapping keys between inputs and outputs
    if self.inputs and self.outputs:
        overlapping_keys = set(self.inputs.keys()) & set(self.outputs.keys())
        if overlapping_keys:
            raise ValueError(
                f"Op '{self.name}' has overlapping input/output keys: {overlapping_keys}. "
                "Input and output variable names must be different."
            )

Attributes

full_name property

full_name: str

Fully-qualified hierarchical path of this op. Cached after build().

specific_metadata property

specific_metadata: Dict[str, Any]

Return subclass-specific metadata. Override in subclasses.

metadata property

metadata: Dict[str, Any]

Build a metadata dictionary for this op.

Methods:

get

get(key: str) -> Ref

op["key"] for an output this op may not produce.

The build-time check that op["key"] names one of the op's outputs is skipped, and when the op does not write the key the reading op gets its own parameter default — the same as for any input that arrives empty. For an op handed in by someone else, whose contract makes some outputs optional::

ended = how_it_ended(finish_reason=model.get("finish_reason"))

There is no default= here: the reading op's parameter default is the one place a default lives.

Source code in operonx/core/ops/base.py

def get(self, key: str) -> "Ref":
    """``op["key"]`` for an output this op may not produce.

    The build-time check that ``op["key"]`` names one of the op's
    outputs is skipped, and when the op does not write the key the
    reading op gets its own parameter default — the same as for any
    input that arrives empty. For an op handed in by someone else,
    whose contract makes some outputs optional::

        ended = how_it_ended(finish_reason=model.get("finish_reason"))

    There is no ``default=`` here: the reading op's parameter default
    is the one place a default lives.
    """
    return Ref(self, key, optional=True)

on_error

on_error(handler: BaseOp) -> BaseOp

Error edge: handler runs when this op fails, and only then.

::

@op
def apologise(error: str, op: str, inputs: dict) -> dict:
    return {"reply": "Sorry, try again later."}

look = lookup(order=order)
sorry = apologise()
look.on_error(sorry)          # sorry runs once if look failed
[look, sorry] >> reply        # reply merges the two by itself

The handler runs once per failed invocation of this op — after its last attempt when it has retry= — and waits for its other predecessors like any op. Its parameters named error (the error as "TypeName: message"), op (this op's full name) and inputs (what this op was called with) receive the failure, unless they are wired to something else. The failure is still recorded in "$errors"; Operon(g, errors="raise") does not end the run for it.

Returns handler, so look.on_error(sorry) >> reply reads on.

Source code in operonx/core/ops/base.py

def on_error(self, handler: "BaseOp") -> "BaseOp":
    """Error edge: *handler* runs when this op fails, and only then.

    ::

        @op
        def apologise(error: str, op: str, inputs: dict) -> dict:
            return {"reply": "Sorry, try again later."}

        look = lookup(order=order)
        sorry = apologise()
        look.on_error(sorry)          # sorry runs once if look failed
        [look, sorry] >> reply        # reply merges the two by itself

    The handler runs once per failed invocation of this op — after its
    last attempt when it has ``retry=`` — and waits for its other
    predecessors like any op. Its parameters named ``error`` (the error
    as ``"TypeName: message"``), ``op`` (this op's full name) and
    ``inputs`` (what this op was called with) receive the failure,
    unless they are wired to something else. The failure is still
    recorded in ``"$errors"``; ``Operon(g, errors="raise")`` does not
    end the run for it.

    Returns *handler*, so ``look.on_error(sorry) >> reply`` reads on.
    """
    if self.type == "dummy" or getattr(handler, "type", None) == "dummy":
        raise TypeError(
            "on_error() connects two ops; START, END and PARENT are sentinels. "
            "Call it on the op that may fail, with the op that handles it."
        )
    if not isinstance(handler, BaseOp):
        raise TypeError(f"on_error() takes an op, got {type(handler).__name__}")
    if self.type == "branch":
        raise TypeError(
            f"on_error() on branch '{self.name}': a branch only routes; put the "
            f"error edge on the op whose failure you mean."
        )
    add_edge = getattr(self.parent, "add_edge", None)
    if add_edge is None:
        raise RuntimeError(
            f"on_error() on '{self.name}' wires an edge: call it inside the graph body."
        )
    if handler.parent is not self.parent:
        raise ValueError(
            f"on_error(): '{handler.name}' is not in the same graph as '{self.name}'. "
            f"Build the handler in the graph body where '{self.name}' is."
        )
    add_edge(self.name, handler.name, "error")
    return handler

warmup

warmup() -> None

Called by Operon engine after graph.build() when a ResourceHub is available.

Override in provider ops (LLMOp, EmbeddingOp, etc.) to eagerly initialize backends and eliminate cold-start latency on the first user request.

The default implementation is a no-op — subclasses opt in by overriding.

Source code in operonx/core/ops/base.py

def warmup(self) -> None:
    """Called by Operon engine after graph.build() when a ResourceHub is available.

    Override in provider ops (LLMOp, EmbeddingOp, etc.) to eagerly initialize
    backends and eliminate cold-start latency on the first user request.

    The default implementation is a no-op — subclasses opt in by overriding.
    """

get_inputs

get_inputs(state: MemoryState, context_id: str) -> Dict[str, Any]

Retrieve input values from state based on connection mappings.

Each input's cell index is resolved once per schema (see _build_input_cache); every call, the first included, then reads each input the same way:

1. a shared cell — its one value;
2. the op's own cell at exactly this context;
3. with a pull ref, the source cell (nearest enclosing context with a value), cached into the op's cell;
4. without one, the op's own cell at the nearest enclosing context — a value pushed at a parent context;

then the literal or declared default in place of None, SCRATCH[...] resolved, and a top-level Media unwrapped so the op body receives the raw value.

The first call used to go through state[...] instead, and the two disagreed: only the first unwrapped Media, and only later calls did step 4's walk. The cache is keyed by the schema, which every run of an engine and every item of a fan-out share — so an op's body got a different type, or None instead of a value, depending on whether it had run before.

Source code in operonx/core/ops/base.py

def get_inputs(self, state: "MemoryState", context_id: str) -> Dict[str, Any]:
    """Retrieve input values from state based on connection mappings.

    Each input's cell index is resolved once per schema (see
    ``_build_input_cache``); every call, the first included, then reads
    each input the same way:

    1. a shared cell — its one value;
    2. the op's own cell at exactly this context;
    3. with a pull ref, the source cell (nearest enclosing context
       with a value), cached into the op's cell;
    4. without one, the op's own cell at the nearest enclosing
       context — a value pushed at a parent context;

    then the literal or declared default in place of ``None``,
    ``SCRATCH[...]`` resolved, and a top-level ``Media`` unwrapped so
    the op body receives the raw value.

    The first call used to go through ``state[...]`` instead, and the
    two disagreed: only the first unwrapped ``Media``, and only later
    calls did step 4's walk. The cache is keyed by the schema, which
    every run of an engine and every item of a fan-out share — so an
    op's body got a different type, or ``None`` instead of a value,
    depending on whether it had run before.
    """
    cached = self._input_cache
    if cached is None or cached[0] is not state.schema:
        cached = self._build_input_cache(state.schema)
    # Normalised once so every read below is a plain dict lookup;
    # a cell treats None as the default context anyway.
    ctx = DEFAULT_CONTEXT if context_id is None else context_id
    cells = state._cells
    pull_refs = state.schema._pull_refs
    result = {}
    for var_name, idx, fallback in cached[1]:
        if idx < 0:
            # Not in this schema: nothing to read, only the literal.
            value = None
        else:
            # The hot path of a Cell read, inlined: this runs per input
            # per invocation, 25 times a second per live audio stream.
            cell = cells[idx]
            contexts = cell.contexts
            if cell.is_shared:
                value = cell[ctx]
            elif ctx in contexts:
                value = contexts[ctx]
            else:
                pull_ref = pull_refs[idx]
                if pull_ref is not None and not pull_ref.is_output and pull_ref.idx >= 0:
                    source = cells[pull_ref.idx]
                    source_val = source[ctx]
                    if source_val is not None or source.default_value is not None:
                        value = pull_ref._fn(source_val)
                        contexts[ctx] = value
                    else:
                        value = cell.default_value
                else:
                    value = cell[ctx]  # hierarchy walk
        if value is None:
            if fallback is None:
                continue
            value = fallback
        if isinstance(value, ScratchRef):
            value = state._scratch.get(value.key)
        if isinstance(value, Media):
            # The collector's raw state read keeps the wrapper; only the
            # op body gets the raw value. Nested Media (inside a dict or
            # list) is left alone — rare, and usually intentional.
            value = value.data
        result[var_name] = value
    return result

get_outputs

get_outputs(state: MemoryState, context_id: str) -> Dict[str, Any]

Read output values from state.

Reads directly from this op's output variables. Output connections (outputs={...}) are resolved by the schema at build time — they create refs at the destination, not at this op.

Parameters:

Name	Type	Description	Default
state	MemoryState	Workflow state.	required
context_id	str	Context of this op.	required

Source code in operonx/core/ops/base.py

def get_outputs(self, state: "MemoryState", context_id: str) -> Dict[str, Any]:
    """Read output values from state.

    Reads directly from this op's output variables. Output connections
    (outputs={...}) are resolved by the schema at build time —
    they create refs at the destination, not at this op.

    Args:
        state: Workflow state.
        context_id: Context of this op.
    """
    return {var: state[self.full_name, var, context_id] for var in self.outputs}

normalize_trace_io

normalize_trace_io(inputs: Dict[str, Any], outputs: Dict[str, Any]) -> tuple

Produce a trace-time view of this op's I/O.

Called on the trace's own copies — already filtered by @op(exclude=/include=) — each time an OpExecution is built (:meth:_trace_inputs, :meth:_trace_outputs). Subclasses override to add what the trace should show and the raw I/O does not: LLMOp adds the rendered request (messages) and wraps OpenAI chat-format image_url blocks as Media. The real state value is untouched — return new dicts, never edit these in place.

Built from the filtered copy, so a value hidden from the trace cannot come back through something derived from it.

Default is identity: most ops never override.

Source code in operonx/core/ops/base.py

def normalize_trace_io(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> tuple:
    """Produce a trace-time view of this op's I/O.

    Called on the trace's own copies — already filtered by
    ``@op(exclude=/include=)`` — each time an ``OpExecution`` is built
    (:meth:`_trace_inputs`, :meth:`_trace_outputs`). Subclasses
    override to add what the trace should show and the raw I/O does
    not: LLMOp adds the rendered request (``messages``) and wraps
    OpenAI chat-format ``image_url`` blocks as ``Media``. The real state
    value is untouched — return new dicts, never edit these in place.

    Built from the filtered copy, so a value hidden from the trace
    cannot come back through something derived from it.

    Default is identity: most ops never override.
    """
    return inputs, outputs

store_result

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

Store result dict into state.

Uses state[op, var, ctx] = value for O(1) index-based storage. Extracts $tags special key for dynamic tagging. After all writes commit, calls state.advance_step() so the checkpointer / tracer see a fresh step_id for the next op's writes.

A bare return value has already been wrapped into {output: value} by _as_result_dict, so return False arrives here as a non-empty dict and survives the falsy early-out below. That ordering matters: skipping the write would leave the cell unset, which every reader downstream sees as "the op did not run" rather than "it said no".

Source code in operonx/core/ops/base.py

def store_result(self, state: "MemoryState", result: Dict[str, Any], context_id: str) -> None:
    """Store result dict into state.

    Uses state[op, var, ctx] = value for O(1) index-based storage.
    Extracts $tags special key for dynamic tagging. After all writes
    commit, calls ``state.advance_step()`` so the checkpointer /
    tracer see a fresh ``step_id`` for the next op's writes.

    A bare return value has already been wrapped into `{output: value}`
    by `_as_result_dict`, so `return False` arrives here as a non-empty
    dict and survives the falsy early-out below. That ordering matters:
    skipping the write would leave the cell unset, which every reader
    downstream sees as "the op did not run" rather than "it said no".
    """
    if not result:
        return

    # Extract $tags before storing (don't store as output variable)
    tags = result.pop("$tags", None)
    if tags:
        state.add_tags(tags)

    for key, value in result.items():
        state[self.full_name, key, context_id] = value

    # Phase 2: bump step_id after this op's outputs commit.
    state.advance_step()

save_all_caches staticmethod

save_all_caches() -> int

Write every file-backed op cache to its file. Returns entries written.

Source code in operonx/core/ops/base.py

@staticmethod
def save_all_caches() -> int:
    """Write every file-backed op cache to its file. Returns entries written."""
    return sum(store.save() for store in BaseOp._cache_stores.values())

run async

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

Execute this op as a uniform async generator.

Every op — whether it uses return or yield — is driven through the same three-layer model:

1. User function (@op): plain return {"k": v} or yield {"k": v}. No awareness of Frame/EOF.
2. BaseOp.run() (this method): wraps the user function via _exec_core() into a uniform async generator that yields (context_id, result) tuples. Normal op → one yield. Generator op → N yields, each in its own stream context [i].
3. Scheduler._pump(): consumes this generator. Each yield becomes a Frame event on the queue; when the generator exhausts naturally, _pump emits one EOF event. The user never writes Frame or EOF.

Yields:

Type	Description
AsyncGenerator[tuple[Optional[str], Dict[str, Any]], None]	(context_id, result) — one tuple per item produced by the op.

Source code in operonx/core/ops/base.py

async def run(
    self,
    state: "MemoryState",
    context_id: Optional[str] = None,
) -> "AsyncGenerator[tuple[Optional[str], Dict[str, Any]], None]":
    """Execute this op as a uniform async generator.

    Every op — whether it uses ``return`` or ``yield`` — is driven through
    the same three-layer model:

    1. **User function** (``@op``): plain ``return {"k": v}`` or ``yield {"k": v}``.
       No awareness of Frame/EOF.
    2. **``BaseOp.run()``** (this method): wraps the user function via
       ``_exec_core()`` into a uniform async generator that yields
       ``(context_id, result)`` tuples.  Normal op → one yield.
       Generator op → N yields, each in its own stream context ``[i]``.
    3. **``Scheduler._pump()``**: consumes this generator.  Each yield becomes
       a ``Frame`` event on the queue; when the generator exhausts naturally,
       ``_pump`` emits one ``EOF`` event.  The user never writes Frame or EOF.

    Yields:
        ``(context_id, result)`` — one tuple per item produced by the op.
    """
    if not self.enabled:
        # Completed, produced nothing — not "never completed". A bare
        # `return` ends the generator without yielding, and a successor
        # waiting on this op never becomes ready, so disabling one stage
        # silently stalls everything downstream of it. Yielding an empty
        # result lets the graph flow; `store_result` skips a falsy
        # payload, so the consumer reads None for every field this op
        # would have written.
        yield (context_id if context_id is not None else DEFAULT_CONTEXT), {}
        return

    _tracing = state.tracing
    start_time = None
    end_time = None
    perf_start = perf_counter()
    duration_ms = 0.0
    _inputs = {}
    _outputs = {}
    error_msg = None
    # "TypeName: message" of the op's own failure, for a TaskFailed.
    error_text = ""

    # The frame `_current_frame` held before this op set its own;
    # _MISSING until it does. See the restore in `finally`.
    prev_frame = _MISSING
    idx = 0
    op_started = False
    op_cancelled = False
    ctx_for_end = context_id if context_id is not None else DEFAULT_CONTEXT

    # V3 tracing: `_wf_trace` is None when no consumer is attached
    # (i.e. Operon.start() didn't install a WorkflowTrace). All V3
    # code paths guard on `_wf_trace is not None` so the no-tracing
    # case pays only one ContextVar.get. `_v3_upstreams` is computed
    # once at entry and shared across every OpExecution this op emits
    # (streaming ops keep the same upstreams for all yields).
    _wf_trace = _current_trace.get()
    _v3_upstreams: List[UpstreamRef] = []
    # A transient generator emits one span for the whole stream instead
    # of one per yield; these carry the count to that span.
    _transient_yields = 0
    _transient_last_ctx = None
    # A generator's trace inputs, built at its first yield (see there),
    # and the op_id of the record that holds them: every later record of
    # this invocation names it instead of repeating them.
    _trace_in = None
    _trace_in_at = None
    # This invocation as its body sees it (`run_context()`): its ctx, the
    # attempt running, the deadline. Set once the inputs resolve.
    frame = _Frame(self.full_name, ctx_for_end, self, _wf_trace, state._run_info)
    if self.is_gen and not self.transient:
        # A step the body records while producing item i hangs under
        # that item's yield record. A transient generator has one record.
        frame.item = 0

    try:
        if self.delay > 0:
            await asyncio.sleep(self.delay)

        _inputs = self.get_inputs(state, context_id)

        # V3 tracing: upstreams are stable across the op's lifetime,
        # so compute once at entry. Cheap when disabled.
        if _wf_trace is not None:
            _v3_upstreams = self._v3_infer_upstreams(state, context_id)

        # Record timing AFTER inputs resolved — measures actual processing,
        # not time spent waiting for upstream ops in the scheduler queue.
        start_time = datetime.now(timezone.utc) if _tracing else None
        perf_start = perf_counter()

        op_started = True
        # The body reads this invocation through `_current_frame`:
        # `run_context()`, and the ctx InterruptOp, EmitOp and LLMOp
        # stamp on what they emit.
        prev_frame = _current_frame.get()
        _current_frame.set(frame)
        if _wf_trace is not None and _wf_trace._task_listeners:
            _wf_trace.emit_task(TaskStarted(self.full_name, ctx_for_end, 1))

        # Cache check
        if self.cache is not None:
            _cache_store, _cache_key = self._cache_lookup(_inputs)
            _hit = _cache_store.get(_cache_key)
            if _hit is not _cache.MISS:
                _outputs = _hit
                self.store_result(state, _outputs, context_id)
                yield context_id, _outputs
                return

        base_ctx = context_id if context_id is not None else DEFAULT_CONTEXT
        _yield_start = perf_counter()
        if self._policy is None:
            _source = self._exec_core(_inputs)
        else:
            _source = self._exec_with_policy(
                _inputs, frame, state, _wf_trace, _v3_upstreams, ctx_for_end
            )
        async for result in _source:
            ctx = base_ctx + (f"[{idx}]",) if self.is_gen else context_id
            # Interrupt is a scheduler control event, not a result dict.
            # Forward it untouched — _pump puts it on the queue as-is.
            if isinstance(result, Interrupt):
                yield ctx, result
                idx += 1
                if frame.item is not None:
                    frame.item, frame.counts = idx, None
                _yield_start = perf_counter()
                continue
            self.store_result(state, result, ctx)
            _outputs = result
            if self.is_gen and _tracing:
                _yield_end = perf_counter()
                _now = datetime.now(timezone.utc)
                self._store_metrics(
                    state,
                    ctx,
                    start_time=_now,
                    end_time=_now,
                    duration_ms=(_yield_end - _yield_start) * 1000,
                )
            elif not self.is_gen and _tracing:
                # Batch ops: record timing BEFORE yield. The yield
                # suspends this generator until the scheduler resumes
                # it, inflating duration if measured in finally.
                _batch_end = perf_counter()
                end_time = datetime.now(timezone.utc)
                duration_ms = (_batch_end - perf_start) * 1000
            if not self.is_gen and self.cache is not None:
                _cache_store.put(_cache_key, result)
            # V3 tracing: one OpExecution per yield for generator ops.
            # ctx already carries the yield sub-index (`("main","[T]","[i]")`),
            # so op_id is unique per yield → downstream consumers of
            # yield i produce matching UpstreamRef.from_op_id.
            if _wf_trace is not None and self.is_gen and self.transient:
                # A transient op is a stream: one node per yield means
                # 30,000 spans for a ten-minute call, and every one of
                # them holds its ctx for the life of the run. Count them
                # and emit a single span at EOF instead — see the summary
                # node below.
                _transient_yields += 1
                _transient_last_ctx = ctx
            elif _wf_trace is not None and self.is_gen:
                _yield_id = make_op_id(self.full_name, ctx)
                _inputs_from = _trace_in_at
                if _trace_in is None:
                    _trace_in, _trace_in_at = self._trace_inputs(_inputs), _yield_id
                _wf_trace.record(
                    OpExecution(
                        op_id=_yield_id,
                        op_name=self.name,
                        op_full_name=self.full_name,
                        ctx=ctx,
                        start_time=_yield_start,
                        end_time=perf_counter(),
                        # Shallow copies — the op may reuse its dicts —
                        # with @op(exclude=/include=) applied. Every
                        # trace consumer reads these, so filtering at
                        # the source is what makes the filter mean
                        # anything at all. The inputs are one per
                        # invocation: every yield shares them, and an
                        # LLM stream renders its request once, not
                        # once per frame — and stores it once.
                        inputs=_trace_in,
                        inputs_from=_inputs_from,
                        outputs=self._trace_outputs(result)
                        if isinstance(result, dict)
                        else {"_": result},
                        upstreams=_v3_upstreams,
                        status=STATUS_OK,
                        op_type=str(self.type),
                        is_yield=True,
                        attempt=frame.attempt,
                    )
                )
            yield ctx, result
            idx += 1
            if frame.item is not None:
                frame.item, frame.counts = idx, None
            _yield_start = perf_counter()

    except asyncio.CancelledError:
        # Mark for the OP_END emit below. finally still runs because Python
        # guarantees it on CancelledError. Re-raise so the scheduler sees
        # cancellation propagate through normally (Phase B Interrupt sweep).
        op_cancelled = True
        raise
    except Exception:
        # The one handler for an op's own failure: nothing above this
        # frame sees the exception. The run carries on — one bad op must
        # not end a live call — so the failure is recorded where the
        # caller reads it (`handle.errors`, `"$errors"`), besides the
        # log, the `error` cell and the trace node.
        #
        # ObserveBudgetExceeded is a BaseException subclass and skips
        # this handler by design — the circuit-breaker propagates up
        # to the scheduler and halts the run.

        import sys

        error_msg = (
            traceback.format_exc()
            if LOGGER.isEnabledFor(ERROR)
            else f"{type(sys.exc_info()[1]).__name__}: {sys.exc_info()[1]}"
        )
        LOGGER.error(
            format_event(
                "op_error",
                request_id=state.request_id or "unknown",
                name=self.name,
                error=error_msg.rstrip(),
            ),
        )
        # an error edge takes it: seen, but it does not fail the run
        state.record_op_error(
            self.full_name, sys.exc_info()[1], ctx_for_end, handled=bool(self._error_routes)
        )
        error_text = f"{type(sys.exc_info()[1]).__name__}: {sys.exc_info()[1]}"
        # Recorded first, so "$errors" and the trace show it either way.
        # Then an error edge handles it — `errors="raise"` leaves a
        # handled failure alone — or a fail-fast run ends here.
        if self._error_routes:
            failed = sys.exc_info()[1]
            yield (
                ctx_for_end,
                Failure(
                    self.name, ctx_for_end, 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:
        if self.is_gen or not _tracing:
            # Generators: measure total wall-clock time across all yields.
            # Non-tracing: just compute duration for metrics.
            duration_ms = (perf_counter() - perf_start) * 1000
            end_time = datetime.now(timezone.utc) if _tracing else None
        # else: batch ops already set end_time/duration_ms before yield

        self._store_metrics(
            state,
            context_id,
            start_time=start_time,
            end_time=end_time,
            duration_ms=duration_ms,
        )

        # Write error state via cached index
        err_cached = self._error_idx
        if err_cached is not None and err_cached[0] is state.schema:
            err_idx = err_cached[1]
        else:
            err_idx = state.schema.get_index(self.full_name, "error")
            self._error_idx = (state.schema, err_idx)
        if err_idx >= 0:
            ctx_key = context_id if context_id is not None else DEFAULT_CONTEXT
            if error_msg is not None:
                state._cells[err_idx][ctx_key] = error_msg
            elif "error" not in _outputs:
                state._cells[err_idx][ctx_key] = None

        if _tracing:
            self._log(state.request_id, context_id, _inputs, _outputs, duration_ms)

        # New tracing: OP_END fires regardless of legacy `_tracing` flag
        # — the emitter is NullEmitter when no pipeline is bound, so this
        # is free in the no-tracer path. Status reflects what happened.
        #
        # Cancel-before-start case (op_started=False): the body never ran,
        # no OP_START was emitted, so we skip OP_END too — no orphan event.
        if op_started:
            if op_cancelled:
                status = "cancelled"
            elif error_msg is not None:
                status = "error"
            else:
                status = "ok"
            # V3 tracing:
            #   - batch ops: emit one OpExecution here (per-yield emit
            #     inside the loop only fires for generators)
            #   - streaming ops: yields already recorded; append one
            #     final OpExecution only for cancelled/errored runs so
            #     the trace shows the failure attempt
            if _wf_trace is not None:
                v3_status = (
                    STATUS_CANCELLED
                    if op_cancelled
                    else (STATUS_ERROR if error_msg is not None else STATUS_OK)
                )
                if self.is_gen and self.transient and _transient_yields:
                    # One span for the stream. `items` is the payload —
                    # the values themselves were summarised or never
                    # copied, which is the whole point.
                    _wf_trace.record(
                        OpExecution(
                            op_id=make_op_id(self.full_name, ctx_for_end),
                            op_name=self.name,
                            op_full_name=self.full_name,
                            ctx=ctx_for_end,
                            start_time=perf_start,
                            end_time=perf_start + duration_ms / 1000.0,
                            inputs=_trace_in or self._trace_inputs(_inputs),
                            outputs={
                                "_transient_stream": True,
                                "items": _transient_yields,
                                "last_ctx": str(_transient_last_ctx),
                            },
                            upstreams=_v3_upstreams,
                            status=v3_status,
                            error=error_msg,
                            op_type=str(self.type),
                            attempt=frame.attempt,
                        )
                    )
                # A transient op in a stream emits no per-item node on
                # success: at 30,000 items that is 30,000 spans, each
                # holding its ctx for the life of the run, and it is the
                # last linear term once the cells are released. Failures
                # always emit — a stream you cannot debug is worse than
                # one you cannot count.
                should_emit = (
                    (not self.is_gen) and not self.transient
                ) or v3_status != STATUS_OK
                if should_emit:
                    _wf_trace.record(
                        OpExecution(
                            op_id=make_op_id(self.full_name, ctx_for_end),
                            op_name=self.name,
                            op_full_name=self.full_name,
                            ctx=ctx_for_end,
                            start_time=perf_start,
                            end_time=perf_start + duration_ms / 1000.0,
                            inputs=_trace_in
                            if _trace_in is not None
                            else self._trace_inputs(_inputs),
                            inputs_from=_trace_in_at,
                            outputs=self._trace_outputs(_outputs)
                            if isinstance(_outputs, dict)
                            else {},
                            upstreams=_v3_upstreams,
                            status=v3_status,
                            error=error_msg,
                            op_type=str(self.type),
                            attempt=frame.attempt,
                        )
                    )
                if _wf_trace._task_listeners:
                    if v3_status == STATUS_OK:
                        _wf_trace.emit_task(
                            TaskFinished(
                                self.full_name, ctx_for_end, frame.attempt, duration_ms
                            )
                        )
                    else:
                        _wf_trace.emit_task(
                            TaskFailed(
                                self.full_name,
                                ctx_for_end,
                                frame.attempt,
                                duration_ms,
                                error=error_text,
                                cancelled=op_cancelled,
                            )
                        )
        # Restored by value, not with `ContextVar.reset(token)`: this
        # `finally` does not always run in the context that set it. A
        # generator abandoned by a cancelled pump is closed later by the
        # event loop's async-generator finalizer, in a task of its own,
        # and `reset` refuses a token from another context — the
        # ValueError ended this block there and reached the loop's
        # exception handler on every such cancellation. Setting the
        # previous value reads the same in the right context and is
        # harmless in the finalizer's throwaway one.
        if prev_frame is not _MISSING:
            _current_frame.set(prev_frame)

        if (
            duration_ms > slow_op_ms()
            and LOGGER.isEnabledFor(WARNING)
            and self.type not in _SLOW_EXEMPT
            and self.door is None
            and not getattr(self, "is_gen", False)
        ):
            LOGGER.warning(
                format_event(
                    "op_slow",
                    request_id=state.request_id or "unknown",
                    full_name=self.full_name,
                    duration_ms=f"{duration_ms:.1f}",
                ),
            )

serialize

serialize() -> dict

This op as a config dict (see GraphOp.serialize).

Source code in operonx/core/ops/base.py

def serialize(self) -> dict:
    """This op as a config dict (see ``GraphOp.serialize``)."""
    is_async = inspect.iscoroutinefunction(self.core)
    is_gen = inspect.isgeneratorfunction(self.core) or inspect.isasyncgenfunction(self.core)
    base = {
        "type": self.type,
        "full_name": self.full_name,
        "name": self.name,
        "func_name": getattr(self.core, "_func_name", getattr(self.core, "__name__", None)),
        "python_callable": self.core,
        "is_async": is_async,
        "is_generator": is_gen,
        "enabled": self.enabled,
        "verbose": self.verbose,
        "stream": self.stream,
        "bound": self.bound,
        "inputs": self._serialize_params(self.inputs),
        "outputs": self._serialize_params(self.outputs),
    }
    if self.cache is not None:
        base["cache"] = self.cache
    if self.delay > 0:
        base["delay"] = self.delay
    return base

DummyOp

DummyOp(name: str)

Bases: BaseOp

Sentinel op used as START, END, and PARENT markers.

Source code in operonx/core/ops/_edges.py

def __init__(self, name: str):
    super().__init__(name=name)

Methods:

declare

declare(
    *,
    reducers: dict | None = None,
    allow_race: bool | list | tuple | set = False,
    **vars: Any,
) -> None

Declare shared vars on the current graph, with optional reducers.

Shared vars persist across all stream contexts within the graph. Normal PARENT vars are copied per stream context. When a reducer is registered for a var, concurrent/repeated writes to its shared cell go through reducer(old, new) instead of overwriting. Without a reducer, shared cell writes use last-write-wins semantics.

Parameters:

Name	Type	Description	Default
reducers	dict | None	Optional {var_name: fn} mapping. Each fn is a (old, new) -> merged reducer applied at cell write time. Every key MUST be one of the vars declared in this call (build-time check).	None
allow_race	bool | list | tuple | set	Two ops that may run at once writing a declared cell without a reducer fail the build: the cell keeps whichever write lands last. True says that is intended for every var of this call; a list names some of them.	False
**vars	Any	var_name=initial_value declarations. Same semantics as PARENT.shared().	{}

Raises:

Type	Description
TypeError	if called on anything other than PARENT.
RuntimeError	if called outside a @graph body.
ValueError	if a reducer key is not in the declared vars.

Usage::

from operonx.reducers import add_messages

@graph
def agent():
    PARENT.declare(
        count=0,
        messages=[],
        reducers={"messages": add_messages},
    )

Source code in operonx/core/ops/_edges.py

def declare(
    self,
    *,
    reducers: "dict | None" = None,
    allow_race: "bool | list | tuple | set" = False,
    **vars: "Any",
) -> None:
    """Declare shared vars on the current graph, with optional reducers.

    Shared vars persist across all stream contexts within the graph. Normal
    PARENT vars are copied per stream context. When a reducer is registered
    for a var, concurrent/repeated writes to its shared cell go through
    ``reducer(old, new)`` instead of overwriting. Without a reducer, shared
    cell writes use last-write-wins semantics.

    Args:
        reducers: Optional ``{var_name: fn}`` mapping. Each ``fn`` is a
            ``(old, new) -> merged`` reducer applied at cell write time.
            Every key MUST be one of the vars declared in this call
            (build-time check).
        allow_race: Two ops that may run at once writing a declared cell
            without a reducer fail the build: the cell keeps whichever
            write lands last. ``True`` says that is intended for every
            var of this call; a list names some of them.
        **vars: ``var_name=initial_value`` declarations. Same semantics as
            ``PARENT.shared()``.

    Raises:
        TypeError: if called on anything other than PARENT.
        RuntimeError: if called outside a ``@graph`` body.
        ValueError: if a reducer key is not in the declared vars.

    Usage::

        from operonx.reducers import add_messages

        @graph
        def agent():
            PARENT.declare(
                count=0,
                messages=[],
                reducers={"messages": add_messages},
            )
    """
    if self.name != "__PARENT__":
        raise TypeError("declare() can only be called on PARENT")
    current_graph = get_current()
    if current_graph is None:
        raise RuntimeError("PARENT.declare() must be called inside a @graph function body")

    reducers = reducers or {}
    if not isinstance(reducers, dict):
        raise TypeError(f"declare() reducers must be a dict, got {type(reducers).__name__}")

    # E4: reducer keys must reference declared vars
    unknown = set(reducers) - set(vars)
    if unknown:
        raise ValueError(
            f"declare() reducers reference undeclared vars: {sorted(unknown)}. "
            f"declared vars: {sorted(vars)}"
        )

    if not hasattr(current_graph, "_shared_vars"):
        current_graph._shared_vars = {}
    current_graph._shared_vars.update(vars)

    if not hasattr(current_graph, "_reducer_vars"):
        current_graph._reducer_vars = {}
    current_graph._reducer_vars.update(reducers)

    if allow_race is True:
        racy = set(vars)
    elif allow_race is False:
        racy = set()
    elif isinstance(allow_race, (list, tuple, set, frozenset)):
        racy = set(allow_race)
        unknown = racy - set(vars)
        if unknown:
            raise ValueError(
                f"declare() allow_race names undeclared vars: {sorted(unknown)}. "
                f"declared vars: {sorted(vars)}"
            )
    else:
        raise TypeError(
            f"declare() allow_race takes True or a list of var names, got {allow_race!r}"
        )
    if not hasattr(current_graph, "_race_vars"):
        current_graph._race_vars = set()
    current_graph._race_vars.update(racy)

ScratchAccessor

Dict-like accessor for per-call scratch space.

• Inside an op body (ContextVar bound): reads/writes the live scratch dict on the current MemoryState.
• At graph-construction time (ContextVar unbound): __getitem__ returns a ScratchRef marker, post-resolved by BaseOp.get_inputs(). __setitem__ raises — write-outside-run is a programming error.

Methods:

get

get(key: str, default: Any = None) -> Any

Read key, or default when it is absent.

SCRATCH[key] already returns None for a missing key, so this adds nothing but the idiom people reach for first. Without it, SCRATCH.get("k") raised AttributeError from inside an op body — where BaseOp.run records it into state rather than raising, so it surfaced as an op error rather than an obvious typo.

Outside a run this returns default rather than a :class:ScratchRef: a ref is a wiring marker and get() with a default reads as a value lookup, so returning one would smuggle a marker into a place expecting data.

Source code in operonx/core/ops/_edges.py

def get(self, key: str, default: Any = None) -> Any:
    """Read ``key``, or ``default`` when it is absent.

    ``SCRATCH[key]`` already returns ``None`` for a missing key, so
    this adds nothing but the idiom people reach for first. Without
    it, ``SCRATCH.get("k")`` raised ``AttributeError`` from inside an
    op body — where ``BaseOp.run`` records it into state rather than
    raising, so it surfaced as an op error rather than an obvious typo.

    Outside a run this returns ``default`` rather than a
    :class:`ScratchRef`: a ref is a *wiring* marker and ``get()`` with
    a default reads as a value lookup, so returning one would smuggle
    a marker into a place expecting data.
    """
    try:
        state = _current_state_var.get()
    except LookupError:
        return default
    return state._scratch.get(key, default)

keys

keys()

Keys currently in scratch — empty outside a run.

Source code in operonx/core/ops/_edges.py

def keys(self):
    """Keys currently in scratch — empty outside a run."""
    try:
        state = _current_state_var.get()
    except LookupError:
        return ()
    return tuple(state._scratch)

items

items()

(key, value) pairs — a snapshot, empty outside a run.

Source code in operonx/core/ops/_edges.py

def items(self):
    """``(key, value)`` pairs — a snapshot, empty outside a run."""
    try:
        state = _current_state_var.get()
    except LookupError:
        return ()
    return tuple(state._scratch.items())

SoftEdge

SoftEdge(op: BaseOp)

Marker for soft-edge connection. Use ~op syntax.

Source code in operonx/core/ops/_edges.py

def __init__(self, op: "_BaseOp"):
    self.op = op

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()

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

EmitOp

EmitOp(payload: Any = None, channel: str = 'default', **kwargs)

Bases: BaseOp

Op that emits a CustomEvent to mode="custom" subscribers.

Parameters:

Name	Type	Description	Default
payload	Any	Value (typically a Ref) to include in the emitted event.	None
channel	str	Free-form string label used by consumers for filtering. Defaults to "default".	'default'

Behaviour

• Fires once per invocation (per ctx for streaming/parallel graphs).
• Fire-and-forget: no subscriber → payload dropped, no error.
• Emits CustomEvent(step_id, op, ctx, channel, payload) via state._notify_custom — scheduler provides step_id and ctx via ContextVar plumbing (matches SCRATCH's pattern).

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

def __init__(
    self,
    payload: Any = None,
    channel: str = "default",
    **kwargs,
):
    # Payload becomes a declared input — enables ref wiring.
    kwargs.setdefault("inputs", {})
    if payload is not None and "payload" not in kwargs["inputs"]:
        kwargs["inputs"]["payload"] = payload

    super().__init__(**kwargs)
    self.channel = channel

    # Auto-declare "payload" input if missing so the schema tracks it.
    if self.inputs is None:
        self.inputs = {}
    if "payload" not in self.inputs:
        self.inputs["payload"] = Param(type=Any, required=False, default=None)

    # No outputs — fire-and-forget.
    if self.outputs is None:
        self.outputs = {}

    self._set_core(self._emit_impl)

InterruptOp

InterruptOp(payload: Any = None, timeout: float = 0, **kwargs)

Bases: BaseOp

Op that suspends until the caller posts a resume value.

Parameters:

Name	Type	Description	Default
payload	Any	Value (typically a Ref) sent to the caller via :class:~operonx.checkpoint.InterruptEvent.	None
timeout	float	Optional wall-clock seconds. Falsey (0/None) = wait forever. A run with a journal does not wait: it parks (below), and the timeout does not apply.	0

Behaviour

• Emits InterruptEvent(step_id, op, ctx, payload, interrupt_id) when dispatched. interrupt_id is :func:~operonx.core.runtime.invocation_key of (run, op, ctx): deterministic, the same as the op's run_context().idempotency_key.
• Awaits state._interrupt_responses[interrupt_id] (a dict[str, Future]) — the caller resolves this future.
• Returns {"response": <value>} — downstream refs like approve["response"] read the answer.
• Also exposes timed_out (bool) and interrupt_id (str) outputs.
• With Operon(journal=…) the run parks instead: the question is journalled, the ops in flight finish, nothing new starts, and the run stops with status interrupted — its result holds "$interrupted": [{interrupt_id, op, ctx, payload}]. await engine.resume(run_id, answers={interrupt_id: value}), in any process, continues it with value as the response.

Design note

This op piggybacks on the state's interrupt bus. engine.stream(mode="updates" | "interrupts") subscribes to it and yields each event; the event's resume(value) resolves the future this op awaits.

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

def __init__(
    self,
    payload: Any = None,
    timeout: float = 0,
    **kwargs,
):
    kwargs.setdefault("inputs", {})
    if payload is not None and "payload" not in kwargs["inputs"]:
        kwargs["inputs"]["payload"] = payload

    super().__init__(**kwargs)
    self.timeout = timeout

    if self.inputs is None:
        self.inputs = {}
    if "payload" not in self.inputs:
        self.inputs["payload"] = Param(type=Any, required=False, default=None)

    # Declared outputs — downstream ops read approve["response"] etc.
    if self.outputs is None:
        self.outputs = {}
    for out_name in ("response", "timed_out", "interrupt_id"):
        if out_name not in self.outputs:
            self.outputs[out_name] = Param(type=Any, required=False, default=None)

    self._set_core(self._interrupt_impl)

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)

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.

Functions:

shorthand

shorthand(fn)

Decorator for Op.of() classmethods.

Registers the function for auto-naming frame skip via register_skip() and wraps as classmethod.

Usage::

class MyOp(BaseOp):
    @shorthand
    def of(cls, my_param=None, **kwargs):
        inputs, init_kwargs = split_shorthand_kwargs(kwargs)
        return cls(my_param=my_param, inputs=inputs or None, **init_kwargs)

Source code in operonx/core/ops/_shortcuts.py

def shorthand(fn):
    """Decorator for ``Op.of()`` classmethods.

    Registers the function for auto-naming frame skip via ``register_skip()``
    and wraps as ``classmethod``.

    Usage::

        class MyOp(BaseOp):
            @shorthand
            def of(cls, my_param=None, **kwargs):
                inputs, init_kwargs = split_shorthand_kwargs(kwargs)
                return cls(my_param=my_param, inputs=inputs or None, **init_kwargs)
    """
    register_skip(fn)
    return classmethod(fn)

split_shorthand_kwargs

split_shorthand_kwargs(
    kwargs: dict, extra_init_keys: set = None, own: frozenset = frozenset()
) -> tuple

Split flat kwargs into (inputs, init_kwargs).

Used by shorthand functions (llm_, for_, op, etc.) to separate op constructor kwargs from input mappings.

Parameters:

Name	Type	Description	Default
kwargs	dict	Flat keyword arguments from shorthand function.	required
extra_init_keys	set	Additional op-specific init keys beyond base keys (e.g., {'max_concurrency', 'callback'} for iteration ops).	None
own	frozenset	The function's own parameter names. A keyword among them is always an input — @op def f(id) called f(id=7) gets the 7 — whatever op setting shares its name; that setting is given with f.configure(...).	frozenset()

Returns:

Type	Description
tuple	(inputs, init_kwargs) tuple where:
tuple	inputs: Dict of input variable mappings
tuple	init_kwargs: Dict of op constructor arguments

Example

Provider ops - just base keys

inputs, init_kwargs = split_shorthand_kwargs(kwargs)

Iteration ops - with extra keys

inputs, init_kwargs = split_shorthand_kwargs( kwargs, {'max_concurrency', 'until', 'callback'} )

Source code in operonx/core/ops/_shortcuts.py

def split_shorthand_kwargs(
    kwargs: dict, extra_init_keys: set = None, own: frozenset = frozenset()
) -> tuple:
    """Split flat kwargs into (inputs, init_kwargs).

    Used by shorthand functions (llm_, for_, op, etc.) to separate
    op constructor kwargs from input mappings.

    Args:
        kwargs: Flat keyword arguments from shorthand function.
        extra_init_keys: Additional op-specific init keys beyond base keys
            (e.g., {'max_concurrency', 'callback'} for iteration ops).

        own: The function's own parameter names. A keyword among them is
            always an input — ``@op def f(id)`` called ``f(id=7)`` gets
            the 7 — whatever op setting shares its name; that setting is
            given with ``f.configure(...)``.

    Returns:
        (inputs, init_kwargs) tuple where:
        - inputs: Dict of input variable mappings
        - init_kwargs: Dict of op constructor arguments

    Example:
        # Provider ops - just base keys
        inputs, init_kwargs = split_shorthand_kwargs(kwargs)

        # Iteration ops - with extra keys
        inputs, init_kwargs = split_shorthand_kwargs(
            kwargs,
            {'max_concurrency', 'until', 'callback'}
        )
    """
    init_keys = _BASE_INIT_KEYS | (extra_init_keys or set())

    inputs = {}
    init_kwargs = {}

    for key, value in kwargs.items():
        if key in own:
            inputs[key] = value
            continue
        policy_type = _POLICY_KEYS.get(key)
        if policy_type is not None and key not in init_keys:
            # By type, not by name: `fetch(url=u, timeout=10)` is the
            # function's own argument, `fetch(url=u, timeout=Timeout(run=10))`
            # is the op's deadline.
            if isinstance(value, policy_type):
                init_kwargs[key] = value
            else:
                inputs[key] = value
            continue
        if key in init_keys and not isinstance(value, Ref):
            init_kwargs[key] = value
        else:
            if key in init_keys:
                LOGGER.warning(
                    "Keyword '%s' is a reserved op parameter (one of %s) but received a Ref value. "
                    "It will be treated as an input mapping instead of an op constructor arg. "
                    "Consider renaming this parameter to avoid ambiguity.",
                    key,
                    sorted(init_keys),
                )
            inputs[key] = value

    return inputs, init_kwargs

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)

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