operonx.app

The application layer: what puts work into an Operon. Services listen (Service), jobs loop over items (Job), and Application declares both. Graphs never import this package; the dependency runs one way.

Guides: Deployment, Jobs, The playground bridge, Evals.

Application

Application

Application(
    manifest: Union[Manifest, str],
    *,
    services: Sequence[ServeSpec] = (),
    jobs: Sequence[Any] = (),
    on_startup: Sequence[Any] = (),
    root: Union[str, Path, None] = None,
    src: Sequence[str] = ("src", "."),
    resources: Optional[str] = "resources.yaml",
    description: str = "",
    trace: Optional[Sequence[Any]] = None,
)

See the module docstring.

Source code in operonx/app/application.py

def __init__(
    self,
    manifest: Union[Manifest, str],
    *,
    services: Sequence[ServeSpec] = (),
    jobs: Sequence[Any] = (),
    on_startup: Sequence[Any] = (),
    root: Union[str, Path, None] = None,
    src: Sequence[str] = ("src", "."),
    resources: Optional[str] = "resources.yaml",
    description: str = "",
    trace: Optional[Sequence[Any]] = None,
):
    if isinstance(manifest, Manifest):
        self.manifest = manifest
        self._jobs: Dict[str, Any] = {}
    else:
        here = Path(inspect.stack()[1].filename).resolve().parent
        base = Path(root).resolve() if root else project_root(here)
        self.manifest = manifest_from(
            manifest,
            services=services,
            jobs=jobs,
            on_startup=on_startup,
            root=base,
            src=src,
            resources=resources,
            description=description,
            trace=trace,
        )
        self._jobs = {j.name: j for j in jobs}
    self._bootstrapped = False

Attributes

graphs property

graphs: List[GraphRef]

Every graph the manifest names, once each, with who uses it; a door with variants contributes one graph per variant.

jobs property

jobs: List[Any]

The declared jobs, tracing where the application says, each recording under the project's jobs folder unless told otherwise.

Methods:

find classmethod

find(start: Union[str, Path] = '.') -> 'Application'

The nearest operonx.toml at or above start — or, when that file says [project] app = "module:APP", the object it points at.

Source code in operonx/app/application.py

@classmethod
def find(cls, start: Union[str, Path] = ".") -> "Application":
    """The nearest ``operonx.toml`` at or above *start* — or, when that
    file says ``[project] app = "module:APP"``, the object it points at."""
    return cls._from(Manifest.find(start))

bootstrap

bootstrap() -> None

Make the project importable and its resources resolvable. Once.

Source code in operonx/app/application.py

def bootstrap(self) -> None:
    """Make the project importable and its resources resolvable. Once."""
    if self._bootstrapped:
        return
    for src in reversed(self.manifest.src):
        path = str((self.root / src).resolve())
        if path not in sys.path:
            sys.path.insert(0, path)
    overlay = self.manifest.resources_overlay
    if overlay and (self.root / overlay).exists():
        import operonx

        operonx.bootstrap(resources=self.root / overlay)
    stamp_process(self.root, self.name)  # every run: project + code version
    self._bootstrapped = True

serve

serve(only: Optional[Sequence[str]] = None, **bind: Any) -> None

Boot every listener the manifest declares, and block. host= / port= bind elsewhere than declared (port: one listener only).

Source code in operonx/app/application.py

def serve(self, only: Optional[Sequence[str]] = None, **bind: Any) -> None:
    """Boot every listener the manifest declares, and block. ``host=`` /
    ``port=`` bind elsewhere than declared (``port``: one listener only)."""
    from .serve.app import serve_manifest

    self.bootstrap()
    self.jobs  # settled (tracing, record folder) before a clock runs one
    serve_manifest(self.manifest, only=list(only) if only else None, **bind)

asgi

asgi(
    *,
    port: Optional[int] = None,
    only: Optional[Sequence[str]] = None,
    startup: bool = True,
) -> Any

One listener's ASGI app — for a process that runs uvicorn itself, or a test client. By port, or the only one. startup=False leaves the on_startup hooks — the application's and the services' — to another process.

Source code in operonx/app/application.py

def asgi(
    self,
    *,
    port: Optional[int] = None,
    only: Optional[Sequence[str]] = None,
    startup: bool = True,
) -> Any:
    """One listener's ASGI app — for a process that runs uvicorn itself, or
    a test client. By *port*, or the only one. ``startup=False`` leaves the
    ``on_startup`` hooks — the application's and the services' — to
    another process."""
    from .serve.app import build_app

    self.bootstrap()
    self.jobs  # settled (tracing, record folder) before a clock runs one
    specs = [s for s in self.services if s.kind != "asgi" or s.app]
    if only:
        specs = [s for s in specs if s.name in set(only)]
    if port is not None:
        specs = [s for s in specs if s.port == port]
    listeners = {s.listener for s in specs}
    if len(listeners) != 1:
        raise ManifestError(
            f"asgi() needs exactly one listener; {self.name} has "
            f"{sorted(listeners) or 'none'} — pass port= or only="
        )
    return build_app(tuple(specs), on_startup=self.manifest.on_startup, startup=startup)

run async

run(name: str, *, resume: bool = False) -> Any

Run a job once and return its record.

Source code in operonx/app/application.py

async def run(self, name: str, *, resume: bool = False) -> Any:
    """Run a job once and return its record."""
    check_sinks(self.name, jobs=[self.job(name)])  # before the first item, not in it
    return await self.job(name).run(resume=resume)

describe

describe() -> Dict[str, Any]

The three lists as plain data: what --list and the studio read. Names of things only; a TOML application is described without importing the project.

Source code in operonx/app/application.py

def describe(self) -> Dict[str, Any]:
    """The three lists as plain data: what ``--list`` and the studio read.
    Names of things only; a TOML application is described without
    importing the project."""
    jobs = [describe_job(j, self.manifest) for j in self._jobs.values()]
    return {
        "name": self.name,
        "root": str(self.root),
        "graphs": [
            {
                "name": g.name,
                "entry": g.entry,
                "used_by": list(g.used_by),
                "bind": {k: ref_name(v) for k, v in g.bind.items()},
            }
            for g in self.graphs
        ],
        "services": [describe_service(s) for s in self.services],
        "jobs": jobs,
    }

GraphRef dataclass

GraphRef(
    name: str,
    entry: str,
    used_by: Tuple[str, ...] = tuple(),
    bind: Dict[str, Any] = dict(),
    graph: Any = None,
)

A graph the application knows: by name, by entry point, and by who uses it. Compiled on demand, never at load.

Methods:

compile

compile(**kwargs: Any) -> Any

An Operon for this graph, the way a served graph is compiled: every unbound parameter becomes a runtime input.

Source code in operonx/app/application.py

def compile(self, **kwargs: Any) -> Any:
    """An ``Operon`` for this graph, the way a served graph is compiled:
    every unbound parameter becomes a runtime input."""
    from .serve.app import compile_graph

    target = self.graph if self.graph is not None else self.entry
    return compile_graph(target, bind=self.bind or None, where=f"graph {self.name!r}", **kwargs)

Declaring it in Python

Service

Service(
    name: str,
    listener: Listener,
    *,
    graph: Any = None,
    app: Any = None,
    session: Optional[str] = None,
    max_inflight: Optional[int] = None,
    concurrency: Optional[int] = None,
    trace: Optional[Sequence[str]] = None,
    variants: Optional[Mapping[str, Mapping[str, Any]]] = None,
    on_session: Any = None,
    on_close: Any = None,
    on_startup: Sequence[Any] = (),
    resume: Any = None,
    description: str = "",
    key_ops: Optional[Sequence[str]] = None,
    playground: Any = None,
    replay: bool = False,
    input: Optional[str] = None,
    **options: Any,
) -> ServeSpec

One endpoint, as :class:ServeSpec — the same record [[serve]] parses to, with objects where the manifest has module:attr.

graph is a module-level @graph (variants bind its parameters), app an ASGI app for an asgi listener. on_startup hooks run before the listener accepts, in each of its worker processes and nowhere else — a model warmed for the call workers is not warmed again for an admin port. The door's inputs are the graph's own runtime parameters; the door ops declare themselves with @op(door=...).

key_ops names the ops whose latency the team watches first (time to first audio, the LLM call) — a dashboard pins them above the rest, the way show_keys names the outputs that stand for an op.

playground is the codec the studio's playground drives this door with (:class:operonx.app.play.Codec, or "module:attr") — needed only when the door's protocol is its own; http and websocket doors have built-in ones.

resume is the graph that continues a run this http door left waiting for a human (an agent's approval): the door also answers POST <path>/resume with it, with the door's own codec, hooks and streaming. A websocket door resumes on its connection and takes none.

input="payload" hands a graph without doors the whole request body as that one parameter, instead of filling parameters by the body's fields: for a payload someone else shapes (a mail server's webhook).

replay=True records what clients send this door — text and JSON as they are, audio and bytes only counted — on each run, so a real session can be replayed in the playground later. Off by default: a script holds what users typed, and lives as long as the run.

Source code in operonx/app/declare.py

def Service(  # noqa: N802 — reads as a declaration
    name: str,
    listener: Listener,
    *,
    graph: Any = None,
    app: Any = None,
    session: Optional[str] = None,
    max_inflight: Optional[int] = None,
    concurrency: Optional[int] = None,
    trace: Optional[Sequence[str]] = None,
    variants: Optional[Mapping[str, Mapping[str, Any]]] = None,
    on_session: Any = None,
    on_close: Any = None,
    on_startup: Sequence[Any] = (),
    resume: Any = None,
    description: str = "",
    key_ops: Optional[Sequence[str]] = None,
    playground: Any = None,
    replay: bool = False,
    input: Optional[str] = None,  # noqa: A002 — the graph parameter's name
    **options: Any,
) -> ServeSpec:
    """One endpoint, as :class:`ServeSpec` — the same record ``[[serve]]``
    parses to, with objects where the manifest has ``module:attr``.

    ``graph`` is a module-level ``@graph`` (``variants`` bind its parameters), ``app`` an
    ASGI app for an ``asgi`` listener. ``on_startup`` hooks run before the
    listener accepts, in each of its worker processes and nowhere else — a
    model warmed for the call workers is not warmed again for an admin
    port. The door's inputs are the graph's own runtime parameters; the
    door ops declare themselves with ``@op(door=...)``.

    ``key_ops`` names the ops whose latency the team watches first (time
    to first audio, the LLM call) — a dashboard pins them above the rest,
    the way ``show_keys`` names the outputs that stand for an op.

    ``playground`` is the codec the studio's playground drives this door
    with (:class:`operonx.app.play.Codec`, or ``"module:attr"``) — needed
    only when the door's protocol is its own; http and websocket doors have
    built-in ones.

    ``resume`` is the graph that continues a run this ``http`` door left
    waiting for a human (an agent's approval): the door also answers
    ``POST <path>/resume`` with it, with the door's own codec, hooks and
    streaming. A websocket door resumes on its connection and takes none.

    ``input="payload"`` hands a graph without doors the whole request body
    as that one parameter, instead of filling parameters by the body's
    fields: for a payload someone else shapes (a mail server's webhook).

    ``replay=True`` records what clients send this door — text and JSON as
    they are, audio and bytes only counted — on each run, so a real session
    can be replayed in the playground later. Off by default: a script holds
    what users typed, and lives as long as the run.
    """
    label = f"Service({name!r})"
    _refuse_moved(options, label)
    kind = listener.kind
    if kind == "asgi":
        if app is None:
            raise ManifestError(f"{label} is an asgi listener and needs app=")
        if graph is not None:
            raise ManifestError(f"{label} is an asgi listener and cannot also take a graph")
    elif graph is None:
        raise ManifestError(f"{label} needs graph=")
    elif app is not None:
        raise ManifestError(f"{label} takes a graph or an app, not both")

    session = session or _default_session(kind)
    if session not in SESSION_MODES:
        raise ManifestError(
            f"{label} has session {session!r}; expected one of {', '.join(sorted(SESSION_MODES))}"
        )
    if kind in STREAM_KINDS and max_inflight is None:
        raise ManifestError(
            f"{label} is a {kind} listener and must set max_inflight= — "
            f"a stream transport needs a bound to push back against"
        )
    if max_inflight is not None and (not isinstance(max_inflight, int) or max_inflight < 1):
        raise ManifestError(
            f"{label} has max_inflight={max_inflight!r}; expected a positive integer"
        )

    variants_out: Dict[str, Dict[str, Any]] = {}
    for v_name, bind in (variants or {}).items():
        if not isinstance(bind, Mapping):
            raise ManifestError(f"{label} variant {v_name!r} must be a mapping of parameters")
        variants_out[str(v_name)] = dict(bind)
    check_resume(resume, kind, variants_out, label)

    opts: Dict[str, Any] = {**listener.options, **dict(options)}
    if concurrency is not None:
        opts["concurrency"] = int(concurrency)
    # None inherits the application's consumers; [] says "trace nothing"
    if trace is not None:
        opts["trace"] = list(trace)
    if key_ops:
        opts["key_ops"] = [str(k) for k in key_ops]
    if playground is not None:
        opts["playground"] = playground
    if replay:
        opts["replay"] = True
    if input is not None:
        if not (isinstance(input, str) and input):
            raise ManifestError(f"{label}: input= is the name of a graph parameter")
        opts["input"] = input

    return ServeSpec(
        name=name,
        kind=kind,
        graph=graph if graph is not None else "",
        path=listener.path,
        method=listener.method,
        host=listener.host,
        port=int(listener.port),
        session=session,
        max_inflight=max_inflight,
        on_session=on_session,
        on_close=on_close,
        app=app,
        resume=resume,
        description=description,
        options=opts,
        variants=variants_out,
        workers=listener.workers,
        on_startup=tuple(on_startup),
        trace_own=tuple(trace) if trace is not None else None,
        trace_from="service" if trace is not None else "default",
    )

websocket

websocket(
    path: str,
    port: int = 8000,
    host: str = "0.0.0.0",
    workers: int = 1,
    codec: Optional[str] = None,
) -> Listener

A websocket door: each text frame is one ingress item, decoded by codec — "json" (the default) or "text". Bytes frames are bytes. A frame the codec cannot read is answered with an error frame and does not reach the graph.

Source code in operonx/app/declare.py

def websocket(
    path: str,
    port: int = 8000,
    host: str = "0.0.0.0",
    workers: int = 1,
    codec: Optional[str] = None,
) -> Listener:
    """A websocket door: each text frame is one ingress item, decoded by
    ``codec`` — ``"json"`` (the default) or ``"text"``. Bytes frames are
    bytes. A frame the codec cannot read is answered with an error frame
    and does not reach the graph."""
    options = _codec_options(codec, f"websocket({path!r})")
    return Listener("websocket", path, int(port), host, workers=_workers(workers), options=options)

http

http(
    method: str,
    path: str,
    port: int = 8000,
    host: str = "0.0.0.0",
    workers: int = 1,
    codec: Optional[str] = None,
) -> Listener

An HTTP door: the body is the one ingress item, decoded by codec — "json" (the default) or "text". An empty body is None; a body the codec cannot read is answered 400 and starts no run.

Source code in operonx/app/declare.py

def http(
    method: str,
    path: str,
    port: int = 8000,
    host: str = "0.0.0.0",
    workers: int = 1,
    codec: Optional[str] = None,
) -> Listener:
    """An HTTP door: the body is the one ingress item, decoded by ``codec``
    — ``"json"`` (the default) or ``"text"``. An empty body is ``None``; a
    body the codec cannot read is answered ``400`` and starts no run."""
    options = _codec_options(codec, f"http({method!r}, {path!r})")
    return Listener(
        "http", path, int(port), host, str(method).upper(), _workers(workers), options=options
    )

asgi

asgi(
    path: str = "/", port: int = 8000, host: str = "0.0.0.0", workers: int = 1
) -> Listener

Source code in operonx/app/declare.py

def asgi(path: str = "/", port: int = 8000, host: str = "0.0.0.0", workers: int = 1) -> Listener:
    return Listener("asgi", path, int(port), host, workers=_workers(workers))

env

env(name: str, default: Any = None) -> Any

${NAME:default} for Python: the variable's value, cast to the default's type when the default is a number.

Source code in operonx/app/declare.py

def env(name: str, default: Any = None) -> Any:
    """``${NAME:default}`` for Python: the variable's value, cast to the
    default's type when the default is a number."""
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    if isinstance(default, bool):
        return raw.strip().lower() in ("1", "true", "yes", "on")
    if isinstance(default, int):
        return int(raw)
    if isinstance(default, float):
        return float(raw)
    return raw

The manifest

Manifest dataclass

Manifest(
    project: Dict[str, Any],
    serves: Tuple[ServeSpec, ...],
    graphs: Tuple[GraphSpec, ...],
    fixtures: Dict[str, Dict[str, Any]],
    resources_overlay: Optional[str],
    source: Optional[Path],
    on_startup: Tuple[Any, ...] = (),
    src: Tuple[str, ...] = (".",),
    base: Optional[Path] = None,
    tracing: Any = None,
    queues: Tuple[Any, ...] = (),
    jobs_dir: Optional[str] = None,
)

A parsed, validated operonx.toml.

Attributes

root property

root: Path

The directory the manifest lives in; relative paths resolve here.

app_entry property

app_entry: Optional[str]

[project] app = "module:APP" — the application declared in Python, which this file then only points at.

Methods:

listeners

listeners() -> Dict[Tuple[str, int], Tuple[ServeSpec, ...]]

Endpoints grouped onto the servers that will carry them.

Source code in operonx/app/manifest.py

def listeners(self) -> Dict[Tuple[str, int], Tuple[ServeSpec, ...]]:
    """Endpoints grouped onto the servers that will carry them."""
    out: Dict[Tuple[str, int], list] = {}
    for spec in self.serves:
        out.setdefault(spec.listener, []).append(spec)
    return {addr: tuple(specs) for addr, specs in out.items()}

find classmethod

find(start: Path | str = '.') -> 'Manifest'

Walk up from start to the first operonx.toml.

Source code in operonx/app/manifest.py

@classmethod
def find(cls, start: Path | str = ".") -> "Manifest":
    """Walk up from `start` to the first `operonx.toml`."""
    here = Path(start).resolve()
    if here.is_file():
        here = here.parent
    for candidate in (here, *here.parents):
        found = candidate / MANIFEST_FILENAME
        if found.is_file():
            return cls.from_file(found)
    raise ManifestError(f"no {MANIFEST_FILENAME} at or above {here}")

ServeSpec dataclass

ServeSpec(
    name: str,
    kind: str,
    graph: Any = "",
    path: str = "/",
    method: str = "POST",
    host: str = "0.0.0.0",
    port: int = 8000,
    session: str = "per_request",
    max_inflight: Optional[int] = None,
    on_session: Any = None,
    on_close: Any = None,
    app: Any = None,
    resume: Any = None,
    description: str = "",
    options: Dict[str, Any] = dict(),
    variants: Dict[str, Dict[str, Any]] = dict(),
    workers: int = 1,
    on_startup: Tuple[Any, ...] = (),
    trace_own: Optional[Tuple[Any, ...]] = None,
    trace_from: str = "default",
)

One endpoint: a transport, a graph, and how runs are minted.

Attributes:

Name	Type	Description
name	str	Identifies the endpoint in logs, metrics and operonx serve --only. Defaults to the path when absent.
kind	str	A registered transport name (websocket, http, asgi, file, queue) or a module:Class import path to a transport this project wrote itself. The built-ins hold no privileged position — they register the same way.
graph	Any	module:function entry point of the graph to run.
session	str	One of :data:SESSION_MODES. Decides run cardinality, and with it what an unhandled op failure means: a per_request run has exactly one caller waiting and maps a failure to 5xx, where a per_connection run is a phone call that must survive a bad turn.
max_inflight	Optional[int]	Bound on undelivered items. Required for :data:STREAM_KINDS, because an unbounded queue behind a network socket is how this project's Channels came to exist.
on_session	Any	module:function returning a RunRequest from a connection. Where ?call_id=&customer_info= becomes real inputs — project logic no amount of TOML expresses.
on_close	Any	module:function(session, handle) run after the run ends, however it ended. The symmetric half of on_session: whatever was opened at the door gets closed here — a counter decremented, a record written, a last message sent to a peer that may already be gone.
app	Any	For kind = "asgi": the foreign app to mount. Health, CRUD and admin routes stay someone else's code.
resume	Any	For an http door whose runs can stop for a human (an agent waiting on an approval): the graph that continues one. The door then also answers POST <path>/resume with it — same codec, hooks and streaming as the door itself (:meth:resume_spec). A websocket door resumes on its own connection and takes none.

