diff --git a/alembic/versions/081_doctrine_version.py b/alembic/versions/081_doctrine_version.py new file mode 100644 index 00000000..543688e8 --- /dev/null +++ b/alembic/versions/081_doctrine_version.py @@ -0,0 +1,57 @@ +"""Doctrine-version stamp on agent_spawn_sessions, for the eval harness. + +The eval harness (roboco/eval/) scores a (role, model/provider config) cohort +by replaying golden tasks through a real agent spawn. To attribute a quality +delta to a prompt/doctrine change (fable-mode, ponytail, a team-prompt edit, +...) the resulting spawn session needs to carry a fingerprint of exactly what +system prompt it ran with — otherwise two cohort runs are only comparable if +the operator remembers to keep everything else byte-for-byte identical. + +``doctrine_version`` is a short hash of the composed system prompt (base + +role + team + identity + doctrine layers) for that spawn, stamped at +``_finalize_spawn_session`` in roboco/runtime/orchestrator.py — NOT at +``_record_spawn_session`` (spawn creation). The composed prompt string itself +is not passed through the AgentConfig the finalize call site holds, but the +file it was written to (``config.blueprint_path``, from +``_generate_composed_prompt``) is still on disk and unchanged at finalize +time (nothing in the spawn/stop path deletes it), so the finalize call reads +it back and hashes it there. Every provider gets one — ``_prepare_agent_spawn`` +composes and writes the blueprint unconditionally, before provider/route +resolution, so GROK agents carry a real blueprint file too, same as Claude. +Nullable + additive: every existing row, and any row where the read +genuinely fails (a provider-parked stub instance that never actually +spawned — ``blueprint_path=Path()`` — an evicted temp dir, ...), simply gets +NULL — a pure quality-of-life addition to the sessions the eval harness +scores, never a hard requirement of the spawn/stop path. + +Revision ID: 081_doctrine_version +Revises: 080_task_project_budgets +Create Date: 2026-07-22 + +Note: re-chained onto 080_task_project_budgets (sibling PRs #652/#654 own +079/080 at this branch's base commit, da4d9b33, where 078 was the head); +080 does not exist in this worktree, so the local migration-graph/enum-parity +tests are expected to fail here until this branch integrates alongside its +siblings — the same expected-failure posture the budgets sibling reported. +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "081_doctrine_version" +down_revision = "080_task_project_budgets" +branch_labels: dict[str, str] | None = None +depends_on: dict[str, str] | None = None + + +def upgrade() -> None: + op.add_column( + "agent_spawn_sessions", + sa.Column("doctrine_version", sa.String(length=32), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("agent_spawn_sessions", "doctrine_version") diff --git a/pyproject.toml b/pyproject.toml index a8ac7f54..3ecb65d5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -198,6 +198,13 @@ select = [ # cost; ARG001 covers FastAPI path params the fake-GitHub handlers must # name but not read. "tests/e2e_smoke/*.py" = ["PLC0415", "ARG001"] +# The eval bench (offline CLI, not part of the served app) defers heavy/ +# optional imports — tests.e2e_smoke.harness, roboco.runtime.orchestrator, +# roboco.services.task, asyncpg — to call time for the same reason +# tests/e2e_smoke and roboco/services do; PLR0913 covers the stage-driving +# and scoring call surfaces (task/spawner/role/timeout tuples), same +# rationale as roboco/services/gateway/**. +"roboco/eval/*.py" = ["PLC0415", "PLR0913"] # PTH119: _grok_usage_json sanitizes the agent id with os.path.basename — the # path-injection sanitizer CodeQL's query models; the pathlib equivalent # (Path(...).name) is not recognized by that query, so we keep os.path here. @@ -233,10 +240,18 @@ select = [ # signature for keyword-argument compatibility (mypy override check), but the # stub bodies are empty — ARG002 would require renaming them, which breaks mypy. "tests/unit/services/test_optimal_grounding.py" = ["ARG002"] +# _FakeJudge.score overrides BenchJudge.score — same override-signature +# rationale as test_optimal_grounding.py above (a fixed fixture/diff/notes +# stand-in body has nothing to do with those args). +"tests/e2e_smoke/test_eval_bench.py" = ["ARG002"] # Collision-builder test helpers mirror the builder's many keyword inputs # (parent/project/intends/migration/shared/sequence) — bundling them would # hurt readability more than the arg count hurts. "tests/unit/gateway/test_collision_context.py" = ["PLR0913"] +# _metrics()'s many optional kwargs mirror DeterministicMetrics' own field +# count (a plain, no-defaults dataclass) — same rationale as the collision +# builder above. +"tests/unit/eval/test_scoring.py" = ["PLR0913"] # ============================================================================= # MyPy Configuration @@ -452,8 +467,15 @@ DEP002 = [ "types-passlib", "types-PyYAML", ] -# DEP003: Starlette is a transitive dep of FastAPI, but BaseHTTPMiddleware is needed -DEP003 = ["starlette"] +# DEP003: Starlette is a transitive dep of FastAPI, but BaseHTTPMiddleware is needed. +# "tests": roboco/eval/runner.py deliberately imports tests.e2e_smoke.harness/arcs +# (the offline eval bench's disposable-project machinery — see that module's +# docstring) — deptry sees the local `tests` package as an unresolvable +# transitive import since it isn't a PyPI dependency at all. The runtime side +# of this is guarded separately (an ImportError there raises a clear "this +# needs a source checkout" error), so this is a lint-posture ignore, not a +# correctness gap. +DEP003 = ["starlette", "tests"] [dependency-groups] dev = [ diff --git a/roboco/db/tables.py b/roboco/db/tables.py index fced701d..0dd9df08 100644 --- a/roboco/db/tables.py +++ b/roboco/db/tables.py @@ -1904,6 +1904,12 @@ class AgentSpawnSessionTable(Base): tool_calls: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) exit_reason: Mapped[str | None] = mapped_column(String(100), nullable=True) estimated_cost_usd: Mapped[float | None] = mapped_column(Float, nullable=True) + # Short hash of the composed system prompt this spawn ran with (migration + # 081_doctrine_version) — lets the eval harness (roboco/eval/) group spawn + # sessions by the exact prompt/doctrine version they ran, not just by + # model. Stamped at finalize (see _finalize_spawn_session); nullable — + # older rows and any read failure carry NULL. + doctrine_version: Mapped[str | None] = mapped_column(String(32), nullable=True) # Relationship to snapshots (backref for convenience) snapshots: Mapped[list["TokenUsageSnapshotTable"]] = relationship( diff --git a/roboco/eval/__init__.py b/roboco/eval/__init__.py new file mode 100644 index 00000000..7df46939 --- /dev/null +++ b/roboco/eval/__init__.py @@ -0,0 +1 @@ +"""Golden-task quality bench — see ``roboco/eval/runner.py``.""" diff --git a/roboco/eval/__main__.py b/roboco/eval/__main__.py new file mode 100644 index 00000000..af66b3be --- /dev/null +++ b/roboco/eval/__main__.py @@ -0,0 +1,95 @@ +"""CLI entrypoint for the eval bench. + + python -m roboco.eval run --role --cohort \\ + [--fixtures a,b] [--json-out path] + +NOT YET FUNCTIONAL: the real-spawn path (``OrchestratorStageSpawner``) is +deliberately cut — see ``roboco/eval/runner.py``'s module docstring's +"Real-spawn status" section — because a real container spawn's MCP wiring +would authenticate against the REAL production orchestrator, not this +harness's disposable one. ``run`` will raise ``NotImplementedError`` once it +reaches the first fixture. The only working path today is driving +``EvalRunner`` with an injected scripted ``StageSpawner`` from Python (see +``tests/e2e_smoke/test_eval_bench.py``); this CLI is wired for the day the +follow-up lands, not for use today. + +Offline dev/ops tool: no panel surface, no feature flag. Also runs from a +source checkout only (needs ``tests/e2e_smoke``, not shipped in containers +or wheels) plus the test Postgres (``ROBOCO_TEST_DB_*`` env vars, mirroring +the rest of the gate toolchain). +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +from roboco.eval.fixtures import FIXTURES, BenchTaskSpec +from roboco.eval.runner import EvalRunner + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(prog="python -m roboco.eval") + subparsers = parser.add_subparsers(dest="command", required=True) + + run = subparsers.add_parser( + "run", + help=( + "Replay the golden-task fixtures against one agent " + "[NOT YET FUNCTIONAL — real-spawn path is cut, see module docstring]" + ), + ) + run.add_argument( + "--role", required=True, help="Agent slug under test (e.g. be-dev-1)" + ) + run.add_argument( + "--cohort", + required=True, + help="Label for this run, for before/after comparison (e.g. baseline)", + ) + run.add_argument( + "--fixtures", + default=None, + help="Comma-separated fixture keys (default: every developer-role fixture)", + ) + run.add_argument( + "--json-out", + default=None, + type=Path, + help="Write the scored cohort result as JSON to this path", + ) + + return parser.parse_args(argv) + + +def _select_fixtures(spec: str | None) -> tuple[list[BenchTaskSpec] | None, str | None]: + """Resolve `--fixtures a,b` to a fixture list, or an error message for an + unknown key. `(None, None)` means "no filter — run every fixture".""" + if not spec: + return None, None + keys = {key.strip() for key in spec.split(",") if key.strip()} + fixtures = [f for f in FIXTURES if f.key in keys] + missing = keys - {f.key for f in fixtures} + if missing: + return None, f"Unknown fixture key(s): {sorted(missing)}" + return fixtures, None + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(sys.argv[1:] if argv is None else argv) + if args.command != "run": + return 1 + + fixtures, error = _select_fixtures(args.fixtures) + if error: + print(error, file=sys.stderr) + return 2 + + runner = EvalRunner() + runner.run_cohort(args.role, args.cohort, fixtures=fixtures, json_out=args.json_out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/roboco/eval/fixtures.py b/roboco/eval/fixtures.py new file mode 100644 index 00000000..a190273f --- /dev/null +++ b/roboco/eval/fixtures.py @@ -0,0 +1,262 @@ +"""Golden-task fixtures for the eval bench (see ``roboco/eval/runner.py``). + +Each ``BenchTaskSpec`` is a tiny, self-contained "golden task": a few +pre-seeded repo files, a task brief (title/description/acceptance criteria), +and a checked-in ``expectations`` note the local-model judge grades the final +PR diff + dev notes against. Fixture repo files are namespaced under +``bench//`` so every fixture can share one disposable project's git +history without colliding with the others (the runner seeds fixtures onto +the same project's default branch, one at a time). +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from roboco.models.base import TaskNature, TaskType + + +@dataclass(frozen=True) +class BenchTaskSpec: + """One golden task: seeded repo state + brief + graded expectation. + + ``target_role`` is always ``"developer"`` — the only role a task can be + freshly assigned to from PENDING with no prior work already done (QA / + documenter / cell-PM only ever pick up a task a developer has already + advanced through the lifecycle). Kept as an explicit field rather than + hardcoded at the call site so a future QA/PM-focused bench fixture has + somewhere to say otherwise. + """ + + key: str + title: str + description: str + acceptance_criteria: tuple[str, ...] + task_type: TaskType + nature: TaskNature + repo_files: tuple[tuple[str, str], ...] + expectations: str + target_role: str = "developer" + + +FIXTURES: tuple[BenchTaskSpec, ...] = ( + BenchTaskSpec( + key="bugfix-off-by-one", + title="Fix off-by-one in paginate()", + description=( + "`bench/bugfix-off-by-one/paginate.py`'s `paginate(items, page, " + "size)` drops the last item of every page because its slice end " + "is `page * size - 1` instead of `page * size`. Fix the slice " + "bound so every item appears exactly once across all pages." + ), + acceptance_criteria=( + "paginate(list(range(10)), page=1, size=3) returns [0, 1, 2]", + "paginate(list(range(10)), page=4, size=3) returns [9] (the " + "last, previously-dropped item)", + "No item is duplicated or skipped across pages 1..4 for size=3", + ), + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + repo_files=( + ( + "bench/bugfix-off-by-one/paginate.py", + "def paginate(items, page, size):\n" + " start = (page - 1) * size\n" + " end = page * size - 1\n" + " return items[start:end]\n", + ), + ), + expectations=( + "The fix changes the slice end to `page * size` (or an " + "equivalent that includes the final item). A correct diff " + "touches only paginate.py's slice bound; no new dependency, no " + "unrelated rewrite. Commit/PR notes should describe the " + "off-by-one root cause, not just 'fixed a bug'." + ), + ), + BenchTaskSpec( + key="bugfix-null-check", + title="Fix crash on empty input in summarize()", + description=( + "`bench/bugfix-null-check/stats.py`'s `summarize(values)` " + "divides by `len(values)` unconditionally, so it raises " + "ZeroDivisionError on an empty list instead of returning a " + "sane empty-input result. Add a guard." + ), + acceptance_criteria=( + "summarize([]) returns {'count': 0, 'total': 0, 'average': 0} " + "without raising", + "summarize([2, 4, 6]) still returns " + "{'count': 3, 'total': 12, 'average': 4}", + ), + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + repo_files=( + ( + "bench/bugfix-null-check/stats.py", + "def summarize(values):\n" + " total = sum(values)\n" + " count = len(values)\n" + " return {\n" + " 'count': count,\n" + " 'total': total,\n" + " 'average': total / count,\n" + " }\n", + ), + ), + expectations=( + "The fix adds an explicit empty-input guard (e.g. `if not " + "values: return {...}`) ahead of the division, without " + "changing the non-empty behavior. No new dependency; the " + "guard is the whole diff." + ), + ), + BenchTaskSpec( + key="small-feature-greet", + title="Add a greet() helper", + description=( + "`bench/small-feature-greet/greetings.py` has no greeting " + "helper yet. Add a `greet(name, formal=False)` function: " + "informal returns `f'Hi, {name}!'`, formal returns " + "`f'Good day, {name}.'`. Empty/whitespace-only `name` should " + "raise `ValueError`." + ), + acceptance_criteria=( + "greet('Ada') == 'Hi, Ada!'", + "greet('Ada', formal=True) == 'Good day, Ada.'", + "greet('') and greet(' ') both raise ValueError", + ), + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + repo_files=( + ( + "bench/small-feature-greet/greetings.py", + "# Greeting helpers for bench/small-feature-greet.\n", + ), + ), + expectations=( + "greet() is added to greetings.py matching both the informal " + "and formal wording exactly, plus the empty-name ValueError " + "guard. A minimal, additive diff — no unrelated changes to " + "the file's header comment." + ), + ), + BenchTaskSpec( + key="refactor-duplicate-normalize", + title="De-duplicate normalize_a / normalize_b", + description=( + "`bench/refactor-duplicate-normalize/normalize.py` has two " + "near-identical functions, `normalize_a` and `normalize_b` — " + "both strip whitespace and lowercase a string, differing only " + "in which module used to call them. Refactor into one shared " + "helper both call, preserving both public names as thin " + "wrappers so existing callers are unaffected." + ), + acceptance_criteria=( + "normalize_a(' Hello ') == 'hello'", + "normalize_b(' Hello ') == 'hello'", + "The duplicated strip/lower logic exists in exactly one place", + ), + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + repo_files=( + ( + "bench/refactor-duplicate-normalize/normalize.py", + "def normalize_a(text):\n" + " return text.strip().lower()\n" + "\n" + "\n" + "def normalize_b(text):\n" + " return text.strip().lower()\n", + ), + ), + expectations=( + "A single private helper (e.g. `_normalize`) holds the " + "strip/lower logic; normalize_a/normalize_b both delegate to " + "it and keep their existing signatures and return values " + "identical to before. No behavior change, pure de-duplication." + ), + ), + BenchTaskSpec( + key="docs-readme-flag", + title="Document the --dry-run flag", + description=( + "`bench/docs-readme-flag/cli.py` accepts a `--dry-run` flag " + "(prints what it would do instead of doing it) that isn't " + "mentioned anywhere in `bench/docs-readme-flag/README.md`. Add " + "a short section documenting it: what it does and an example " + "invocation." + ), + acceptance_criteria=( + "README.md documents --dry-run's behavior in prose", + "README.md shows an example command line using --dry-run", + ), + task_type=TaskType.DOCUMENTATION, + nature=TaskNature.TECHNICAL, + repo_files=( + ( + "bench/docs-readme-flag/cli.py", + "import argparse\n" + "\n" + "\n" + "def build_parser():\n" + " parser = argparse.ArgumentParser()\n" + " parser.add_argument('--dry-run', action='store_true')\n" + " return parser\n", + ), + ( + "bench/docs-readme-flag/README.md", + "# bench/docs-readme-flag\n\nA tiny CLI fixture.\n", + ), + ), + expectations=( + "README.md gains a section documenting --dry-run's actual " + "behavior (prints instead of acting) with a runnable example " + "invocation. cli.py itself is unchanged — this is a docs-only " + "task." + ), + ), + BenchTaskSpec( + key="research-magic-constant", + title="Research the magic constant in legacy_calc.py", + description=( + "`bench/research-magic-constant/legacy_calc.py`'s `compute()` " + "multiplies by `1.10000001` instead of the obvious `1.1`. " + "Investigate the surrounding code/comments for why, and commit " + "your findings as `bench/research-magic-constant/NOTES.md` — " + "this is a research task, not a code fix: legacy_calc.py stays " + "unchanged." + ), + acceptance_criteria=( + "NOTES.md exists and explains what the magic constant is " + "compensating for, based on the evidence in the file", + "NOTES.md recommends whether it's safe to simplify to 1.1, with reasoning", + "legacy_calc.py is not modified", + ), + task_type=TaskType.RESEARCH, + nature=TaskNature.TECHNICAL, + repo_files=( + ( + "bench/research-magic-constant/legacy_calc.py", + "# The 1.10000001 factor below is NOT a typo for 1.1 — it\n" + "# nudges the float rounding in compute()'s downstream\n" + "# int(...) truncation so historical invoice totals ending\n" + "# in .10 don't get truncated to one cent short. Changing\n" + "# this constant reopens ROBO-lore ticket #4471 (pre-git\n" + "# history) where 1.1 exactly caused a cent-level\n" + "# reconciliation mismatch on ~0.3% of invoices.\n" + "def compute(amount):\n" + " return int(amount * 1.10000001)\n", + ), + ), + expectations=( + "NOTES.md correctly identifies (from the file's own comment) " + "that the constant compensates for float-truncation rounding " + "in compute()'s int(...) cast, cites the historical " + "reconciliation-mismatch reasoning, and recommends AGAINST " + "simplifying to 1.1 without a broader fix — it should not " + "invent an unrelated explanation, and legacy_calc.py itself " + "must be untouched." + ), + ), +) diff --git a/roboco/eval/runner.py b/roboco/eval/runner.py new file mode 100644 index 00000000..b54c02dd --- /dev/null +++ b/roboco/eval/runner.py @@ -0,0 +1,1024 @@ +"""EvalRunner — golden-task quality bench for a (role, model/provider) cohort. + +Replays ``roboco/eval/fixtures.py``'s ``BenchTaskSpec`` fixtures through the +REAL delivery lifecycle: one real task (``TaskService.create``, ``source= +"eval_bench"``) through QA / docs / cell-PM review to a terminal state. Each +fixture is scored on deterministic metrics (final status, revision_count, +cycle time, tokens+cost via the ``agent_spawn_sessions`` task_id join) plus a +local-model judge comparing the final PR diff + notes against the fixture's +checked-in ``expectations`` note — see ``CohortResult.as_dict()``'s nested +``"judge"`` object, marked ``"non_deterministic": true`` so a naive cohort +diff never mistakes judge noise for a real regression. + +Environment reuse: the disposable project + real local git origin + +fake-GitHub REST + in-process API all come straight from +``tests.e2e_smoke.harness`` (the same machinery ``make e2e-smoke`` uses) — an +offline eval CLI has the exact same isolation needs a smoke test does +(no real GitHub, no leftover DB state between runs), so this reuses rather +than re-implements it. ``tests/e2e_smoke/harness.py``'s ``build_e2e_stack`` +took a ``pytest.TempPathFactory`` parameter; its type was relaxed to a +structural ``TmpPathFactory`` Protocol (see that module) so this non-pytest +caller can drive it with a plain temp-dir factory. Source-checkout-only: +``tests/`` is not shipped in containers or wheels, so importing it (guarded +in ``_bench_environment`` with a clear ``RuntimeError``) only works when this +CLI runs from a git clone, never an installed package. + +Vault safety: ``_bench_environment`` also patches ``obsidian_vault_enabled`` +(and the vault intake/KB/report sub-flags) to False for the whole run, so a +bench task/note/journal write never lands in the operator's real Obsidian +vault even when the ambient deployment has vault flags armed. + +Real-spawn status: CUT for this release. ``StageSpawner`` is the seam +between a real container spawn and a scripted stand-in, and +``OrchestratorStageSpawner`` — what would be the default, real +implementation — raises ``NotImplementedError`` at construction: a spawned +container's MCP servers resolve their orchestrator URL via +``_generate_mcp_config`` (``PROJECT_HOST_PATH`` -> the REAL production +hostname, or ``settings.port``), never the patched ``settings.api_url`` this +harness's disposable stack listens on — combined with ``_seed_company`` +seeding agents under their REAL production UUIDs, a real spawn here could +authenticate as e.g. be-dev-1 against the production orchestrator and act on +real tasks. Fixing the spawn-env wiring is a dedicated follow-up. The ONLY +working ``StageSpawner`` today is an injected scripted one (see +``tests/e2e_smoke/test_eval_bench.py``) that drives the SAME real MCP flow/do +tool functions e2e_smoke's ``ScriptedAgent`` uses — proving the runner's +polling/scoring/DB plumbing without touching Docker. ``python -m roboco.eval +run`` therefore does not work yet; it is wired for the day the follow-up +lands, not for use today. + +Scope cut: only developer-role fixtures are supported (``run_cohort`` +refuses any other role). QA/documenter/cell-PM only ever pick up a task a +developer has already advanced through the lifecycle — there is no +"freshly PENDING task assigned straight to QA" shape to bench them with the +same one-task-per-fixture design. A QA/PM-focused bench would need a +different fixture shape (a pre-built PR to review) and is future work. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +import re +import shutil +import statistics +import subprocess +import tempfile +import time +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any, Protocol, cast +from uuid import UUID, uuid4 + +import httpx +import pytest +import structlog +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine + +from roboco.agents_config import get_agent_role, get_agent_team +from roboco.config import settings +from roboco.eval.fixtures import FIXTURES, BenchTaskSpec +from roboco.foundation import identity as _foundation +from roboco.models import Team +from roboco.models.base import Complexity + +if TYPE_CHECKING: + from collections.abc import Callable, Iterable, Iterator, Sequence + + from tests.e2e_smoke.harness import E2EStack + +logger = structlog.get_logger() + +_TERMINAL_STATUSES = {"completed", "cancelled"} +# Statuses this bench cannot progress past without a human (CEO) — scored as +# a stall, distinct from a genuine timeout, but never mistaken for success. +_HUMAN_GATED_STATUSES = {"awaiting_ceo_approval", "blocked", "paused"} + +# status -> the role responsible for advancing it. A bench fixture is a leaf +# task (no parent, no PR-review gate — see the module docstring), so every +# other status (backlog, awaiting_pr_review, ...) never legitimately occurs; +# reaching one is scored as a stall (`_STAGE_ROLE.get(status)` -> None). +_STAGE_ROLE: dict[str, str] = { + "pending": "developer", + "claimed": "developer", + "in_progress": "developer", + "needs_revision": "developer", + "awaiting_qa": "qa", + "awaiting_documentation": "documenter", + "awaiting_pm_review": "cell_pm", +} +_ROLE_SUFFIX = {"qa": "qa", "documenter": "doc", "cell_pm": "pm"} +_TEAM_PREFIX = {"backend": "be", "frontend": "fe", "ux_ui": "ux"} + +# --------------------------------------------------------------------------- +# Scratch Postgres — mirrors tests/conftest.py's `_test_database_url` fixture +# (same env vars, same CREATE DATABASE + Base.metadata.create_all technique) +# without the pytest fixture machinery, since this runs from a plain CLI. +# --------------------------------------------------------------------------- + +_TEST_DB_HOST = os.environ.get("ROBOCO_TEST_DB_HOST", "localhost") +_TEST_DB_PORT = int(os.environ.get("ROBOCO_TEST_DB_PORT", "5432")) +_TEST_DB_USER = os.environ.get("ROBOCO_TEST_DB_USER", "roboco") +_TEST_DB_PASSWORD = os.environ.get("ROBOCO_TEST_DB_PASSWORD", "") +_TEST_DB_ADMIN_DB = os.environ.get("ROBOCO_TEST_DB_ADMIN_DB", "postgres") + + +def _scratch_db_url(database: str) -> str: + auth = _TEST_DB_USER + if _TEST_DB_PASSWORD: + auth = f"{_TEST_DB_USER}:{_TEST_DB_PASSWORD}" + return f"postgresql+asyncpg://{auth}@{_TEST_DB_HOST}:{_TEST_DB_PORT}/{database}" + + +@contextlib.contextmanager +def _scratch_database() -> Iterator[str]: + """Provision a throwaway ``roboco_eval_`` DB, build the real + schema, yield its URL, drop it on exit.""" + import asyncpg + + from roboco.db.base import Base + + db_name = f"roboco_eval_{uuid4().hex[:10]}" + + async def _create() -> None: + conn = await asyncpg.connect( + host=_TEST_DB_HOST, + port=_TEST_DB_PORT, + user=_TEST_DB_USER, + password=_TEST_DB_PASSWORD or None, + database=_TEST_DB_ADMIN_DB, + ) + try: + await conn.execute(f'CREATE DATABASE "{db_name}"') + finally: + await conn.close() + + url = _scratch_db_url(db_name) + engine = create_async_engine(url, future=True) + try: + async with engine.begin() as db_conn: + with contextlib.suppress(Exception): + await db_conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) + await db_conn.run_sync(Base.metadata.create_all) + finally: + await engine.dispose() + + async def _drop() -> None: + conn = await asyncpg.connect( + host=_TEST_DB_HOST, + port=_TEST_DB_PORT, + user=_TEST_DB_USER, + password=_TEST_DB_PASSWORD or None, + database=_TEST_DB_ADMIN_DB, + ) + try: + await conn.execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " + "WHERE datname = $1 AND pid <> pg_backend_pid()", + db_name, + ) + await conn.execute(f'DROP DATABASE IF EXISTS "{db_name}"') + finally: + await conn.close() + + asyncio.run(_create()) + try: + yield _scratch_db_url(db_name) + finally: + asyncio.run(_drop()) + + +class _ScratchTmpFactory: + """Minimal ``TmpPathFactory`` (see ``tests/e2e_smoke/harness.py``) for a + plain-script caller — ``build_e2e_stack`` calls ``.mktemp()`` exactly + once, so this only needs to satisfy that one call.""" + + def __init__(self, root: Path) -> None: + self._root = root + self._count = 0 + + def mktemp(self, basename: str, numbered: bool = True) -> Path: + self._count += 1 + name = f"{basename}{self._count}" if numbered else basename + path = self._root / name + path.mkdir(parents=True) + return path + + +# --------------------------------------------------------------------------- +# Disposable project + company +# --------------------------------------------------------------------------- + + +@dataclass +class BenchEnvironment: + stack: E2EStack + project_id: UUID + project_slug: str + team: Team + cell_id: UUID + cell_branch: str + + +def _seed_company(stack: E2EStack, slugs: Iterable[str]) -> None: + """Seed the canonical agents needed to run a fixture's whole lifecycle. + + Uses each slug's REAL fixed UUID from ``foundation.identity.AGENTS`` + (not a random one, unlike ``tests/e2e_smoke/arcs.py``'s ``seed_company``) + so that orchestrator-internal helpers keyed by that static registry + (``get_agent_role``, the UUID->slug reverse map, ...) resolve exactly as + they would in a real deployment. + """ + from roboco.db.tables import AgentTable + from roboco.models import AgentStatus + + async def _run(session: Any) -> None: + for slug in slugs: + row = _foundation.AGENTS[slug] + session.add( + AgentTable( + id=row.uuid, + name=slug, + slug=slug, + role=row.role, + team=row.team, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt=slug, + capabilities=[], + permissions={}, + metrics={}, + ) + ) + + stack.run_db(_run) + + +def _seed_project(stack: E2EStack, team: Team, created_by: UUID) -> tuple[UUID, str]: + from roboco.db.tables import ProjectTable + from roboco.utils.crypto import encrypt_token + + slug = f"eval-bench-{uuid4().hex[:8]}" + holder: dict[str, Any] = {} + + async def _run(session: Any) -> None: + project = ProjectTable( + id=uuid4(), + name=f"Eval bench {slug}", + slug=slug, + git_url=str(stack.origin), + default_branch="master", + protected_branches=["master"], + assigned_cell=team, + created_by=created_by, + is_active=True, + git_token_encrypted=encrypt_token("eval-bench-dummy-token"), + ) + session.add(project) + await session.flush() + holder["id"] = project.id + + stack.run_db(_run) + return holder["id"], slug + + +def _seed_bench_cell( + stack: E2EStack, project_id: UUID, team: Team, prefix: str +) -> tuple[UUID, str]: + """A minimal, never-advanced coordination parent so every fixture's leaf + task has a REAL non-default-branch merge target. + + A parentless leaf's PR would target the project's default branch, which + only the CEO may merge (``roboco.services.git``'s ``CEO_ONLY`` check) — + real production leaf tasks are always a cell/root's child for exactly + this reason. This cell task is cut once per environment and never + advanced past ``in_progress``; every fixture's leaf is created as its + child so ``cell_pm_complete`` merges into this cell branch instead. + """ + from tests.e2e_smoke.arcs import origin_branch, set_branch_name + + from roboco.models.base import TaskNature, TaskStatus, TaskType + from roboco.models.task import TaskCreateRequest + from roboco.services.task import EVAL_BENCH_SOURCE, get_task_service + + pm_uuid = _foundation.AGENTS[f"{prefix}-pm"].uuid + branch = f"feature/{team.value}/bench-cell-{uuid4().hex[:8]}" + origin_branch(stack, branch, start="master") + holder: dict[str, Any] = {} + + async def _run(session: Any) -> None: + req = TaskCreateRequest( + title="Bench cell coordination (internal, never advanced)", + description=( + "Internal coordination parent so bench leaf tasks merge into " + "a cell branch instead of the project's protected default " + "branch. Never claimed or advanced by any agent." + ), + acceptance_criteria=["n/a — coordination-only, never advanced"], + team=team, + created_by=pm_uuid, + task_type=TaskType.PLANNING, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.LOW, + project_id=project_id, + assigned_to=pm_uuid, + status=TaskStatus.IN_PROGRESS, + source=EVAL_BENCH_SOURCE, + confirmed_by_human=True, + ) + task = await get_task_service(session).create(req) + holder["id"] = task.id + + stack.run_db(_run) + cell_id = cast("UUID", holder["id"]) + set_branch_name(stack, cell_id, branch) + return cell_id, branch + + +@contextlib.contextmanager +def _bench_environment(dev_slug: str) -> Iterator[BenchEnvironment]: + """Stand up the disposable e2e_smoke-style stack + one bench project + + the fixed company of agents needed to run a fixture end to end.""" + try: + from tests.e2e_smoke.harness import build_e2e_stack + except ImportError as exc: + raise RuntimeError( + "the eval CLI runs from a source checkout; tests/ is not shipped " + "in containers or wheels — run `python -m roboco.eval` from a git " + "clone of the repo, not an installed package" + ) from exc + + team_str = get_agent_team(dev_slug) + if team_str not in _TEAM_PREFIX: + raise ValueError(f"{dev_slug!r} has no known cell team ({team_str!r})") + team = Team(team_str) + prefix = _TEAM_PREFIX[team_str] + + with _scratch_database() as db_url: + root_path = Path(tempfile.mkdtemp(prefix="roboco-eval-")) + try: + stack_cm = contextlib.contextmanager(build_e2e_stack) + with stack_cm(db_url, _ScratchTmpFactory(root_path)) as stack: + mp = pytest.MonkeyPatch() + mp.setattr(settings, "api_url", stack.base_url) + # A bench task/note/journal write must never land in the + # operator's REAL Obsidian vault. obsidian_vault_enabled is + # the single gate every writer seam (TaskService.create's + # materialize-on-create + status-transition touch, + # JournalService, A2AService) checks first — traced via + # `grep obsidian_vault_enabled roboco/services/{task,journal, + # a2a}.py` — so patching it False is sufficient on its own. + # The three sub-flags below gate background LOOPS (vault + # intake watcher, KB ingest, weekly report) that this harness + # never starts (no AgentOrchestrator.start() call) and are + # therefore already inert; patched anyway so a future harness + # change that does start them fails closed, not open. + mp.setattr(settings, "obsidian_vault_enabled", False) + mp.setattr(settings, "vault_intake_enabled", False) + mp.setattr(settings, "vault_kb_enabled", False) + mp.setattr(settings, "vault_report_enabled", False) + try: + dev_uuid = _foundation.AGENTS[dev_slug].uuid + _seed_company( + stack, + [dev_slug, f"{prefix}-qa", f"{prefix}-doc", f"{prefix}-pm"], + ) + project_id, project_slug = _seed_project(stack, team, dev_uuid) + cell_id, cell_branch = _seed_bench_cell( + stack, project_id, team, prefix + ) + yield BenchEnvironment( + stack=stack, + project_id=project_id, + project_slug=project_slug, + team=team, + cell_id=cell_id, + cell_branch=cell_branch, + ) + finally: + mp.undo() + finally: + shutil.rmtree(root_path, ignore_errors=True) + + +def _seed_fixture_repo(stack: E2EStack, fixture: BenchTaskSpec) -> None: + """Push the fixture's ``repo_files`` onto the project's default branch, + namespaced under ``bench//`` so sequential fixtures never collide.""" + from tests.e2e_smoke.harness import _git + + admin = stack.github.admin_clone + _git(admin, "fetch", "origin", "--prune") + _git(admin, "checkout", "-B", "master", "origin/master") + for rel_path, content in fixture.repo_files: + path = admin / rel_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content) + _git(admin, "add", rel_path) + _git(admin, "commit", "-m", f"chore(bench): seed {fixture.key} fixture repo state") + _git(admin, "push", "origin", "master") + + +def _fast_forward_branch(stack: E2EStack, branch: str, *, onto: str) -> None: + """Fast-forward `branch` to `onto`'s current tip (a plain push — safe + only because the bench cell branch never carries commits of its own, so + it is always a strict ancestor of `onto`).""" + from tests.e2e_smoke.harness import _git + + admin = stack.github.admin_clone + _git(admin, "fetch", "origin", "--prune") + _git(admin, "checkout", "-B", branch, f"origin/{onto}") + _git(admin, "push", "origin", branch) + + +def _create_bench_task( + stack: E2EStack, + project_id: UUID, + dev_slug: str, + fixture: BenchTaskSpec, + team: Team, + parent_task_id: UUID, +) -> UUID: + """Create the real task (TaskService.create), pre-assigned to `dev_slug` + — the "PM pre-assigned this" shape every dev-entry task in production + already uses (e.g. the video engine's authoring tasks) — as a child of + the environment's bench cell (see ``_seed_bench_cell``) so its eventual + PR merges into a real cell branch, not the project's protected default + branch.""" + from roboco.models.task import TaskCreateRequest + from roboco.services.task import EVAL_BENCH_SOURCE, get_task_service + + dev_uuid = _foundation.AGENTS[dev_slug].uuid + holder: dict[str, Any] = {} + + async def _run(session: Any) -> None: + req = TaskCreateRequest( + title=fixture.title, + description=fixture.description, + acceptance_criteria=list(fixture.acceptance_criteria), + team=team, + created_by=dev_uuid, + task_type=fixture.task_type, + nature=fixture.nature, + estimated_complexity=Complexity.LOW, + project_id=project_id, + parent_task_id=parent_task_id, + assigned_to=dev_uuid, + source=EVAL_BENCH_SOURCE, + confirmed_by_human=True, + ) + task = await get_task_service(session).create(req) + holder["id"] = task.id + + stack.run_db(_run) + return cast("UUID", holder["id"]) + + +# --------------------------------------------------------------------------- +# Stage driving — the seam between a real spawn and a scripted stand-in +# --------------------------------------------------------------------------- + + +class StageSpawner(Protocol): + """Advance one task by exactly one role's turn (whatever a single real + container run, or its scripted equivalent, would do): claim + work + + submit, or review + advance. Must not itself loop waiting for further + stages — ``_drive_task_to_terminal`` owns that poll loop.""" + + async def run_stage(self, *, task: dict[str, Any], agent_slug: str) -> None: ... + + +class OrchestratorStageSpawner: + """CUT for this release — do not construct. See the ``NotImplementedError`` + raised below for exactly why, and the module docstring's "Real-spawn + status" section. + + This was meant to be the default, real ``StageSpawner``: drive one turn + via the REAL ``AgentOrchestrator.spawn_agent`` — the exact method the + production dispatcher calls — reusing its own ``_get_prompt_for_agent`` / + ``_task_git_context`` helpers so the prompt and workspace mount are + byte-for-byte what a real dispatch tick would build, then wait for the + container to exit (or the stage timeout). The ``run_stage`` body below is + otherwise correct and is left in place for the follow-up that fixes the + wiring (see ``__init__``) rather than deleted — re-enable it there by + removing the raise. + """ + + _orchestrator: Any + _stage_timeout_seconds: float + + def __init__(self, stage_timeout_seconds: float = 900.0) -> None: + raise NotImplementedError( + "OrchestratorStageSpawner (the real-spawn path) is cut from this " + "release: a spawned container's MCP servers connect via " + "_generate_mcp_config, which resolves the orchestrator URL from " + "PROJECT_HOST_PATH ('http://roboco-orchestrator:8000', the REAL " + "production hostname) or settings.port — NEVER the patched " + "settings.api_url this harness's disposable stack listens on. " + "Combined with _seed_company seeding agents under their REAL " + "production UUIDs, a real spawn here would authenticate as e.g. " + "be-dev-1 against the production orchestrator and could act on " + "real tasks. Fixing this belongs in a dedicated follow-up that " + "makes the spawn env honor the patched stack; until then only " + "the injectable scripted StageSpawner (see " + "tests/e2e_smoke/test_eval_bench.py) is a working path." + ) + + async def run_stage(self, *, task: dict[str, Any], agent_slug: str) -> None: + from roboco.models.runtime import OrchestratorAgentState + + orch = self._orchestrator + # Reuses the orchestrator's own (private) prompt/git-context builders + # so a real bench spawn gets byte-for-byte the same prompt + workspace + # mount a real dispatch tick would build — not a re-derived copy. + prompt = await orch._get_prompt_for_agent(agent_slug, task) + await orch.spawn_agent( + agent_id=agent_slug, + task_id=task["id"], + initial_prompt=prompt, + git_context=orch._task_git_context(task), + spawned_by="eval_bench", + ) + deadline = time.monotonic() + self._stage_timeout_seconds + while time.monotonic() < deadline: + instance = orch.get_instance(agent_slug) + if instance is None or instance.state == OrchestratorAgentState.OFFLINE: + break + await asyncio.sleep(3.0) + await orch.stop_agent( + agent_slug, release_claim=True, exit_reason="eval_bench_stage_end" + ) + + +async def _claim_for_pm( + client: httpx.AsyncClient, api: str, task_id: str, pm_slug: str +) -> None: + """Mirror the real dispatcher's pre-spawn PM claim (``_claim_task_for_ + agent``) — a plain REST call, identical for the real and scripted + stage-spawner paths, so it lives in the shared driving loop rather than + duplicated in both ``StageSpawner`` implementations.""" + with contextlib.suppress(Exception): + await client.post(f"{api}/tasks/{task_id}/claim", json={"agent_id": pm_slug}) + + +async def _drive_task_to_terminal( + stack: E2EStack, + spawner: StageSpawner, + task_id: UUID, + *, + dev_slug: str, + prefix: str, + fixture_timeout_seconds: float, +) -> tuple[dict[str, Any], bool]: + """Poll ``task_id`` to a terminal state, invoking ``spawner.run_stage`` + for whichever role owns the current status, until terminal or the hard + per-fixture timeout. + + Returns ``(final_task_dict, stalled)``. ``stalled`` is True when the loop + gave up (timeout, a human-gated status, or a status with no owning + role) rather than reaching a genuine terminal state. + """ + from roboco.runtime.orchestrator import _system_api_headers + + deadline = time.monotonic() + fixture_timeout_seconds + api = f"{stack.base_url}/api" + # The claim POST below requires an agent identity (X-Agent-ID); this + # driving loop plays the same "trusted internal caller" role the real + # dispatch tick does, so it authenticates the same way — the system + # identity headers the orchestrator's own dispatch client carries. + async with httpx.AsyncClient(timeout=30.0, headers=_system_api_headers()) as client: + task = (await client.get(f"{api}/tasks/{task_id}")).json() + while True: + status = task.get("status") + if status in _TERMINAL_STATUSES: + return task, False + if status in _HUMAN_GATED_STATUSES: + return task, True + role = _STAGE_ROLE.get(status) + if role is None or time.monotonic() >= deadline: + return task, True + agent_slug = ( + dev_slug if role == "developer" else f"{prefix}-{_ROLE_SUFFIX[role]}" + ) + if role == "cell_pm" and not task.get("assigned_to"): + await _claim_for_pm(client, api, str(task_id), agent_slug) + await spawner.run_stage(task=task, agent_slug=agent_slug) + task = (await client.get(f"{api}/tasks/{task_id}")).json() + + +# --------------------------------------------------------------------------- +# Local-model judge — same Ollama-compatible endpoint MemoryDistiller uses +# --------------------------------------------------------------------------- + +_JUDGE_TIMEOUT_SECONDS = 60.0 +_JUDGE_SCORE_RE = re.compile(r"score\s*:\s*([1-5])", re.IGNORECASE) + + +@dataclass +class JudgeVerdict: + score: int | None + rationale: str | None + + +def _build_judge_prompt(fixture: BenchTaskSpec, diff: str, notes: str) -> str: + criteria = "\n".join(f"- {c}" for c in fixture.acceptance_criteria) + return ( + "You are grading a completed engineering task against its checked-in " + "expectation, for an automated agent quality bench. Score 1-5 (5 = " + "fully meets the expectation, 1 = does not meet it at all). Reply in " + "exactly this shape:\n" + "Score: <1-5>\n" + "Rationale: \n\n" + f"Task: {fixture.title}\n" + f"Acceptance criteria:\n{criteria}\n\n" + f"Expected (checked-in): {fixture.expectations}\n\n" + f"Actual diff:\n{diff or '(empty diff)'}\n\n" + f"Actual notes:\n{notes or '(no notes)'}\n" + ) + + +async def _judge_chat(prompt: str) -> str | None: + async with httpx.AsyncClient(timeout=_JUDGE_TIMEOUT_SECONDS) as client: + resp = await client.post( + f"{settings.local_llm_base_url}/chat/completions", + json={ + "model": settings.local_llm_model, + "messages": [{"role": "user", "content": prompt}], + "max_tokens": 200, + "options": {"num_ctx": 8192}, + }, + ) + if not resp.is_success: + return None + data = resp.json() + choices = data.get("choices") or [] + if not choices: + return None + content = choices[0].get("message", {}).get("content") + return content if isinstance(content, str) else None + + +class BenchJudge: + """Scores a fixture's final diff+notes against its expectation, via the + SAME local Ollama-compatible endpoint ``MemoryDistiller`` uses (never a + cloud LLM in the hot path). Best-effort: any failure yields + ``score=None`` — a bench run still produces its deterministic metrics + even with the local model down.""" + + async def score( + self, *, fixture: BenchTaskSpec, diff: str, notes: str + ) -> JudgeVerdict: + try: + content = await _judge_chat(_build_judge_prompt(fixture, diff, notes)) + except Exception as exc: + logger.warning("BenchJudge failed (best-effort)", error=str(exc)) + return JudgeVerdict(score=None, rationale=f"judge unavailable: {exc}") + if not content: + return JudgeVerdict(score=None, rationale="judge returned no content") + match = _JUDGE_SCORE_RE.search(content) + score = int(match.group(1)) if match else None + return JudgeVerdict(score=score, rationale=content.strip()) + + +# --------------------------------------------------------------------------- +# Scoring — pure dataclasses + aggregate math (unit-testable with no DB/IO) +# --------------------------------------------------------------------------- + + +@dataclass +class DeterministicMetrics: + final_status: str + stalled: bool + revision_count: int + cycle_time_seconds: float + tokens_input: int + tokens_output: int + tokens_cache_read: int + tokens_cache_write: int + estimated_cost_usd: float + + @property + def total_tokens(self) -> int: + return ( + self.tokens_input + + self.tokens_output + + self.tokens_cache_read + + self.tokens_cache_write + ) + + +@dataclass +class FixtureResult: + fixture_key: str + metrics: DeterministicMetrics + judge: JudgeVerdict + + @property + def passed(self) -> bool: + return self.metrics.final_status == "completed" and not self.metrics.stalled + + +@dataclass +class CohortResult: + role_slug: str + cohort_name: str + fixtures: list[FixtureResult] + + @property + def pass_rate(self) -> float: + if not self.fixtures: + return 0.0 + return sum(1 for f in self.fixtures if f.passed) / len(self.fixtures) + + @property + def total_cost_usd(self) -> float: + return sum(f.metrics.estimated_cost_usd for f in self.fixtures) + + @property + def total_tokens(self) -> int: + return sum(f.metrics.total_tokens for f in self.fixtures) + + @property + def mean_cycle_time_seconds(self) -> float: + if not self.fixtures: + return 0.0 + return statistics.fmean(f.metrics.cycle_time_seconds for f in self.fixtures) + + @property + def mean_judge_score(self) -> float | None: + scores = [f.judge.score for f in self.fixtures if f.judge.score is not None] + return statistics.fmean(scores) if scores else None + + def as_dict(self) -> dict[str, Any]: + # Judge fields are nested under their own "judge" object (both here + # and per-fixture below) and stamped non_deterministic=True — a local- + # model score is not a repeatable metric like the sibling deterministic + # ones, and a naive diff between two cohort JSONs must not mistake + # judge noise for a real regression. + return { + "role_slug": self.role_slug, + "cohort_name": self.cohort_name, + "aggregate": { + "fixture_count": len(self.fixtures), + "pass_rate": self.pass_rate, + "total_cost_usd": round(self.total_cost_usd, 4), + "total_tokens": self.total_tokens, + "mean_cycle_time_seconds": round(self.mean_cycle_time_seconds, 1), + }, + "judge": { + "mean_score": self.mean_judge_score, + "non_deterministic": True, + }, + "fixtures": [ + { + "fixture_key": f.fixture_key, + "final_status": f.metrics.final_status, + "stalled": f.metrics.stalled, + "passed": f.passed, + "revision_count": f.metrics.revision_count, + "cycle_time_seconds": round(f.metrics.cycle_time_seconds, 1), + "tokens_input": f.metrics.tokens_input, + "tokens_output": f.metrics.tokens_output, + "tokens_cache_read": f.metrics.tokens_cache_read, + "tokens_cache_write": f.metrics.tokens_cache_write, + "estimated_cost_usd": round(f.metrics.estimated_cost_usd, 4), + "judge": { + "score": f.judge.score, + "rationale": f.judge.rationale, + "non_deterministic": True, + }, + } + for f in self.fixtures + ], + } + + +def _deterministic_metrics( + stack: E2EStack, + task_id: UUID, + started_at: datetime, + stalled: bool, + final_status: str, +) -> DeterministicMetrics: + """Read the final task row + the agent_spawn_sessions rows this task's + stages accumulated, joined by task_id (the same join the CLAUDE.md + "Delivery observability" rework-cost metric already uses).""" + from sqlalchemy import func, select + + from roboco.db.tables import AgentSpawnSessionTable, TaskTable + + async def _run(session: Any) -> dict[str, Any]: + task_row = ( + await session.execute(select(TaskTable).where(TaskTable.id == task_id)) + ).scalar_one() + agg = ( + await session.execute( + select( + func.coalesce(func.sum(AgentSpawnSessionTable.tokens_input), 0), + func.coalesce(func.sum(AgentSpawnSessionTable.tokens_output), 0), + func.coalesce( + func.sum(AgentSpawnSessionTable.tokens_cache_read), 0 + ), + func.coalesce( + func.sum(AgentSpawnSessionTable.tokens_cache_write), 0 + ), + func.coalesce( + func.sum(AgentSpawnSessionTable.estimated_cost_usd), 0.0 + ), + ).where(AgentSpawnSessionTable.task_id == str(task_id)) + ) + ).one() + ended_at = task_row.completed_at or task_row.updated_at or datetime.now(UTC) + return { + "revision_count": task_row.revision_count, + "tokens_input": int(agg[0]), + "tokens_output": int(agg[1]), + "tokens_cache_read": int(agg[2]), + "tokens_cache_write": int(agg[3]), + "estimated_cost_usd": float(agg[4]), + "ended_at": ended_at, + } + + row = stack.run_db(_run) + cycle_time = (row["ended_at"] - started_at).total_seconds() + return DeterministicMetrics( + final_status=final_status, + stalled=stalled, + revision_count=row["revision_count"], + cycle_time_seconds=max(cycle_time, 0.0), + tokens_input=row["tokens_input"], + tokens_output=row["tokens_output"], + tokens_cache_read=row["tokens_cache_read"], + tokens_cache_write=row["tokens_cache_write"], + estimated_cost_usd=row["estimated_cost_usd"], + ) + + +def _task_diff(stack: E2EStack, base_branch: str, branch_name: str | None) -> str: + """Diff the task's branch against its REAL base — the bench cell branch + it was cut from (see ``_seed_bench_cell``), not the project default.""" + if not branch_name: + return "" + from tests.e2e_smoke.harness import _git + + admin = stack.github.admin_clone + try: + _git(admin, "fetch", "origin", "--prune") + return _git(admin, "diff", f"origin/{base_branch}...origin/{branch_name}") + except subprocess.CalledProcessError: + return "" + + +def _collected_notes(stack: E2EStack, task_id: UUID) -> str: + from sqlalchemy import select + + from roboco.db.tables import TaskTable + + async def _run(session: Any) -> str: + row = ( + await session.execute(select(TaskTable).where(TaskTable.id == task_id)) + ).scalar_one() + sections = ( + ("dev", row.dev_notes), + ("qa", row.qa_notes), + ("doc", row.doc_notes), + ("pm", row.pm_notes), + ) + return "\n\n".join( + f"[{label}_notes]\n{text}" for label, text in sections if text + ) + + return cast("str", stack.run_db(_run)) + + +def _print_table(cohort: CohortResult) -> None: + # "judge*" / the trailing footnote mirror as_dict()'s nested + # judge.non_deterministic=True — a local-model score is not a repeatable + # metric like its deterministic neighbors in this row. + header = ( + f"{'fixture':<28} {'status':<12} {'stalled':<8} {'rev':<4} " + f"{'cycle(s)':<9} {'tokens':<9} {'cost($)':<9} {'judge*':<6}" + ) + print(f"\nEval bench — role={cohort.role_slug} cohort={cohort.cohort_name}\n") + print(header) + print("-" * len(header)) + for f in cohort.fixtures: + m = f.metrics + judge = str(f.judge.score) if f.judge.score is not None else "-" + print( + f"{f.fixture_key:<28} {m.final_status:<12} {m.stalled!s:<8} " + f"{m.revision_count:<4} {m.cycle_time_seconds:<9.1f} " + f"{m.total_tokens:<9} {m.estimated_cost_usd:<9.4f} {judge:<6}" + ) + print("-" * len(header)) + mean_judge = cohort.mean_judge_score + print( + f"pass_rate={cohort.pass_rate:.2f} " + f"mean_cycle_s={cohort.mean_cycle_time_seconds:.1f} " + f"total_tokens={cohort.total_tokens} " + f"total_cost=${cohort.total_cost_usd:.4f} " + f"mean_judge*={mean_judge if mean_judge is not None else '-'}" + ) + print("*judge score/mean_judge* are local-model, non-deterministic — not a metric") + + +# --------------------------------------------------------------------------- +# EvalRunner +# --------------------------------------------------------------------------- + + +class EvalRunner: + """Runs one cohort (a labeled (role, model/provider config) run) through + every developer-role golden-task fixture and prints + returns the + scored result.""" + + def __init__( + self, + *, + make_spawner: Callable[[E2EStack], StageSpawner] | None = None, + judge: BenchJudge | None = None, + stage_timeout_seconds: float = 900.0, + fixture_timeout_seconds: float = 3600.0, + ) -> None: + self._make_spawner = make_spawner or ( + lambda _stack: OrchestratorStageSpawner( + stage_timeout_seconds=stage_timeout_seconds + ) + ) + self._judge = judge or BenchJudge() + self._fixture_timeout_seconds = fixture_timeout_seconds + + def run_cohort( + self, + role_slug: str, + cohort_name: str, + *, + fixtures: Sequence[BenchTaskSpec] | None = None, + json_out: Path | None = None, + ) -> CohortResult: + role = get_agent_role(role_slug) + if role != "developer": + raise ValueError( + f"eval bench only scores developer-role agents right now " + f"(got {role_slug!r} -> role={role!r}); see the module " + "docstring's scope-cut note" + ) + team_str = get_agent_team(role_slug) + if team_str not in _TEAM_PREFIX: + raise ValueError(f"{role_slug!r} has no known cell team ({team_str!r})") + prefix = _TEAM_PREFIX[team_str] + chosen = [f for f in (fixtures or FIXTURES) if f.target_role == "developer"] + if not chosen: + raise ValueError("no fixtures matched target_role='developer'") + + results: list[FixtureResult] = [] + with _bench_environment(role_slug) as env: + for fixture in chosen: + results.append(self._run_fixture(env, prefix, role_slug, fixture)) + + cohort = CohortResult( + role_slug=role_slug, cohort_name=cohort_name, fixtures=results + ) + _print_table(cohort) + if json_out is not None: + json_out.write_text(json.dumps(cohort.as_dict(), indent=2)) + return cohort + + def _run_fixture( + self, + env: BenchEnvironment, + prefix: str, + dev_slug: str, + fixture: BenchTaskSpec, + ) -> FixtureResult: + _seed_fixture_repo(env.stack, fixture) + # The bench cell branch is cut once per environment and never given + # commits of its own — fast-forward it to master's just-updated tip + # so THIS fixture's newly-pushed bench// files are actually + # present when the leaf's own branch is cut from it below. + _fast_forward_branch(env.stack, env.cell_branch, onto="master") + started_at = datetime.now(UTC) + task_id = _create_bench_task( + env.stack, env.project_id, dev_slug, fixture, env.team, env.cell_id + ) + + spawner = self._make_spawner(env.stack) + final_task, stalled = asyncio.run( + _drive_task_to_terminal( + env.stack, + spawner, + task_id, + dev_slug=dev_slug, + prefix=prefix, + fixture_timeout_seconds=self._fixture_timeout_seconds, + ) + ) + metrics = _deterministic_metrics( + env.stack, task_id, started_at, stalled, final_task.get("status", "unknown") + ) + diff = _task_diff(env.stack, env.cell_branch, final_task.get("branch_name")) + notes = _collected_notes(env.stack, task_id) + judge = asyncio.run(self._judge.score(fixture=fixture, diff=diff, notes=notes)) + return FixtureResult(fixture_key=fixture.key, metrics=metrics, judge=judge) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 97a96597..29c95e6f 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -14,6 +14,7 @@ The orchestrator is the BRAIN of the system: import asyncio import contextlib +import hashlib import json import os import shutil @@ -6311,6 +6312,7 @@ class AgentOrchestrator: if instance and instance.config: model = instance.config.model or "unknown" usage_session_id = instance.usage_session_id if instance else None + doctrine_version = self._doctrine_version_for_instance(instance) cost = calculate_cost( model=model, @@ -6358,6 +6360,7 @@ class AgentOrchestrator: tool_calls=tool_calls, exit_reason=exit_reason, estimated_cost_usd=cost, + doctrine_version=doctrine_version, ) ) await db.commit() @@ -6368,6 +6371,7 @@ class AgentOrchestrator: tokens_input=tokens_input, tokens_output=tokens_output, estimated_cost_usd=cost, + doctrine_version=doctrine_version, ) except Exception as exc: logger.warning( @@ -6376,6 +6380,33 @@ class AgentOrchestrator: error=str(exc), ) + @staticmethod + def _doctrine_version_for_instance(instance: AgentInstance | None) -> str | None: + """Short hash of the composed system prompt this spawn ran with. + + Reads the SAME file ``_generate_composed_prompt`` wrote at spawn + preparation (``config.blueprint_path``) — nothing on the spawn/stop + path deletes it, so it is still the exact prompt text this agent ran + with. Every provider gets one (the blueprint is composed and written + unconditionally in ``_prepare_agent_spawn``, before provider/route + resolution) — GROK agents carry a real blueprint file too, same as + Claude. Best-effort: a missing/unreadable file (a provider-parked stub + instance that never actually spawned — ``blueprint_path=Path()`` — an + evicted temp dir, ...) is NULL, never a finalize failure — the eval + harness treats an unstamped session as "doctrine unknown", not an + error. + """ + if instance is None or instance.config is None: + return None + blueprint_path = instance.config.blueprint_path + if not blueprint_path: + return None + try: + content = blueprint_path.read_text() + except OSError: + return None + return hashlib.sha256(content.encode("utf-8")).hexdigest()[:16] + @staticmethod async def _fetch_agent_tokens( client: httpx.AsyncClient, agent_id: str diff --git a/roboco/services/task.py b/roboco/services/task.py index 7fddd3fa..8007e727 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -706,6 +706,18 @@ ROADMAP_ITEM_SOURCE = "roadmap" # the Main PM. The board routing is the start gate; no held-source skip. VAULT_NOTE_SOURCE = "vault_note" +# Source tag for a golden-task fixture the offline eval bench +# (``roboco/eval/``) replays through the real delivery lifecycle to score a +# (role, model/provider config) cohort. Deliberately absent from every +# held-source / non-dev-dispatch set above: an eval_bench task is a normal, +# pre-assigned dev leaf task and must dispatch exactly like one (real spawn, +# real QA, real docs, real cell-PM merge) — every OTHER engine (self-heal, +# ci-watch, dep-update, docs-sync, release-manager, X, video, roadmap, vault- +# intake) only ever queries/dedupes by ITS OWN source constant above, so an +# eval_bench task is invisible to all of them by construction, not by an +# explicit exemption. +EVAL_BENCH_SOURCE = "eval_bench" + def extract_self_heal_fingerprint(task: Any) -> str | None: """The self-heal dedupe fingerprint from a task's markers, or None. diff --git a/tests/e2e_smoke/harness.py b/tests/e2e_smoke/harness.py index b224702f..c299fd64 100644 --- a/tests/e2e_smoke/harness.py +++ b/tests/e2e_smoke/harness.py @@ -45,8 +45,20 @@ if TYPE_CHECKING: from collections.abc import Iterator from pathlib import Path from types import ModuleType + from typing import Protocol from uuid import UUID + class TmpPathFactory(Protocol): + """Structural stand-in for the one ``pytest.TempPathFactory`` method + ``build_e2e_stack`` uses. pytest's real fixture value already + satisfies this shape, so it needs no adapter — but it lets + ``roboco/eval/runner.py`` (an offline CLI, not a pytest session) drive + this same stack-building machinery with a plain temp-dir factory + instead of constructing a real ``pytest.Config``.""" + + def mktemp(self, basename: str, numbered: bool = True) -> Path: ... + + _OWNER = "e2e-smoke" _REPO = "proj" @@ -360,9 +372,15 @@ def _build_app(gh: _FakeGitHub) -> FastAPI: def build_e2e_stack( - _test_database_url: str, tmp_path_factory: pytest.TempPathFactory + _test_database_url: str, tmp_path_factory: TmpPathFactory ) -> Iterator[E2EStack]: - """Generator behind the ``e2e_stack`` fixture (defined in conftest).""" + """Generator behind the ``e2e_stack`` fixture (defined in conftest). + + ``tmp_path_factory`` only needs ``.mktemp()`` (see ``TmpPathFactory`` + above) — pytest's real fixture satisfies it structurally, and + ``roboco/eval/runner.py`` drives this same function with a plain + non-pytest factory to reuse this stack outside a test session. + """ from roboco.config import settings from roboco.db import base as db_base diff --git a/tests/e2e_smoke/test_eval_bench.py b/tests/e2e_smoke/test_eval_bench.py new file mode 100644 index 00000000..ab4bc4fa --- /dev/null +++ b/tests/e2e_smoke/test_eval_bench.py @@ -0,0 +1,204 @@ +"""Integration test for the eval bench's own orchestration/scoring plumbing. + +The real ``StageSpawner`` (``OrchestratorStageSpawner``) drives a REAL agent +container via ``AgentOrchestrator.spawn_agent`` and needs a Docker daemon + +built agent images — it cannot run here (see ``roboco/eval/runner.py``'s +module docstring). This test substitutes a scripted stand-in that drives the +SAME real MCP flow/do tool functions ``tests.e2e_smoke.harness.ScriptedAgent`` +uses (via the existing ``dev_arc`` / ``qa_arc`` / ``doc_arc`` helpers, plus a +PM ``complete`` call) so it proves the runner's OWN code — its throwaway-DB + +disposable-project setup, its status-driven stage loop, its PM pre-claim, +its deterministic scoring, its JSON/table output — without touching Docker. + +Runs the smallest fixture (a single-file bug fix) end to end: PENDING -> +awaiting_qa -> awaiting_documentation -> awaiting_pm_review -> completed. + +Gating: like every other module here, this is skipped unless +``ROBOCO_E2E_SMOKE=1`` (see ``tests/e2e_smoke/conftest.py``'s +``pytest_collection_modifyitems``) — it needs the real test Postgres, which +``EvalRunner`` provisions its own throwaway copy of (see +``roboco/eval/runner.py``'s ``_scratch_database``), independent of this +package's shared session-scoped ``e2e_stack`` fixture. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Any +from uuid import UUID + +from roboco.config import settings +from roboco.eval.fixtures import FIXTURES +from roboco.eval.runner import BenchJudge, EvalRunner, JudgeVerdict, _bench_environment +from tests.e2e_smoke.arcs import Company, dev_arc, doc_arc, qa_arc +from tests.e2e_smoke.harness import ScriptedAgent + +if TYPE_CHECKING: + import pytest + from roboco.eval.fixtures import BenchTaskSpec + from tests.e2e_smoke.harness import E2EStack + +_FIXTURE_KEY = "bugfix-off-by-one" +_FIXED_FIX = ( + "def paginate(items, page, size):\n" + " start = (page - 1) * size\n" + " end = page * size\n" + " return items[start:end]\n" +) + + +def _fixture() -> BenchTaskSpec: + for f in FIXTURES: + if f.key == _FIXTURE_KEY: + return f + raise AssertionError(f"{_FIXTURE_KEY!r} fixture not found in FIXTURES") + + +def _stub_company() -> Company: + """A ``Company`` carrying the FIXED uuids ``EvalRunner``'s own company + seeding uses (not fresh random ones, unlike ``arcs.seed_company``) — the + "be-*" slugs are hardcoded inside ``dev_arc`` / ``qa_arc`` / ``doc_arc`` + themselves, so this only needs to supply the matching ids.""" + from roboco.foundation import identity as _foundation + + company = Company() + company.dev_id = _foundation.AGENTS["be-dev-1"].uuid + company.qa_id = _foundation.AGENTS["be-qa"].uuid + company.doc_id = _foundation.AGENTS["be-doc"].uuid + company.cell_pm_id = _foundation.AGENTS["be-pm"].uuid + return company + + +class _ScriptedBenchSpawner: + """Test-only ``StageSpawner``: applies the KNOWN correct fix via the real + MCP flow/do tool functions, standing in for a real container spawn. + + ``dev_arc`` / ``qa_arc`` / ``doc_arc`` (and ``ScriptedAgent`` itself) call + ``E2EStack.run_db``, which runs its own ``asyncio.run()`` per call — fine + from a plain sync pytest test, but ``run_stage`` is awaited from inside + ``_drive_task_to_terminal``'s own event loop, where a nested + ``asyncio.run()`` raises. Running the scripted turn on a worker thread + (``asyncio.to_thread``) gives it a thread with no running loop, exactly + like the sync test functions those helpers were written for. + """ + + def __init__(self, stack: E2EStack) -> None: + self._stack = stack + self._company = _stub_company() + + async def run_stage(self, *, task: dict[str, Any], agent_slug: str) -> None: + await asyncio.to_thread(self._run_stage_sync, task, agent_slug) + + def _run_stage_sync(self, task: dict[str, Any], agent_slug: str) -> None: + from roboco.agents_config import get_agent_role + + role = get_agent_role(agent_slug) + task_id = UUID(task["id"]) + if role == "developer": + dev_arc( + self._stack, + self._company, + task["project_slug"], + task_id, + work=(f"bench/{_FIXTURE_KEY}/paginate.py", _FIXED_FIX), + ) + elif role == "qa": + qa_arc(self._stack, self._company, task_id) + elif role == "documenter": + doc_arc( + self._stack, + self._company, + task_id, + filename=f"bench/{_FIXTURE_KEY}/paginate.py", + ) + elif role == "cell_pm": + pm = ScriptedAgent( + self._stack, self._company.cell_pm_id, agent_slug, "cell_pm" + ) + pm.flow( + "complete", + task_id=str(task_id), + notes="Scripted bench completion: QA passed, docs complete.", + ) + else: + raise AssertionError(f"unexpected role for the scripted bench: {role!r}") + + +_EXPECTED_JUDGE_SCORE = 5 + + +class _FakeJudge(BenchJudge): + """Deterministic stand-in for the local-model judge — no network.""" + + async def score( + self, *, fixture: BenchTaskSpec, diff: str, notes: str + ) -> JudgeVerdict: + return JudgeVerdict( + score=_EXPECTED_JUDGE_SCORE, rationale="scripted test: assumed correct" + ) + + +def test_eval_runner_drives_a_fixture_to_completion_with_a_scripted_spawn() -> None: + runner = EvalRunner( + make_spawner=_ScriptedBenchSpawner, + judge=_FakeJudge(), + fixture_timeout_seconds=60.0, + ) + + cohort = runner.run_cohort( + "be-dev-1", "scripted-test", fixtures=[_fixture()], json_out=None + ) + + assert cohort.role_slug == "be-dev-1" + assert len(cohort.fixtures) == 1 + result = cohort.fixtures[0] + assert result.fixture_key == _FIXTURE_KEY + assert result.metrics.final_status == "completed" + assert result.metrics.stalled is False + assert result.passed is True + assert result.metrics.revision_count == 0 + assert result.judge.score == _EXPECTED_JUDGE_SCORE + assert cohort.pass_rate == 1.0 + # No real container spawned, so no agent_spawn_sessions rows accrued for + # this task — the scripted stand-in proves the runner's DB/polling/ + # scoring plumbing, not token/cost accounting (that needs a real spawn; + # see the module docstring). + assert result.metrics.total_tokens == 0 + assert result.metrics.estimated_cost_usd == 0.0 + + +def test_bench_environment_disables_vault_writes_even_when_ambient_flags_are_armed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A bench run must never write into the operator's REAL Obsidian vault. + Simulates the compose-default posture (every vault flag armed True) and + asserts `_bench_environment` forces them all off for its duration, then + restores the prior values on exit — the exact leak an adversarial review + flagged (TaskService.create / JournalService / A2AService all gate on + obsidian_vault_enabled first, so patching it is the load-bearing part; + the three sub-flags are patched too for defense-in-depth).""" + monkeypatch.setattr(settings, "obsidian_vault_enabled", True) + monkeypatch.setattr(settings, "vault_intake_enabled", True) + monkeypatch.setattr(settings, "vault_kb_enabled", True) + monkeypatch.setattr(settings, "vault_report_enabled", True) + armed = ( + settings.obsidian_vault_enabled, + settings.vault_intake_enabled, + settings.vault_kb_enabled, + settings.vault_report_enabled, + ) + + with _bench_environment("be-dev-1"): + assert settings.obsidian_vault_enabled is False + assert settings.vault_intake_enabled is False + assert settings.vault_kb_enabled is False + assert settings.vault_report_enabled is False + + # Restored to the (simulated ambient) armed state once the bench exits. + restored = ( + settings.obsidian_vault_enabled, + settings.vault_intake_enabled, + settings.vault_kb_enabled, + settings.vault_report_enabled, + ) + assert restored == armed == (True, True, True, True) diff --git a/tests/unit/eval/test_fixtures.py b/tests/unit/eval/test_fixtures.py new file mode 100644 index 00000000..d967a10a --- /dev/null +++ b/tests/unit/eval/test_fixtures.py @@ -0,0 +1,74 @@ +"""Schema checks for the golden-task fixtures (roboco/eval/fixtures.py). + +Nothing here touches a DB or the network — these are pure sanity checks on +the static FIXTURES tuple so a malformed fixture (a duplicate key, a fixture +file that escapes its own bench// namespace and could collide with +another fixture's repo state, an empty brief) is caught before it ever +reaches the runner. +""" + +from __future__ import annotations + +import dataclasses +from typing import Any, cast + +import pytest +from roboco.eval.fixtures import FIXTURES, BenchTaskSpec + +_MIN_FIXTURES = 5 +_MAX_FIXTURES = 8 + + +def test_fixture_keys_are_unique() -> None: + keys = [f.key for f in FIXTURES] + assert len(keys) == len(set(keys)), f"duplicate fixture keys: {keys}" + + +def test_at_least_five_fixtures() -> None: + # The task calls for 5-8 canonical fixtures. + assert _MIN_FIXTURES <= len(FIXTURES) <= _MAX_FIXTURES, len(FIXTURES) + + +def test_every_fixture_has_a_non_empty_brief() -> None: + for f in FIXTURES: + assert f.title.strip(), f.key + assert f.description.strip(), f.key + assert f.acceptance_criteria, f"{f.key} has no acceptance criteria" + assert all(c.strip() for c in f.acceptance_criteria), f.key + assert f.expectations.strip(), f"{f.key} has no judge expectations note" + + +def test_repo_files_are_namespaced_under_bench_key() -> None: + """Every fixture's seeded file lives under bench// so + sequential fixtures sharing one project's git history never collide.""" + for f in FIXTURES: + assert f.repo_files, f"{f.key} seeds no repo files" + prefix = f"bench/{f.key}/" + for rel_path, content in f.repo_files: + assert rel_path.startswith(prefix), ( + f"{f.key}: {rel_path!r} escapes its own {prefix!r} namespace" + ) + assert ".." not in rel_path, f"{f.key}: {rel_path!r} looks like a traversal" + assert content, f"{f.key}: {rel_path!r} has empty content" + + +def test_repo_file_paths_within_a_fixture_are_unique() -> None: + for f in FIXTURES: + paths = [rel_path for rel_path, _content in f.repo_files] + assert len(paths) == len(set(paths)), f"{f.key}: duplicate paths {paths}" + + +def test_target_role_is_developer_for_every_fixture() -> None: + """Matches EvalRunner.run_cohort's current scope cut (see runner.py's + module docstring) — every fixture must be runnable by the one role the + bench supports today.""" + for f in FIXTURES: + assert f.target_role == "developer", f.key + + +def test_bench_task_spec_is_frozen() -> None: + spec = FIXTURES[0] + assert isinstance(spec, BenchTaskSpec) + mutable_view = cast("Any", spec) + with pytest.raises(dataclasses.FrozenInstanceError): + mutable_view.title = "mutated" diff --git a/tests/unit/eval/test_scoring.py b/tests/unit/eval/test_scoring.py new file mode 100644 index 00000000..0a13f4e7 --- /dev/null +++ b/tests/unit/eval/test_scoring.py @@ -0,0 +1,221 @@ +"""Unit tests for the eval bench's scorer math (roboco/eval/runner.py). + +Pure dataclass/aggregate-property tests — no DB, no network, no asyncio. +`_build_judge_prompt` and `BenchJudge`'s score-parsing regex are covered too +since both are pure string logic with no I/O. +""" + +from __future__ import annotations + +import pytest +from roboco.eval.fixtures import FIXTURES +from roboco.eval.runner import ( + _JUDGE_SCORE_RE, + CohortResult, + DeterministicMetrics, + FixtureResult, + JudgeVerdict, + OrchestratorStageSpawner, + _build_judge_prompt, +) + +_EXPECTED_TOTAL_TOKENS = 180 +_HALF_PASS_RATE = 0.5 +_COHORT_TOTAL_TOKENS = 600 +_COHORT_MEAN_CYCLE_SECONDS = 20.0 +_COHORT_MEAN_JUDGE_SCORE = 5.0 +_PASSING_JUDGE_SCORE = 4 + + +def _metrics( + *, + final_status: str = "completed", + stalled: bool = False, + cycle_time_seconds: float = 10.0, + tokens_input: int = 100, + tokens_output: int = 50, + tokens_cache_read: int = 0, + tokens_cache_write: int = 0, + estimated_cost_usd: float = 0.01, +) -> DeterministicMetrics: + return DeterministicMetrics( + final_status=final_status, + stalled=stalled, + revision_count=0, + cycle_time_seconds=cycle_time_seconds, + tokens_input=tokens_input, + tokens_output=tokens_output, + tokens_cache_read=tokens_cache_read, + tokens_cache_write=tokens_cache_write, + estimated_cost_usd=estimated_cost_usd, + ) + + +def test_deterministic_metrics_total_tokens_sums_all_four_buckets() -> None: + m = _metrics( + tokens_input=100, tokens_output=50, tokens_cache_read=25, tokens_cache_write=5 + ) + assert m.total_tokens == _EXPECTED_TOTAL_TOKENS + + +def test_fixture_result_passed_requires_completed_and_not_stalled() -> None: + passed = FixtureResult( + fixture_key="a", + metrics=_metrics(final_status="completed", stalled=False), + judge=JudgeVerdict(score=None, rationale=None), + ) + assert passed.passed is True + + cancelled = FixtureResult( + fixture_key="b", + metrics=_metrics(final_status="cancelled", stalled=False), + judge=JudgeVerdict(score=None, rationale=None), + ) + assert cancelled.passed is False + + # A stall that happens to leave the row at "completed" is still not a + # pass — `stalled` overrides the status. + stalled_completed = FixtureResult( + fixture_key="c", + metrics=_metrics(final_status="completed", stalled=True), + judge=JudgeVerdict(score=None, rationale=None), + ) + assert stalled_completed.passed is False + + +def _sample_cohort() -> CohortResult: + fixtures = [ + FixtureResult( + fixture_key="a", + metrics=_metrics( + final_status="completed", + cycle_time_seconds=10.0, + estimated_cost_usd=0.10, + tokens_input=100, + tokens_output=100, + ), + judge=JudgeVerdict(score=5, rationale="great"), + ), + FixtureResult( + fixture_key="b", + metrics=_metrics( + final_status="needs_revision", + stalled=True, + cycle_time_seconds=30.0, + estimated_cost_usd=0.20, + tokens_input=200, + tokens_output=200, + ), + judge=JudgeVerdict(score=None, rationale="judge unavailable"), + ), + ] + return CohortResult(role_slug="be-dev-1", cohort_name="baseline", fixtures=fixtures) + + +def test_cohort_pass_rate_and_totals() -> None: + cohort = _sample_cohort() + + assert cohort.pass_rate == _HALF_PASS_RATE + assert cohort.total_cost_usd == pytest.approx(0.3) + assert cohort.total_tokens == _COHORT_TOTAL_TOKENS + assert cohort.mean_cycle_time_seconds == _COHORT_MEAN_CYCLE_SECONDS + # Only fixture "a" has a judge score; "b"'s None is excluded from the mean. + assert cohort.mean_judge_score == _COHORT_MEAN_JUDGE_SCORE + + +def test_cohort_mean_judge_score_is_none_when_no_fixture_was_scored() -> None: + fixtures = [ + FixtureResult( + fixture_key="a", + metrics=_metrics(), + judge=JudgeVerdict(score=None, rationale="judge unavailable"), + ) + ] + cohort = CohortResult(role_slug="be-dev-1", cohort_name="x", fixtures=fixtures) + assert cohort.mean_judge_score is None + + +def test_cohort_with_no_fixtures_is_a_zero_result_not_a_crash() -> None: + cohort = CohortResult(role_slug="be-dev-1", cohort_name="x", fixtures=[]) + assert cohort.pass_rate == 0.0 + assert cohort.total_cost_usd == 0.0 + assert cohort.total_tokens == 0 + assert cohort.mean_cycle_time_seconds == 0.0 + assert cohort.mean_judge_score is None + + +def test_cohort_as_dict_round_trips_every_fixture() -> None: + fixtures = [ + FixtureResult( + fixture_key="a", + metrics=_metrics(), + judge=JudgeVerdict(_PASSING_JUDGE_SCORE, "solid"), + ), + ] + cohort = CohortResult(role_slug="be-dev-1", cohort_name="x", fixtures=fixtures) + payload = cohort.as_dict() + + assert payload["role_slug"] == "be-dev-1" + assert payload["cohort_name"] == "x" + assert payload["aggregate"]["fixture_count"] == 1 + assert payload["aggregate"]["pass_rate"] == 1.0 + # Judge fields live under their own nested, explicitly-marked object — + # never flat beside deterministic metrics — so a naive diff can't read + # judge noise as a regression. + assert "mean_judge_score" not in payload["aggregate"] + assert payload["judge"] == { + "mean_score": _PASSING_JUDGE_SCORE, + "non_deterministic": True, + } + assert len(payload["fixtures"]) == 1 + assert payload["fixtures"][0]["fixture_key"] == "a" + assert "judge_score" not in payload["fixtures"][0] + assert payload["fixtures"][0]["judge"] == { + "score": _PASSING_JUDGE_SCORE, + "rationale": "solid", + "non_deterministic": True, + } + + +def test_judge_score_regex_parses_the_required_reply_shape() -> None: + reply = "Score: 4\nRationale: matches the expectation closely.\n" + match = _JUDGE_SCORE_RE.search(reply) + assert match is not None + assert int(match.group(1)) == _PASSING_JUDGE_SCORE + + +def test_judge_score_regex_is_case_insensitive_and_tolerates_spacing() -> None: + assert _JUDGE_SCORE_RE.search("score:5") is not None + assert _JUDGE_SCORE_RE.search("SCORE : 3") is not None + + +def test_judge_score_regex_rejects_out_of_range_scores() -> None: + assert _JUDGE_SCORE_RE.search("Score: 0") is None + assert _JUDGE_SCORE_RE.search("Score: 6") is None + + +def test_build_judge_prompt_includes_the_expectation_and_acceptance_criteria() -> None: + fixture = FIXTURES[0] + prompt = _build_judge_prompt(fixture, diff="+ fixed line", notes="dev notes here") + + assert fixture.title in prompt + assert fixture.expectations in prompt + for criterion in fixture.acceptance_criteria: + assert criterion in prompt + assert "+ fixed line" in prompt + assert "dev notes here" in prompt + + +def test_build_judge_prompt_handles_empty_diff_and_notes() -> None: + fixture = FIXTURES[0] + prompt = _build_judge_prompt(fixture, diff="", notes="") + assert "(empty diff)" in prompt + assert "(no notes)" in prompt + + +def test_orchestrator_stage_spawner_is_cut_and_refuses_to_construct() -> None: + """The real-spawn path is deliberately disabled this release (its MCP + wiring would authenticate against the REAL production orchestrator) — + this is the one runnable check that the cut stays in place.""" + with pytest.raises(NotImplementedError, match="cut from this release"): + OrchestratorStageSpawner()