Revert "feat(benchmarks): HTTP user-journey performance harness (seeded corpu…"

This reverts commit 7572d965a0.

.github/workflows/benchmark.yml

@@ -1,156 +0,0 @@
-name: Performance Benchmark
-
-# Nightly run of the HTTP user-journey performance benchmark
-# (dev/benchmarks/omnigent). Seeds a sizeable corpus, boots a real server
-# against it, drives the journeys, and uploads the JSON report as an artifact.
-# Runs a backend matrix — SQLite (in-process) and Postgres (a service
-# container, matching prod's Lakebase/Postgres round-trip + pooling profile).
-# A workspace Databricks notebook pulls these artifacts via the GitHub API into
-# a Delta table for the trend dashboard (see dev/benchmarks/omnigent/README.md)
-# — so this workflow only produces artifacts; it never touches Databricks.
-#
-# Scheduled -> runs on the trusted default branch with the repo GITHUB_TOKEN;
-# it reads no PR-authored code. Also dispatchable for an ad-hoc run.
-
-on:
-  schedule:
-    - cron: "37 7 * * *" # 07:37 UTC nightly (off-peak, off the :00 mark)
-  workflow_dispatch:
-    inputs:
-      iterations:
-        description: "Sequential operations per latency run"
-        required: false
-        default: "200"
-      runs:
-        description: "Timed runs per journey"
-        required: false
-        default: "3"
-      sessions:
-        description: "Seeded sessions"
-        required: false
-        default: "5000"
-      items_per_session:
-        description: "Seeded items per session"
-        required: false
-        default: "50"
-
-permissions:
-  contents: read
-
-env:
-  # No web SPA build during `uv sync` (setup.py _build_web_ui): this job never
-  # serves the bundle, and the build otherwise times out on public npm.
-  OMNIGENT_SKIP_WEB_UI: "true"
-  UV_INDEX_URL: https://pypi.org/simple
-  PIP_INDEX_URL: https://pypi.org/simple
-  ITERATIONS: ${{ github.event_name == 'workflow_dispatch' && inputs.iterations || '200' }}
-  RUNS: ${{ github.event_name == 'workflow_dispatch' && inputs.runs || '3' }}
-  SESSIONS: ${{ github.event_name == 'workflow_dispatch' && inputs.sessions || '5000' }}
-  ITEMS: ${{ github.event_name == 'workflow_dispatch' && inputs.items_per_session || '50' }}
-
-concurrency:
-  # Never cancel a scheduled run mid-flight (each is a distinct data point);
-  # coalesce manual dispatches per ref.
-  group: benchmark-${{ github.event_name }}-${{ github.ref }}
-  cancel-in-progress: ${{ github.event_name == 'workflow_dispatch' }}
-
-jobs:
-  benchmark:
-    name: Run benchmark (${{ matrix.backend }})
-    if: github.repository == 'omnigent-ai/omnigent'
-    runs-on: ubuntu-latest
-    timeout-minutes: 30
-    strategy:
-      fail-fast: false
-      matrix:
-        backend: [sqlite, postgres]
-    services:
-      # A Postgres service is defined unconditionally (GitHub Actions has no
-      # per-matrix-value service gating), but only the postgres leg connects to
-      # it — the sqlite leg simply ignores it. postgres:16 mirrors Lakebase's
-      # major version.
-      postgres:
-        image: postgres:16
-        env:
-          POSTGRES_PASSWORD: bench
-          POSTGRES_DB: benchdb
-        ports:
-          - 5432:5432
-        options: >-
-          --health-cmd "pg_isready -U postgres"
-          --health-interval 5s
-          --health-timeout 5s
-          --health-retries 10
-    steps:
-      - name: Check out repo
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
-        with:
-          persist-credentials: false
-
-      - name: Set up Python
-        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
-        with:
-          python-version-file: ".python-version"
-
-      - name: Set up uv
-        uses: astral-sh/setup-uv@38f3f104447c67c051c4a08e39b64a148898af3a # v3
-        with:
-          enable-cache: true
-
-      - name: Install dependencies
-        # `databricks` extra carries psycopg[binary] for the Postgres backend.
-        run: uv sync --extra dev --extra databricks
-
-      # Resolve the DB URI + a stable seed-cache key for this backend. The
-      # cache key binds the DB schema head + seed.py contents + corpus config,
-      # so a schema change or seed edit busts the cache and forces a reseed —
-      # the "you changed the schema, refresh the seed" contract (SQLite only;
-      # the Postgres service is fresh each run so its DB is never cached).
-      - name: Resolve DB target
-        id: db
-        run: |
-          HEAD="$(uv run --no-sync dev/benchmarks/omnigent/seed.py --print-head)"
-          if [[ "${{ matrix.backend }}" == "postgres" ]]; then
-            echo "uri=postgresql+psycopg://postgres:bench@localhost:5432/benchdb" >> "$GITHUB_OUTPUT"
-            echo "cache_path=" >> "$GITHUB_OUTPUT"
-          else
-            echo "uri=sqlite:///$PWD/bench.db" >> "$GITHUB_OUTPUT"
-            echo "cache_path=bench.db" >> "$GITHUB_OUTPUT"
-          fi
-          echo "cache_key=benchdb-${{ matrix.backend }}-$HEAD-${SESSIONS}x${ITEMS}-${{ hashFiles('dev/benchmarks/omnigent/seed.py') }}" >> "$GITHUB_OUTPUT"
-
-      # Reuse a previously-seeded SQLite corpus when schema + seed + config are
-      # unchanged. No-op for the postgres leg (empty path).
-      - name: Restore seeded SQLite corpus
-        if: matrix.backend == 'sqlite'
-        id: seedcache
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v4
-        with:
-          path: ${{ steps.db.outputs.cache_path }}
-          key: ${{ steps.db.outputs.cache_key }}
-
-      - name: Seed corpus
-        # Postgres always seeds (fresh service each run); SQLite seeds only on a
-        # cache miss. seed.py is itself idempotent, so a stray hit is harmless.
-        if: matrix.backend == 'postgres' || steps.seedcache.outputs.cache-hit != 'true'
-        run: |
-          uv run --no-sync dev/benchmarks/omnigent/seed.py \
-            --database-uri "${{ steps.db.outputs.uri }}" \
-            --sessions "$SESSIONS" --items-per-session "$ITEMS"
-
-      - name: Run benchmark
-        run: |
-          uv run --no-sync dev/benchmarks/omnigent/run.py \
-            --database-uri "${{ steps.db.outputs.uri }}" \
-            --iterations "$ITERATIONS" \
-            --runs "$RUNS" \
-            --output "benchmark-results-${{ matrix.backend }}.json"
-
-      - name: Upload benchmark results
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
-        if: always()
-        with:
-          name: benchmark-results-${{ matrix.backend }}-${{ github.run_id }}
-          path: benchmark-results-${{ matrix.backend }}.json
-          retention-days: 90
-          if-no-files-found: warn

.pre-commit-config.yaml

@@ -94,16 +94,6 @@ repos:
         files: ^(pyproject\.toml|omnigent/version\.py)$
         pass_filenames: false
 
-      # Fail if the DB schema changed without refreshing the performance
-      # benchmark's seed (dev/benchmarks/omnigent/seed.py). Check-only — the
-      # fix (regenerate the seed corpus, bump SEED_SCHEMA_REVISION) is manual.
-      - id: check-benchmark-seed-schema
-        name: benchmark seed matches DB schema head
-        language: system
-        entry: .venv/bin/python scripts/check_benchmark_seed_schema.py
-        files: ^omnigent/db/(db_models\.py|migrations/)
-        pass_filenames: false
-
       # Local `uv` runs rewrite uv.lock's registry to whatever index is
       # configured on the developer's machine (e.g. the Databricks PyPI
       # proxy). This OSS repo must always commit the public PyPI URL, so

dev/benchmarks/__init__.py

@@ -1 +0,0 @@
-"""Performance benchmarks (runnable via ``uv run``, not shipped)."""

dev/benchmarks/omnigent/README.md

