#!/usr/bin/env python3 """Repeatable profile evaluation, metrics extraction, and run comparison.""" from __future__ import annotations import contextlib import copy import datetime as dt import json import math import re import shutil import subprocess import uuid from collections import Counter from collections.abc import Callable, Mapping, Sequence from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any from mmo_gateway import route_availability from mmo_profiles import derive_coordination_capacities from mmo_runtime import cancel_session, iter_jobs, load_session, run_root_exec, stop_session from mmo_snapshot import compile_profile, compile_resolved_profile from mmo_util import ( atomic_write_json, config_root, event_usage, filtered_environment, install_root, read_json, read_json_object, read_toml, safe_name, sha256_bytes, stable_hash, state_root, strict_json_loads, terminate_process_group, utc_now, validate_id, ) from mmo_version import MMO_SCHEMA_VERSION SUITE_FIELDS = { "schema_version", "id", "name", "description", "profile", "fixture", "development_trials", "release_trials", "promotion", "variants", "tasks", } PROMOTION_FIELDS = { "primary_metric", "primary_baseline", "direction", "strongest_success_tolerance", "minimum_relative_improvement", "minimum_absolute_improvement", "worker_minimum_success_contribution", "worker_minimum_metric_contribution", "no_regression_higher_metrics", "no_regression_lower_metrics", "require_complete_api_cost", "scarce_model_keys", } TASK_FIELDS = { "id", "description", "sandbox", "wall_timeout_seconds", "validation_timeout_seconds", "prompt", "images", "difficulty", "negative_control", "route_faults", "outcome_assertions", "orchestration_assertions", } ROUTE_FAULT_TYPES = {"credential_loss", "rate_limit", "timeout"} OUTCOME_ASSERTION_FIELDS = { "expected_patterns", "forbidden_patterns", "validation_commands", } ORCHESTRATION_ASSERTION_FIELDS = { "required_agents", "forbidden_agents", "min_peak_mcp_workers", "max_jobs", "min_result_acceptance_rate", "max_contract_failures", "max_observed_mcp_wait_ratio", } VARIANT_FIELDS = { "id", "purpose", "profile", "bindings", "topology", "worker", "comparison_class", "access_product", } VARIANT_TOPOLOGIES = {"root_only", "root_plus_worker", "full", "full_without_worker"} VARIANT_COMPARISON_CLASSES = { "configured_root_alone", "strongest_single_agent", "access_service_single_agent", "root_plus_highest_value", "full_profile", "ablation", "control", } EVALUATION_IMAGE_SUFFIXES = {".gif", ".jpeg", ".jpg", ".png", ".webp"} RUN_ID_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,199}") def _reject_unknown_fields(value: Mapping[str, Any], allowed: set[str], label: str) -> None: unknown = sorted(set(value) - allowed) if unknown: raise ValueError(f"{label} has unknown fields: {', '.join(unknown)}") def _required_string(value: Any, label: str) -> str: if not isinstance(value, str) or not value.strip(): raise ValueError(f"{label} must be a non-empty string") return value def _required_id(value: Any, label: str) -> str: return validate_id(_required_string(value, label), label) def builtin_evals_root() -> Path: return install_root() / "evals" def user_evals_root() -> Path: return config_root() / "evals.d" def evaluations_state_root() -> Path: root = state_root() / "evaluations" root.mkdir(parents=True, exist_ok=True, mode=0o700) return root def discover_suites() -> dict[str, dict[str, Any]]: results: dict[str, dict[str, Any]] = {} for source, root in (("builtin", builtin_evals_root()), ("user", user_evals_root())): if not root.is_dir(): continue for directory in sorted(path for path in root.iterdir() if path.is_dir()): path = directory / "suite.toml" if not path.is_file(): continue with contextlib.suppress(Exception): data = read_toml(path) suite_id = _required_id(data["id"], "evaluation suite id") results[suite_id] = { "id": suite_id, "name": data.get("name", suite_id), "description": data.get("description", ""), "source": source, "path": str(directory), "task_count": len(data.get("tasks", [])), } return results def resolve_suite(value: str | Path) -> tuple[Path, dict[str, Any]]: candidate = Path(value).expanduser() if candidate.exists(): directory = candidate.resolve() if directory.is_file(): directory = directory.parent else: suite_id = validate_id(str(value), "evaluation suite id") suites = discover_suites() if suite_id not in suites: raise FileNotFoundError(f"unknown evaluation suite: {suite_id}") directory = Path(suites[suite_id]["path"]) path = directory / "suite.toml" data = read_toml(path) _reject_unknown_fields(data, SUITE_FIELDS, f"evaluation suite {path}") schema_version = data.get("schema_version") if ( not isinstance(schema_version, int) or isinstance(schema_version, bool) or schema_version != MMO_SCHEMA_VERSION ): raise ValueError(f"unsupported suite schema in {path}") _required_id(data.get("id"), "evaluation suite id") _required_id(data.get("profile"), "evaluation suite profile") for field in ("name", "description"): if field in data and not isinstance(data[field], str): raise ValueError(f"evaluation suite {field} must be a string") tasks = data.get("tasks") if not isinstance(tasks, list) or not tasks: raise ValueError("evaluation suite must contain [[tasks]]") seen: set[str] = set() for field, default in (("development_trials", 3), ("release_trials", 5)): trials = data.get(field, default) if not isinstance(trials, int) or isinstance(trials, bool) or not 1 <= trials <= 20: raise ValueError(f"evaluation suite {field} must be an integer from 1 to 20") variants = data.get("variants", []) if not isinstance(variants, list): raise ValueError("evaluation suite variants must be an array of tables") variant_ids: set[str] = set() for index, variant in enumerate(variants): if not isinstance(variant, Mapping): raise ValueError(f"evaluation variant {index} must be a table") _reject_unknown_fields(variant, VARIANT_FIELDS, f"evaluation variant {index}") variant_id = _required_id(variant.get("id"), f"evaluation variant {index} id") if variant_id in variant_ids: raise ValueError(f"duplicate evaluation variant id: {variant_id}") variant_ids.add(variant_id) if not isinstance(variant.get("purpose"), str) or not variant["purpose"].strip(): raise ValueError(f"evaluation variant {variant_id}: purpose is required") if "profile" in variant and not isinstance(variant["profile"], str): raise ValueError(f"evaluation variant {variant_id}: profile must be a string") bindings = variant.get("bindings", {}) if not isinstance(bindings, Mapping) or not all( isinstance(key, str) and isinstance(value, str) for key, value in bindings.items() ): raise ValueError(f"evaluation variant {variant_id}: bindings must be a string table") topology = variant.get("topology") if topology not in VARIANT_TOPOLOGIES: raise ValueError( f"evaluation variant {variant_id}: topology must be one of " f"{sorted(VARIANT_TOPOLOGIES)}" ) comparison_class = variant.get("comparison_class") if comparison_class not in VARIANT_COMPARISON_CLASSES: raise ValueError( f"evaluation variant {variant_id}: comparison_class must be one of " f"{sorted(VARIANT_COMPARISON_CLASSES)}" ) if (topology == "full_without_worker") != (comparison_class == "ablation"): raise ValueError( f"evaluation variant {variant_id}: full_without_worker topology and " "ablation comparison_class must be used together" ) worker = variant.get("worker") if topology in {"root_plus_worker", "full_without_worker"}: _required_id(worker, f"evaluation variant {variant_id} worker") elif worker is not None: raise ValueError( f"evaluation variant {variant_id}: worker is allowed only for " "root_plus_worker topology" ) access_product = variant.get("access_product") if comparison_class == "access_service_single_agent": _required_id( access_product, f"evaluation variant {variant_id} access_product", ) elif access_product is not None: raise ValueError( f"evaluation variant {variant_id}: access_product is reserved for " "access-service controls" ) required_comparisons = { "configured_root_alone", "strongest_single_agent", "root_plus_highest_value", "full_profile", } comparison_classes = {str(variant.get("comparison_class")) for variant in variants} missing_comparisons = sorted(required_comparisons - comparison_classes) if missing_comparisons: raise ValueError( "evaluation suite lacks required matched comparison classes: " + ", ".join(missing_comparisons) ) if "access_service_single_agent" not in comparison_classes: raise ValueError( "evaluation suite requires at least one access_service_single_agent variant" ) promotion = data.get("promotion") if not isinstance(promotion, Mapping): raise ValueError("evaluation suite promotion table is required") _reject_unknown_fields(promotion, PROMOTION_FIELDS, "evaluation promotion") primary_metric = _required_id( promotion.get("primary_metric"), "evaluation promotion primary_metric" ) if primary_metric == "score": raise ValueError("promotion primary_metric must be success_rate, not legacy score") primary_baseline = promotion.get("primary_baseline", "configured_root_alone") if primary_baseline not in {"configured_root_alone", "strongest_single_agent"}: raise ValueError( "evaluation promotion primary_baseline must be configured_root_alone or " "strongest_single_agent" ) if promotion.get("direction") not in {"higher", "lower"}: raise ValueError("evaluation promotion direction must be higher or lower") for field, default_number in ( ("strongest_success_tolerance", 0.02), ("minimum_relative_improvement", 0.10), ("minimum_absolute_improvement", 0.05), ("worker_minimum_success_contribution", 0.02), ("worker_minimum_metric_contribution", 0.10), ): number = promotion.get(field, default_number) if ( not isinstance(number, (int, float)) or isinstance(number, bool) or not math.isfinite(float(number)) or float(number) < 0 ): raise ValueError(f"evaluation promotion {field} must be a non-negative number") metric_sets: list[set[str]] = [] for field in ("no_regression_higher_metrics", "no_regression_lower_metrics"): metrics = promotion.get(field, []) if not isinstance(metrics, list) or not all( isinstance(metric, str) and metric.strip() for metric in metrics ): raise ValueError(f"evaluation promotion {field} must be strings") normalized = {validate_id(metric, f"evaluation promotion {field}") for metric in metrics} if len(normalized) != len(metrics): raise ValueError(f"evaluation promotion {field} contains duplicates") metric_sets.append(normalized) if metric_sets[0] & metric_sets[1]: raise ValueError("a non-regression metric cannot have both directions") scarce_model_keys = promotion.get("scarce_model_keys", []) if not isinstance(scarce_model_keys, list) or not all( isinstance(key, str) and key.strip() for key in scarce_model_keys ): raise ValueError("evaluation promotion scarce_model_keys must be an array of model keys") if len(scarce_model_keys) != len(set(scarce_model_keys)): raise ValueError("evaluation promotion scarce_model_keys contains duplicates") if primary_metric == "scarce_tier_request_units" and not scarce_model_keys: raise ValueError( "evaluation promotion scarce_tier_request_units requires scarce_model_keys" ) if not isinstance(promotion.get("require_complete_api_cost", True), bool): raise ValueError("evaluation promotion require_complete_api_cost must be boolean") for index, task in enumerate(tasks): if not isinstance(task, Mapping): raise ValueError(f"evaluation task {index} must be a table") _reject_unknown_fields(task, TASK_FIELDS, f"evaluation task {index}") task_id = _required_id(task.get("id"), f"task {index} id") if task_id in seen: raise ValueError(f"duplicate evaluation task id: {task_id}") seen.add(task_id) if not isinstance(task.get("prompt"), str) or not task["prompt"].strip(): raise ValueError(f"task {task_id}: prompt is required") if "description" in task and not isinstance(task["description"], str): raise ValueError(f"task {task_id}: description must be a string") sandbox = task.get("sandbox", "read-only") if not isinstance(sandbox, str) or sandbox not in {"read-only", "workspace-write"}: raise ValueError(f"task {task_id}: invalid sandbox") for field, default, maximum in ( ("wall_timeout_seconds", 1800, 172_800), ("validation_timeout_seconds", 300, 86_400), ): timeout = task.get(field, default) if ( not isinstance(timeout, int) or isinstance(timeout, bool) or not 1 <= timeout <= maximum ): raise ValueError(f"task {task_id}: {field} must be an integer from 1 to {maximum}") for field in ("images",): value = task.get(field, []) if not isinstance(value, list) or not all(isinstance(item, str) for item in value): raise ValueError(f"task {task_id}: {field} must be an array of strings") if any(not item.strip() for item in value): raise ValueError(f"task {task_id}: {field} entries must be non-empty strings") difficulty = task.get("difficulty", "medium") if difficulty not in {"easy", "medium", "hard", "adversarial"}: raise ValueError(f"task {task_id}: invalid difficulty") if not isinstance(task.get("negative_control", False), bool): raise ValueError(f"task {task_id}: negative_control must be boolean") route_faults = task.get("route_faults", {}) if not isinstance(route_faults, Mapping) or not all( isinstance(route, str) and route.strip() and isinstance(fault, str) and fault in ROUTE_FAULT_TYPES for route, fault in route_faults.items() ): raise ValueError( f"task {task_id}: route_faults must map route IDs to one of " f"{sorted(ROUTE_FAULT_TYPES)}" ) for route in route_faults: validate_id(route, f"task {task_id} disabled route") for image in task.get("images", []): image_path = Path(image) if ( image_path.is_absolute() or ".." in image_path.parts or image_path.suffix.lower() not in EVALUATION_IMAGE_SUFFIXES ): raise ValueError(f"task {task_id}: images must be supported relative fixture paths") outcomes = task.get("outcome_assertions") if not isinstance(outcomes, Mapping): raise ValueError(f"task {task_id}: outcome_assertions table is required") _reject_unknown_fields(outcomes, OUTCOME_ASSERTION_FIELDS, f"task {task_id} outcomes") if not any(outcomes.get(field) for field in OUTCOME_ASSERTION_FIELDS): raise ValueError(f"task {task_id}: outcome_assertions cannot be empty") for field in OUTCOME_ASSERTION_FIELDS: entries = outcomes.get(field, []) if not isinstance(entries, list) or not all(isinstance(item, str) for item in entries): raise ValueError(f"task {task_id}: {field} must be an array of strings") if any(not item.strip() for item in entries): raise ValueError(f"task {task_id}: {field} entries must be non-empty") if field != "validation_commands": for pattern in entries: re.compile(pattern) assertions = task.get("orchestration_assertions", {}) if not isinstance(assertions, Mapping): raise ValueError(f"task {task_id}: orchestration_assertions must be a table") _reject_unknown_fields( assertions, ORCHESTRATION_ASSERTION_FIELDS, f"task {task_id} orchestration_assertions", ) for field in ("required_agents", "forbidden_agents"): agents = assertions.get(field, []) if not isinstance(agents, list) or not all(isinstance(item, str) for item in agents): raise ValueError(f"task {task_id}: {field} must be an array of agent IDs") if len(agents) != len(set(agents)): raise ValueError(f"task {task_id}: {field} contains duplicates") for agent in agents: validate_id(agent, f"task {task_id} {field} agent") overlap = set(assertions.get("required_agents", [])) & set( assertions.get("forbidden_agents", []) ) if overlap: raise ValueError( f"task {task_id}: agents cannot be both required and forbidden: {sorted(overlap)}" ) for field in ("min_peak_mcp_workers", "max_jobs", "max_contract_failures"): if field not in assertions: continue number = assertions[field] if not isinstance(number, int) or isinstance(number, bool) or number < 0: raise ValueError(f"task {task_id}: {field} must be a non-negative integer") for field in ("min_result_acceptance_rate", "max_observed_mcp_wait_ratio"): if field not in assertions: continue number = assertions[field] if ( not isinstance(number, (int, float)) or isinstance(number, bool) or not math.isfinite(float(number)) or not 0 <= float(number) <= 1 ): raise ValueError(f"task {task_id}: {field} must be a number from 0 to 1") fixture = data.get("fixture") fixture_path: Path | None = None if fixture is not None: if not isinstance(fixture, str) or not fixture.strip(): raise ValueError("suite fixture must be a non-empty relative path") if Path(fixture).is_absolute(): raise ValueError("suite fixture must be a non-empty relative path") candidate = directory / fixture if candidate.is_symlink(): raise ValueError("suite fixture may not be a symbolic link") fixture_path = candidate.resolve() if directory not in fixture_path.parents or not fixture_path.is_dir(): raise ValueError("suite fixture must be a directory inside the suite") for member in fixture_path.rglob("*"): relative = member.relative_to(fixture_path) if member.is_symlink(): raise ValueError(f"evaluation fixtures may not contain symlinks: {relative}") if not member.is_dir() and not member.is_file(): raise ValueError(f"evaluation fixtures may not contain special files: {relative}") for task in tasks: if task.get("images") and fixture_path is None: raise ValueError(f"task {task['id']}: images require a suite fixture") for image in task.get("images", []): assert fixture_path is not None candidate = fixture_path / image if candidate.is_symlink(): raise ValueError(f"task {task['id']}: image may not be a symbolic link: {image}") resolved_image = candidate.resolve() if fixture_path not in resolved_image.parents or not resolved_image.is_file(): raise ValueError(f"task {task['id']}: image is not a fixture file: {image}") task_ids = {str(task["id"]) for task in tasks} for auxiliary_name in ("holdout", "mutations"): auxiliary_root = directory / auxiliary_name if not auxiliary_root.exists(): continue if auxiliary_root.is_symlink() or not auxiliary_root.is_dir(): raise ValueError(f"evaluation {auxiliary_name} must be a static directory") for member in auxiliary_root.rglob("*"): relative = member.relative_to(auxiliary_root) if member.is_symlink(): raise ValueError( f"evaluation {auxiliary_name} may not contain symlinks: {relative}" ) if not member.is_dir() and not member.is_file(): raise ValueError( f"evaluation {auxiliary_name} may not contain special files: {relative}" ) if relative.parts and relative.parts[0] not in task_ids: raise ValueError( f"evaluation {auxiliary_name} references unknown task: {relative.parts[0]}" ) if auxiliary_name == "mutations" and member.is_file() and member.suffix != ".patch": raise ValueError(f"evaluation mutations must be .patch files: {relative}") return directory, data def validate_suite(value: str | Path) -> dict[str, Any]: directory, data = resolve_suite(value) return { "id": data["id"], "name": data.get("name", data["id"]), "path": str(directory), "tasks": [task["id"] for task in data["tasks"]], "hidden_mutation_tasks": sorted(path.name for path in (directory / "mutations").iterdir()) if (directory / "mutations").is_dir() else [], "holdout_tasks": sorted(path.name for path in (directory / "holdout").iterdir()) if (directory / "holdout").is_dir() else [], "valid": True, } def _variant_snapshot( configured_profile: str | Path, configured_bindings: Mapping[str, str] | None, variant: Mapping[str, Any], ) -> dict[str, Any]: """Compile one matched evaluation topology as an immutable v2 snapshot.""" selected_profile: str | Path = variant.get("profile") or configured_profile bindings = dict(configured_bindings or {}) bindings.update(dict(variant.get("bindings") or {})) base = compile_profile(selected_profile, bindings=bindings) topology = str(variant["topology"]) if topology == "full": return base resolved = copy.deepcopy(base["resolved"]) root_id = str(resolved["profile"]["root"]) worker = variant.get("worker") if topology == "root_only": resolved["agents"][root_id]["can_spawn"] = [] resolved["coordination"].update( { "max_active_agents": 1, "max_depth": 0, "max_active_writers": 0, "feasible_max_active_agents": 1, } ) elif topology == "root_plus_worker": if not isinstance(worker, str) or worker not in resolved["agents"]: raise ValueError( f"evaluation variant {variant['id']}: unknown root-plus worker {worker!r}" ) if worker not in resolved["agents"][root_id]["can_spawn"]: raise ValueError( f"evaluation variant {variant['id']}: root cannot spawn worker {worker!r}" ) resolved["agents"][root_id]["can_spawn"] = [worker] resolved["agents"][worker]["can_spawn"] = [] writable = ( resolved["agents"][worker]["permissions"] == "workspace-write" and "mcp" in resolved["agents"][worker]["backends"] ) resolved["coordination"].update( { "max_active_agents": 2, "max_depth": 1, "max_children_per_agent": 1, "max_active_writers": 1 if writable else 0, "feasible_max_active_agents": 2, } ) for agent_id, agent in resolved["agents"].items(): if agent_id not in {root_id, worker}: agent["can_spawn"] = [] else: if not isinstance(worker, str) or worker not in resolved["agents"]: raise ValueError( f"evaluation variant {variant['id']}: unknown ablated worker {worker!r}" ) if worker == root_id: raise ValueError(f"evaluation variant {variant['id']}: root cannot be ablated") for agent in resolved["agents"].values(): agent["can_spawn"] = [child for child in agent["can_spawn"] if child != worker] resolved["agents"][worker]["can_spawn"] = [] # A matched topology may not retain dormant control edges to roles that # its spawn graph cannot create. Prune both outgoing spawn and control # authority to the exact root-reachable role set before compiling guidance # and Agent-MCP tools for the variant. reachable = {root_id} frontier = [root_id] while frontier: parent = frontier.pop() for child in resolved["agents"][parent]["can_spawn"]: if child not in reachable: reachable.add(child) frontier.append(child) for agent_id, agent in resolved["agents"].items(): if agent_id not in reachable: agent["can_spawn"] = [] agent["controls"] = {} continue agent["controls"] = { target: grant for target, grant in agent.get("controls", {}).items() if target in reachable } coordination = resolved["coordination"] capacities = derive_coordination_capacities( resolved["agents"], resolved["resources"], root_agent=root_id, max_depth=int(coordination["max_depth"]), ) feasible_active = int(capacities["feasible_max_active_agents"]) if resolved["agents"][root_id]["can_spawn"] and feasible_active < 2: raise ValueError( f"evaluation variant {variant['id']}: no configured worker can run within " "the root's resource reservation" ) coordination.update( { "max_active_agents": min(int(coordination["max_active_agents"]), feasible_active), "max_active_writers": min( int(coordination["max_active_writers"]), int(capacities["writable_slots"]) ), "feasible_max_active_agents": feasible_active, } ) resolved["evaluation_variant"] = { "id": variant["id"], "comparison_class": variant["comparison_class"], "topology": topology, "worker": worker, "source_snapshot_hash": base["manifest"]["snapshot_hash"], } hash_payload = copy.deepcopy(resolved) hash_payload.pop("logical_hash", None) resolved["logical_hash"] = stable_hash(hash_payload) return compile_resolved_profile(resolved) def _run_id(profile_id: str, suite_id: str) -> str: stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") return f"{stamp}-{safe_name(profile_id)}-{safe_name(suite_id)}-{uuid.uuid4().hex[:8]}" def _copy_fixture(source: Path | None, destination: Path) -> None: destination.mkdir(parents=True, exist_ok=False) if source is None: return for path in source.rglob("*"): relative = path.relative_to(source) target = destination / relative if path.is_symlink(): raise ValueError(f"evaluation fixtures may not contain symlinks: {relative}") if path.is_dir(): target.mkdir(parents=True, exist_ok=True) elif path.is_file(): target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(path, target) else: raise ValueError(f"evaluation fixtures may not contain special files: {relative}") def _copy_overlay(source: Path, destination: Path) -> list[str]: copied: list[str] = [] for path in sorted(source.rglob("*")): relative = path.relative_to(source) target = destination / relative if path.is_symlink(): raise ValueError(f"evaluation holdouts may not contain symlinks: {relative}") if path.is_dir(): target.mkdir(parents=True, exist_ok=True) elif path.is_file(): target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(path, target) copied.append(relative.as_posix()) else: raise ValueError(f"evaluation holdouts may not contain special files: {relative}") return copied def _prepare_hidden_trial( suite_dir: Path, task_id: str, trial: int, workspace: Path ) -> dict[str, Any]: """Apply one deterministic hidden mutation shared by every matched variant.""" root = suite_dir / "mutations" / task_id candidates = sorted(root.glob("*.patch")) if root.is_dir() else [] if not candidates: return {"configured": False, "selected": None, "sha256": None} selected = candidates[(trial - 1) % len(candidates)].resolve() completed = subprocess.run( ["git", "apply", "--binary", str(selected)], cwd=workspace, text=True, capture_output=True, check=False, env=filtered_environment(extra={"GIT_TERMINAL_PROMPT": "0"}), ) if completed.returncode != 0: raise RuntimeError( f"hidden mutation failed for {task_id}: {selected.name}: {completed.stderr.strip()}" ) return { "configured": True, "selected": selected.name, "sha256": sha256_bytes(selected.read_bytes()), "candidate_count": len(candidates), } def _initialize_fixture_repository(workspace: Path) -> None: """Create the Git boundary required by isolated writable Agent MCP workers.""" commands = ( ["git", "init", "--quiet"], ["git", "config", "user.name", "MMO Evaluation"], ["git", "config", "user.email", "eval@invalid.example"], ["git", "add", "--all"], ["git", "commit", "--quiet", "--allow-empty", "-m", "matched evaluation fixture"], ) for command in commands: completed = subprocess.run( command, cwd=workspace, text=True, capture_output=True, check=False, env=filtered_environment(extra={"GIT_TERMINAL_PROMPT": "0"}), ) if completed.returncode != 0: raise RuntimeError( f"unable to initialize evaluation fixture repository: {completed.stderr.strip()}" ) def _install_holdout(suite_dir: Path, task_id: str, workspace: Path) -> dict[str, Any]: source = suite_dir / "holdout" / task_id if not source.is_dir(): return {"configured": False, "files": []} files = _copy_overlay(source, workspace) return {"configured": True, "files": files} def _run_validation(command: str, cwd: Path, timeout: int) -> dict[str, Any]: started = dt.datetime.now(dt.UTC) process = subprocess.Popen( ["sh", "-lc", command], cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True, env=filtered_environment(extra={"GIT_TERMINAL_PROMPT": "0", "NO_COLOR": "1"}), ) try: stdout, stderr = process.communicate(timeout=timeout) return { "command": command, "exit_code": process.returncode, "passed": process.returncode == 0, "stdout": stdout[-12000:], "stderr": stderr[-12000:], "elapsed_seconds": (dt.datetime.now(dt.UTC) - started).total_seconds(), } except subprocess.TimeoutExpired: terminate_process_group(process.pid) stdout, stderr = process.communicate() return { "command": command, "exit_code": 124, "passed": False, "stdout": stdout[-12000:], "stderr": stderr[-12000:], "elapsed_seconds": (dt.datetime.now(dt.UTC) - started).total_seconds(), "error": "validation command timed out", } def _numeric_validation_metrics(validations: Sequence[Mapping[str, Any]]) -> dict[str, float]: """Read typed numeric metrics from a validator's final JSON object.""" buckets: dict[str, list[float]] = {} for validation in validations: stdout = str(validation.get("stdout") or "") payload: Mapping[str, Any] | None = None for line in reversed(stdout.splitlines()): with contextlib.suppress(json.JSONDecodeError, ValueError): candidate = strict_json_loads(line) if isinstance(candidate, Mapping): payload = candidate break if payload is None or not isinstance(payload.get("metrics"), Mapping): continue for name, value in payload["metrics"].items(): if ( isinstance(name, str) and validate_id(name, "validator metric") and isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)) ): buckets.setdefault(name, []).append(float(value)) return {name: sum(values) / len(values) for name, values in sorted(buckets.items())} def _pattern_results(output: str, task: Mapping[str, Any]) -> dict[str, Any]: assertions = task["outcome_assertions"] expected = [ {"pattern": pattern, "matched": bool(re.search(pattern, output, flags=re.MULTILINE))} for pattern in assertions.get("expected_patterns", []) ] forbidden = [ {"pattern": pattern, "matched": bool(re.search(pattern, output, flags=re.MULTILINE))} for pattern in assertions.get("forbidden_patterns", []) ] return { "expected": expected, "forbidden": forbidden, "passed": all(item["matched"] for item in expected) and not any(item["matched"] for item in forbidden), } def _jsonl_rows(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] if not path.is_file(): return rows with path.open("r", encoding="utf-8", errors="replace") as handle: for line in handle: with contextlib.suppress(json.JSONDecodeError, ValueError): value = strict_json_loads(line) if isinstance(value, dict): rows.append(value) return rows def _event_usage(path: Path) -> dict[str, int]: return event_usage(path) def _retry_count(path: Path) -> int: observed = 0 def visit(value: Any) -> None: nonlocal observed if isinstance(value, Mapping): for key, child in value.items(): normalized = str(key).lower() if ( normalized in {"retry_count", "retries"} and isinstance(child, int) and not isinstance(child, bool) ): observed += max(0, child) visit(child) elif isinstance(value, list): for child in value: visit(child) for row in _jsonl_rows(path): visit(row) return observed def _estimated_token_cost(usage: Mapping[str, Any], model: Mapping[str, Any]) -> float | None: input_price = model.get("input_cost_per_million") output_price = model.get("output_cost_per_million") if input_price is None or output_price is None: return None input_tokens = usage.get("input_tokens", 0) cached_tokens = usage.get("cached_input_tokens", 0) cache_write_tokens = usage.get("cache_write_input_tokens", 0) output_tokens = usage.get("output_tokens", 0) if any( not isinstance(value, int) or isinstance(value, bool) or value < 0 for value in (input_tokens, cached_tokens, cache_write_tokens, output_tokens) ): return None if cached_tokens + cache_write_tokens > input_tokens: return None cached_price = model.get("cached_input_cost_per_million") cache_write_price = model.get("cache_write_input_cost_per_million") if cached_tokens and cached_price is None: return None if cache_write_tokens and cache_write_price is None: return None try: amount = ( Decimal(input_tokens - cached_tokens - cache_write_tokens) * Decimal(str(input_price)) + Decimal(cached_tokens) * Decimal(str(cached_price or 0)) + Decimal(cache_write_tokens) * Decimal(str(cache_write_price or 0)) + Decimal(output_tokens) * Decimal(str(output_price)) ) / Decimal(1_000_000) result = float(amount) except (InvalidOperation, OverflowError, ValueError): return None return result if math.isfinite(result) else None def _observed_api_cost(path: Path) -> float | None: maxima: list[float] = [] def visit(value: Any) -> None: if isinstance(value, Mapping): for key, child in value.items(): if ( str(key).casefold() in { "cost", "cost_usd", "total_cost", "total_cost_usd", } and isinstance(child, (int, float)) and not isinstance(child, bool) ): amount = float(child) if math.isfinite(amount) and amount >= 0: maxima.append(amount) visit(child) elif isinstance(value, list): for child in value: visit(child) for row in _jsonl_rows(path): visit(row) return max(maxima) if maxima else None def _call_cost_ledgers( *, usage: Mapping[str, Any], model: Mapping[str, Any], route: Mapping[str, Any], events_path: Path, elapsed_seconds: float, ) -> dict[str, Any]: billing = str(route["billing_mode"]) access_product = str(route["access_product"]) tokens = sum( int(usage.get(field, 0)) for field in ("input_tokens", "output_tokens") if isinstance(usage.get(field, 0), int) and not isinstance(usage.get(field, 0), bool) ) actual = _observed_api_cost(events_path) if billing == "api" else None estimate = _estimated_token_cost(usage, model) subscription: dict[str, Any] = {} local: dict[str, float] = {} if billing in {"subscription", "chatgpt_subscription"}: subscription[access_product] = {"request_units": 1, "observed_tokens": tokens} if billing == "local": local[access_product] = max(0.0, elapsed_seconds) return { "actual_api_usd": actual, "actual_api_usd_complete": billing != "api" or actual is not None, "api_equivalent_estimate_usd": estimate, "api_equivalent_estimate_complete": estimate is not None, "subscription_units": subscription, "local_resource_seconds": local, } def _merge_cost_ledgers(items: Sequence[Mapping[str, Any]]) -> dict[str, Any]: actual = 0.0 estimated = 0.0 actual_complete = True estimated_complete = True subscription: dict[str, Counter[str]] = {} local: Counter[str] = Counter() for item in items: actual_complete = actual_complete and bool(item.get("actual_api_usd_complete")) estimated_complete = estimated_complete and bool( item.get("api_equivalent_estimate_complete") ) if item.get("actual_api_usd") is not None: actual += float(item["actual_api_usd"]) if item.get("api_equivalent_estimate_usd") is not None: estimated += float(item["api_equivalent_estimate_usd"]) for product, units in item.get("subscription_units", {}).items(): bucket = subscription.setdefault(str(product), Counter()) bucket.update({str(key): int(value) for key, value in units.items()}) local.update( { str(product): float(seconds) for product, seconds in item.get("local_resource_seconds", {}).items() } ) return { "actual_api_usd": actual if actual_complete else None, "actual_api_usd_complete": actual_complete, "api_equivalent_estimate_usd": estimated if estimated_complete else None, "api_equivalent_estimate_complete": estimated_complete, "subscription_units": { product: dict(units) for product, units in sorted(subscription.items()) }, "local_resource_seconds": dict(sorted(local.items())), } def _audit_metrics(session_id: str, jobs: Sequence[Mapping[str, Any]]) -> dict[str, Any]: audit_path = state_root() / "sessions" / session_id / "audit.jsonl" rows = _jsonl_rows(audit_path) root_wait_seconds = 0.0 worker_wait_seconds = 0.0 read: set[str] = set() cancel_reasons: Counter[str] = Counter() scope_conflicts = 0 for row in rows: event = row.get("event") if event == "agents_wait_finished": seconds = float(row.get("waited_seconds") or 0.0) if row.get("caller_job_id"): worker_wait_seconds += seconds else: root_wait_seconds += seconds elif event == "agent_result_read" and row.get("terminal"): target = row.get("target_job_id") if target: read.add(str(target)) elif event == "agent_cancelled": reason = str(row.get("reason") or "unspecified") cancel_reasons[reason] += 1 elif ( event in {"spawn_rejected", "spawn_batch_rejected"} and row.get("reason") == "write_scope_conflict" ): scope_conflicts += 1 terminal = { str(job.get("job_id")) for job in jobs if job.get("status") not in {"queued", "running", "cancelling"} } accepted = { str(job.get("job_id")) for job in jobs if job.get("result_state") in {"accepted", "integrated"} } integrated = {str(job.get("job_id")) for job in jobs if job.get("result_state") == "integrated"} return { "explicit_root_mcp_wait_seconds": root_wait_seconds, "explicit_worker_mcp_wait_seconds": worker_wait_seconds, "terminal_results_read": len(terminal & read), "result_read_rate": len(terminal & read) / len(terminal) if terminal else None, "terminal_results_accepted": len(terminal & accepted), "result_acceptance_rate": len(terminal & accepted) / len(terminal) if terminal else None, "patches_integrated": len(integrated), "cancel_reasons": dict(cancel_reasons), "cancelled_duplicates": sum( count for reason, count in cancel_reasons.items() if "duplicate" in reason.lower() ), "write_scope_conflicts": scope_conflicts, "audit_event_count": len(rows), } def _timestamp(value: Any) -> dt.datetime | None: if not isinstance(value, str): return None with contextlib.suppress(ValueError): parsed = dt.datetime.fromisoformat(value) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=dt.UTC) return parsed.astimezone(dt.UTC) return None def _interval_metrics(session_id: str, jobs: Sequence[Mapping[str, Any]]) -> dict[str, Any]: intervals: list[tuple[dt.datetime, dt.datetime]] = [] terminal_times: list[dt.datetime] = [] for job in jobs: started = _timestamp(job.get("started_at")) finished = _timestamp(job.get("finished_at")) if finished is not None: terminal_times.append(finished) if started is not None and finished is not None and finished >= started: intervals.append((started, finished)) events: dict[dt.datetime, int] = {} for started, finished in intervals: events[started] = events.get(started, 0) + 1 events[finished] = events.get(finished, 0) - 1 active = 0 peak = 0 overlap_seconds = 0.0 previous: dt.datetime | None = None for stamp in sorted(events): if previous is not None and active >= 2: overlap_seconds += max(0.0, (stamp - previous).total_seconds()) active += events[stamp] peak = max(peak, active) previous = stamp session_started = None with contextlib.suppress(FileNotFoundError, RuntimeError, ValueError): session_started = _timestamp(load_session(session_id).get("started_at")) first_terminal = min(terminal_times) if terminal_times else None time_to_first = ( max(0.0, (first_terminal - session_started).total_seconds()) if first_terminal is not None and session_started is not None else None ) return { "peak_mcp_workers": peak, "worker_overlap_seconds": overlap_seconds, "time_to_first_terminal_result_seconds": time_to_first, } def _root_activity_metrics(events_path: Path, session_id: str) -> dict[str, Any]: rows = _jsonl_rows(events_path) timestamps = [stamp for row in rows if (stamp := _timestamp(row.get("timestamp"))) is not None] jobs = [job for job in iter_jobs(strict=True) if job.get("session_id") == session_id] intervals = [ (started, finished) for job in jobs if (started := _timestamp(job.get("started_at"))) is not None and (finished := _timestamp(job.get("finished_at"))) is not None ] overlap_events = sum( any(started <= stamp <= finished for started, finished in intervals) for stamp in timestamps ) return { "telemetry_coverage": { "event_count": len(rows), "timestamped_event_count": len(timestamps), "timestamp_coverage_ratio": len(timestamps) / len(rows) if rows else None, "activity_overlap_seconds": None, "activity_overlap_seconds_reason": ( "point events do not establish continuous root activity intervals" ), }, "observed_root_activity_events_during_worker_execution": overlap_events, } def _integration_correction_metrics(jobs: Sequence[Mapping[str, Any]]) -> dict[str, Any]: integrated_paths = 0 corrected_paths = 0 corrected_jobs = 0 for job in jobs: if job.get("result_state") != "integrated": continue canonical_cwd = Path(str(job.get("canonical_cwd", ""))).resolve(strict=False) patch = job.get("patch") changed = patch.get("changed_paths", []) if isinstance(patch, Mapping) else [] artifacts = { str(item.get("relative_path")): str(item.get("sha256")) for item in job.get("artifacts", []) if isinstance(item, Mapping) and isinstance(item.get("relative_path"), str) and isinstance(item.get("sha256"), str) } job_corrected = False for raw_path in changed: if not isinstance(raw_path, str): continue relative = Path(raw_path) if relative.is_absolute() or ".." in relative.parts: continue integrated_paths += 1 target = (canonical_cwd / relative).resolve(strict=False) expected_hash = artifacts.get(relative.as_posix()) if expected_hash is None: corrected = target.exists() else: corrected = ( not target.is_file() or target.is_symlink() or sha256_bytes(target.read_bytes()) != expected_hash ) if corrected: corrected_paths += 1 job_corrected = True corrected_jobs += int(job_corrected) return { "integrated_patch_paths": integrated_paths, "integration_corrected_paths": corrected_paths, "integrated_jobs_with_corrections": corrected_jobs, "integration_correction_rate": ( corrected_paths / integrated_paths if integrated_paths else None ), } def _session_job_metrics(session_id: str, resolved: Mapping[str, Any]) -> dict[str, Any]: jobs = [job for job in iter_jobs(strict=True) if job.get("session_id") == session_id] statuses = Counter(str(job.get("status")) for job in jobs) agents = Counter(str(job.get("agent")) for job in jobs) models = Counter(str(job.get("model")) for job in jobs) routes = Counter(str(job.get("route")) for job in jobs) makers = Counter(str(job.get("maker")) for job in jobs) api_operators: Counter[str] = Counter() access_products: Counter[str] = Counter() gateway_drivers: Counter[str] = Counter() serving_providers: Counter[str] = Counter() serving_endpoints: Counter[str] = Counter() route_telemetry_incomplete = 0 usage: Counter[str] = Counter() warnings = 0 contract_failures = 0 conflicts = 0 cost_ledgers: list[dict[str, Any]] = [] for job in jobs: if job.get("warning"): warnings += 1 if job.get("contract_valid") is False: contract_failures += 1 result_path = Path(str(job.get("structured_result_path", ""))) if result_path.is_file(): with contextlib.suppress(Exception): value = read_json(result_path) if value.get("status") == "conflict" or value.get("conflicts"): conflicts += 1 for key, value in (job.get("usage") or {}).items(): if isinstance(value, int) and not isinstance(value, bool): usage[key] += value model_key = job.get("model_key") model = resolved["models"].get(model_key, {}) route = resolved["routes"].get(model.get("route"), {}) if model and route: api_operators[str(route["api_operator"])] += 1 access_products[str(route["access_product"])] += 1 gateway_drivers[str(route["driver"])] += 1 route_telemetry = job.get("route_telemetry") or {} for value in route_telemetry.get("actual_serving_provider_slugs", []): serving_providers[str(value)] += 1 for value in route_telemetry.get("actual_serving_endpoint_tags", []): serving_endpoints[str(value)] += 1 if job.get("requested_route_policy") is not None and not route_telemetry.get( "complete" ): route_telemetry_incomplete += 1 cost_ledgers.append( _call_cost_ledgers( usage=job.get("usage") or {}, model=model, route=route, events_path=Path(str(job.get("events_path", ""))), elapsed_seconds=float(job.get("elapsed_seconds") or 0.0), ) ) terminal = sum( value for key, value in statuses.items() if key not in {"queued", "running", "cancelling"} ) successful = statuses.get("completed", 0) + statuses.get("completed_with_warnings", 0) audit = _audit_metrics(session_id, jobs) intervals = _interval_metrics(session_id, jobs) integration_corrections = _integration_correction_metrics(jobs) retries_observed = sum(_retry_count(Path(str(job.get("events_path", "")))) for job in jobs) root = str(resolved["profile"]["root"]) reachable = {root} frontier = [root] while frontier: parent = frontier.pop() for child in resolved["agents"][parent].get("can_spawn", []): if child not in reachable: reachable.add(child) frontier.append(child) return { "profile_agents": sorted(reachable), "job_count": len(jobs), "statuses": dict(statuses), "agents": dict(agents), "models": dict(models), "routes": dict(routes), "makers": dict(makers), "api_operators": dict(api_operators), "access_products": dict(access_products), "gateway_drivers": dict(gateway_drivers), "serving_providers": dict(serving_providers), "serving_endpoints": dict(serving_endpoints), "route_telemetry_incomplete": route_telemetry_incomplete, "usage": dict(usage), "worker_success_rate": successful / terminal if terminal else None, "warnings": warnings, "contract_failures": contract_failures, "reported_conflicts": conflicts, "retries_observed": retries_observed, **audit, **intervals, **integration_corrections, "cost_ledgers": _merge_cost_ledgers(cost_ledgers), } def _orchestration_assertion_results( task: Mapping[str, Any], worker_metrics: Mapping[str, Any], root_elapsed: float ) -> dict[str, Any]: assertions = task.get("orchestration_assertions", {}) checks: list[dict[str, Any]] = [] def add(name: str, expected: Any, actual: Any, passed: bool) -> None: checks.append({"name": name, "expected": expected, "actual": actual, "passed": passed}) observed_agents = set(worker_metrics.get("agents", {})) required = set(assertions.get("required_agents", [])) forbidden = set(assertions.get("forbidden_agents", [])) applicable_required = required & set(worker_metrics.get("profile_agents", observed_agents)) absent_required = required - set(worker_metrics.get("profile_agents", observed_agents)) if required: add( "required_agents", sorted(required), sorted(observed_agents), applicable_required <= observed_agents, ) if absent_required: checks[-1]["not_applicable"] = sorted(absent_required) if forbidden: add( "forbidden_agents", sorted(forbidden), sorted(observed_agents), not bool(forbidden & observed_agents), ) for field, metric, comparison in ( ("min_peak_mcp_workers", "peak_mcp_workers", lambda actual, expected: actual >= expected), ("max_jobs", "job_count", lambda actual, expected: actual <= expected), ("max_contract_failures", "contract_failures", lambda actual, expected: actual <= expected), ): if field in assertions: actual_count = int(worker_metrics.get(metric) or 0) expected_count = int(assertions[field]) add( field, expected_count, actual_count, comparison(actual_count, expected_count), ) if "min_result_acceptance_rate" in assertions: actual_rate = worker_metrics.get("result_acceptance_rate") expected_rate = float(assertions["min_result_acceptance_rate"]) add( "min_result_acceptance_rate", expected_rate, actual_rate, actual_rate is not None and float(actual_rate) >= expected_rate, ) if "max_observed_mcp_wait_ratio" in assertions: wait_seconds = min( root_elapsed, float(worker_metrics.get("explicit_root_mcp_wait_seconds") or 0.0) ) actual_wait_ratio = wait_seconds / root_elapsed if root_elapsed else 0.0 expected_wait_ratio = float(assertions["max_observed_mcp_wait_ratio"]) add( "max_observed_mcp_wait_ratio", expected_wait_ratio, actual_wait_ratio, actual_wait_ratio <= expected_wait_ratio, ) return {"checks": checks, "passed": all(check["passed"] for check in checks)} def _variant_root_availability(snapshot: Mapping[str, Any]) -> dict[str, Any]: resolved = snapshot["resolved"] root = resolved["agents"][resolved["profile"]["root"]] status = route_availability(snapshot)[root["route"]] return {"route": root["route"], **status} def _mean(values: Sequence[float]) -> float | None: return sum(values) / len(values) if values else None def _variant_summary( variant: Mapping[str, Any], records: Sequence[Mapping[str, Any]], *, scarce_model_keys: set[str] | None = None, ) -> dict[str, Any]: passed = sum(bool(record["passed"]) for record in records) root_usage: Counter[str] = Counter() worker_usage: Counter[str] = Counter() worker_agents: Counter[str] = Counter() identities: dict[str, Counter[str]] = { name: Counter() for name in ( "makers", "api_operators", "access_products", "gateway_drivers", "routes", "serving_providers", "serving_endpoints", ) } metric_values: dict[str, list[float]] = {} cost_items: list[Mapping[str, Any]] = [] terminal_jobs = 0 accepted_results = 0 first_results: list[float] = [] scarcity = scarce_model_keys or set() scarce_requests = 0 for record in records: root_usage.update(record.get("root_usage") or {}) worker = record["worker_metrics"] worker_usage.update(worker.get("usage") or {}) worker_agents.update(worker.get("agents") or {}) root_identity = record["root_identity"] if str(root_identity["model"]) in scarcity: scarce_requests += 1 scarce_requests += sum( int(count) for model, count in (worker.get("models") or {}).items() if str(model) in scarcity ) for field in ("maker", "api_operator", "access_product", "gateway_driver", "route"): identities[{"maker": "makers", "route": "routes"}.get(field, field + "s")][ str(root_identity[field]) ] += 1 for field in identities: if field in {"serving_providers", "serving_endpoints"}: identities[field].update(worker.get(field) or {}) elif field == "routes": identities[field].update(worker.get("routes") or {}) else: identities[field].update(worker.get(field) or {}) for name, value in (record.get("outcome_metrics") or {}).items(): metric_values.setdefault(str(name), []).append(float(value)) cost_items.extend((record["root_cost_ledgers"], worker["cost_ledgers"])) terminal_jobs += sum( int(count) for status, count in worker.get("statuses", {}).items() if status not in {"queued", "running", "cancelling"} ) accepted_results += int(worker.get("terminal_results_accepted") or 0) if worker.get("time_to_first_terminal_result_seconds") is not None: first_results.append(float(worker["time_to_first_terminal_result_seconds"])) success_rate = passed / len(records) if records else None metrics = {name: _mean(values) for name, values in sorted(metric_values.items())} metrics["success_rate"] = success_rate root_seconds = sum(float(record.get("root_elapsed_seconds") or 0.0) for record in records) wait_seconds = sum( float(record.get("explicit_root_mcp_wait_seconds") or 0.0) for record in records ) cost_ledgers = _merge_cost_ledgers(cost_items) subscription_request_units = sum( int(units.get("request_units", 0)) for units in cost_ledgers["subscription_units"].values() ) local_resource_seconds = sum(cost_ledgers["local_resource_seconds"].values()) metrics["subscription_request_units"] = float(subscription_request_units) metrics["scarce_tier_request_units"] = float(scarce_requests) metrics["local_resource_seconds"] = float(local_resource_seconds) if cost_ledgers["actual_api_usd"] is not None: metrics["actual_api_usd"] = float(cost_ledgers["actual_api_usd"]) if cost_ledgers["api_equivalent_estimate_usd"] is not None: metrics["api_equivalent_estimate_usd"] = float(cost_ledgers["api_equivalent_estimate_usd"]) integrated_patch_paths = sum( int(record["worker_metrics"].get("integrated_patch_paths") or 0) for record in records ) integration_corrected_paths = sum( int(record["worker_metrics"].get("integration_corrected_paths") or 0) for record in records ) return { "id": variant["id"], "purpose": variant["purpose"], "comparison_class": variant["comparison_class"], "topology": variant["topology"], "worker": variant.get("worker"), "access_product": variant.get("access_product"), "trial_records": len(records), "passed": passed, "failed": len(records) - passed, "success_rate": success_rate, "outcome_metrics": metrics, "mean_root_elapsed_seconds": root_seconds / len(records) if records else None, "explicit_root_mcp_wait_seconds": wait_seconds, "explicit_root_mcp_wait_ratio": wait_seconds / root_seconds if root_seconds else 0.0, "orchestration_diagnostic_pass_rate": ( sum(bool(record["orchestration_diagnostics_passed"]) for record in records) / len(records) if records else None ), "worker_jobs": sum(int(record["worker_metrics"]["job_count"]) for record in records), "contract_failures": sum( int(record["worker_metrics"]["contract_failures"]) for record in records ), "write_scope_conflicts": sum( int(record["worker_metrics"]["write_scope_conflicts"]) for record in records ), "integrated_patch_paths": integrated_patch_paths, "integration_corrected_paths": integration_corrected_paths, "integrated_jobs_with_corrections": sum( int(record["worker_metrics"].get("integrated_jobs_with_corrections") or 0) for record in records ), "integration_correction_rate": ( integration_corrected_paths / integrated_patch_paths if integrated_patch_paths else None ), "route_telemetry_incomplete": sum( int(record["worker_metrics"].get("route_telemetry_incomplete") or 0) for record in records ), "result_acceptance_rate": (accepted_results / terminal_jobs if terminal_jobs else None), "mean_time_to_first_terminal_result_seconds": _mean(first_results), "root_activity_telemetry": [record.get("root_activity") for record in records], "root_usage": dict(root_usage), "worker_usage": dict(worker_usage), "worker_agents": dict(worker_agents), "usage": dict(root_usage + worker_usage), "cost_ledgers": cost_ledgers, "diversity": {name: dict(counter) for name, counter in identities.items()}, } def _directed_improvement( baseline: float, candidate: float, direction: str ) -> dict[str, float | bool | None]: absolute = candidate - baseline if direction == "higher" else baseline - candidate if baseline: relative: float | None = absolute / abs(baseline) relative_unbounded = False elif absolute > 0: # Improvement from an exact zero baseline is positive but has no finite # percentage representation. Keep the JSON document strict and expose # that boundary explicitly instead of serializing Infinity. relative = None relative_unbounded = True else: relative = 0.0 if absolute == 0 else None relative_unbounded = False return { "absolute": absolute, "relative": relative, "relative_unbounded": relative_unbounded, } def _passes_relative_improvement( improvement: Mapping[str, float | bool | None], minimum: float ) -> bool: if improvement.get("relative_unbounded") is True: return True relative = improvement.get("relative") return ( isinstance(relative, (int, float)) and not isinstance(relative, bool) and relative >= minimum ) def _promotion_verdict( *, resolved: Mapping[str, Any], suite: Mapping[str, Any], variants: Sequence[Mapping[str, Any]], summaries: Mapping[str, Mapping[str, Any]], skipped: Sequence[Mapping[str, Any]], ) -> dict[str, Any]: is_lab = resolved["profile"]["maturity"] == "lab" config = suite["promotion"] checks: list[dict[str, Any]] = [] def finish() -> dict[str, Any]: hypothesis_passed = all(check["passed"] for check in checks) if is_lab: return { "eligible": False, "passed": None, "status": "experimental_lab", "reason": "labs report hypothesis checks but make no bundled superiority claim", "hypothesis_passed": hypothesis_passed, "checks": checks, } return {"eligible": True, "passed": hypothesis_passed, "checks": checks} def add(name: str, passed: bool, **evidence: Any) -> None: checks.append({"name": name, "passed": passed, **evidence}) by_class: dict[str, list[Mapping[str, Any]]] = {} for variant in variants: summary = summaries.get(str(variant["id"])) if summary is not None: by_class.setdefault(str(variant["comparison_class"]), []).append(summary) full_candidates = by_class.get("full_profile", []) root_candidates = by_class.get("configured_root_alone", []) if len(full_candidates) != 1 or len(root_candidates) != 1: add( "unique_full_and_configured_root", False, full_count=len(full_candidates), configured_root_count=len(root_candidates), ) return finish() full = full_candidates[0] root = root_candidates[0] add( "full_profile_has_success", float(full["success_rate"] or 0.0) > 0.0, full_success_rate=full["success_rate"], ) single_candidates = [ *by_class.get("strongest_single_agent", []), *by_class.get("access_service_single_agent", []), root, ] strongest = max(single_candidates, key=lambda item: float(item["success_rate"] or 0.0)) tolerance = float(config.get("strongest_success_tolerance", 0.02)) add( "strongest_single_success_tolerance", float(full["success_rate"] or 0.0) + tolerance >= float(strongest["success_rate"] or 0.0), full_success_rate=full["success_rate"], strongest_variant=strongest["id"], strongest_success_rate=strongest["success_rate"], tolerance=tolerance, ) metric = str(config["primary_metric"]) direction = str(config["direction"]) full_metric = full["outcome_metrics"].get(metric) baseline_class = str(config.get("primary_baseline", "configured_root_alone")) baseline_candidates = by_class.get(baseline_class, []) baseline = root if baseline_class == "configured_root_alone" else None if baseline is None and len(baseline_candidates) == 1: baseline = baseline_candidates[0] baseline_metric = baseline["outcome_metrics"].get(metric) if baseline else None if full_metric is None or baseline_metric is None: add( "primary_metric_improvement", False, metric=metric, baseline_class=baseline_class, reason="metric missing from full or primary-baseline validation output", ) else: assert baseline is not None improvement = _directed_improvement(float(baseline_metric), float(full_metric), direction) relative_gate = float(config.get("minimum_relative_improvement", 0.10)) absolute_gate = float(config.get("minimum_absolute_improvement", 0.05)) add( "primary_metric_improvement", _passes_relative_improvement(improvement, relative_gate) or float(improvement["absolute"] or 0.0) >= absolute_gate, metric=metric, direction=direction, baseline_class=baseline_class, baseline_variant=baseline["id"], baseline=baseline_metric, full_profile=full_metric, improvement=improvement, minimum_relative=relative_gate, minimum_absolute=absolute_gate, ) for field, expected in ( ("write_scope_conflicts", 0), ("contract_failures", 0), ("route_telemetry_incomplete", 0), ): add(f"no_{field}", int(full[field]) == expected, actual=full[field], expected=expected) if bool(config.get("require_complete_api_cost", True)): complete = bool(full["cost_ledgers"]["actual_api_usd_complete"]) add("complete_actual_api_cost", complete, ledger=full["cost_ledgers"]) for field, metric_direction in ( ("no_regression_higher_metrics", "higher"), ("no_regression_lower_metrics", "lower"), ): for name in config.get(field, []): candidate = full["outcome_metrics"].get(name) baseline = strongest["outcome_metrics"].get(name) if candidate is None or baseline is None: add( f"non_regression_{name}", False, reason="metric missing", full=candidate, strongest=baseline, ) else: passed = ( float(candidate) >= float(baseline) if metric_direction == "higher" else float(candidate) <= float(baseline) ) add( f"non_regression_{name}", passed, direction=metric_direction, full=candidate, strongest=baseline, ) root_id = str(resolved["profile"]["root"]) reachable = {root_id} frontier = [root_id] while frontier: parent = frontier.pop() for child in resolved["agents"][parent]["can_spawn"]: if child not in reachable: reachable.add(child) frontier.append(child) expected_workers = reachable - {root_id} ablations = {str(item["worker"]): item for item in by_class.get("ablation", [])} missing_ablations = sorted(expected_workers - set(ablations)) add( "complete_worker_ablation_coverage", not missing_ablations, expected_workers=sorted(expected_workers), missing=missing_ablations, ) for worker, ablation in sorted(ablations.items()): success_contribution = float(full["success_rate"] or 0.0) - float( ablation["success_rate"] or 0.0 ) ablated_metric = ablation["outcome_metrics"].get(metric) contribution = None if full_metric is not None and ablated_metric is not None: contribution = _directed_improvement( float(ablated_metric), float(full_metric), direction ) success_gate = float(config.get("worker_minimum_success_contribution", 0.02)) metric_gate = float(config.get("worker_minimum_metric_contribution", 0.10)) add( f"worker_ablation_{worker}", success_contribution >= success_gate or ( contribution is not None and _passes_relative_improvement(contribution, metric_gate) ), success_contribution=success_contribution, metric_contribution=contribution, minimum_success=success_gate, minimum_metric_relative=metric_gate, ) unavailable_access = [ item for item in skipped if item.get("comparison_class") == "access_service_single_agent" ] executed_access = by_class.get("access_service_single_agent", []) add( "accessible_service_controls_executed", bool(executed_access), executed=[item["id"] for item in executed_access], unavailable=unavailable_access, ) return finish() def _run_evaluation( *, profile: str | Path, suite: str | Path, bindings: Mapping[str, str] | None = None, wall_timeout_override: int | None = None, dry_run: bool = False, trial_mode: str = "development", progress: Callable[[str], None] | None = None, _failure_context: dict[str, Any], ) -> dict[str, Any]: if wall_timeout_override is not None and ( not isinstance(wall_timeout_override, int) or isinstance(wall_timeout_override, bool) or not 1 <= wall_timeout_override <= 172_800 ): raise ValueError("wall-timeout override must be an integer from 1 to 172800") if trial_mode not in {"development", "release"}: raise ValueError("trial_mode must be development or release") if progress: progress(f"Resolving evaluation suite {suite} and profile {profile}...") suite_dir, suite_data = resolve_suite(suite) configured_snapshot = compile_profile(profile, bindings=bindings) configured_resolved = configured_snapshot["resolved"] if configured_resolved["profile"]["id"] != suite_data["profile"]: raise ValueError( f"evaluation suite {suite_data['id']!r} requires profile " f"{suite_data['profile']!r}, not {configured_resolved['profile']['id']!r}" ) configured_agents = set(configured_resolved["agents"]) for task in suite_data["tasks"]: assertions = task.get("orchestration_assertions", {}) asserted_agents = set(assertions.get("required_agents", [])) | set( assertions.get("forbidden_agents", []) ) unknown_agents = sorted(asserted_agents - configured_agents) if unknown_agents: raise ValueError( f"task {task['id']}: orchestration assertion agents absent from profile: " + ", ".join(unknown_agents) ) configured_availability = _variant_root_availability(configured_snapshot) if not configured_availability["available"]: raise RuntimeError( f"configured root route {configured_availability['route']!r} is unavailable: " f"{configured_availability['reason']}" ) variant_entries = [] for variant in suite_data["variants"]: snapshot = _variant_snapshot(profile, bindings, variant) variant_entries.append((variant, snapshot, _variant_root_availability(snapshot))) known_variant_models = { model_key for _variant, snapshot, _availability in variant_entries for model_key in snapshot["resolved"]["models"] } unknown_scarce_models = sorted( set(suite_data["promotion"].get("scarce_model_keys", [])) - known_variant_models ) if unknown_scarce_models: raise ValueError( "evaluation promotion scarce_model_keys are absent from every matched variant: " + ", ".join(unknown_scarce_models) ) trial_count = int( suite_data["release_trials" if trial_mode == "release" else "development_trials"] ) run_id = _run_id(configured_resolved["profile"]["id"], suite_data["id"]) directory = evaluations_state_root() / run_id directory.mkdir(mode=0o700) fixture = suite_data.get("fixture") fixture_source = (suite_dir / str(fixture)).resolve() if fixture else None manifest: dict[str, Any] = { "schema_version": MMO_SCHEMA_VERSION, "run_id": run_id, "profile_id": configured_resolved["profile"]["id"], "profile_version": configured_resolved["profile"]["version"], "snapshot_hash": configured_snapshot["manifest"]["snapshot_hash"], "suite_id": suite_data["id"], "suite_name": suite_data.get("name", suite_data["id"]), "bindings": dict(bindings or {}), "trial_mode": trial_mode, "trial_count": trial_count, "promotion_policy": dict(suite_data["promotion"]), "variants": [ { **dict(variant), "snapshot_hash": variant_snapshot["manifest"]["snapshot_hash"], "resolved_profile_id": variant_snapshot["resolved"]["profile"]["id"], "root_route_availability": availability, } for variant, variant_snapshot, availability in variant_entries ], "skipped_variants": [ { "id": variant["id"], "comparison_class": variant["comparison_class"], "access_product": variant.get("access_product"), "reason": availability["reason"], "root_route": availability["route"], } for variant, _snapshot, availability in variant_entries if not availability["available"] ], "created_at": utc_now(), "status": "validated" if dry_run else "running", "tasks": [], } atomic_write_json(directory / "run.json", manifest) _failure_context.update(directory=directory, manifest=manifest) if dry_run: manifest["tasks"] = [ { "id": task["id"], "variant_id": variant["id"], "comparison_class": variant["comparison_class"], "trial": trial, "snapshot_hash": variant_snapshot["manifest"]["snapshot_hash"], "sandbox": task.get("sandbox", "read-only"), "images": list(task.get("images", [])), "route_faults": dict(task.get("route_faults", {})), "outcome_assertions": dict(task["outcome_assertions"]), "orchestration_assertions": dict(task.get("orchestration_assertions", {})), "validated": True, } for task in suite_data["tasks"] for variant, variant_snapshot, availability in variant_entries if availability["available"] for trial in range(1, trial_count + 1) ] manifest["finished_at"] = utc_now() atomic_write_json(directory / "run.json", manifest) if progress: progress(f"Validated {len(manifest['tasks'])} planned evaluation trial(s).") return manifest total_trials = sum( 1 for _task in suite_data["tasks"] for _variant, _snapshot, availability in variant_entries if availability["available"] for _trial in range(1, trial_count + 1) ) trial_index = 0 for task in suite_data["tasks"]: for variant, snapshot, availability in variant_entries: if not availability["available"]: continue for trial in range(1, trial_count + 1): trial_index += 1 if progress: progress( f"Evaluation trial {trial_index}/{total_trials}: " f"task={task['id']} variant={variant['id']} trial={trial}." ) resolved = snapshot["resolved"] task_dir = ( directory / "tasks" / task["id"] / str(variant["id"]) / f"trial-{trial:02d}" ) workspace = task_dir / "workspace" task_dir.mkdir(parents=True, exist_ok=True) _copy_fixture(fixture_source, workspace) mutation = _prepare_hidden_trial(suite_dir, task["id"], trial, workspace) _initialize_fixture_repository(workspace) timeout = wall_timeout_override or int(task.get("wall_timeout_seconds", 1800)) root_exception: str | None = None try: root_result = run_root_exec( profile=None, cwd=workspace, prompt=task["prompt"], images=task.get("images", []), snapshot_hash=snapshot["manifest"]["snapshot_hash"], wall_timeout_seconds=timeout, sandbox_mode=task.get("sandbox", "read-only"), label=( f"eval-{suite_data['id']}-{task['id']}-{variant['id']}-trial-{trial}" ), route_faults={ route: fault for route, fault in task.get("route_faults", {}).items() if route in snapshot["resolved"]["routes"] }, ) except Exception as exc: root_exception = f"{type(exc).__name__}: {exc}" failed_session_id = getattr(exc, "mmo_session_id", None) failed_session_status = getattr(exc, "mmo_session_status", None) root_result = { "status": ( "detached" if failed_session_status == "detached" else "failed_before_or_during_root_execution" ), "root_status": "failed_before_or_during_root_execution", "exit_code": 1, "elapsed_seconds": 0.0, "events_path": "", "result": "", } if isinstance(failed_session_id, str): root_result["session"] = {"session_id": failed_session_id} session_id = ( str(root_result["session"]["session_id"]) if isinstance(root_result.get("session"), Mapping) else "" ) # A harness wall limit detaches the durable app-server session; # it does not terminate it. Stop it before installing holdouts # or running validators so detached agents cannot keep mutating # the evaluation workspace behind the harness. if root_result.get("status") == "detached" and session_id: try: root_result["harness_cleanup"] = stop_session(session_id, grace_seconds=0) except Exception as stop_error: root_result["harness_cleanup_error"] = ( f"{type(stop_error).__name__}: {stop_error}" ) try: root_result["harness_cancel"] = cancel_session(session_id) except Exception as cancel_error: raise RuntimeError( "evaluation cannot install holdouts while detached agents may " "still be running; graceful stop and immediate cancellation " "both failed: " f"{type(stop_error).__name__}: {stop_error}; " f"{type(cancel_error).__name__}: {cancel_error}" ) from cancel_error (workspace / ".mmo-eval-final.txt").write_text( str(root_result.get("result") or ""), encoding="utf-8" ) patterns = _pattern_results(root_result.get("result", ""), task) holdout = _install_holdout(suite_dir, task["id"], workspace) validations = [ _run_validation( command, workspace, int(task.get("validation_timeout_seconds", 300)), ) for command in task["outcome_assertions"].get("validation_commands", []) ] outcome_metrics = _numeric_validation_metrics(validations) worker_metrics = _session_job_metrics(session_id, resolved) events_path = Path(str(root_result.get("events_path", ""))) root_usage = _event_usage(events_path) root_agent = resolved["agents"][resolved["profile"]["root"]] root_model = resolved["models"][root_agent["model"]] root_route = resolved["routes"][root_model["route"]] root_retries = _retry_count(events_path) root_elapsed = float(root_result.get("elapsed_seconds") or 0.0) root_cost_ledgers = ( _call_cost_ledgers( usage=root_usage, model=root_model, route=root_route, events_path=events_path, elapsed_seconds=root_elapsed, ) if root_exception is None else _merge_cost_ledgers([]) ) observed_wait = min( root_elapsed, float(worker_metrics.get("explicit_root_mcp_wait_seconds") or 0.0), ) root_activity = _root_activity_metrics(events_path, session_id) orchestration_assertions = _orchestration_assertion_results( task, worker_metrics, root_elapsed ) task_passed = ( root_result.get("exit_code") == 0 and patterns["passed"] and all(item["passed"] for item in validations) ) record = { "id": task["id"], "variant_id": variant["id"], "comparison_class": variant["comparison_class"], "topology": variant["topology"], "access_product": variant.get("access_product"), "trial": trial, "difficulty": task.get("difficulty", "medium"), "negative_control": bool(task.get("negative_control", False)), "route_faults": dict(task.get("route_faults", {})), "hidden_mutation": mutation, "holdout": holdout, "snapshot_hash": snapshot["manifest"]["snapshot_hash"], "resolved_profile_id": resolved["profile"]["id"], "description": task.get("description", ""), "workspace": str(workspace), "session_id": session_id, "root_status": root_result.get("status"), "root_error": root_exception, "root_exit_code": root_result.get("exit_code"), "root_elapsed_seconds": root_elapsed, "explicit_root_mcp_wait_seconds": observed_wait, "explicit_root_mcp_wait_ratio": ( observed_wait / root_elapsed if root_elapsed else 0.0 ), "root_activity": root_activity, "root_usage": root_usage, "root_cost_ledgers": root_cost_ledgers, "root_retries_observed": root_retries, "root_identity": { "model": root_agent["model"], "maker": root_model["maker"], "route": root_model["route"], "api_operator": root_route["api_operator"], "access_product": root_route["access_product"], "gateway_driver": root_route["driver"], }, "result": root_result.get("result", ""), "patterns": patterns, "validation": validations, "outcome_metrics": outcome_metrics, "worker_metrics": worker_metrics, "outcome_assertions": { "patterns": patterns, "validation": validations, "passed": patterns["passed"] and all(item["passed"] for item in validations), }, "orchestration_assertions": orchestration_assertions, "orchestration_diagnostics_passed": orchestration_assertions["passed"], "passed": task_passed, } atomic_write_json(task_dir / "result.json", record) manifest["tasks"].append(record) atomic_write_json(directory / "run.json", manifest) if progress: progress("Aggregating evaluation results and promotion checks...") variant_summaries = { str(variant["id"]): _variant_summary( variant, [record for record in manifest["tasks"] if record["variant_id"] == variant["id"]], scarce_model_keys=set(suite_data["promotion"].get("scarce_model_keys", [])), ) for variant, _snapshot, availability in variant_entries if availability["available"] } promotion = _promotion_verdict( resolved=configured_resolved, suite=suite_data, variants=suite_data["variants"], summaries=variant_summaries, skipped=manifest["skipped_variants"], ) manifest["summary"] = { "matched_task_count": len(suite_data["tasks"]), "executed_trial_records": len(manifest["tasks"]), "variant_summaries": variant_summaries, "promotion": promotion, "aggregate_score": None, "aggregate_score_reason": ( "comparison variants are controls, not interchangeable tasks; use per-variant " "outcomes and the promotion verdict" ), } manifest["status"] = ( "completed" if manifest["tasks"] and all(bool(record["passed"]) for record in manifest["tasks"]) else "completed_with_failures" ) manifest["finished_at"] = utc_now() atomic_write_json(directory / "run.json", manifest) return manifest def run_evaluation( *, profile: str | Path, suite: str | Path, bindings: Mapping[str, str] | None = None, wall_timeout_override: int | None = None, dry_run: bool = False, trial_mode: str = "development", progress: Callable[[str], None] | None = None, ) -> dict[str, Any]: failure_context: dict[str, Any] = {} try: return _run_evaluation( profile=profile, suite=suite, bindings=bindings, wall_timeout_override=wall_timeout_override, dry_run=dry_run, trial_mode=trial_mode, progress=progress, _failure_context=failure_context, ) except BaseException as exc: manifest = failure_context.get("manifest") directory = failure_context.get("directory") if isinstance(manifest, dict) and isinstance(directory, Path): manifest["status"] = "failed" manifest["finished_at"] = utc_now() manifest["error"] = f"{type(exc).__name__}: {exc}" with contextlib.suppress(Exception): atomic_write_json(directory / "run.json", manifest) raise def list_runs(limit: int = 100) -> list[dict[str, Any]]: if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1: raise ValueError("evaluation run limit must be a positive integer") results: list[dict[str, Any]] = [] for directory in sorted(evaluations_state_root().iterdir(), reverse=True): path = directory / "run.json" if not path.is_file(): continue with contextlib.suppress(Exception): data = read_json_object(path, label="evaluation run state") results.append( { "run_id": data.get("run_id"), "profile_id": data.get("profile_id"), "suite_id": data.get("suite_id"), "status": data.get("status"), "created_at": data.get("created_at"), "summary": data.get("summary"), } ) if len(results) >= limit: break return results def load_run(run_id: str) -> dict[str, Any]: if not isinstance(run_id, str) or RUN_ID_PATTERN.fullmatch(run_id) is None: raise ValueError(f"invalid evaluation run ID: {run_id!r}") path = evaluations_state_root() / run_id / "run.json" if not path.is_file(): raise FileNotFoundError(f"unknown evaluation run: {run_id}") return read_json_object(path, label="evaluation run state") def compare_runs(run_ids: Sequence[str]) -> dict[str, Any]: if len(run_ids) < 2: raise ValueError("compare requires at least two run IDs") runs = [load_run(item) for item in run_ids] suite_ids = {item.get("suite_id") for item in runs} warning = None if len(suite_ids) != 1: warning = "runs use different suites; scores are not directly comparable" rows: list[dict[str, Any]] = [] for run in runs: summary = run.get("summary") or {} variants = summary.get("variant_summaries") or {} full: Mapping[str, Any] = next( ( value for value in variants.values() if value.get("comparison_class") == "full_profile" ), {}, ) rows.append( { "run_id": run["run_id"], "profile_id": run["profile_id"], "suite_id": run["suite_id"], "status": run["status"], "promotion": summary.get("promotion"), "full_profile_success_rate": full.get("success_rate"), "full_profile_outcome_metrics": full.get("outcome_metrics"), "full_profile_mean_root_elapsed_seconds": full.get("mean_root_elapsed_seconds"), "full_profile_cost_ledgers": full.get("cost_ledgers"), "variant_summaries": variants, } ) ranked = sorted( rows, key=lambda item: ( -( float(item["full_profile_success_rate"]) if item["full_profile_success_rate"] is not None else -1.0 ), float(item["full_profile_mean_root_elapsed_seconds"] or float("inf")), ), ) return { "runs": rows, "ranked_by_full_profile_success_then_time": ranked, "warning": warning, "aggregate_score": None, "aggregate_score_reason": "controls and full profiles are not pooled into one score", }