Attributes

listener property

listener: Tuple[str, int]

Endpoints sharing this pair share one server.

Grouping by (host, port) rather than by a declared server name keeps a second concept out of the manifest: a listener is only ever an address, so let the address be the identity.

Methods:

resume_spec

resume_spec() -> Optional['ServeSpec']

The door resume adds: POST <path>/resume running the resume graph, named <name>.resume, with this door's codec, hooks, trace and address. None when there is no resume graph.

Source code in operonx/app/manifest.py

def resume_spec(self) -> Optional["ServeSpec"]:
    """The door ``resume`` adds: ``POST <path>/resume`` running the
    resume graph, named ``<name>.resume``, with this door's codec,
    hooks, trace and address. ``None`` when there is no resume graph."""
    if self.resume is None:
        return None
    return replace(
        self,
        name=f"{self.name}.resume",
        graph=self.resume,
        path=self.path.rstrip("/") + "/resume",
        method="POST",
        resume=None,
        description=f"resumes {self.name}'s runs",
    )

Tracing

[tracing] in operonx.toml: which trace sinks are on, and the precedence between it and trace= in Python. See Tracing.

tracing

[tracing] in operonx.toml — which trace sinks are on, in one place.

::

[tracing]
sinks = ["local", "trace_langfuse:edupia", "trace_clickhouse:default"]

[tracing.services.call]        # one service, overridden
sinks = ["local", "trace_langfuse:edupia"]

[tracing.jobs.backfill_call_logs]
sinks = []                     # this job is not traced

The file says which sinks a run goes to; resources.yaml keeps how to reach each one. "local" is the built-in local consumer (the one a job records to when nothing is configured); any other entry is a resource key, resolved through the hub exactly as trace=[...] entries are.

Precedence, most specific first:

1. [tracing.services.<name>] / [tracing.jobs.<name>];
2. the service's or job's own trace= (Python, or trace = on its [[serve]] / [[job]] block);
3. [tracing] sinks;
4. Application(trace=...) (or [project] trace);
5. the built-in default: a job records locally, a service traces nothing.

The toml is the operator's switch, so its per-service entry beats everything; a setting on one service beats any project-wide one, wherever it was written. sinks = [] is "not traced" at its level, never "not set".

Every level resolves to one list handed to one engine, and the engine hands its one WorkflowTrace to each consumer, so every sink of a run sees the same trace id — the caller's ?trace_id= included.

Classes

Tracing dataclass

Tracing(
    sinks: Optional[Tuple[str, ...]] = None,
    services: Dict[str, Tuple[str, ...]] = dict(),
    jobs: Dict[str, Tuple[str, ...]] = dict(),
)

A parsed [tracing] table. sinks is None when the table does not set it — the levels below then decide.

Functions:

parse_tracing

parse_tracing(raw: Any, where: str) -> Optional[Tracing]

[tracing] as a :class:Tracing, or None when absent.

Every mistake raises here, naming the key: a typo such as sink = must not read as "nothing configured" and quietly trace nowhere.

Source code in operonx/app/tracing.py

def parse_tracing(raw: Any, where: str) -> Optional[Tracing]:
    """``[tracing]`` as a :class:`Tracing`, or ``None`` when absent.

    Every mistake raises here, naming the key: a typo such as ``sink =``
    must not read as "nothing configured" and quietly trace nowhere."""
    if raw is None:
        return None
    if not isinstance(raw, dict):
        raise ManifestError(f"{where}: [tracing] must be a table")
    _no_stray(raw, _KEYS, f"{where}: [tracing]")
    sinks = _sinks(raw["sinks"], f"{where}: [tracing] sinks") if "sinks" in raw else None
    overrides = {}
    for kind in ("services", "jobs"):
        table = raw.get(kind, {})
        if not isinstance(table, dict):
            raise ManifestError(
                f"{where}: [tracing.{kind}] must be a table of [tracing.{kind}.<name>]"
            )
        out: Dict[str, Tuple[str, ...]] = {}
        for name, block in table.items():
            label = f"{where}: [tracing.{kind}.{name}]"
            if not isinstance(block, dict):
                raise ManifestError(f"{label} must be a table with `sinks`")
            _no_stray(block, ("sinks",), label)
            if "sinks" not in block:
                raise ManifestError(f"{label} has no `sinks` (write `sinks = []` to trace nothing)")
            out[str(name)] = _sinks(block["sinks"], f"{label} sinks")
        overrides[kind] = out
    return Tracing(sinks=sinks, services=overrides["services"], jobs=overrides["jobs"])

pick

pick(
    *,
    overrides: Sequence[Tuple[Optional[Sequence[Any]], str]],
    own: Optional[Sequence[Any]],
    own_label: str,
    tracing: Optional[Tracing],
    app: Optional[Sequence[Any]],
) -> Tuple[Optional[List[Any]], str]

The sinks one service or job uses, and the level they came from.

overrides are the [tracing.*.<name>] lists that name it, most specific first (a step's own entry, then its job of steps'). None at a level means "not set there"; [] is a choice. Returns (None, "default") when no level says anything.

Source code in operonx/app/tracing.py

def pick(
    *,
    overrides: Sequence[Tuple[Optional[Sequence[Any]], str]],
    own: Optional[Sequence[Any]],
    own_label: str,
    tracing: Optional[Tracing],
    app: Optional[Sequence[Any]],
) -> Tuple[Optional[List[Any]], str]:
    """The sinks one service or job uses, and the level they came from.

    ``overrides`` are the ``[tracing.*.<name>]`` lists that name it, most
    specific first (a step's own entry, then its job of steps').
    ``None`` at a level means "not set there"; ``[]`` is a choice. Returns
    ``(None, "default")`` when no level says anything."""
    for sinks, label in overrides:
        if sinks is not None:
            return list(sinks), label
    if own is not None:
        return list(own), own_label
    if tracing is not None and tracing.sinks is not None:
        return list(tracing.sinks), "[tracing]"
    if app is not None:
        return list(app), "application"
    return None, "default"

settle_serves

settle_serves(
    serves: Sequence[ServeSpec],
    app: Optional[Sequence[Any]],
    tracing: Optional[Tracing],
) -> Tuple[ServeSpec, ...]

Each service with options["trace"] set to what it will use.

A pure function of what was declared (trace_own), so applying it again — the file's [tracing] after a Python application already took its own default — gives the same answer as applying it once.

Source code in operonx/app/tracing.py

def settle_serves(
    serves: Sequence[ServeSpec], app: Optional[Sequence[Any]], tracing: Optional[Tracing]
) -> Tuple[ServeSpec, ...]:
    """Each service with ``options["trace"]`` set to what it will use.

    A pure function of what was declared (``trace_own``), so applying it
    again — the file's ``[tracing]`` after a Python application already
    took its own default — gives the same answer as applying it once."""
    out = []
    for spec in serves:
        if spec.kind == "asgi":
            out.append(spec)
            continue
        own = spec.trace_own
        if own is None and spec.trace_from == "default" and "trace" in spec.options:
            # a ServeSpec built by hand, with `options={"trace": ...}`:
            # that is its own choice, not something inherited
            own = tuple(spec.options["trace"] or ())
        override = tracing.services.get(spec.name) if tracing else None
        sinks, source = pick(
            overrides=[(override, f"[tracing.services.{spec.name}]")],
            own=own,
            own_label="service",
            tracing=tracing,
            app=app,
        )
        options = {k: v for k, v in spec.options.items() if k != "trace"}
        if sinks is not None:
            options["trace"] = sinks
        out.append(replace(spec, options=options, trace_own=own, trace_from=source))
    return tuple(out)

check_names

check_names(
    tracing: Optional[Tracing],
    where: str,
    services: Optional[Sequence[ServeSpec]],
    jobs: Optional[Iterable[str]],
) -> None

Every [tracing.services.<n>] names a service with runs, every [tracing.jobs.<n>] a job. None skips that check — a job of steps' members, say, are not known until it is imported.

Source code in operonx/app/tracing.py

def check_names(
    tracing: Optional[Tracing],
    where: str,
    services: Optional[Sequence[ServeSpec]],
    jobs: Optional[Iterable[str]],
) -> None:
    """Every ``[tracing.services.<n>]`` names a service with runs, every
    ``[tracing.jobs.<n>]`` a job. ``None`` skips that check — a job of steps'
    members, say, are not known until it is imported."""
    if tracing is None:
        return
    by_name = {s.name: s for s in services or ()}
    for name in tracing.services if services is not None else ():
        spec = by_name.get(name)
        if spec is None:
            have = ", ".join(by_name) or "none"
            raise ManifestError(
                f"{where}: [tracing.services.{name}] names no service (have: {have})"
            )
        if spec.kind == "asgi":
            raise ManifestError(
                f"{where}: [tracing.services.{name}] is an asgi mount — it runs no graph, "
                "so it has no runs to trace"
            )
    if jobs is None:
        return
    known = list(dict.fromkeys(jobs))
    for name in tracing.jobs:
        if name not in known:
            raise ManifestError(
                f"{where}: [tracing.jobs.{name}] names no job (have: {', '.join(known) or 'none'})"
            )

check_sinks

check_sinks(
    app: str, services: Sequence[ServeSpec] = (), jobs: Sequence[Any] = ()
) -> None

Every trace sink these services and jobs name is in the resource hub — checked as they start, so a sink missing from resources.yaml stops the start naming the key and the level that chose it, instead of surfacing as a stack trace from inside an engine (or, in a pooled worker, in a log nobody reads).

Source code in operonx/app/tracing.py

def check_sinks(app: str, services: Sequence[ServeSpec] = (), jobs: Sequence[Any] = ()) -> None:
    """Every trace sink these services and jobs name is in the resource
    hub — checked as they start, so a sink missing from
    ``resources.yaml`` stops the start naming the key and the level
    that chose it, instead of surfacing as a stack trace from inside an
    engine (or, in a pooled worker, in a log nobody reads)."""
    # (key, level that chose it) -> who uses it
    wanted: Dict[Tuple[str, str], List[str]] = {}
    for s in services:
        for sink in s.options.get("trace") or []:
            if isinstance(sink, str) and sink != LOCAL:
                wanted.setdefault((sink, s.trace_from), []).append(f"service {s.name!r}")

    def members(job: Any) -> List[Any]:
        if getattr(job, "steps", None) is None:
            return [job]
        return [m for step in job.steps for m in members(step)]

    for job in jobs:
        for j in members(job):
            for sink in getattr(j, "trace", None) or []:
                if isinstance(sink, str) and sink != LOCAL:
                    source = getattr(j, "_trace_from", "job")
                    wanted.setdefault((sink, source), []).append(f"job {j.name!r}")
    _check_wanted(app, wanted)

Services

compile_graph

compile_graph(
    entry: Any,
    *,
    bind: Optional[Dict[str, Any]] = None,
    trace: Any = None,
    concurrency: Optional[int] = None,
    where: str = "graph",
) -> Any

An Operon from a module:attr entry point, the way anything declared in the manifest is compiled.

Every parameter of the graph factory becomes a runtime input port. A graph compiled without declaring them keeps the literal defaults it was built with, which is the failure mode where a deep op holds None forever and every call fails on it.

With bind — a variant of one door — the named parameters of the @graph are fixed at build time and the rest stay runtime inputs. Each bound value that reads as module:attr is loaded; anything else is a literal; the decorator passes a static value into the body as-is. A plain function that builds and returns a graph is refused: graphs are defined at module level (guide 05).

Source code in operonx/app/serve/app.py

def compile_graph(
    entry: Any,
    *,
    bind: Optional[Dict[str, Any]] = None,
    trace: Any = None,
    concurrency: Optional[int] = None,
    where: str = "graph",
) -> Any:
    """An ``Operon`` from a ``module:attr`` entry point, the way anything
    declared in the manifest is compiled.

    Every parameter of the graph factory becomes a runtime input port. A
    graph compiled without declaring them keeps the literal defaults it
    was built with, which is the failure mode where a deep op holds
    ``None`` forever and every call fails on it.

    With ``bind`` — a variant of one door — the named parameters of the
    ``@graph`` are fixed at build time and the rest stay runtime inputs.
    Each bound value that reads as ``module:attr`` is loaded; anything
    else is a literal; the decorator passes a static value into the body
    as-is. A plain function that builds and returns a graph is refused:
    graphs are defined at module level (guide 05).
    """
    from operonx.core import Operon

    graph_fn = resolve_ref(entry, field=f"{where} graph")
    entry = ref_name(entry)
    bound: Dict[str, Any] = {}
    if bind:
        if not getattr(graph_fn, "_operonx_graph", False):
            raise TypeError(
                f"{where} graph {entry!r} is a plain function; variants bind a @graph's own "
                f"parameters {sorted(bind)}. Define the @graph at module level with them as "
                "parameters, not built inside a function (operonx guide 05)."
            )
        bound = _resolve_bind(bind, where)
    # `Operon(...)` on something that is not a graph fails as
    # `AttributeError: 'str' object has no attribute 'name'`, which names
    # neither the manifest entry nor what was actually wrong.
    if not hasattr(graph_fn, "name") and not callable(graph_fn):
        raise TypeError(
            f"{where} graph {entry!r} resolved to a "
            f"{type(graph_fn).__name__}, which is not a @graph"
        )
    try:
        signature = inspect.signature(graph_fn).parameters
    except (TypeError, ValueError):
        signature = {}
    params = {name: None for name in signature}
    defaults = {
        name: p.default for name, p in signature.items() if p.default is not inspect.Parameter.empty
    }
    unknown = set(bound) - set(params)
    if unknown:
        raise TypeError(
            f"{where} graph {entry!r} has no parameter {sorted(unknown)}; "
            f"it takes {sorted(params) or 'none'}"
        )
    params.update(bound)
    runtime = tuple(name for name in params if name not in bound)

    kwargs = {}
    if params:
        kwargs["params"] = params
    if trace:
        kwargs["trace"] = list(trace) if isinstance(trace, (list, tuple)) else [str(trace)]

    if getattr(graph_fn, "_operonx_graph", False):
        # Named after the graph, explicitly: built inside `Operon(...)` it
        # took the name of the local below, and every service run was
        # called `engine`.
        graph_fn = graph_fn(name=graph_fn.__name__, **params)
        kwargs.pop("params", None)
    engine = Operon(graph_fn, **kwargs)
    if concurrency:
        engine.graph.concurrency = int(concurrency)
    # What a door's RunRequest must carry: the graph's signature is the
    # contract, checked at the gate (`ServeRunner._inputs_fit`).
    engine.inputs_expected = runtime
    # What a doorless door fills a parameter with when the caller leaves it
    # out: the @graph's own default (`operonx.app.doors.serve_inputs`).
    engine.inputs_defaults = {k: v for k, v in defaults.items() if k in runtime}
    return engine

ingress async

ingress(items=None)

Yield each item the session receives, one per item.

transient=True is what makes per-item dispatch affordable: the item is delivered to its consumer and the cell released with the context, so a long-lived session does not retain everything that ever crossed it. Without it a call retains every packet it received — measured at 23 KB per item before the fix.

The generator ending is end-of-input, and it is the only such signal. Downstream ops drain and the run completes on its own; nothing outside cancels it. That is deliberate — work that has to happen after the peer has gone, like writing a call record, only survives if the run is allowed to finish.

items is an escape hatch for tests and for direct engine.start(): an iterable there is used when no session exists.

Source code in operonx/app/serve/ops.py

@op(bound="io", transient=True, door="ingress")
async def ingress(items=None):
    """Yield each item the session receives, one per item.

    ``transient=True`` is what makes per-item dispatch affordable: the
    item is delivered to its consumer and the cell released with the
    context, so a long-lived session does not retain everything that ever
    crossed it. Without it a call retains every packet it received —
    measured at 23 KB per item before the fix.

    The generator ending *is* end-of-input, and it is the only such
    signal. Downstream ops drain and the run completes on its own; nothing
    outside cancels it. That is deliberate — work that has to happen after
    the peer has gone, like writing a call record, only survives if the
    run is allowed to finish.

    `items` is an escape hatch for tests and for direct
    ``engine.start()``: an iterable there is used when no session exists.
    """
    session = current_session()
    if session is None:
        for item in items or ():
            yield {"item": item}
        return

    count = 0
    try:
        async for item in session.recv():
            count += 1
            yield {"item": item}
    finally:
        LOGGER.debug(f"[serve] ingress ended after {count} item(s)")

egress async

egress(item=None) -> dict

Write one item back to the session that minted this run.

A missing session is not an error — the graph is simply not being served — but it is reported, because a graph that believes it is replying to someone and is not would otherwise look identical to one that succeeded.

Source code in operonx/app/serve/ops.py

@op(bound="io", door="egress")
async def egress(item=None) -> dict:
    """Write one item back to the session that minted this run.

    A missing session is not an error — the graph is simply not being
    served — but it is reported, because a graph that believes it is
    replying to someone and is not would otherwise look identical to one
    that succeeded.
    """
    session = current_session()
    if session is None:
        LOGGER.debug("[serve] egress with no session; item dropped")
        return {"sent": False}
    try:
        return {"sent": bool(await session.send(item))}
    except Exception as exc:  # noqa: BLE001
        # `Session.send` is specified to report failure rather than raise,
        # and a transport that breaks that promise should lose its item,
        # not the run. A peer disappearing mid-reply is ordinary; the run
        # may still have a record to write.
        LOGGER.error(f"[serve] session.send raised: {type(exc).__name__}: {exc}")
        return {"sent": False}

current_session

current_session() -> Optional[Session]

The session that minted the run this op is running inside.

Reads the run's state through the same ContextVar SCRATCH uses, so it is bound inside an op body and unbound everywhere else. Returns None when the graph was started directly rather than served — which is what makes an ingress-bearing graph still testable by calling engine.start() with ordinary inputs.

Source code in operonx/app/serve/protocol.py

def current_session() -> Optional[Session]:
    """The session that minted the run this op is running inside.

    Reads the run's state through the same ContextVar `SCRATCH` uses, so
    it is bound inside an op body and unbound everywhere else. Returns
    ``None`` when the graph was started directly rather than served —
    which is what makes an `ingress`-bearing graph still testable by
    calling ``engine.start()`` with ordinary inputs.
    """
    from operonx.core.ops._edges import _current_state_var

    try:
        state = _current_state_var.get()
    except LookupError:
        return None
    return state._scratch.get(SESSION_KEY)

serve_session async

serve_session(
    engine: Any,
    session: Session,
    request: Optional[RunRequest] = None,
    metadata: Optional[Dict[str, Any]] = None,
    timeout: Optional[float] = None,
    on_start: Optional[Callable[[Any], None]] = None,
    reply: bool = False,
) -> Any

Run engine for one session, and return its handle when it ends.

The session is seeded into the run's scratch under a reserved key, so ingress and egress resolve it from the run rather than from a string the author had to remember.

The run is always allowed to finish. A peer that disappears surfaces as the session's recv ending, which ends ingress, which drains the graph — one signal travelling one way, instead of a race between a cancel and a teardown.

metadata is merged onto the run's trace before it ends, so every consumer sees it: a job tags its runs here (job, job_run, key and the same three as tags) and a transport could name its route the same way. tags extend the trace's list; other keys are set.

timeout is a deadline in seconds for the whole run. Past it the run is cancelled and :class:RunTimeout is raised — the one case in which a transport cancels the run it minted, and it does so because the caller asked for exactly that. A job's item_timeout is this.

on_start is called with the run's handle as soon as the run has started, before it is drained — for a caller that follows the run while it happens (the playground watches handle.trace grow, so a canvas can light each op as it finishes). A callback that raises is logged; it never stops the run.