@@ -1,219 +0,0 @@
-# Omnigent performance benchmark
-
-Baseline, repeatable latency/throughput numbers for key Omnigent user
-journeys, so we can track them over time and catch regressions. Modeled on
-MLflow's `dev/benchmarks/gateway/` workflow.
-
-The harness boots a real `omnigent server`, drives the selected journeys under
-load, prints latency/throughput tables, and writes a versioned JSON report.
-Two families: **HTTP/API journeys** (server + DB, no runner/LLM — fast and
-low-noise) and **full-turn journeys** (a real agent turn through the runner +
-a zero-latency mock LLM). See *Journeys* below.
-
-By default the server boots a fresh, empty SQLite DB, which gives best-case
-numbers that don't move with load. For meaningful results, point it at a
-**pre-seeded corpus** (`seed.py`) and, ideally, at **Postgres** — production
-runs on Databricks Lakebase (Postgres), whose per-query round-trip + pooling
-cost SQLite doesn't have. See *Seeding* and *Backends* below.
-
-## Run it
-
-```bash
-# All journeys, sequential latency (100 iterations × 3 runs each).
-uv run --no-sync dev/benchmarks/omnigent/run.py
-
-# A subset, writing a report for CI artifact upload.
-uv run --no-sync dev/benchmarks/omnigent/run.py \
-    --journeys list_sessions,load_conversation_history \
-    --iterations 200 --runs 3 --output bench.json
-
-# Throughput mode: >1 concurrency drives concurrency-safe journeys as load.
-uv run --no-sync dev/benchmarks/omnigent/run.py \
-    --requests 500 --concurrency 25 --runs 3
-
-# CI gating: exit 1 if a threshold is breached.
-uv run --no-sync dev/benchmarks/omnigent/run.py --max-p50-ms 25 --max-p99-ms 100
-```
-
-`--no-sync` runs against the already-installed venv. (A bare `uv run` may try to
-rebuild the project, which fails in a git worktree without a Node web-UI build;
-`OMNIGENT_SKIP_WEB_UI=true uv sync` prepares the venv once, then use
-`--no-sync`.)
-
-Key flags (`--help` for all): `--journeys A,B`, `--database-uri URI` (seeded
-corpus / Postgres; default: throwaway empty SQLite), `--iterations N` (per
-latency run), `--requests N` / `--concurrency N` (throughput), `--runs N`,
-`--warmup N`, `--output FILE`, `--min-rps` / `--max-p50-ms` / `--max-p99-ms`
-(CI thresholds).
-
-## Journeys
-
-### HTTP/API (server + DB, runner-free)
-
-| Journey | Operation timed | Stressed by |
-| --- | --- | --- |
-| `list_sessions` | `GET /v1/sessions` — session-list read | session count |
-| `create_session` | `POST /v1/sessions` then `DELETE` — session create | write path |
-| `get_session` | `GET /v1/sessions/{id}` — single-session snapshot | (O(1)) |
-| `load_conversation_history` | `GET /v1/sessions/{id}/items` — history read | items/session |
-| `search_sessions` | `GET /v1/sessions?search_query=` — unindexed `LIKE` | total item count |
-
-Read journeys target a **pre-seeded** session when the DB has a corpus; against
-an empty DB they self-seed a small fallback session over HTTP (the
-`external_conversation_item` event — appends items without starting a task), so
-they still work with no runner or LLM.
-
-### Full-turn (runner + mock LLM)
-
-These drive a real agent turn end-to-end — `POST …/events` → server → **runner**
-→ in-process executor → mock LLM → stream back → `idle`. Selecting any of them
-boots `BenchEnvironment(with_runner=True)` automatically.
-
-| Journey | Operation timed |
-| --- | --- |
-| `session_cold_start` | Create+bind a fresh session and drive its first turn to `idle` (runner spawn + executor construction + turn) |
-| `warm_turn` | Drive a turn on an already-warm session — steady-state dispatch overhead |
-| `time_to_first_token` | Post a turn; time to the first streamed `output_text` delta |
-| `interrupt` | Interrupt a running (gated) turn; time to cancellation |
-
-**Only measure what we control.** Full-turn journeys always use the
-**`openai-agents`** SDK harness, which runs **in-process** (a call into the
-`agents` library + an HTTP call to the mock LLM) — no vendor binary, no external
-process. Native harnesses (e.g. `claude-native`) launch the real vendor CLI
-into a tmux pane, whose startup we don't control, so they're deliberately
-excluded. The mock LLM is zero-latency, so every number is omnigent
-dispatch/streaming/cancel overhead, not model latency.
-
-Add a journey by registering a `Journey` in `journeys.py` (set `needs_runner`
-for full-turn journeys).
-
-## Seeding a realistic corpus
-
-`seed.py` writes a sizeable, deterministic corpus directly through the store
-API (no HTTP, no runner) into the same DB the server then boots against:
-
-```bash
-# Seed 5000 sessions × 50 items into a SQLite file, then benchmark against it.
-uv run --no-sync dev/benchmarks/omnigent/seed.py \
-    --database-uri sqlite:////abs/path/bench.db --sessions 5000 --items-per-session 50
-uv run --no-sync dev/benchmarks/omnigent/run.py \
-    --database-uri sqlite:////abs/path/bench.db --output bench.json
-```
-
-Seeding is **idempotent**: a matching corpus (same sessions/items/schema) is
-detected and reused, so re-running is a fast no-op — pass `--reseed` to force,
-or a differing config to be warned. SQLite absolute paths need four slashes
-(`sqlite:////abs/...`). The seed is pinned to the DB schema via
-`SEED_SCHEMA_REVISION`; a schema change requires bumping it (guarded — see
-*Schema drift*, phase 2).
-
-## Backends
-
-`--database-uri` selects the DB; the report's `backend` field (`sqlite` /
-`postgres`) is derived from the URI scheme so results group by backend.
-
-- **SQLite** (default) — in-process; fast, but not prod-representative.
-- **Postgres** — `postgresql+psycopg://user@host:5432/db` (the fully-qualified
-  `+psycopg` form; the server CLI does not normalize a bare `postgresql://`).
-  Requires `psycopg[binary]` (the `databricks` extra). Matches prod's
-  round-trip/pooling profile. Stand up a local one with
-  `docker run -e POSTGRES_PASSWORD=… -p 5432:5432 postgres:16`.
-
-## Output → Databricks → dashboard
-
-The harness writes JSON only. Storage and charting live in Databricks:
-
-```
-run.py --output bench.json   →   GitHub Actions artifact   →   Databricks notebook (ETL)   →   Delta table   →   AI/BI dashboard
-        (this repo)                    (CI, follow-up)              (workspace, yours)
-```
-
-The repo's contract is the **JSON schema** below. A workspace notebook (owned
-outside this repo, modeled on MLflow's gateway ETL) pulls the CI artifacts via
-the GitHub API, flattens each run's `summary` + `runs` + metadata, and
-`saveAsTable`s into a Delta table the dashboard reads. `sample_output.json` is a
-committed, faithful example so the notebook can be written against a real
-document without running the harness.
-
-### JSON schema (`schema.py`, `SCHEMA_VERSION`)
-
-```jsonc
-{
-  "schema_version": 1,
-  "generated_at": "<ISO-8601 UTC>",
-  "git_sha": "<HEAD sha>",
-  "git_branch": "<branch>",
-  "host": {"platform": "...", "python": "...", "cpu_count": 12},
-  "harness": "http-only",
-  "config": {"iterations": 100, "requests": 500, "concurrency": 1,
-             "runs": 3, "warmup": 10, "with_runner": false,
-             "backend": "sqlite"},
-  "journeys": {
-    "<journey name>": {
-      "kind": "latency" | "throughput",
-      "backend": "sqlite" | "postgres",
-      "runs": [                       // one per --runs
-        {"n_success": N, "n_failures": N, "failures": {"HTTP 500": 1},
-         "wall_time_s": …, "mean_ms": …, "p50_ms": …, "p95_ms": …,
-         "p99_ms": …, "max_ms": …, "rps": …}
-      ],
-      "summary": {"avg_mean_ms": …, "avg_p50_ms": …, "avg_p95_ms": …,
-                  "avg_p99_ms": …, "avg_rps": …}    // averaged across runs
-    }
-  }
-}
-```
-
-The per-journey `summary` + `runs` shape mirrors MLflow's gateway benchmark, so
-the same ETL flatten works — keyed by `journey` and `backend`. Bump
-`SCHEMA_VERSION` on any breaking shape change so the notebook can branch on it.
-
-## Layout
-
-| File | Role |
-| --- | --- |
-| `run.py` | CLI orchestrator + entrypoint |
-| `seed.py` | deterministic corpus seeder (store API) + `SEED_SCHEMA_REVISION` |
-| `journeys.py` | `Journey` dataclass, latency/throughput runners, registry |
-| `environment.py` | server (± runner + mock LLM) lifecycle; `--database-uri` |
-| `measure.py` | `RunResult`, percentile, aggregation, thresholds, tables |
-| `schema.py` | `SCHEMA_VERSION`, `build_report`, git/host metadata |
-| `sample_output.json` | committed example of the JSON contract |
-
-The smoke test is `tests/benchmarks/test_benchmark_smoke.py` (boots the server
-with tiny counts + a seeded-corpus unit test; runs on the normal CI lane, no
-creds). `scripts/check_benchmark_seed_schema.py` is the schema-drift guard.
-
-## CI
-
-`.github/workflows/benchmark.yml` runs nightly (and on dispatch) as a backend
-matrix — `sqlite` and `postgres` (a `postgres:16` service container). Each leg
-seeds a corpus (SQLite reuses a cache keyed on schema head + `seed.py` + corpus
-config; Postgres is fresh per run), runs the benchmark, and uploads
-`benchmark-results-<backend>-<run_id>.json`. The workspace notebook pulls those
-artifacts.
-
-### Schema drift
-
-`seed.py` pins `SEED_SCHEMA_REVISION` to the DB's Alembic head. The pre-commit
-hook `check-benchmark-seed-schema` (→ `scripts/check_benchmark_seed_schema.py`)
-fails when `omnigent/db/db_models.py` or a migration changes the head without
-the seed being refreshed. On failure: update `seed.py` to seed any new schema,
-bump `SEED_SCHEMA_REVISION` (`seed.py --print-head`), and regenerate the cached
-corpus / `sample_output.json`.
-
-## Follow-ups
-
-- **Excluded journeys** (agent-behaviour-dependent, deliberately not measured):
-  multi-turn and tool-calling turns (dominated by the agent's own choices) and
-  large-history turns (the O(N) `history_to_input_items` conversion is real app
-  work but only fires on a cold runner cache, so isolating it entangles with
-  cold-start cost).
-- **CI matrix.** Runner journeys are backend-agnostic (they exercise runner
-  dispatch, not big DB reads), so the nightly workflow can run them on the
-  SQLite leg only rather than both — wire a runner `--journeys` set into
-  `benchmark.yml` when desired.
-- **Simulated provider latency.** The mock LLM returns at ~zero latency, which
-  is what isolates omnigent overhead. A fixed per-response delay knob would let
-  turns model end-user wall-clock instead; it's a small change behind the
-  `configure_mock` / `set_mock_fallback` seam if that's ever wanted.

dev/benchmarks/omnigent/__init__.py

@@ -1,7 +0,0 @@
-"""Omnigent user-journey performance benchmark.
-
-Stands up a real server + runner against a zero-latency mock LLM, drives
-key user journeys under load, and emits a versioned JSON report of latency
-percentiles and throughput. See ``README.md`` for the workflow and how the
-workspace ETL notebook consumes the JSON.
-"""

dev/benchmarks/omnigent/environment.py

@@ -1,645 +0,0 @@
-"""Benchmark environment lifecycle.
-
-:class:`BenchEnvironment` is an async context manager that stands up a real
-Omnigent ``server`` with no Databricks credentials. Two modes:
-
-- ``with_runner=False`` (default): server + SQLite DB only. Enough for the
-  HTTP/API journeys, which never drive an agent turn.
-- ``with_runner=True``: additionally spawns a zero-latency mock LLM and a
-  sibling ``runner``, routes the server-side prompt-policy classifier at the
-  mock (via ``--config``), and sets an ALLOW fallback — everything the
-  full-turn journeys need.
-
-A full env is a strict superset of the HTTP-only env, so both modes share one
-class; the runner mode is gated behind the flag rather than forked into a
-separate type. It mirrors the proven ``live_server`` e2e recipe
-(``tests/e2e/conftest.py``) and reuses the credential-free spawn core: the
-compat helpers (so subprocesses import this worktree) and
-``token_bound_runner_id``.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import io
-import os
-import signal
-import socket
-import subprocess
-import sys
-import tarfile
-import time
-import uuid
-from pathlib import Path
-from typing import IO
-
-import httpx
-import yaml
-
-from omnigent.runner.identity import OMNIGENT_INTERNAL_WS_ORIGIN, token_bound_runner_id
-from tests._helpers.compat import (
-    apply_runner_env,
-    apply_server_env,
-    compat_runner_cwd,
-    compat_server_cwd,
-    runner_executable,
-    server_executable,
-)
-
-_REPO_ROOT = Path(__file__).resolve().parents[3]
-_MOCK_SERVER = _REPO_ROOT / "tests" / "server" / "integration" / "mock_llm_server.py"
-
-_HEALTH_TIMEOUT_S = 90.0
-_MOCK_TIMEOUT_S = 15.0
-_POLL_INTERVAL_S = 0.2
-_TURN_TIMEOUT_S = 180.0
-
-# Terminal SSE events — if one arrives before any delta, the turn produced no
-# streamed text (a failure for the TTFT journey).
-_STREAM_TERMINAL_EVENTS = frozenset(
-    {"response.completed", "response.failed", "response.cancelled"}
-)
-# The server persists an interrupted turn as a synthetic user message whose
-# text contains this marker (see tests/e2e/test_cancel_history.py).
-_CANCELLATION_MARKER = "interrupted"
-
-# Default full-turn agent (with_runner=True). The mock ignores the model for
-# routing (its "default" queue serves any request), but the key is baked into
-# the spec so the harness has a concrete model to send.
-_DEFAULT_MODEL = "mock-bench-brain"
-_DEFAULT_HARNESS = "openai-agents"
-
-# Server-side prompt-policy classifier queue key. In runner mode we set an
-# ALLOW fallback here so a classifier call (if the agent trips one) never
-# blocks or returns non-verdict text.
-_POLICY_LLM_KEY = "_policy_llm_"
-_POLICY_ALLOW = '{"action": "allow", "reason": ""}'
-
-
-def _find_free_port() -> int:
-    """Bind an ephemeral port and return it (races are tolerated by retries)."""
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-        sock.bind(("127.0.0.1", 0))
-        return int(sock.getsockname()[1])
-
-
-class BenchEnvironment:
-    """Async context manager owning the benchmark's server (± runner + mock).
-
-    :param with_runner: When ``False`` (default), boot the server only — the
-        v1 HTTP-journey path. When ``True``, also spawn the mock LLM and a
-        runner and wire the policy classifier at the mock — the phase-2
-        full-turn path.
-    :param database_uri: SQLAlchemy URI the server boots against. ``None``
-        (default) uses a fresh throwaway SQLite file in the temp dir — the
-        empty-DB path. Pass a pre-seeded URI (e.g. a seeded SQLite file, or a
-        ``postgresql+psycopg://…`` instance) to benchmark against a realistic
-        corpus. Postgres must be the fully-qualified ``+psycopg`` form — the
-        server CLI does not normalize it.
-    :param harness: Harness for full-turn agents when ``with_runner`` (default
-        ``openai-agents``, a base dependency needing no vendor CLI binary).
-    :param model: Model string baked into registered agent specs.
-    """
-
-    def __init__(
-        self,
-        *,
-        with_runner: bool = False,
-        database_uri: str | None = None,
-        harness: str = _DEFAULT_HARNESS,
-        model: str = _DEFAULT_MODEL,
-    ) -> None:
-        self.with_runner = with_runner
-        self.database_uri = database_uri
-        self.harness = harness
-        self.model = model
-        self.base_url = ""
-        self.mock_url = ""
-        self.runner_id = ""
-        self.client: httpx.AsyncClient | None = None
-
-        self._tmp = Path("/tmp") / f"omni-bench-{uuid.uuid4().hex[:8]}"
-        self._mock_proc: subprocess.Popen[bytes] | None = None
-        self._server_proc: subprocess.Popen[bytes] | None = None
-        self._runner_proc: subprocess.Popen[bytes] | None = None
-        self._log_handles: list[IO[bytes]] = []
-        self._agent_cache: dict[str, str] = {}
-
-    # ── lifecycle ────────────────────────────────────────────
-
-    async def __aenter__(self) -> BenchEnvironment:
-        await asyncio.to_thread(self._start)
-        self.client = httpx.AsyncClient(
-            base_url=self.base_url,
-            timeout=300.0,
-            headers={"Origin": OMNIGENT_INTERNAL_WS_ORIGIN},
-        )
-        if self.with_runner:
-            # ALLOW fallback so a server-side classifier call resolves against
-            # the mock (never api.openai.com) and returns a valid verdict.
-            await self._mock_post(
-                "/mock/set_fallback", {"key": _POLICY_LLM_KEY, "text": _POLICY_ALLOW}
-            )
-        return self
-
-    async def __aexit__(self, *exc: object) -> None:
-        if self.client is not None:
-            await self.client.aclose()
-        await asyncio.to_thread(self._stop)
-
-    def _start(self) -> None:
-        """Spawn the server (± mock + runner) and block until ready."""
-        self._tmp.mkdir(mode=0o700, parents=True, exist_ok=True)
-        artifact_dir = self._tmp / "artifacts"
-        artifact_dir.mkdir(exist_ok=True)
-
-        if self.with_runner:
-            mock_port = _find_free_port()
-            self.mock_url = f"http://127.0.0.1:{mock_port}"
-            self._mock_proc = self._spawn_mock(mock_port)
-            self._wait_mock_ready()
-
-        port = _find_free_port()
-        self.base_url = f"http://localhost:{port}"
-        binding_token = uuid.uuid4().hex
-
-        base_env = {**os.environ}
-        if self.with_runner:
-            self.runner_id = token_bound_runner_id(binding_token)
-            base_env["OPENAI_API_KEY"] = "mock-key"
-            # The OpenAI SDK appends /responses, so include /v1 in the base.
-            base_env["OPENAI_BASE_URL"] = f"{self.mock_url}/v1"
-        # Prepend the worktree so subprocesses import this branch's source.
-        apply_server_env(base_env, _REPO_ROOT)
-
-        self._server_proc = self._spawn_server(port, base_env, binding_token, artifact_dir)
-        if self.with_runner:
-            self._runner_proc = self._spawn_runner(base_env, binding_token)
-        self._wait_ready()
-
-    def _stop(self) -> None:
-        """Terminate runner, server, and mock; remove the temp dir."""
-        for proc in (self._runner_proc, self._server_proc, self._mock_proc):
-            if proc is not None and proc.poll() is None:
-                proc.send_signal(signal.SIGTERM)
-                try:
-                    proc.wait(timeout=8)
-                except subprocess.TimeoutExpired:
-                    proc.kill()
-                    proc.wait(timeout=5)
-        for handle in self._log_handles:
-            handle.close()
-        import shutil
-
-        shutil.rmtree(self._tmp, ignore_errors=True)
-
-    # ── spawns ───────────────────────────────────────────────
-
-    def _log(self, name: str) -> IO[bytes]:
-        handle = (self._tmp / name).open("wb")
-        self._log_handles.append(handle)
-        return handle
-
-    def _spawn_mock(self, port: int) -> subprocess.Popen[bytes]:
-        return subprocess.Popen(
-            [sys.executable, str(_MOCK_SERVER), str(port)],
-            env={**os.environ, "PYTHONPATH": str(_REPO_ROOT)},
-            stdout=self._log("mock.log"),
-            stderr=subprocess.STDOUT,
-        )
-
-    def _spawn_server(
-        self,
-        port: int,
-        base_env: dict[str, str],
-        binding_token: str,
-        artifact_dir: Path,
-    ) -> subprocess.Popen[bytes]:
-        # Pre-seeded URI when given (realistic corpus), else a throwaway SQLite
-        # file in the temp dir (the empty-DB path). SQLite absolute paths need
-        # four slashes; the temp path is absolute.
-        db_uri = self.database_uri or f"sqlite:///{self._tmp / 'bench.db'}"
-        args = [
-            server_executable(),
-            "-m",
-            "omnigent.cli",
-            "server",
-            "--port",
-            str(port),
-            "--database-uri",
-            db_uri,
-            "--artifact-location",
-            str(artifact_dir),
-        ]
-        env = {**base_env}
-        if self.with_runner:
-            # Route the server-side policy-classifier LLM at the mock, mirroring
-            # live_server. Without this the classifier's client defaults to
-            # api.openai.com and errors. Server-only mode needs no llm config —
-            # the classifier only builds under OMNIGENT_SMART_ROUTING=1.
-            server_cfg = self._tmp / "server.yaml"
-            server_cfg.write_text(
-                yaml.safe_dump(
-                    {
-                        "llm": {
-                            "model": _POLICY_LLM_KEY,
-                            "connection": {
-                                "base_url": f"{self.mock_url}/v1",
-                                "api_key": "mock-key",
-                            },
-                        }
-                    }
-                )
-            )
-            args.extend(["--config", str(server_cfg)])
-            env["OMNIGENT_RUNNER_TUNNEL_TOKEN"] = binding_token
-        return subprocess.Popen(
-            args,
-            env=env,
-            cwd=compat_server_cwd(),
-            stdout=self._log("server.log"),
-            stderr=subprocess.STDOUT,
-        )
-
-    def _spawn_runner(
-        self, base_env: dict[str, str], binding_token: str
-    ) -> subprocess.Popen[bytes]:
-        runner_env = apply_runner_env(
-            {
-                **base_env,
-                "OMNIGENT_RUNNER_ID": self.runner_id,
-                "OMNIGENT_RUNNER_TUNNEL_BINDING_TOKEN": binding_token,
-                "OMNIGENT_RUNNER_PARENT_PID": str(os.getpid()),
-                "RUNNER_SERVER_URL": self.base_url,
-            }
-        )
-        return subprocess.Popen(
-            [runner_executable(), "-m", "omnigent.runner._entry"],
-            env=runner_env,
-            cwd=compat_runner_cwd(),
-            stdout=self._log("runner.log"),
-            stderr=subprocess.STDOUT,
-        )
-
-    # ── readiness ────────────────────────────────────────────
-
-    def _wait_mock_ready(self) -> None:
-        deadline = time.monotonic() + _MOCK_TIMEOUT_S
-        while time.monotonic() < deadline:
-            try:
-                if httpx.get(f"{self.mock_url}/stats", timeout=1).status_code == 200:
-                    return
-            except httpx.HTTPError:
-                pass
-            time.sleep(0.1)
-        raise RuntimeError(f"mock LLM not ready within {_MOCK_TIMEOUT_S}s; logs in {self._tmp}")
-
-    def _wait_ready(self) -> None:
-        """Wait for ``/health`` (and, in runner mode, the runner online)."""
-        deadline = time.monotonic() + _HEALTH_TIMEOUT_S
-        while time.monotonic() < deadline:
-            try:
-                health = httpx.get(f"{self.base_url}/health", timeout=2)
-                if health.status_code == 200 and self._runner_ready():
-                    return
-            except httpx.HTTPError:
-                pass
-            time.sleep(_POLL_INTERVAL_S)
-        raise RuntimeError(f"server not ready within {_HEALTH_TIMEOUT_S}s; logs in {self._tmp}")
-
-    def _runner_ready(self) -> bool:
-        """Whether the runner reports online (always ``True`` server-only)."""
-        if not self.with_runner:
-            return True
-        status = httpx.get(f"{self.base_url}/v1/runners/{self.runner_id}/status", timeout=2)
-        return status.status_code == 200 and status.json().get("online") is True
-
-    # ── mock control (runner mode only) ──────────────────────
-
-    async def _mock_post(self, path: str, body: dict[str, object]) -> None:
-        async with httpx.AsyncClient(timeout=5.0) as client:
-            resp = await client.post(f"{self.mock_url}{path}", json=body)
-            resp.raise_for_status()
-
-    async def configure_mock(
-        self,
-        responses: list[dict[str, object]],
-        *,
-        key: str = "default",
-        match: str | None = None,
-    ) -> None:
-        """Load a keyed response queue on the mock (see e2e ``configure_mock_llm``)."""
-        payload: dict[str, object] = {"key": key, "responses": responses}
-        if match is not None:
-            payload["match"] = match
-        await self._mock_post("/mock/configure", payload)
-
-    async def set_mock_fallback(
-        self, text: str, *, key: str = "default", stream: bool = False
-    ) -> None:
-        """Set a reset-surviving fallback response for a mock queue *key*.
-
-        :param stream: When ``True`` the fallback emits per-word
-            ``output_text.delta`` events before completing — needed for the
-            time-to-first-token journey to observe streamed deltas.
-        """
-        await self._mock_post("/mock/set_fallback", {"key": key, "text": text, "stream": stream})
-
-    # ── agent + session primitives ───────────────────────────
-
-    def _agent_bundle(self, name: str) -> bytes:
-        """Build a ``spec_version: 1`` agent bundle.
-
-        In runner mode the executor is wired at the mock LLM (auth +
-        connection). Server-only, no LLM is ever called, so the bundle just
-        needs to be a valid spec the server can register and bind sessions to.
-        """
-        executor: dict[str, object] = {
-            "type": "omnigent",
-            "model": self.model,
-            "config": {"harness": self.harness},
-        }
-        if self.with_runner:
-            executor["auth"] = {
-                "type": "api_key",
-                "api_key": "mock-key",
-                "base_url": f"{self.mock_url}/v1",
-            }
-            executor["connection"] = {"base_url": f"{self.mock_url}/v1", "api_key": "mock-key"}
-        config: dict[str, object] = {
-            "spec_version": 1,
-            "name": name,
-            "prompt": "You are a helpful assistant used for performance benchmarking.",
-            "executor": executor,
-        }
-        buf = io.BytesIO()
-        with tarfile.open(fileobj=buf, mode="w:gz") as tar:
-            payload = yaml.safe_dump(config).encode()
-            info = tarfile.TarInfo("config.yaml")
-            info.size = len(payload)
-            tar.addfile(info, io.BytesIO(payload))
-        return buf.getvalue()
-
-    async def ensure_agent(self, name: str = "bench-agent") -> str:
-        """Register the benchmark agent once, returning its name (idempotent)."""
-        assert self.client is not None
-        if name in self._agent_cache:
-            return name
-        resp = await self.client.post(
-            "/v1/sessions",
-            data={"metadata": "{}"},
-            files={"bundle": ("agent.tar.gz", self._agent_bundle(name), "application/gzip")},
-        )
-        if resp.status_code not in (200, 201, 409):
-            raise RuntimeError(f"agent register failed: {resp.status_code} {resp.text[:400]}")
-        self._agent_cache[name] = name
-        return name
-
-    async def agent_id(self, agent_name: str) -> str:
-        """Resolve a registered agent's id by name."""
-        assert self.client is not None
-        listing = await self.client.get(
-            "/v1/sessions", params={"agent_name": agent_name, "limit": 1}
-        )
-        listing.raise_for_status()
-        return str(listing.json()["data"][0]["agent_id"])
-
-    async def create_session(self, agent_id: str) -> str:
-        """Create an (unbound) session for *agent_id*, returning its id."""
-        assert self.client is not None
-        created = await self.client.post("/v1/sessions", json={"agent_id": agent_id})
-        created.raise_for_status()
-        return str(created.json()["id"])
-
-    async def seed_items(self, session_id: str, count: int) -> None:
-        """Append *count* history items over HTTP, with no runner or LLM.
-
-        Uses the ``external_conversation_item`` event, which the server
-        appends "without starting or steering a task" — the runner-free path
-        for giving ``load_conversation_history`` something to read back.
-
-        Items are user messages: assistant messages require an ``agent`` field
-        the server only has after a real turn, and the read path this seeds is
-        role-agnostic — item count and size, not role, drive its cost.
-        """
-        assert self.client is not None
-        for i in range(count):
-            body = {
-                "type": "external_conversation_item",
-                "data": {
-                    "item_type": "message",
-                    "item_data": {
-                        "role": "user",
-                        "content": [{"type": "input_text", "text": f"benchmark seed item {i}"}],
-                    },
-                },
-            }
-            resp = await self.client.post(f"/v1/sessions/{session_id}/events", json=body)
-            resp.raise_for_status()
-
-    # ── runner-mode session driving (phase 2) ────────────────
-
-    async def create_bound_session(self, agent_id: str) -> str:
-        """Create a session for *agent_id* and bind it to the runner."""
-        assert self.client is not None
-        if not self.with_runner:
-            raise RuntimeError("create_bound_session requires with_runner=True")
-        session_id = await self.create_session(agent_id)
-        bound = await self.client.patch(
-            f"/v1/sessions/{session_id}", json={"runner_id": self.runner_id}
-        )
-        bound.raise_for_status()
-        return session_id
-
-    async def drive_turn(
-        self, session_id: str, text: str, *, timeout: float = _TURN_TIMEOUT_S
-    ) -> None:
-        """Post a user message and poll the session to a terminal state.
-
-        :raises RuntimeError: If not in runner mode, the turn fails, or it does
-            not settle within *timeout* seconds.
-        """
-        assert self.client is not None
-        if not self.with_runner:
-            raise RuntimeError("drive_turn requires with_runner=True")
-        body = {
-            "type": "message",
-            "data": {"role": "user", "content": [{"type": "input_text", "text": text}]},
-        }
-        posted = await self.client.post(f"/v1/sessions/{session_id}/events", json=body)
-        posted.raise_for_status()
-
-        deadline = time.monotonic() + timeout
-        seen_running = False
-        while time.monotonic() < deadline:
-            snap = await self.client.get(f"/v1/sessions/{session_id}")
-            snap.raise_for_status()
-            status = snap.json().get("status")
-            if status in ("running", "waiting"):
-                seen_running = True
-            elif status == "failed":
-                raise RuntimeError(f"turn failed: {snap.json().get('last_task_error')}")
-            elif status == "idle" and seen_running:
-                return
-            await asyncio.sleep(_POLL_INTERVAL_S)
-        raise RuntimeError(f"turn did not settle within {timeout}s (session {session_id})")
-
-    async def _wait_idle(self, session_id: str, *, timeout: float = _TURN_TIMEOUT_S) -> None:
-        """Poll until the session is ``idle`` (a prior turn has settled)."""
-        assert self.client is not None
-        deadline = time.monotonic() + timeout
-        while time.monotonic() < deadline:
-            snap = await self.client.get(f"/v1/sessions/{session_id}")
-            snap.raise_for_status()
-            if snap.json().get("status") == "idle":
-                return
-            await asyncio.sleep(_POLL_INTERVAL_S)
-        raise RuntimeError(f"session did not reach idle within {timeout}s ({session_id})")
-
-    async def time_to_first_delta(
-        self, session_id: str, text: str, *, timeout: float = _TURN_TIMEOUT_S
-    ) -> None:
-        """Post a turn and return once the first output-text delta streams back.
-
-        The session SSE stream (``GET …/stream``) is separate from the message
-        POST, so we subscribe first (as a concurrent task), post the turn, then
-        return when the first ``response.output_text.delta`` event arrives. This
-        times omnigent's streaming-pipeline overhead to first token — with the
-        zero-latency mock there is no model latency in the number.
-
-        :raises RuntimeError: If not in runner mode, or no delta / a terminal
-            event arrives within *timeout*.
-        """
-        assert self.client is not None
-        if not self.with_runner:
-            raise RuntimeError("time_to_first_delta requires with_runner=True")
-
-        connected = asyncio.Event()
-        first_delta = asyncio.Event()
-        outcome: dict[str, str] = {}
-
-        async def _read_stream() -> None:
-            try:
-                async with self.client.stream(  # type: ignore[union-attr]
-                    "GET", f"/v1/sessions/{session_id}/stream", timeout=timeout
-                ) as resp:
-                    # Any first line means the SSE connection is live (the server
-                    # emits a heartbeat on connect). Signalling here lets us post
-                    # the turn only once subscribed — without a blind sleep that
-                    # would otherwise inflate the measured time-to-first-delta.
-                    connected.set()
-                    async for line in resp.aiter_lines():
-                        if not line.startswith("event:"):
-                            continue
-                        etype = line[len("event:") :].strip()
-                        if etype == "response.output_text.delta":
-                            first_delta.set()
-                            return
-                        if etype in _STREAM_TERMINAL_EVENTS:
-                            outcome["terminal"] = etype
-                            first_delta.set()
-                            return
-            except httpx.HTTPError as exc:
-                outcome["error"] = repr(exc)
-                connected.set()
-                first_delta.set()
-
-        # Ensure any prior turn has settled so the fresh subscription's first
-        # terminal event can't be the previous turn completing (which would
-        # otherwise race ahead of this turn's delta).
-        await self._wait_idle(session_id, timeout=timeout)
-
-        reader = asyncio.create_task(_read_stream())
-        try:
-            # Wait until the stream is actually connected (not a fixed sleep) so
-            # the measured window is post → first delta, not subscription setup.
-            await asyncio.wait_for(connected.wait(), timeout=timeout)
-            posted = await self.client.post(
-                f"/v1/sessions/{session_id}/events",
-                json={
-                    "type": "message",
-                    "data": {"role": "user", "content": [{"type": "input_text", "text": text}]},
-                },
-            )
-            posted.raise_for_status()
-            try:
-                await asyncio.wait_for(first_delta.wait(), timeout=timeout)
-            except TimeoutError as exc:
-                raise RuntimeError(
-                    f"no output_text.delta within {timeout}s (session {session_id})"
-                ) from exc
-            if "error" in outcome:
-                raise RuntimeError(f"stream error: {outcome['error']}")
-            if "terminal" in outcome:
-                raise RuntimeError(
-                    f"turn reached {outcome['terminal']} before any delta (session {session_id})"
-                )
-        finally:
-            reader.cancel()
-
-    async def drive_and_interrupt(
-        self, session_id: str, *, timeout: float = _TURN_TIMEOUT_S
-    ) -> None:
-        """Drive a gated turn, interrupt it mid-flight, return when cancelled.
-
-        The caller configures a ``block=True`` mock response first (see
-        :meth:`configure_mock`), so the turn parks in ``running`` on the
-        executor's LLM call. We post an ``interrupt`` once running, wait for the
-        server's cancellation marker, then release the gate so the runner
-        unwinds cleanly. Times the server → runner → executor cancel path.
-
-        :raises RuntimeError: If not in runner mode, or the interrupt is not
-            honored within *timeout*.
-        """
-        assert self.client is not None
-        if not self.with_runner:
-            raise RuntimeError("drive_and_interrupt requires with_runner=True")
-        body = {
-            "type": "message",
-            "data": {"role": "user", "content": [{"type": "input_text", "text": "Interrupt me."}]},
-        }
-        posted = await self.client.post(f"/v1/sessions/{session_id}/events", json=body)
-        posted.raise_for_status()
-
-        deadline = time.monotonic() + timeout
-        interrupted = False
-        try:
-            while time.monotonic() < deadline:
-                snap = (await self.client.get(f"/v1/sessions/{session_id}")).json()
-                status = snap.get("status")
-                items = snap.get("items", [])
-                if status in ("running", "waiting") and not interrupted:
-                    await self.client.post(
-                        f"/v1/sessions/{session_id}/events", json={"type": "interrupt"}
-                    )
-                    interrupted = True
-                if _has_cancellation_marker(items):
-                    return
-                if status == "idle" and interrupted:
-                    if _has_cancellation_marker(items):
-                        return
-                    raise RuntimeError("turn settled without a cancellation marker")
-                await asyncio.sleep(_POLL_INTERVAL_S)
-            raise RuntimeError(f"interrupt not honored within {timeout}s (session {session_id})")
-        finally:
-            # Always release the gate so the blocked runner turn unwinds and
-            # teardown doesn't hang, even if the interrupt path errored above.
-            with contextlib.suppress(httpx.HTTPError):
-                await self._mock_post("/gate/release", {})
-
-
-def _has_cancellation_marker(items: list[dict[str, object]]) -> bool:
-    """Whether items include the synthetic 'interrupted' user message."""
-    for raw in items:
-        data = raw.get("data", raw)
-        if not isinstance(data, dict):
-            continue
-        if raw.get("type") == "message" and data.get("role") == "user":
-            content = data.get("content") or []
-            if isinstance(content, list) and any(
-                isinstance(b, dict) and _CANCELLATION_MARKER in str(b.get("text", ""))
-                for b in content
-            ):
-                return True
-    return False

dev/benchmarks/omnigent/journeys.py

@@ -1,422 +0,0 @@
-"""User-journey definitions and the runners that time them.
-
-A :class:`Journey` names a user-facing operation, an optional per-journey
-``setup`` that returns a context object, and a ``measure`` coroutine — the
-timed unit. :func:`run_latency` times ``measure`` sequentially; journeys marked
-``concurrency_safe`` can also be driven by :func:`run_throughput` with many
-operations in flight.
-
-v1 journeys are pure HTTP/API (server + DB, no runner, no LLM):
-
-- ``list_sessions`` — the session-list read behind the sidebar/home.
-- ``create_session`` — session creation cost (POST then DELETE).
-- ``get_session`` — single-session snapshot load.
-- ``load_conversation_history`` — history read, seeded runner-free via
-  ``external_conversation_item`` (see :meth:`BenchEnvironment.seed_items`).
-
-The framework (``Journey`` + the two runners) is harness-agnostic and reused
-verbatim by phase-2 full-turn journeys.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import time
-from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
-from typing import Literal, cast
-
-import httpx
-
-from .environment import BenchEnvironment
-from .measure import RunResult
-
-# Per-journey context returned by ``setup`` and threaded to ``measure``. Its
-# concrete type varies by journey (an agent id, a session id, or nothing), so
-# it is opaque at the framework level; each measure op casts it as needed.
-JourneyContext = object
-
-JourneyKind = Literal["latency", "throughput"]
-
-# Items requested per history-read page. Also the count self-seeded into a
-# fallback session when the DB has no corpus (empty-DB smoke path).
-_HISTORY_PAGE_LIMIT = 20
-_HISTORY_SEED_ITEMS = _HISTORY_PAGE_LIMIT
-
-
-@dataclass
-class Journey:
-    """One benchmarkable user journey.
-
-    :param name: Stable identifier used on the CLI and as the report key.
-    :param kind: ``"latency"`` (time each operation) or ``"throughput"``
-        (fixed request count under concurrency). A latency journey that is
-        ``concurrency_safe`` can additionally be run as throughput.
-    :param measure: Coroutine performing exactly one timed operation, given
-        the environment and the setup context.
-    :param setup: Optional coroutine run once before timing; its return value
-        is passed to ``measure`` (and ``teardown``) as ``ctx``.
-    :param teardown: Optional coroutine run once after timing, given ``ctx``.
-    :param concurrency_safe: Whether many ``measure`` calls may run at once
-        against a shared setup (true for read-only / independent-write HTTP
-        journeys).
-    :param needs_runner: Whether this journey drives a full agent turn and so
-        requires ``BenchEnvironment(with_runner=True)`` (mock LLM + runner).
-        HTTP/DB journeys leave this ``False``.
-    :param description: Human-readable one-liner for ``--list``.
-    """
-
-    name: str
-    kind: JourneyKind
-    measure: Callable[[BenchEnvironment, JourneyContext], Awaitable[None]]
-    setup: Callable[[BenchEnvironment], Awaitable[JourneyContext]] | None = None
-    teardown: Callable[[BenchEnvironment, JourneyContext], Awaitable[None]] | None = None
-    concurrency_safe: bool = False
-    needs_runner: bool = False
-    description: str = ""
-
-    async def run_setup(self, env: BenchEnvironment) -> JourneyContext:
-        return await self.setup(env) if self.setup is not None else None
-
-    async def run_teardown(self, env: BenchEnvironment, ctx: JourneyContext) -> None:
-        if self.teardown is not None:
-            await self.teardown(env, ctx)
-
-
-# ── timed operation (shared by both runners) ─────────────────
-
-
-async def _timed(
-    journey: Journey, env: BenchEnvironment, ctx: JourneyContext, result: RunResult
-) -> None:
-    """Run one ``measure`` op, recording its latency or a failure reason."""
-    start = time.perf_counter()
-    try:
-        await journey.measure(env, ctx)
-    except httpx.HTTPStatusError as exc:
-        result.record_failure(f"HTTP {exc.response.status_code}")
-    except Exception as exc:  # noqa: BLE001 — any failure is a recorded data point
-        result.record_failure(exc.__class__.__name__)
-    else:
-        result.latencies_ms.append((time.perf_counter() - start) * 1000)
-
-
-# ── runners ──────────────────────────────────────────────────
-
-
-async def run_latency(
-    journey: Journey, env: BenchEnvironment, *, iterations: int, warmup: int
-) -> RunResult:
-    """Time *iterations* sequential operations after discarding *warmup*.
-
-    Warmup operations run through the same path but are excluded from the
-    result, so first-call import/JIT/connection costs don't skew the numbers.
-    """
-    ctx = await journey.run_setup(env)
-    try:
-        for _ in range(warmup):
-            with contextlib.suppress(Exception):  # warmup errors are non-fatal
-                await journey.measure(env, ctx)
-        result = RunResult()
-        wall_start = time.perf_counter()
-        for _ in range(iterations):
-            await _timed(journey, env, ctx, result)
-        result.wall_time = time.perf_counter() - wall_start
-        return result
-    finally:
-        await journey.run_teardown(env, ctx)
-
-
-async def run_throughput(
-    journey: Journey,
-    env: BenchEnvironment,
-    *,
-    requests: int,
-    concurrency: int,
-    warmup: int,
-) -> RunResult:
-    """Fire *requests* operations with at most *concurrency* in flight.
-
-    Wall time spans from the first dispatch to the last completion, so
-    ``throughput`` reflects sustained req/s under load (MLflow's ``_run_once``
-    shape, with an :class:`asyncio.Semaphore` gate).
-    """
-    ctx = await journey.run_setup(env)
-    try:
-        sem = asyncio.Semaphore(concurrency)
-
-        async def _one(count_it: bool, result: RunResult) -> None:
-            async with sem:
-                if count_it:
-                    await _timed(journey, env, ctx, result)
-                else:
-                    with contextlib.suppress(Exception):  # warmup errors are non-fatal
-                        await journey.measure(env, ctx)
-
-        if warmup:
-            throwaway = RunResult()
-            await asyncio.gather(*[_one(False, throwaway) for _ in range(warmup)])
-
-        result = RunResult()
-        wall_start = time.perf_counter()
-        await asyncio.gather(*[_one(True, result) for _ in range(requests)])
-        result.wall_time = time.perf_counter() - wall_start
-        return result
-    finally:
-        await journey.run_teardown(env, ctx)
-
-
-# ── journey implementations ──────────────────────────────────
-#
-# Setups return the context each measure op needs. Ops must be independent so
-# concurrency-safe journeys don't interfere across in-flight calls.
-
-
-# A token present in the seeded corpus (titles + item text, see seed.py
-# _FRAGMENTS) so search_sessions exercises the LIKE path with real matches.
-_SEARCH_TOKEN = "runner"
-
-
-async def _setup_agent_id(env: BenchEnvironment) -> str:
-    """Register the benchmark agent and return its id."""
-    name = await env.ensure_agent()
-    return await env.agent_id(name)
-
-
-async def _setup_target_session(env: BenchEnvironment) -> str:
-    """Return a session id to read: an existing corpus session if any, else make one.
-
-    Real runs target a pre-seeded corpus (``seed.py``), so we read a
-    representative existing session. When the DB is empty (e.g. the smoke test
-    against a throwaway DB), fall back to creating one with a little history so
-    the journey still exercises the read path.
-    """
-    assert env.client is not None
-    listing = await env.client.get("/v1/sessions", params={"limit": 1})
-    listing.raise_for_status()
-    data = listing.json().get("data", [])
-    if data:
-        return str(data[0]["id"])
-    # Empty DB: self-seed one session over HTTP (runner-free).
-    name = await env.ensure_agent()
-    agent_id = await env.agent_id(name)
-    session_id = await env.create_session(agent_id)
-    await env.seed_items(session_id, _HISTORY_SEED_ITEMS)
-    return session_id
-
-
-async def _measure_list_sessions(env: BenchEnvironment, _ctx: JourneyContext) -> None:
-    assert env.client is not None
-    resp = await env.client.get("/v1/sessions", params={"limit": 20})
-    resp.raise_for_status()
-
-
-async def _measure_search_sessions(env: BenchEnvironment, _ctx: JourneyContext) -> None:
-    assert env.client is not None
-    resp = await env.client.get(
-        "/v1/sessions", params={"limit": 20, "search_query": _SEARCH_TOKEN}
-    )
-    resp.raise_for_status()
-
-
-async def _measure_create_session(env: BenchEnvironment, ctx: JourneyContext) -> None:
-    assert env.client is not None
-    agent_id = cast(str, ctx)  # _setup_agent_id
-    created = await env.client.post("/v1/sessions", json={"agent_id": agent_id})
-    created.raise_for_status()
-    # Delete inline so a long run doesn't accumulate unbounded sessions; the
-    # POST is the operation of interest and dominates the timed span.
-    session_id = created.json()["id"]
-    deleted = await env.client.delete(f"/v1/sessions/{session_id}")
-    deleted.raise_for_status()
-
-
-async def _measure_get_session(env: BenchEnvironment, ctx: JourneyContext) -> None:
-    assert env.client is not None
-    session_id = cast(str, ctx)  # _setup_target_session
-    resp = await env.client.get(f"/v1/sessions/{session_id}")
-    resp.raise_for_status()
-
-
-async def _measure_load_history(env: BenchEnvironment, ctx: JourneyContext) -> None:
-    assert env.client is not None
-    session_id = cast(str, ctx)  # _setup_target_session
-    resp = await env.client.get(
-        f"/v1/sessions/{session_id}/items",
-        params={"order": "asc", "limit": _HISTORY_PAGE_LIMIT},
-    )
-    resp.raise_for_status()
-
-
-# ── runner (full-turn) journeys ──────────────────────────────
-#
-# These drive a real agent turn through the runner + mock LLM (with_runner=True,
-# openai-agents). The mock is zero-latency, so every number is omnigent dispatch
-# / streaming / cancel overhead, not model latency. Short deterministic replies.
-
-# A multi-word reply so the streaming path emits several output_text deltas.
-_TURN_REPLY = "Hello there, this is a mock benchmark reply."
-_TURN_PROMPT = "Say hello."
-
-
-async def _setup_turn_agent(env: BenchEnvironment, *, stream: bool = False) -> str:
-    """Register the agent + a reset-surviving reply; return the agent id.
-
-    The fallback survives per-call queue exhaustion, so every turn in the run
-    gets the same reply regardless of how many turns consume the queue. When
-    *stream* is set the reply emits per-word deltas (for the TTFT journey).
-    """
-    name = await env.ensure_agent()
-    await env.set_mock_fallback(_TURN_REPLY, stream=stream)
-    return await env.agent_id(name)
-
-
-async def _setup_warm_session(env: BenchEnvironment) -> str:
-    """Create+bind a session and drive one warm-up turn; return the session id.
-
-    The warm-up pays the cold-start cost (runner spawn + executor construction)
-    so the measured op times only steady-state per-turn overhead.
-    """
-    agent_id = await _setup_turn_agent(env)
-    session_id = await env.create_bound_session(agent_id)
-    await env.drive_turn(session_id, _TURN_PROMPT)
-    return session_id
-
-
-async def _setup_streaming_session(env: BenchEnvironment) -> str:
-    """Warm session whose mock reply streams deltas — for the TTFT journey."""
-    agent_id = await _setup_turn_agent(env, stream=True)
-    session_id = await env.create_bound_session(agent_id)
-    await env.drive_turn(session_id, _TURN_PROMPT)
-    return session_id
-
-
-async def _setup_interrupt_session(env: BenchEnvironment) -> str:
-    """Create+bind a session for the interrupt journey; return the session id.
-
-    Configures a ``block=True`` mock response so each turn parks in ``running``
-    until the gate is released — giving the interrupt something to cancel
-    mid-flight, deterministically.
-    """
-    name = await env.ensure_agent()
-    agent_id = await env.agent_id(name)
-    session_id = await env.create_bound_session(agent_id)
-    await env.configure_mock([{"text": _TURN_REPLY, "block": True}])
-    return session_id
-
-
-async def _measure_session_cold_start(env: BenchEnvironment, ctx: JourneyContext) -> None:
-    agent_id = cast(str, ctx)  # _setup_turn_agent
-    session_id = await env.create_bound_session(agent_id)
-    await env.drive_turn(session_id, _TURN_PROMPT)
-
-
-async def _measure_warm_turn(env: BenchEnvironment, ctx: JourneyContext) -> None:
-    session_id = cast(str, ctx)  # _setup_warm_session
-    await env.drive_turn(session_id, _TURN_PROMPT)
-
-
-async def _measure_time_to_first_token(env: BenchEnvironment, ctx: JourneyContext) -> None:
-    session_id = cast(str, ctx)  # _setup_warm_session
-    await env.time_to_first_delta(session_id, _TURN_PROMPT)
-
-
-async def _measure_interrupt(env: BenchEnvironment, ctx: JourneyContext) -> None:
-    session_id = cast(str, ctx)  # _setup_interrupt_session
-    await env.drive_and_interrupt(session_id)
-
-
-# ── registry ─────────────────────────────────────────────────
-
-ALL_JOURNEYS: dict[str, Journey] = {
-    j.name: j
-    for j in (
-        Journey(
-            name="list_sessions",
-            kind="latency",
-            measure=_measure_list_sessions,
-            concurrency_safe=True,
-            description="GET /v1/sessions — session list read.",
-        ),
-        Journey(
-            name="create_session",
-            kind="latency",
-            measure=_measure_create_session,
-            setup=_setup_agent_id,
-            concurrency_safe=True,
-            description="POST /v1/sessions then DELETE — session create.",
-        ),
-        Journey(
-            name="get_session",
-            kind="latency",
-            measure=_measure_get_session,
-            setup=_setup_target_session,
-            concurrency_safe=True,
-            description="GET /v1/sessions/{id} — single-session snapshot.",
-        ),
-        Journey(
-            name="load_conversation_history",
-            kind="latency",
-            measure=_measure_load_history,
-            setup=_setup_target_session,
-            concurrency_safe=True,
-            description="GET /v1/sessions/{id}/items — conversation history read.",
-        ),
-        Journey(
-            name="search_sessions",
-            kind="latency",
-            measure=_measure_search_sessions,
-            concurrency_safe=True,
-            description="GET /v1/sessions?search_query= — unindexed LIKE over titles + items.",
-        ),
-        # Runner (full-turn) journeys — with_runner=True, openai-agents, mock LLM.
-        Journey(
-            name="session_cold_start",
-            kind="latency",
-            measure=_measure_session_cold_start,
-            setup=_setup_turn_agent,
-            needs_runner=True,
-            description="Create+bind a fresh session and drive its first turn to idle.",
-        ),
-        Journey(
-            name="warm_turn",
-            kind="latency",
-            measure=_measure_warm_turn,
-            setup=_setup_warm_session,
-            needs_runner=True,
-            description="Drive a turn on an already-warm session (steady-state overhead).",
-        ),
-        Journey(
-            name="time_to_first_token",
-            kind="latency",
-            measure=_measure_time_to_first_token,
-            setup=_setup_streaming_session,
-            needs_runner=True,
-            description="Post a turn; time to the first streamed output_text delta.",
-        ),
-        Journey(
-            name="interrupt",
-            kind="latency",
-            measure=_measure_interrupt,
-            setup=_setup_interrupt_session,
-            needs_runner=True,
-            description="Interrupt a running (gated) turn; time to cancellation.",
-        ),
-    )
-}
-
-
-def resolve_journeys(names: list[str] | None) -> list[Journey]:
-    """Resolve requested journey *names* (or all when ``None``/empty).
-
-    :raises KeyError: If a requested name isn't registered.
-    """
-    if not names:
-        return list(ALL_JOURNEYS.values())
-    resolved = []
-    for name in names:
-        if name not in ALL_JOURNEYS:
-            raise KeyError(f"unknown journey {name!r}; known: {', '.join(ALL_JOURNEYS)}")
-        resolved.append(ALL_JOURNEYS[name])
-    return resolved

dev/benchmarks/omnigent/measure.py

@@ -1,222 +0,0 @@
-"""Latency/throughput measurement primitives.
-
-Pure and I/O-free: a :class:`RunResult` accumulates per-operation latencies
-and failures for one timed run, :func:`aggregate` folds several runs into the
-``runs`` + ``summary`` shape the workspace ETL flattens, and
-:func:`check_thresholds` gates a run in CI. Adapted from MLflow's
-``dev/benchmarks/gateway/benchmark.py``.
-"""
-
-from __future__ import annotations
-
-import math
-import statistics
-from dataclasses import dataclass, field
-
-from rich.console import Console
-from rich.table import Table
-
-console = Console()
-
-
-@dataclass
-class RunResult:
-    """Latencies and failures collected during one timed run.
-
-    :param latencies_ms: Per-operation wall-clock latency in milliseconds,
-        one entry per successful operation.
-    :param failures: Failure reason (e.g. ``"HTTP 500"`` / an exception
-        class name) mapped to how many times it occurred.
-    :param wall_time: Total elapsed seconds for the run, used for throughput.
-    """
-
-    latencies_ms: list[float] = field(default_factory=list)
-    failures: dict[str, int] = field(default_factory=dict)
-    wall_time: float = 0.0
-
-    @property
-    def n_success(self) -> int:
-        """Number of operations that completed without error."""
-        return len(self.latencies_ms)
-
-    @property
-    def n_failures(self) -> int:
-        """Total failed operations across all reasons."""
-        return sum(self.failures.values())
-
-    @property
-    def throughput(self) -> float:
-        """Successful operations per second over the run's wall time."""
-        return self.n_success / self.wall_time if self.wall_time > 0 else 0.0
-
-    def record_failure(self, reason: str) -> None:
-        """Increment the count for one failure *reason*."""
-        self.failures[reason] = self.failures.get(reason, 0) + 1
-
-    def percentile(self, p: float) -> float:
-        """Return the *p*-th percentile latency in ms (ceil-index method).
-
-        :param p: Percentile in ``[0, 100]``, e.g. ``99`` for p99.
-        :returns: The latency at that percentile, or ``0.0`` when no
-            successful operation was recorded.
-        """
-        if not self.latencies_ms:
-            return 0.0
-        ordered = sorted(self.latencies_ms)
-        idx = max(0, math.ceil(p / 100 * len(ordered)) - 1)
-        return ordered[idx]
-
-    def mean_ms(self) -> float:
-        """Mean latency in ms, or ``0.0`` when no operation succeeded."""
-        return statistics.mean(self.latencies_ms) if self.latencies_ms else 0.0
-
-    def max_ms(self) -> float:
-        """Maximum latency in ms, or ``0.0`` when no operation succeeded."""
-        return max(self.latencies_ms) if self.latencies_ms else 0.0
-
-
-def _run_to_dict(result: RunResult) -> dict[str, object]:
-    """Flatten one :class:`RunResult` into a JSON-serializable per-run row."""
-    return {
-        "n_success": result.n_success,
-        "n_failures": result.n_failures,
-        "failures": dict(result.failures),
-        "wall_time_s": result.wall_time,
-        "mean_ms": result.mean_ms(),
-        "p50_ms": result.percentile(50),
-        "p95_ms": result.percentile(95),
-        "p99_ms": result.percentile(99),
-        "max_ms": result.max_ms(),
-        "rps": result.throughput,
-    }
-
-
-def aggregate(results: list[RunResult]) -> dict[str, object]:
-    """Fold per-run results into ``{"runs": [...], "summary": {...}}``.
-
-    The ``summary`` averages each metric across runs. Its keys mirror
-    MLflow's gateway benchmark (``avg_mean_ms`` / ``avg_p50_ms`` /
-    ``avg_p99_ms`` / ``avg_rps``) plus ``avg_p95_ms``, so the workspace ETL
-    that flattens ``summary`` works unchanged.
-
-    :param results: One :class:`RunResult` per timed run (warmup excluded).
-    :returns: A dict with a per-run ``runs`` list and an averaged
-        ``summary`` (empty ``summary`` when *results* is empty).
-    """
-    runs = [_run_to_dict(r) for r in results]
-    if not results:
-        return {"runs": runs, "summary": {}}
-    summary = {
-        "avg_mean_ms": statistics.mean(r.mean_ms() for r in results),
-        "avg_p50_ms": statistics.mean(r.percentile(50) for r in results),
-        "avg_p95_ms": statistics.mean(r.percentile(95) for r in results),
-        "avg_p99_ms": statistics.mean(r.percentile(99) for r in results),
-        "avg_rps": statistics.mean(r.throughput for r in results),
-    }
-    return {"runs": runs, "summary": summary}
-
-
-def check_thresholds(
-    results: list[RunResult],
-    *,
-    min_rps: float | None = None,
-    max_p50_ms: float | None = None,
-    max_p99_ms: float | None = None,
-) -> bool:
-    """Check averaged results against optional CI thresholds.
-
-    :param results: Timed runs for one journey.
-    :param min_rps: Fail if average throughput is below this (req/s).
-    :param max_p50_ms: Fail if average p50 latency exceeds this (ms).
-    :param max_p99_ms: Fail if average p99 latency exceeds this (ms).
-    :returns: ``True`` when every supplied threshold passes (vacuously
-        true when none are supplied or *results* is empty).
-    """
-    if not results:
-        return True
-    avg_rps = statistics.mean(r.throughput for r in results)
-    avg_p50 = statistics.mean(r.percentile(50) for r in results)
-    avg_p99 = statistics.mean(r.percentile(99) for r in results)
-    passed = True
-
-    if min_rps is not None and avg_rps < min_rps:
-        console.print(
-            f"  [red]THRESHOLD FAILED:[/red] avg throughput {avg_rps:.0f} req/s"
-            f" < minimum {min_rps:.0f} req/s"
-        )
-        passed = False
-    if max_p50_ms is not None and avg_p50 > max_p50_ms:
-        console.print(
-            f"  [red]THRESHOLD FAILED:[/red] avg P50 {avg_p50:.1f} ms"
-            f" > maximum {max_p50_ms:.1f} ms"
-        )
-        passed = False
-    if max_p99_ms is not None and avg_p99 > max_p99_ms:
-        console.print(
-            f"  [red]THRESHOLD FAILED:[/red] avg P99 {avg_p99:.1f} ms"
-            f" > maximum {max_p99_ms:.1f} ms"
-        )
-        passed = False
-    return passed
-
-
-def print_results(journey_name: str, results: list[RunResult]) -> None:
-    """Render per-run and averaged metrics for one journey as a rich table.
-
-    :param journey_name: Journey label used as the table title.
-    :param results: Timed runs to display.
-    """
-    table = Table(
-        title=journey_name,
-        show_header=True,
-        header_style="bold cyan",
-        box=None,
-        padding=(0, 2),
-        title_justify="left",
-    )
-    table.add_column("Run", style="dim", width=5)
-    table.add_column("Mean ms", justify="right")
-    table.add_column("P50 ms", justify="right")
-    table.add_column("P95 ms", justify="right")
-    table.add_column("P99 ms", justify="right")
-    table.add_column("Max ms", justify="right")
-    table.add_column("Req/s", justify="right")
-    table.add_column("Failures", justify="right")
-
-    for i, r in enumerate(results):
-        fail_str = f"[red]{r.n_failures}[/red]" if r.n_failures else "0"
-        table.add_row(
-            str(i + 1),
-            f"{r.mean_ms():.1f}",
-            f"{r.percentile(50):.1f}",
-            f"{r.percentile(95):.1f}",
-            f"{r.percentile(99):.1f}",
-            f"{r.max_ms():.1f}",
-            f"{r.throughput:.0f}",
-            fail_str,
-        )
-
-    if len(results) > 1:
-        table.add_section()
-        table.add_row(
-            "[bold]avg[/bold]",
-            f"[bold]{statistics.mean(r.mean_ms() for r in results):.1f}[/bold]",
-            f"[bold]{statistics.mean(r.percentile(50) for r in results):.1f}[/bold]",
-            f"[bold]{statistics.mean(r.percentile(95) for r in results):.1f}[/bold]",
-            f"[bold]{statistics.mean(r.percentile(99) for r in results):.1f}[/bold]",
-            f"[bold]{statistics.mean(r.max_ms() for r in results):.1f}[/bold]",
-            f"[bold]{statistics.mean(r.throughput for r in results):.0f}[/bold]",
-            "",
-        )
-
-    console.print()
-    console.print(table)
-
-    combined: dict[str, int] = {}
-    for r in results:
-        for reason, count in r.failures.items():
-            combined[reason] = combined.get(reason, 0) + count
-    if combined:
-        console.print("  [red]Failure breakdown:[/red]")
-        for reason, count in sorted(combined.items(), key=lambda kv: -kv[1]):
-            console.print(f"    {reason}: {count}")

dev/benchmarks/omnigent/run.py

@@ -1,253 +0,0 @@
-"""Omnigent user-journey benchmark runner.
-
-Boots a real ``omnigent server`` against a SQLite DB (no runner, no LLM),
-drives the selected HTTP journeys under load, prints per-journey latency /
-throughput tables, and writes a versioned JSON report. Exits non-zero if any
-supplied threshold is breached.
-
-Runs in the project venv — it imports ``omnigent`` and ``tests._helpers`` and
-spawns the real server, so it is NOT a standalone PEP 723 script. Invoke with
-``--no-sync`` so ``uv`` uses the existing environment instead of rebuilding the
-project (which triggers a web-UI build that fails in a worktree)::
-
-    uv run --no-sync dev/benchmarks/omnigent/run.py
-    uv run --no-sync dev/benchmarks/omnigent/run.py --journeys list_sessions,get_session
-    uv run --no-sync dev/benchmarks/omnigent/run.py --requests 500 --concurrency 25 --runs 3
-    uv run --no-sync dev/benchmarks/omnigent/run.py --output bench.json --max-p50-ms 25
-
-The JSON is the contract consumed by the workspace Databricks ETL notebook —
-see ``README.md``.
-"""
-
-from __future__ import annotations
-
-import argparse
-import asyncio
-import datetime
-import json
-import sys
-from pathlib import Path
-
-# Allow ``uv run <path>`` (no package context) to import the sibling modules.
-sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
-
-from dev.benchmarks.omnigent.environment import BenchEnvironment
-from dev.benchmarks.omnigent.journeys import (
-    ALL_JOURNEYS,
-    Journey,
-    resolve_journeys,
-    run_latency,
-    run_throughput,
-)
-from dev.benchmarks.omnigent.measure import (
-    RunResult,
-    aggregate,
-    check_thresholds,
-    console,
-    print_results,
-)
-from dev.benchmarks.omnigent.schema import build_report
-
-# Harness label stamped in the report: HTTP/DB journeys drive no agent turn;
-# runner journeys drive turns through the in-process openai-agents SDK harness.
-_HTTP_HARNESS = "http-only"
-_RUNNER_HARNESS = "openai-agents"
-
-
-def _backend_of(database_uri: str | None) -> str:
-    """Classify the DB URI into a coarse backend label for the report.
-
-    ``None`` is the harness's throwaway SQLite temp file. Otherwise key off the
-    URI scheme so the report (and the workspace dashboard) can group by backend.
-    """
-    if database_uri is None or database_uri.startswith("sqlite"):
-        return "sqlite"
-    if database_uri.startswith("postgres"):
-        return "postgres"
-    return "other"
-
-
-async def _run_journey(
-    journey: Journey, env: BenchEnvironment, args: argparse.Namespace
-) -> tuple[str, list[RunResult]]:
-    """Run one journey's timed runs, returning its report kind + per-run results.
-
-    A journey runs as throughput when ``--concurrency > 1`` and it is
-    concurrency-safe; otherwise as sequential latency.
-    """
-    as_throughput = args.concurrency > 1 and journey.concurrency_safe
-    results: list[RunResult] = []
-    for _ in range(args.runs):
-        if as_throughput:
-            results.append(
-                await run_throughput(
-                    journey,
-                    env,
-                    requests=args.requests,
-                    concurrency=args.concurrency,
-                    warmup=args.warmup,
-                )
-            )
-        else:
-            results.append(
-                await run_latency(journey, env, iterations=args.iterations, warmup=args.warmup)
-            )
-    return ("throughput" if as_throughput else "latency"), results
-
-
-async def run_benchmark(args: argparse.Namespace) -> tuple[dict[str, object], bool]:
-    """Run all selected journeys and build the report.
-
-    :returns: ``(report, passed)`` where *passed* is ``False`` if any journey
-        breached a supplied threshold.
-    """
-    journeys = resolve_journeys(args.journeys)
-    journey_results: dict[str, dict[str, object]] = {}
-    passed = True
-    backend = _backend_of(args.database_uri)
-
-    # Any full-turn journey needs the runner + mock LLM. A full env is a
-    # superset — HTTP journeys still run against it — so a mixed selection just
-    # boots with_runner=True. The harness label reflects what drove the turns.
-    with_runner = any(j.needs_runner for j in journeys)
-    harness = _RUNNER_HARNESS if with_runner else _HTTP_HARNESS
-
-    async with BenchEnvironment(with_runner=with_runner, database_uri=args.database_uri) as env:
-        for journey in journeys:
-            console.print(f"\n[bold]Benchmarking[/bold] {journey.name} [dim]({backend})[/dim]")
-            kind, results = await _run_journey(journey, env, args)
-            print_results(journey.name, results)
-            block = aggregate(results)
-            block["kind"] = kind
-            block["backend"] = backend
-            journey_results[journey.name] = block
-            if not check_thresholds(
-                results,
-                min_rps=args.min_rps,
-                max_p50_ms=args.max_p50_ms,
-                max_p99_ms=args.max_p99_ms,
-            ):
-                passed = False
-
-    config = {
-        "iterations": args.iterations,
-        "requests": args.requests,
-        "concurrency": args.concurrency,
-        "runs": args.runs,
-        "warmup": args.warmup,
-        "with_runner": with_runner,
-        "backend": backend,
-    }
-    generated_at = datetime.datetime.now(datetime.timezone.utc).isoformat()
-    report = build_report(
-        journey_results,
-        generated_at=generated_at,
-        config=config,
-        harness=harness,
-    )
-    return report, passed
-
-
-def _parse_args(argv: list[str]) -> argparse.Namespace:
-    parser = argparse.ArgumentParser(
-        prog="omnigent-benchmark",
-        description=__doc__,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
-    )
-    parser.add_argument(
-        "--journeys",
-        type=lambda s: [p.strip() for p in s.split(",") if p.strip()],
-        default=None,
-        metavar="A,B,C",
-        help=f"Comma-separated journeys to run. Default: all ({', '.join(ALL_JOURNEYS)}).",
-    )
-    parser.add_argument(
-        "--database-uri",
-        default=None,
-        metavar="URI",
-        help="DB the server boots against — a pre-seeded SQLite file or a "
-        "postgresql+psycopg://… instance (see seed.py). Default: a fresh "
-        "throwaway SQLite DB (empty — best-case numbers). The report's "
-        "`backend` field is derived from this.",
-    )
-    parser.add_argument(
-        "--iterations",
-        type=int,
-        default=100,
-        metavar="N",
-        help="Sequential operations per latency run (default: 100).",
-    )
-    parser.add_argument(
-        "--requests",
-        type=int,
-        default=500,
-        metavar="N",
-        help="Total operations per throughput run — used when --concurrency>1 (default: 500).",
-    )
-    parser.add_argument(
-        "--concurrency",
-        type=int,
-        default=1,
-        metavar="N",
-        help="Max in-flight operations. >1 runs concurrency-safe journeys as "
-        "throughput (default: 1 = sequential latency).",
-    )
-    parser.add_argument(
-        "--runs",
-        type=int,
-        default=3,
-        metavar="N",
-        help="Timed runs per journey; results are per-run and averaged (default: 3).",
-    )
-    parser.add_argument(
-        "--warmup",
-        type=int,
-        default=10,
-        metavar="N",
-        help="Warmup operations discarded before each run (default: 10).",
-    )
-    parser.add_argument(
-        "--output",
-        type=Path,
-        default=None,
-        metavar="FILE",
-        help="Write the JSON report to FILE (for CI artifact upload).",
-    )
-    parser.add_argument(
-        "--min-rps",
-        type=float,
-        default=None,
-        metavar="N",
-        help="Exit 1 if any journey's avg throughput falls below N req/s.",
-    )
-    parser.add_argument(
-        "--max-p50-ms",
-        type=float,
-        default=None,
-        metavar="N",
-        help="Exit 1 if any journey's avg P50 latency exceeds N ms.",
-    )
-    parser.add_argument(
-        "--max-p99-ms",
-        type=float,
-        default=None,
-        metavar="N",
-        help="Exit 1 if any journey's avg P99 latency exceeds N ms.",
-    )
-    return parser.parse_args(argv)
-
-
-def main(argv: list[str] | None = None) -> int:
-    args = _parse_args(argv if argv is not None else sys.argv[1:])
-    report, passed = asyncio.run(run_benchmark(args))
-    if args.output is not None:
-        args.output.write_text(json.dumps(report, indent=2))
-        console.print(f"\n  Results written to [cyan]{args.output}[/cyan]")
-    if not passed:
-        console.print("\n[red]One or more thresholds failed.[/red]")
-        return 1
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

dev/benchmarks/omnigent/sample_output.json

@@ -1,268 +0,0 @@
-{
-  "schema_version": 1,
-  "generated_at": "2026-07-08T18:30:00+00:00",
-  "git_sha": "0000000000000000000000000000000000000000",
-  "git_branch": "main",
-  "host": {
-    "platform": "macOS-15.5-arm64-arm-64bit",
-    "python": "3.12.8",
-    "cpu_count": 12
-  },
-  "harness": "http-only",
-  "config": {
-    "iterations": 100,
-    "requests": 500,
-    "concurrency": 1,
-    "runs": 3,
-    "warmup": 10,
-    "with_runner": false,
-    "backend": "sqlite"
-  },
-  "journeys": {
-    "list_sessions": {
-      "runs": [
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.6587757079978473,
-          "mean_ms": 6.586994149838574,
-          "p50_ms": 6.261250004172325,
-          "p95_ms": 7.65325000975281,
-          "p99_ms": 7.9127089702524245,
-          "max_ms": 38.27937500318512,
-          "rps": 151.79673261468648
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.6238234580378048,
-          "mean_ms": 6.237602901528589,
-          "p50_ms": 6.126708001829684,
-          "p95_ms": 7.152916979975998,
-          "p99_ms": 7.425624993629754,
-          "max_ms": 7.667875033803284,
-          "rps": 160.30176280087855
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.5675292499945499,
-          "mean_ms": 5.674769564066082,
-          "p50_ms": 5.528832960408181,
-          "p95_ms": 6.237249996047467,
-          "p99_ms": 7.393250009045005,
-          "max_ms": 11.83562504593283,
-          "rps": 176.20237194992208
-        }
-      ],
-      "summary": {
-        "avg_mean_ms": 6.166455538477749,
-        "avg_p50_ms": 5.972263655470063,
-        "avg_p95_ms": 7.014472328592092,
-        "avg_p99_ms": 7.577194657642394,
-        "avg_rps": 162.7669557884957
-      },
-      "kind": "latency",
-      "backend": "sqlite"
-    },
-    "create_session": {
-      "runs": [
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 2.4451411250047386,
-          "mean_ms": 24.450668342760764,
-          "p50_ms": 24.028874991927296,
-          "p95_ms": 27.25041698431596,
-          "p99_ms": 29.374166973866522,
-          "max_ms": 29.56758299842477,
-          "rps": 40.89743490769115
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 2.498600291030016,
-          "mean_ms": 24.985182073432952,
-          "p50_ms": 24.459875014144927,
-          "p95_ms": 28.391665953677148,
-          "p99_ms": 29.0600000298582,
-          "max_ms": 34.39550002804026,
-          "rps": 40.02240788932922
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 2.455423459003214,
-          "mean_ms": 24.553418274153955,
-          "p50_ms": 24.073333013802767,
-          "p95_ms": 27.43383398046717,
-          "p99_ms": 29.375000041909516,
-          "max_ms": 29.430416005197912,
-          "rps": 40.72617276394162
-        }
-      ],
-      "summary": {
-        "avg_mean_ms": 24.663089563449223,
-        "avg_p50_ms": 24.187361006624997,
-        "avg_p95_ms": 27.691972306153428,
-        "avg_p99_ms": 29.269722348544747,
-        "avg_rps": 40.548671853654
-      },
-      "kind": "latency",
-      "backend": "sqlite"
-    },
-    "get_session": {
-      "runs": [
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.5546917500323616,
-          "mean_ms": 5.5460037814918905,
-          "p50_ms": 5.360124981962144,
-          "p95_ms": 6.925499998033047,
-          "p99_ms": 7.144333969336003,
-          "max_ms": 7.331291970331222,
-          "rps": 180.28030882767922
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.49645508296089247,
-          "mean_ms": 4.963922067545354,
-          "p50_ms": 4.782959003932774,
-          "p95_ms": 5.978292028885335,
-          "p99_ms": 6.7617910099215806,
-          "max_ms": 6.881375040393323,
-          "rps": 201.42809174919327
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.460644083970692,
-          "mean_ms": 4.605880451854318,
-          "p50_ms": 4.526541975792497,
-          "p95_ms": 5.18629199359566,
-          "p99_ms": 5.445250018965453,
-          "max_ms": 5.790999974124134,
-          "rps": 217.08734244020465
-        }
-      ],
-      "summary": {
-        "avg_mean_ms": 5.0386021002971875,
-        "avg_p50_ms": 4.889875320562472,
-        "avg_p95_ms": 6.030028006838013,
-        "avg_p99_ms": 6.450458332741012,
-        "avg_rps": 199.59858100569238
-      },
-      "kind": "latency",
-      "backend": "sqlite"
-    },
-    "load_conversation_history": {
-      "runs": [
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.20811437495285645,
-          "mean_ms": 2.080691678565927,
-          "p50_ms": 2.037000027485192,
-          "p95_ms": 2.5742079596966505,
-          "p99_ms": 2.768124977592379,
-          "max_ms": 2.784749958664179,
-          "rps": 480.50501087516284
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.19513549999101087,
-          "mean_ms": 1.9509158097207546,
-          "p50_ms": 1.9018329912796617,
-          "p95_ms": 2.284207963384688,
-          "p99_ms": 2.4481670116074383,
-          "max_ms": 2.5021659675985575,
-          "rps": 512.4644157757384
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 0.19278304203180596,
-          "mean_ms": 1.9274150469573215,
-          "p50_ms": 1.8819589749909937,
-          "p95_ms": 2.2150420118123293,
-          "p99_ms": 2.2878749878145754,
-          "max_ms": 2.316958038136363,
-          "rps": 518.7178236532945
-        }
-      ],
-      "summary": {
-        "avg_mean_ms": 1.9863408450813342,
-        "avg_p50_ms": 1.9402639979186158,
-        "avg_p95_ms": 2.3578193116312227,
-        "avg_p99_ms": 2.5013889923381307,
-        "avg_rps": 503.8957501013986
-      },
-      "kind": "latency",
-      "backend": "sqlite"
-    },
-    "search_sessions": {
-      "runs": [
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 8.150427000015043,
-          "mean_ms": 81.50338126753923,
-          "p50_ms": 80.11816703947261,
-          "p95_ms": 90.9090840141289,
-          "p99_ms": 94.67683301772922,
-          "max_ms": 96.44366696011275,
-          "rps": 12.26929582950874
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 8.152122708968818,
-          "mean_ms": 81.5203430026304,
-          "p50_ms": 79.57212498877198,
-          "p95_ms": 95.20591603359208,
-          "p99_ms": 98.80137501750141,
-          "max_ms": 99.93629204109311,
-          "rps": 12.266743714490682
-        },
-        {
-          "n_success": 100,
-          "n_failures": 0,
-          "failures": {},
-          "wall_time_s": 8.053999124967959,
-          "mean_ms": 80.53909589187242,
-          "p50_ms": 79.52124997973442,
-          "p95_ms": 91.39445802429691,
-          "p99_ms": 93.2748339837417,
-          "max_ms": 94.79766699951142,
-          "rps": 12.416192061654566
-        }
-      ],
-      "summary": {
-        "avg_mean_ms": 81.18760672068068,
-        "avg_p50_ms": 79.73718066932634,
-        "avg_p95_ms": 92.50315269067262,
-        "avg_p99_ms": 95.58434733965744,
-        "avg_rps": 12.317410535217997
-      },
-      "kind": "latency",
-      "backend": "sqlite"
-    }
-  }
-}

