operonx.core

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

Engine

Operon

Operon(
    graph: Union[GraphOp, Callable[..., GraphOp]],
    *,
    params: Optional[Dict[str, Any]] = None,
    trace: Optional[Union[str, Consumer, List[Union[str, Consumer]]]] = None,
)

Workflow execution engine.

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

Attributes:

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

Example

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

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

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

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

Initialize Operon engine with a GraphOp or a graph factory.

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

Parameters:

Name	Type	Description	Default
graph	Union[GraphOp, Callable[..., GraphOp]]	A GraphOp workflow, or a callable that returns one. When a callable is passed, it is invoked with **params immediately — call :func:operonx.bootstrap first if the factory needs the hub.	required
params	Optional[Dict[str, Any]]	Keyword arguments passed to the graph factory. Ignored when graph is already a GraphOp. Defaults to {}.	None
trace	Optional[Union[str, Consumer, List[Union[str, Consumer]]]]	V3 tracing. Accepts a ResourceHub key (str), a :class:Consumer instance, or a list of either. Each consumer gets handle.trace at the end of every run and writes its own view (disk, Langfuse, report, …). Failures are caught + logged per-consumer so one bad backend never affects the call. Requires :func:operonx.bootstrap when using string keys. Examples:: trace="trace_local:default" trace=CallbotLocalConsumer(config={"root": "/tmp/x"}) trace=["trace_langfuse:edupia", MyDebugConsumer()]	None

Raises:

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

Source code in operonx/core/engine.py

def __init__(
    self,
    graph: Union[GraphOp, Callable[..., GraphOp]],
    *,
    params: Optional[Dict[str, Any]] = None,
    trace: Optional[Union[str, "Consumer", List[Union[str, "Consumer"]]]] = None,
):
    """Initialize Operon engine with a GraphOp or a graph factory.

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

    Args:
        graph: A GraphOp workflow, or a callable that returns one.
               When a callable is passed, it is invoked with ``**params``
               immediately — call :func:`operonx.bootstrap` first if the
               factory needs the hub.
        params: Keyword arguments passed to the graph factory. Ignored
                when *graph* is already a GraphOp. Defaults to ``{}``.
        trace: V3 tracing. Accepts a ResourceHub key (str),
               a :class:`Consumer` instance, or a list of either.
               Each consumer gets ``handle.trace`` at the end of
               every run and writes its own view (disk, Langfuse,
               report, …). Failures are caught + logged per-consumer
               so one bad backend never affects the call. Requires
               :func:`operonx.bootstrap` when using string keys.
               Examples::

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

    Raises:
        RuntimeError: If a provider op needs the hub but none has been
            installed. The message points at ``operonx.bootstrap()``.
        TypeError: If a ``trace=`` item is neither a str nor a
            Consumer instance.
    """
    if callable(graph) and not isinstance(graph, GraphOp):
        graph = graph(**(params or {}))

    self.graph = graph
    self.name = graph.name
    self._trace_consumers = self._resolve_trace_consumers(trace)

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

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

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

Attributes

schema property

schema: StateSchema

Access the workflow state schema.

Methods:

start

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

Start workflow execution and return a streaming handle immediately.

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

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

Parameters:

Name	Type	Description	Default
inputs	Dict[str, Any]	Input data for the workflow	required
user_id	Optional[str]	Optional user identifier (auto-generated if not provided)	None
session_id	Optional[str]	Optional session identifier (auto-generated if not provided)	None
request_id	Optional[str]	Optional request identifier (auto-generated if not provided)	None
scratch	Optional[Dict[str, Any]]	Optional initial values for per-call scratch space. Applied synchronously before the scheduler task is created — race-free. Equivalent to writing handle.scratch[k] = v before the first await after start(), but guaranteed to be visible to entry ops.	None

Returns:

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

