operonx.telemetry.runs

Where finished runs live, and how they are asked for. A store keeps each run in full and as a summary with per-op rollups; it is a trace consumer and a resource (run_store:). The guide: Runs: stores, retention and alerts.

runs

Run stores — where finished runs live, and how they are asked for.

A :class:RunStore keeps each run twice over: in full (every execution with its values) and as a summary with per-op rollups, so one run opens fully and a month of runs summarises without opening any. It is a trace consumer, so the engine writes into it like into any other; it is a resource (run_store: in resources.yaml), so a project picks its backend in configuration:

• files — run directories plus a SQLite index (the default)
• sqlite — one file
• langfuse — read-only, over runs a LangfuseConsumer shipped

The contract is five methods on purpose; see :mod:.base. :func:project_stores names the stores a project's own [tracing] sinks can be read from, without importing the project.

The contract

RunStore

RunStore(config: Optional[Dict[str, Any]] = None)

Bases: Consumer

See the module docstring.

Source code in operonx/telemetry/runs/base.py

def __init__(self, config: Optional[Dict[str, Any]] = None):
    super().__init__(config=config or {})

Methods:

consume

consume(trace: Any) -> Any

A finished run arrives from the engine: store it.

Source code in operonx/telemetry/runs/base.py

def consume(self, trace: Any) -> Any:
    """A finished run arrives from the engine: store it."""
    return self.put_trace(trace)

put_trace abstractmethod

put_trace(trace: Any) -> RunSummary

Store one finished WorkflowTrace; return its summary.

Source code in operonx/telemetry/runs/base.py

@abstractmethod
def put_trace(self, trace: Any) -> RunSummary:
    """Store one finished ``WorkflowTrace``; return its summary."""

list_runs abstractmethod

list_runs(
    where: Optional[RunFilter] = None,
    order: str = "started_desc",
    limit: int = 50,
    cursor: Optional[str] = None,
) -> Page

Runs matching where, one page at a time.

Source code in operonx/telemetry/runs/base.py

@abstractmethod
def list_runs(
    self,
    where: Optional[RunFilter] = None,
    order: str = "started_desc",
    limit: int = 50,
    cursor: Optional[str] = None,
) -> Page:
    """Runs matching *where*, one page at a time."""

get_run abstractmethod

get_run(trace_id: str) -> Optional[RunRecord]

One run fully, or None when the store has no such run.

Source code in operonx/telemetry/runs/base.py

@abstractmethod
def get_run(self, trace_id: str) -> Optional[RunRecord]:
    """One run fully, or ``None`` when the store has no such run."""

rollups abstractmethod

rollups(where: Optional[RunFilter] = None) -> List[OpRollup]

Every per-op, per-run rollup of the runs where matches.

Source code in operonx/telemetry/runs/base.py

@abstractmethod
def rollups(self, where: Optional[RunFilter] = None) -> List[OpRollup]:
    """Every per-op, per-run rollup of the runs *where* matches."""

op_stats

op_stats(where: Optional[RunFilter] = None) -> List[OpStats]

Each op across the runs where matches — slowest total first. Backends with a query engine may override; the default combines :meth:rollups in Python.

Source code in operonx/telemetry/runs/base.py

def op_stats(self, where: Optional[RunFilter] = None) -> List[OpStats]:
    """Each op across the runs *where* matches — slowest total first.
    Backends with a query engine may override; the default combines
    :meth:`rollups` in Python."""
    return combine_rollups(self.rollups(where))

groups

groups(
    where: Optional[RunFilter] = None, by: Sequence[str] = ("origin", "name")
) -> List[Dict[str, Any]]

Runs where matches, counted per group of by (columns from :data:GROUP_FIELDS): runs, errors, first_started, last_started, cost_usd (None when nothing was priced) and duration_ms (summed). Newest group first. SQL backends override; this default pages through :meth:list_runs.

Source code in operonx/telemetry/runs/base.py

def groups(
    self, where: Optional[RunFilter] = None, by: Sequence[str] = ("origin", "name")
) -> List[Dict[str, Any]]:
    """Runs *where* matches, counted per group of *by* (columns from
    :data:`GROUP_FIELDS`): ``runs``, ``errors``, ``first_started``,
    ``last_started``, ``cost_usd`` (None when nothing was priced) and
    ``duration_ms`` (summed). Newest group first. SQL backends
    override; this default pages through :meth:`list_runs`."""
    by = _check_by(by)
    acc: Dict[tuple, Dict[str, Any]] = {}
    cursor = None
    while True:
        page = self.list_runs(where, limit=500, cursor=cursor)
        for s in page.items:
            key = tuple(getattr(s, f) for f in by)
            g = acc.get(key)
            if g is None:
                g = acc[key] = {
                    **dict(zip(by, key)),
                    "runs": 0,
                    "errors": 0,
                    "first_started": s.started_at,
                    "last_started": s.started_at,
                    "cost_usd": None,
                    "duration_ms": 0.0,
                }
            g["runs"] += 1
            g["errors"] += 1 if s.status == "error" else 0
            g["first_started"] = min(g["first_started"], s.started_at)
            g["last_started"] = max(g["last_started"], s.started_at)
            g["duration_ms"] += s.duration_ms or 0.0
            if s.cost_usd is not None:
                g["cost_usd"] = (g["cost_usd"] or 0.0) + s.cost_usd
        cursor = page.next_cursor
        if not cursor:
            break
    return sorted(acc.values(), key=lambda g: -g["last_started"])

delete_runs abstractmethod

delete_runs(where: RunFilter) -> int

Remove the runs where matches; return how many.

Source code in operonx/telemetry/runs/base.py

@abstractmethod
def delete_runs(self, where: RunFilter) -> int:
    """Remove the runs *where* matches; return how many."""

count

count(where: Optional[RunFilter] = None) -> int

How many runs where matches.

Source code in operonx/telemetry/runs/base.py

def count(self, where: Optional[RunFilter] = None) -> int:
    """How many runs *where* matches."""
    page = self.list_runs(where, limit=1)
    if page.total is not None:
        return page.total
    n, cursor = 0, None
    while True:
        page = self.list_runs(where, limit=500, cursor=cursor)
        n += len(page.items)
        cursor = page.next_cursor
        if not cursor:
            return n

refresh

refresh() -> int

Pick up runs written by something other than this store (the files backend indexes directories a plain LocalConsumer wrote). Returns how many were added; most backends have nothing to do.

Source code in operonx/telemetry/runs/base.py

def refresh(self) -> int:
    """Pick up runs written by something other than this store (the
    files backend indexes directories a plain LocalConsumer wrote).
    Returns how many were added; most backends have nothing to do."""
    return 0

close

close() -> None

Release connections. Idempotent.

Source code in operonx/telemetry/runs/base.py

def close(self) -> None:
    """Release connections. Idempotent."""

The data

RunFilter dataclass

RunFilter(
    origin: Optional[str] = None,
    name: Optional[str] = None,
    status: Optional[str] = None,
    since: Optional[float] = None,
    until: Optional[float] = None,
    version: Optional[str] = None,
    job_run: Optional[str] = None,
    runbook_run: Optional[str] = None,
    trace_ids: Optional[Sequence[str]] = None,
    metadata: Dict[str, Any] = dict(),
    search: Optional[str] = None,
)

Which runs — a small data object, never a query language, so every backend implements it natively.

metadata matches metadata keys exactly ({"session_id": "0912"}); search is a case-insensitive substring over the trace id, the key and the metadata. Times are epoch seconds, since inclusive.

Methods:

matches