dev/benchmarks/omnigent/schema.py

@@ -1,89 +0,0 @@
-"""Benchmark report schema + metadata capture.
-
-:func:`build_report` assembles the single JSON document the harness writes.
-Its per-journey ``summary`` + ``runs`` shape mirrors MLflow's gateway
-benchmark so the workspace ETL notebook flattens it unchanged — keyed by
-journey (and ``harness``) instead of ``backend``. Bump :data:`SCHEMA_VERSION`
-whenever the document's shape changes so the ETL can branch on it.
-"""
-
-from __future__ import annotations
-
-import platform
-import subprocess
-
-# Incremented on any breaking change to the report document shape below.
-SCHEMA_VERSION = 1
-
-
-def _git(*args: str) -> str:
-    """Run ``git *args`` at the repo root, returning stripped stdout or ``""``.
-
-    Never raises: a missing git, detached checkout, or non-zero exit all
-    surface as an empty string so a benchmark run outside a clean checkout
-    still produces a valid report.
-    """
-    try:
-        out = subprocess.run(
-            ["git", *args],
-            capture_output=True,
-            text=True,
-            timeout=10,
-            check=False,
-        )
-    except (OSError, subprocess.SubprocessError):
-        return ""
-    return out.stdout.strip() if out.returncode == 0 else ""
-
-
-def git_sha() -> str:
-    """Return the current commit SHA, or ``""`` when unavailable."""
-    return _git("rev-parse", "HEAD")
-
-
-def git_branch() -> str:
-    """Return the current branch name, or ``""`` when detached/unavailable."""
-    return _git("rev-parse", "--abbrev-ref", "HEAD")
-
-
-def host_info() -> dict[str, object]:
-    """Capture coarse host facts for cross-machine result comparison."""
-    import os
-
-    return {
-        "platform": platform.platform(),
-        "python": platform.python_version(),
-        "cpu_count": os.cpu_count(),
-    }
-
-
-def build_report(
-    journey_results: dict[str, dict[str, object]],
-    *,
-    generated_at: str,
-    config: dict[str, object],
-    harness: str,
-) -> dict[str, object]:
-    """Assemble the full benchmark report document.
-
-    :param journey_results: Per-journey ``{"kind", "runs", "summary"}``
-        blocks (each ``runs``/``summary`` produced by
-        :func:`measure.aggregate`), keyed by journey name.
-    :param generated_at: ISO-8601 timestamp stamped by the caller (kept out
-        of this pure function so it stays deterministic under test).
-    :param config: The run's knobs (iterations, requests, concurrency, runs,
-        mock_llm) for provenance.
-    :param harness: Harness driving full-turn journeys, e.g.
-        ``"openai-agents"``.
-    :returns: The JSON-serializable report document.
-    """
-    return {
-        "schema_version": SCHEMA_VERSION,
-        "generated_at": generated_at,
-        "git_sha": git_sha(),
-        "git_branch": git_branch(),
-        "host": host_info(),
-        "harness": harness,
-        "config": config,
-        "journeys": journey_results,
-    }

