-
+
- 基线解决率
- {percent(comparison.resolution.baseline_rate_percent)}
- 当前解决率
- {percent(comparison.resolution.current_rate_percent)}
- 差值
- {signed(comparison.resolution.delta_points, " pp")}
- 可比较任务
- {number(comparison.coverage.comparable_tasks)}
diff --git a/evaluation/README.md b/evaluation/README.md
index b7ff650ca..b78c1c941 100644
--- a/evaluation/README.md
+++ b/evaluation/README.md
@@ -1,7 +1,8 @@
# PowerContext evaluation console
-This directory contains a self-progressing SWE-bench Pro OFF/ON evaluation service. The web process owns the HTTP
-API and report UI; the worker owns task execution, retries, resource cleanup, and durable recovery.
+This directory contains a self-progressing SWE-bench Pro evaluation service. A batch can run the paired OFF/ON
+experiment or only one Arm to reduce cost and latency. The web process owns the HTTP API and report UI; the worker
+owns task execution, retries, resource cleanup, and durable recovery.
The service is intentionally deployment-neutral. Host names, operators, filesystem roots, optional proxy endpoints, Docker
network ranges, credentials, and service locations are supplied by the operator. The repository does not contain a
@@ -165,14 +166,17 @@ Verify the control plane before submitting work:
curl --fail --silent http://127.0.0.1:8787/api/health
```
-Open `http://127.0.0.1:8787/`, choose `swebench-pro-stability-v1`, preview the request, and confirm it. This 24-task
-set is the deployment regression suite; use it before the 731-task `swebench-pro-public-v2` set. The same bounded
-batch can be created from the CLI after Web and Worker are healthy:
+Open `http://127.0.0.1:8787/`, choose `swebench-pro-stability-v1` and an `OFF + ON`, `ON only`, or `OFF only` run.
+This 24-task set is the deployment regression suite; use it before the 731-task `swebench-pro-public-v2` set. A
+completed aggregate report can freeze any executed Arm as an immutable baseline. Reports may select multiple
+compatible baselines for historical comparison without rerunning evaluation, while the baseline library lists the
+newest saved baselines first. The same bounded batch can be created from the CLI after Web and Worker are healthy:
```bash
uv run --project evaluation powercontext-eval swebench-pro create-batch \
--console-url http://127.0.0.1:8787 \
--task-set swebench-pro-stability-v1 \
+ --treatment-mode on_only \
--powercontext-ref latest \
--idempotency-key "stability-$(date -u +%Y%m%dT%H%M%SZ)"
```
diff --git a/evaluation/src/powercontext_eval/cli.py b/evaluation/src/powercontext_eval/cli.py
index 7d94dc5da..ee1eedee3 100644
--- a/evaluation/src/powercontext_eval/cli.py
+++ b/evaluation/src/powercontext_eval/cli.py
@@ -34,6 +34,7 @@
from powercontext_eval.benchmarks.swebench_pro.catalog import PUBLIC_V2_TASK_SET, SweBenchProCatalog, TaskSet
from powercontext_eval.codex import DEFAULT_CODEX_MODEL, DEFAULT_REASONING_EFFORT
+from powercontext_eval.models import TreatmentMode
from powercontext_eval.powercontext_sut import DEFAULT_DOCKER_NETWORK_POOL, run_codex_contract_smoke
from powercontext_eval.runner import RunConfig, run_swebench_pro_instance
from powercontext_eval.web.batches import BatchCreate
@@ -258,6 +259,7 @@ def swebench_pro_create_batch(
powercontext_ref: str = typer.Option("latest", "--powercontext-ref"),
task_set: str = typer.Option(PUBLIC_V2_TASK_SET, "--task-set"),
model: str = typer.Option(DEFAULT_CODEX_MODEL, "--model"),
+ treatment_mode: TreatmentMode = typer.Option(TreatmentMode.OFF_ON, "--treatment-mode"),
usage_pause_percent: int = typer.Option(80, "--usage-pause-percent", min=1, max=100),
start_paused: bool = typer.Option(False, "--start-paused/--start-running"),
) -> None:
@@ -271,7 +273,7 @@ def swebench_pro_create_batch(
task_set=cast(TaskSet, task_set),
model=model,
reasoning_effort=DEFAULT_REASONING_EFFORT,
- treatment_mode="off_on",
+ treatment_mode=treatment_mode,
idempotency_key=idempotency_key,
usage_pause_percent=usage_pause_percent,
initial_control_intent="pause" if start_paused else "run",
diff --git a/evaluation/src/powercontext_eval/models.py b/evaluation/src/powercontext_eval/models.py
index 587b5e846..da476af90 100644
--- a/evaluation/src/powercontext_eval/models.py
+++ b/evaluation/src/powercontext_eval/models.py
@@ -28,6 +28,22 @@ class Arm(StrEnum):
ON = "on"
+class TreatmentMode(StrEnum):
+ """The exact treatment arms executed for one evaluation task."""
+
+ OFF_ON = "off_on"
+ ON_ONLY = "on_only"
+ OFF_ONLY = "off_only"
+
+ @property
+ def arms(self) -> tuple[Arm, ...]:
+ if self is TreatmentMode.OFF_ON:
+ return (Arm.OFF, Arm.ON)
+ if self is TreatmentMode.ON_ONLY:
+ return (Arm.ON,)
+ return (Arm.OFF,)
+
+
class PowerContextRef(BaseModel):
"""An explicit, immutable PowerContext source reference."""
diff --git a/evaluation/src/powercontext_eval/powercontext_sut.py b/evaluation/src/powercontext_eval/powercontext_sut.py
index 6c4972aa1..239ea544a 100644
--- a/evaluation/src/powercontext_eval/powercontext_sut.py
+++ b/evaluation/src/powercontext_eval/powercontext_sut.py
@@ -46,7 +46,7 @@
import time
from collections.abc import Callable, Iterator, Mapping, Sequence
from contextlib import contextmanager
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, field, replace
from datetime import UTC, datetime
from enum import StrEnum
from pathlib import Path
@@ -519,10 +519,9 @@ class SutConfig:
reasoning_effort: str = DEFAULT_REASONING_EFFORT
recorder_script: Path = _DEFAULT_RECORDER_SCRIPT
limits: ContainerLimits = ContainerLimits()
- plugin_version: str = "0.1.0"
codex_timeout: float = 3600
finalization_registrar: TokensFlowFinalizationRegistrar | None = None
- container_env: Mapping[str, str] = MappingProxyType({})
+ container_env: Mapping[str, str] = field(default_factory=lambda: MappingProxyType({}))
def __post_init__(self) -> None:
if _SAFE_RUN_ID.fullmatch(self.run_id) is None:
@@ -866,8 +865,6 @@ def _verify_source(self, config: SutConfig) -> SourceProvenance:
raise TypeError
except (OSError, json.JSONDecodeError, KeyError, TypeError) as error:
raise InvalidTreatment("PowerContext plugin manifest is invalid") from error
- if version != config.plugin_version:
- raise InvalidTreatment("PowerContext plugin manifest version does not match configuration")
lockfile = config.source_checkout / _PLUGIN_RELATIVE / "uv.lock"
try:
metadata = lockfile.stat(follow_symlinks=False)
diff --git a/evaluation/src/powercontext_eval/report.py b/evaluation/src/powercontext_eval/report.py
index fa4126735..09f3a6d42 100644
--- a/evaluation/src/powercontext_eval/report.py
+++ b/evaluation/src/powercontext_eval/report.py
@@ -39,6 +39,7 @@
SOURCE595_DATASET_PATCH_SHA256,
SOURCE595_INSTANCE_ID,
)
+from powercontext_eval.models import Arm, TreatmentMode
class MetricSet(BaseModel):
@@ -190,10 +191,27 @@ class ReportBundle(BaseModel):
title: str
revisions: Mapping[str, str]
configuration: Mapping[str, str]
- off: ArmReport
- on: ArmReport
+ treatment_mode: TreatmentMode = TreatmentMode.OFF_ON
+ off: ArmReport | None = None
+ on: ArmReport | None = None
gold_validation: GoldValidationAudit | None = None
+ @field_validator("treatment_mode", mode="before")
+ @classmethod
+ def parse_treatment_mode(cls, value: object) -> object:
+ return TreatmentMode(value) if isinstance(value, str) else value
+
+ @model_validator(mode="after")
+ def validate_arms(self) -> Self:
+ present = {arm for arm, report in ((Arm.OFF, self.off), (Arm.ON, self.on)) if report is not None}
+ if present != set(self.treatment_mode.arms):
+ raise ValueError("Report arms do not match the treatment mode")
+ if self.off is not None and self.off.arm != "off":
+ raise ValueError("OFF report is bound to the wrong arm")
+ if self.on is not None and self.on.arm != "on":
+ raise ValueError("ON report is bound to the wrong arm")
+ return self
+
@model_validator(mode="after")
def validate_gold_binding(self) -> Self:
instance_id = self.configuration.get("instance")
@@ -332,6 +350,8 @@ def _gold_validation_section(audit: GoldValidationAudit) -> list[str]:
def _comparison(bundle: ReportBundle) -> list[str]:
lines = ["## Comparison", ""]
+ if bundle.off is None or bundle.on is None:
+ return lines + ["Comparison unavailable: this run contains one treatment arm.", ""]
comparable_states = {ArmState.TREATMENT_VALIDATED, ArmState.REPORTED}
if not (
bundle.off.state in comparable_states
@@ -374,10 +394,9 @@ def _validated_bundle(bundle: ReportBundle) -> ReportBundle:
expected_bundle_fields = set(ReportBundle.model_fields)
if set(bundle.__dict__) != expected_bundle_fields:
raise ValueError
- for model, model_type in (
- (bundle.off, ArmReport),
- (bundle.on, ArmReport),
- ):
+ for model, model_type in ((bundle.off, ArmReport), (bundle.on, ArmReport)):
+ if model is None:
+ continue
if type(model) is not model_type or set(model.__dict__) != set(model_type.model_fields):
raise ValueError
if type(model.metrics) is not MetricSet or set(model.metrics.__dict__) != set(MetricSet.model_fields):
@@ -400,14 +419,15 @@ def render_report(bundle: ReportBundle) -> str:
"""Render only the supplied validated bundle, with no external reads."""
bundle = _validated_bundle(bundle)
- if bundle.off.arm != "off" or bundle.on.arm != "on":
- raise ValueError("Report arms must be supplied in OFF then ON roles")
lines = [f"# {_cell(bundle.title)}", ""]
lines.extend(_mapping_table("Resolved revisions", bundle.revisions))
lines.extend(_mapping_table("Configuration", bundle.configuration))
if bundle.gold_validation is not None:
lines.extend(_gold_validation_section(bundle.gold_validation))
- lines.extend(_arm_section("OFF", bundle.off))
- lines.extend(_arm_section("ON", bundle.on))
- lines.extend(_comparison(bundle))
+ if bundle.off is not None:
+ lines.extend(_arm_section("OFF", bundle.off))
+ if bundle.on is not None:
+ lines.extend(_arm_section("ON", bundle.on))
+ if bundle.off is not None and bundle.on is not None:
+ lines.extend(_comparison(bundle))
return "\n".join(lines)
diff --git a/evaluation/src/powercontext_eval/runner.py b/evaluation/src/powercontext_eval/runner.py
index b9594fd7e..0d939f276 100644
--- a/evaluation/src/powercontext_eval/runner.py
+++ b/evaluation/src/powercontext_eval/runner.py
@@ -49,7 +49,7 @@
from powercontext_eval.context_trace import write_context_trace
from powercontext_eval.errors import CommandFailed
from powercontext_eval.git_source import GitSource
-from powercontext_eval.models import Arm, PowerContextRef
+from powercontext_eval.models import Arm, PowerContextRef, TreatmentMode
from powercontext_eval.paths import EvaluationPaths
from powercontext_eval.powercontext_sut import (
DEFAULT_DOCKER_NETWORK_POOL,
@@ -152,6 +152,7 @@ class RunConfig:
registry_binary: Path
auth_json: Path
run_id: str
+ treatment_mode: TreatmentMode = TreatmentMode.OFF_ON
tokensflow_enabled: bool = False
tokensflow_binary: Path | None = None
tokensflow_user_home: Path | None = None
@@ -163,10 +164,12 @@ class RunConfig:
model: str = DEFAULT_CODEX_MODEL
reasoning_effort: str = DEFAULT_REASONING_EFFORT
finalization_registrar: TokensFlowFinalizationRegistrar | None = None
- container_env: Mapping[str, str] = MappingProxyType({})
+ container_env: Mapping[str, str] = field(default_factory=lambda: MappingProxyType({}))
cancel_event: threading.Event | None = field(default=None, repr=False, compare=False)
def __post_init__(self) -> None:
+ if isinstance(self.treatment_mode, str):
+ object.__setattr__(self, "treatment_mode", TreatmentMode(self.treatment_mode))
if not is_safe_codex_model(self.model):
raise ValueError("Codex model is unsafe")
if self.reasoning_effort != DEFAULT_REASONING_EFFORT:
@@ -188,8 +191,8 @@ class RunResult:
run_id: str
report_path: Path
- off_resolved: bool
- on_resolved: bool
+ off_resolved: bool | None
+ on_resolved: bool | None
# Public compatibility name retained while the web task schema migrates from a single task to batches.
@@ -352,11 +355,13 @@ def _run_swebench_pro_instance(
)
run_store.write_json("gold/validation.json", gold_audit.model_dump(mode="json"))
- def arms() -> tuple[OfficialEvaluation, OfficialEvaluation, Mapping[Arm, SutOutcome], dict[Arm, int]]:
+ selected_arms = config.treatment_mode.arms
+
+ def arms() -> tuple[dict[Arm, OfficialEvaluation], Mapping[Arm, SutOutcome], dict[Arm, int]]:
codex_secrets = auth_secret_variants(config.auth_json)
arm_paths: dict[Arm, ArmPaths] = {}
stores: dict[Arm, ArtifactStore] = {}
- for arm in (Arm.OFF, Arm.ON):
+ for arm in selected_arms:
arm_work = layout.arm_work(arm)
runtime = arm_work / "runtime"
root_home = runtime / "root-home"
@@ -387,34 +392,43 @@ def arms() -> tuple[OfficialEvaluation, OfficialEvaluation, Mapping[Arm, SutOutc
secrets += tokensflow_secret_variants(tokensflow_credentials)
stores[arm] = ArtifactStore(layout.arm_artifacts(arm), forbidden_values=secrets)
prompt = instance.codex_prompt().encode()
- outcomes = DockerSut(process).run_pair(
- SutConfig(
- run_id=run_id,
- task_image=task_image_id,
- codex_binary=config.codex_binary,
- uv_binary=config.uv_binary,
- source_checkout=materialized,
- plugin_checkout_sha=resolved.sha,
- proxy=ProxyRelayConfig(config.proxy_url) if config.proxy_url is not None else None,
- docker_network_pool=config.docker_network_pool,
- extra_no_proxy_hosts=config.extra_no_proxy_hosts,
- tokensflow_enabled=config.tokensflow_enabled,
- tokensflow_binary=config.tokensflow_binary,
- tokensflow_egress_network=config.tokensflow_egress_network,
- model=config.model,
- reasoning_effort=config.reasoning_effort,
- finalization_registrar=config.finalization_registrar,
- container_env=config.container_env,
- ),
- paths=arm_paths,
- prompts={Arm.OFF: prompt, Arm.ON: prompt},
- stores=stores,
- before_arm=lambda arm: emit_phase(RunPhase.RUNNING_OFF if arm is Arm.OFF else RunPhase.RUNNING_ON),
+ sut_config = SutConfig(
+ run_id=run_id,
+ task_image=task_image_id,
+ codex_binary=config.codex_binary,
+ uv_binary=config.uv_binary,
+ source_checkout=materialized,
+ plugin_checkout_sha=resolved.sha,
+ proxy=ProxyRelayConfig(config.proxy_url) if config.proxy_url is not None else None,
+ docker_network_pool=config.docker_network_pool,
+ extra_no_proxy_hosts=config.extra_no_proxy_hosts,
+ tokensflow_enabled=config.tokensflow_enabled,
+ tokensflow_binary=config.tokensflow_binary,
+ tokensflow_egress_network=config.tokensflow_egress_network,
+ model=config.model,
+ reasoning_effort=config.reasoning_effort,
+ finalization_registrar=config.finalization_registrar,
+ container_env=config.container_env,
)
+ sut = DockerSut(process)
+ if config.treatment_mode is TreatmentMode.OFF_ON:
+ outcomes = sut.run_pair(
+ sut_config,
+ paths=arm_paths,
+ prompts={arm: prompt for arm in selected_arms},
+ stores=stores,
+ before_arm=lambda arm: emit_phase(RunPhase.RUNNING_OFF if arm is Arm.OFF else RunPhase.RUNNING_ON),
+ )
+ else:
+ arm = selected_arms[0]
+ emit_phase(RunPhase.RUNNING_OFF if arm is Arm.OFF else RunPhase.RUNNING_ON)
+ outcomes = {
+ arm: sut.run_arm(sut_config, arm, arm_paths[arm], prompt, stores[arm]),
+ }
official: dict[Arm, OfficialEvaluation] = {}
patch_sizes: dict[Arm, int] = {}
emit_phase(RunPhase.OFFICIAL_EVALUATION)
- for arm in (Arm.OFF, Arm.ON):
+ for arm in selected_arms:
patch = process.run(
("git", "diff", "--binary", "--full-index", instance.base_commit, "--"),
cwd=arm_paths[arm].workspace,
@@ -451,17 +465,20 @@ def arms() -> tuple[OfficialEvaluation, OfficialEvaluation, Mapping[Arm, SutOutc
official=official[arm],
official_observed_at=datetime.now(UTC),
)
- return official[Arm.OFF], official[Arm.ON], outcomes, patch_sizes
+ return official, outcomes, patch_sizes
- off_eval, on_eval, outcomes, patch_sizes = run_after_gold(
+ official, outcomes, patch_sizes = run_after_gold(
GoldResult(instance.instance_id, gold.resolved),
arms,
)
- off_outcome = outcomes[Arm.OFF]
- on_outcome = outcomes[Arm.ON]
emit_phase(RunPhase.GENERATING_REPORT)
+ arm_reports = {arm: _arm_report(arm, official[arm], outcomes[arm], patch_sizes[arm]) for arm in selected_arms}
report = ReportBundle(
- title="PowerContext Codex SWE-bench Pro comparison",
+ title=(
+ "PowerContext Codex SWE-bench Pro comparison"
+ if config.treatment_mode is TreatmentMode.OFF_ON
+ else f"PowerContext Codex SWE-bench Pro {selected_arms[0].value.upper()} run"
+ ),
revisions={
"dataset": DATASET_REVISION,
"harness": HARNESS_COMMIT,
@@ -476,9 +493,11 @@ def arms() -> tuple[OfficialEvaluation, OfficialEvaluation, Mapping[Arm, SutOutc
"extra_no_proxy_hosts": ",".join(config.extra_no_proxy_hosts),
"proxy": "enabled" if config.proxy_url is not None else "disabled",
"tokensflow": "enabled" if config.tokensflow_enabled else "disabled",
+ "treatment_mode": config.treatment_mode.value,
},
- off=_arm_report(Arm.OFF, off_eval, off_outcome, patch_sizes[Arm.OFF]),
- on=_arm_report(Arm.ON, on_eval, on_outcome, patch_sizes[Arm.ON]),
+ treatment_mode=config.treatment_mode,
+ off=arm_reports.get(Arm.OFF),
+ on=arm_reports.get(Arm.ON),
gold_validation=gold_audit,
)
rendered = render_report(report)
@@ -486,7 +505,12 @@ def arms() -> tuple[OfficialEvaluation, OfficialEvaluation, Mapping[Arm, SutOutc
raise RuntimeError("Report rendering is not deterministic")
report_path = run_store.create_text("report.md", rendered)
run_store.create_json("report.json", report.model_dump(mode="json"))
- return RunResult(run_id, report_path, off_eval.resolved, on_eval.resolved)
+ return RunResult(
+ run_id,
+ report_path,
+ official[Arm.OFF].resolved if Arm.OFF in official else None,
+ official[Arm.ON].resolved if Arm.ON in official else None,
+ )
def _evaluator_test_requirements(
diff --git a/evaluation/src/powercontext_eval/web/api.py b/evaluation/src/powercontext_eval/web/api.py
index 7bf526658..4b331cb9d 100644
--- a/evaluation/src/powercontext_eval/web/api.py
+++ b/evaluation/src/powercontext_eval/web/api.py
@@ -42,7 +42,14 @@
from powercontext_eval.codex import DEFAULT_REASONING_EFFORT
from powercontext_eval.errors import GitSourceError
from powercontext_eval.git_source import GitSource
-from powercontext_eval.models import PowerContextRef
+from powercontext_eval.models import Arm, PowerContextRef
+from powercontext_eval.web.baselines import (
+ BaselineCandidate,
+ BaselineComparisonResponse,
+ BaselineCreate,
+ BaselineSelectionUpdate,
+ CompatibilityStatus,
+)
from powercontext_eval.web.batches import (
BatchCreate,
BatchPreviewResponse,
@@ -61,7 +68,11 @@
BenchmarkCatalog,
InvalidReportArtifact,
ReportingError,
+ StaleReportRevision,
UnsafeReportPath,
+ baseline_compatibility,
+ compare_batch_to_baseline,
+ create_baseline_snapshot,
load_batch_estimate_samples,
load_batch_report,
load_batch_task_detail,
@@ -74,7 +85,14 @@
)
from powercontext_eval.web.resources import FilesystemResourceProbe, ResourceProbe, ResourceUnavailable
from powercontext_eval.web.revision import RUNTIME_SCHEMA_VERSION, current_build_revision
-from powercontext_eval.web.store import BatchNotFound, TaskAdmissionRejected, TaskConflict, TaskNotFound, TaskStore
+from powercontext_eval.web.store import (
+ BaselineNotFound,
+ BatchNotFound,
+ TaskAdmissionRejected,
+ TaskConflict,
+ TaskNotFound,
+ TaskStore,
+)
from powercontext_eval.web.usage import AccountUsage, UsageSnapshot, is_fresh
_TERMINAL = {TaskStatus.SUCCEEDED, TaskStatus.FAILED, TaskStatus.INTERRUPTED, TaskStatus.CANCELLED}
@@ -403,7 +421,7 @@ def historical_estimate(request: BatchPreviewRequest, *, total_tasks: int) -> Ba
or candidate.task_set != request.task_set
or candidate.model != request.model
or candidate.reasoning_effort != DEFAULT_REASONING_EFFORT
- or candidate.treatment_mode != "off_on"
+ or candidate.treatment_mode != request.treatment_mode
):
continue
try:
@@ -516,7 +534,7 @@ def preview_batch(request: BatchPreviewRequest) -> Response:
task_set=request.task_set,
model=request.model,
reasoning_effort=DEFAULT_REASONING_EFFORT,
- treatment_mode="off_on",
+ treatment_mode=request.treatment_mode,
total_tasks=total_tasks,
usage_pause_percent=request.usage_pause_percent,
usage=snapshot,
@@ -577,6 +595,171 @@ def list_batches() -> Response:
headers=_NO_STORE,
)
+ @app.post("/api/baselines")
+ def create_baseline(request: BaselineCreate) -> Response:
+ try:
+ batch = task_store.get_batch(request.source_batch_id)
+ tasks = task_store.list_batch_tasks(request.source_batch_id)
+ snapshot = create_baseline_snapshot(
+ batch,
+ tasks,
+ arm=request.source_arm,
+ expected_report_revision=request.expected_report_revision,
+ runs_root=config.run_root / "runs",
+ catalog=get_catalog(),
+ )
+ baseline, created = task_store.create_baseline(request, snapshot, now=datetime.now(UTC))
+ except BatchNotFound:
+ return _error(404, "batch_not_found", "The requested evaluation batch does not exist.")
+ except StaleReportRevision:
+ return _error(409, "report_revision_conflict", "The batch report changed before the baseline was saved.")
+ except TaskConflict:
+ return _error(409, "idempotency_conflict", "The idempotency key belongs to a different request.")
+ except (CatalogError, ReportingError, OSError, ValueError):
+ return _error(409, "baseline_unavailable", "The selected batch arm cannot be saved as a baseline.")
+ return JSONResponse(
+ status_code=201 if created else 200,
+ content=baseline.model_dump(mode="json"),
+ headers=_NO_STORE,
+ )
+
+ @app.get("/api/baselines")
+ def list_baselines() -> Response:
+ return JSONResponse(
+ content=[baseline.model_dump(mode="json") for baseline in task_store.list_baselines()],
+ headers=_NO_STORE,
+ )
+
+ @app.get("/api/baselines/{baseline_id}")
+ def get_baseline(baseline_id: str) -> Response:
+ try:
+ baseline = task_store.get_baseline(baseline_id)
+ except BaselineNotFound:
+ return _error(404, "baseline_not_found", "The requested baseline does not exist.")
+ return JSONResponse(content=baseline.model_dump(mode="json"), headers=_NO_STORE)
+
+ @app.get("/api/batches/{batch_id}/baseline-candidates")
+ def baseline_candidates(batch_id: str, current_arm: Arm) -> Response:
+ selected = batch_inputs(batch_id)
+ if isinstance(selected, JSONResponse):
+ return selected
+ batch, tasks = selected
+ try:
+ report = load_batch_report(
+ batch,
+ tasks,
+ runs_root=config.run_root / "runs",
+ catalog=get_catalog(),
+ )
+ candidates = [
+ BaselineCandidate(
+ baseline=baseline,
+ compatibility=baseline_compatibility(
+ baseline,
+ batch,
+ tasks,
+ report,
+ current_arm=current_arm,
+ ),
+ )
+ for baseline in task_store.list_baselines()
+ ]
+ except (CatalogError, ReportingError, OSError, ValueError):
+ return _error(409, "report_unavailable", "The batch report is not available.")
+ return JSONResponse(
+ content=[candidate.model_dump(mode="json") for candidate in candidates],
+ headers=_NO_STORE,
+ )
+
+ @app.get("/api/batches/{batch_id}/baseline-selections")
+ def baseline_selections(batch_id: str) -> Response:
+ try:
+ selections = task_store.list_baseline_selections(batch_id)
+ except BatchNotFound:
+ return _error(404, "batch_not_found", "The requested evaluation batch does not exist.")
+ return JSONResponse(
+ content=[selection.model_dump(mode="json") for selection in selections],
+ headers=_NO_STORE,
+ )
+
+ @app.put("/api/batches/{batch_id}/baseline-selections")
+ def update_baseline_selections(batch_id: str, request: BaselineSelectionUpdate) -> Response:
+ selected = batch_inputs(batch_id)
+ if isinstance(selected, JSONResponse):
+ return selected
+ batch, tasks = selected
+ try:
+ report = load_batch_report(
+ batch,
+ tasks,
+ runs_root=config.run_root / "runs",
+ catalog=get_catalog(),
+ )
+ for selection in request.selections:
+ baseline = task_store.get_baseline(selection.baseline_id)
+ compatibility = baseline_compatibility(
+ baseline,
+ batch,
+ tasks,
+ report,
+ current_arm=selection.current_arm,
+ )
+ if compatibility.status is CompatibilityStatus.INCOMPATIBLE:
+ return _error(409, "baseline_incompatible", "The selected baseline is not compatible.")
+ selections = task_store.replace_baseline_selections(
+ batch_id,
+ request.selections,
+ now=datetime.now(UTC),
+ )
+ except BaselineNotFound:
+ return _error(404, "baseline_not_found", "The requested baseline does not exist.")
+ except TaskConflict:
+ return _error(409, "baseline_selection_conflict", "The baseline selection is invalid.")
+ except (CatalogError, ReportingError, OSError, ValueError):
+ return _error(409, "report_unavailable", "The batch report is not available.")
+ return JSONResponse(
+ content=[selection.model_dump(mode="json") for selection in selections],
+ headers=_NO_STORE,
+ )
+
+ @app.get("/api/batches/{batch_id}/baseline-comparisons")
+ def baseline_comparisons(batch_id: str) -> Response:
+ selected = batch_inputs(batch_id)
+ if isinstance(selected, JSONResponse):
+ return selected
+ batch, tasks = selected
+ try:
+ report = load_batch_report(
+ batch,
+ tasks,
+ runs_root=config.run_root / "runs",
+ catalog=get_catalog(),
+ )
+ comparisons = []
+ for selection in task_store.list_baseline_selections(batch_id):
+ baseline = task_store.get_baseline(selection.baseline_id)
+ comparisons.append(
+ compare_batch_to_baseline(
+ batch,
+ tasks,
+ report,
+ baseline,
+ task_store.list_baseline_items(baseline.baseline_id),
+ current_arm=selection.current_arm,
+ runs_root=config.run_root / "runs",
+ )
+ )
+ response = BaselineComparisonResponse(
+ batch_id=batch_id,
+ report_revision=report.report_revision,
+ comparisons=tuple(comparisons),
+ )
+ except BaselineNotFound:
+ return _error(409, "baseline_unavailable", "A selected baseline no longer exists.")
+ except (CatalogError, ReportingError, OSError, ValueError):
+ return _error(409, "report_unavailable", "The batch comparison is not available.")
+ return JSONResponse(content=response.model_dump(mode="json"), headers=_NO_STORE)
+
@app.get("/api/batches/{batch_id}")
def get_batch(batch_id: str) -> Response:
try:
diff --git a/evaluation/src/powercontext_eval/web/baselines.py b/evaluation/src/powercontext_eval/web/baselines.py
new file mode 100644
index 000000000..8e3b336bb
--- /dev/null
+++ b/evaluation/src/powercontext_eval/web/baselines.py
@@ -0,0 +1,229 @@
+# Copyright (c) 2026 OceanBase.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Immutable single-arm baseline and historical-comparison contracts."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from powercontext_eval.benchmarks.swebench_pro.catalog import TaskSet
+from powercontext_eval.models import Arm
+from powercontext_eval.web.models import TaskStatus
+
+
+class _FrozenModel(BaseModel):
+ model_config = ConfigDict(extra="forbid", strict=True, frozen=True)
+
+
+def _require_utc(value: datetime) -> datetime:
+ if value.tzinfo is None or value.utcoffset() != UTC.utcoffset(value):
+ raise ValueError("Timestamps must use UTC")
+ return value
+
+
+class BaselineCreate(_FrozenModel):
+ name: str = Field(min_length=1, max_length=120)
+ source_batch_id: str = Field(min_length=1, max_length=200)
+ source_arm: Arm
+ expected_report_revision: Annotated[int, Field(ge=0)]
+ idempotency_key: str = Field(min_length=8, max_length=128, pattern=r"^[A-Za-z0-9._-]+$")
+
+ @field_validator("name")
+ @classmethod
+ def normalize_name(cls, value: str) -> str:
+ normalized = " ".join(value.split())
+ if not normalized:
+ raise ValueError("Baseline name must not be blank")
+ return normalized
+
+ @field_validator("source_arm", mode="before")
+ @classmethod
+ def parse_arm(cls, value: object) -> object:
+ return Arm(value) if isinstance(value, str) else value
+
+
+class BaselineRecord(_FrozenModel):
+ baseline_id: str
+ name: str
+ source_batch_id: str
+ source_arm: Arm
+ source_report_revision: Annotated[int, Field(ge=0)]
+ benchmark: Literal["swebench-pro"]
+ task_set: TaskSet
+ instance_set_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
+ total_tasks: Annotated[int, Field(ge=1)]
+ resolved_tasks: Annotated[int, Field(ge=0)]
+ execution_failures: Annotated[int, Field(ge=0)]
+ model: str
+ reasoning_effort: Literal["medium"]
+ dataset_revision: str
+ harness_revision: str
+ powercontext_sha: str | None = Field(default=None, pattern=r"^[0-9a-f]{40}$")
+ codex_version: str | None = None
+ created_at: datetime
+
+ _created_at_utc = field_validator("created_at")(_require_utc)
+
+ @field_validator("source_arm", mode="before")
+ @classmethod
+ def parse_arm(cls, value: object) -> object:
+ return Arm(value) if isinstance(value, str) else value
+
+
+class BaselineItemRecord(_FrozenModel):
+ baseline_id: str
+ instance_id: str
+ source_index: Annotated[int, Field(ge=0)]
+ source_task_id: str
+ source_attempt_id: str | None = None
+ status: TaskStatus
+ resolved: bool | None = None
+ input_tokens: Annotated[int, Field(ge=0)] | None = None
+ output_tokens: Annotated[int, Field(ge=0)] | None = None
+ total_tokens: Annotated[int, Field(ge=0)] | None = None
+
+ @model_validator(mode="after")
+ def validate_outcome(self) -> BaselineItemRecord:
+ if self.status is TaskStatus.SUCCEEDED:
+ if self.source_attempt_id is None or self.resolved is None:
+ raise ValueError("Successful baseline items require an exact attempt and result")
+ elif any(
+ value is not None for value in (self.resolved, self.input_tokens, self.output_tokens, self.total_tokens)
+ ):
+ raise ValueError("Unsuccessful baseline items cannot contain arm outcomes")
+ if (
+ self.input_tokens is not None
+ and self.output_tokens is not None
+ and self.total_tokens != self.input_tokens + self.output_tokens
+ ):
+ raise ValueError("Baseline token totals are inconsistent")
+ return self
+
+
+class BaselineSnapshot(_FrozenModel):
+ benchmark: Literal["swebench-pro"]
+ task_set: TaskSet
+ instance_set_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
+ total_tasks: Annotated[int, Field(ge=1)]
+ resolved_tasks: Annotated[int, Field(ge=0)]
+ execution_failures: Annotated[int, Field(ge=0)]
+ model: str
+ reasoning_effort: Literal["medium"]
+ dataset_revision: str
+ harness_revision: str
+ powercontext_sha: str | None = Field(default=None, pattern=r"^[0-9a-f]{40}$")
+ codex_version: str | None = None
+ items: tuple[BaselineItemRecord, ...]
+
+
+class CompatibilityStatus(StrEnum):
+ COMPATIBLE = "compatible"
+ WARNING = "warning"
+ INCOMPATIBLE = "incompatible"
+
+
+class BaselineCompatibility(_FrozenModel):
+ status: CompatibilityStatus
+ reasons: tuple[str, ...] = ()
+
+
+class BaselineCandidate(_FrozenModel):
+ baseline: BaselineRecord
+ compatibility: BaselineCompatibility
+
+
+class BaselineSelection(_FrozenModel):
+ baseline_id: str
+ current_arm: Arm
+
+ @field_validator("current_arm", mode="before")
+ @classmethod
+ def parse_arm(cls, value: object) -> object:
+ return Arm(value) if isinstance(value, str) else value
+
+
+class BaselineSelectionUpdate(_FrozenModel):
+ selections: tuple[BaselineSelection, ...] = Field(max_length=10)
+
+ @field_validator("selections", mode="before")
+ @classmethod
+ def parse_json_array(cls, value: object) -> object:
+ return tuple(value) if isinstance(value, list) else value
+
+ @model_validator(mode="after")
+ def unique_selections(self) -> BaselineSelectionUpdate:
+ keys = {(selection.baseline_id, selection.current_arm) for selection in self.selections}
+ if len(keys) != len(self.selections):
+ raise ValueError("Baseline selections must be unique")
+ return self
+
+
+class ComparisonCoverage(_FrozenModel):
+ matched_tasks: Annotated[int, Field(ge=0)]
+ comparable_tasks: Annotated[int, Field(ge=0)]
+ current_execution_failures: Annotated[int, Field(ge=0)]
+ baseline_execution_failures: Annotated[int, Field(ge=0)]
+
+
+class HistoricalResolutionComparison(_FrozenModel):
+ baseline_resolved: Annotated[int, Field(ge=0)]
+ current_resolved: Annotated[int, Field(ge=0)]
+ total: Annotated[int, Field(ge=0)]
+ baseline_rate_percent: Annotated[float, Field(ge=0, le=100, allow_inf_nan=False)]
+ current_rate_percent: Annotated[float, Field(ge=0, le=100, allow_inf_nan=False)]
+ delta_points: Annotated[float, Field(allow_inf_nan=False)]
+
+
+class HistoricalTokenComparison(_FrozenModel):
+ baseline: Annotated[int, Field(ge=0)]
+ current: Annotated[int, Field(ge=0)]
+ delta: int
+ baseline_measured_tasks: Annotated[int, Field(ge=0)]
+ current_measured_tasks: Annotated[int, Field(ge=0)]
+
+
+class BaselineComparison(_FrozenModel):
+ baseline: BaselineRecord
+ current_arm: Arm
+ compatibility: BaselineCompatibility
+ coverage: ComparisonCoverage
+ resolution: HistoricalResolutionComparison
+ outcome_categories: dict[
+ Literal[
+ "baseline_fail_current_pass",
+ "baseline_pass_current_fail",
+ "both_pass",
+ "both_fail",
+ ],
+ Annotated[int, Field(ge=0)],
+ ]
+ input_tokens: HistoricalTokenComparison | None = None
+ output_tokens: HistoricalTokenComparison | None = None
+ total_tokens: HistoricalTokenComparison | None = None
+
+ @field_validator("current_arm", mode="before")
+ @classmethod
+ def parse_arm(cls, value: object) -> object:
+ return Arm(value) if isinstance(value, str) else value
+
+
+class BaselineComparisonResponse(_FrozenModel):
+ batch_id: str
+ report_revision: Annotated[int, Field(ge=0)]
+ comparisons: tuple[BaselineComparison, ...]
diff --git a/evaluation/src/powercontext_eval/web/batches.py b/evaluation/src/powercontext_eval/web/batches.py
index affdb3e8f..1ab24e00e 100644
--- a/evaluation/src/powercontext_eval/web/batches.py
+++ b/evaluation/src/powercontext_eval/web/batches.py
@@ -24,7 +24,7 @@
from powercontext_eval.benchmarks.swebench_pro.catalog import TaskSet
from powercontext_eval.codex import DEFAULT_CODEX_MODEL, DEFAULT_REASONING_EFFORT, is_safe_codex_model
-from powercontext_eval.models import PowerContextRef
+from powercontext_eval.models import PowerContextRef, TreatmentMode
from powercontext_eval.web.controls import BatchControlState
from powercontext_eval.web.estimation import BatchEstimate
from powercontext_eval.web.models import FailureCategory, FailureCode, TaskPhase, TaskStatus
@@ -41,12 +41,17 @@ class BatchCreate(_FrozenModel):
task_set: TaskSet
model: str = DEFAULT_CODEX_MODEL
reasoning_effort: Literal["medium"] = DEFAULT_REASONING_EFFORT
- treatment_mode: Literal["off_on"]
+ treatment_mode: TreatmentMode
idempotency_key: str = Field(min_length=8, max_length=128, pattern=r"^[A-Za-z0-9._-]+$")
usage_pause_percent: Annotated[int, Field(ge=1, le=100)] = 80
initial_control_intent: Literal["run", "pause"] = "run"
container_env: dict[str, str] = Field(default_factory=dict)
+ @field_validator("treatment_mode", mode="before")
+ @classmethod
+ def parse_treatment_mode(cls, value: object) -> object:
+ return TreatmentMode(value) if isinstance(value, str) else value
+
@field_validator("powercontext_ref")
@classmethod
def validate_ref(cls, value: str) -> str:
@@ -69,7 +74,7 @@ class BatchPreviewResponse(_FrozenModel):
task_set: TaskSet
model: str
reasoning_effort: Literal["medium"]
- treatment_mode: Literal["off_on"]
+ treatment_mode: TreatmentMode
total_tasks: Annotated[int, Field(ge=1)]
usage_pause_percent: Annotated[int, Field(ge=1, le=100)]
usage: UsageSnapshot | None
@@ -77,6 +82,11 @@ class BatchPreviewResponse(_FrozenModel):
can_start: bool
block_reason: Literal["usage_threshold_reached"] | None = None
+ @field_validator("treatment_mode", mode="before")
+ @classmethod
+ def parse_treatment_mode(cls, value: object) -> object:
+ return TreatmentMode(value) if isinstance(value, str) else value
+
class TaskRetryRequest(_FrozenModel):
idempotency_key: str = Field(min_length=8, max_length=128, pattern=r"^[A-Za-z0-9._-]+$")
@@ -211,11 +221,11 @@ class ResolutionAggregate(_FrozenModel):
class TokenMetricAggregate(_FrozenModel):
- off: Annotated[int, Field(ge=0)]
- on: Annotated[int, Field(ge=0)]
- delta: int
- off_measured_tasks: Annotated[int, Field(ge=0)]
- on_measured_tasks: Annotated[int, Field(ge=0)]
+ off: Annotated[int, Field(ge=0)] | None = None
+ on: Annotated[int, Field(ge=0)] | None = None
+ delta: int | None = None
+ off_measured_tasks: Annotated[int, Field(ge=0)] | None = None
+ on_measured_tasks: Annotated[int, Field(ge=0)] | None = None
class TokenAggregate(_FrozenModel):
@@ -226,16 +236,17 @@ class TokenAggregate(_FrozenModel):
class BatchReportResponse(_FrozenModel):
batch_id: str
+ treatment_mode: TreatmentMode
report_revision: Annotated[int, Field(ge=0)]
total_tasks: Annotated[int, Field(ge=1)]
terminal_tasks: Annotated[int, Field(ge=0)]
- comparable_pairs: Annotated[int, Field(ge=0)]
+ comparable_pairs: Annotated[int, Field(ge=0)] | None = None
execution_failures: Annotated[int, Field(ge=0)]
cancelled_tasks: Annotated[int, Field(ge=0)]
- off: ResolutionAggregate
- on: ResolutionAggregate
- resolution_rate_delta_points: Annotated[float, Field(allow_inf_nan=False)]
- pair_categories: dict[PairCategory, Annotated[int, Field(ge=0)]]
+ off: ResolutionAggregate | None = None
+ on: ResolutionAggregate | None = None
+ resolution_rate_delta_points: Annotated[float, Field(allow_inf_nan=False)] | None = None
+ pair_categories: dict[PairCategory, Annotated[int, Field(ge=0)]] | None = None
task_statuses: dict[TaskStatus, Annotated[int, Field(ge=0)]]
tokens: TokenAggregate
control: BatchControlState
@@ -244,6 +255,11 @@ class BatchReportResponse(_FrozenModel):
revisions: dict[str, str]
configuration: dict[str, str]
+ @field_validator("treatment_mode", mode="before")
+ @classmethod
+ def parse_report_treatment_mode(cls, value: object) -> object:
+ return TreatmentMode(value) if isinstance(value, str) else value
+
class TaskArmSummary(_FrozenModel):
resolved: bool
diff --git a/evaluation/src/powercontext_eval/web/controls.py b/evaluation/src/powercontext_eval/web/controls.py
index 188a81574..3f8a8e1f1 100644
--- a/evaluation/src/powercontext_eval/web/controls.py
+++ b/evaluation/src/powercontext_eval/web/controls.py
@@ -24,7 +24,7 @@
from powercontext_eval.benchmarks.swebench_pro.catalog import PUBLIC_V2_TASK_SET, TaskSet
from powercontext_eval.codex import DEFAULT_CODEX_MODEL, is_safe_codex_model
-from powercontext_eval.models import PowerContextRef
+from powercontext_eval.models import PowerContextRef, TreatmentMode
from powercontext_eval.web.models import TaskStatus
if TYPE_CHECKING:
@@ -61,8 +61,14 @@ class BatchPreviewRequest(_FrozenModel):
powercontext_ref: str
task_set: TaskSet = PUBLIC_V2_TASK_SET
model: str = DEFAULT_CODEX_MODEL
+ treatment_mode: TreatmentMode = TreatmentMode.OFF_ON
usage_pause_percent: Annotated[int, Field(ge=1, le=100)] = 80
+ @field_validator("treatment_mode", mode="before")
+ @classmethod
+ def parse_treatment_mode(cls, value: object) -> object:
+ return TreatmentMode(value) if isinstance(value, str) else value
+
@field_validator("powercontext_ref")
@classmethod
def validate_ref(cls, value: str) -> str:
diff --git a/evaluation/src/powercontext_eval/web/models.py b/evaluation/src/powercontext_eval/web/models.py
index 7417f259e..d960d195c 100644
--- a/evaluation/src/powercontext_eval/web/models.py
+++ b/evaluation/src/powercontext_eval/web/models.py
@@ -24,7 +24,7 @@
from powercontext_eval.artifacts import ArmState
from powercontext_eval.codex import DEFAULT_CODEX_MODEL, DEFAULT_REASONING_EFFORT, is_safe_codex_model
-from powercontext_eval.models import PowerContextRef
+from powercontext_eval.models import Arm, PowerContextRef, TreatmentMode
from powercontext_eval.report import GoldValidationAudit
from powercontext_eval.runner import INSTANCE_ID
@@ -120,10 +120,15 @@ class TaskCreate(FrozenModel):
instance_id: str = Field(min_length=1, max_length=300, pattern=r"^[A-Za-z0-9._-]+$")
model: str = DEFAULT_CODEX_MODEL
reasoning_effort: Literal["medium"] = DEFAULT_REASONING_EFFORT
- treatment_mode: Literal["off_on"]
+ treatment_mode: TreatmentMode
idempotency_key: str = Field(min_length=8, max_length=128, pattern=r"^[A-Za-z0-9._-]+$")
container_env: dict[str, str] = Field(default_factory=dict)
+ @field_validator("treatment_mode", mode="before")
+ @classmethod
+ def parse_treatment_mode(cls, value: object) -> object:
+ return TreatmentMode(value) if isinstance(value, str) else value
+
@field_validator("powercontext_ref")
@classmethod
def validate_ref(cls, value: str) -> str:
@@ -157,8 +162,11 @@ class SafeFailure(FrozenModel):
class TaskResult(FrozenModel):
artifact_dir: str
report_path: str
- off_resolved: bool
- on_resolved: bool
+ off_resolved: bool | None = None
+ on_resolved: bool | None = None
+
+ def resolved_for(self, arm: Arm) -> bool | None:
+ return self.off_resolved if arm is Arm.OFF else self.on_resolved
class TaskRecord(FrozenModel):
@@ -325,7 +333,14 @@ class Capabilities(FrozenModel):
instances: tuple[Literal["instance_flipt-io__flipt-518ec324b66a07fdd95464a5e9ca5fe7681ad8f9"], ...] = (INSTANCE_ID,)
models: tuple[str, ...] = (DEFAULT_CODEX_MODEL,)
reasoning_efforts: tuple[Literal["medium"], ...] = ("medium",)
- treatment_modes: tuple[Literal["off_on"], ...] = ("off_on",)
+ treatment_modes: tuple[TreatmentMode, ...] = tuple(TreatmentMode)
+
+ @field_validator("treatment_modes", mode="before")
+ @classmethod
+ def parse_treatment_modes(cls, value: object) -> object:
+ if isinstance(value, (list, tuple)):
+ return tuple(TreatmentMode(item) if isinstance(item, str) else item for item in value)
+ return value
class HealthResponse(FrozenModel):
@@ -391,16 +406,17 @@ class TreatmentEvidence(FrozenModel):
class EvidenceResponse(FrozenModel):
- off: TreatmentEvidence
- on: TreatmentEvidence
+ off: TreatmentEvidence | None = None
+ on: TreatmentEvidence | None = None
class ReportResponse(FrozenModel):
task_id: str
acceptance_valid: bool
- off: ArmResponse
- on: ArmResponse
- comparison: ComparisonResponse
+ treatment_mode: TreatmentMode = TreatmentMode.OFF_ON
+ off: ArmResponse | None = None
+ on: ArmResponse | None = None
+ comparison: ComparisonResponse | None = None
evidence: EvidenceResponse
gold_validation: GoldValidationAudit | None = None
revisions: Mapping[str, str]
@@ -409,6 +425,11 @@ class ReportResponse(FrozenModel):
_utc_timestamp = field_validator("generated_at")(_require_utc)
+ @field_validator("treatment_mode", mode="before")
+ @classmethod
+ def parse_report_treatment_mode(cls, value: object) -> object:
+ return TreatmentMode(value) if isinstance(value, str) else value
+
@field_validator("revisions", "configuration")
@classmethod
def freeze_mapping(cls, value: Mapping[str, str]) -> Mapping[str, str]:
@@ -420,6 +441,16 @@ def serialize_mapping(self, value: Mapping[str, str]) -> dict[str, str]:
@model_validator(mode="after")
def require_distinct_arms(self) -> Self:
- if self.off.arm != "off" or self.on.arm != "on":
- raise ValueError("Report arms must preserve OFF/ON roles")
+ present = {arm for arm, response in ((Arm.OFF, self.off), (Arm.ON, self.on)) if response is not None}
+ evidence = {
+ arm for arm, response in ((Arm.OFF, self.evidence.off), (Arm.ON, self.evidence.on)) if response is not None
+ }
+ if present != set(self.treatment_mode.arms) or evidence != present:
+ raise ValueError("Report arms and evidence must match the treatment mode")
+ if self.off is not None and self.off.arm != "off":
+ raise ValueError("Report OFF arm has the wrong role")
+ if self.on is not None and self.on.arm != "on":
+ raise ValueError("Report ON arm has the wrong role")
+ if (self.comparison is None) is (self.treatment_mode is TreatmentMode.OFF_ON):
+ raise ValueError("Only paired reports contain an OFF/ON comparison")
return self
diff --git a/evaluation/src/powercontext_eval/web/reporting.py b/evaluation/src/powercontext_eval/web/reporting.py
index 5e500e0d8..e462fca35 100644
--- a/evaluation/src/powercontext_eval/web/reporting.py
+++ b/evaluation/src/powercontext_eval/web/reporting.py
@@ -16,11 +16,12 @@
from __future__ import annotations
+import hashlib
import json
import os
import stat
from collections import Counter
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from pathlib import Path
from typing import Literal, Protocol
@@ -29,10 +30,23 @@
from powercontext_eval.artifacts import ArmState
from powercontext_eval.benchmarks.swebench_pro.adapter import DATASET_REVISION, HARNESS_COMMIT, SweBenchProInstance
+from powercontext_eval.models import Arm, TreatmentMode
from powercontext_eval.report import ArmReport, ReportBundle, TestGroupReport
+from powercontext_eval.web.baselines import (
+ BaselineComparison,
+ BaselineCompatibility,
+ BaselineItemRecord,
+ BaselineRecord,
+ BaselineSnapshot,
+ ComparisonCoverage,
+ CompatibilityStatus,
+ HistoricalResolutionComparison,
+ HistoricalTokenComparison,
+)
from powercontext_eval.web.batches import (
BatchRecord,
BatchReportResponse,
+ BatchStatus,
BatchTaskDetailResponse,
BatchTaskItem,
BatchTaskPage,
@@ -117,6 +131,10 @@ def __init__(self) -> None:
super().__init__("Evaluation report artifacts are invalid")
+class StaleReportRevision(ReportingError):
+ """The report changed after the operator chose to save it."""
+
+
def _directory_flags() -> int:
return os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
@@ -196,8 +214,6 @@ def _load_bundle(run_fd: int) -> tuple[ReportBundle, os.stat_result]:
bundle = ReportBundle.model_validate_json(raw, strict=True)
except (ValidationError, ValueError, UnicodeDecodeError):
raise InvalidReportArtifact from None
- if bundle.off.arm != "off" or bundle.on.arm != "on":
- raise InvalidReportArtifact
return bundle, metadata
@@ -216,29 +232,32 @@ def _load_evidence(run_fd: int, arm: str) -> TreatmentEvidence:
def _validate_evidence(
bundle: ReportBundle,
run_id: str,
- off: TreatmentEvidence,
- on: TreatmentEvidence,
+ evidence: Mapping[Arm, TreatmentEvidence],
) -> None:
expected_sha = bundle.revisions.get("powercontext")
configured_plugin_id = bundle.configuration.get("plugin_id", _PLUGIN_ID)
- configured_plugin_version = bundle.configuration.get("plugin_version", off.plugin_version)
- common = (
- expected_sha is not None
- and off.plugin_checkout_sha == expected_sha
- and on.plugin_checkout_sha == expected_sha
- and off.plugin_id == configured_plugin_id == on.plugin_id == _PLUGIN_ID
- and bool(off.plugin_version)
- and off.plugin_version == configured_plugin_version == on.plugin_version
- and off.plugin_installed
- and on.plugin_installed
- and off.server_ready
- and on.server_ready
- and off.scope_id == f"eval:{run_id}:off"
- and on.scope_id == f"eval:{run_id}:on"
- )
- activity = off.prompt_sources == 0 and off.mcp_requests == 0 and on.prompt_sources > 0 and on.mcp_requests > 0
- if not common or not activity:
+ if set(evidence) != set(bundle.treatment_mode.arms) or expected_sha is None:
raise InvalidReportArtifact
+ versions = {item.plugin_version for item in evidence.values()}
+ configured_plugin_version = bundle.configuration.get("plugin_version", next(iter(versions), ""))
+ if len(versions) != 1 or not configured_plugin_version:
+ raise InvalidReportArtifact
+ for arm, item in evidence.items():
+ common = (
+ item.plugin_checkout_sha == expected_sha
+ and item.plugin_id == configured_plugin_id == _PLUGIN_ID
+ and item.plugin_version == configured_plugin_version
+ and item.plugin_installed
+ and item.server_ready
+ and item.scope_id == f"eval:{run_id}:{arm.value}"
+ )
+ activity = (
+ item.prompt_sources == 0 and item.mcp_requests == 0
+ if arm is Arm.OFF
+ else item.prompt_sources > 0 and item.mcp_requests > 0
+ )
+ if not common or not activity:
+ raise InvalidReportArtifact
def _arm_response(arm: Literal["off", "on"], report: ArmReport) -> ArmResponse:
@@ -280,6 +299,11 @@ def _comparisons(off: ArmReport, on: ArmReport) -> ComparisonResponse:
def _acceptance_valid(bundle: ReportBundle) -> bool:
+ if bundle.off is None:
+ assert bundle.on is not None
+ return bundle.on.treatment_valid and bundle.on.state in _COMPARABLE_STATES and bundle.on.resolved
+ if bundle.on is None:
+ return bundle.off.treatment_valid and bundle.off.state in _COMPARABLE_STATES and bundle.off.resolved
lifecycle_is_comparable = bundle.off.state == bundle.on.state and bundle.off.state in _COMPARABLE_STATES
official_outcomes_are_coherent = (
bundle.off.passed is True and bundle.on.passed is True and bundle.off.resolved and bundle.on.resolved
@@ -298,16 +322,18 @@ def load_report(run_dir: Path, run_root: Path | None = None) -> ReportResponse:
run_fd, run_id = _open_run(run_dir, run_root)
try:
bundle, report_metadata = _load_bundle(run_fd)
- off_evidence = _load_evidence(run_fd, "off")
- on_evidence = _load_evidence(run_fd, "on")
- _validate_evidence(bundle, run_id, off_evidence, on_evidence)
+ evidence = {arm: _load_evidence(run_fd, arm.value) for arm in bundle.treatment_mode.arms}
+ _validate_evidence(bundle, run_id, evidence)
return ReportResponse(
task_id=run_id,
acceptance_valid=_acceptance_valid(bundle),
- off=_arm_response("off", bundle.off),
- on=_arm_response("on", bundle.on),
- comparison=_comparisons(bundle.off, bundle.on),
- evidence=EvidenceResponse(off=off_evidence, on=on_evidence),
+ treatment_mode=bundle.treatment_mode,
+ off=_arm_response("off", bundle.off) if bundle.off is not None else None,
+ on=_arm_response("on", bundle.on) if bundle.on is not None else None,
+ comparison=(
+ _comparisons(bundle.off, bundle.on) if bundle.off is not None and bundle.on is not None else None
+ ),
+ evidence=EvidenceResponse(off=evidence.get(Arm.OFF), on=evidence.get(Arm.ON)),
gold_validation=bundle.gold_validation,
revisions=bundle.revisions,
configuration=bundle.configuration,
@@ -359,11 +385,17 @@ def _bundle_for_task(task: TaskRecord, runs_root: Path) -> ReportBundle:
raise InvalidReportArtifact
bundle = _load_batch_bundle(task_run_dir(task, runs_root), runs_root)
if (
- bundle.off.arm != "off"
- or bundle.on.arm != "on"
+ bundle.treatment_mode != task.request.treatment_mode
or bundle.configuration.get("instance") != task.request.instance_id
- or bundle.off.resolved != task.result.off_resolved
- or bundle.on.resolved != task.result.on_resolved
+ ):
+ raise InvalidReportArtifact
+ for arm, report in ((Arm.OFF, bundle.off), (Arm.ON, bundle.on)):
+ if (report is None) != (arm not in task.request.treatment_mode.arms):
+ raise InvalidReportArtifact
+ if report is not None and report.resolved != task.result.resolved_for(arm):
+ raise InvalidReportArtifact
+ if {arm for arm in (Arm.OFF, Arm.ON) if task.result.resolved_for(arm) is not None} != set(
+ task.request.treatment_mode.arms
):
raise InvalidReportArtifact
return bundle
@@ -426,11 +458,20 @@ def _task_item(
tokens = TaskTokenDelta()
if task.status is TaskStatus.SUCCEEDED:
bundle = _bundle_for_task(task, runs_root)
- category = _pair_category(bundle.off.resolved, bundle.on.resolved)
- off = _task_arm(bundle.off)
- on = _task_arm(bundle.on)
- delta = None if off.total_tokens is None or on.total_tokens is None else on.total_tokens - off.total_tokens
- tokens = TaskTokenDelta(off=off.total_tokens, on=on.total_tokens, delta=delta)
+ off = _task_arm(bundle.off) if bundle.off is not None else None
+ on = _task_arm(bundle.on) if bundle.on is not None else None
+ if off is not None and on is not None:
+ category = _pair_category(off.resolved, on.resolved)
+ delta = (
+ None
+ if off is None or on is None or off.total_tokens is None or on.total_tokens is None
+ else on.total_tokens - off.total_tokens
+ )
+ tokens = TaskTokenDelta(
+ off=off.total_tokens if off is not None else None,
+ on=on.total_tokens if on is not None else None,
+ delta=delta,
+ )
elif task.status in _EXECUTION_FAILURE_STATES:
category = PairCategory.EXECUTION_FAILURE
return BatchTaskItem(
@@ -454,15 +495,15 @@ def _task_item(
)
-def _metric_aggregate(values: dict[str, list[int]]) -> TokenMetricAggregate:
- off = sum(values["off"])
- on = sum(values["on"])
+def _metric_aggregate(values: dict[str, list[int]], mode: TreatmentMode) -> TokenMetricAggregate:
+ off = sum(values["off"]) if Arm.OFF in mode.arms else None
+ on = sum(values["on"]) if Arm.ON in mode.arms else None
return TokenMetricAggregate(
off=off,
on=on,
- delta=on - off,
- off_measured_tasks=len(values["off"]),
- on_measured_tasks=len(values["on"]),
+ delta=on - off if off is not None and on is not None else None,
+ off_measured_tasks=len(values["off"]) if Arm.OFF in mode.arms else None,
+ on_measured_tasks=len(values["on"]) if Arm.ON in mode.arms else None,
)
@@ -491,13 +532,12 @@ def load_batch_estimate_samples(
or task.finished_at is None
):
continue
- off_total = _arm_total(bundle.off)
- on_total = _arm_total(bundle.on)
- if off_total is None or on_total is None:
+ arm_totals = [_arm_total(report) for report in (bundle.off, bundle.on) if report is not None]
+ if not arm_totals or any(total is None for total in arm_totals):
continue
samples.append(
EstimateSample(
- tokens=off_total + on_total,
+ tokens=sum(total for total in arm_totals if total is not None),
duration_seconds=max(0, round((task.finished_at - task.started_at).total_seconds())),
)
)
@@ -531,30 +571,45 @@ def load_batch_report(
catalog.require(task.request.instance_id)
if task.status is TaskStatus.SUCCEEDED:
bundle = _bundle_for_task(task, runs_root)
- category = _pair_category(bundle.off.resolved, bundle.on.resolved)
- categories[category] += 1
- comparable += 1
- off_resolved += int(bundle.off.resolved)
- on_resolved += int(bundle.on.resolved)
- metric_pairs = {
- "input": (bundle.off.metrics.input_tokens, bundle.on.metrics.input_tokens),
- "output": (bundle.off.metrics.output_tokens, bundle.on.metrics.output_tokens),
- "total": (_arm_total(bundle.off), _arm_total(bundle.on)),
+ if bundle.off is not None:
+ off_resolved += int(bundle.off.resolved)
+ if bundle.on is not None:
+ on_resolved += int(bundle.on.resolved)
+ if bundle.off is not None and bundle.on is not None:
+ category = _pair_category(bundle.off.resolved, bundle.on.resolved)
+ categories[category] += 1
+ comparable += 1
+ metric_values = {
+ "input": {
+ "off": bundle.off.metrics.input_tokens if bundle.off is not None else None,
+ "on": bundle.on.metrics.input_tokens if bundle.on is not None else None,
+ },
+ "output": {
+ "off": bundle.off.metrics.output_tokens if bundle.off is not None else None,
+ "on": bundle.on.metrics.output_tokens if bundle.on is not None else None,
+ },
+ "total": {
+ "off": _arm_total(bundle.off) if bundle.off is not None else None,
+ "on": _arm_total(bundle.on) if bundle.on is not None else None,
+ },
}
- for metric_name, (off_value, on_value) in metric_pairs.items():
- if off_value is not None and on_value is not None:
- token_values[metric_name]["off"].append(off_value)
- token_values[metric_name]["on"].append(on_value)
- paired_tokens = metric_pairs["total"]
+ for metric_name, arm_values in metric_values.items():
+ if batch.request.treatment_mode is TreatmentMode.OFF_ON and any(
+ value is None for value in arm_values.values()
+ ):
+ continue
+ for arm_name, value in arm_values.items():
+ if value is not None:
+ token_values[metric_name][arm_name].append(value)
+ selected_totals = [metric_values["total"][arm.value] for arm in batch.request.treatment_mode.arms]
if (
- paired_tokens[0] is not None
- and paired_tokens[1] is not None
+ all(value is not None for value in selected_totals)
and task.started_at is not None
and task.finished_at is not None
):
estimate_samples.append(
EstimateSample(
- tokens=paired_tokens[0] + paired_tokens[1],
+ tokens=sum(value for value in selected_totals if value is not None),
duration_seconds=max(0, round((task.finished_at - task.started_at).total_seconds())),
)
)
@@ -566,7 +621,8 @@ def load_batch_report(
elif revisions != candidate_revisions or configuration != candidate_configuration:
raise InvalidReportArtifact
elif task.status in _EXECUTION_FAILURE_STATES:
- categories[PairCategory.EXECUTION_FAILURE] += 1
+ if batch.request.treatment_mode is TreatmentMode.OFF_ON:
+ categories[PairCategory.EXECUTION_FAILURE] += 1
execution_failures += 1
if revisions is None:
@@ -593,21 +649,32 @@ def load_batch_report(
on_rate = on_resolved / denominator * 100
return BatchReportResponse(
batch_id=batch.batch_id,
+ treatment_mode=batch.request.treatment_mode,
report_revision=batch.control.version + sum(task.attempt_count * 100 + task.version for task in tasks),
total_tasks=total,
terminal_tasks=terminal_tasks,
- comparable_pairs=comparable,
+ comparable_pairs=comparable if batch.request.treatment_mode is TreatmentMode.OFF_ON else None,
execution_failures=execution_failures,
cancelled_tasks=status_counts[TaskStatus.CANCELLED],
- off=ResolutionAggregate(resolved=off_resolved, total=terminal_tasks, rate_percent=off_rate),
- on=ResolutionAggregate(resolved=on_resolved, total=terminal_tasks, rate_percent=on_rate),
- resolution_rate_delta_points=on_rate - off_rate,
- pair_categories=categories,
+ off=(
+ ResolutionAggregate(resolved=off_resolved, total=terminal_tasks, rate_percent=off_rate)
+ if Arm.OFF in batch.request.treatment_mode.arms
+ else None
+ ),
+ on=(
+ ResolutionAggregate(resolved=on_resolved, total=terminal_tasks, rate_percent=on_rate)
+ if Arm.ON in batch.request.treatment_mode.arms
+ else None
+ ),
+ resolution_rate_delta_points=(
+ on_rate - off_rate if batch.request.treatment_mode is TreatmentMode.OFF_ON else None
+ ),
+ pair_categories=categories if batch.request.treatment_mode is TreatmentMode.OFF_ON else None,
task_statuses={status: status_counts[status] for status in TaskStatus},
tokens=TokenAggregate(
- input=_metric_aggregate(token_values["input"]),
- output=_metric_aggregate(token_values["output"]),
- total=_metric_aggregate(token_values["total"]),
+ input=_metric_aggregate(token_values["input"], batch.request.treatment_mode),
+ output=_metric_aggregate(token_values["output"], batch.request.treatment_mode),
+ total=_metric_aggregate(token_values["total"], batch.request.treatment_mode),
),
control=batch.control,
latest_usage=latest_usage,
@@ -625,6 +692,244 @@ def load_batch_report(
)
+def instance_set_digest(tasks: Sequence[TaskRecord]) -> str:
+ """Return a stable identity for the exact ordered benchmark instance set."""
+
+ instance_ids = [task.instance_id for task in tasks]
+ if any(instance_id is None for instance_id in instance_ids):
+ raise InvalidReportArtifact
+ payload = json.dumps(instance_ids, ensure_ascii=True, separators=(",", ":")).encode()
+ return hashlib.sha256(payload).hexdigest()
+
+
+def create_baseline_snapshot(
+ batch: BatchRecord,
+ tasks: Sequence[TaskRecord],
+ *,
+ arm: Arm,
+ expected_report_revision: int,
+ runs_root: Path,
+ catalog: BenchmarkCatalog,
+) -> BaselineSnapshot:
+ """Materialize immutable comparison facts for one exact completed batch arm."""
+
+ if batch.status is not BatchStatus.COMPLETED:
+ raise ValueError("Only completed batches can be saved as baselines")
+ if arm not in batch.request.treatment_mode.arms:
+ raise ValueError("The selected baseline arm was not executed")
+ report = load_batch_report(batch, tasks, runs_root=runs_root, catalog=catalog)
+ if report.report_revision != expected_report_revision:
+ raise StaleReportRevision
+ items: list[BaselineItemRecord] = []
+ resolved_tasks = 0
+ for task in tasks:
+ resolved: bool | None = None
+ input_tokens: int | None = None
+ output_tokens: int | None = None
+ total_tokens: int | None = None
+ if task.status is TaskStatus.SUCCEEDED:
+ bundle = _bundle_for_task(task, runs_root)
+ arm_report = bundle.off if arm is Arm.OFF else bundle.on
+ if arm_report is None:
+ raise InvalidReportArtifact
+ resolved = arm_report.resolved
+ input_tokens = arm_report.metrics.input_tokens
+ output_tokens = arm_report.metrics.output_tokens
+ total_tokens = _arm_total(arm_report)
+ resolved_tasks += int(resolved)
+ if task.instance_id is None or task.source_index is None:
+ raise InvalidReportArtifact
+ items.append(
+ BaselineItemRecord(
+ baseline_id="",
+ instance_id=task.instance_id,
+ source_index=task.source_index,
+ source_task_id=task.task_id,
+ source_attempt_id=task.attempt_id,
+ status=task.status,
+ resolved=resolved,
+ input_tokens=input_tokens,
+ output_tokens=output_tokens,
+ total_tokens=total_tokens,
+ )
+ )
+ dataset_revision = report.revisions.get("dataset")
+ harness_revision = report.revisions.get("harness")
+ if not dataset_revision or not harness_revision:
+ raise InvalidReportArtifact
+ return BaselineSnapshot(
+ benchmark=batch.request.benchmark,
+ task_set=batch.request.task_set,
+ instance_set_digest=instance_set_digest(tasks),
+ total_tasks=batch.total_tasks,
+ resolved_tasks=resolved_tasks,
+ execution_failures=report.execution_failures,
+ model=batch.request.model,
+ reasoning_effort=batch.request.reasoning_effort,
+ dataset_revision=dataset_revision,
+ harness_revision=harness_revision,
+ powercontext_sha=report.revisions.get("powercontext"),
+ codex_version=report.configuration.get("codex"),
+ items=tuple(items),
+ )
+
+
+def baseline_compatibility(
+ baseline: BaselineRecord,
+ batch: BatchRecord,
+ tasks: Sequence[TaskRecord],
+ report: BatchReportResponse,
+ *,
+ current_arm: Arm,
+) -> BaselineCompatibility:
+ """Classify comparison safety without treating the PowerContext revision as a gate."""
+
+ hard_reasons: list[str] = []
+ checks = (
+ (baseline.benchmark, batch.request.benchmark, "benchmark differs"),
+ (baseline.task_set, batch.request.task_set, "task set differs"),
+ (baseline.instance_set_digest, instance_set_digest(tasks), "instance set differs"),
+ (baseline.total_tasks, batch.total_tasks, "task count differs"),
+ (baseline.model, batch.request.model, "model differs"),
+ (baseline.reasoning_effort, batch.request.reasoning_effort, "reasoning effort differs"),
+ (baseline.dataset_revision, report.revisions.get("dataset"), "dataset revision differs"),
+ (baseline.harness_revision, report.revisions.get("harness"), "harness revision differs"),
+ )
+ hard_reasons.extend(reason for baseline_value, current_value, reason in checks if baseline_value != current_value)
+ if baseline.codex_version is not None and baseline.codex_version != report.configuration.get("codex"):
+ hard_reasons.append("Codex version differs")
+ if current_arm not in batch.request.treatment_mode.arms:
+ hard_reasons.append("current arm was not executed")
+ if hard_reasons:
+ return BaselineCompatibility(status=CompatibilityStatus.INCOMPATIBLE, reasons=tuple(hard_reasons))
+ if baseline.source_arm is not current_arm:
+ return BaselineCompatibility(
+ status=CompatibilityStatus.WARNING,
+ reasons=("cross-arm comparison",),
+ )
+ return BaselineCompatibility(status=CompatibilityStatus.COMPATIBLE)
+
+
+def _historical_tokens(
+ baseline_items: Sequence[BaselineItemRecord],
+ current_values: Mapping[str, int | None],
+ field: Literal["input_tokens", "output_tokens", "total_tokens"],
+) -> HistoricalTokenComparison | None:
+ baseline_values = [getattr(item, field) for item in baseline_items if getattr(item, field) is not None]
+ measured_current = [value for value in current_values.values() if value is not None]
+ if not baseline_values or not measured_current:
+ return None
+ baseline_total = sum(baseline_values)
+ current_total = sum(measured_current)
+ return HistoricalTokenComparison(
+ baseline=baseline_total,
+ current=current_total,
+ delta=current_total - baseline_total,
+ baseline_measured_tasks=len(baseline_values),
+ current_measured_tasks=len(measured_current),
+ )
+
+
+def compare_batch_to_baseline(
+ batch: BatchRecord,
+ tasks: Sequence[TaskRecord],
+ report: BatchReportResponse,
+ baseline: BaselineRecord,
+ baseline_items: Sequence[BaselineItemRecord],
+ *,
+ current_arm: Arm,
+ runs_root: Path,
+) -> BaselineComparison:
+ """Compare frozen per-instance facts without invoking the evaluation runner."""
+
+ compatibility = baseline_compatibility(baseline, batch, tasks, report, current_arm=current_arm)
+ baseline_by_instance = {item.instance_id: item for item in baseline_items}
+ current_by_instance = {task.instance_id: task for task in tasks if task.instance_id is not None}
+ matched_ids = sorted(set(baseline_by_instance) & set(current_by_instance))
+ categories: dict[
+ Literal[
+ "baseline_fail_current_pass",
+ "baseline_pass_current_fail",
+ "both_pass",
+ "both_fail",
+ ],
+ int,
+ ] = {
+ "baseline_fail_current_pass": 0,
+ "baseline_pass_current_fail": 0,
+ "both_pass": 0,
+ "both_fail": 0,
+ }
+ current_resolved = 0
+ comparable = 0
+ current_failures = 0
+ baseline_failures = 0
+ current_tokens: dict[str, dict[str, int | None]] = {
+ "input_tokens": {},
+ "output_tokens": {},
+ "total_tokens": {},
+ }
+ matched_baseline_items: list[BaselineItemRecord] = []
+ for instance_id in matched_ids:
+ baseline_item = baseline_by_instance[instance_id]
+ task = current_by_instance[instance_id]
+ matched_baseline_items.append(baseline_item)
+ baseline_failures += int(baseline_item.status in _EXECUTION_FAILURE_STATES)
+ current_failures += int(task.status in _EXECUTION_FAILURE_STATES)
+ current_result: bool | None = None
+ values = {"input_tokens": None, "output_tokens": None, "total_tokens": None}
+ if task.status is TaskStatus.SUCCEEDED:
+ bundle = _bundle_for_task(task, runs_root)
+ arm_report = bundle.off if current_arm is Arm.OFF else bundle.on
+ if arm_report is None:
+ raise InvalidReportArtifact
+ current_result = arm_report.resolved
+ current_resolved += int(current_result)
+ values = {
+ "input_tokens": arm_report.metrics.input_tokens,
+ "output_tokens": arm_report.metrics.output_tokens,
+ "total_tokens": _arm_total(arm_report),
+ }
+ for field, value in values.items():
+ current_tokens[field][instance_id] = value
+ if baseline_item.resolved is not None and current_result is not None:
+ comparable += 1
+ if baseline_item.resolved and current_result:
+ categories["both_pass"] += 1
+ elif baseline_item.resolved:
+ categories["baseline_pass_current_fail"] += 1
+ elif current_result:
+ categories["baseline_fail_current_pass"] += 1
+ else:
+ categories["both_fail"] += 1
+ total = len(matched_ids)
+ baseline_rate = baseline.resolved_tasks / total * 100 if total else 0.0
+ current_rate = current_resolved / total * 100 if total else 0.0
+ return BaselineComparison(
+ baseline=baseline,
+ current_arm=current_arm,
+ compatibility=compatibility,
+ coverage=ComparisonCoverage(
+ matched_tasks=total,
+ comparable_tasks=comparable,
+ current_execution_failures=current_failures,
+ baseline_execution_failures=baseline_failures,
+ ),
+ resolution=HistoricalResolutionComparison(
+ baseline_resolved=baseline.resolved_tasks,
+ current_resolved=current_resolved,
+ total=total,
+ baseline_rate_percent=baseline_rate,
+ current_rate_percent=current_rate,
+ delta_points=current_rate - baseline_rate,
+ ),
+ outcome_categories=categories,
+ input_tokens=_historical_tokens(matched_baseline_items, current_tokens["input_tokens"], "input_tokens"),
+ output_tokens=_historical_tokens(matched_baseline_items, current_tokens["output_tokens"], "output_tokens"),
+ total_tokens=_historical_tokens(matched_baseline_items, current_tokens["total_tokens"], "total_tokens"),
+ )
+
+
def load_batch_task_page(
batch: BatchRecord,
tasks: Sequence[TaskRecord],
@@ -740,8 +1045,8 @@ def load_batch_task_detail(
on = None
if task.status is TaskStatus.SUCCEEDED:
bundle = _bundle_for_task(task, runs_root)
- off = _detail_arm(bundle.off)
- on = _detail_arm(bundle.on)
+ off = _detail_arm(bundle.off) if bundle.off is not None else None
+ on = _detail_arm(bundle.on) if bundle.on is not None else None
finalization_by_arm = {record.arm: tokensflow_finalization_summary(record) for record in finalizations}
return BatchTaskDetailResponse(
task=item,
diff --git a/evaluation/src/powercontext_eval/web/revision.py b/evaluation/src/powercontext_eval/web/revision.py
index 803881036..5ba3cd5d9 100644
--- a/evaluation/src/powercontext_eval/web/revision.py
+++ b/evaluation/src/powercontext_eval/web/revision.py
@@ -22,7 +22,7 @@
from pathlib import Path
from re import fullmatch
-RUNTIME_SCHEMA_VERSION = 2
+RUNTIME_SCHEMA_VERSION = 3
_REVISION_ENVIRONMENT = "POWERCONTEXT_EVAL_BUILD_REVISION"
diff --git a/evaluation/src/powercontext_eval/web/store.py b/evaluation/src/powercontext_eval/web/store.py
index 6f3488767..5594734ac 100644
--- a/evaluation/src/powercontext_eval/web/store.py
+++ b/evaluation/src/powercontext_eval/web/store.py
@@ -26,10 +26,17 @@
from enum import StrEnum
from pathlib import Path
from re import fullmatch
-from typing import Any, Literal, TypedDict
+from typing import Any, Literal, TypedDict, cast
from powercontext_eval.codex import DEFAULT_CODEX_MODEL, DEFAULT_REASONING_EFFORT
from powercontext_eval.paths import EvaluationPaths
+from powercontext_eval.web.baselines import (
+ BaselineCreate,
+ BaselineItemRecord,
+ BaselineRecord,
+ BaselineSelection,
+ BaselineSnapshot,
+)
from powercontext_eval.web.batches import (
BatchControlEvent,
BatchControlEventType,
@@ -82,6 +89,10 @@ class BatchNotFound(TaskStoreError):
"""The requested batch does not exist."""
+class BaselineNotFound(TaskStoreError):
+ """The requested immutable baseline does not exist."""
+
+
class TaskConflict(TaskStoreError):
"""The requested transition conflicts with the task lifecycle."""
@@ -388,6 +399,52 @@ def initialize(self) -> None:
lease_expires_at TEXT,
UNIQUE(attempt_id, arm)
);
+ CREATE TABLE IF NOT EXISTS baselines (
+ baseline_seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ baseline_id TEXT NOT NULL UNIQUE,
+ idempotency_key TEXT NOT NULL UNIQUE,
+ request_json TEXT NOT NULL,
+ name TEXT NOT NULL,
+ source_batch_id TEXT NOT NULL REFERENCES batches(batch_id),
+ source_arm TEXT NOT NULL CHECK (source_arm IN ('off', 'on')),
+ source_report_revision INTEGER NOT NULL CHECK (source_report_revision >= 0),
+ benchmark TEXT NOT NULL CHECK (benchmark = 'swebench-pro'),
+ task_set TEXT NOT NULL,
+ instance_set_digest TEXT NOT NULL,
+ total_tasks INTEGER NOT NULL CHECK (total_tasks > 0),
+ resolved_tasks INTEGER NOT NULL CHECK (resolved_tasks >= 0),
+ execution_failures INTEGER NOT NULL CHECK (execution_failures >= 0),
+ model TEXT NOT NULL,
+ reasoning_effort TEXT NOT NULL,
+ dataset_revision TEXT NOT NULL,
+ harness_revision TEXT NOT NULL,
+ powercontext_sha TEXT,
+ codex_version TEXT,
+ created_at TEXT NOT NULL
+ );
+ CREATE TABLE IF NOT EXISTS baseline_items (
+ baseline_id TEXT NOT NULL REFERENCES baselines(baseline_id),
+ instance_id TEXT NOT NULL,
+ source_index INTEGER NOT NULL CHECK (source_index >= 0),
+ source_task_id TEXT NOT NULL,
+ source_attempt_id TEXT,
+ status TEXT NOT NULL,
+ resolved INTEGER CHECK (resolved IN (0, 1)),
+ input_tokens INTEGER CHECK (input_tokens >= 0),
+ output_tokens INTEGER CHECK (output_tokens >= 0),
+ total_tokens INTEGER CHECK (total_tokens >= 0),
+ PRIMARY KEY (baseline_id, instance_id),
+ UNIQUE (baseline_id, source_index)
+ );
+ CREATE TABLE IF NOT EXISTS batch_baseline_selections (
+ batch_id TEXT NOT NULL REFERENCES batches(batch_id),
+ baseline_id TEXT NOT NULL REFERENCES baselines(baseline_id),
+ current_arm TEXT NOT NULL CHECK (current_arm IN ('off', 'on')),
+ sort_order INTEGER NOT NULL CHECK (sort_order >= 0),
+ created_at TEXT NOT NULL,
+ PRIMARY KEY (batch_id, baseline_id, current_arm),
+ UNIQUE (batch_id, sort_order)
+ );
""",
)
task_columns = {str(row["name"]) for row in connection.execute("PRAGMA table_info(tasks)").fetchall()}
@@ -540,6 +597,12 @@ def initialize(self) -> None:
ON tokensflow_finalizations(lease_expires_at);
CREATE INDEX IF NOT EXISTS tokensflow_finalizations_attempt
ON tokensflow_finalizations(attempt_id, arm);
+ CREATE INDEX IF NOT EXISTS baselines_created_desc
+ ON baselines(created_at DESC, baseline_seq DESC);
+ CREATE INDEX IF NOT EXISTS baseline_items_source
+ ON baseline_items(baseline_id, source_index);
+ CREATE INDEX IF NOT EXISTS batch_baseline_selections_order
+ ON batch_baseline_selections(batch_id, sort_order);
""",
)
@@ -676,6 +739,178 @@ def get_batch(self, batch_id: str) -> BatchRecord:
with self._connection() as connection:
return self._batch_record(connection, self._select_batch(connection, batch_id))
+ def create_baseline(
+ self,
+ request: BaselineCreate,
+ snapshot: BaselineSnapshot,
+ *,
+ now: datetime,
+ ) -> tuple[BaselineRecord, bool]:
+ """Freeze one completed batch arm without copying mutable run artifacts."""
+
+ if len(snapshot.items) != snapshot.total_tasks:
+ raise ValueError("Baseline snapshot item count does not match its task count")
+ if [item.source_index for item in snapshot.items] != list(range(snapshot.total_tasks)):
+ raise ValueError("Baseline snapshot items must retain contiguous source order")
+ request_json = request.model_dump_json()
+ created_at = _timestamp(now)
+ with self._write() as connection:
+ self._select_batch(connection, request.source_batch_id)
+ existing = connection.execute(
+ "SELECT * FROM baselines WHERE idempotency_key = ?",
+ (request.idempotency_key,),
+ ).fetchone()
+ if existing is not None:
+ self._require_idempotent_request(existing["request_json"], request_json)
+ return self._baseline_record(existing), False
+ placeholder = f"pending-baseline-{uuid.uuid4().hex}"
+ cursor = connection.execute(
+ """
+ INSERT INTO baselines(
+ baseline_id, idempotency_key, request_json, name, source_batch_id,
+ source_arm, source_report_revision, benchmark, task_set,
+ instance_set_digest, total_tasks, resolved_tasks, execution_failures,
+ model, reasoning_effort, dataset_revision, harness_revision,
+ powercontext_sha, codex_version, created_at
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ """,
+ (
+ placeholder,
+ request.idempotency_key,
+ request_json,
+ request.name,
+ request.source_batch_id,
+ request.source_arm.value,
+ request.expected_report_revision,
+ snapshot.benchmark,
+ snapshot.task_set,
+ snapshot.instance_set_digest,
+ snapshot.total_tasks,
+ snapshot.resolved_tasks,
+ snapshot.execution_failures,
+ snapshot.model,
+ snapshot.reasoning_effort,
+ snapshot.dataset_revision,
+ snapshot.harness_revision,
+ snapshot.powercontext_sha,
+ snapshot.codex_version,
+ created_at,
+ ),
+ )
+ sequence = cursor.lastrowid
+ if sequence is None: # pragma: no cover - guaranteed by SQLite
+ raise TaskStoreError("SQLite did not assign a baseline sequence")
+ baseline_id = _baseline_id(now, sequence)
+ connection.execute(
+ "UPDATE baselines SET baseline_id = ? WHERE baseline_seq = ?",
+ (baseline_id, sequence),
+ )
+ connection.executemany(
+ """
+ INSERT INTO baseline_items(
+ baseline_id, instance_id, source_index, source_task_id,
+ source_attempt_id, status, resolved, input_tokens,
+ output_tokens, total_tokens
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ """,
+ [
+ (
+ baseline_id,
+ item.instance_id,
+ item.source_index,
+ item.source_task_id,
+ item.source_attempt_id,
+ item.status.value,
+ None if item.resolved is None else int(item.resolved),
+ item.input_tokens,
+ item.output_tokens,
+ item.total_tokens,
+ )
+ for item in snapshot.items
+ ],
+ )
+ row = connection.execute("SELECT * FROM baselines WHERE baseline_id = ?", (baseline_id,)).fetchone()
+ assert row is not None
+ return self._baseline_record(row), True
+
+ def get_baseline(self, baseline_id: str) -> BaselineRecord:
+ with self._connection() as connection:
+ row = connection.execute("SELECT * FROM baselines WHERE baseline_id = ?", (baseline_id,)).fetchone()
+ if row is None:
+ raise BaselineNotFound(f"Baseline not found: {baseline_id}")
+ return self._baseline_record(row)
+
+ def list_baselines(self) -> list[BaselineRecord]:
+ """List newest immutable baselines first, with a stable sequence tie-breaker."""
+
+ with self._connection() as connection:
+ rows = connection.execute("SELECT * FROM baselines ORDER BY created_at DESC, baseline_seq DESC").fetchall()
+ return [self._baseline_record(row) for row in rows]
+
+ def list_baseline_items(self, baseline_id: str) -> list[BaselineItemRecord]:
+ with self._connection() as connection:
+ if connection.execute("SELECT 1 FROM baselines WHERE baseline_id = ?", (baseline_id,)).fetchone() is None:
+ raise BaselineNotFound(f"Baseline not found: {baseline_id}")
+ rows = connection.execute(
+ "SELECT * FROM baseline_items WHERE baseline_id = ? ORDER BY source_index ASC",
+ (baseline_id,),
+ ).fetchall()
+ return [self._baseline_item(row) for row in rows]
+
+ def replace_baseline_selections(
+ self,
+ batch_id: str,
+ selections: Sequence[BaselineSelection],
+ *,
+ now: datetime,
+ ) -> tuple[BaselineSelection, ...]:
+ """Replace only report presentation state; no task is queued or rerun."""
+
+ now_text = _timestamp(now)
+ if len(selections) > 10:
+ raise ValueError("At most ten baseline comparisons may be selected")
+ with self._write() as connection:
+ batch = self._batch_record(connection, self._select_batch(connection, batch_id))
+ for selection in selections:
+ if selection.current_arm not in batch.request.treatment_mode.arms:
+ raise TaskConflict("Selected current arm was not executed by this batch")
+ if (
+ connection.execute(
+ "SELECT 1 FROM baselines WHERE baseline_id = ?", (selection.baseline_id,)
+ ).fetchone()
+ is None
+ ):
+ raise BaselineNotFound(f"Baseline not found: {selection.baseline_id}")
+ connection.execute("DELETE FROM batch_baseline_selections WHERE batch_id = ?", (batch_id,))
+ connection.executemany(
+ """
+ INSERT INTO batch_baseline_selections(
+ batch_id, baseline_id, current_arm, sort_order, created_at
+ ) VALUES (?, ?, ?, ?, ?)
+ """,
+ [
+ (batch_id, selection.baseline_id, selection.current_arm.value, index, now_text)
+ for index, selection in enumerate(selections)
+ ],
+ )
+ return tuple(selections)
+
+ def list_baseline_selections(self, batch_id: str) -> tuple[BaselineSelection, ...]:
+ with self._connection() as connection:
+ self._select_batch(connection, batch_id)
+ rows = connection.execute(
+ """
+ SELECT baseline_id, current_arm
+ FROM batch_baseline_selections
+ WHERE batch_id = ?
+ ORDER BY sort_order ASC
+ """,
+ (batch_id,),
+ ).fetchall()
+ return tuple(
+ BaselineSelection(baseline_id=str(row["baseline_id"]), current_arm=str(row["current_arm"])) for row in rows
+ )
+
def save_usage_snapshot(self, snapshot: UsageSnapshot) -> UsageSnapshot:
"""Append one normalized account-wide usage observation."""
@@ -2515,6 +2750,61 @@ def _control_event(row: sqlite3.Row) -> BatchControlEvent:
strict=True,
)
+ @staticmethod
+ def _baseline_record(row: sqlite3.Row) -> BaselineRecord:
+ return BaselineRecord(
+ baseline_id=str(row["baseline_id"]),
+ name=str(row["name"]),
+ source_batch_id=str(row["source_batch_id"]),
+ source_arm=str(row["source_arm"]),
+ source_report_revision=_stored_int(row["source_report_revision"], name="baseline report revision"),
+ benchmark=cast(Literal["swebench-pro"], str(row["benchmark"])),
+ task_set=cast(
+ Literal["swebench-pro-public-v2", "swebench-pro-stability-v1"],
+ str(row["task_set"]),
+ ),
+ instance_set_digest=str(row["instance_set_digest"]),
+ total_tasks=_stored_int(row["total_tasks"], name="baseline task count"),
+ resolved_tasks=_stored_int(row["resolved_tasks"], name="baseline resolved count"),
+ execution_failures=_stored_int(row["execution_failures"], name="baseline failure count"),
+ model=str(row["model"]),
+ reasoning_effort=cast(Literal["medium"], str(row["reasoning_effort"])),
+ dataset_revision=str(row["dataset_revision"]),
+ harness_revision=str(row["harness_revision"]),
+ powercontext_sha=str(row["powercontext_sha"]) if row["powercontext_sha"] is not None else None,
+ codex_version=str(row["codex_version"]) if row["codex_version"] is not None else None,
+ created_at=_parse_timestamp(row["created_at"]),
+ )
+
+ @staticmethod
+ def _baseline_item(row: sqlite3.Row) -> BaselineItemRecord:
+ return BaselineItemRecord(
+ baseline_id=str(row["baseline_id"]),
+ instance_id=str(row["instance_id"]),
+ source_index=_stored_int(row["source_index"], name="baseline source index"),
+ source_task_id=str(row["source_task_id"]),
+ source_attempt_id=str(row["source_attempt_id"]) if row["source_attempt_id"] is not None else None,
+ status=TaskStatus(str(row["status"])),
+ resolved=(
+ bool(_stored_int(row["resolved"], name="baseline resolution")) if row["resolved"] is not None else None
+ ),
+ input_tokens=(
+ _stored_int(row["input_tokens"], name="baseline input tokens")
+ if row["input_tokens"] is not None
+ else None
+ ),
+ output_tokens=(
+ _stored_int(row["output_tokens"], name="baseline output tokens")
+ if row["output_tokens"] is not None
+ else None
+ ),
+ total_tokens=(
+ _stored_int(row["total_tokens"], name="baseline total tokens")
+ if row["total_tokens"] is not None
+ else None
+ ),
+ )
+
def _batch_record(self, connection: sqlite3.Connection, row: sqlite3.Row) -> BatchRecord:
batch_id = row["batch_id"]
if not isinstance(batch_id, str):
@@ -2935,6 +3225,11 @@ def _batch_id(now: datetime, sequence: int) -> str:
return batch_id
+def _baseline_id(now: datetime, sequence: int) -> str:
+ _timestamp(now)
+ return f"baseline-{now.astimezone(UTC):%Y%m%d-%H%M%S-%f}-{sequence:010d}-{uuid.uuid4().hex[:8]}"
+
+
def _batch_task_id(now: datetime, batch_sequence: int, source_index: int) -> str:
_timestamp(now)
task_id = f"run-{now.astimezone(UTC):%Y%m%d-%H%M%S-%f}-b{batch_sequence:010d}-t{source_index:04d}"
diff --git a/evaluation/src/powercontext_eval/web/worker.py b/evaluation/src/powercontext_eval/web/worker.py
index 11f4dbd97..694f5f08e 100644
--- a/evaluation/src/powercontext_eval/web/worker.py
+++ b/evaluation/src/powercontext_eval/web/worker.py
@@ -38,7 +38,7 @@
from powercontext_eval.codex import CodexCapacityError, CodexInfrastructureError, UnsafeCodexInvocation
from powercontext_eval.errors import CommandCancelled, CommandError, GitSourceError, PowerContextEvalError
from powercontext_eval.git_source import GitSource
-from powercontext_eval.models import PowerContextRef
+from powercontext_eval.models import Arm, PowerContextRef
from powercontext_eval.paths import EvaluationPaths
from powercontext_eval.powercontext_sut import (
InvalidTreatment,
@@ -324,6 +324,7 @@ def _batch_run_config(self, task: TaskRecord, cancel_event: threading.Event | No
docker_network_pool=self._config.docker_network_pool,
extra_no_proxy_hosts=self._config.extra_no_proxy_hosts,
run_id=_execution_run_id(task),
+ treatment_mode=task.request.treatment_mode,
model=task.request.model,
reasoning_effort=task.request.reasoning_effort,
finalization_registrar=self._finalization_registrar(task),
@@ -464,6 +465,13 @@ def _validated_result(self, task: TaskRecord, result: MinimalRunResult) -> TaskR
os.close(descriptor)
except (FileNotFoundError, OSError, RuntimeError, ValueError, InvalidReportBundle):
raise InvalidReportBundle("Runner returned an unsafe report path") from None
+ returned_arms = {
+ arm
+ for arm, resolved in ((Arm.OFF, result.off_resolved), (Arm.ON, result.on_resolved))
+ if resolved is not None
+ }
+ if returned_arms != set(task.request.treatment_mode.arms):
+ raise InvalidReportBundle("Runner outcomes do not match the requested treatment mode")
return TaskResult(
artifact_dir=os.fspath(layout.run_artifacts.relative_to(self._config.run_root)),
report_path=os.fspath(expected_report.relative_to(self._config.run_root)),
diff --git a/evaluation/tests/contract/test_codex_contract.py b/evaluation/tests/contract/test_codex_contract.py
index 56b94c0cf..6956f7818 100644
--- a/evaluation/tests/contract/test_codex_contract.py
+++ b/evaluation/tests/contract/test_codex_contract.py
@@ -404,6 +404,7 @@ def __init__(
self,
*,
fail_at: str | None = None,
+ plugin_version: str = "0.1.0",
host_identity: bytes = b"fixture-person\n",
container_identity: bytes | None = None,
host_tokensflow_version: bytes = b"tokensflow 1.0.16\n",
@@ -411,6 +412,7 @@ def __init__(
) -> None:
self.commands: list[tuple[str, ...]] = []
self.fail_at = fail_at
+ self.plugin_version = plugin_version
self.host_identity = host_identity
self.container_identity = container_identity if container_identity is not None else host_identity
self.host_tokensflow_version = host_tokensflow_version
@@ -481,7 +483,7 @@ def run(self, argv: tuple[str, ...], **kwargs: object) -> CommandResult:
"installed": [
{
"pluginId": "powercontext@powercontext",
- "version": "0.1.0",
+ "version": self.plugin_version,
"installed": True,
"enabled": True,
}
@@ -535,10 +537,10 @@ def stop(self) -> None:
self.events.append(("stop", "exact"))
-def sut_config(tmp_path: Path) -> SutConfig:
+def sut_config(tmp_path: Path, *, plugin_version: str = "0.1.0") -> SutConfig:
manifest = tmp_path / "source/integrations/codex/plugins/powercontext/.codex-plugin/plugin.json"
manifest.parent.mkdir(parents=True, exist_ok=True)
- manifest.write_text(json.dumps({"name": "powercontext", "version": "0.1.0"}))
+ manifest.write_text(json.dumps({"name": "powercontext", "version": plugin_version}))
lock = tmp_path / "source/integrations/codex/plugins/powercontext/uv.lock"
lock.write_text("version = 1\n")
tokensflow_binary = tmp_path / "tokensflow"
@@ -948,9 +950,9 @@ def run(self, argv: tuple[str, ...], **kwargs: object) -> CommandResult:
def test_sut_transcript_has_hardening_mount_allowlist_shared_network_and_scope(tmp_path: Path) -> None:
paths = make_paths(tmp_path)
- docker = TranscriptDocker()
+ docker = TranscriptDocker(plugin_version="0.2.0")
relay = FakeRelay()
- config = sut_config(tmp_path)
+ config = sut_config(tmp_path, plugin_version="0.2.0")
config.codex_binary.write_text("binary")
config.uv_binary.write_text("binary")
@@ -1017,7 +1019,7 @@ def test_sut_transcript_has_hardening_mount_allowlist_shared_network_and_scope(t
}
source_provenance = json.loads((paths.result_root / "powercontext/provenance.json").read_text())
assert source_provenance["checkout_sha"] == "a" * 40
- assert source_provenance["plugin_version"] == "0.1.0"
+ assert source_provenance["plugin_version"] == "0.2.0"
assert len(source_provenance["plugin_manifest_sha256"]) == 64
evidence_command = next(command for command in transcript if "evidence" in command)
assert "eval:run-1:on" in evidence_command
@@ -3457,11 +3459,11 @@ def run(self, argv: tuple[str, ...], **kwargs: object) -> CommandResult:
assert docker.commands == [("git", "rev-parse", "--verify", "HEAD^{commit}")]
-def test_manifest_version_mismatch_is_rejected_before_docker(tmp_path: Path) -> None:
+def test_manifest_without_a_version_is_rejected_before_docker(tmp_path: Path) -> None:
paths = make_paths(tmp_path)
config = sut_config(tmp_path)
manifest = config.source_checkout / "integrations/codex/plugins/powercontext/.codex-plugin/plugin.json"
- manifest.write_text(json.dumps({"name": "powercontext", "version": "9.9.9"}))
+ manifest.write_text(json.dumps({"name": "powercontext"}))
docker = TranscriptDocker()
with pytest.raises(InvalidTreatment, match="manifest"):
diff --git a/evaluation/tests/unit/test_cli.py b/evaluation/tests/unit/test_cli.py
index a5602a42d..41622603b 100644
--- a/evaluation/tests/unit/test_cli.py
+++ b/evaluation/tests/unit/test_cli.py
@@ -204,7 +204,7 @@ def fake_run(config: RunConfig, *, instance: object) -> MinimalRunResult:
assert captured[0].proxy_url is None
-def test_cli_creates_a_luna_batch_atomically_paused(monkeypatch) -> None:
+def test_cli_creates_a_single_arm_luna_batch_atomically_paused(monkeypatch) -> None:
calls: list[tuple[Request, float]] = []
class Response:
@@ -238,6 +238,8 @@ def fake_urlopen(request: Request, *, timeout: float) -> Response:
"gpt-5.6-luna",
"--task-set",
"swebench-pro-stability-v1",
+ "--treatment-mode",
+ "on_only",
"--start-paused",
],
)
@@ -252,6 +254,7 @@ def fake_urlopen(request: Request, *, timeout: float) -> Response:
payload = json.loads(request.data)
assert payload["model"] == "gpt-5.6-luna"
assert payload["task_set"] == "swebench-pro-stability-v1"
+ assert payload["treatment_mode"] == "on_only"
assert payload["initial_control_intent"] == "pause"
diff --git a/evaluation/tests/unit/test_runner_phases.py b/evaluation/tests/unit/test_runner_phases.py
index 367931256..27fd9f414 100644
--- a/evaluation/tests/unit/test_runner_phases.py
+++ b/evaluation/tests/unit/test_runner_phases.py
@@ -40,7 +40,7 @@
)
from powercontext_eval.codex import CodexOutcome
from powercontext_eval.errors import CommandFailed
-from powercontext_eval.models import Arm
+from powercontext_eval.models import Arm, TreatmentMode
from powercontext_eval.powercontext_sut import ProxyRelayConfig, SutConfig
from powercontext_eval.process import CommandResult, ProcessRunner
from powercontext_eval.report import ReportBundle
@@ -295,8 +295,9 @@ def _run_with_fakes(
model: str | None = None,
docker_network_pool: str | None = None,
extra_no_proxy_hosts: tuple[str, ...] | None = None,
+ treatment_mode: TreatmentMode = TreatmentMode.OFF_ON,
) -> tuple[RunConfig, MinimalRunResult, dict[str, object]]:
- config = _config(tmp_path)
+ config = replace(_config(tmp_path), treatment_mode=treatment_mode)
if model is not None:
config = replace(config, model=model)
if docker_network_pool is not None:
@@ -396,6 +397,28 @@ def evaluate(self, **kwargs: object) -> OfficialEvaluation:
return OfficialEvaluation(instance.instance_id, True, "", "")
class FakeSut:
+ @staticmethod
+ def _outcome(arm: Arm, store: ArtifactStore) -> object:
+ observed_at = datetime.now(UTC).isoformat(timespec="microseconds").replace("+00:00", "Z")
+ store.write_text(
+ "context/codex-observed.jsonl",
+ f'{{"sequence":1,"observed_at":"{observed_at}","event":{{"type":"agent_message","message":"done"}}}}\n',
+ )
+ events.append(arm)
+ return SimpleNamespace(codex=CodexOutcome("", None))
+
+ def run_arm(
+ self,
+ sut_config: object,
+ arm: Arm,
+ _paths: object,
+ _prompt: bytes,
+ store: ArtifactStore,
+ ) -> object:
+ observed["sut_config"] = sut_config
+ observed["single_arm"] = arm
+ return self._outcome(arm, store)
+
def run_pair(
self,
sut_config: object,
@@ -409,17 +432,11 @@ def run_pair(
stores = cast(dict[Arm, ArtifactStore], kwargs["stores"])
observed["stores"] = stores
assert before_arm is not None
+ outcomes = {}
for arm in (Arm.OFF, Arm.ON):
before_arm(arm)
- events.append(arm)
- observed_at = datetime.now(UTC).isoformat(timespec="microseconds").replace("+00:00", "Z")
- stores[arm].write_text(
- "context/codex-observed.jsonl",
- f'{{"sequence":1,"observed_at":"{observed_at}",'
- '"event":{"type":"agent_message","message":"done"}}\n',
- )
- outcome = SimpleNamespace(codex=CodexOutcome("", None))
- return {Arm.OFF: outcome, Arm.ON: outcome}
+ outcomes[arm] = self._outcome(arm, stores[arm])
+ return outcomes
monkeypatch.setattr("powercontext_eval.runner.ProcessRunner", FakeProcess)
monkeypatch.setattr("powercontext_eval.runner.GitSource", lambda **kwargs: FakeSource())
@@ -447,6 +464,36 @@ def test_runner_propagates_batch_model_to_codex_pair_and_report(
)
assert (sut_config.model, sut_config.reasoning_effort) == ("gpt-5.6-luna", "medium")
assert report.configuration["model"] == "gpt-5.6-luna"
+
+
+@pytest.mark.parametrize(
+ ("mode", "arm"),
+ ((TreatmentMode.ON_ONLY, Arm.ON), (TreatmentMode.OFF_ONLY, Arm.OFF)),
+)
+def test_runner_executes_and_reports_exactly_one_requested_arm(
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ mode: TreatmentMode,
+ arm: Arm,
+) -> None:
+ config, result, observed = _run_with_fakes(
+ tmp_path,
+ monkeypatch,
+ [],
+ treatment_mode=mode,
+ )
+
+ assert observed["single_arm"] is arm
+ assert result.off_resolved is (True if arm is Arm.OFF else None)
+ assert result.on_resolved is (True if arm is Arm.ON else None)
+ report = ReportBundle.model_validate_json(
+ (config.root / "runs" / result.run_id / "report.json").read_text(),
+ strict=True,
+ )
+ assert report.treatment_mode is mode
+ assert (report.off is not None) is (arm is Arm.OFF)
+ assert (report.on is not None) is (arm is Arm.ON)
+ assert len(cast(list[EvaluatorCall], observed["evaluator_calls"])) == 2
assert report.configuration["reasoning_effort"] == "medium"
diff --git a/evaluation/tests/web/test_api.py b/evaluation/tests/web/test_api.py
index d91588e59..af1179f10 100644
--- a/evaluation/tests/web/test_api.py
+++ b/evaluation/tests/web/test_api.py
@@ -154,7 +154,7 @@ def test_health_and_capabilities_are_server_owned_and_secret_free(client: TestCl
assert health_payload.pop("filesystem_min_free_inodes") == 1_000_000
assert health_payload.pop("web_revision") != "unknown"
assert health_payload.pop("worker_revision") is None
- assert health_payload.pop("web_schema_version") == 2
+ assert health_payload.pop("web_schema_version") == 3
assert health_payload.pop("worker_schema_version") is None
assert health_payload.pop("deployment_consistent") is False
assert health_payload == {
@@ -170,12 +170,19 @@ def test_health_and_capabilities_are_server_owned_and_secret_free(client: TestCl
"instances": [INSTANCE],
"models": ["gpt-5.6-sol"],
"reasoning_efforts": ["medium"],
- "treatment_modes": ["off_on"],
+ "treatment_modes": ["off_on", "on_only", "off_only"],
}
assert_safe(health)
assert_safe(capabilities)
+def test_baseline_selection_accepts_a_json_array_before_batch_lookup(client: TestClient) -> None:
+ response = client.put("/api/batches/missing/baseline-selections", json={"selections": []})
+
+ assert response.status_code == 404
+ assert response.json()["error"]["code"] == "batch_not_found"
+
+
def test_health_fails_resource_admission_closed_when_capacity_is_unavailable(
config: WebConfig,
store: TaskStore,
@@ -378,7 +385,7 @@ def test_health_reads_four_active_pairs_and_published_capacity_from_store(
assert health_payload.pop("filesystem_min_free_inodes") == 1_000_000
assert health_payload.pop("web_revision") != "unknown"
assert health_payload.pop("worker_revision") is None
- assert health_payload.pop("web_schema_version") == 2
+ assert health_payload.pop("web_schema_version") == 3
assert health_payload.pop("worker_schema_version") is None
assert health_payload.pop("deployment_consistent") is False
assert health_payload == {
diff --git a/evaluation/tests/web/test_reporting.py b/evaluation/tests/web/test_reporting.py
index cc39ba0b1..0ee1af921 100644
--- a/evaluation/tests/web/test_reporting.py
+++ b/evaluation/tests/web/test_reporting.py
@@ -21,6 +21,7 @@
import pytest
from powercontext_eval.artifacts import ArmState
+from powercontext_eval.models import TreatmentMode
from powercontext_eval.report import ArmReport, MetricSet, ReportBundle
from powercontext_eval.web.reporting import InvalidReportArtifact, UnsafeReportPath, load_raw_report, load_report
@@ -117,6 +118,37 @@ def test_loads_validated_report_and_derives_exact_comparisons(tmp_path: Path) ->
assert dict(response.configuration) == dict(_bundle().configuration)
+@pytest.mark.parametrize("mode", (TreatmentMode.ON_ONLY, TreatmentMode.OFF_ONLY))
+def test_loads_single_arm_report_without_inventing_the_missing_arm(
+ tmp_path: Path,
+ mode: TreatmentMode,
+) -> None:
+ arm = "on" if mode is TreatmentMode.ON_ONLY else "off"
+ source = _bundle().on if arm == "on" else _bundle().off
+ assert source is not None
+ bundle = ReportBundle(
+ title="single arm",
+ revisions=_bundle().revisions,
+ configuration=_bundle().configuration,
+ treatment_mode=mode,
+ off=source if arm == "off" else None,
+ on=source if arm == "on" else None,
+ )
+ runs_root = tmp_path / "runs"
+ run_dir = runs_root / "run-single"
+ evidence_dir = run_dir / "arms" / arm / "powercontext"
+ evidence_dir.mkdir(parents=True)
+ (evidence_dir / "treatment.json").write_text(json.dumps(_evidence("run-single", arm)))
+ (run_dir / "report.json").write_text(bundle.model_dump_json())
+
+ response = load_report(run_dir, runs_root)
+
+ assert response.treatment_mode is mode
+ assert (response.off is not None) is (arm == "off")
+ assert (response.on is not None) is (arm == "on")
+ assert response.comparison is None
+
+
@pytest.mark.parametrize(
("relative_path", "contents"),
[
diff --git a/evaluation/tests/web/test_store.py b/evaluation/tests/web/test_store.py
index 011b27ff9..a6cfccd30 100644
--- a/evaluation/tests/web/test_store.py
+++ b/evaluation/tests/web/test_store.py
@@ -21,7 +21,14 @@
import pytest
+from powercontext_eval.models import Arm, TreatmentMode
from powercontext_eval.paths import EvaluationPaths
+from powercontext_eval.web.baselines import (
+ BaselineCreate,
+ BaselineItemRecord,
+ BaselineSelection,
+ BaselineSnapshot,
+)
from powercontext_eval.web.batches import BatchControlEventType, BatchCreate, BatchStatus
from powercontext_eval.web.controls import BatchControlIntent, BatchPauseReason
from powercontext_eval.web.models import (
@@ -72,6 +79,37 @@ def batch_request(key: str, *, model: str = "gpt-5.6-sol") -> BatchCreate:
)
+def baseline_snapshot(*, resolved: bool) -> BaselineSnapshot:
+ return BaselineSnapshot(
+ benchmark="swebench-pro",
+ task_set="swebench-pro-public-v2",
+ instance_set_digest="d" * 64,
+ total_tasks=1,
+ resolved_tasks=int(resolved),
+ execution_failures=0,
+ model="gpt-5.6-sol",
+ reasoning_effort="medium",
+ dataset_revision="dataset-v1",
+ harness_revision="harness-v1",
+ powercontext_sha="a" * 40,
+ codex_version="0.145.0",
+ items=(
+ BaselineItemRecord(
+ baseline_id="",
+ instance_id="instance-one",
+ source_index=0,
+ source_task_id="source-task",
+ source_attempt_id="source-task.attempt-0001",
+ status=TaskStatus.SUCCEEDED,
+ resolved=resolved,
+ input_tokens=10,
+ output_tokens=5,
+ total_tokens=15,
+ ),
+ ),
+ )
+
+
def usage_snapshot(*, used_percent: int, observed_at: datetime = NOW) -> UsageSnapshot:
return UsageSnapshot(
limit_id="codex",
@@ -2021,3 +2059,76 @@ def test_batch_record_round_trips_container_env(store: TaskStore) -> None:
"OPENROUTER_API_KEY": "sk-test",
"POWERCONTEXT_SERVER_HTTP_PORT": "8000",
}
+
+
+def test_baselines_are_single_arm_immutable_and_listed_newest_first(store: TaskStore) -> None:
+ source, _ = store.create_batch(batch_request("baseline-source"), ("instance_a",), now=NOW)
+ older, created = store.create_baseline(
+ BaselineCreate(
+ name="Older ON",
+ source_batch_id=source.batch_id,
+ source_arm=Arm.ON,
+ expected_report_revision=101,
+ idempotency_key="baseline-older",
+ ),
+ baseline_snapshot(resolved=False),
+ now=NOW,
+ )
+ newer, _ = store.create_baseline(
+ BaselineCreate(
+ name="Newer OFF",
+ source_batch_id=source.batch_id,
+ source_arm=Arm.OFF,
+ expected_report_revision=102,
+ idempotency_key="baseline-newer",
+ ),
+ baseline_snapshot(resolved=True),
+ now=NOW + timedelta(seconds=1),
+ )
+
+ assert created is True
+ assert [baseline.baseline_id for baseline in store.list_baselines()] == [newer.baseline_id, older.baseline_id]
+ assert store.list_baseline_items(older.baseline_id)[0].resolved is False
+ assert store.list_baseline_items(newer.baseline_id)[0].resolved is True
+
+
+def test_baseline_selections_are_multiple_and_do_not_change_batch_execution(store: TaskStore) -> None:
+ request = batch_request("single-arm-selection").model_copy(
+ update={"treatment_mode": TreatmentMode.ON_ONLY},
+ )
+ batch, _ = store.create_batch(request, ("instance_a",), now=NOW)
+ first, _ = store.create_baseline(
+ BaselineCreate(
+ name="First",
+ source_batch_id=batch.batch_id,
+ source_arm=Arm.ON,
+ expected_report_revision=1,
+ idempotency_key="selection-first",
+ ),
+ baseline_snapshot(resolved=True),
+ now=NOW,
+ )
+ second, _ = store.create_baseline(
+ BaselineCreate(
+ name="Second",
+ source_batch_id=batch.batch_id,
+ source_arm=Arm.OFF,
+ expected_report_revision=1,
+ idempotency_key="selection-second",
+ ),
+ baseline_snapshot(resolved=False),
+ now=NOW + timedelta(seconds=1),
+ )
+
+ selected = store.replace_baseline_selections(
+ batch.batch_id,
+ (
+ BaselineSelection(baseline_id=first.baseline_id, current_arm=Arm.ON),
+ BaselineSelection(baseline_id=second.baseline_id, current_arm=Arm.ON),
+ ),
+ now=NOW + timedelta(seconds=2),
+ )
+
+ assert selected == store.list_baseline_selections(batch.batch_id)
+ assert store.get_batch(batch.batch_id).request.treatment_mode is TreatmentMode.ON_ONLY
+ assert all(task.status is TaskStatus.QUEUED for task in store.list_batch_tasks(batch.batch_id))
diff --git a/evaluation/web/src/App.test.tsx b/evaluation/web/src/App.test.tsx
index b9b7eaea3..668cf6f42 100644
--- a/evaluation/web/src/App.test.tsx
+++ b/evaluation/web/src/App.test.tsx
@@ -32,14 +32,14 @@ describe("App batch report navigation", () => {
expect(screen.getByRole("link", { name: "任务详细报告" })).toHaveAttribute("aria-disabled", "true");
expect(screen.getByRole("link", { name: "当前运行任务" })).toHaveAttribute("aria-disabled", "true");
expect(screen.getAllByRole("navigation")).toHaveLength(1);
- expect(screen.getAllByRole("link")).toHaveLength(4);
+ expect(screen.getAllByRole("link")).toHaveLength(5);
expect(screen.queryByRole("link", { name: /工作台|测试任务|验收报告|单任务详情/ })).not.toBeInTheDocument();
- expect(await screen.findByRole("button", { name: "预览评测" })).toBeVisible();
+ expect(await screen.findByRole("button", { name: "开始评测" })).toBeVisible();
expect(await screen.findByText("Worker 工作中")).toBeVisible();
expect(screen.getByText("任务对 3 / 4")).toBeVisible();
expect(screen.getByText("队列 1")).toBeVisible();
expect(screen.getByText("资源门禁开放")).toBeVisible();
- expect(screen.getByText("Worker 按配置并行运行独立任务对")).toBeVisible();
+ expect(screen.getByText("Worker 按配置并行运行独立任务")).toBeVisible();
expect(screen.queryByText("全局同时只运行一个任务,其余任务排队")).not.toBeInTheDocument();
});
@@ -97,8 +97,7 @@ describe("App batch report navigation", () => {
const api = apiStub({ createBatch: vi.fn().mockResolvedValue(created) });
render(
基线库暂时无法加载。
Historical baselines
+每条基线冻结一个历史 Arm;默认按保存时间从新到旧排列。
+暂无基线,请从已完成批次的总体报告保存。
: ( +| 名称 | Arm | 解决率 | 任务集 | 模型 | +PowerContext | 保存时间 | 来源 | +
|---|---|---|---|---|---|---|---|
| {baseline.name}{baseline.baseline_id} | +{baseline.source_arm.toUpperCase()} | +{baseline.resolved_tasks} / {baseline.total_tasks} | +{baseline.task_set} | +{baseline.model} · {baseline.reasoning_effort} | +{baseline.powercontext_sha?.slice(0, 12) ?? "—"} | +{new Date(baseline.created_at).toLocaleString("zh-CN", { hour12: false })} | +onLink(event, path)}>查看总体报告 | +
确认信息
-当前用量已达到暂停阈值
- )} - -{message}
}可比较任务 {number(report.comparable_pairs)} / {number(report.total_tasks)}
+可比较任务 {number(comparablePairs)} / {number(report.total_tasks)}
可选择多个基线;增删对比不会重新运行任务。
+没有兼容基线。可以在下方先保存当前结果。
+ ) : candidates.filter((candidate) => candidate.compatibility.status !== "incompatible").map((candidate) => ( + + ))}尚未选择基线;当前报告只展示本次真实结果。
+ )}每次保存一个不可变 Arm 快照。
{baselineMessage}
} +{number(measured ?? 0)} / {number(total)} 个任务有记录
+{taskId} · 按实际观察时间排序
报告保留实际结果,但不构成有效验收结论。
} -当前报告不具备有效的 OFF / ON 对照数据。
}{task.repository}
{task.failure_summary}
}任务未执行,因此没有上下文时间线。
- ) : !hasComparison ? ( + ) : off === null && on === null ? (@@ -266,6 +265,10 @@ export function TaskRunDetail({ api, batchId, taskId, search, navigate }: TaskRu batchId={batchId} taskId={taskId} {...(task.attempt_id === null ? {} : { attemptId: task.attempt_id })} + availableArms={[ + ...(off === null ? [] : ["off" as const]), + ...(on === null ? [] : ["on" as const]), + ]} /> )}