matches(s: RunSummary) -> bool

The filter in Python — for backends without a query engine.

Source code in operonx/telemetry/runs/model.py

def matches(self, s: RunSummary) -> bool:
    """The filter in Python — for backends without a query engine."""
    if self.origin and s.origin != self.origin:
        return False
    if self.name and s.name != self.name:
        return False
    if self.status and s.status != self.status:
        return False
    if self.since is not None and s.started_at < self.since:
        return False
    if self.until is not None and s.started_at >= self.until:
        return False
    if self.version and s.version != self.version:
        return False
    if self.job_run and s.job_run != self.job_run:
        return False
    if self.runbook_run and s.runbook_run != self.runbook_run:
        return False
    if self.trace_ids is not None and s.trace_id not in set(self.trace_ids):
        return False
    for k, v in self.metadata.items():
        if str(s.metadata.get(k)) != str(v):
            return False
    if self.search:
        q = self.search.lower()
        hay = " ".join([s.trace_id, s.key or "", json.dumps(s.metadata, default=str)]).lower()
        if q not in hay:
            return False
    return True

RunSummary dataclass

RunSummary(
    trace_id: str,
    workflow: str = "",
    origin: str = "adhoc",
    name: str = "",
    status: str = "ok",
    started_at: float = 0.0,
    duration_ms: float = 0.0,
    executions: int = 0,
    ops: int = 0,
    errors: int = 0,
    first_error: Optional[str] = None,
    cost_usd: Optional[float] = None,
    unpriced: int = 0,
    llm_calls: int = 0,
    tokens_in: int = 0,
    tokens_out: int = 0,
    tokens_cached: int = 0,
    version: Optional[str] = None,
    version_dirty: Optional[bool] = None,
    service: Optional[str] = None,
    transport: Optional[str] = None,
    variant: Optional[str] = None,
    job: Optional[str] = None,
    job_run: Optional[str] = None,
    key: Optional[str] = None,
    runbook: Optional[str] = None,
    runbook_run: Optional[str] = None,
    project: Optional[str] = None,
    session_id: Optional[str] = None,
    user_id: Optional[str] = None,
    request_id: Optional[str] = None,
    metadata: Dict[str, Any] = dict(),
    location: Optional[str] = None,
)

One run, in a line: who made it, how it went, what it cost.

RunRecord dataclass

RunRecord(
    summary: RunSummary,
    nodes: List[Dict[str, Any]] = list(),
    meta: Dict[str, Any] = dict(),
    media_root: Optional[str] = None,
)

One run in full: the summary, the trace-level metadata, and every execution as the row a consumer wrote (inputs and outputs included, media as references relative to media_root).

A row stored with inputs_from (a later record of a generator invocation, see :func:row_of) reads with its invocation's inputs: every store's get_run builds one of these, so every reader gets them, by reference, never copied.

OpRollup dataclass

OpRollup(
    trace_id: str,
    op: str,
    op_type: str = "",
    count: int = 0,
    total_ms: float = 0.0,
    max_ms: float = 0.0,
    errors: int = 0,
    cost_usd: Optional[float] = None,
    unpriced: int = 0,
    tokens_in: int = 0,
    tokens_out: int = 0,
    samples: List[float] = list(),
    exact: bool = True,
)

One op in one run: how often, how long, how costly.

OpStats dataclass

OpStats(
    op: str,
    op_type: str = "",
    runs: int = 0,
    count: int = 0,
    total_ms: float = 0.0,
    avg_ms: float = 0.0,
    p50_ms: float = 0.0,
    p95_ms: float = 0.0,
    p99_ms: float = 0.0,
    max_ms: float = 0.0,
    errors: int = 0,
    cost_usd: Optional[float] = None,
    unpriced: int = 0,
    tokens_in: int = 0,
    tokens_out: int = 0,
    exact: bool = True,
)

One op across every run a filter matched — a Monitor table row.

Page dataclass

Page(
    items: List[RunSummary],
    next_cursor: Optional[str] = None,
    total: Optional[int] = None,
)

One page of runs, and the cursor for the next (None: the end).

summarize

summarize(
    trace_id: str,
    rows: Iterable[Dict[str, Any]],
    meta: Optional[Dict[str, Any]] = None,
    location: Optional[str] = None,
) -> Tuple[RunSummary, List[OpRollup]]

A run's summary and per-op rollups from its rows (the shape nodes.jsonl holds) and its meta.json.

Cost follows operonx's own rule: an op that reports cost_usd is priced when the value is a number (zero included — a declared zero) and unpriced when it is None; the run's cost is None only when nothing reported a price at all.

Source code in operonx/telemetry/runs/model.py

def summarize(
    trace_id: str,
    rows: Iterable[Dict[str, Any]],
    meta: Optional[Dict[str, Any]] = None,
    location: Optional[str] = None,
) -> Tuple[RunSummary, List[OpRollup]]:
    """A run's summary and per-op rollups from its rows (the shape
    ``nodes.jsonl`` holds) and its ``meta.json``.

    Cost follows operonx's own rule: an op that reports ``cost_usd`` is
    priced when the value is a number (zero included — a declared zero)
    and *unpriced* when it is ``None``; the run's cost is ``None`` only
    when nothing reported a price at all.
    """
    meta = dict(meta or {})
    md = dict(meta.get("metadata") or {})
    per_op: Dict[str, OpRollup] = {}
    durations: Dict[str, List[float]] = {}
    first_start: Optional[float] = None
    last_end: Optional[float] = None
    first_wall: Optional[float] = None
    s = RunSummary(
        trace_id=trace_id, workflow=str(meta.get("workflow_name") or ""), location=location
    )

    rows = list(rows)
    # steps that failed inside a retried attempt: the attempt after it decides
    gone = superseded_ids(rows, lambda r, k: r.get(k))
    for row in rows:
        op = str(row.get("op_name") or row.get("op_full_name") or "?")
        dur = float(row.get("duration_ms") or 0.0)
        r = per_op.get(op)
        if r is None:
            r = per_op[op] = OpRollup(
                trace_id=trace_id, op=op, op_type=str(row.get("op_type") or "")
            )
            durations[op] = []
        r.count += 1
        r.total_ms += dur
        r.max_ms = max(r.max_ms, dur)
        durations[op].append(dur)
        s.executions += 1
        start = row.get("start_time")
        if isinstance(start, (int, float)):
            first_start = start if first_start is None else min(first_start, start)
            end = row.get("end_time")
            end = end if isinstance(end, (int, float)) else start + dur / 1000.0
            last_end = end if last_end is None else max(last_end, end)
        wall = row.get("wall_start")
        if isinstance(wall, (int, float)):
            first_wall = wall if first_wall is None else min(first_wall, wall)
        # A retried attempt is not a failure: the attempt after it decides.
        if (
            row.get("status") not in (None, STATUS_OK, STATUS_RETRIED)
            and row.get("op_id") not in gone
        ):
            r.errors += 1
            s.errors += 1
            if s.first_error is None:
                s.first_error = f"{op}: {_first_line(row.get('error') or row.get('status'))}"
        outputs = row.get("outputs")
        if isinstance(outputs, dict) and "cost_usd" in outputs:
            s.llm_calls += 1
            cost = outputs.get("cost_usd")
            if isinstance(cost, (int, float)):
                r.cost_usd = (r.cost_usd or 0.0) + float(cost)
                s.cost_usd = (s.cost_usd or 0.0) + float(cost)
            else:
                r.unpriced += 1
                s.unpriced += 1
            t_in, t_out, t_cached = _usage(outputs)
            r.tokens_in += t_in
            r.tokens_out += t_out
            s.tokens_in += t_in
            s.tokens_out += t_out
            s.tokens_cached += t_cached

    for op, r in per_op.items():
        r.samples, r.exact = _thin(durations[op], MAX_SAMPLES)

    s.ops = len(per_op)
    # The run's own records (`meta.json` "errors") know failures no row
    # shows: a structured LLM step that returned `error`, a subgraph that
    # failed around its children, a loop stopped at its cap. They are in
    # the order the ops failed, so the first is the likeliest cause — the
    # parse failure, not the `PromptError` it caused downstream.
    recorded = meta.get("errors") or {}
    if recorded:
        op_name, record = next(iter(recorded.items()))
        text = record.get("message") or record.get("type")
        s.first_error = f"{op_name.rsplit('.', 1)[-1]}: {_first_line(text)}"
    if meta.get("status") == "running":
        # not ended: what failed so far is counted, the verdict waits
        s.status = "running"
    else:
        s.status = "error" if s.errors or recorded or meta.get("status") == "error" else "ok"
    if meta.get("duration_ms") is not None:
        s.duration_ms = float(meta["duration_ms"])
    elif first_start is not None and last_end is not None:
        s.duration_ms = (last_end - first_start) * 1000.0
    s.started_at = float(meta.get("wall_started_at") or first_wall or 0.0)
    s.metadata = md
    for key in _NAMED:
        value = md.get(key)
        if value is not None:
            setattr(s, key, value if key == "origin" else str(value))
    if md.get("version_dirty") is not None:
        s.version_dirty = bool(md["version_dirty"])
    s.origin = str(md.get("origin") or "adhoc")
    if s.origin in ("job", "eval"):
        s.name = str(md.get("job") or s.workflow)
    elif s.origin in ("service", "playground"):
        s.name = str(md.get("service") or s.workflow)
    else:
        s.name = s.workflow
    return s, list(per_op.values())