dev/benchmarks/omnigent/seed.py

@@ -1,226 +0,0 @@
-"""Deterministic corpus seeder for the performance benchmark.
-
-The v1 harness booted an empty DB, so the read journeys measured a best-case
-near-empty table. This seeds a sizeable, realistic corpus directly through the
-store API (no HTTP, no runner) so ``list_sessions`` / ``get_session`` /
-``load_conversation_history`` read a production-shaped volume.
-
-Writes to the same DB URI the server later boots against; startup migrations
-are an idempotent no-op on an at-head DB. The seed is deterministic (fixed RNG,
-fixed counts) so the same config always yields the same corpus — which is what
-makes "seed once, reuse" and the schema-drift guard sound.
-
-Listable-corpus recipe, per session (the permission grant is the gotcha — the
-loopback server resolves every request to user ``"local"`` and
-``list_sessions`` filters by it):
-
-1. ``create_session_with_agent`` — conversation + session-scoped agent row.
-2. ``permission_store.grant("local", sid, LEVEL_OWNER)`` — makes it listable.
-3. one batched ``append(sid, items)`` — user-role message items.
-
-Run standalone::
-
-    uv run --no-sync dev/benchmarks/omnigent/seed.py \
-        --database-uri sqlite:///tmp/bench.db --sessions 5000 --items-per-session 50
-"""
-
-from __future__ import annotations
-
-import argparse
-import random
-import sys
-from pathlib import Path
-
-# Allow ``uv run <path>`` (no package context) to import omnigent + siblings.
-sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
-
-from omnigent.db.utils import _get_head_db_revision, generate_agent_id
-from omnigent.entities import MessageData, NewConversationItem
-from omnigent.server.auth import LEVEL_OWNER, RESERVED_USER_LOCAL
-from omnigent.stores.conversation_store.sqlalchemy_store import SqlAlchemyConversationStore
-from omnigent.stores.permission_store.sqlalchemy_store import SqlAlchemyPermissionStore
-
-# Alembic head the current seed logic was authored against. The drift guard
-# (scripts/check_benchmark_seed_schema.py) fails CI when the repo's head moves
-# past this without the seed being regenerated + this bumped. Keep in sync via
-# ``_get_head_db_revision(...)`` (see the guard script).
-SEED_SCHEMA_REVISION = "v1a2b3c4d5e6"
-
-# Label key stamped on the first seeded session recording the corpus config, so
-# a later run can detect an existing (and matching) seed and skip re-seeding.
-_SEED_META_LABEL = "omni_bench_seed"
-
-# Fixed identifiers so the corpus is byte-stable across runs at a given config.
-_AGENT_NAME = "bench-agent"
-_DEFAULT_SESSIONS = 5000
-_DEFAULT_ITEMS = 50
-_DEFAULT_RNG_SEED = 1234
-
-# A pool of realistic-ish message fragments; the RNG assembles item text from
-# these so search_text has lexical variety without external data.
-_FRAGMENTS = (
-    "investigate the failing migration",
-    "the runner keeps disconnecting under load",
-    "add pagination to the sessions endpoint",
-    "why does the policy classifier time out",
-    "refactor the conversation store append path",
-    "benchmark the list endpoints against postgres",
-    "the web UI drops the last streamed token",
-    "trace the tunnel handshake for this runner id",
-    "summarize the changes in this pull request",
-    "reproduce the elicitation race on reconnect",
-)
-
-
-def _meta_value(sessions: int, items_per_session: int, rng_seed: int) -> str:
-    """Serialize the corpus config into the seed-marker label value."""
-    return (
-        f"sessions={sessions};items={items_per_session};rng={rng_seed};rev={SEED_SCHEMA_REVISION}"
-    )
-
-
-def _existing_seed_meta(conv: SqlAlchemyConversationStore) -> str | None:
-    """Return the seed-marker label value if a bench corpus already exists.
-
-    Looks up the most recent ``bench-agent`` session and reads its
-    ``omni_bench_seed`` label. ``None`` means no (recognizable) seed present.
-    """
-    listing = conv.list_conversations(limit=1, agent_name=_AGENT_NAME)
-    if not listing.data:
-        return None
-    marked = conv.get_conversation(listing.data[0].id)
-    return marked.labels.get(_SEED_META_LABEL) if marked is not None else None
-
-
-def _make_items(rng: random.Random, count: int) -> list[NewConversationItem]:
-    """Build *count* deterministic user-role message items.
-
-    User-role only: assistant messages require an ``agent`` field the store
-    only assigns after a real turn, and the seeded read path is role-agnostic.
-    """
-    items: list[NewConversationItem] = []
-    for i in range(count):
-        text = f"{rng.choice(_FRAGMENTS)} (item {i})"
-        items.append(
-            NewConversationItem(
-                type="message",
-                response_id=f"resp_seed_{i}",
-                data=MessageData(role="user", content=[{"type": "input_text", "text": text}]),
-            )
-        )
-    return items
-
-
-def seed(
-    db_uri: str,
-    *,
-    sessions: int = _DEFAULT_SESSIONS,
-    items_per_session: int = _DEFAULT_ITEMS,
-    rng_seed: int = _DEFAULT_RNG_SEED,
-    reseed: bool = False,
-) -> int:
-    """Seed *sessions* sessions × *items_per_session* items into *db_uri*.
-
-    Idempotent: if a matching seed already exists (same config + schema
-    revision) it is left untouched unless *reseed* is set. Constructing the
-    store runs migrations to head on first init, so *db_uri* need not
-    pre-exist.
-
-    :param db_uri: SQLAlchemy URI the server will also boot against, e.g.
-        ``"sqlite:///abs/bench.db"`` or ``"postgresql+psycopg://…"``.
-    :param sessions: Number of listable sessions to create.
-    :param items_per_session: Conversation items appended to each session.
-    :param rng_seed: Seed for the deterministic text RNG.
-    :param reseed: Seed even when a matching corpus is already present.
-    :returns: The number of sessions created (0 when a matching seed is reused).
-    """
-    conv = SqlAlchemyConversationStore(db_uri)
-    perms = SqlAlchemyPermissionStore(db_uri)
-
-    want = _meta_value(sessions, items_per_session, rng_seed)
-    if not reseed:
-        existing = _existing_seed_meta(conv)
-        if existing == want:
-            print(f"seed: matching corpus already present ({want}); skipping")
-            return 0
-        if existing is not None:
-            print(f"seed: existing corpus differs ({existing!r} != {want!r}); pass --reseed")
-            return 0
-
-    perms.ensure_user(RESERVED_USER_LOCAL)
-    rng = random.Random(rng_seed)
-
-    last_sid = ""
-    for s in range(sessions):
-        created = conv.create_session_with_agent(
-            agent_id=generate_agent_id(),
-            agent_name=_AGENT_NAME,
-            agent_bundle_location="bench/seed",  # never validated on the read path
-            agent_description=None,
-            title=f"bench session {s}: {rng.choice(_FRAGMENTS)}",
-        )
-        sid = created.conversation.id
-        last_sid = sid
-        perms.grant(RESERVED_USER_LOCAL, sid, LEVEL_OWNER)
-        if items_per_session:
-            conv.append(sid, _make_items(rng, items_per_session))
-        if sessions >= 100 and s % (sessions // 10) == 0 and s:
-            print(f"seed: {s}/{sessions} sessions")
-
-    # Stamp the corpus config on the LAST (newest) session — that's the one
-    # ``_existing_seed_meta``'s default desc listing returns, so the reuse
-    # check finds it regardless of corpus size.
-    if last_sid:
-        conv.set_labels(last_sid, {_SEED_META_LABEL: want})
-
-    print(f"seed: created {sessions} sessions × {items_per_session} items ({want})")
-    return sessions
-
-
-def _parse_args(argv: list[str]) -> argparse.Namespace:
-    parser = argparse.ArgumentParser(
-        prog="omnigent-benchmark-seed",
-        description=__doc__,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
-    )
-    parser.add_argument(
-        "--database-uri",
-        metavar="URI",
-        help="DB to seed. Required unless --print-head.",
-    )
-    parser.add_argument("--sessions", type=int, default=_DEFAULT_SESSIONS, metavar="N")
-    parser.add_argument("--items-per-session", type=int, default=_DEFAULT_ITEMS, metavar="N")
-    parser.add_argument("--rng-seed", type=int, default=_DEFAULT_RNG_SEED, metavar="N")
-    parser.add_argument(
-        "--reseed",
-        action="store_true",
-        help="Seed even if a matching corpus is already present.",
-    )
-    parser.add_argument(
-        "--print-head",
-        action="store_true",
-        help="Print the repo's Alembic head revision and exit (drift-check helper).",
-    )
-    return parser.parse_args(argv)
-
-
-def main(argv: list[str] | None = None) -> int:
-    args = _parse_args(argv if argv is not None else sys.argv[1:])
-    if args.print_head:
-        print(_get_head_db_revision("sqlite:///:memory:"))
-        return 0
-    if not args.database_uri:
-        print("seed: --database-uri is required (unless --print-head)", file=sys.stderr)
-        return 2
-    seed(
-        args.database_uri,
-        sessions=args.sessions,
-        items_per_session=args.items_per_session,
-        rng_seed=args.rng_seed,
-        reseed=args.reseed,
-    )
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/check_benchmark_seed_schema.py

@@ -1,72 +0,0 @@
-"""Guard: the benchmark seed's pinned schema revision must equal the DB head.
-
-The performance-benchmark seeder (``dev/benchmarks/omnigent/seed.py``) writes a
-corpus shaped for the *current* DB schema and pins the schema it was authored
-against in ``SEED_SCHEMA_REVISION``. When a migration moves the Alembic head,
-the seed logic (and any cached/generated corpus) may no longer match the schema
-— e.g. a new NOT NULL column the seeder doesn't populate. This check fails so
-the schema change and the seed are updated together.
-
-Unlike ``sync_version_py.py`` this is **check-only**: there's no safe automatic
-fix — regenerating the seed corpus and confirming the seeder still populates
-every required column is a human step. On drift, do:
-
-1. Update ``seed.py`` if the new schema needs new columns/values seeded.
-2. Set ``SEED_SCHEMA_REVISION`` to the new head (``seed.py --print-head``).
-3. Regenerate any cached corpus / ``sample_output.json``.
-
-Wired as a pre-commit hook on ``omnigent/db/db_models.py`` and
-``omnigent/db/migrations/`` (mirrors the ``sync-version-py`` local hook).
-
-Usage::
-
-    python scripts/check_benchmark_seed_schema.py
-"""
-
-from __future__ import annotations
-
-import argparse
-import sys
-from pathlib import Path
-
-# scripts/ -> repo root is one level up; ensure the package imports resolve
-# when pre-commit invokes this with an arbitrary CWD.
-_REPO_ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(_REPO_ROOT))
-
-from dev.benchmarks.omnigent.seed import SEED_SCHEMA_REVISION  # noqa: E402
-from omnigent.db.utils import _get_head_db_revision  # noqa: E402
-
-
-def main(argv: list[str] | None = None) -> int:
-    """Compare the repo's Alembic head to the seed's pinned revision.
-
-    :param argv: Argument list (defaults to ``sys.argv[1:]``). Any positional
-        filenames pre-commit passes are accepted and ignored.
-    :returns: ``0`` when in sync, ``1`` on drift (with a fix hint on stderr).
-    """
-    parser = argparse.ArgumentParser(description=__doc__)
-    # pre-commit passes the matched filenames; we check a fixed invariant, so
-    # accept and ignore them.
-    parser.add_argument("files", nargs="*", help=argparse.SUPPRESS)
-    parser.parse_args(argv)
-
-    # Head is read from the migration scripts on disk — no DB is contacted.
-    head = _get_head_db_revision("sqlite:///:memory:")
-    if head == SEED_SCHEMA_REVISION:
-        return 0
-
-    print(
-        f"benchmark seed schema drift: SEED_SCHEMA_REVISION is "
-        f"{SEED_SCHEMA_REVISION!r} but the DB head is {head!r}.\n"
-        f"The DB schema changed without the benchmark seed being refreshed. "
-        f"Update dev/benchmarks/omnigent/seed.py to seed any new schema, set "
-        f"SEED_SCHEMA_REVISION = {head!r}, and regenerate the cached corpus / "
-        f"sample_output.json.",
-        file=sys.stderr,
-    )
-    return 1
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