reply=True is the doorless shape: when the run ends without an unhandled error, its outputs (without the $ keys) are sent to the session as its one reply, before the session is closed. A failed run sends nothing.

Source code in operonx/app/serve/runner.py

async def serve_session(
    engine: Any,
    session: Session,
    request: Optional[RunRequest] = None,
    metadata: Optional[Dict[str, Any]] = None,
    timeout: Optional[float] = None,
    on_start: Optional[Callable[[Any], None]] = None,
    reply: bool = False,
) -> Any:
    """Run `engine` for one session, and return its handle when it ends.

    The session is seeded into the run's scratch under a reserved key, so
    `ingress` and `egress` resolve it from the run rather than from a
    string the author had to remember.

    The run is always allowed to finish. A peer that disappears surfaces as
    the session's `recv` ending, which ends `ingress`, which drains the
    graph — one signal travelling one way, instead of a race between a
    cancel and a teardown.

    ``metadata`` is merged onto the run's trace before it ends, so every
    consumer sees it: a job tags its runs here (``job``, ``job_run``,
    ``key`` and the same three as ``tags``) and a transport could name
    its route the same way. ``tags`` extend the trace's list; other keys
    are set.

    ``timeout`` is a deadline in seconds for the whole run. Past it the
    run is cancelled and :class:`RunTimeout` is raised — the one case in
    which a transport cancels the run it minted, and it does so because
    the caller asked for exactly that. A job's ``item_timeout`` is this.

    ``on_start`` is called with the run's handle as soon as the run has
    started, before it is drained — for a caller that follows the run
    while it happens (the playground watches ``handle.trace`` grow, so
    a canvas can light each op as it finishes). A callback that raises
    is logged; it never stops the run.

    ``reply=True`` is the doorless shape: when the run ends without an
    unhandled error, its outputs (without the ``$`` keys) are sent to the
    session as its one reply, before the session is closed. A failed run
    sends nothing.
    """
    request = request or RunRequest()
    scratch = dict(request.scratch)
    scratch[SESSION_KEY] = session

    handle = engine.start(
        inputs=dict(request.inputs),
        scratch=scratch,
        request_id=request.request_id,
        user_id=request.user_id,
        session_id=request.session_id,
        trace_id=request.trace_id,
    )
    trace = getattr(handle, "trace", None)
    if metadata and trace is not None:
        trace.tag(metadata)
    if on_start is not None:
        try:
            on_start(handle)
        except Exception as exc:  # noqa: BLE001 — a follower's failure is its own
            LOGGER.error(f"[serve] on_start failed: {type(exc).__name__}: {exc}")
    try:
        if timeout is None:
            await _drain(handle)
        else:
            try:
                await asyncio.wait_for(_drain(handle), timeout)
            except asyncio.TimeoutError:
                handle.cancel()
                raise RunTimeout(
                    f"run exceeded {timeout:g}s", trace_id=getattr(trace, "trace_id", None)
                ) from None
        if reply and not unhandled(handle.errors):
            await session.send(plain(await handle.result()))
    finally:
        # A transport whose `close` raises must not turn a completed run
        # into a failed one. Closing is teardown; its failure is reported
        # where it happened rather than replacing whatever the run did.
        try:
            await session.close()
        except Exception as exc:  # noqa: BLE001
            LOGGER.error(f"[serve] session close failed: {type(exc).__name__}: {exc}")
    return handle

Session

Bases: Protocol

One connection, for as long as it lives.

recv ending is the only end-of-input signal, and a transport never cancels the run it minted. That is not a new rule — it is what the callbot already relies on: terminal_events writes the CRM record after the caller has hung up, so a framework that killed the run on disconnect would silently lose the record for every completed call. A transport wanting cancellation cancels the run explicitly.

RunRequest dataclass

RunRequest(
    inputs: Dict[str, Any] = dict(),
    scratch: Dict[str, Any] = dict(),
    trace_id: Optional[str] = None,
    request_id: Optional[str] = None,
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
    variant: Optional[str] = None,
)

A connection, turned into the arguments for one run.

Returned by a project's on_session hook. Turning ?call_id=&customer_info= into real inputs, refusing an unknown agent, hitting a customer store — that is project logic, and no amount of TOML will express it. One declared hook is the honest answer; frameworks that pretended config was enough grew a plugin system instead.

Returning None from the hook refuses the connection.

variant names one of the door's [serve.variants] — the compiled graph this session runs. Required when the door declares variants, refused when it names one the door does not have.

BoundedSession

BoundedSession(
    meta: Optional[Mapping[str, Any]] = None, max_inflight: Optional[int] = None
)

A Session whose inbound side has a ceiling.

The bound lives here rather than in the ingress op, because this is where items enter. When the buffer is full the transport stops reading its socket, and for TCP that is the connection's own flow control — correct, free, and applied before anything is allocated inside the graph.

It is mandatory for stream transports rather than defaulted, because operonx guards concurrency and — since transient ports — retention, but it has never guarded volume. An unbounded queue behind a network socket is precisely how this project's Channels came to exist, and a ceiling nobody chose is how one gets shipped again.

Subclasses implement :meth:_send. Anything that overflows is counted and reported, never dropped in silence.

Source code in operonx/app/serve/protocol.py

def __init__(
    self, meta: Optional[Mapping[str, Any]] = None, max_inflight: Optional[int] = None
):
    self.meta: Mapping[str, Any] = dict(meta or {})
    self.max_inflight = max_inflight
    self._queue: asyncio.Queue = asyncio.Queue(maxsize=max_inflight or 0)
    self._closed = asyncio.Event()
    self.overflowed = 0
    #: The trace id of the run this session minted, set by the serve
    #: runner as the run starts — so a transport can hand it to its
    #: peer (an HTTP reply's ``x-operonx-trace-id`` header).
    self.trace_id: Optional[str] = None

Methods:

feed async

feed(item: Any) -> None

Push one inbound item, waiting when the bound is reached.

Source code in operonx/app/serve/protocol.py

async def feed(self, item: Any) -> None:
    """Push one inbound item, waiting when the bound is reached."""
    await self._queue.put(item)

feed_nowait

feed_nowait(item: Any) -> bool

Push without waiting. False means the bound was hit.

Counted rather than silent — a transport that drops has to be able to say how much.

Source code in operonx/app/serve/protocol.py

def feed_nowait(self, item: Any) -> bool:
    """Push without waiting. ``False`` means the bound was hit.

    Counted rather than silent — a transport that drops has to be able
    to say how much.
    """
    try:
        self._queue.put_nowait(item)
        return True
    except asyncio.QueueFull:
        self.overflowed += 1
        return False

end_input

end_input() -> None

No more inbound items. recv drains what is left, then stops.

Source code in operonx/app/serve/protocol.py

def end_input(self) -> None:
    """No more inbound items. `recv` drains what is left, then stops."""
    self._closed.set()
    try:
        self._queue.put_nowait(_EOF)
    except asyncio.QueueFull:
        # The sentinel only exists to wake a reader parked in `get()`,
        # and nothing is parked there while the queue is full. `recv`
        # sees the closed flag once it has drained what is there. Found
        # by a stream job whose source outran its graph: the same shape
        # as a peer hanging up while its packets are still queued.
        pass

send async

send(item: Any) -> bool

Write one item out. False means it did not reach the peer.

A bool rather than None because a caller that paces audio has to report what actually landed: the callbot's play_frame sends forty-odd chunks per spoken frame and its audit line — chunks, ok, fail, the indices that failed — is the row you read when audio sounds wrong. Swallowing per-item failure here would have forced that op to keep talking to the socket directly, which is the one coupling the transport interface exists to remove.

Failure is reported, never raised: a peer that goes away mid-reply is ordinary, and one lost frame must not tear down a run that still has a record to write.

Source code in operonx/app/serve/protocol.py

async def send(self, item: Any) -> bool:
    """Write one item out. ``False`` means it did not reach the peer.

    A bool rather than ``None`` because a caller that paces audio has
    to report what actually landed: the callbot's `play_frame` sends
    forty-odd chunks per spoken frame and its audit line — chunks, ok,
    fail, the indices that failed — is the row you read when audio
    sounds wrong. Swallowing per-item failure here would have forced
    that op to keep talking to the socket directly, which is the one
    coupling the transport interface exists to remove.

    Failure is reported, never raised: a peer that goes away mid-reply
    is ordinary, and one lost frame must not tear down a run that still
    has a record to write.
    """
    return await self._send(item)

run_failed async

run_failed() -> None

The run this session minted ended with an error.

Called by the serve runner after the run, before on_close. A transport whose peer would otherwise wait in silence tells it here; the default does nothing, because a transport such as HTTP answers from its own endpoint (a 500). Never pass the error text on: it is a traceback, and it stays in the log and the trace.

Source code in operonx/app/serve/protocol.py

async def run_failed(self) -> None:
    """The run this session minted ended with an error.

    Called by the serve runner after the run, before ``on_close``. A
    transport whose peer would otherwise wait in silence tells it
    here; the default does nothing, because a transport such as HTTP
    answers from its own endpoint (a ``500``). Never pass the error
    text on: it is a traceback, and it stays in the log and the trace.
    """

Evals

Eval

Eval(
    name: str,
    *,
    graph: Any,
    dataset: Any,
    evaluators: Sequence[Any] = (),
    threshold: Optional[float] = None,
    repeats: int = 1,
    cluster: Optional[str] = None,
    gate: Optional[Gate] = None,
    root: Union[str, Path, None] = None,
    record_dir: Union[str, Path, None] = None,
    scores: Any = None,
    scores_timeout: float = 10.0,
    variant: Optional[str] = None,
    judge_cache: bool = True,
    judge_concurrency: int = 8,
    **kwargs: Any,
)

Bases: Job

A dataset, a graph, evaluators — run as a job with origin=eval.

Everything but the arguments below is a :class:Job argument (concurrency, timeout, inputs, input, trace…).

Parameters:

Name	Type	Description	Default
dataset	Any	A :class:Dataset, a JSONL path, or "dataset:name".	required
evaluators	Sequence[Any]	Functions judging each case (see the module docstring).	()
threshold	Optional[float]	With it, the run fails when the pass rate is under it; without, it fails when any case fails. With a gate, put it in Gate(threshold=…) instead.	None
repeats	int	Runs per case. Above 1, each case is repeats items keyed "<id>#<r>", and the summary says which cases are flaky and what pass^k is.	1
cluster	Optional[str]	The case field that groups cases (one scenario's variants, one conversation's turns) for the clustered SE. Without it a case's own cluster field is read; a case with neither is its own cluster.	None
gate	Optional[Gate]	A :class:~operonx.app.evals.gate.Gate: thresholds, a baseline to compare against, a must-pass tier, an error budget, and the exit code they make. Without one the run passes or fails exactly as threshold says.	None
root	Union[str, Path, None]	The project root, where the fingerprint asks git for the commit (the manifest's directory; else the working directory).	None
scores	Any	Where the experiment, its items and every check's score are written as the run goes: a :class:~operonx.telemetry.scores.ScoreStore, a "score_store:<name>" key, or a spec ({"backend": "files"}). Unset, nothing is written but the job record.	None
scores_timeout	float	How long the run waits, at its end, for the store to take what is queued. What it did not take is counted and logged; the job record still holds every verdict, and :func:~operonx.app.evals.publish sends it later.	10.0
variant	Optional[str]	A free label for what this experiment tries ("prompt v7"), kept on its record and its experiment row.	None
judge_cache	bool	With a score store, a judge's verdict is kept there, keyed by its version and what it was shown, and an unchanged case is not judged again (no call). False asks every time.	True
judge_concurrency	int	Judge runs in flight at once, over all cases.	8

Source code in operonx/app/evals/job.py

def __init__(
    self,
    name: str,
    *,
    graph: Any,
    dataset: Any,
    evaluators: Sequence[Any] = (),
    threshold: Optional[float] = None,
    repeats: int = 1,
    cluster: Optional[str] = None,
    gate: Optional[Gate] = None,
    root: Union[str, Path, None] = None,
    record_dir: Union[str, Path, None] = None,
    scores: Any = None,
    scores_timeout: float = 10.0,
    variant: Optional[str] = None,
    judge_cache: bool = True,
    judge_concurrency: int = 8,
    **kwargs: Any,
):
    self.dataset = dataset if isinstance(dataset, Dataset) else Dataset(dataset_path(dataset))
    self.evaluators = [evaluator_of(ev) for ev in evaluators]
    self._prepared = [prepare(ev) for ev in self.evaluators]
    # a case's trace view is built only when some evaluator can take it
    self._wants_trace = any(p.wants("trace") for p in self._prepared)
    # text, so never lazy: made only for an evaluator that names it (one taking
    # **kwargs has `trace`, and `trace.as_text()` when it wants it)
    self._wants_summary = any(
        p.params is not None and "trace_summary" in p.params for p in self._prepared
    )
    self._wants_judging = any(p.wants("judging") for p in self._prepared)
    if isinstance(judge_concurrency, bool) or int(judge_concurrency) < 1:
        raise ValueError(f"eval {name!r}: judge_concurrency is a number of judge runs, ≥ 1")
    self.judge_cache = bool(judge_cache)
    self.judge_concurrency = int(judge_concurrency)
    if threshold is not None and not 0 <= float(threshold) <= 1:
        raise ValueError(f"eval {name!r}: threshold is a pass rate in [0, 1]")
    self.threshold = float(threshold) if threshold is not None else None
    if isinstance(repeats, bool) or not isinstance(repeats, int) or repeats < 1:
        raise ValueError(f"eval {name!r}: repeats is a whole number of runs per case, ≥ 1")
    self.repeats = repeats
    self.cluster = cluster
    if gate is not None and not isinstance(gate, Gate):
        raise TypeError(f"eval {name!r}: gate is a Gate, not a {type(gate).__name__}")
    if gate is not None and threshold is not None:
        raise ValueError(
            f"eval {name!r}: threshold= is the gate-less shorthand; with a gate, "
            "give it as Gate(threshold=…)"
        )
    self.gate = gate
    if gate is not None:
        known = {PASS_METRIC} | {_name(ev) for ev in self.evaluators}
        unknown = [m for m in gate.named_metrics() if m not in known]
        if unknown:
            raise ValueError(
                f"eval {name!r}: the gate names {unknown}, which are not metrics of "
                f"this eval; metrics are {sorted(known)} ('pass' = every check passed)"
            )
    self.root = Path(root) if root is not None else None
    self.scores = _check_scores(name, scores)
    self.scores_timeout = float(scores_timeout)
    self.variant = str(variant) if variant else None
    self._judges = [
        _name(ev) for ev in self.evaluators if getattr(ev, "eval_kind", None) == "judge"
    ]
    self._judging: Optional[Judging] = None
    self._judge_counts: Dict[str, Dict[str, Any]] = {}
    self._alignments: Optional[Dict[str, Dict[str, Dict[str, Any]]]] = None
    self._store: Optional[ScoreStore] = None
    self._writer: Optional[ScoreWriter] = None
    self._experiment: Optional[Experiment] = None
    self._rows: Dict[str, Dict[str, Any]] = {}
    self._verdicts: List[Dict[str, Any]] = []
    self._fingerprint: Optional[Dict[str, Any]] = None
    self._code: Optional["Future[Dict[str, Any]]"] = None
    self._baseline: Optional[Tuple[str, Dict[str, CaseOutcome], Optional[Dict]]] = None
    # a "main" / "git:<ref>" baseline: found before the record opens
    self._from_git: Optional[Tuple[str, Dict[str, CaseOutcome], Dict, str]] = None
    self._complete = False
    super().__init__(
        name, graph=graph, items=self._cases, key="id", record_dir=record_dir, **kwargs
    )

Methods:

begin

begin(run_id: str, started: str) -> None

The runner opened the record: the judges get their context (the eval's sinks, this run's ids, the cache); with scores=, the experiment's running row goes to the store before any case.

Source code in operonx/app/evals/job.py

def begin(self, run_id: str, started: str) -> None:
    """The runner opened the record: the judges get their context (the
    eval's sinks, this run's ids, the cache); with ``scores=``, the
    experiment's ``running`` row goes to the store before any case."""
    self._writer, self._experiment = None, None
    self._judge_counts = {}
    self._judging = Judging(
        trace=self.engine().trace_consumers,
        metadata={"job": self.name, "job_run": run_id},
        cache=self._store if self.judge_cache else None,
        concurrency=self.judge_concurrency,
    )
    if self._store is None:
        return
    opened = JobRun(
        job=self.name,
        run_id=run_id,
        path=self.records() / self.name / run_id,
        status=RUN_RUNNING,
        started=started,
        ended=None,
        counts={},
        meta=self.describe(),
    )
    self._experiment = experiment_of(opened)
    self._writer = ScoreWriter(self._store, self.name)
    self._writer.submit([self._experiment])

run async

run(*, resume: bool = False, **kwargs: Any) -> JobRun

Run the eval once; with scores=, then wait (up to scores_timeout) for the store to take the experiment.

Source code in operonx/app/evals/job.py

async def run(self, *, resume: bool = False, **kwargs: Any) -> JobRun:
    """Run the eval once; with ``scores=``, then wait (up to
    ``scores_timeout``) for the store to take the experiment."""
    # opened before the record: a store that cannot be named fails the
    # run before it starts, not with a record stuck at "running"
    self._store = _open_scores(self.scores) if self.scores is not None else None
    # so is a commit's baseline: no experiment there, no run (and no cost)
    ref = git_ref(self.gate.baseline) if self.gate is not None else None
    self._from_git = await asyncio.to_thread(self._git_baseline, ref) if ref else None
    # each judge's alignment record, read once: the report warns on an unvalidated one
    self._alignments = (
        await asyncio.to_thread(self._read_alignments) if self._store is not None else None
    )
    run: Optional[JobRun] = None
    try:
        run = await super().run(resume=resume, **kwargs)
        if self._writer is not None:
            self._writer.submit([experiment_of(run)])
    finally:
        writer, self._writer = self._writer, None
        if writer is not None:  # even when the run raised: no writer thread outlives it
            _, lost = await asyncio.to_thread(writer.finish, self.scores_timeout)
            if lost and run is not None:
                warn_lost(self.name, lost, run)
    return run

judge async

judge(raw: Any, result: Any) -> None

After a case's run: its evaluators, and the verdict on its record.

Source code in operonx/app/evals/job.py

async def judge(self, raw: Any, result: Any) -> None:
    """After a case's run: its evaluators, and the verdict on its record."""
    if isinstance(raw, _Trial):
        row, repeat, digest = raw.row, raw.repeat, raw.case_hash
    else:  # items given on the command line: its rows are the cases
        row = self._rows.get(result.key) or (
            raw if isinstance(raw, Mapping) else {"input": raw}
        )
        repeat, digest = 0, case_hash(row)
    case = str(row.get("id", result.key))
    output = result.result
    # every frame the case sent: one result is one frame, several are a list
    sent = output if result.sent > 1 else ([output] if output is not None else [])
    if result.status not in (ITEM_OK, ITEM_EMPTY):
        verdict = {"passed": False, "error": result.error or result.status, "checks": {}}
    else:
        avail = {
            "input": row.get("input"),
            "output": output,
            "expected": row.get("expected"),
            "row": row,
            "outputs": sent,
        }
        trace = getattr(result, "trace", None)
        if self._wants_trace or self._wants_summary:
            view = TraceView.from_trace(trace) if trace is not None else None
            avail["trace"] = view
            if self._wants_summary:
                avail["trace_summary"] = view.as_text() if view is not None else None
        if self._wants_judging and self._judging is not None:
            avail["judging"] = self._judging.for_case(
                key=result.key, case=case, judged_trace=result.trace_id
            )
        checks = await judge_all(self._prepared, avail)
        self._count_judges(checks)
        verdict = case_verdict(checks)
    result.verdict = recorded_verdict(
        verdict,
        row,
        output,
        case=case,
        repeat=repeat,
        digest=digest,
        cluster=self._cluster_of(row),
        trace=getattr(result, "trace", None),
    )
    self._verdicts.append(trial_of(result.key, result.verdict, result.ms))
    if self._writer is not None:
        meta = {"eval": {"fingerprint": self._fingerprint}, "judges": self._judges}
        item = item_of(self._experiment.experiment_id, result)
        self._writer.submit([item, *scores_of(meta, self._experiment, result)])

summarize

summarize(status: str) -> tuple

What run.json says about the eval, and the run's status.

Without a gate: failed when a trial failed, or with a threshold, when the rate is under it (1.9.0, unchanged). With one: whatever the gate decides (see :mod:operonx.app.evals.gate).

Source code in operonx/app/evals/job.py

def summarize(self, status: str) -> tuple:
    """What run.json says about the eval, and the run's status.

    Without a gate: failed when a trial failed, or with a threshold,
    when the rate is under it (1.9.0, unchanged). With one: whatever
    the gate decides (see :mod:`operonx.app.evals.gate`).
    """
    vs, self._verdicts = self._verdicts, []
    nums, cases, metrics = numbers(vs, self.repeats)
    summary: Dict[str, Any] = {
        "dataset": str(self.dataset.path),
        "threshold": self.threshold,
        **nums,
    }
    if self.variant:
        summary["variant"] = self.variant
    if self.dataset.selection:
        summary["selection"] = self.dataset.selection
    fp = self._fingerprint
    if fp is not None and self._code is not None:
        code = self._code.result()
        fp = {
            **fp,
            "code_version": code.get("version"),
            "version_dirty": code.get("version_dirty"),
        }
    self._fingerprint = fp
    summary["fingerprint"] = fp

    summary["gate"], status = gate_run(
        self.gate,
        self.threshold,
        status=status,
        nums=nums,
        cases=cases,
        metrics=metrics,
        complete=self._complete,
        baseline=self._baseline,
        fingerprint=fp,
        baseline_ref=self._from_git[3] if self._from_git is not None else None,
    )
    if self._judges:
        summary["judges"], warned = self._judge_report(fp)
        if warned:  # loud, but not a verdict: an unvalidated judge can still gate (T4)
            summary["gate"]["warnings"] = list(summary["gate"].get("warnings") or []) + warned
    return {"eval": summary}, status

rescore async

rescore(
    run_id: str,
    evaluators: Optional[Sequence[Any]] = None,
    *,
    store: Any = None,
    dataset: Any = None,
    scores: Any = None,
) -> Any

Judge this eval's recorded run run_id again without running the graph (:func:~operonx.app.evals.rescore). Without evaluators, the eval's own — its judges left out, and named in skipped. store is the run store its traces went to; scores a score store the new scores go to.

Source code in operonx/app/evals/job.py

async def rescore(
    self,
    run_id: str,
    evaluators: Optional[Sequence[Any]] = None,
    *,
    store: Any = None,
    dataset: Any = None,
    scores: Any = None,
) -> Any:
    """Judge this eval's recorded run *run_id* again without running the
    graph (:func:`~operonx.app.evals.rescore`). Without
    *evaluators*, the eval's own — its judges left out, and named in
    ``skipped``. *store* is the run store its traces went to; *scores*
    a score store the new scores go to."""
    from .rescoring import is_judge, rescore

    path = self.records() / self.name / run_id
    if not (path / "run.json").is_file():
        raise ValueError(f"eval {self.name!r}: no run {run_id!r} under {path.parent}")
    chosen = list(self.evaluators if evaluators is None else evaluators)
    skipped: List[str] = []
    if evaluators is None:
        skipped = [_name(ev) for ev in chosen if is_judge(ev)]
        chosen = [ev for ev in chosen if not is_judge(ev)]
    out = await rescore(
        path,
        chosen,
        store=store,
        dataset=self.dataset if dataset is None else dataset,
        scores=None if scores is None else _open_scores(_check_scores(self.name, scores)),
    )
    out.skipped = skipped
    return out

Dataset

Dataset(
    path: Union[str, Path],
    *,
    split: Optional[str] = None,
    tags: Optional[Sequence[str]] = None,
    ids: Optional[Sequence[str]] = None,
    sample: Optional[int] = None,
)

A JSONL file of cases. Reading is lazy; adding appends and dedupes.

:meth:select gives a view of some of the cases (a split, tags, ids, a stable sample) — the same file, read through the selection.

Source code in operonx/app/evals/dataset.py

def __init__(
    self,
    path: Union[str, Path],
    *,
    split: Optional[str] = None,
    tags: Optional[Sequence[str]] = None,
    ids: Optional[Sequence[str]] = None,
    sample: Optional[int] = None,
):
    self.path = Path(path)
    if sample is not None and (isinstance(sample, bool) or int(sample) < 1):
        raise ValueError(f"Dataset: sample is a number of cases, ≥ 1, not {sample!r}")
    self.split = split
    self.tags = tuple(tags) if tags else None
    self.ids = tuple(str(i) for i in ids) if ids else None
    self.sample = int(sample) if sample is not None else None

Attributes

selection property

selection: Dict[str, Any]

What :meth:select chose, as data ({} for every case).

Methods:

select

select(
    *,
    split: Optional[str] = None,
    tags: Optional[Sequence[str]] = None,
    ids: Optional[Sequence[str]] = None,
    sample: Optional[int] = None,
) -> "Dataset"

The cases of split, carrying any of tags, among ids (each must exist), then the sample of them with the smallest sha256(id) — the same N on every machine and every run. A criterion given replaces this view's own; one not given is kept.

Source code in operonx/app/evals/dataset.py

def select(
    self,
    *,
    split: Optional[str] = None,
    tags: Optional[Sequence[str]] = None,
    ids: Optional[Sequence[str]] = None,
    sample: Optional[int] = None,
) -> "Dataset":
    """The cases of *split*, carrying any of *tags*, among *ids* (each
    must exist), then the *sample* of them with the smallest
    ``sha256(id)`` — the same N on every machine and every run. A
    criterion given replaces this view's own; one not given is kept."""
    return Dataset(
        self.path,
        split=split if split is not None else self.split,
        tags=tags if tags else self.tags,
        ids=ids if ids else self.ids,
        sample=sample if sample is not None else self.sample,
    )

all_rows

all_rows() -> List[Dict[str, Any]]

Every case in the file, archived ones too, whatever the selection.

Source code in operonx/app/evals/dataset.py

def all_rows(self) -> List[Dict[str, Any]]:
    """Every case in the file, archived ones too, whatever the selection."""
    if not self.path.is_file():
        return []
    with self.path.open("r", encoding="utf-8") as fh:
        return parse_rows(fh, where=str(self.path))

rows

rows() -> List[Dict[str, Any]]

The selected active cases, in file order (archived ones are left out).

Source code in operonx/app/evals/dataset.py

def rows(self) -> List[Dict[str, Any]]:
    """The selected active cases, in file order (archived ones are left out)."""
    rows = [r for r in self.all_rows() if r.get("status") != "archived"]
    if not self.selection:
        return rows
    if self.split is not None:
        rows = [r for r in rows if r.get("split") == self.split]
    if self.tags:
        wanted = set(self.tags)
        rows = [r for r in rows if wanted & set(r.get("tags") or ())]
    if self.ids:
        have = {r["id"] for r in rows}
        missing = [i for i in self.ids if i not in have]
        if missing:
            raise ValueError(f"cases not in {self.path} (or not selected): {missing}")
        wanted = set(self.ids)
        rows = [r for r in rows if r["id"] in wanted]
    if self.sample is not None and len(rows) > self.sample:
        chosen = set(sorted((r["id"] for r in rows), key=_stable)[: self.sample])
        rows = [r for r in rows if r["id"] in chosen]
    return rows

problems

problems() -> List[Tuple[int, str]]

What is wrong with the file, each with its line (0: the file).

Source code in operonx/app/evals/dataset.py

def problems(self) -> List[Tuple[int, str]]:
    """What is wrong with the file, each with its line (0: the file)."""
    if not self.path.is_file():
        return [(0, "no such file")]
    out: List[Tuple[int, str]] = []
    first: Dict[str, int] = {}
    with self.path.open("r", encoding="utf-8") as fh:
        for n, line in enumerate(fh, 1):
            line = line.strip()
            if not line:
                continue
            try:
                row = json.loads(line)
            except ValueError as exc:
                out.append((n, f"not JSON: {getattr(exc, 'msg', exc)}"))
                continue
            if not isinstance(row, dict) or "input" not in row:
                row = {"input": row}
            cid = case_id(row)
            if cid in first:
                out.append((n, f"duplicate id {cid!r} (first on line {first[cid]})"))
            else:
                first[cid] = n
            out.extend((n, why) for why in _row_problems(row))
    return out

add

add(rows: Iterable[Mapping]) -> List[str]

Append rows whose id is not already there; returns the ids added.

Source code in operonx/app/evals/dataset.py

def add(self, rows: Iterable[Mapping]) -> List[str]:
    """Append *rows* whose id is not already there; returns the ids added."""
    self.path.parent.mkdir(parents=True, exist_ok=True)
    with _writing(self.path.parent):
        return self._append(rows)

update

update(case: str, changes: Mapping[str, Any]) -> Dict[str, Any]

Change one case in place and return it as it now reads.

changes maps keys of :data:EDITABLE to their new values; None removes the key ("status": "active" too: active is the default). The file is rewritten through a temporary file beside it, the case's line changed and every other line kept byte for byte, so a diff shows exactly the edit. A change that would make the case malformed, an unknown or duplicated case, and a file another writer changed meanwhile are refused with the file untouched.

Source code in operonx/app/evals/dataset.py

def update(self, case: str, changes: Mapping[str, Any]) -> Dict[str, Any]:
    """Change one case in place and return it as it now reads.

    *changes* maps keys of :data:`EDITABLE` to their new values; ``None``
    removes the key (``"status": "active"`` too: active is the default).
    The file is rewritten through a temporary file beside it, the
    case's line changed and every other line kept byte for byte, so a
    diff shows exactly the edit. A change that would make the case
    malformed, an unknown or duplicated case, and a file another writer
    changed meanwhile are refused with the file untouched.
    """
    unknown = [k for k in changes if k not in EDITABLE]
    if unknown:
        why = (
            " — a different input is a different case: add it, and archive this one"
            if "input" in unknown
            else ""
        )
        raise ValueError(
            f"cannot change {unknown[0]!r} of a case{why}; "
            f"an edit changes {', '.join(EDITABLE)}"
        )
    if not changes:
        raise ValueError(f"nothing to change in case {case!r}")
    if not self.path.is_file():
        raise ValueError(f"{self.path}: no such file")
    # one writer at a time among add() and update(); a writer that is
    # not a Dataset is caught by the size/mtime check before the replace
    with _writing(self.path.parent):
        before = self.path.stat()
        with self.path.open("r", encoding="utf-8", newline="") as fh:
            lines = fh.readlines()
        found: List[int] = []
        for n, line in enumerate(lines):
            if not line.strip():
                continue
            try:
                row = json.loads(line)
            except ValueError:
                continue  # problems() names it; it is not the case asked for
            if not isinstance(row, dict) or "input" not in row:
                row = {"input": row}
            if case_id(row) == str(case):
                found.append(n)
        if not found:
            raise ValueError(f"no case {case!r} in {self.path}")
        if len(found) > 1:
            on = " and ".join(str(n + 1) for n in found)
            raise ValueError(
                f"case {case!r} is on lines {on} of {self.path}: fix the file first"
            )
        n = found[0]
        raw = json.loads(lines[n])
        row = dict(raw) if isinstance(raw, dict) and "input" in raw else {"input": raw}
        if "id" not in row:
            row = {"id": case_id(row), **row}
        for key, value in changes.items():
            if value is None or (key == "status" and value == "active"):
                row.pop(key, None)
            else:
                row[key] = value
        bad = _row_problems(row)
        if bad:
            raise ValueError(f"case {case!r}: {bad[0]}")
        end = "\n" if lines[n].endswith("\n") else ""
        lines[n] = json.dumps(row, ensure_ascii=False, default=str) + end
        fd, tmp = tempfile.mkstemp(
            prefix=f".{self.path.name}.", suffix=".tmp", dir=self.path.parent
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8", newline="") as fh:
                fh.writelines(lines)
            now = self.path.stat()
            if (now.st_size, now.st_mtime_ns) != (before.st_size, before.st_mtime_ns):
                raise ValueError(
                    f"{self.path} changed while it was being edited (another writer): "
                    "nothing was written; read it again and repeat the edit"
                )
            os.replace(tmp, self.path)
        finally:
            if os.path.exists(tmp):
                os.unlink(tmp)
    return row

exact

exact(field: Optional[str] = None) -> Callable

Passes when the output (or its field, dotted) equals expected.

Source code in operonx/app/evals/evaluators.py

def exact(field: Optional[str] = None) -> Callable:
    """Passes when the output (or its ``field``, dotted) equals ``expected``."""

    def exact_match(output: Any = None, expected: Any = None) -> Dict[str, Any]:
        got = _pick(output, field)
        want = _pick(expected, field) if isinstance(expected, Mapping) and field else expected
        ok = got == want
        return {"passed": ok, "reason": None if ok else f"got {got!r}, expected {want!r}"}

    exact_match.eval_name = f"exact({field})" if field else "exact"
    return exact_match

contains

contains(
    *needles: str, field: Optional[str] = None, case: bool = False
) -> Callable

Passes when the output text contains every needle — the given ones, else expected (a string or a list of strings).

Source code in operonx/app/evals/evaluators.py

def contains(*needles: str, field: Optional[str] = None, case: bool = False) -> Callable:
    """Passes when the output text contains every needle — the given ones,
    else ``expected`` (a string or a list of strings)."""

    def contains_all(output: Any = None, expected: Any = None) -> Dict[str, Any]:
        text = _text(_pick(output, field))
        want = list(needles) or ([expected] if isinstance(expected, str) else list(expected or []))
        hay = text if case else text.lower()
        missing = [n for n in want if (n if case else str(n).lower()) not in hay]
        return {
            "passed": not missing and bool(want),
            "reason": f"missing {missing}"
            if missing
            else (None if want else "nothing to look for"),
        }

    contains_all.eval_name = "contains"
    return contains_all

fuzzy

fuzzy(threshold: float = 0.8, field: Optional[str] = None) -> Callable

Similarity of the output text to expected (0–1); passes at threshold.

Source code in operonx/app/evals/evaluators.py

def fuzzy(threshold: float = 0.8, field: Optional[str] = None) -> Callable:
    """Similarity of the output text to ``expected`` (0–1); passes at *threshold*."""

    def fuzzy_match(output: Any = None, expected: Any = None) -> Dict[str, Any]:
        a, b = _text(_pick(output, field)), _text(expected)
        score = difflib.SequenceMatcher(None, a.lower(), b.lower()).ratio()
        return {"passed": score >= threshold, "score": round(score, 4)}

    fuzzy_match.eval_name = f"fuzzy({threshold:g})"
    return fuzzy_match

json_match

json_match(keys: Optional[Sequence[str]] = None) -> Callable

Passes when the output object agrees with expected on keys (every key expected has, when none are named); score is the share that agree.

Source code in operonx/app/evals/evaluators.py

def json_match(keys: Optional[Sequence[str]] = None) -> Callable:
    """Passes when the output object agrees with ``expected`` on *keys*
    (every key ``expected`` has, when none are named); score is the share
    that agree."""

    def json_agree(output: Any = None, expected: Any = None) -> Dict[str, Any]:
        if not isinstance(output, Mapping) or not isinstance(expected, Mapping):
            return {"passed": False, "reason": "output and expected must both be objects"}
        names = list(keys or expected.keys())
        wrong = [k for k in names if _pick(output, k) != _pick(expected, k)]
        score = 1 - len(wrong) / len(names) if names else 1.0
        return {
            "passed": not wrong,
            "score": round(score, 4),
            "reason": f"differs on {wrong}" if wrong else None,
        }

    json_agree.eval_name = "json_match"
    return json_agree

llm_judge

llm_judge(resource: str, rubric: str, *, name: str = 'llm_judge') -> Judge

An LLM grades the case against rubric: {passed, score, reason} parsed from its JSON answer, with the call's cost and usage kept on the verdict. resource is an llm: key (or a bare name) the hub resolves. The 1.9.0 judge, now a traced, versioned, cached :class:Judge; :func:judge (PASS/FAIL with a reason) is the one to reach for.

Source code in operonx/app/evals/judges.py

def llm_judge(resource: str, rubric: str, *, name: str = "llm_judge") -> Judge:
    """An LLM grades the case against *rubric*: ``{passed, score, reason}``
    parsed from its JSON answer, with the call's cost and usage kept on
    the verdict. *resource* is an ``llm:`` key (or a bare name) the hub
    resolves. The 1.9.0 judge, now a traced, versioned, cached
    :class:`Judge`; :func:`judge` (PASS/FAIL with a reason) is the one
    to reach for."""
    return _JsonJudge(resource, rubric, name=name, check_name=name)

Gate dataclass

Gate(
    threshold: PerMetric = None,
    baseline: Optional[str] = None,
    tolerance: PerMetric = None,
    metrics: Optional[Sequence[str]] = None,
    must_pass_tag: Optional[str] = "critical",
    max_error_rate: float = 0.05,
    alpha: float = 0.05,
    strict: bool = False,
    bootstrap: int = 2000,
)

When an eval run passes, as data. See the module docstring.

Parameters:

Name	Type	Description	Default
threshold	PerMetric	Absolute floor(s): a number for the pass rate, or {metric: floor}. A metric under its floor fails the run — exactly Eval(threshold=…)'s 1.9.0 rule.	None
baseline	Optional[str]	What to compare against: "latest" (this eval's last finished run, fixed when the run starts), a run id under its record_dir, or a commit's experiment from the score store: "git:<ref>" is the one at git merge-base HEAD <ref>, "main" is "git:origin/main".	None
tolerance	PerMetric	How large a drop matters, per gated metric (a number, or {metric: number}). Required with a baseline.	None
metrics	Optional[Sequence[str]]	The metrics compared against the baseline that can gate (default ["pass"]); the other checks are compared too, as exploratory.	None
must_pass_tag	Optional[str]	Cases with this tag are a deterministic tier (see the module docstring). None turns it off.	'critical'
max_error_rate	float	The share of trials that may error (failed or timed out) before the run is an infrastructure error.	0.05
alpha	float	The significance level, after Holm adjustment.	0.05
strict	bool	An inconclusive run exits 2 instead of 0.	False
bootstrap	int	Bootstrap replicates for each comparison's CI.	2000

Methods:

gated

gated() -> List[str]

The metrics a baseline comparison may fail the run on.

Source code in operonx/app/evals/gate.py

def gated(self) -> List[str]:
    """The metrics a baseline comparison may fail the run on."""
    return list(self.metrics) if self.metrics else [PASS_METRIC]

tolerance_for

tolerance_for(metric: str) -> Optional[float]

How large a drop of metric matters; None when the gate does not say (a comparison then reports it without a verdict).

Source code in operonx/app/evals/gate.py

def tolerance_for(self, metric: str) -> Optional[float]:
    """How large a drop of *metric* matters; ``None`` when the gate does
    not say (a comparison then reports it without a verdict)."""
    return self._tolerances.get(metric)

named_metrics

named_metrics() -> List[str]

Every metric name the gate mentions, to check against the eval's.

Source code in operonx/app/evals/gate.py

def named_metrics(self) -> List[str]:
    """Every metric name the gate mentions, to check against the eval's."""
    names = set(self._thresholds) | set(self._tolerances)
    if self.metrics:
        names |= set(self.metrics)
    return sorted(names)

compare

compare — any two experiments, paired, the way the gate compares them.

The gate compares a run with its baseline as the run finishes; this is the same comparison (:func:~operonx.app.evals.gate.compare_runs) for any two experiments after the fact — from records or the score store::

a = load_experiment("20261004T101500-000001", store=store)
b = load_experiment("20261004T113000-000002", store=store)
got = compare(a, b, tolerance=0.03)
got["verdict"], got["comparison"]["tests"][0]["diff"]

Without a tolerance every metric is reported — difference, interval, p-value — and nothing is judged: a verdict needs someone to have said how large a drop matters (D15).

Classes

Functions:

compare

compare(
    a: ExperimentData,
    b: ExperimentData,
    *,
    tolerance: PerMetric = None,
    metrics: Optional[Sequence[str]] = None,
    alpha: float = 0.05,
    strict: bool = False,
    bootstrap: int = 2000,
) -> Dict[str, Any]

b against a (the baseline), on the cases both hold unchanged.

Returns {"baseline", "candidate", "verdict", "exit_code", "reasons", "warnings", "comparison"}; verdict and exit_code are None without a tolerance. metrics are the gated ones (default ["pass"]; Holm-adjusted), the rest exploratory (BH).

Source code in operonx/app/evals/compare.py

def compare(
    a: ExperimentData,
    b: ExperimentData,
    *,
    tolerance: PerMetric = None,
    metrics: Optional[Sequence[str]] = None,
    alpha: float = 0.05,
    strict: bool = False,
    bootstrap: int = 2000,
) -> Dict[str, Any]:
    """*b* against *a* (the baseline), on the cases both hold unchanged.

    Returns ``{"baseline", "candidate", "verdict", "exit_code",
    "reasons", "warnings", "comparison"}``; ``verdict`` and ``exit_code``
    are ``None`` without a *tolerance*. *metrics* are the gated ones
    (default ``["pass"]``; Holm-adjusted), the rest exploratory (BH).
    """
    gate = Gate(
        tolerance=tolerance,
        metrics=list(metrics) if metrics else None,
        alpha=alpha,
        strict=strict,
        bootstrap=bootstrap,
        must_pass_tag=None,
    )
    comparison = compare_runs(gate, a.outcomes(), b.outcomes(), baseline_id=a.experiment_id)
    comparison["baseline"] = a.experiment_id
    warnings = comparability(a.experiment_id, a.fingerprint, b.fingerprint, comparison["cases"])
    if a.eval != b.eval:
        warnings.insert(0, f"two different evals: {a.eval!r} and {b.eval!r}")
    judged = [t for t in comparison["tests"] if t.get("verdict")]
    reasons = [
        f"{t['metric']}: {t['diff'] * 100:+.1f} pts [{t['ci_lo'] * 100:+.1f}, "
        f"{t['ci_hi'] * 100:+.1f}], p={t['p_holm']:.3g} ({t['method']}, Holm), tolerance "
        f"{t['tolerance'] * 100:.1f} pts → {t['verdict']}"
        for t in judged
        if t["verdict"] != PASS
    ]
    found = [t["verdict"] for t in judged]
    for m in gate.gated():
        if m in comparison["not_compared"]:
            found.append(INCONCLUSIVE)
            reasons.append(f"{m}: no shared case measured it in both experiments")
    verdict = _worst(found) if tolerance is not None else None
    return {
        "baseline": a.experiment_id,
        "candidate": b.experiment_id,
        "eval": b.eval,
        "verdict": verdict,
        "exit_code": exit_code(verdict, strict) if verdict is not None else None,
        "reasons": reasons,
        "warnings": warnings,
        "alpha": alpha,
        "comparison": comparison,
    }

calibrate

The noise floor and the power of an eval — evidence before a tolerance.

calibrate. Run the same code k times (an A/A test) and measure how much the numbers move on their own. Each case passes with its own probability p_i, estimated from its k·r trials; a case that passed every trial is simulated as deterministic, so more runs see more of the rare flakes. The tolerance the gate needs is then measured through the gate: pairs of synthetic A/A runs drawn from those p_i are compared by :func:~operonx.app.evals.gate.compare_runs — the same interval, McNemar or bootstrap, clusters and all — and a gated metric passes exactly when ci_lo ≥ −tolerance. The tolerance at which 95% of A/A comparisons pass is the 95th percentile of −ci_lo. It includes the case-sampling uncertainty the interval carries, not just the flakes: 40 cases that always pass still cannot rule out a 9-point drop. The table gives it for r = 1, 2, 3, 5 repeats (and the configured r)::

got = calibrate([exp1, exp2, exp3], target=0.03)
got["rows"], got["recommended_repeats"], got["notes"]

power. :func:~operonx.app.evals.stats.paired_sample_size with the discordance two experiments of the eval actually show (:func:discordance).

Classes

Functions:

calibrate

calibrate(
    experiments: Sequence[ExperimentData],
    *,
    target: Optional[float] = None,
    alpha: float = 0.05,
    repeats: Sequence[int] = DEFAULT_REPEATS,
    simulations: int = 200,
    bootstrap: int = 2000,
) -> Dict[str, Any]

The A/A noise floor of experiments — runs of one eval on one commit — and the tolerance the gate needs at each number of repeats.

target is the tolerance wanted (the gate's): the recommendation is the fewest repeats that reach it. See the module docstring.

Source code in operonx/app/evals/calibrate.py

def calibrate(
    experiments: Sequence[ExperimentData],
    *,
    target: Optional[float] = None,
    alpha: float = 0.05,
    repeats: Sequence[int] = DEFAULT_REPEATS,
    simulations: int = 200,
    bootstrap: int = 2000,
) -> Dict[str, Any]:
    """The A/A noise floor of *experiments* — runs of one eval on one
    commit — and the tolerance the gate needs at each number of repeats.

    *target* is the tolerance wanted (the gate's): the recommendation is
    the fewest repeats that reach it. See the module docstring.
    """
    if len(experiments) < 2:
        raise ValueError(
            "calibrate needs at least two runs of the same code (three is the usual: "
            "--runs 3) — one run shows no run-to-run noise"
        )
    evals = sorted({e.eval for e in experiments})
    if len(evals) > 1:
        raise ValueError(f"calibrate compares runs of one eval, not of {evals}")
    shas = sorted(
        {
            str(e.fingerprint["code_version"])
            for e in experiments
            if e.fingerprint.get("code_version")
        }
    )
    ids = _shared(experiments)
    if not ids:
        raise ValueError("the experiments share no unchanged case")
    sets = [e.outcomes() for e in experiments]
    cluster = {c: sets[0][c].cluster for c in ids}
    passes = {c: sum(sum(s[c].passed) for s in sets) for c in ids}
    trials = {c: sum(len(s[c].passed) for s in sets) for c in ids}
    p = {c: passes[c] / trials[c] for c in ids}
    var = {c: _unbiased_var(passes[c], trials[c]) for c in ids}
    configured = int(experiments[0].summary.get("repeats") or 1)
    n = len(ids)

    gate = Gate(alpha=alpha, bootstrap=bootstrap, must_pass_tag=None)
    rows: List[Dict[str, Any]] = []
    for r in sorted({*repeats, configured}):
        rng = random.Random(seed_of("calibrate", *sorted(e.experiment_id for e in experiments), r))

        def draw() -> Dict[str, CaseOutcome]:
            return {
                c: CaseOutcome(
                    c, cluster=cluster[c], passed=[rng.random() < p[c] for _ in range(r)]
                )
                for c in ids
            }

        lows = [_ci_lo(gate, draw(), draw(), f"aa{r}#{s}") for s in range(simulations)]
        row: Dict[str, Any] = {
            "repeats": r,
            # the smallest tolerance (to a tenth of a point) 95% of A/A runs pass
            "tolerance": math.ceil(max(0.0, _quantile([-lo for lo in lows], AA_PASS)) * 1000 - 1e-9)
            / 1000,
        }
        if target is not None:
            row["aa_pass_at_target"] = round(sum(lo >= -target for lo in lows) / len(lows), 4)
        rows.append(row)

    observed: List[Dict[str, Any]] = []
    for i in range(len(experiments)):
        for j in range(i + 1, len(experiments)):
            a, b = experiments[i], experiments[j]
            sa = {c: sets[i][c] for c in ids}
            sb = {c: sets[j][c] for c in ids}
            cmp = compare_runs(gate, sa, sb, baseline_id=a.experiment_id)
            t = next(t for t in cmp["tests"] if t["metric"] == PASS_METRIC)
            observed.append(
                {
                    "a": a.experiment_id,
                    "b": b.experiment_id,
                    "diff": t["diff"],
                    "ci_lo": t["ci_lo"],
                    "ci_hi": t["ci_hi"],
                }
            )

    metrics: Dict[str, Any] = {}
    for name in sorted({m for e in experiments for m in e.metrics}):
        means = [e.metrics[name]["mean"] for e in experiments if name in e.metrics]
        metrics[name] = {
            "means": means,
            "sd": round(stdev(means), 6) if len(means) > 1 else None,
        }

    flaky = [c for c in ids if 0 < passes[c] < trials[c]]
    at_configured = next(row for row in rows if row["repeats"] == configured)
    recommended: Optional[int] = configured
    notes: List[str] = []
    if len(shas) > 1:
        notes.append(
            f"the runs are of different commits {shas}: what moves between them is not "
            "only noise — calibrate runs of one commit"
        )
    if target is not None:
        reach = [row["repeats"] for row in rows if row["tolerance"] <= target]
        recommended = min(reach) if reach else None
        if recommended is None:
            best = min(rows, key=lambda row: row["tolerance"])
            notes.append(
                f"too noisy to gate at tolerance {target * 100:.1f} pts with {n} cases: the "
                f"best of the table ({best['repeats']} repeat(s)) needs "
                f"{best['tolerance'] * 100:.1f} pts. More cases "
                "narrow the interval (`operonx eval power` says how many); so does a larger "
                "tolerance"
            )
        elif at_configured["tolerance"] > target:
            notes.append(
                f"at the configured {configured} repeat(s) A/A runs pass at tolerance "
                f"{target * 100:.1f} pts only {at_configured['aa_pass_at_target']:.0%} of the "
                f"time; {recommended} repeats reach it"
            )
    if flaky:
        notes.append(
            f"{len(flaky)} of {n} cases flipped across the runs; a case that never flipped in "
            f"{min(trials.values())} trials is treated as deterministic, so more runs measure "
            "rare flakes better"
        )
    return {
        "eval": evals[0],
        "experiments": [e.experiment_id for e in experiments],
        "code_versions": shas,
        "runs": len(experiments),
        "cases": n,
        "repeats": configured,
        "alpha": alpha,
        "simulations": simulations,
        "flaky_share": round(len(flaky) / n, 6),
        "flaky_cases": flaky[:50],
        # P(two trials of a case disagree), averaged over cases
        "flip_rate": round(sum(2 * v for v in var.values()) / n, 6),
        "metrics": metrics,
        "rows": rows,
        "observed": observed,
        "target": target,
        "recommended_repeats": recommended,
        "suggested_tolerance": (
            next(row["tolerance"] for row in rows if row["repeats"] == recommended)
            if recommended is not None
            else None
        ),
        "notes": notes,
    }

discordance

discordance(a: ExperimentData, b: ExperimentData) -> Tuple[float, int]

The share of the cases two experiments share (unchanged) on which they differ — mean |s_B − s_A| of the pass share, which for one trial per case is the McNemar discordance (b + c) / n. Returns (p_d, n).

Source code in operonx/app/evals/calibrate.py

def discordance(a: ExperimentData, b: ExperimentData) -> Tuple[float, int]:
    """The share of the cases two experiments share (unchanged) on which
    they differ — ``mean |s_B − s_A|`` of the ``pass`` share, which for one
    trial per case is the McNemar discordance ``(b + c) / n``. Returns
    ``(p_d, n)``."""
    oa, ob = a.outcomes(), b.outcomes()
    ids = [
        c
        for c in sorted(set(oa) & set(ob))
        if not (oa[c].case_hash and ob[c].case_hash and oa[c].case_hash != ob[c].case_hash)
    ]
    if not ids:
        raise ValueError(f"{a.experiment_id} and {b.experiment_id} share no unchanged case")
    diffs = [abs(ob[c].share(PASS_METRIC) - oa[c].share(PASS_METRIC)) for c in ids]  # type: ignore[operator]
    return sum(diffs) / len(diffs), len(ids)

load_experiment

load_experiment(
    ref: Union[str, Path, JobRun],
    *,
    store: Optional[ScoreStore] = None,
    record_dirs: Iterable[Union[str, Path]] = (),
) -> ExperimentData

An experiment by ref: a :class:JobRun, a record directory, or an id — looked up under record_dirs first (<dir>/<eval>/<id>: a record holds the expected values and every verdict field), then in store.

Source code in operonx/app/evals/experiments.py

def load_experiment(
    ref: Union[str, Path, JobRun],
    *,
    store: Optional[ScoreStore] = None,
    record_dirs: Iterable[Union[str, Path]] = (),
) -> ExperimentData:
    """An experiment by *ref*: a :class:`JobRun`, a record directory, or an
    id — looked up under *record_dirs* first (``<dir>/<eval>/<id>``: a
    record holds the expected values and every verdict field), then in
    *store*."""
    if isinstance(ref, JobRun):
        return ExperimentData.from_run(ref)
    path = Path(str(ref))
    if (path / "run.json").is_file():
        return ExperimentData.from_run(JobRun.load(path))
    dirs = [Path(d) for d in record_dirs]
    text = str(ref)
    if "/" not in text and text not in (".", ".."):
        for d in dirs:
            for hit in sorted(d.glob(f"*/{text}/run.json")):
                return ExperimentData.from_run(JobRun.load(hit.parent))
    if store is not None:
        found = ExperimentData.from_store(store, text)
        if found is not None:
            return found
    looked = [f"record dirs {[str(d) for d in dirs]}"] if dirs else []
    if store is not None:
        looked.append("the score store")
    raise ValueError(
        f"no experiment {text!r}: not a run directory"
        + (f", and not in {' or '.join(looked)}" if looked else "")
    )

ExperimentData dataclass

ExperimentData(
    experiment_id: str,
    eval: str,
    status: str,
    started: Optional[str],
    ended: Optional[str],
    summary: Dict[str, Any],
    items: List[Dict[str, Any]] = list(),
    source: str = "",
    path: Optional[Path] = None,
)

One experiment: summary in run.json["eval"]'s shape (counts, metrics, gate, fingerprint, reliability…), and one item per trial — key, case, repeat, status, passed, checks (name → {passed, score?, reason?, …}), error, output, expected (records only), tags, cluster, case_hash, trace_id, ms, cost_usd. source says where it was read: record <path> or store.

Methods:

trials

trials() -> List[Dict[str, Any]]

The items as the statistics read them (gate.outcomes).

Source code in operonx/app/evals/experiments.py

def trials(self) -> List[Dict[str, Any]]:
    """The items as the statistics read them (``gate.outcomes``)."""
    return [
        {
            "case": i["case"],
            "passed": bool(i.get("passed")),
            "checks": {k: bool(c.get("passed")) for k, c in (i.get("checks") or {}).items()},
            "error": bool(i.get("error")) and i.get("status") not in ("ok", "empty"),
            "case_hash": i.get("case_hash"),
            "cluster": i.get("cluster"),
            "tags": i.get("tags") or (),
        }
        for i in self.items
    ]

outcomes

outcomes() -> Dict[str, CaseOutcome]

Every trial grouped by case.

Source code in operonx/app/evals/experiments.py

def outcomes(self) -> Dict[str, CaseOutcome]:
    """Every trial grouped by case."""
    return outcomes(self.trials())

from_run classmethod

from_run(run: JobRun) -> 'ExperimentData'

An eval's job record.

Source code in operonx/app/evals/experiments.py

@classmethod
def from_run(cls, run: JobRun) -> "ExperimentData":
    """An eval's job record."""
    summary = run.meta.get("eval")
    if not isinstance(summary, dict):
        raise ValueError(
            f"run {run.run_id} of {run.job!r} at {run.path} is not an eval run "
            "(its run.json has no 'eval' block)"
        )
    items = []
    for item in run.items:
        v = item.verdict
        if not v:
            continue  # skipped on resume: judged in the run it came from
        row = {
            "key": item.key,
            "case": str(v.get("case", item.key)),
            "repeat": int(v.get("repeat") or 0),
            "status": item.status,
            "passed": bool(v.get("passed")),
            "checks": dict(v.get("checks") or {}),
            "error": item.error or v.get("error"),
            "output": v.get("output"),
            "tags": list(v.get("tags") or []),
            "cluster": v.get("cluster"),
            "case_hash": v.get("case_hash"),
            "trace_id": item.trace_id,
            "ms": item.ms,
            "cost_usd": v.get("cost_usd"),
        }
        if "expected" in v:
            row["expected"] = v["expected"]
        if v.get("output_clipped"):
            row["output_clipped"] = True
        items.append(row)
    return cls(
        experiment_id=run.run_id,
        eval=run.job,
        status=run.status,
        started=run.started,
        ended=run.ended,
        summary=dict(summary),
        items=items,
        source=f"record {run.path}",
        path=run.path,
    )

from_experiment classmethod

from_experiment(exp: Experiment) -> 'ExperimentData'

One store row as an experiment with its summary and no items — what a list shows, read without the items and scores.

Source code in operonx/app/evals/experiments.py

@classmethod
def from_experiment(cls, exp: Experiment) -> "ExperimentData":
    """One store row as an experiment with its summary and no items —
    what a list shows, read without the items and scores."""
    meta = exp.metadata or {}
    summary: Dict[str, Any] = {
        "dataset": meta.get("dataset_path") or exp.dataset,
        **{
            k: meta[k]
            for k in (
                "pass_rate",
                "passed",
                "failed",
                "trials",
                "threshold",
                "reliability",
                "judges",
            )
            if k in meta
        },
        "cases": exp.cases,
        "errored": exp.errored,
        "repeats": exp.repeats,
        "p50_ms": exp.p50_ms,
        "p95_ms": exp.p95_ms,
        "cost_usd": exp.cost_usd,
        "judge_cost_usd": exp.judge_cost_usd,
        "metrics": dict(exp.metrics or {}),
        "gate": dict(exp.gate or {}),
        "fingerprint": {
            **{k: getattr(exp, k) for k in _FINGERPRINT},
            **({"models": meta["models"]} if "models" in meta else {}),
        },
    }
    for k in ("variant", "split"):
        if getattr(exp, k):
            summary[k] = getattr(exp, k)
    return cls(
        experiment_id=exp.experiment_id,
        eval=exp.eval,
        status=exp.status,
        started=_iso(exp.started_at),
        ended=_iso(exp.ended_at),
        summary=summary,
        source="store",
    )

from_store classmethod

from_store(store: ScoreStore, experiment_id: str) -> Optional['ExperimentData']

An experiment's rows in store: the experiment, its items, and each item's scores as its checks. None when it is not there.

Source code in operonx/app/evals/experiments.py

@classmethod
def from_store(cls, store: ScoreStore, experiment_id: str) -> Optional["ExperimentData"]:
    """An experiment's rows in *store*: the experiment, its items, and
    each item's scores as its checks. ``None`` when it is not there."""
    record = store.get_experiment(experiment_id)
    if record is None:
        return None
    exp = record.experiment
    checks: Dict[tuple, Dict[str, Any]] = {}
    limit = max(1, len(record.items)) * max(1, len(exp.metrics)) + 1
    for s in store.scores(ScoreFilter(experiment_id=experiment_id, target="item"), limit):
        check: Dict[str, Any] = {"passed": s.passed}
        if s.data_type == "numeric":
            check["score"] = s.value
        if s.label is not None:
            check["label"] = s.label
        if s.reason:
            check["reason"] = s.reason
        if s.op_id:
            check["op"] = s.op_id
        if s.cost_usd is not None:
            check["cost_usd"] = s.cost_usd
        check.update(s.metadata or {})
        checks.setdefault((s.case_id, int(s.repeat or 0)), {})[s.score_name] = check
    out = cls.from_experiment(exp)
    out.items = [
        {
            "key": it.case_id if exp.repeats <= 1 else f"{it.case_id}#{it.repeat}",
            "case": it.case_id,
            "repeat": it.repeat,
            "status": it.status,
            "passed": bool(it.passed),
            "checks": checks.get((it.case_id, it.repeat), {}),
            "error": it.error,
            "output": it.output,
            "tags": list(it.tags or []),
            "cluster": it.cluster,
            "case_hash": it.case_hash or None,
            "trace_id": it.trace_id,
            "ms": it.ms,
            "cost_usd": it.cost_usd,
        }
        for it in record.items
    ]
    return out

TraceView

TraceView(
    trace_id: str,
    *,
    rows: Optional[Sequence[Mapping[str, Any]]] = None,
    meta: Optional[Mapping[str, Any]] = None,
    trace: Any = None,
)

See the module docstring. Build one with :meth:from_trace, :meth:from_rows, :meth:from_record or :meth:from_store.

Source code in operonx/app/evals/traceview.py

def __init__(
    self,
    trace_id: str,
    *,
    rows: Optional[Sequence[Mapping[str, Any]]] = None,
    meta: Optional[Mapping[str, Any]] = None,
    trace: Any = None,
):
    self.trace_id = str(trace_id)
    self._raw = rows
    self._meta = dict(meta or {})
    self._trace = trace
    self._rows: Optional[List[OpRow]] = None
    self._counted: Optional[List[OpRow]] = None
    self._summary: Optional[RunSummary] = None

Attributes

rows property

rows: List[OpRow]

Every execution, in start order.

counted property

counted: List[OpRow]

The executions that count, in start order: every row but a retried attempt and the steps recorded under one.

input property

input: Any

What came into the run: the first op's inputs at the root, or — when that op takes none, a door's ingress or a source — what it put out. None for a run with no ops.

output property

output: Any

What the run answered: the last op's outputs at the root. None for a run with no ops.

summary property

summary: RunSummary

The run store's summary of these rows (its totals, its counts).

cost_usd property

cost_usd: Optional[float]

Every priced call's cost; None when nothing was priced.

unpriced property

unpriced: int

Calls that reported no price: a cost nobody measured.

Methods:

from_trace classmethod

from_trace(trace: Any) -> 'TraceView'

A live WorkflowTrace. Nothing is read until the view is.

Source code in operonx/app/evals/traceview.py

@classmethod
def from_trace(cls, trace: Any) -> "TraceView":
    """A live ``WorkflowTrace``. Nothing is read until the view is."""
    return cls(str(trace.trace_id), trace=trace)

from_rows classmethod

from_rows(
    rows: Sequence[Mapping[str, Any]], meta: Optional[Mapping[str, Any]] = None
) -> "TraceView"

A stored run: its rows (the shape nodes.jsonl holds) and its meta.json (trace_id, metadata, duration_ms, …).

Source code in operonx/app/evals/traceview.py

@classmethod
def from_rows(
    cls, rows: Sequence[Mapping[str, Any]], meta: Optional[Mapping[str, Any]] = None
) -> "TraceView":
    """A stored run: its rows (the shape ``nodes.jsonl`` holds) and its
    ``meta.json`` (``trace_id``, ``metadata``, ``duration_ms``, …)."""
    meta = dict(meta or {})
    return cls(str(meta.get("trace_id") or ""), rows=list(rows), meta=meta)

from_record classmethod

from_record(record: Any) -> 'TraceView'

A :class:~operonx.telemetry.runs.RunRecord (store.get_run).

Source code in operonx/app/evals/traceview.py

@classmethod
def from_record(cls, record: Any) -> "TraceView":
    """A :class:`~operonx.telemetry.runs.RunRecord` (``store.get_run``)."""
    meta = {"trace_id": record.summary.trace_id, **(record.meta or {})}
    return cls.from_rows(record.nodes, meta)

from_store classmethod

from_store(store: Any, trace_id: str) -> 'TraceView'

The run trace_id from a run store.

Source code in operonx/app/evals/traceview.py

@classmethod
def from_store(cls, store: Any, trace_id: str) -> "TraceView":
    """The run *trace_id* from a run store."""
    record = store.get_run(trace_id)
    if record is None:
        raise LookupError(
            f"no run {trace_id!r} in {type(store).__name__}: was the eval traced into "
            "this store, and is the run still within its retention?"
        )
    return cls.from_record(record)

ops

ops(
    name: Optional[str] = None,
    *,
    type: Optional[str] = None,
    under: Optional[str] = None,
    status: Optional[str] = None,
) -> List[OpRow]

Executions that count (:attr:counted) matching every filter given. name is an op's name or a dotted tail of its full name ("handle.classify"); under is an enclosing subgraph's name (the root's name — the variable that held the engine — never matches).

Source code in operonx/app/evals/traceview.py

def ops(
    self,
    name: Optional[str] = None,
    *,
    type: Optional[str] = None,  # noqa: A002 — the row's field is op_type
    under: Optional[str] = None,
    status: Optional[str] = None,
) -> List[OpRow]:
    """Executions that count (:attr:`counted`) matching every filter
    given. *name* is an op's name or a dotted tail of its full name
    (``"handle.classify"``); *under* is an enclosing subgraph's name
    (the root's name — the variable that held the engine — never
    matches)."""
    out = []
    for r in self.counted:
        if name is not None and not self._named(r, name):
            continue
        if type is not None and r.op_type != type:
            continue
        if under is not None and under not in r.op_full_name.split(".")[1:-1]:
            continue
        if status is not None and r.status != status:
            continue
        out.append(r)
    return out

path

path(
    *,
    types: Optional[Sequence[str]] = None,
    collapse: bool = False,
    children: bool = False,
) -> List[str]

Op names in start order, of the executions that count. Without types, every execution but routing and containers (branch, graph); with it, only those types. collapse merges consecutive repeats (a generator's yields). children adds the steps ops recorded with child(), in their start order.

Source code in operonx/app/evals/traceview.py

def path(
    self,
    *,
    types: Optional[Sequence[str]] = None,
    collapse: bool = False,
    children: bool = False,
) -> List[str]:
    """Op names in start order, of the executions that count. Without
    *types*, every execution but routing and containers (``branch``,
    ``graph``); with it, only those types. ``collapse`` merges
    consecutive repeats (a generator's yields). ``children`` adds the
    steps ops recorded with ``child()``, in their start order."""
    wanted = set(types) if types is not None else None
    names = [
        r.op_name
        for r in self.counted
        if (r.op_type in wanted if wanted is not None else r.op_type not in STRUCTURAL)
        and (children or not r.is_child)
    ]
    if collapse:
        names = [n for i, n in enumerate(names) if i == 0 or names[i - 1] != n]
    return names

llm_calls

llm_calls() -> List[OpRow]

Executions that report cost_usd — LLM calls, as the run store counts them.

Source code in operonx/app/evals/traceview.py

def llm_calls(self) -> List[OpRow]:
    """Executions that report ``cost_usd`` — LLM calls, as the run
    store counts them."""
    return [r for r in self.rows if r.is_llm_call]

tool_calls

tool_calls() -> List[ToolCall]

Every tool call the counted LLM calls asked for, in order, with the tool message an op returned for it when one did.

Source code in operonx/app/evals/traceview.py

def tool_calls(self) -> List[ToolCall]:
    """Every tool call the counted LLM calls asked for, in order, with
    the tool message an op returned for it when one did."""
    answers: Dict[str, Mapping[str, Any]] = {}
    for r in self.counted:
        msg = r.outputs.get("tool_message") if isinstance(r.outputs, Mapping) else None
        if isinstance(msg, Mapping) and msg.get("tool_call_id"):
            answers[str(msg["tool_call_id"])] = msg
    out = []
    for r in self.counted:
        if not r.is_llm_call:
            continue
        for raw in r.outputs.get("tool_calls") or ():
            call = _call_of(raw, r.op_id)
            if call is None:
                continue
            answer = answers.get(str(call.id)) if call.id else None
            if answer is not None:
                call = ToolCall(
                    call.name,
                    call.args,
                    call.id,
                    call.op_id,
                    result=answer.get("content"),
                    status=answer.get("status"),
                )
            out.append(call)
    return out

errors

errors() -> List[OpRow]

Executions that count and did not end ok.

Source code in operonx/app/evals/traceview.py

def errors(self) -> List[OpRow]:
    """Executions that count and did not end ``ok``."""
    return [r for r in self.counted if r.status != "ok"]

as_text

as_text(limit: int = 4000) -> str

The run as a model reads it (a judge's trace_summary): one line per step of :meth:path — name, type, status, time, and its outputs clipped to one line — cut to limit characters.

Source code in operonx/app/evals/traceview.py

def as_text(self, limit: int = 4000) -> str:
    """The run as a model reads it (a judge's ``trace_summary``): one
    line per step of :meth:`path` — name, type, status, time, and its
    outputs clipped to one line — cut to *limit* characters."""
    lines = []
    for r in self.rows:
        if r.op_type in STRUCTURAL:
            continue
        out = json.dumps(r.outputs, ensure_ascii=False, default=str)
        out = out if len(out) <= 200 else out[:199] + "…"
        err = (
            f" error={r.error.strip().splitlines()[-1]}" if r.error and r.error.strip() else ""
        )
        lines.append(
            f"{r.op_name} ({r.op_type}) {r.status} {r.duration_ms:.0f} ms{err} → {out}"
        )
    text = "\n".join(lines)
    return text if len(text) <= limit else text[: limit - 1] + "…"

OpRow dataclass

OpRow(
    op_id: str,
    op_name: str,
    op_full_name: str,
    op_type: str,
    ctx: Tuple[str, ...],
    inputs: Any,
    outputs: Any,
    status: str,
    error: Optional[str],
    start: float,
    duration_ms: float,
    is_yield: bool = False,
    start_time: float = 0.0,
    attempt: int = 1,
    attrs: Mapping[str, Any] = dict(),
)

One execution of one op. start is wall-clock epoch seconds; cost_usd is what the execution reported (None: it reported no price, or is not an LLM call — priced says which).

Attributes

is_child property

is_child: bool

A step an op recorded itself with child(), not an op.

is_llm_call property

is_llm_call: bool

An execution that reports cost_usd — the rule the run store counts llm_calls by. A streaming LLM's token frames do not.

ToolCall dataclass

ToolCall(
    name: str,
    args: Any = dict(),
    id: Optional[str] = None,
    op_id: Optional[str] = None,
    result: Any = None,
    status: Optional[str] = None,
)

A tool call an LLM made: its name and arguments (a dict, or the raw text when it was not JSON), the call id, the LLM execution that made it, and — when an op returned the tool message for that id (the agents' dispatch does) — its result and status.

trajectory

Evaluators over the run, not just its output: the path, the tool calls, one op's output, and the budget.

Each takes the case's trace (a :class:~.traceview.TraceView)::

evaluators=[
    trajectory.ops(["classify", "lookup_order", "reply"], mode="strict"),
    trajectory.tool_calls([{"name": "lookup", "args": {"order_id": "42"}}],
                          mode="superset", args="subset"),
    trajectory.op_output("classify", exact("kind")),
    budget(ms=1500, cost_usd=0.002, llm_calls=2),
]

Modes (AgentEvals'): strict — the same calls in the same order; unordered — the same calls in any order; subset — nothing beyond the reference (every actual call matches a distinct reference call); superset — at least the reference (every reference call matches a distinct actual call). Repeats count: a reference with one a is not a superset of a, a's needs twice. Matching is a maximum matching, so a loose reference entry is never used up by the call a stricter one needed.

Arguments of a tool call: exact — equal; subset — every argument the reference gives is in the call with the same value (the call may have more); ignore — names only. A reference entry may be a bare name, or omit args: it then matches on the name alone.

Without a reference, the case's trajectory gives it ({"ops": [...], "tool_calls": [{"name", "args"}]}); a case with neither is an error on that case, never a silent pass. The score is the share of the reference matched (subset: of the actual calls).

Functions:

ops

ops(
    reference: Optional[Sequence[str]] = None,
    *,
    mode: str = "strict",
    types: Optional[Sequence[str]] = None,
    collapse: bool = False,
    name: Optional[str] = None,
) -> Callable

The op path (trace.path(types=…, collapse=…)) against reference.

Source code in operonx/app/evals/trajectory.py

def ops(
    reference: Optional[Sequence[str]] = None,
    *,
    mode: str = "strict",
    types: Optional[Sequence[str]] = None,
    collapse: bool = False,
    name: Optional[str] = None,
) -> Callable:
    """The op path (``trace.path(types=…, collapse=…)``) against *reference*."""
    _check(mode, MODES, "trajectory.ops mode")
    types = list(types) if types is not None else None

    def trajectory_ops(trace: Any = None, row: Any = None) -> Dict[str, Any]:
        path = _need_trace(trace, "trajectory.ops").path(types=types, collapse=collapse)
        ref = [str(r) for r in _reference(reference, row, "ops", "trajectory.ops")]
        return _compare(path, ref, mode, lambda a, r: a == r, lambda x: x)

    trajectory_ops.eval_name = name or f"trajectory.ops({mode})"
    return trajectory_ops

tool_calls

tool_calls(
    reference: Optional[Sequence[Any]] = None,
    *,
    mode: str = "strict",
    args: str = "exact",
    name: Optional[str] = None,
) -> Callable

The tool calls the LLM calls made (trace.tool_calls()) against reference: names, and arguments as args says.

Source code in operonx/app/evals/trajectory.py

def tool_calls(
    reference: Optional[Sequence[Any]] = None,
    *,
    mode: str = "strict",
    args: str = "exact",
    name: Optional[str] = None,
) -> Callable:
    """The tool calls the LLM calls made (``trace.tool_calls()``) against
    *reference*: names, and arguments as *args* says."""
    _check(mode, MODES, "trajectory.tool_calls mode")
    _check(args, ARGS, "trajectory.tool_calls args")

    def same(call: Any, ref: Tuple[str, Any]) -> bool:
        ref_name, ref_args = ref
        if call.name != ref_name:
            return False
        if args == "ignore" or ref_args is _ANY:
            return True
        if args == "exact":
            return call.args == ref_args
        return (
            isinstance(call.args, Mapping)
            and isinstance(ref_args, Mapping)
            and all(k in call.args and call.args[k] == v for k, v in ref_args.items())
        )

    def show(x: Any) -> Any:
        if isinstance(x, tuple):
            return x[0] if x[1] is _ANY or args == "ignore" else {"name": x[0], "args": x[1]}
        return x.name if args == "ignore" else {"name": x.name, "args": x.args}

    def trajectory_tool_calls(trace: Any = None, row: Any = None) -> Dict[str, Any]:
        calls = _need_trace(trace, "trajectory.tool_calls").tool_calls()
        ref = [
            _ref_call(e) for e in _reference(reference, row, "tool_calls", "trajectory.tool_calls")
        ]
        return _compare(calls, ref, mode, same, show)

    trajectory_tool_calls.eval_name = name or f"trajectory.tool_calls({mode}, args={args})"
    return trajectory_tool_calls

op_output

op_output(
    op: str, check: Any, *, at: str = "last", name: Optional[str] = None
) -> Callable

check (any evaluator) on one op's outputs instead of the case's: output is that execution's outputs (at="first" or "last"), and the verdict's op is its op_id. An op that never ran fails.

Source code in operonx/app/evals/trajectory.py

def op_output(op: str, check: Any, *, at: str = "last", name: Optional[str] = None) -> Callable:
    """*check* (any evaluator) on one op's outputs instead of the case's:
    ``output`` is that execution's outputs (``at="first"`` or ``"last"``),
    and the verdict's ``op`` is its ``op_id``. An op that never ran fails."""
    _check(at, ("first", "last"), "op_output at")
    inner = prepare(check)

    def op_output_check(
        trace: Any = None,
        input: Any = None,
        expected: Any = None,
        row: Any = None,  # noqa: A002
    ) -> Any:
        view = _need_trace(trace, "op_output")
        found = view.first(op) if at == "first" else view.last(op)
        if found is None:
            return {"passed": False, "reason": f"op {op!r} did not run"}
        avail = {
            "input": input,
            "output": found.outputs,
            "expected": expected,
            "row": row,
            "outputs": [found.outputs],
            "trace": view,
        }
        result = inner.fn(**inner.kwargs(avail))
        if inspect.isawaitable(result):

            async def settled() -> Dict[str, Any]:
                return {**verdict_of(await result), "op": found.op_id}

            return settled()
        return {**verdict_of(result), "op": found.op_id}

    op_output_check.eval_name = name or f"op_output({op}:{_name(check)})"
    return op_output_check

budget

budget(
    ms: Optional[float] = None,
    cost_usd: Optional[float] = None,
    tokens: Optional[int] = None,
    llm_calls: Optional[int] = None,
    *,
    name: str = "budget",
) -> Callable

Limits on the case's run, each inclusive: ms (the graph run's duration), cost_usd, tokens (in + out), llm_calls. A cost limit over a run with unpriced calls fails: that cost is unknown.

Source code in operonx/app/evals/trajectory.py

def budget(
    ms: Optional[float] = None,
    cost_usd: Optional[float] = None,
    tokens: Optional[int] = None,
    llm_calls: Optional[int] = None,
    *,
    name: str = "budget",
) -> Callable:
    """Limits on the case's run, each inclusive: ``ms`` (the graph run's
    duration), ``cost_usd``, ``tokens`` (in + out), ``llm_calls``. A cost
    limit over a run with unpriced calls fails: that cost is unknown."""
    limits = {"ms": ms, "cost_usd": cost_usd, "tokens": tokens, "llm_calls": llm_calls}
    limits = {k: v for k, v in limits.items() if v is not None}
    if not limits:
        raise ValueError("budget() needs at least one limit: ms, cost_usd, tokens or llm_calls")

    def within_budget(trace: Any = None) -> Dict[str, Any]:
        view = _need_trace(trace, "budget")
        measured = {
            "ms": round(view.duration_ms, 3),
            # unknown when any call went unpriced; nothing priced and no call costs 0
            "cost_usd": None if view.unpriced else (view.cost_usd or 0.0),
            "tokens": view.tokens,
            "llm_calls": len(view.llm_calls()),
        }
        over = []
        for key, limit in limits.items():
            got = measured[key]
            if got is None:
                over.append(f"cost unknown: {view.unpriced} unpriced call(s)")
            elif got > limit:
                over.append(f"{key} {got:g} > {limit:g}")
        return {
            "passed": not over,
            "reason": "; ".join(over) or None,
            "measured": {k: measured[k] for k in limits},
        }

    within_budget.eval_name = name
    return within_budget

rescore async

rescore(
    run: Union[JobRun, str, Path],
    evaluators: Sequence[Any],
    *,
    store: Any = None,
    dataset: Any = None,
    scores: Any = None,
) -> Rescored

Judge run (a :class:JobRun, or its directory) again with evaluators. store is the run store its traces went to — needed when an evaluator reads trace; dataset overrides the dataset path the run recorded; scores (a ScoreStore) receives the new scores, one per check per judged run, ids by trace and evaluator version — beside the experiment's own, never over them.

Source code in operonx/app/evals/rescoring.py

async def rescore(
    run: Union[JobRun, str, Path],
    evaluators: Sequence[Any],
    *,
    store: Any = None,
    dataset: Any = None,
    scores: Any = None,
) -> Rescored:
    """Judge *run* (a :class:`JobRun`, or its directory) again with
    *evaluators*. ``store`` is the run store its traces went to — needed
    when an evaluator reads ``trace``; ``dataset`` overrides the dataset
    path the run recorded; ``scores`` (a ScoreStore) receives the new
    scores, one per check per judged run, ids by trace and evaluator
    version — beside the experiment's own, never over them."""
    found = _load(run)
    prepared = [prepare(ev) for ev in evaluators]
    judges = [p.name for p in prepared if is_judge(p.ev)]
    if judges:
        raise ValueError(
            f"rescore re-runs deterministic evaluators; {judges} ask a model. "
            "Run the eval again to judge with them"
        )
    wants_trace = any(p.wants("trace") for p in prepared)
    if wants_trace and store is None:
        names = [p.name for p in prepared if p.wants("trace")]
        raise ValueError(
            f"{names} read the case's trace: pass store= (the run store the eval traced "
            "into, e.g. open_run_store({'backend': 'files'}))"
        )
    rows = _rows(found, dataset)
    ev_meta = found.meta["eval"]

    verdicts: Dict[str, Dict[str, Any]] = {}
    trials = []
    for item in found.items:
        old = item.verdict
        if not old:
            continue  # skipped on resume: judged in the run it came from
        verdict = dict(old)
        if item.status in (ITEM_OK, ITEM_EMPTY):
            for key in ("passed", "checks", "judge_cost_usd", "error"):
                verdict.pop(key, None)
            verdict.update(await _again(item, old, rows, prepared, store if wants_trace else None))
        verdicts[item.key] = verdict
        trials.append(trial_of(item.key, verdict, item.ms))
    nums, _, _ = numbers(trials, int(ev_meta.get("repeats") or 1))
    if scores is not None:
        rows = _scores(found, verdicts, prepared)
        await asyncio.to_thread(scores.put_scores, rows)
    return Rescored(found.run_id, found.job, verdicts, nums)

Rescored dataclass

Rescored(
    run_id: str,
    eval: str,
    verdicts: Dict[str, Dict[str, Any]],
    summary: Dict[str, Any],
    skipped: List[str] = list(),
)

A rescore's outcome: the run it judged again, each item's new verdict (by item key), the summary numbers run.json["eval"] would hold for them, and the evaluators left out (an eval's judges).

publish

An eval run as ScoreStore rows — written live, or published from its record.

One set of converters turns a job record into an :class:~operonx.telemetry.scores.Experiment, its :class:~operonx.telemetry.scores.ExperimentItem rows and one :class:~operonx.telemetry.scores.Score per check per item. An eval with scores= sends them while it runs (through :class:ScoreWriter, so a slow or dead store costs the run nothing); :func:publish sends a recorded run — one whose store was down, or one run before the store existed. Both use the same converters on the same record, and score ids are derived from what was judged, so publishing what was already written writes nothing new.

Classes

ScoreWriter

ScoreWriter(store: ScoreStore, name: str)

A ScoreStore behind a :class:BackgroundWriter: :meth:submit queues a row and returns; one thread writes batches, retries, and past that drops and counts. :meth:finish waits up to a timeout, stops, and says what was lost.

Source code in operonx/app/evals/publish.py

def __init__(self, store: ScoreStore, name: str):
    self.store = store
    self.writer = BackgroundWriter(
        self._write,
        name=f"scores:{name}",
        max_queue=100_000,
        batch_size=500,
        flush_interval=0.2,
    )

Methods:

finish

finish(timeout: float) -> Tuple[int, int]

Wait up to timeout seconds, stop; (written, lost) rows.

Source code in operonx/app/evals/publish.py

def finish(self, timeout: float) -> Tuple[int, int]:
    """Wait up to *timeout* seconds, stop; ``(written, lost)`` rows."""
    self.writer.flush(timeout)
    self.writer.close(timeout=0)
    stats = self.writer.stats
    return stats["written"], stats["submitted"] - stats["written"]

Functions:

experiment_of

experiment_of(run: JobRun) -> Experiment

A job record (finished or just started) as its experiment row.

Source code in operonx/app/evals/publish.py

def experiment_of(run: JobRun) -> Experiment:
    """A job record (finished or just started) as its experiment row."""
    meta = run.meta
    ev: Mapping[str, Any] = meta.get("eval") or {}
    fp: Mapping[str, Any] = ev.get("fingerprint") or {}
    gate: Mapping[str, Any] = ev.get("gate") or {}
    dataset = str(ev.get("dataset") or meta.get("dataset") or "")
    verdicts = [i.verdict for i in run.items if i.verdict]
    costs = [v["cost_usd"] for v in verdicts if v.get("cost_usd") is not None]
    ms = [i.ms for i in run.items if i.verdict and i.ms]
    return Experiment(
        experiment_id=run.run_id,
        eval=run.job,
        project=str(meta.get("project") or ""),
        dataset=Path(dataset).stem if dataset else "",
        dataset_version=str(fp.get("dataset_version") or ""),
        graph=str(meta.get("graph") or ""),
        code_version=fp.get("code_version"),
        version_dirty=fp.get("version_dirty"),
        graph_hash=str(fp.get("graph_hash") or ""),
        config_hash=str(fp.get("config_hash") or ""),
        evaluators_hash=str(fp.get("evaluators_hash") or ""),
        operonx_version=str(fp.get("operonx_version") or ""),
        split=(ev.get("selection") or meta.get("selection") or {}).get("split"),
        variant=ev.get("variant") or meta.get("variant"),
        repeats=int(ev.get("repeats") or meta.get("repeats") or 1),
        baseline_id=(gate.get("comparison") or {}).get("baseline"),
        status=run.status,
        started_at=_epoch(run.started) or 0.0,
        ended_at=_epoch(run.ended),
        cases=int(ev.get("cases") or 0),
        errored=int(ev.get("errored") or 0),
        cost_usd=ev["cost_usd"] if "cost_usd" in ev else (round(sum(costs), 10) if costs else None),
        judge_cost_usd=ev.get("judge_cost_usd"),
        p50_ms=ev.get("p50_ms"),
        p95_ms=ev["p95_ms"] if "p95_ms" in ev else (percentile(ms, 95) if ms else None),
        metrics=dict(ev.get("metrics") or {}),
        gate=dict(gate),
        metadata={
            "dataset_path": dataset,
            "record": str(run.path),
            **{
                k: ev[k]
                for k in (
                    "pass_rate",
                    "passed",
                    "failed",
                    "trials",
                    "threshold",
                    "reliability",
                    "judges",
                )
                if k in ev
            },
            **({"models": fp["models"]} if fp.get("models") is not None else {}),
        },
    )

item_of

item_of(experiment_id: str, item: ItemResult) -> Optional[ExperimentItem]

One recorded item as its row; None for one with no verdict (skipped on resume: it belongs to the run that judged it).

Source code in operonx/app/evals/publish.py

def item_of(experiment_id: str, item: ItemResult) -> Optional[ExperimentItem]:
    """One recorded item as its row; ``None`` for one with no verdict
    (skipped on resume: it belongs to the run that judged it)."""
    v = item.verdict
    if not v:
        return None
    return ExperimentItem(
        experiment_id=experiment_id,
        case_id=str(v.get("case", item.key)),
        repeat=int(v.get("repeat") or 0),
        case_hash=str(v.get("case_hash") or ""),
        trace_id=item.trace_id,
        status=item.status,
        ms=float(item.ms or 0.0),
        cost_usd=v.get("cost_usd"),
        tokens_in=int(v.get("tokens_in") or 0),
        tokens_out=int(v.get("tokens_out") or 0),
        passed=v.get("passed"),
        tags=list(v.get("tags") or []),
        cluster=v.get("cluster"),
        output=v.get("output"),
        error=item.error or v.get("error"),
    )

check_score

check_score(
    score_name: str,
    check: Mapping[str, Any],
    *,
    source: str,
    evaluator_version: str,
    **ids: Any,
) -> Score

One check's verdict as a score: numeric when it gave a score.

Source code in operonx/app/evals/publish.py

def check_score(
    score_name: str,
    check: Mapping[str, Any],
    *,
    source: str,
    evaluator_version: str,
    **ids: Any,
) -> Score:
    """One check's verdict as a score: numeric when it gave a score."""
    numeric = check.get("score") is not None
    return Score(
        score_name=score_name,
        source=source,
        data_type="numeric" if numeric else "bool",
        value=float(check["score"]) if numeric else None,
        passed=check.get("passed"),
        label=check.get("label"),
        reason=str(check.get("reason") or check.get("error") or ""),
        op_id=check.get("op"),
        evaluator_version=evaluator_version,
        judge_trace_id=check.get("judge_trace_id"),
        cost_usd=check.get("cost_usd"),
        metadata={
            **{k: v for k, v in check.items() if k not in _SCORE_FIELDS},
            # a check that could not decide (a broken evaluator, a judge answer that
            # did not parse) says so: its `passed` False is not a verdict
            **({"error": str(check["error"])} if check.get("error") else {}),
        },
        **ids,
    )

scores_of

scores_of(
    run_meta: Mapping[str, Any], experiment: Experiment, item: ItemResult
) -> List[Score]

One item's checks as scores. run_meta is the record's (or, live, the eval's describe() with its fingerprint): evaluator versions and which checks are judges come from it.

Source code in operonx/app/evals/publish.py

def scores_of(run_meta: Mapping[str, Any], experiment: Experiment, item: ItemResult) -> List[Score]:
    """One item's checks as scores. *run_meta* is the record's (or, live,
    the eval's ``describe()`` with its fingerprint): evaluator versions and
    which checks are judges come from it."""
    v = item.verdict
    if not v:
        return []
    versions = ((run_meta.get("eval") or {}).get("fingerprint") or {}).get("evaluators") or {}
    judges = set(run_meta.get("judges") or ())
    return [
        check_score(
            name,
            check,
            source="judge" if name in judges else "code",
            evaluator_version=str(versions.get(name) or ""),
            target="item",
            experiment_id=experiment.experiment_id,
            case_id=str(v.get("case", item.key)),
            repeat=int(v.get("repeat") or 0),
            trace_id=item.trace_id,
            origin="eval",
            name=experiment.eval,
            created_at=experiment.started_at,  # one experiment, one point in time
        )
        for name, check in (v.get("checks") or {}).items()
    ]

publish

publish(run: Any, store: ScoreStore) -> Dict[str, int]

Write a recorded eval run (a :class:JobRun or its directory) to store: its experiment, items and scores. Safe to repeat.

Source code in operonx/app/evals/publish.py

def publish(run: Any, store: ScoreStore) -> Dict[str, int]:
    """Write a recorded eval run (a :class:`JobRun` or its directory) to
    *store*: its experiment, items and scores. Safe to repeat."""
    run = run if isinstance(run, JobRun) else JobRun.load(Path(run))
    experiment = experiment_of(run)
    items = [x for x in (item_of(run.run_id, i) for i in run.items) if x is not None]
    scores = [s for i in run.items for s in scores_of(run.meta, experiment, i)]
    store.put_experiment(experiment)
    store.put_items(items)
    store.put_scores(scores)
    return {"experiments": 1, "items": len(items), "scores": len(scores)}

fingerprint

An experiment's identity: what produced its numbers.

Two eval runs are comparable when they ran the same cases with the same evaluators; the difference between them is meant to be the system under test. The fingerprint records each of those, as short hashes::

code_version, version_dirty   the git commit at the eval's root, and
                              whether the tree had changes no commit holds
graph_hash                    the graph's topology, literal params and
                              inline prompts, and each op's source
config_hash                   the resolved resources its ops use (model,
                              temperature, endpoint) — secrets left out
models                        the models those resources name (what a
                              judge of the same model would favour)
dataset_version               the cases, sorted by id
evaluators, evaluators_hash   each evaluator's version, and all of them
operonx_version               the engine

Every hash is sha256 over canonical JSON (sorted keys, no whitespace), cut to 12 hex characters. Nothing here runs per case.

Functions:

fingerprint

fingerprint(
    *,
    graph: Any,
    rows: Sequence[Mapping],
    evaluators: Mapping[str, Any],
    root: Union[str, Path, None] = None,
    code: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]

The fingerprint of an experiment (see the module docstring).

graph is the compiled root GraphOp; evaluators maps each evaluator's name to it; root is where git is asked for the commit, unless code already holds what origin.code_version said (an eval asks git in the background, while its cases run). A graph that cannot serialize (a cycle-rewritten loop refuses to) has graph_hash, config_hash and models None and says why.

Source code in operonx/app/evals/fingerprint.py

def fingerprint(
    *,
    graph: Any,
    rows: Sequence[Mapping],
    evaluators: Mapping[str, Any],
    root: Union[str, Path, None] = None,
    code: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
    """The fingerprint of an experiment (see the module docstring).

    *graph* is the compiled root ``GraphOp``; *evaluators* maps each
    evaluator's name to it; *root* is where git is asked for the commit,
    unless *code* already holds what ``origin.code_version`` said (an
    eval asks git in the background, while its cases run). A graph that
    cannot serialize (a cycle-rewritten loop refuses to) has
    ``graph_hash``, ``config_hash`` and ``models`` ``None`` and says why.
    """
    import operonx
    from operonx.app.origin import code_version

    if code is None:
        code = code_version(Path(root) if root is not None else Path.cwd())
    out: Dict[str, Any] = {
        "code_version": code.get("version"),
        "version_dirty": code.get("version_dirty"),
    }
    try:
        spec = graph.serialize()
    except NotImplementedError as exc:
        out.update(graph_hash=None, config_hash=None, models=None, graph_hash_error=str(exc))
    else:
        configs = config_spec(spec)
        out.update(
            graph_hash=digest(graph_spec(spec)),
            config_hash=digest(configs),
            models=models_of(configs),
        )
    versions, versions_hash = evaluators_version(evaluators)
    out.update(
        dataset_version=dataset_version(rows),
        evaluators=versions,
        evaluators_hash=versions_hash,
        operonx_version=operonx.__version__,
    )
    return out

case_hash

case_hash(row: Mapping) -> str

What a case asks and expects: an edited expected is a new case even though its id stayed.

Source code in operonx/app/evals/fingerprint.py

def case_hash(row: Mapping) -> str:
    """What a case asks and expects: an edited ``expected`` is a new case
    even though its id stayed."""
    return digest({"input": row.get("input"), "expected": row.get("expected")})

dataset_version

dataset_version(rows: Iterable[Mapping]) -> str

The dataset's content, cases sorted by id.

Source code in operonx/app/evals/fingerprint.py

def dataset_version(rows: Iterable[Mapping]) -> str:
    """The dataset's content, cases sorted by id."""
    return digest(sorted((dict(r) for r in rows), key=lambda r: str(r.get("id"))))

evaluator_version

evaluator_version(ev: Any) -> str

An evaluator's eval_version when it declares one; else a hash of its source and of what its closure and defaults hold — so contains("a") and contains("b") differ, and an llm_judge whose rubric changed is a new version. An @op is its body.

Source code in operonx/app/evals/fingerprint.py

def evaluator_version(ev: Any) -> str:
    """An evaluator's ``eval_version`` when it declares one; else a hash
    of its source and of what its closure and defaults hold — so
    ``contains("a")`` and ``contains("b")`` differ, and an ``llm_judge``
    whose rubric changed is a new version. An ``@op`` is its body."""
    declared = getattr(ev, "eval_version", None)
    if declared is not None:
        return str(declared)
    fn = getattr(ev, "__wrapped__", ev)
    state: Dict[str, Any] = {"code": _source_digest(fn)}
    code = getattr(fn, "__code__", None)
    if code is not None and getattr(fn, "__closure__", None):
        state["closure"] = {
            name: canonical(cell.cell_contents)
            for name, cell in zip(code.co_freevars, fn.__closure__)
        }
    if getattr(fn, "__defaults__", None):
        state["defaults"] = canonical(fn.__defaults__)
    if getattr(fn, "__kwdefaults__", None):
        state["kwdefaults"] = canonical(fn.__kwdefaults__)
    if not inspect.isroutine(fn) and hasattr(fn, "__dict__"):
        state["fields"] = canonical(vars(fn))
    return digest(state)

stats

The statistics an eval reports and gates on, in pure Python.

Core operonx has no numpy, and none of this needs it: every function here works on plain lists of floats and is deterministic — the bootstrap takes a seed, and :func:seed_of derives one from text, never from hash() (which Python salts per process).

What each answers:

• :func:wilson — a pass rate's 95% interval that behaves at 0, 1 and small n (45/50 → [0.786, 0.957]).
• :func:mean_se / :func:clustered_se — the standard error of a mean, naive or with cases grouped (turns of one conversation, one scenario's variants), which can be several times the naive one.
• :func:estimate — picks the right one of the three for a metric.
• :func:pass_hat_k — the chance that k trials of a case all pass.
• :func:mcnemar / :func:newcombe_paired — the exact paired test and the paired interval for a binary check.
• :func:paired_bootstrap — a CI and p for the mean paired difference, resampling cases or whole clusters.
• :func:compare_paired — picks McNemar or the bootstrap for one metric.
• :func:holm / :func:benjamini_hochberg — adjusted p-values for several metrics at once (gated, and exploratory).
• :func:paired_sample_size / :func:detectable_drop — how many cases a drop needs, and what drop a number of cases can see.

Functions:

wilson

wilson(successes: int, n: int, confidence: float = 0.95) -> Tuple[float, float]

The Wilson score interval for successes / n.

Source code in operonx/app/evals/stats.py

def wilson(successes: int, n: int, confidence: float = 0.95) -> Tuple[float, float]:
    """The Wilson score interval for ``successes / n``."""
    if n <= 0:
        raise ValueError("wilson() needs at least one trial")
    if not 0 <= successes <= n:
        raise ValueError(f"wilson(): {successes} successes out of {n} trials")
    z = z_of(confidence)
    p = successes / n
    z2n = z * z / n
    center = (p + z2n / 2) / (1 + z2n)
    half = z * math.sqrt(p * (1 - p) / n + z2n / (4 * n)) / (1 + z2n)
    # at 0 and n the bound is exactly 0 (or 1); the formula leaves float dust
    lo = 0.0 if successes == 0 else max(0.0, center - half)
    hi = 1.0 if successes == n else min(1.0, center + half)
    return lo, hi

estimate

estimate(
    values: Sequence[float],
    clusters: Optional[Sequence[Any]] = None,
    *,
    confidence: float = 0.95,
    bounds: Optional[Tuple[float, float]] = None,
) -> Estimate

The mean of per-case values with a confidence interval.

Grouped cases (some cluster holds more than one) → the clustered SE; 0/1 values otherwise → Wilson; anything else → the CLT. bounds clips a normal interval to the values' range (a pass rate's [0, 1]).

Source code in operonx/app/evals/stats.py

def estimate(
    values: Sequence[float],
    clusters: Optional[Sequence[Any]] = None,
    *,
    confidence: float = 0.95,
    bounds: Optional[Tuple[float, float]] = None,
) -> Estimate:
    """The mean of per-case *values* with a confidence interval.

    Grouped cases (some cluster holds more than one) → the clustered SE;
    0/1 values otherwise → Wilson; anything else → the CLT. *bounds*
    clips a normal interval to the values' range (a pass rate's [0, 1]).
    """
    n = len(values)
    if n == 0:
        raise ValueError("estimate() of no values")
    z = z_of(confidence)
    if _clustered(clusters):
        mean = math.fsum(values) / n
        se: Optional[float] = clustered_se(values, clusters)  # type: ignore[arg-type]
        method = "clustered"
    elif _is_binary(values):
        k = int(sum(values))
        lo, hi = wilson(k, n, confidence)
        _, se = mean_se(values)
        return Estimate(n, k / n, se, lo, hi, "wilson")
    else:
        mean, se = mean_se(values)
        method = "clt"
    if se is None:
        return Estimate(n, mean, None, None, None, method)
    lo, hi = mean - z * se, mean + z * se
    if bounds is not None:
        lo, hi = max(bounds[0], lo), min(bounds[1], hi)
    return Estimate(n, mean, se, lo, hi, method)

clustered_se

clustered_se(values: Sequence[float], clusters: Sequence[Any]) -> float

The cluster-robust standard error of the mean: sqrt(Σ_c (Σ_{i∈c} (s_i − s̄))²) / n.

Cases in one cluster move together, so they carry less information than as many independent cases; this is the SE that says so.

Source code in operonx/app/evals/stats.py

def clustered_se(values: Sequence[float], clusters: Sequence[Any]) -> float:
    """The cluster-robust standard error of the mean:
    ``sqrt(Σ_c (Σ_{i∈c} (s_i − s̄))²) / n``.

    Cases in one cluster move together, so they carry less information
    than as many independent cases; this is the SE that says so.
    """
    if len(values) != len(clusters):
        raise ValueError("clustered_se(): one cluster per value")
    n = len(values)
    if n == 0:
        raise ValueError("clustered_se() of no values")
    mean = math.fsum(values) / n
    sums: Dict[Any, float] = {}
    for v, c in zip(values, clusters):
        sums[c] = sums.get(c, 0.0) + (v - mean)
    return math.sqrt(math.fsum(s * s for s in sums.values())) / n

pass_hat_k

pass_hat_k(passes: int, trials: int, k: int) -> float

pass^k: the chance that k trials drawn from a case's trials all pass, given passes of them did — C(c, k) / C(n, k), the unbiased estimate (τ-bench). 4 passes of 5 → pass^3 = 0.4.

Source code in operonx/app/evals/stats.py

def pass_hat_k(passes: int, trials: int, k: int) -> float:
    """pass^k: the chance that *k* trials drawn from a case's *trials*
    all pass, given *passes* of them did — ``C(c, k) / C(n, k)``, the
    unbiased estimate (τ-bench). 4 passes of 5 → pass^3 = 0.4."""
    if not 1 <= k <= trials:
        raise ValueError(f"pass^{k} needs 1 ≤ k ≤ trials ({trials})")
    if not 0 <= passes <= trials:
        raise ValueError(f"pass_hat_k(): {passes} passes out of {trials} trials")
    return math.comb(passes, k) / math.comb(trials, k)

mcnemar

mcnemar(b: int, c: int) -> float

The exact two-sided McNemar test on the discordant pairs: b cases went pass → fail, c fail → pass. p = min(1, 2·P(X ≤ min(b, c))) with X ~ Bin(b + c, ½).

Source code in operonx/app/evals/stats.py

def mcnemar(b: int, c: int) -> float:
    """The exact two-sided McNemar test on the discordant pairs: *b*
    cases went pass → fail, *c* fail → pass.
    ``p = min(1, 2·P(X ≤ min(b, c)))`` with ``X ~ Bin(b + c, ½)``."""
    if b < 0 or c < 0:
        raise ValueError("mcnemar() counts cases; they cannot be negative")
    m = b + c
    if m == 0:
        return 1.0
    tail = sum(math.comb(m, i) for i in range(min(b, c) + 1))
    return min(1.0, 2 * tail / 2**m)

newcombe_paired

newcombe_paired(
    both: int,
    went_bad: int,
    went_good: int,
    neither: int,
    confidence: float = 0.95,
) -> Tuple[float, float]

The CI for B's pass rate minus A's on the same cases (Newcombe 1998, method 10: Wilson intervals for each rate, joined by the correlation of the two runs). both passed in A and B, went_bad passed only in A, went_good only in B, neither in neither.

Unlike a bootstrap it stays honest with few discordant cases: three cases that agree in both runs give about ±56 points, not [0, 0]. A positive correlation is shrunk by n/2 first (Newcombe's correction): without it, when most cases always pass or always fail, the interval covered a 1.5-point drop only 81% of the time at n = 40 (measured); with it, 99.9%.

Source code in operonx/app/evals/stats.py

def newcombe_paired(
    both: int, went_bad: int, went_good: int, neither: int, confidence: float = 0.95
) -> Tuple[float, float]:
    """The CI for B's pass rate minus A's on the same cases (Newcombe 1998,
    method 10: Wilson intervals for each rate, joined by the correlation
    of the two runs). *both* passed in A and B, *went_bad* passed only in
    A, *went_good* only in B, *neither* in neither.

    Unlike a bootstrap it stays honest with few discordant cases: three
    cases that agree in both runs give about ±56 points, not [0, 0]. A
    positive correlation is shrunk by n/2 first (Newcombe's correction):
    without it, when most cases always pass or always fail, the interval
    covered a 1.5-point drop only 81% of the time at n = 40 (measured);
    with it, 99.9%.
    """
    n = both + went_bad + went_good + neither
    if n <= 0:
        raise ValueError("newcombe_paired() of no cases")
    p1, p2 = (both + went_bad) / n, (both + went_good) / n  # A's rate, B's rate
    l1, u1 = wilson(both + went_bad, n, confidence)
    l2, u2 = wilson(both + went_good, n, confidence)
    margins = (both + went_bad) * (went_good + neither) * (both + went_good) * (went_bad + neither)
    agree = both * neither - went_bad * went_good
    if agree > 0:
        agree = max(agree - n / 2, 0.0)
    phi = agree / math.sqrt(margins) if margins else 0.0
    d = p1 - p2  # Newcombe's direction, A − B
    lo = d - math.sqrt(max(0.0, (p1 - l1) ** 2 - 2 * phi * (p1 - l1) * (u2 - p2) + (u2 - p2) ** 2))
    hi = d + math.sqrt(max(0.0, (u1 - p1) ** 2 - 2 * phi * (u1 - p1) * (p2 - l2) + (p2 - l2) ** 2))
    return -hi, -lo

paired_bootstrap

paired_bootstrap(
    diffs: Sequence[float],
    clusters: Optional[Sequence[Any]] = None,
    *,
    replicates: int = 2000,
    seed: int = 0,
    confidence: float = 0.95,
) -> Tuple[float, float, float]

(ci_lo, ci_hi, p) for the mean of per-case differences.

Resamples cases, or whole clusters when clusters groups them. The CI is the percentile interval; p is two-sided against "no change", read off the same replicates (so p < 1 − confidence exactly when the CI leaves out 0, up to the +1 that keeps p above zero).

Source code in operonx/app/evals/stats.py

def paired_bootstrap(
    diffs: Sequence[float],
    clusters: Optional[Sequence[Any]] = None,
    *,
    replicates: int = 2000,
    seed: int = 0,
    confidence: float = 0.95,
) -> Tuple[float, float, float]:
    """``(ci_lo, ci_hi, p)`` for the mean of per-case differences.

    Resamples cases, or whole clusters when *clusters* groups them. The
    CI is the percentile interval; p is two-sided against "no change",
    read off the same replicates (so ``p < 1 − confidence`` exactly when
    the CI leaves out 0, up to the ``+1`` that keeps p above zero).
    """
    if not diffs:
        raise ValueError("paired_bootstrap() of no differences")
    if replicates < 100:
        raise ValueError("paired_bootstrap(): use at least 100 replicates")
    if clusters is not None:
        if len(clusters) != len(diffs):
            raise ValueError("paired_bootstrap(): one cluster per difference")
        groups: Dict[Any, List[float]] = {}
        for d, c in zip(diffs, clusters):
            groups.setdefault(c, []).append(d)
        units = [(_unit(math.fsum(g)), len(g)) for g in groups.values()]
    else:
        units = [(_unit(d), 1) for d in diffs]
    reps = _bootstrap_means(units, replicates, random.Random(seed))
    alpha = 1 - confidence
    lo = reps[max(0, math.floor(replicates * alpha / 2))]
    hi = reps[min(replicates - 1, math.ceil(replicates * (1 - alpha / 2)) - 1)]
    below = sum(1 for r in reps if r <= 0)
    above = sum(1 for r in reps if r >= 0)
    p = min(1.0, 2 * min(below + 1, above + 1) / (replicates + 1))
    return lo, hi, p

compare_paired

compare_paired(
    a: Sequence[float],
    b: Sequence[float],
    clusters: Optional[Sequence[Any]] = None,
    *,
    replicates: int = 2000,
    seed: int = 0,
    confidence: float = 0.95,
) -> Comparison

Compare per-case values of the same cases under A and B.

0/1 values in independent cases → the exact McNemar test and the Newcombe interval (:func:newcombe_paired), both from the 2×2 table. Otherwise (shares over repeats, or grouped cases) → the paired bootstrap's p and percentile interval, resampling cases or clusters.

Source code in operonx/app/evals/stats.py

def compare_paired(
    a: Sequence[float],
    b: Sequence[float],
    clusters: Optional[Sequence[Any]] = None,
    *,
    replicates: int = 2000,
    seed: int = 0,
    confidence: float = 0.95,
) -> Comparison:
    """Compare per-case values of the same cases under A and B.

    0/1 values in independent cases → the exact McNemar test and the
    Newcombe interval (:func:`newcombe_paired`), both from the 2×2 table.
    Otherwise (shares over repeats, or grouped cases) → the paired
    bootstrap's p and percentile interval, resampling cases or clusters.
    """
    if len(a) != len(b):
        raise ValueError("compare_paired(): A and B must hold the same cases")
    if not a:
        raise ValueError("compare_paired() of no cases")
    n = len(a)
    mean_a, mean_b = math.fsum(a) / n, math.fsum(b) / n
    grouped = _clustered(clusters)
    if not grouped and _is_binary(a) and _is_binary(b):
        table = {(1, 1): 0, (1, 0): 0, (0, 1): 0, (0, 0): 0}
        for x, y in zip(a, b):
            table[int(x), int(y)] += 1
        went_bad, went_good = table[1, 0], table[0, 1]
        lo, hi = newcombe_paired(table[1, 1], went_bad, went_good, table[0, 0], confidence)
        return Comparison(
            n=n,
            a=mean_a,
            b=mean_b,
            diff=mean_b - mean_a,
            ci_lo=lo,
            ci_hi=hi,
            p=mcnemar(went_bad, went_good),
            method="mcnemar",
            corr=_corr(a, b),
            regressed=went_bad,
            fixed=went_good,
        )
    lo, hi, p = paired_bootstrap(
        [y - x for x, y in zip(a, b)],
        clusters if grouped else None,
        replicates=replicates,
        seed=seed,
        confidence=confidence,
    )
    return Comparison(
        n=n,
        a=mean_a,
        b=mean_b,
        diff=mean_b - mean_a,
        ci_lo=lo,
        ci_hi=hi,
        p=p,
        method="bootstrap-clustered" if grouped else "bootstrap",
        corr=_corr(a, b),
        units=len(set(clusters)) if grouped else n,  # type: ignore[arg-type]
    )

holm

holm(pvalues: Sequence[float]) -> List[float]

Holm–Bonferroni adjusted p-values (step-down, family-wise error), in the input's order.

Source code in operonx/app/evals/stats.py

def holm(pvalues: Sequence[float]) -> List[float]:
    """Holm–Bonferroni adjusted p-values (step-down, family-wise error),
    in the input's order."""
    m = len(pvalues)
    order = sorted(range(m), key=lambda i: pvalues[i])
    out = [0.0] * m
    running = 0.0
    for rank, i in enumerate(order):
        running = max(running, min(1.0, (m - rank) * pvalues[i]))
        out[i] = running
    return out

benjamini_hochberg

benjamini_hochberg(pvalues: Sequence[float]) -> List[float]

Benjamini–Hochberg adjusted p-values (step-up, false discovery rate), in the input's order.

Source code in operonx/app/evals/stats.py

def benjamini_hochberg(pvalues: Sequence[float]) -> List[float]:
    """Benjamini–Hochberg adjusted p-values (step-up, false discovery
    rate), in the input's order."""
    m = len(pvalues)
    order = sorted(range(m), key=lambda i: pvalues[i], reverse=True)
    out = [0.0] * m
    running = 1.0
    for rank_from_top, i in enumerate(order):
        rank = m - rank_from_top  # 1-based rank in ascending order
        running = min(running, pvalues[i] * m / rank)
        out[i] = min(1.0, running)
    return out

The playground bridge

play

The playground bridge — a service's doors, driven from outside.

The studio's playground puts a toy on a service's doors: a form, a chat box, later a microphone. The run itself must happen in the project's own interpreter (the studio never imports a project), so this module is a small process the studio starts there and talks to in JSON lines::

python -m operonx.app.play [--root DIR]      # stdin: requests, stdout: events

Nothing here is a second way to run a service. A playground session is a :class:~operonx.app.serve.protocol.BoundedSession driven through the same :class:~operonx.app.serve.runner.ServeRunner gate (the service's own on_session hook, its variants, its input contract), the same :func:~operonx.app.serve.runner.serve_session, the same door ops and the same trace consumers as production — only the origin differs: origin=playground, so the run is filed apart and kept 7 days.

Codecs. Toys speak one small protocol — messages {"kind": "text", "text": …}, {"kind": "json", "value": …}, {"kind": "bytes", "size": …} — and a door speaks whatever it speaks. A codec translates between them: :meth:Codec.to_door turns a toy message into an ingress item, :meth:Codec.from_door an egress item into a toy message. The built-ins cover plain transports (http JSON, websocket text); a service whose protocol is its own declares Service(..., playground=MyCodec) (or playground = "module:attr" in [[serve]]). A door with no codec offers no toy. Session hooks can tell a playground session apart by session.meta["playground"].

Audio. A codec that sets audio = {"rate": 8000, …} offers the Voice toy: the browser sends {"kind": "audio", "b64": <pcm16 mono at that rate>} in short batches and plays what comes back the same way. The codec cuts a batch into the door's own frames (to_door_items).

Conditions. A session can run in a worse world than the one it was built in: latency_ms before each inbound item, drop (the share of inbound items lost), noise_dbfs (white noise mixed into audio), silence_ms (quiet audio before the first message) and fail (resource keys that raise for this session only — the resource objects are wrapped once and fail only where a session asked them to).

A simulated user. simulate opens a session and lets an LLM persona play the other side for a number of turns: it waits for the service to go quiet, reads the conversation, answers in character, and ends it when the persona says so. The run records what the persona said, like any session.

Where it is recorded. A playground run goes to the service's local consumers only — files and run stores in the project — so a test session never lands in a production Langfuse; "remote": true on a session sends it everywhere the service traces to.

Re-running one op. rerun runs a single op of a service's graph with the inputs it had in a recorded run — a failing op deep inside a call retried in a second, without making the call. It is recorded as a run of its own (toy=rerun, rerun_of=<run>).

Requests (one JSON object per line; id is echoed back)::

{"op": "describe"}
{"op": "open",  "sid": "s1", "service": "chat", "query": {...}, "variant": null,
                "send": [msg, ...], "end": false}
{"op": "send",  "sid": "s1", "msg": {"kind": "text", "text": "hi"}}
{"op": "end",   "sid": "s1"}
{"op": "simulate", "sid": "s2", "service": "chat", "persona": "a busy parent…", "llm": "llm:x",
                   "turns": 6, "first": "user", "conditions": {...}}
{"op": "rerun", "service": "chat", "op_name": "reply", "inputs": {...}, "of": "run-id"}
{"op": "rerun", "job": "score_calls", "op_name": "scored", "inputs": {...}}

Events: doors, opened, refused, out (an egress item, as a toy message), said (what a simulated user sent), ops (the op executions that finished since the last batch — {"op", "status", "ms"} each — so a canvas can follow the run live; the last batch comes before ended), ended (status, error, duration), rerun, error.

Codec

Toy messages ↔ door items. Subclass for a door with its own protocol.

toys names the toys the codec can drive (form, chat; the voice toy arrives with audio codecs).

Methods:

to_door_items

to_door_items(message: Dict[str, Any]) -> List[Any]

The ingress items one toy message becomes — one, unless the door takes audio in frames smaller than the toy's batches.

Source code in operonx/app/play.py

def to_door_items(self, message: Dict[str, Any]) -> List[Any]:
    """The ingress items one toy message becomes — one, unless the door
    takes audio in frames smaller than the toy's batches."""
    return [self.to_door(message)]

to_toy

to_toy(item: Any) -> Dict[str, Any]

What a client sent the door, as a toy message — the inverse of :meth:to_door, for recording a real session so it can be replayed (Service(replay=True)). A door whose items carry audio inside JSON overrides this to say so ({"kind": "audio", "size": …}): audio is never kept.

Source code in operonx/app/play.py

def to_toy(self, item: Any) -> Dict[str, Any]:
    """What a client sent the door, as a toy message — the inverse of
    :meth:`to_door`, for recording a real session so it can be replayed
    (``Service(replay=True)``). A door whose items carry audio inside
    JSON overrides this to say so (``{"kind": "audio", "size": …}``):
    audio is never kept."""
    if isinstance(item, (bytes, bytearray)):
        return {"kind": "bytes", "size": len(item)}
    return toy_message(item)

JsonCodec

Bases: Codec

An http door: a JSON payload in, JSON replies out — the Form toy.

TextCodec

Bases: Codec

A websocket door: text frames both ways (the Chat toy); JSON frames are sent as objects and come back as events.

PcmCodec

PcmCodec(rate: int = 16000, frame_ms: int = 20)

Bases: Codec

A door that takes raw 16-bit mono PCM frames and sends them back — the Voice toy. Declared, never guessed: Service(..., playground=PcmCodec(rate=16000, frame_ms=20)). What the door sends that is not audio (a transcript, an event) still reaches the toy.

Source code in operonx/app/play.py

def __init__(self, rate: int = 16000, frame_ms: int = 20):
    self.audio = {"rate": int(rate), "encoding": "pcm16", "frame_ms": int(frame_ms)}

codec_for

codec_for(spec: Any) -> Optional[Codec]

The codec a service's playground uses: its declared one, else the built-in for its transport, else None (no toy).

Source code in operonx/app/play.py

def codec_for(spec: Any) -> Optional[Codec]:
    """The codec a service's playground uses: its declared one, else the
    built-in for its transport, else None (no toy)."""
    declared = spec.options.get("playground")
    if declared:
        obj = resolve_ref(declared, field=f"[[serve]] {spec.name!r} playground")
        return obj() if isinstance(obj, type) else obj
    make = BUILTIN_CODECS.get(spec.kind)
    return make() if make else None

Where a run came from

origin

Where a run came from — the tags every run carries.

A run is one graph execution, and every run has exactly one origin: a service answering a client, a job working through a source, an eval, the studio's playground, or anything else (a test, a script — "ad hoc"). The origin travels as trace metadata, which every consumer already reads: the local consumer files the run under it, Langfuse receives it as tags, a run store indexes it. Nothing here does I/O except :func:code_version, which asks git once per process.

::

origin_metadata("service", service="call", transport="websocket")
# {"origin": "service", "service": "call", "transport": "websocket",
#  "tags": ["origin:service", "service:call", "transport:websocket"]}

A job of steps runs its steps inside :func:in_runbook; a step's runs then carry runbook and runbook_run (the names run stores keep, from when that job was a Runbook) beside their own job fields, so its run → its steps' runs → their items → their traces is one path.

origin_metadata

origin_metadata(origin: str, **fields: Any) -> Dict[str, Any]

Trace metadata naming a run's origin: origin, the given fields, and the same pairs as tags (Langfuse filters on tags). Fields that are None are left out.

Source code in operonx/app/origin.py

def origin_metadata(origin: str, **fields: Any) -> Dict[str, Any]:
    """Trace metadata naming a run's origin: ``origin``, the given fields,
    and the same pairs as ``tags`` (Langfuse filters on tags). Fields
    that are ``None`` are left out."""
    if origin not in ORIGINS:
        raise ValueError(f"unknown origin {origin!r}; expected one of {', '.join(ORIGINS)}")
    out: Dict[str, Any] = {"origin": origin}
    out.update({k: v for k, v in fields.items() if v is not None})
    out["tags"] = [f"{k}:{v}" for k, v in out.items() if not isinstance(v, (dict, list))]
    return out

in_runbook

in_runbook(name: str, run_id: str) -> Iterator[None]

Everything a job of steps starts inside this block — its steps, their runs — knows the run it belongs to. Tasks created inside inherit it (asyncio copies the context), so it reaches every item.

Source code in operonx/app/origin.py

@contextmanager
def in_runbook(name: str, run_id: str) -> Iterator[None]:
    """Everything a job of steps starts inside this block — its steps,
    their runs — knows the run it belongs to. Tasks created inside
    inherit it (asyncio copies the context), so it reaches every item."""
    token = _RUNBOOK.set((name, run_id))
    try:
        yield
    finally:
        _RUNBOOK.reset(token)

code_version

code_version(root: Union[str, Path]) -> Dict[str, Any]

{"version": <commit>, "version_dirty": bool} for the git checkout at root, or {} when it is not one (or git is absent).

Read once per process at bootstrap — never per run. "p95 went up" is only actionable as "p95 went up after commit abc123"; a dirty tree says the numbers came from code no commit holds.

Source code in operonx/app/origin.py

def code_version(root: Union[str, Path]) -> Dict[str, Any]:
    """``{"version": <commit>, "version_dirty": bool}`` for the git
    checkout at *root*, or ``{}`` when it is not one (or git is absent).

    Read once per process at bootstrap — never per run. "p95 went up" is
    only actionable as "p95 went up after commit abc123"; a dirty tree
    says the numbers came from code no commit holds."""
    try:
        head = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=str(root),
            capture_output=True,
            text=True,
            timeout=5,
        )
        if head.returncode != 0:
            return {}
        status = subprocess.run(
            ["git", "status", "--porcelain", "--untracked-files=no"],
            cwd=str(root),
            capture_output=True,
            text=True,
            timeout=5,
        )
    except (OSError, subprocess.SubprocessError):
        return {}
    return {
        "version": head.stdout.strip()[:12],
        "version_dirty": bool(status.stdout.strip()) if status.returncode == 0 else False,
    }

Jobs and runbooks are on their own page: operonx.app.jobs. Run stores, retention and alerts: operonx.telemetry.runs.