percentile

percentile(values: Sequence[float], q: float) -> float

The q-th percentile (0–100), linear between closest ranks.

Source code in operonx/telemetry/runs/model.py

def percentile(values: Sequence[float], q: float) -> float:
    """The q-th percentile (0–100), linear between closest ranks."""
    if not values:
        return 0.0
    xs = sorted(values)
    if len(xs) == 1:
        return float(xs[0])
    pos = (len(xs) - 1) * q / 100.0
    lo = int(pos)
    hi = min(lo + 1, len(xs) - 1)
    return float(xs[lo] + (xs[hi] - xs[lo]) * (pos - lo))

Configuration

RunStoreConfig

Bases: YamlModel

YAML-configurable :class:RunStore.

open_run_store

open_run_store(spec: Optional[Dict[str, Any]] = None) -> RunStore

A store from a plain mapping (the YAML block's fields). The studio opens a project's store this way without importing the project.

Source code in operonx/telemetry/runs/config.py

def open_run_store(spec: Optional[Dict[str, Any]] = None) -> RunStore:
    """A store from a plain mapping (the YAML block's fields). The studio
    opens a project's store this way without importing the project."""
    spec = dict(spec or {})
    backend = str(spec.get("backend") or "files")
    if backend == "files":
        from .files import FilesRunStore

        return FilesRunStore(root=spec.get("root") or "", layout=spec.get("layout") or "origin")
    if backend == "sqlite":
        from .sqlite import SqliteRunStore

        return SqliteRunStore(path=spec.get("path") or "", live=_flag(spec.get("live", True)))
    if backend == "postgres":
        from .postgres import PostgresRunStore

        if not spec.get("dsn"):
            raise ValueError("run_store backend 'postgres' needs dsn")
        return PostgresRunStore(
            spec["dsn"],
            prefix=spec.get("prefix") or "operonx_",
            media_dir=spec.get("media_dir") or "",
            live=_flag(spec.get("live", True)),
        )
    if backend == "mongo":
        from .mongo import MongoRunStore

        if not spec.get("uri"):
            raise ValueError("run_store backend 'mongo' needs uri")
        return MongoRunStore(
            spec["uri"],
            database=spec.get("database") or "operonx",
            prefix=spec.get("prefix") or "",
            media_dir=spec.get("media_dir") or "",
        )
    if backend == "langfuse":
        from .langfuse import LangfuseRunStore

        missing = [k for k in ("host", "public_key", "secret_key") if not spec.get(k)]
        if missing:
            raise ValueError(f"run_store backend 'langfuse' needs {', '.join(missing)}")
        return LangfuseRunStore(spec["host"], spec["public_key"], spec["secret_key"])
    if backend == "clickhouse":
        from .clickhouse import ClickHouseRunStore

        if not spec.get("host"):
            raise ValueError("run_store backend 'clickhouse' needs host")
        ttl = spec.get("ttl_days")
        return ClickHouseRunStore(
            host=spec["host"],
            port=int(spec.get("port") or 0),
            user=spec.get("user") or "default",
            password=spec.get("password") or "",
            database=spec.get("database") or "operonx",
            secure=_flag(spec.get("secure")),
            ttl_days=None if ttl in (None, "") else float(ttl),
            media_dir=spec.get("media_dir") or "",
            media_threshold=int(spec.get("media_threshold") or 1024),
            batch_size=int(spec.get("batch_size") or 10000),
            flush_interval=float(spec.get("flush_interval") or 1.0),
            queue_size=int(spec.get("queue_size") or 1000),
            timeout=float(spec.get("timeout") or 10.0),
            media=str(spec.get("media") or "local"),
            live=_flag(spec.get("live", True)),
        )
    raise ValueError(f"unknown run_store backend {backend!r}; one of {', '.join(BACKENDS)}")

A project's own stores

project_stores

project_stores(
    root: Any, env: Optional[Mapping[str, str]] = None
) -> List[StoreSource]

One :class:StoreSource per trace sink the project's operonx.toml names in [tracing] — the project-wide list, every [tracing.services.<n>] and [tracing.jobs.<n>], and the trace = of any [[serve]] / [[job]] block none of those overrides — each sink once, in that order.

env is the environment to resolve ${VAR} in (default: this process's); the project's .env fills in what it lacks. An empty list when there is no manifest or no [tracing] table. A malformed [tracing] raises :class:~operonx.app.manifest.ManifestError.

Source code in operonx/telemetry/runs/project.py

def project_stores(root: Any, env: Optional[Mapping[str, str]] = None) -> List[StoreSource]:
    """One :class:`StoreSource` per trace sink the project's ``operonx.toml``
    names in ``[tracing]`` — the project-wide list, every
    ``[tracing.services.<n>]`` and ``[tracing.jobs.<n>]``, and the
    ``trace =`` of any ``[[serve]]`` / ``[[job]]`` block none of those
    overrides — each sink once, in that order.

    ``env`` is the environment to resolve ``${VAR}`` in (default: this
    process's); the project's ``.env`` fills in what it lacks. An empty
    list when there is no manifest or no ``[tracing]`` table. A malformed
    ``[tracing]`` raises :class:`~operonx.app.manifest.ManifestError`."""
    raw, resolver = project_files(root, env)
    if raw is None or raw.get("tracing") is None:
        return []
    used = _sinks_in_use(raw, str(resolver.root / "operonx.toml"))
    if not used:
        return []
    levels: Dict[str, List[str]] = {}
    for sink, level in used:
        have = levels.setdefault(sink, [])
        if level not in have:
            have.append(level)
    out = []
    for sink, where in levels.items():
        spec, reason = resolver.store(sink)
        out.append(StoreSource(sink=sink, spec=spec, levels=tuple(where), reason=reason))
    return out

StoreSource dataclass

StoreSource(
    sink: str,
    spec: Optional[Dict[str, Any]],
    levels: Tuple[str, ...],
    reason: str = "",
)

One trace sink of a project, as a store that can be read.

spec is what :func:~operonx.telemetry.runs.open_run_store takes, or None when the sink cannot be read — reason says why. levels are the places in operonx.toml that name the sink ([tracing], [tracing.services.call], ...); source says the same in one line: "[tracing] → trace_clickhouse:default".

Methods:

describe

describe() -> str

The store in a few words, with no credentials: "ClickHouse callbot_traces at 192.168.1.12:8123".

Source code in operonx/telemetry/runs/project.py

def describe(self) -> str:
    """The store in a few words, with no credentials:
    ``"ClickHouse callbot_traces at 192.168.1.12:8123"``."""
    if self.spec is None:
        return f"{self.sink} (unreadable: {self.reason})"
    return describe_spec(self.spec)

open

open() -> Any

The :class:~operonx.telemetry.runs.RunStore itself.

Source code in operonx/telemetry/runs/project.py

def open(self) -> Any:
    """The :class:`~operonx.telemetry.runs.RunStore` itself."""
    if self.spec is None:
        raise ValueError(f"{self.sink} cannot be read: {self.reason}")
    from .config import open_run_store

    return open_run_store(self.spec)

ClickHouse

ClickHouseRunStore

ClickHouseRunStore(
    host: str = "localhost",
    port: int = 0,
    user: str = "default",
    password: str = "",
    database: str = "operonx",
    secure: bool = False,
    ttl_days: Optional[float] = None,
    media_dir: Any = "",
    media_threshold: int = 1024,
    batch_size: int = 10000,
    flush_interval: float = 1.0,
    queue_size: int = 1000,
    timeout: float = 10.0,
    read_timeout: float = 5.0,
    media: Any = "local",
    client: Any = None,
    media_batch_bytes: int = 32 << 20,
    live: bool = True,
)

Bases: ClickHouseConnection, RunStore

See the module docstring.

client= takes a ready client (anything with query, command and insert the way clickhouse_connect's has): tests hand in a fake. read_timeout is how long a read waits for this process's queued runs to land first.

media is where blobs go: "local" (the default) a directory, media_dir; "clickhouse" the media table, written in the same batches as the runs (media_dir is then unused, and media_batch_bytes caps the blob bytes one insert carries); or a :class:~operonx.core.media_store.MediaStore of your own.

live (the default) writes a run while it goes, on live_writer (bounded at :attr:live_queue_size items; past it, live rows are dropped and counted, and the final write still has them all).

Source code in operonx/telemetry/runs/clickhouse.py

def __init__(
    self,
    host: str = "localhost",
    port: int = 0,
    user: str = "default",
    password: str = "",
    database: str = "operonx",
    secure: bool = False,
    ttl_days: Optional[float] = None,
    media_dir: Any = "",
    media_threshold: int = 1024,
    batch_size: int = 10000,
    flush_interval: float = 1.0,
    queue_size: int = 1000,
    timeout: float = 10.0,
    read_timeout: float = 5.0,
    media: Any = "local",
    client: Any = None,
    media_batch_bytes: int = 32 << 20,
    live: bool = True,
):
    if ttl_days is not None and float(ttl_days) < 0:
        raise ValueError(f"ttl_days is {ttl_days}; must be >= 0, or unset")
    self._connect_init(
        host=host,
        port=port,
        user=user,
        password=password,
        database=database,
        secure=secure,
        timeout=timeout,
        client=client,
    )
    super().__init__(config={"host": host, "database": database})
    self.ttl_days = None if ttl_days is None else float(ttl_days)
    self.media_threshold = int(media_threshold)
    self.read_timeout = float(read_timeout)
    if media is None or media in ("", "local"):
        root = resolve_root(media_dir) if media_dir else resolve_root("") / "media"
        media = LocalMediaStore(root)
    elif media == "clickhouse":
        media = ClickHouseMediaStore(self, batch_bytes=media_batch_bytes)
    elif not isinstance(media, MediaStore):
        raise ValueError(f"media is {media!r}; one of 'local', 'clickhouse', or a MediaStore")
    self.media: MediaStore = media
    # The two writer threads take turns: a blob both a live row and the
    # final write reference is then put once (the second sees it in the
    # media store's `seen`), and inserts never interleave on the client.
    self._write_lock = threading.Lock()
    self.writer = BackgroundWriter(
        self._write_batch,
        name=f"clickhouse:{self.host}/{database}",
        max_queue=queue_size,
        batch_size=batch_size,
        flush_interval=flush_interval,
        weight=lambda trace: len(getattr(trace, "nodes", ()) or ()) + 1,
    )
    self.live_writer: Optional[BackgroundWriter] = None
    if live:
        self.live_writer = BackgroundWriter(
            self._write_live,
            name=f"clickhouse-live:{self.host}/{database}",
            max_queue=self.live_queue_size,
            batch_size=batch_size,
            flush_interval=flush_interval,
        )

Methods:

consume

consume(trace: Any) -> bool

The engine's hook: queue the run and return. False when the queue was full and the run was dropped (counted in writer.stats).

Source code in operonx/telemetry/runs/clickhouse.py

def consume(self, trace: Any) -> bool:
    """The engine's hook: queue the run and return. ``False`` when the
    queue was full and the run was dropped (counted in
    ``writer.stats``)."""
    return self.writer.submit(trace)

on_start

on_start(trace: Any) -> None

Live: queue the run's running row.

Source code in operonx/telemetry/runs/clickhouse.py

def on_start(self, trace: Any) -> None:
    """Live: queue the run's ``running`` row."""
    self.live_writer.submit(_Live("start", trace))

on_execution

on_execution(trace: Any, execution: Any) -> None

Live: queue one execution's row at its index in trace.nodes.

Source code in operonx/telemetry/runs/clickhouse.py

def on_execution(self, trace: Any, execution: Any) -> None:
    """Live: queue one execution's row at its index in ``trace.nodes``."""
    self.live_writer.submit(_Live("node", trace, len(trace.nodes) - 1, execution))

build

build(
    trace: Any,
) -> Tuple[RunSummary, List[Any], List[List[Any]], List[List[Any]]]

A finished trace as its rows: (summary, run row, node rows, rollup rows). Blobs go to :attr:media on the way.

Source code in operonx/telemetry/runs/clickhouse.py

def build(self, trace: Any) -> Tuple[RunSummary, List[Any], List[List[Any]], List[List[Any]]]:
    """A finished trace as its rows: ``(summary, run row, node rows,
    rollup rows)``. Blobs go to :attr:`media` on the way."""
    # Values stay raw until node_rows serialises them, in one orjson pass
    # whose default= hook sanitises and offloads: a sanitize walk, an
    # offload walk and a dumps cost ~60 us per execution in Python, and
    # that CPU is taken from the event loop running the next call.
    rows = [row_of(n, trace) for n in trace.nodes]
    meta = meta_of_trace(trace)
    summary, rollups = summarize(str(trace.trace_id), rows, meta, location=None)
    exp = expires_at(summary.started_at, summary.origin, self.ttl_days)
    if isinstance(self.media, ClickHouseMediaStore):
        self.media.expire_with(exp)
    return (
        summary,
        run_row(summary, meta, exp),
        node_rows(summary, rows, exp, json_default(self.media, self.media_threshold)),
        rollup_rows(summary, rollups, exp),
    )

put_trace

put_trace(trace: Any) -> RunSummary

Store one run now, synchronously (scripts, backfills). The engine's path is :meth:consume.

Source code in operonx/telemetry/runs/clickhouse.py

def put_trace(self, trace: Any) -> RunSummary:
    """Store one run now, synchronously (scripts, backfills). The
    engine's path is :meth:`consume`."""
    with self._media_batch():
        summary, run, nodes, rollups = self.build(trace)
        self._insert_all([run], nodes, rollups)
    return summary

flush

flush(timeout: Optional[float] = None) -> bool

Wait for queued runs (and live rows) to be written (or dropped).

Source code in operonx/telemetry/runs/clickhouse.py

def flush(self, timeout: Optional[float] = None) -> bool:
    """Wait for queued runs (and live rows) to be written (or dropped)."""
    live = self.live_writer is None or self.live_writer.flush(timeout)
    return self.writer.flush(timeout) and live

op_stats

op_stats(where: Optional[RunFilter] = None) -> List[OpStats]

Native: sums in ClickHouse, percentiles with quantilesExactInclusive over the kept samples — the same linear-between-ranks rule :func:percentile uses.

Source code in operonx/telemetry/runs/clickhouse.py

def op_stats(self, where: Optional[RunFilter] = None) -> List[OpStats]:
    """Native: sums in ClickHouse, percentiles with
    ``quantilesExactInclusive`` over the kept samples — the same
    linear-between-ranks rule :func:`percentile` uses."""
    self._settle()
    sql, params = self._where(where)
    db = self.database
    rows = self._query(
        "SELECT op, anyIf(op_type, op_type != ''), uniqExact(trace_id), sum(count), "
        "sum(total_ms), max(max_ms), sum(errors), "
        "if(countIf(cost_usd IS NOT NULL) > 0, sum(cost_usd), NULL), sum(unpriced), "
        "sum(tokens_in), sum(tokens_out), "
        "quantilesExactInclusiveArray(0.5, 0.95, 0.99)(samples), min(exact) "
        f"FROM {db}.op_rollups FINAL "
        f"WHERE trace_id IN (SELECT trace_id FROM {db}.runs FINAL{sql}) "
        "GROUP BY op ORDER BY sum(total_ms) DESC",
        params,
    )
    out = []
    for r in rows:
        q = [0.0 if x != x else float(x) for x in (r[11] or [0.0, 0.0, 0.0])]  # NaN → 0
        count, total = int(r[3]), float(r[4])
        out.append(
            OpStats(
                op=r[0],
                op_type=r[1] or "",
                runs=int(r[2]),
                count=count,
                total_ms=total,
                avg_ms=total / count if count else 0.0,
                p50_ms=q[0],
                p95_ms=q[1],
                p99_ms=q[2],
                max_ms=float(r[5]),
                errors=int(r[6]),
                cost_usd=r[7],
                unpriced=int(r[8]),
                tokens_in=int(r[9]),
                tokens_out=int(r[10]),
                exact=bool(r[12]),
            )
        )
    return out

delete_runs

delete_runs(where: RunFilter) -> int

Lightweight DELETE of the matching runs from every table. Blobs stay (other runs may share them): see :meth:prune_media.

Source code in operonx/telemetry/runs/clickhouse.py

def delete_runs(self, where: RunFilter) -> int:
    """Lightweight ``DELETE`` of the matching runs from every table.
    Blobs stay (other runs may share them): see :meth:`prune_media`."""
    self._settle()
    sql, params = self._where(where)
    db = self.database
    ids = [r[0] for r in self._query(f"SELECT trace_id FROM {db}.runs FINAL{sql}", params)]
    for i in range(0, len(ids), 5000):
        chunk = ids[i : i + 5000]
        for table in ("runs", "op_rollups", "nodes"):
            self._command(
                f"DELETE FROM {db}.{table} WHERE trace_id IN {{ids:Array(String)}}",
                {"ids": chunk},
            )
    return len(ids)

prune_media

prune_media(older_than_s: float = 86400.0) -> int

Delete blobs no stored execution references any more, written more than older_than_s ago (younger ones may belong to runs still on their way in). Returns how many were deleted. With media="clickhouse" blobs also expire on their own, with the last run that wrote them.

Source code in operonx/telemetry/runs/clickhouse.py

def prune_media(self, older_than_s: float = 86400.0) -> int:
    """Delete blobs no stored execution references any more, written
    more than *older_than_s* ago (younger ones may belong to runs still
    on their way in). Returns how many were deleted. With
    ``media="clickhouse"`` blobs also expire on their own, with the
    last run that wrote them."""
    self._settle()
    rows = self._query(
        "SELECT DISTINCT arrayJoin(extractAll(concat(inputs, ' ', outputs), {pat:String})) "
        f"FROM {self.database}.nodes",
        {"pat": r'"\$media":\s*"([0-9a-f]{64})"'},
    )
    keep = {r[0] for r in rows}
    cutoff = time.time() - float(older_than_s)
    orphans = [
        sha for sha, written in self.media.keys() if sha not in keep and written < cutoff
    ]
    if isinstance(self.media, ClickHouseMediaStore):
        return self.media.delete_many(orphans)
    return sum(1 for sha in orphans if self.media.delete(sha))

ClickHouseConsumerConfig

Bases: YamlModel

YAML-configurable ClickHouse trace consumer.

Media

media_store

Media blobs: what one is, and where it is kept.

A blob (audio, an image, a PDF, an array) is addressed by the SHA-256 of its bytes, so the same bytes are kept once however many records mention them.

• :func:detect_media names a blob's type from its first bytes (WAV, MP3, OGG/Opus, FLAC, WebM, PNG, JPEG, GIF, WebP, PDF, .npy) and reads a duration where the header carries one. Raw PCM has no header: it takes the rate a :class:~operonx.core.media.Media declares in its mime parameters (audio/L16;rate=16000;channels=1).
• :class:MediaStore is the five-method contract (put, get, exists, delete, keys); :class:LocalMediaStore keeps blobs in a directory. Another backend (S3, MinIO, a database table) implements the same five methods.

Traces use it (:mod:operonx.telemetry.media swaps a run's blobs for references to a store), and so can anything else that keeps blobs by content, such as a document store's page images. It lived in operonx.telemetry.media until 1.14; that module still exports every name.

Nothing here imports a database driver, or anything outside the stdlib.

detect_media

detect_media(data: Any, declared: Optional[str] = None) -> MediaInfo

A blob's type from its magic bytes, else from declared.

The bytes win whenever they identify a format: a producer that labels WAV as audio/mp3 gets audio/wav. When they do not, a declared type is kept, and a declared raw-PCM type (audio/L16, audio/L8, audio/L24, audio/pcm, audio/raw) with a rate= parameter gets a duration from its size, channels= (default 1) and the bit depth (from the L number, else bits=, default 16). Never raises.

Source code in operonx/core/media_store.py

def detect_media(data: Any, declared: Optional[str] = None) -> MediaInfo:
    """A blob's type from its magic bytes, else from *declared*.

    The bytes win whenever they identify a format: a producer that labels
    WAV as ``audio/mp3`` gets ``audio/wav``. When they do not, a declared
    type is kept, and a declared raw-PCM type (``audio/L16``, ``audio/L8``,
    ``audio/L24``, ``audio/pcm``, ``audio/raw``) with a ``rate=`` parameter
    gets a duration from its size, ``channels=`` (default 1) and the bit
    depth (from the ``L`` number, else ``bits=``, default 16). Never raises.
    """
    raw = bytes(data) if isinstance(data, _BYTES) else b""
    try:
        info = _magic(raw)
    except Exception:  # noqa: BLE001 — a malformed header is just unknown
        info = None
    if info is None:
        info = _declared(raw, declared) if declared else MediaInfo()
    info.size = len(raw)
    return info

MediaInfo dataclass

MediaInfo(
    mime: str = OCTET,
    ext: str = "bin",
    size: int = 0,
    duration_s: Optional[float] = None,
    sample_rate: Optional[int] = None,
    channels: Optional[int] = None,
)

What a blob is. duration_s, sample_rate and channels are set only when the bytes (or a declared PCM type) say so.

Methods:

ref

ref(sha: str, store: str) -> Dict[str, Any]

The reference a record (a trace row, say) keeps in place of the bytes.

Source code in operonx/core/media_store.py

def ref(self, sha: str, store: str) -> Dict[str, Any]:
    """The reference a record (a trace row, say) keeps in place of the bytes."""
    out: Dict[str, Any] = {"$media": sha, "mime": self.mime, "size": self.size}
    if self.duration_s is not None:
        out["duration_s"] = round(self.duration_s, 6)
    if self.sample_rate is not None:
        out["sample_rate"] = self.sample_rate
    if self.channels is not None:
        out["channels"] = self.channels
    out["store"] = store
    return out

MediaStore

Bases: ABC

Where blobs live, by SHA-256. A store writes a blob once: :meth:put of bytes it already holds does nothing but return the hash.

Methods:

put abstractmethod

put(data: bytes, info: Optional[MediaInfo] = None) -> str

Keep data (once); return its SHA-256 hex digest.

Source code in operonx/core/media_store.py

@abstractmethod
def put(self, data: bytes, info: Optional[MediaInfo] = None) -> str:
    """Keep *data* (once); return its SHA-256 hex digest."""

get abstractmethod

get(sha: str) -> Optional[bytes]

The bytes, or None when the store has no such blob.

Source code in operonx/core/media_store.py

@abstractmethod
def get(self, sha: str) -> Optional[bytes]:
    """The bytes, or ``None`` when the store has no such blob."""

exists abstractmethod

exists(sha: str) -> bool

Whether the store holds sha.

Source code in operonx/core/media_store.py

@abstractmethod
def exists(self, sha: str) -> bool:
    """Whether the store holds *sha*."""

delete abstractmethod

delete(sha: str) -> bool

Remove sha; return whether there was anything to remove.

Source code in operonx/core/media_store.py

@abstractmethod
def delete(self, sha: str) -> bool:
    """Remove *sha*; return whether there was anything to remove."""

keys abstractmethod

keys() -> Iterator[Tuple[str, float]]

Every blob held, as (sha, written_at epoch seconds).

Source code in operonx/core/media_store.py

@abstractmethod
def keys(self) -> Iterator[Tuple[str, float]]:
    """Every blob held, as ``(sha, written_at epoch seconds)``."""

LocalMediaStore

LocalMediaStore(root: Any)

Bases: MediaStore

Blobs in a directory: <root>/<sha[:2]>/<sha>.<ext>.

The extension is the detected type's, so a file opens in the right tool. Writes go through a temporary file and :func:os.replace, so a reader never sees half a blob and two writers of the same bytes both succeed.

Source code in operonx/core/media_store.py

def __init__(self, root: Any):
    self.root = Path(root)

Methods:

path

path(sha: str) -> Optional[Path]

The file holding sha, or None.

Source code in operonx/core/media_store.py

def path(self, sha: str) -> Optional[Path]:
    """The file holding *sha*, or ``None``."""
    if not isinstance(sha, str) or not _HEX64.match(sha):
        return None
    d = self._dir(sha)
    if not d.is_dir():
        return None
    for p in d.iterdir():
        if p.name.split(".", 1)[0] == sha:
            return p
    return None

media

Trace media: swapping a run's blobs for references to a store.

A trace's large values (audio, images, arrays) are kept apart from the row that mentions them. The row holds a small reference::

{"$media": "<sha256>", "mime": "audio/wav", "size": 48044,
 "duration_s": 1.5, "sample_rate": 16000, "channels": 1, "store": "local"}

and the bytes live in a :class:~operonx.core.media_store.MediaStore, addressed by their SHA-256, so the same audio is kept once however many runs carry it.

• :func:offload_to_store walks a sanitised value and swaps blobs for references; any run store can use it.
• :func:json_default does the same while orjson serialises.

The detector and the store (:func:detect_media, :class:MediaInfo, :class:MediaStore, :class:LocalMediaStore) live in :mod:operonx.core.media_store, since nothing about them is specific to traces. They are exported here too, so imports from this module keep working.

Nothing here imports a database driver.

offload_to_store

offload_to_store(payload: Any, store: MediaStore, threshold: int = 1024) -> Any

payload with its blobs moved into store and replaced by refs.

A :class:~operonx.core.media.Media is always stored, whatever its size: its producer said it is media. bytes-like values and numpy arrays are stored at threshold bytes or more and stay inline below it. A Media holding a URL or path keeps it as {"$media_url": ..., "mime": ...}. The input is not mutated.

Source code in operonx/telemetry/media.py

def offload_to_store(payload: Any, store: MediaStore, threshold: int = 1024) -> Any:
    """*payload* with its blobs moved into *store* and replaced by refs.

    A :class:`~operonx.core.media.Media` is always stored, whatever its
    size: its producer said it is media. ``bytes``-like values and numpy
    arrays are stored at *threshold* bytes or more and stay inline below
    it. A ``Media`` holding a URL or path keeps it as
    ``{"$media_url": ..., "mime": ...}``. The input is not mutated.
    """
    if isinstance(payload, dict):
        return {k: offload_to_store(v, store, threshold) for k, v in payload.items()}
    if isinstance(payload, (list, tuple)):
        return [offload_to_store(v, store, threshold) for v in payload]
    if _is_media(payload):
        data, mime = payload.data, payload.mime_type
        if isinstance(data, _BYTES):
            return _put(bytes(data), store, mime)
        return {"$media_url": str(data), "mime": mime}
    if isinstance(payload, _BYTES):
        if len(payload) >= threshold:
            return _put(bytes(payload), store, None)
        return payload
    if _is_ndarray(payload):
        import io

        import numpy as np

        buf = io.BytesIO()
        np.save(buf, payload, allow_pickle=False)
        raw = buf.getvalue()
        return _put(raw, store, None) if len(raw) >= threshold else payload
    return payload

json_default

json_default(store: MediaStore, threshold: int = 1024) -> Any

A default= hook for orjson.dumps / json.dumps that does what :meth:~operonx.telemetry.consumer.Consumer.sanitize and :func:offload_to_store do, while the value is being serialised.

One pass, and the walk itself runs in orjson's C code: the hook is only called for what JSON cannot hold. A Media (passed through with OPT_PASSTHROUGH_DATACLASS), large bytes and large arrays become refs; small bytes their repr text, small arrays lists, numpy scalars numbers; anything else {"$unserializable": "<type>"}.

Source code in operonx/telemetry/media.py

def json_default(store: MediaStore, threshold: int = 1024) -> Any:
    """A ``default=`` hook for ``orjson.dumps`` / ``json.dumps`` that does
    what :meth:`~operonx.telemetry.consumer.Consumer.sanitize` and
    :func:`offload_to_store` do, while the value is being serialised.

    One pass, and the walk itself runs in orjson's C code: the hook is only
    called for what JSON cannot hold. A ``Media`` (passed through with
    ``OPT_PASSTHROUGH_DATACLASS``), large ``bytes`` and large arrays become
    refs; small ``bytes`` their ``repr`` text, small arrays lists, numpy
    scalars numbers; anything else ``{"$unserializable": "<type>"}``.
    """

    def default(v: Any) -> Any:
        if isinstance(v, _BYTES):
            if len(v) >= threshold:
                return _put(bytes(v), store, None)
            return str(bytes(v))
        if _is_media(v):
            if isinstance(v.data, _BYTES):
                return _put(bytes(v.data), store, v.mime_type)
            return {"$media_url": str(v.data), "mime": v.mime_type}
        if _is_ndarray(v):
            if v.nbytes >= threshold:
                return offload_to_store(v, store, threshold)
            return v.tolist()
        if type(v).__module__ == "numpy" and hasattr(v, "item"):
            return v.item()
        return {"$unserializable": type(v).__name__}

    return default

The background writer

BackgroundWriter

BackgroundWriter(
    sink: Callable[[List[Any]], Any],
    *,
    name: str = "writer",
    max_queue: int = 1000,
    batch_size: int = 1000,
    flush_interval: float = 1.0,
    weight: Optional[Callable[[Any], int]] = None,
    max_retries: int = 3,
    retry_backoff: Tuple[float, float] = (0.5, 30.0),
    exit_timeout: float = 5.0,
)

See the module docstring.

Source code in operonx/telemetry/writer.py

def __init__(
    self,
    sink: Callable[[List[Any]], Any],
    *,
    name: str = "writer",
    max_queue: int = 1000,
    batch_size: int = 1000,
    flush_interval: float = 1.0,
    weight: Optional[Callable[[Any], int]] = None,
    max_retries: int = 3,
    retry_backoff: Tuple[float, float] = (0.5, 30.0),
    exit_timeout: float = 5.0,
):
    if int(max_queue) < 1:
        raise ValueError(f"max_queue must be >= 1, got {max_queue}")
    if int(batch_size) < 1:
        raise ValueError(f"batch_size must be >= 1, got {batch_size}")
    self.sink = sink
    self.name = name
    self.max_queue = int(max_queue)
    self.batch_size = int(batch_size)
    self.flush_interval = max(0.0, float(flush_interval))
    self.weight = weight
    self.max_retries = max(0, int(max_retries))
    self.retry_backoff = (float(retry_backoff[0]), float(retry_backoff[1]))
    self.exit_timeout = float(exit_timeout)
    self.stats: Dict[str, int] = {
        "submitted": 0,
        "written": 0,
        "dropped_full": 0,
        "dropped_failed": 0,
        "failed_batches": 0,
    }
    self._closed = False
    self._outage = False  # a drop or a failure since the last good write
    self._lost_mark = 0
    self._st = _State()
    ref = weakref.ref(self)
    atexit.register(lambda: (lambda w: w and w._at_exit())(ref()))

Attributes

queued property

queued: int

Items waiting, not counting a batch being written.

Methods:

submit

submit(item: Any) -> bool

Queue item; False when it was dropped (full, or closed).

Source code in operonx/telemetry/writer.py

def submit(self, item: Any) -> bool:
    """Queue *item*; ``False`` when it was dropped (full, or closed)."""
    try:
        st = self._st
        if st.pid != os.getpid():
            st = self._st = _State()
        if self._closed:
            return False
        with st.cond:
            if len(st.queue) >= self.max_queue:
                self.stats["dropped_full"] += 1
                first, self._outage = not self._outage, True
            else:
                st.queue.append(item)
                st.pending += 1
                self.stats["submitted"] += 1
                st.cond.notify()
                first = None
        if first is None:
            if st.thread is None:
                self._start(st)
            return True
        if first:
            LOGGER.warning(
                "%s: queue full (%d waiting); runs are being dropped until the "
                "backend keeps up — counted in .stats",
                self.name,
                self.max_queue,
            )
        return False
    except Exception:  # noqa: BLE001 — the caller's run must never see this
        return False

flush

flush(timeout: Optional[float] = None) -> bool

Wait until everything submitted so far is written or dropped. False if timeout (seconds) ran out first.

Source code in operonx/telemetry/writer.py

def flush(self, timeout: Optional[float] = None) -> bool:
    """Wait until everything submitted so far is written or dropped.
    ``False`` if *timeout* (seconds) ran out first."""
    st = self._st
    if st.pid != os.getpid():
        return True
    with st.cond:
        if st.pending == 0:
            return True
        st.flushing += 1
        st.cond.notify_all()
        try:
            end = None if timeout is None else time.monotonic() + timeout
            while st.pending > 0:
                left = None if end is None else end - time.monotonic()
                if left is not None and left <= 0:
                    return False
                st.cond.wait(left)
            return True
        finally:
            st.flushing -= 1

close

close(timeout: float = 5.0) -> None

Write what is queued (up to timeout), then stop the thread. Nothing is accepted afterwards. Idempotent.

Source code in operonx/telemetry/writer.py

def close(self, timeout: float = 5.0) -> None:
    """Write what is queued (up to *timeout*), then stop the thread.
    Nothing is accepted afterwards. Idempotent."""
    if self._closed:
        return
    self._closed = True
    st = self._st
    self.flush(timeout)
    st.stop.set()
    with st.cond:
        st.cond.notify_all()
    if st.thread is not None and st.thread is not threading.current_thread():
        st.thread.join(min(timeout, 1.0))

Retention

retention

How long runs are kept, per origin.

A policy maps an origin to days, or to None for "forever"; an origin the policy does not name is kept. The defaults are the platform plan's: services 30 days, jobs and evals forever, the playground 7 days, ad hoc 30 days. A project overrides them (the studio's Settings writes [studio.retention] in operonx.toml); :func:apply_retention enforces whichever applies.

Classes

Functions:

plan_retention

plan_retention(
    store: RunStore,
    policy: Optional[Mapping[str, Optional[float]]] = None,
    now: Optional[float] = None,
) -> Dict[str, int]

How many runs per origin policy would delete — without deleting.

Source code in operonx/telemetry/runs/retention.py

def plan_retention(
    store: RunStore,
    policy: Optional[Mapping[str, Optional[float]]] = None,
    now: Optional[float] = None,
) -> Dict[str, int]:
    """How many runs per origin *policy* would delete — without deleting."""
    now = time.time() if now is None else now
    policy = DEFAULT_RETENTION if policy is None else policy
    return {
        origin: store.count(RunFilter(origin=origin, until=cut))
        for origin, cut in _cutoffs(policy, now).items()
    }

apply_retention

apply_retention(
    store: RunStore,
    policy: Optional[Mapping[str, Optional[float]]] = None,
    now: Optional[float] = None,
) -> Dict[str, int]

Delete runs older than policy allows; return how many per origin.

Source code in operonx/telemetry/runs/retention.py

def apply_retention(
    store: RunStore,
    policy: Optional[Mapping[str, Optional[float]]] = None,
    now: Optional[float] = None,
) -> Dict[str, int]:
    """Delete runs older than *policy* allows; return how many per origin."""
    now = time.time() if now is None else now
    policy = DEFAULT_RETENTION if policy is None else policy
    if not store.writable:
        return {}
    return {
        origin: store.delete_runs(RunFilter(origin=origin, until=cut))
        for origin, cut in _cutoffs(policy, now).items()
    }

Alerts

alerts

Alerts on runs: a threshold per service or job, checked on the summaries.

An alert watches one origin and name (service:call, job:score_calls) over a trailing window and fires when a number crosses its threshold:

• error_rate — the share of runs that failed;
• p95_ms — the 95th-percentile run duration, or one op's when op is set (a key op: time to first audio, the LLM call);
• cost_per_hour — priced cost in the window, per hour (unpriced runs are counted beside it, never as $0);
• runs — fires when there are fewer than the threshold (a service that went quiet);
• score_mean:<score> — the mean of a score an online eval wrote for those runs; fires when it drops below the threshold (quality going down);
• score_fail_rate:<score> — the share of those scores that failed.

The score metrics read a score store (evaluate(..., scores=)) over the scores written in the window; min_runs counts scores there. It is evaluated from what the run store already keeps — summaries and op rollups, no extra recording. min_runs keeps a window of two runs from paging anyone. :func:step turns an evaluation into what to send: a firing once when it crosses, a reminder every repeat_min while it stays over, a resolved when it comes back. :func:deliver posts the message to a webhook as {"text": …} plus the fields — the shape Slack and Teams incoming webhooks take, and any relay can read.

Classes

AlertState dataclass

AlertState(
    firing: bool = False,
    value: Optional[float] = None,
    runs: int = 0,
    unpriced: int = 0,
    since: float = 0.0,
    until: float = 0.0,
    evaluated_at: float = 0.0,
    fired_at: Optional[float] = None,
    sent_at: Optional[float] = None,
    note: str = "",
    extra: Dict[str, Any] = dict(),
)

What an evaluation found, and what has been sent about it.

Functions:

score_metric

score_metric(metric: str) -> Optional[tuple]

("score_mean", "polite") for "score_mean:polite"; None for a run metric.

Source code in operonx/telemetry/runs/alerts.py

def score_metric(metric: str) -> Optional[tuple]:
    """``("score_mean", "polite")`` for ``"score_mean:polite"``; ``None``
    for a run metric."""
    kind, sep, name = metric.partition(":")
    return (kind, name) if sep and kind in SCORE_METRICS else None

evaluate

evaluate(
    store: RunStore,
    alert: Alert,
    now: Optional[float] = None,
    *,
    scores: Any = None,
) -> AlertState

The alert's number over its window, and whether it crosses. A score metric reads scores (a ScoreStore).

Source code in operonx/telemetry/runs/alerts.py

def evaluate(
    store: RunStore, alert: Alert, now: Optional[float] = None, *, scores: Any = None
) -> AlertState:
    """The alert's number over its window, and whether it crosses. A score
    metric reads *scores* (a ScoreStore)."""
    now = time.time() if now is None else now
    since = now - alert.window_min * 60
    scored = score_metric(alert.metric)
    if scored is not None:
        return _evaluate_scores(scores, alert, scored, since, now)
    f = RunFilter(origin=alert.origin, name=alert.target, since=since, until=now)
    runs = store.list_runs(f, "started_desc", 5000).items
    st = AlertState(runs=len(runs), since=since, until=now, evaluated_at=now)
    if alert.metric == "runs":
        st.value = float(len(runs))
        st.firing = len(runs) < alert.threshold
        return st
    if len(runs) < max(1, alert.min_runs):
        st.note = f"{len(runs)} runs in the window — fewer than {alert.min_runs}, not judged"
        return st
    if alert.metric == "error_rate":
        st.value = sum(1 for r in runs if r.status == "error") / len(runs)
    elif alert.metric == "p95_ms":
        if alert.op:
            rolls = [r for r in store.rollups(f) if r.op == alert.op]
            samples = [x for r in rolls for x in (r.samples or [])] or [r.max_ms for r in rolls]
            if not samples:
                st.note = f"no {alert.op} in the window"
                return st
            st.value = percentile(samples, 95)
        else:
            st.value = percentile([r.duration_ms for r in runs], 95)
    elif alert.metric == "cost_per_hour":
        priced = [r.cost_usd for r in runs if r.cost_usd is not None]
        st.unpriced = sum(1 for r in runs if r.cost_usd is None)
        if not priced:
            st.note = "nothing in the window was priced"
            return st
        st.value = sum(priced) / (alert.window_min / 60)
    st.firing = st.value is not None and st.value > alert.threshold
    return st

step

step(
    alert: Alert, prev: Optional[AlertState], cur: AlertState
) -> Optional[str]

What to send now: "firing", "reminder", "resolved" or nothing. Carries fired_at / sent_at over from prev into cur.

Source code in operonx/telemetry/runs/alerts.py

def step(alert: Alert, prev: Optional[AlertState], cur: AlertState) -> Optional[str]:
    """What to send now: ``"firing"``, ``"reminder"``, ``"resolved"`` or
    nothing. Carries ``fired_at`` / ``sent_at`` over from *prev* into *cur*."""
    was = bool(prev and prev.firing)
    if prev is not None:
        cur.fired_at, cur.sent_at = prev.fired_at, prev.sent_at
    if cur.firing and not was:
        cur.fired_at = cur.sent_at = cur.evaluated_at
        return "firing"
    if cur.firing and was:
        if cur.sent_at is None or cur.evaluated_at - cur.sent_at >= alert.repeat_min * 60:
            cur.sent_at = cur.evaluated_at
            return "reminder"
        return None
    if was and not cur.firing and cur.value is not None:
        cur.fired_at = None
        cur.sent_at = cur.evaluated_at
        return "resolved"
    return None

message

message(
    alert: Alert,
    st: AlertState,
    kind: str,
    *,
    project: str = "",
    link: str = "",
) -> Dict[str, Any]

The webhook body: a text line people read, and the fields.

Source code in operonx/telemetry/runs/alerts.py

def message(
    alert: Alert, st: AlertState, kind: str, *, project: str = "", link: str = ""
) -> Dict[str, Any]:
    """The webhook body: a ``text`` line people read, and the fields."""
    what = (
        f"p95 of {alert.op}"
        if alert.op
        else {
            "error_rate": "error rate",
            "p95_ms": "p95 duration",
            "cost_per_hour": "cost per hour",
            "runs": "runs",
        }[alert.metric]
    )
    head = {"firing": "FIRING", "reminder": "STILL FIRING", "resolved": "RESOLVED", "test": "TEST"}[
        kind
    ]
    cmp = "<" if alert.metric == "runs" else ">"
    text = (
        f"[{head}] {project + ' · ' if project else ''}{alert.origin} {alert.target}: {what} "
        f"{_fmt(alert.metric, st.value)} ({cmp} {_fmt(alert.metric, alert.threshold)}) over the last "
        f"{alert.window_min:g} min, {st.runs} runs" + (f" — {link}" if link else "")
    )
    return {
        "text": text,
        "alert": alert.name,
        "state": kind,
        "metric": alert.metric,
        "op": alert.op,
        "value": st.value,
        "threshold": alert.threshold,
        "runs": st.runs,
        "since": st.since,
        "until": st.until,
        "origin": alert.origin,
        "target": alert.target,
        "project": project,
        "link": link,
    }

deliver

deliver(webhook: str, body: Dict[str, Any], timeout: float = 5.0) -> int

POST body as JSON to webhook; the HTTP status.

Source code in operonx/telemetry/runs/alerts.py

def deliver(webhook: str, body: Dict[str, Any], timeout: float = 5.0) -> int:
    """POST *body* as JSON to *webhook*; the HTTP status."""
    if not webhook.startswith(("http://", "https://")):
        raise ValueError("a webhook is an http(s) URL")
    req = urllib.request.Request(
        webhook,
        data=json.dumps(body).encode("utf-8"),
        method="POST",
        headers={"content-type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=timeout) as res:
        return int(res.status)