Source code in operonx/core/engine.py

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

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

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

    Args:
        inputs: Input data for the workflow
        user_id: Optional user identifier (auto-generated if not provided)
        session_id: Optional session identifier (auto-generated if not provided)
        request_id: Optional request identifier (auto-generated if not provided)
        scratch: Optional initial values for per-call scratch space. Applied
            synchronously before the scheduler task is created — race-free.
            Equivalent to writing ``handle.scratch[k] = v`` before the first
            ``await`` after ``start()``, but guaranteed to be visible to
            entry ops.

    Returns:
        ExecutionHandle — async-iterable, supports ``await handle["op","var"]``
        and ``await handle.collect()``
    """
    user_id = user_id or str(uuid.uuid4())
    session_id = session_id or str(uuid.uuid4())
    request_id = request_id or str(uuid.uuid4())

    state = self._schema.create_state(
        inputs=inputs,
        user_id=user_id,
        session_id=session_id,
        request_id=request_id,
    )
    # Legacy ``state.tracing`` flag retained for back-compat with code
    # that reads it (e.g. inside op.run() for per-op metric writes).
    # No longer load-bearing for trace dispatch.
    state.tracing = False

    # Seed scratch synchronously before the scheduler task is created.
    if scratch:
        state._scratch.update(scratch)

    # Phase 2: wire the checkpointer to the state's write funnel BEFORE
    # the scheduler runs, so no writes are missed. Unsubscribe hook is
    # invoked in the run's finally-block to detach cleanly.
    _cp_unsubscribe = None
    if checkpointer is not None:
        from operonx.checkpoint.bridge import bind_checkpointer

        _cp_unsubscribe = bind_checkpointer(
            state,
            checkpointer,
            op_registry=self._all_ops_registry(),
        )

    # ``@op(observe_max=N)`` is enforced on every run, checkpointer or
    # not. It used to be counted inside bind_checkpointer's closure,
    # which made the circuit breaker a no-op under plain ``run()``.
    # Binds nothing (and returns None) when no op declares a budget.
    from operonx.checkpoint.bridge import bind_observe_budget

    _budget_unsubscribe = bind_observe_budget(state, self._all_ops_registry())

    LOGGER.info(format_event("workflow_start", request_id=request_id, graph_name=self.name))

    graph_name = self.name
    queue: asyncio.Queue = asyncio.Queue()

    # V3 tracing: per-run WorkflowTrace buffer, ContextVar-scoped.
    # Always created — consumers read `handle.trace` after the run.
    # Ops append `OpExecution` records automatically via the
    # `BaseOp.run()` recording hook — no author code required.
    from operonx.core.workflow_trace import WorkflowTrace
    from operonx.core.workflow_trace import _current_trace as _v3_trace_var

    _wf_trace = WorkflowTrace(
        trace_id=trace_id or request_id,
        workflow_name=self.name,
        started_at=perf_counter(),
        ended_at=0.0,
        metadata={
            "request_id": request_id,
            "user_id": user_id,
            "session_id": session_id,
            **({"tags": list(state.tags)} if getattr(state, "tags", None) else {}),
        },
    )

    async def _run() -> None:
        v3_token = _v3_trace_var.set(_wf_trace)
        try:
            await self.graph._scheduler.run(state, ("main",), output_queue=queue)
        except asyncio.CancelledError:
            # Phase 2b3 T5: notify the checkpointer of run-level cancel so
            # audit trails and speculative-chain teardown observers hear it.
            if checkpointer is not None:
                try:
                    checkpointer.on_cancel(("main",))
                except Exception:
                    LOGGER.exception("checkpointer.on_cancel failed")
            queue.put_nowait(None)
            raise
        except BaseException as e:  # includes ObserveBudgetExceeded (Phase 2)
            queue.put_nowait(e)
            # Do NOT re-raise a BaseException — the ExecutionHandle re-raises
            # it to the caller via _pump when the value is dequeued. Bubbling
            # here would surface as an unhandled task exception.
        finally:
            _v3_trace_var.reset(v3_token)
            _wf_trace.ended_at = perf_counter()
            # Detach any Phase 2 observers bound at start().
            if _cp_unsubscribe is not None:
                try:
                    _cp_unsubscribe()
                except Exception:
                    LOGGER.exception("checkpointer unsubscribe failed")
            if _budget_unsubscribe is not None:
                try:
                    _budget_unsubscribe()
                except Exception:
                    LOGGER.exception("observe budget unsubscribe failed")
            # Phase 2b3 H3: drain any pending InterruptOp futures so the
            # state's response bus doesn't leak entries after the run.
            if state._interrupt_responses:
                for _iid, _fut in list(state._interrupt_responses.items()):
                    if not _fut.done():
                        _fut.cancel()
                state._interrupt_responses.clear()
            # V3 consumers — auto-invoke on the completed trace.
            # `asyncio.to_thread` so a slow HTTP consumer (Langfuse)
            # doesn't block the event loop; per-consumer try/except
            # so one broken backend never affects the call.
            for _consumer in self._trace_consumers:
                try:
                    await asyncio.to_thread(_consumer.consume, _wf_trace)
                except Exception:
                    LOGGER.exception(
                        "trace consumer %r failed on trace %s",
                        type(_consumer).__name__,
                        _wf_trace.trace_id,
                    )
            LOGGER.info(
                format_event("workflow_done", request_id=request_id, graph_name=graph_name)
            )

    scheduler_task = asyncio.create_task(_run())
    return ExecutionHandle(queue, scheduler_task, state, trace=_wf_trace)

run async

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

Execute the workflow with given inputs.

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

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

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

Parameters:

Name	Type	Description	Default
inputs	Dict[str, Any]	Input data for the workflow	required
user_id	Optional[str]	Optional user identifier (auto-generated if not provided)	None
session_id	Optional[str]	Optional session identifier (auto-generated if not provided)	None
request_id	Optional[str]	Optional request identifier (auto-generated if not provided)	None
scratch	Optional[Dict[str, Any]]	Optional initial values for per-call scratch space.	None

Returns:

Type	Description
Dict[str, Any]	Dictionary containing workflow outputs plus "$state" key
Dict[str, Any]	with the MemoryState for debugging/tracing access.

Source code in operonx/core/engine.py

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

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

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

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

    Args:
        inputs: Input data for the workflow
        user_id: Optional user identifier (auto-generated if not provided)
        session_id: Optional session identifier (auto-generated if not provided)
        request_id: Optional request identifier (auto-generated if not provided)
        scratch: Optional initial values for per-call scratch space.

    Returns:
        Dictionary containing workflow outputs plus "$state" key
        with the MemoryState for debugging/tracing access.
    """
    user_id = user_id or str(uuid.uuid4())
    session_id = session_id or str(uuid.uuid4())
    request_id = request_id or str(uuid.uuid4())

    handle = self.start(
        inputs,
        user_id=user_id,
        session_id=session_id,
        request_id=request_id,
        trace_id=trace_id,
        scratch=scratch,
        checkpointer=checkpointer,
    )

    result = await handle.collect(unwrap=True)
    result["$state"] = handle.state

    return result