tests/benchmarks/test_benchmark_smoke.py

@@ -1,233 +0,0 @@
-"""Fast smoke test for the HTTP-journey benchmark harness.
-
-Runs the real harness (boots an ``omnigent server``, no runner / no LLM /
-no Databricks) with tiny counts and asserts the report shape and threshold
-logic. Runs on the normal CI lane — no creds, no ``databricks`` marker.
-
-The measurement and schema layers also get direct unit checks so their logic
-is covered without paying the server-boot cost.
-"""
-
-from __future__ import annotations
-
-import argparse
-from pathlib import Path
-from typing import cast
-
-import pytest
-
-from dev.benchmarks.omnigent import run as bench_run
-from dev.benchmarks.omnigent.journeys import ALL_JOURNEYS
-from dev.benchmarks.omnigent.measure import RunResult, aggregate, check_thresholds
-from dev.benchmarks.omnigent.schema import SCHEMA_VERSION, build_report
-
-_SMOKE_JOURNEYS = [
-    "list_sessions",
-    "create_session",
-    "get_session",
-    "load_conversation_history",
-    "search_sessions",
-]
-
-
-def _d(value: object) -> dict[str, object]:
-    """Narrow an opaque report node to a dict for indexing (test-side JSON nav)."""
-    assert isinstance(value, dict)
-    return cast(dict[str, object], value)
-
-
-def _smoke_args(**overrides: object) -> argparse.Namespace:
-    """Tiny-count args so the smoke run boots the server once and finishes fast."""
-    base: dict[str, object] = {
-        "journeys": _SMOKE_JOURNEYS,
-        "database_uri": None,  # empty throwaway SQLite — journeys self-seed a fallback
-        "iterations": 2,
-        "requests": 5,
-        "concurrency": 1,
-        "runs": 1,
-        "warmup": 1,
-        "output": None,
-        "min_rps": None,
-        "max_p50_ms": None,
-        "max_p99_ms": None,
-    }
-    base.update(overrides)
-    return argparse.Namespace(**base)
-
-
-# ── pure-layer unit checks (no server) ───────────────────────
-
-
-def test_percentile_and_throughput() -> None:
-    r = RunResult(latencies_ms=[10.0, 20.0, 30.0, 40.0], wall_time=2.0)
-    assert r.n_success == 4
-    assert r.percentile(50) == 20.0  # ceil-index: idx = ceil(0.5*4)-1 = 1
-    assert r.percentile(100) == 40.0
-    assert r.throughput == 2.0  # 4 successes / 2.0s
-
-
-def test_aggregate_summary_keys() -> None:
-    runs = [RunResult(latencies_ms=[5.0, 15.0], wall_time=1.0) for _ in range(2)]
-    block = aggregate(runs)
-    run_rows = cast(list[dict[str, object]], block["runs"])
-    assert len(run_rows) == 2
-    assert set(_d(block["summary"])) == {
-        "avg_mean_ms",
-        "avg_p50_ms",
-        "avg_p95_ms",
-        "avg_p99_ms",
-        "avg_rps",
-    }
-    assert run_rows[0]["n_success"] == 2
-
-
-def test_check_thresholds_pass_and_fail() -> None:
-    runs = [RunResult(latencies_ms=[10.0, 10.0], wall_time=1.0)]
-    assert check_thresholds(runs, min_rps=None, max_p50_ms=1000.0, max_p99_ms=None)
-    assert not check_thresholds(runs, min_rps=None, max_p50_ms=0.001, max_p99_ms=None)
-
-
-def test_build_report_shape() -> None:
-    block = aggregate([RunResult(latencies_ms=[1.0], wall_time=1.0)])
-    block["kind"] = "latency"
-    report = build_report(
-        {"list_sessions": block},
-        generated_at="2026-07-08T00:00:00+00:00",
-        config={"iterations": 2},
-        harness="http-only",
-    )
-    assert report["schema_version"] == SCHEMA_VERSION
-    assert report["generated_at"] == "2026-07-08T00:00:00+00:00"
-    assert set(report) >= {
-        "schema_version",
-        "generated_at",
-        "git_sha",
-        "git_branch",
-        "host",
-        "harness",
-        "config",
-        "journeys",
-    }
-    assert "list_sessions" in _d(report["journeys"])
-
-
-# ── end-to-end smoke (boots the server) ──────────────────────
-
-
-@pytest.mark.timeout(180)
-async def test_benchmark_smoke_end_to_end() -> None:
-    """Boot the server, run every HTTP journey once, validate the report."""
-    report, passed = await bench_run.run_benchmark(_smoke_args())
-
-    assert passed  # no thresholds supplied → vacuously passes
-    assert report["schema_version"] == SCHEMA_VERSION
-    assert _d(report["config"])["with_runner"] is False
-    # No --database-uri → the throwaway-SQLite path, labelled "sqlite".
-    assert _d(report["config"])["backend"] == "sqlite"
-
-    journeys = _d(report["journeys"])
-    for name in _SMOKE_JOURNEYS:
-        assert name in ALL_JOURNEYS
-        block = _d(journeys[name])
-        assert block["kind"] == "latency"
-        assert block["backend"] == "sqlite"
-        run_rows = cast(list[dict[str, object]], block["runs"])
-        assert run_rows, f"{name} produced no runs"
-        # Zero failures — a failure here means the HTTP path itself broke.
-        assert run_rows[0]["n_failures"] == 0, f"{name}: {run_rows[0]['failures']}"
-        assert cast(float, _d(block["summary"])["avg_p50_ms"]) >= 0.0
-
-
-@pytest.mark.timeout(180)
-async def test_benchmark_smoke_threshold_failure_exits_nonzero() -> None:
-    """An impossible p50 bound trips the threshold gate (passed=False)."""
-    _, passed = await bench_run.run_benchmark(
-        _smoke_args(journeys=["list_sessions"], max_p50_ms=0.0001)
-    )
-    assert not passed
-
-
-# ── runner (full-turn) journeys ──────────────────────────────
-
-_RUNNER_JOURNEYS = ["session_cold_start", "warm_turn", "time_to_first_token", "interrupt"]
-
-
-@pytest.mark.timeout(300)
-async def test_benchmark_smoke_runner_journeys() -> None:
-    """Run each full-turn journey once through server + runner + mock LLM.
-
-    First exercise of the ``with_runner=True`` path end-to-end. Tiny counts —
-    each cold-start iteration spawns a runner, so this is the slow smoke.
-    """
-    report, passed = await bench_run.run_benchmark(
-        _smoke_args(journeys=_RUNNER_JOURNEYS, iterations=1, warmup=1)
-    )
-
-    assert passed
-    # A runner journey was selected → env booted with_runner, harness stamped.
-    assert _d(report["config"])["with_runner"] is True
-    assert report["harness"] == "openai-agents"
-
-    journeys = _d(report["journeys"])
-    for name in _RUNNER_JOURNEYS:
-        assert ALL_JOURNEYS[name].needs_runner
-        block = _d(journeys[name])
-        run_rows = cast(list[dict[str, object]], block["runs"])
-        assert run_rows, f"{name} produced no runs"
-        # Zero failures — a failure here means the full-turn path broke.
-        assert run_rows[0]["n_failures"] == 0, f"{name}: {run_rows[0]['failures']}"
-
-
-# ── seeder (direct store, no server) ─────────────────────────
-
-
-def test_seed_creates_listable_corpus(tmp_path: Path) -> None:
-    """Seed a tiny corpus and confirm it is listable as "local" with history."""
-    from dev.benchmarks.omnigent import seed as seed_mod
-    from omnigent.server.auth import RESERVED_USER_LOCAL
-    from omnigent.stores.conversation_store.sqlalchemy_store import (
-        SqlAlchemyConversationStore,
-    )
-
-    db_uri = f"sqlite:///{tmp_path / 'seed.db'}"
-
-    created = seed_mod.seed(db_uri, sessions=6, items_per_session=4)
-    assert created == 6
-
-    conv = SqlAlchemyConversationStore(db_uri)
-    listing = conv.list_conversations(
-        limit=100,
-        agent_name="bench-agent",
-        accessible_by=RESERVED_USER_LOCAL,
-        has_agent_id=True,
-    )
-    assert len(listing.data) == 6  # all seeded sessions listable as "local"
-    assert len(conv.list_items(listing.data[0].id, limit=100).data) == 4
-
-    # Idempotent: a matching re-seed is a no-op.
-    assert seed_mod.seed(db_uri, sessions=6, items_per_session=4) == 0
-
-
-def test_seed_schema_revision_matches_head() -> None:
-    """SEED_SCHEMA_REVISION must equal the repo's current Alembic head.
-
-    This is the invariant the drift guard (scripts/check_benchmark_seed_schema)
-    enforces in CI/pre-commit; asserting it here fails fast in the unit suite
-    when a migration lands without the seed being refreshed.
-    """
-    from dev.benchmarks.omnigent import seed as seed_mod
-    from omnigent.db.utils import _get_head_db_revision
-
-    assert _get_head_db_revision("sqlite:///:memory:") == seed_mod.SEED_SCHEMA_REVISION
-
-
-def test_seed_schema_guard_passes_and_detects_drift(monkeypatch: pytest.MonkeyPatch) -> None:
-    """The drift-guard script exits 0 at head and 1 when the revision is stale."""
-    import scripts.check_benchmark_seed_schema as guard
-
-    assert guard.main([]) == 0
-
-    # The guard binds SEED_SCHEMA_REVISION by value at import (``from … import``),
-    # so patch the name in the guard's own namespace, not the seed module's.
-    monkeypatch.setattr(guard, "SEED_SCHEMA_REVISION", "stale_revision")
-    assert guard.main([]) == 1

tests/server/integration/mock_llm_server.py

@@ -1002,20 +1002,14 @@ async def set_fallback(request: Request) -> dict[str, object]:
     valid response even when per-test resets clear the regular queue
     (e.g. the server-level policy-classifier LLM queue).
 
-    Body: ``{"key": "<key>", "text": "<response-text>", "stream": false}``
-
-    ``stream`` (optional, default false): when true the fallback emits
-    ``response.output_text.delta`` events (one per word) before the completed
-    event — so a caller that subscribes to a streaming response sees
-    incremental deltas from the fallback, not just a single completed body.
+    Body: ``{"key": "<key>", "text": "<response-text>"}``
     """
     body = await request.json()
     key = body.get("key", _DEFAULT_KEY)
     text = body.get("text", "Mock LLM response")
-    stream = bool(body.get("stream", False))
     async with _state._lock:
         queue = _state.get_queue(key)
-        queue.fallback = QueuedResponse(text=text, stream=stream)
+        queue.fallback = QueuedResponse(text=text)
     return {"fallback_set": True, "key": key}