invoke async

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

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

Source code in operonx/core/engine.py

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

stream async

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

LangGraph-familiar streaming iterator.

Parameters:

Name	Type	Description	Default
inputs	Dict[str, Any]	workflow inputs (same as run/invoke)	required
mode	str	one of - "updates" — yields {op_name: {var: value, ...}} per op completion (matches LangGraph's stream_mode="updates"). Covers every op, including generators in the middle of the graph, and delivers each write as it lands. - "values" — yields the full state snapshot per step; requires a checkpointer (auto-created in-memory if omitted) - "frames" — yields (op, ctx, data) for ops that write a graph output (PARENT- or END-bound). An op feeding only a downstream consumer emits nothing here, whatever it yields — use "updates" to watch those. - "custom" — yields :class:~operonx.checkpoint.CustomEvent emitted by any :class:~operonx.EmitOp, optionally filtered by channels=[...]	'updates'
channels	Optional[List[str]]	for mode="custom" only — restrict to these channel names	None
checkpointer	Optional[Any]	for mode="values"; auto-created InMemory if None	None
**kwargs	Any	forwarded to start() (user_id, session_id, etc.)	{}

Yields:

Type	Description
AsyncGenerator[Any, None]	mode-specific chunks (see above).

Source code in operonx/core/engine.py

async def stream(  # noqa: C901 — routing on mode; keeps engine surface compact
    self,
    inputs: Dict[str, Any],
    *,
    mode: str = "updates",
    channels: Optional[List[str]] = None,
    checkpointer: Optional[Any] = None,
    **kwargs: Any,
) -> "asyncio.AsyncGenerator[Any, None]":
    """LangGraph-familiar streaming iterator.

    Args:
        inputs: workflow inputs (same as ``run``/``invoke``)
        mode: one of
            - ``"updates"`` — yields ``{op_name: {var: value, ...}}`` per op
              completion (matches LangGraph's ``stream_mode="updates"``).
              Covers **every** op, including generators in the middle of
              the graph, and delivers each write as it lands.
            - ``"values"`` — yields the full state snapshot per step;
              requires a checkpointer (auto-created in-memory if omitted)
            - ``"frames"`` — yields ``(op, ctx, data)`` for ops that
              write a graph **output** (PARENT- or END-bound). An op
              feeding only a downstream consumer emits nothing here,
              whatever it yields — use ``"updates"`` to watch those.
            - ``"custom"`` — yields :class:`~operonx.checkpoint.CustomEvent`
              emitted by any :class:`~operonx.EmitOp`, optionally filtered
              by ``channels=[...]``
        channels: for ``mode="custom"`` only — restrict to these channel names
        checkpointer: for ``mode="values"``; auto-created InMemory if None
        **kwargs: forwarded to ``start()`` (user_id, session_id, etc.)

    Yields:
        mode-specific chunks (see above).
    """
    import asyncio

    from operonx.checkpoint import InMemoryCheckpointer
    from operonx.checkpoint.base import CustomEvent
    from operonx.checkpoint.bridge import bind_custom_bus

    # Only "values" mode strictly needs a checkpointer. Create an in-memory
    # one on demand so callers don't have to plumb it manually.
    if mode == "values" and checkpointer is None:
        checkpointer = InMemoryCheckpointer()

    # For mode="custom", subscribe to the state's custom bus and buffer
    # events into a local queue. Handle is used for its scheduler task.
    if mode == "custom":
        queue: asyncio.Queue = asyncio.Queue()

        def _sink(evt: CustomEvent):
            if channels is None or evt.channel in channels:
                queue.put_nowait(evt)

        handle = self.start(inputs, checkpointer=checkpointer, **kwargs)
        op_registry = self._all_ops_registry()
        _unbind = bind_custom_bus(handle.state, _sink, op_registry=op_registry)
        drainer = None
        getter = None
        try:
            # Drain the frame queue in the background so scheduler can
            # progress; we ignore frames here — only custom events.
            async def _drain_frames():
                async for _ in handle:
                    pass

            drainer = asyncio.create_task(_drain_frames())
            while not (drainer.done() and queue.empty()):
                getter = asyncio.create_task(queue.get())
                done, _pending = await asyncio.wait(
                    {getter, drainer},
                    return_when=asyncio.FIRST_COMPLETED,
                )
                if getter in done:
                    yield getter.result()
                else:
                    getter.cancel()
                    # Drain any remaining events after scheduler done.
                    while not queue.empty():
                        yield queue.get_nowait()
                    break
        finally:
            _unbind()
            # Phase 2b3 B3: cancel the scheduler on caller break/error so
            # long-running ops (LLM calls, DB writes) don't keep burning
            # resources with no consumer.
            handle.cancel()
            if drainer and not drainer.done():
                drainer.cancel()
            if getter and not getter.done():
                getter.cancel()
        return

    # For "updates" and "values" we need per-op-completion granularity,
    # not just the scheduler's output-frame stream (which only fires for
    # ops that push to PARENT / END). We piggy-back on the state's write
    # bus so every op invocation yields exactly once.
    if mode in ("updates", "values"):
        if mode == "values" and checkpointer is None:
            checkpointer = InMemoryCheckpointer()
        handle = self.start(inputs, checkpointer=checkpointer, **kwargs)
        state = handle.state

        # Reverse index built once.
        idx_to_key = {
            idx: (op_name, var) for (op_name, var), idx in state.schema._var_to_idx.items()
        }

        # Buffer per-step writes into a list of updates. Each element
        # in step_updates is {op_name: {var: value, ...}} for one step.
        step_updates: Dict[int, Dict[str, Dict[str, Any]]] = {}

        def _record(idx: int, ctx_key: tuple, value):
            op_name, var = idx_to_key.get(idx, ("?", "?"))
            step = state._current_step
            step_updates.setdefault(step, {}).setdefault(op_name, {})[var] = value

        # A write signals the pacer. This loop used to be driven by
        # ``async for _ in handle``, which only ticks on *output*
        # frames — so a graph whose single output lands at the end
        # buffered every intermediate update and released them all at
        # once. Measured: four generator yields 150ms apart, all
        # delivered together after the run. Streaming an LLM into a
        # consumer is exactly that shape.
        signal: asyncio.Queue = asyncio.Queue()

        def _record(idx: int, ctx_key: tuple, value):
            op_name, var = idx_to_key.get(idx, ("?", "?"))
            step = state._current_step
            step_updates.setdefault(step, {}).setdefault(op_name, {})[var] = value
            signal.put_nowait(step)

        def _flush(upto: int, last: int):
            """Yield completed steps in ``(last, upto]``; return the new last."""
            out = []
            while last < upto:
                last += 1
                if mode == "updates":
                    batch = step_updates.pop(last, {})
                    if batch:
                        out.append(batch)
                elif checkpointer is not None:
                    try:
                        out.append(checkpointer.get_state(last))
                    except Exception:
                        pass
            return out, last

        state.subscribe_writes(_record)
        drainer = None
        getter = None
        try:
            # The scheduler still needs its frames consumed to make
            # progress; that just isn't the pacer any more.
            async def _drain_frames():
                async for _ in handle:
                    pass

            drainer = asyncio.create_task(_drain_frames())
            last_yielded = -1
            while not (drainer.done() and signal.empty()):
                getter = asyncio.create_task(signal.get())
                done, _pending = await asyncio.wait(
                    {getter, drainer},
                    return_when=asyncio.FIRST_COMPLETED,
                )
                if getter not in done:
                    getter.cancel()
                # A step is only complete once the counter has moved
                # past it, so flush up to current - 1 while running.
                batches, last_yielded = _flush(state._current_step - 1, last_yielded)
                for batch in batches:
                    yield batch
                while not signal.empty():
                    signal.get_nowait()
            # The run is over, so the last step is complete too.
            batches, last_yielded = _flush(state._current_step, last_yielded)
            for batch in batches:
                yield batch
        finally:
            state.unsubscribe_writes(_record)
            # Phase 2b3 B3: cancel scheduler on caller break so a partial
            # consume doesn't leave the graph running with no listener.
            handle.cancel()
            if drainer is not None and not drainer.done():
                drainer.cancel()
            if getter is not None and not getter.done():
                getter.cancel()
        return

    handle = self.start(inputs, checkpointer=checkpointer, **kwargs)

    if mode == "frames":
        try:
            async for frame in handle:
                yield frame
        finally:
            # Phase 2b3 B3: cancel on caller break/error.
            handle.cancel()
        return

    raise ValueError(
        f"engine.stream(mode={mode!r}) — valid modes: 'updates', 'values', 'frames', 'custom'"
    )

serve

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

Serve this workflow as an HTTP API.

Convenience wrapper around operonx.serve.OperonApp. Requires operonx-serve to be installed.

Parameters:

Name	Type	Description	Default
path	str	URL path for the endpoint (default: "/").	'/'
host	str	Bind address.	'0.0.0.0'
port	int	Bind port.	8000
stream	Optional[bool]	Enable SSE streaming endpoint. None = auto-detect.	None
websocket	bool	Enable WebSocket endpoint.	False
backend	str	"python" (FastAPI/uvicorn) or "rust" (Axum).	'python'
**kwargs	Any	Extra arguments forwarded to OperonApp.serve().	{}

Source code in operonx/core/engine.py

def serve(
    self,
    *,
    path: str = "/",
    host: str = "0.0.0.0",
    port: int = 8000,
    stream: Optional[bool] = None,
    websocket: bool = False,
    backend: str = "python",
    **kwargs: Any,
) -> None:
    """Serve this workflow as an HTTP API.

    Convenience wrapper around ``operonx.serve.OperonApp``. Requires operonx-serve
    to be installed.

    Args:
        path: URL path for the endpoint (default: "/").
        host: Bind address.
        port: Bind port.
        stream: Enable SSE streaming endpoint. None = auto-detect.
        websocket: Enable WebSocket endpoint.
        backend: "python" (FastAPI/uvicorn) or "rust" (Axum).
        **kwargs: Extra arguments forwarded to ``OperonApp.serve()``.
    """
    try:
        from operonx.serve import OperonApp
    except ImportError:
        raise ImportError(
            "operonx-serve is required for engine.serve(). Install it with: pip install operonx-serve"
        ) from None

    app = OperonApp()
    app.endpoint(path, graph=self.graph, stream=stream, websocket=websocket)
    app.serve(host=host, port=port, backend=backend, **kwargs)

batch async

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

Run the workflow concurrently on multiple inputs.

Parameters:

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

Returns:

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

Source code in operonx/core/engine.py

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

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

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

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

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

cli

cli() -> None

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

Source code in operonx/core/engine.py

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

input_schema

input_schema() -> Dict[str, Any]

Return JSON Schema describing the workflow's expected inputs.

Source code in operonx/core/engine.py

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

output_schema

output_schema() -> Dict[str, Any]

Return JSON Schema describing the workflow's outputs.

Source code in operonx/core/engine.py

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

show

show() -> None

Display workflow structure for debugging.

Source code in operonx/core/engine.py

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

Decorators

The two decorators that turn ordinary Python into Operonx ops:

op

op(
    func: Optional[Callable] = None,
    *,
    bound: Optional[str] = None,
    cache: Optional[Any] = None,
    delay: float = 0,
    exclude: Optional[Any] = None,
    include: Optional[Any] = None,
    observe_max: Optional[int] = None,
) -> 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
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

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,
) -> 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.
        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.
    """
    # 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)
        collisions = set(sig.parameters.keys()) & _BASE_INIT_KEYS
        if collisions:
            LOGGER.warning(
                "@op function '%s' has parameter(s) %s that collide with reserved op keywords %s. "
                "When called via shorthand (e.g. %s(name=PARENT['name'])), these may be misinterpreted "
                "as op constructor args instead of function inputs. Consider renaming them.",
                fn.__name__,
                sorted(collisions),
                sorted(_BASE_INIT_KEYS),
                fn.__name__,
            )

        @wraps(fn)
        def wrapper(**kwargs):
            mappings, init_kwargs = split_shorthand_kwargs(kwargs, {"return_keys"})
            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)
            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,
                _mappings=mappings or None,
                **init_kwargs,
            )

        register_skip(wrapper)
        wrapper.__wrapped__ = fn
        return wrapper

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

graph

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

Classes

GraphOp

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

Bases: BaseOp

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

Lifecycle::

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

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

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

4. EXPORT        serialize()  config dict for Rust backend
                 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)
    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=...)
    self._adj = {}  # {op_name: [Link(dst, soft), ...]}
    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
    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
    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 rewrite_cycles_to_loops

    rewrite_cycles_to_loops(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.
    if self.bound is None:
        if all(getattr(op, "bound", None) == "sync" for op in self._ops.values()):
            self.bound = "sync"
        else:
            self.bound = "io"

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."""

    if context_id is None:
        context_id = DEFAULT_CONTEXT

    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)

        if self._is_building:
            self.build()

        _outputs, stream_ctxs, _interrupted = await self._scheduler.run(state, context_id)

        _has_generators = any(op.is_gen for op in self._ops.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 stream_ctxs and not _has_generators:
            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 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(),
        )

    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

Serialize full graph to config dict for the Rust backend.

Note: the key "initial_ready_count" is kept as-is for Rust backend compatibility even though the internal Python attribute was renamed to _initial_ready during the scheduler rewrite.

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

def serialize(self) -> dict:
    """Serialize full graph to config dict for the Rust backend.

    Note: the key ``"initial_ready_count"`` is kept as-is for Rust backend
    compatibility even though the internal Python attribute was renamed to
    ``_initial_ready`` during the scheduler rewrite.
    """
    # 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()},
            "edges": [
                {"from": src, "to": dst, "soft": edge.soft}
                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,
            "loop_config": {
                "until": self._loop_config.until
                if isinstance(self._loop_config.until, str)
                else None,
                "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,
    )

show

show(indent=0)

Display graph structure (debug).

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

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

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

GraphValidationError

GraphValidationError(result: ValidationResult)

Bases: Exception

Exception raised when graph validation fails.

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

def __init__(self, result: ValidationResult):
    self.result = result
    super().__init__(
        f"Graph '{result.graph_name}' validation failed with "
        f"{len(result.errors)} error(s). See logs above for details."
    )

ValidationIssue dataclass

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

A single validation issue found in the graph.

ValidationLevel

Bases: Enum

Severity level for validation issues.

ValidationResult dataclass

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

Result of graph validation.

Methods:

raise_if_errors

raise_if_errors()

Raise exception if there are any errors.

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

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

Op types

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

GraphOp

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

Bases: BaseOp

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

Lifecycle::

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

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

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

4. EXPORT        serialize()  config dict for Rust backend
                 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)
    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=...)
    self._adj = {}  # {op_name: [Link(dst, soft), ...]}
    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
    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
    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 rewrite_cycles_to_loops

    rewrite_cycles_to_loops(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.
    if self.bound is None:
        if all(getattr(op, "bound", None) == "sync" for op in self._ops.values()):
            self.bound = "sync"
        else:
            self.bound = "io"

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."""

    if context_id is None:
        context_id = DEFAULT_CONTEXT

    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)

        if self._is_building:
            self.build()

        _outputs, stream_ctxs, _interrupted = await self._scheduler.run(state, context_id)

        _has_generators = any(op.is_gen for op in self._ops.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 stream_ctxs and not _has_generators:
            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 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(),
        )

    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

Serialize full graph to config dict for the Rust backend.

Note: the key "initial_ready_count" is kept as-is for Rust backend compatibility even though the internal Python attribute was renamed to _initial_ready during the scheduler rewrite.

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

def serialize(self) -> dict:
    """Serialize full graph to config dict for the Rust backend.

    Note: the key ``"initial_ready_count"`` is kept as-is for Rust backend
    compatibility even though the internal Python attribute was renamed to
    ``_initial_ready`` during the scheduler rewrite.
    """
    # 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()},
            "edges": [
                {"from": src, "to": dst, "soft": edge.soft}
                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,
            "loop_config": {
                "until": self._loop_config.until
                if isinstance(self._loop_config.until, str)
                else None,
                "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,
    )

show

show(indent=0)

Display graph structure (debug).

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

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

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

BranchOp

BranchOp(
    cases: Optional[List[Tuple[Ref, str]]] = None,
    candidates: Optional[List[str]] = None,
    default: Optional[str] = None,
    inputs: Dict[str, Any] = None,
    outputs: Dict[str, Any] = None,
    **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,
    **kwargs,
):
    # Parse inputs/outputs from cases
    parsed_inputs, parsed_outputs = 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]

    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 conditions for Rust backend.

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

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

FuncOp

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

Bases: BaseOp

Op that executes a Python function.

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

Inputs

Auto-parsed from the function's parameter list.

Outputs

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

Example::

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

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

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

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

    # 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._set_core(code_fn)

    # Lấy source code
    try:
        self.source = inspect.getsource(code_fn) if code_fn else ""
    except:
        self.source = str(code_fn) 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.

Methods:

run async

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

Execute FuncOp with CodeError wrapping.

Delegates to BaseOp.run() (async generator) and re-raises any exception as a CodeError with full op context attached.

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

async def run(
    self,
    state: "MemoryState",
    context_id: Optional[str] = None,
) -> AsyncGenerator[Tuple[Optional[str], Dict[str, Any]], None]:
    """Execute FuncOp with CodeError wrapping.

    Delegates to ``BaseOp.run()`` (async generator) and re-raises any
    exception as a ``CodeError`` with full op context attached.
    """
    try:
        async for ctx, result in super().run(state, context_id):
            yield ctx, result
    except CodeError:
        raise  # Đã wrapped, không wrap lại
    except Exception as e:
        # Lấy inputs để có context cho error
        _inputs = self.get_inputs(state, context_id)
        raise CodeError(
            message=f"Function '{self.code_fn.__name__ if self.code_fn else 'unknown'}' raised an exception",
            function_name=self.code_fn.__name__ if self.code_fn else "unknown",
            source=self.source,
            inputs=_inputs,
            original_error=e,
        ) from e

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

Branch helpers

Branch

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

Fluent builder for creating a BranchOp.

Two usage flavors — both supported, both idiomatic:

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

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

Auto-name resolves to the LHS if there is one (stt_route = if_(...)) or falls back to a semantic "branch_<target>_or_<default>" name.

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: Ref, target: Union[str, BaseOp]) -> Branch

Add a condition–target case.

Parameters:

Name	Type	Description	Default
condition	Ref	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: Ref, target: Union[str, BaseOp]) -> "Branch":
    """Add a condition–target case.

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

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

else_

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

Set default target and build the BranchOp.

Parameters:

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

Returns:

Type	Description
BranchOp	The constructed BranchOp.

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

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

    Args:
        target: Fallback op instance or name string.

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

build

build() -> BranchOp

Build the BranchOp without a default target.

Returns:

Type	Description
BranchOp	The constructed BranchOp.

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

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

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

if_

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

Start a branch declaration with the first condition.

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

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

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

    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
    """
    return Branch().if_(condition, target)

State markers

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

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

Top-level convenience

bootstrap

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

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

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

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

Source code in operonx/__init__.py

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

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

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

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

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

    return ResourceHub.auto()

Provider-neutral types

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

ChatMessage

Bases: TypedDict

A single message in a chat conversation.

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

Required fields

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

Optional fields

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

ChatRole module-attribute

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