From b02b8ab61990c1da7243be024193ba553c885e92 Mon Sep 17 00:00:00 2001 From: Fengzdadi <453788063@qq.com> Date: Fri, 25 Sep 2026 16:39:40 -0400 Subject: [PATCH 1/5] feat(e2e): compare PowerContext off and on for Bub continuation workloads Add a paired command that runs continuation workloads with PowerContext off and on, in separate containers, over several trials with the arm order alternating, and reports paired outcomes. - OFF installs Bub without the PowerContext plugin and passes no POWERCONTEXT_* settings. ON binds the plugin to a new Scope and enables event capture, so Bub captures what the user says like the other host integrations do. - After each ON session, a Harbor agent-end hook flushes the Scope and snapshots its Server statistics. The hook runs after the agent's timed phase, so the flush does not use the agent's time budget. - An ON run counts only when Sources were captured and turned into Memory before the recall session and PowerContext supplied context during it. Integration failures and infrastructure errors are counted but not scored; agent timeouts count as failures in both arms. - The first continuation task mentions a decision only in conversation and asks for it in a later session. The recall step's own tests hold the answer key, so the earlier session cannot read it. Refs #1705 Co-Authored-By: Claude Opus 5.5 --- Makefile | 5 + e2e/bub/README.md | 54 ++++ .../environment/Dockerfile | 19 ++ .../steps/capture/instruction.md | 6 + .../steps/capture/tests/test.sh | 23 ++ .../steps/recall/instruction.md | 2 + .../steps/recall/tests/grade.py | 44 +++ .../steps/recall/tests/test.sh | 18 ++ .../project-decision-continuation/task.toml | 31 ++ .../project-decision-continuation.yaml | 30 ++ e2e/bub/src/powercontext_e2e/__main__.py | 31 ++ e2e/bub/src/powercontext_e2e/catalog.py | 12 +- e2e/bub/src/powercontext_e2e/harbor_agent.py | 9 +- e2e/bub/src/powercontext_e2e/hosts.py | 53 ++-- e2e/bub/src/powercontext_e2e/models.py | 68 ++++ e2e/bub/src/powercontext_e2e/paired.py | 294 ++++++++++++++++++ e2e/bub/src/powercontext_e2e/report.py | 40 ++- e2e/bub/src/powercontext_e2e/runner.py | 21 +- e2e/bub/src/powercontext_e2e/sessions.py | 72 +++++ e2e/bub/tests/test_harbor_agent.py | 7 + e2e/bub/tests/test_harbor_job_config.py | 18 ++ e2e/bub/tests/test_paired.py | 250 +++++++++++++++ 22 files changed, 1074 insertions(+), 33 deletions(-) create mode 100644 e2e/bub/harbor-tasks/project-decision-continuation/environment/Dockerfile create mode 100644 e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/instruction.md create mode 100644 e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/tests/test.sh create mode 100644 e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md create mode 100644 e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py create mode 100644 e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh create mode 100644 e2e/bub/harbor-tasks/project-decision-continuation/task.toml create mode 100644 e2e/bub/paired-tasks/project-decision-continuation.yaml create mode 100644 e2e/bub/src/powercontext_e2e/paired.py create mode 100644 e2e/bub/src/powercontext_e2e/sessions.py create mode 100644 e2e/bub/tests/test_paired.py diff --git a/Makefile b/Makefile index 977b2c8668..378fdd1852 100644 --- a/Makefile +++ b/Makefile @@ -97,6 +97,11 @@ harness-acceptance: ## Evaluate workloads by ID or category against an existing @uv run --project e2e/bub powercontext-e2e acceptance \ --output "$${POWERCONTEXT_E2E_OUTPUT:-e2e/bub/results}" $(ARGS) +.PHONY: harness-paired +harness-paired: ## Compare PowerContext off and on for continuation workloads against an existing Server. + @uv run --project e2e/bub powercontext-e2e paired \ + --output "$${POWERCONTEXT_E2E_OUTPUT:-e2e/bub/results/paired}" $(ARGS) + .PHONY: harness-rescore harness-rescore: ## Rescore REPLAY without rerunning Bub or PowerContext. @test -n "$${REPLAY:-}" || { echo "REPLAY is required" >&2; exit 2; } diff --git a/e2e/bub/README.md b/e2e/bub/README.md index 199c04e1e7..8ca37cfeae 100644 --- a/e2e/bub/README.md +++ b/e2e/bub/README.md @@ -31,6 +31,7 @@ summary. ```text e2e/bub/ tasks/ # PowerContext manifests and evaluation expectations + paired-tasks/ # OFF/ON continuation manifests for the paired command harbor-tasks/ # Local Harbor tasks used by built-in samples src/powercontext_e2e/ # One Harbor runner and one Memory evaluator ``` @@ -139,6 +140,59 @@ evaluation and report at `batch-/`. `collect-all` reports every failed tas Harbor trial at its first failed step. Runtime batch steps are flat and task-prefixed. Each agent invocation starts an independent ACP session and Bub tape. +## Compare PowerContext off and on + +A continuation workload is a Harbor multi-step task written in plain language, so any agent host can run it. An +earlier session mentions a fact only in the conversation, next to an unrelated small job. The final recall session +asks for that fact and has the agent write its answer to a file. The recall step's own tests grade the answer, and +the answer key lives only there, because Harbor leaves every uploaded test directory in the container for later +steps. The task reward is the final step's reward. + +The `paired` command runs each selected workload with PowerContext off and on, in separate containers, and repeats +this for `--trials` trials. The arm that runs first alternates between trials. + +- OFF installs the host without its PowerContext integration and passes no `POWERCONTEXT_*` settings. +- ON installs the integration bound to a new Scope. For Bub this means the plugin with `capture_events` enabled, so + that, like the other host integrations, it captures what the user says without relying on the model to call a + memory tool. This is not the plugin's default setting. +- Everything else is the same in both arms: image, host version, model, and budget. + +After each ON session the harness flushes the Scope, standing in for the time that passes between real sessions. It +repeats the flush until the Scope has processed every captured Source, a flush makes no progress, or 20 rounds pass. +The flush runs from a Harbor agent-end hook after the agent's timed phase, so it does not use the agent's time +budget. Host plugins flush on different schedules, so the harness flushes the same way for every host. The Server's +generation model therefore takes part in the ON arm; the run fails early when the Server does not report +`memory_extraction`. + +An ON run counts only when Server statistics for its Scope show that Sources were captured and turned into Memory +before the recall session, and that PowerContext supplied context during it. Otherwise it is an integration failure. +Integration failures and harness or infrastructure errors are reported but left out of success rates and paired +differences. An agent timeout counts as a failed attempt in either arm. + +```bash +export POWERCONTEXT_CLIENT_SERVER_URL=http://127.0.0.1:8000 +export POWERCONTEXT_BUB_BASE_URL=http://host-gateway:8000 +export POWERCONTEXT_BUB_TRUST_TRANSPORT_SECURITY=true +export BUB_MODEL=openrouter:openai/gpt-5.4 +export BUB_API_KEY="$OPENROUTER_API_KEY" +make harness-paired ARGS='--trials 2' +``` + +Each arm writes `observation.json`, which includes the per-session Server snapshots for ON, and its Harbor jobs: + +```text +/ + paired-report.json + report.md + /trial-// + observation.json + harbor-jobs/ +``` + +The command exits non-zero when any arm could not be scored; a task that fails in either arm is a result, not a +command failure. The report is marked preliminary. It does not yet estimate uncertainty, check the Default Scope for +leaks, record latency or token usage, or run in the fixed Compose harness. + ## Long-horizon task The Terminal-Bench manifest pins its task checksum, model requirement, step budget, capture cadence, recall probes, diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/environment/Dockerfile b/e2e/bub/harbor-tasks/project-decision-continuation/environment/Dockerfile new file mode 100644 index 0000000000..9c07a63352 --- /dev/null +++ b/e2e/bub/harbor-tasks/project-decision-continuation/environment/Dockerfile @@ -0,0 +1,19 @@ +# 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. + +FROM python:3.12-slim-bookworm + +WORKDIR /workspace + +RUN printf '# Ticket service\n\nThe service will recieve tickets from the support queue.\n' > /workspace/README.md diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/instruction.md b/e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/instruction.md new file mode 100644 index 0000000000..27e48aafa2 --- /dev/null +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/instruction.md @@ -0,0 +1,6 @@ +Some context before we start: last week the team decided that the ticket service database will run on OceanBase, +with the shard count fixed at 12, because it has to stay compatible with our existing MySQL clients. No action is +needed on that yet. + +For now, please fix the spelling mistake "recieve" in /workspace/README.md; it should be "receive". Do not change +anything else. diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/tests/test.sh b/e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/tests/test.sh new file mode 100644 index 0000000000..89974d16c5 --- /dev/null +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/capture/tests/test.sh @@ -0,0 +1,23 @@ +#!/bin/sh +# 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. + +set -eu + +# Diagnostic only: the final step's reward decides the trial. +if grep -q 'receive' /workspace/README.md && ! grep -q 'recieve' /workspace/README.md; then + echo 1 > /logs/verifier/reward.txt +else + echo 0 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md new file mode 100644 index 0000000000..5181d47f69 --- /dev/null +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md @@ -0,0 +1,2 @@ +Which database did we decide to use for the ticket service, and what shard count did we agree on? Write your answer +as one short sentence to /workspace/answer.txt. If you do not know, write "unknown". diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py new file mode 100644 index 0000000000..9aac7346fa --- /dev/null +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py @@ -0,0 +1,44 @@ +# 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. + +"""Grade the recall answer without depending on the agent host. + +This file is uploaded only with the recall step's tests, so no earlier session can read the expected answer. +""" + +from __future__ import annotations + +import re +import sys +import unicodedata +from pathlib import Path + +# Every group must match; any alternative inside a group is enough. +REQUIRED = (("oceanbase",), ("12", "twelve")) + + +def score(answer: str) -> int: + text = unicodedata.normalize("NFC", answer.casefold()) + return int( + all(any(re.search(rf"\b{re.escape(term)}\b", text) for term in group) for group in REQUIRED), + ) + + +def main(answer_path: Path, reward_path: Path) -> None: + answer = answer_path.read_text(encoding="utf-8", errors="replace") if answer_path.is_file() else "" + reward_path.write_text(f"{score(answer)}\n", encoding="utf-8") + + +if __name__ == "__main__": + main(Path(sys.argv[1]), Path(sys.argv[2])) diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh new file mode 100644 index 0000000000..2fe882fd4a --- /dev/null +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh @@ -0,0 +1,18 @@ +#!/bin/sh +# 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. + +set -eu + +python3 /tests/grade.py /workspace/answer.txt /logs/verifier/reward.txt diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/task.toml b/e2e/bub/harbor-tasks/project-decision-continuation/task.toml new file mode 100644 index 0000000000..eb0ee11d80 --- /dev/null +++ b/e2e/bub/harbor-tasks/project-decision-continuation/task.toml @@ -0,0 +1,31 @@ +# 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. + +version = "1.3" +multi_step_reward_strategy = "final" + +[agent] +timeout_sec = 600.0 + +[verifier] +timeout_sec = 60.0 + +[environment] +build_timeout_sec = 300.0 + +[[steps]] +name = "capture" + +[[steps]] +name = "recall" diff --git a/e2e/bub/paired-tasks/project-decision-continuation.yaml b/e2e/bub/paired-tasks/project-decision-continuation.yaml new file mode 100644 index 0000000000..4156574250 --- /dev/null +++ b/e2e/bub/paired-tasks/project-decision-continuation.yaml @@ -0,0 +1,30 @@ +# 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. + +schema: powercontext.e2e-task/v1 +id: project-decision-continuation +categories: + - paired + - sample +dataset: + path: e2e/bub/harbor-tasks + task_id: project-decision-continuation + checksum: c517d77f6fdaf4b7675357b3fd826fc42e24caf2604deb4258223711a1652812 +execution: + type: bub + model: true + max_steps: 30 + max_tokens: 16384 +evaluation: + recall_step: recall diff --git a/e2e/bub/src/powercontext_e2e/__main__.py b/e2e/bub/src/powercontext_e2e/__main__.py index 529683f959..7eb876c44e 100644 --- a/e2e/bub/src/powercontext_e2e/__main__.py +++ b/e2e/bub/src/powercontext_e2e/__main__.py @@ -84,6 +84,24 @@ def main() -> None: help="Continue through case failures or stop the Harbor trial at the first failed step.", ) + paired_parser = subparsers.add_parser("paired", help="Compare PowerContext off and on for continuation workloads.") + paired_parser.add_argument("--manifest", type=Path, default=Path("e2e/bub/paired-tasks")) + paired_parser.add_argument( + "--id", + action="append", + default=[], + metavar="WORKLOAD_ID", + help="Select one workload; repeat to select more than one. Defaults to the paired category.", + ) + paired_parser.add_argument( + "--category", + action="append", + default=[], + help="Select one category; repeat to select more than one.", + ) + paired_parser.add_argument("--trials", type=int, default=2, help="Trials per arm; the arm order alternates.") + paired_parser.add_argument("--output", type=Path, required=True) + rescore_parser = subparsers.add_parser("rescore") rescore_parser.add_argument("replay", type=Path) rescore_parser.add_argument("--output", type=Path, required=True) @@ -94,6 +112,19 @@ def main() -> None: from .rescore import rescore_replay passed = rescore_replay(args.replay, args.output, settings) + elif args.command == "paired": + from .catalog import load_tasks, select_tasks + from .paired import run_paired + + selected = select_tasks( + load_tasks(args.manifest), + ids=tuple(args.id), + categories=tuple(args.category) or (() if args.id else ("paired",)), + ) + report = asyncio.run(run_paired(selected, output_dir=args.output, settings=settings, trials=args.trials)) + # The run is valid when every arm produced a measurement; the task outcome itself is the result. + unscored = report.total.off.errors + report.total.on.errors + report.total.on.integration_failures + passed = unscored == 0 else: from .catalog import load_tasks, select_tasks from .runner import run_tasks diff --git a/e2e/bub/src/powercontext_e2e/catalog.py b/e2e/bub/src/powercontext_e2e/catalog.py index 1c701267a2..001e94a9b8 100644 --- a/e2e/bub/src/powercontext_e2e/catalog.py +++ b/e2e/bub/src/powercontext_e2e/catalog.py @@ -123,6 +123,16 @@ class OutcomeEvaluationSpec(CatalogModel): expected_execution: ExpectedExecutionSpec +class ContinuationEvaluationSpec(CatalogModel): + """Compare PowerContext off and on for a task whose final session depends on an earlier one. + + The Harbor task's own verifier grades the final step. ``recall_step`` names that step so the harness can check + that PowerContext supplied context during it. + """ + + recall_step: str = Field(min_length=1) + + class E2ETask(CatalogModel): schema_: Literal["powercontext.e2e-task/v1"] = Field(alias="schema") id: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]*$") @@ -130,7 +140,7 @@ class E2ETask(CatalogModel): provenance: Provenance | None = None dataset: HarborDatasetSpec execution: BubExecutionSpec - evaluation: MemoryEvaluationSpec | OutcomeEvaluationSpec + evaluation: MemoryEvaluationSpec | OutcomeEvaluationSpec | ContinuationEvaluationSpec class TaskSelectionError(ValueError): diff --git a/e2e/bub/src/powercontext_e2e/harbor_agent.py b/e2e/bub/src/powercontext_e2e/harbor_agent.py index b0cd144290..cb18c413ef 100644 --- a/e2e/bub/src/powercontext_e2e/harbor_agent.py +++ b/e2e/bub/src/powercontext_e2e/harbor_agent.py @@ -45,6 +45,7 @@ class PowerContextBubAcpAgent(harbor_acp.AcpAgent): def __init__(self, **kwargs: Any) -> None: self._invocation_scopes = tuple(kwargs.pop("invocation_scopes", ())) + self._powercontext = bool(kwargs.pop("powercontext", True)) self._step_index = 0 super().__init__( registry_entry={ @@ -86,7 +87,7 @@ async def install(self, environment: BaseEnvironment) -> None: command=self._build_dependencies_command("uvx"), env={"DEBIAN_FRONTEND": "noninteractive"}, ) - await self.exec_as_root(environment, command=_install_bub_command()) + await self.exec_as_root(environment, command=_install_bub_command(powercontext=self._powercontext)) await self.exec_as_root(environment, command=_install_acp_server_command()) agent_user = shlex.quote(str(environment.default_user or "root")) await self.exec_as_root( @@ -119,8 +120,9 @@ def _tool_environment() -> str: return f"UV_TOOL_BIN_DIR={shlex.quote(REMOTE_BIN_DIR)} UV_TOOL_DIR={shlex.quote(REMOTE_TOOL_DIR)}" -def _install_bub_command() -> str: +def _install_bub_command(*, powercontext: bool = True) -> str: uv = f"{harbor_acp.AcpAgent._RUNNER_VENV_PATH}/bin/uv" + plugin = f"--overrides {REMOTE_SOURCE_OVERRIDE} --with {REMOTE_SOURCE}/integrations/bub " if powercontext else "" return ( "set -eu; " f"mkdir -p {REMOTE_BIN_DIR} {REMOTE_BUB_HOME} {REMOTE_BUB_PROJECT} {REMOTE_CODEX_HOME}; " @@ -129,8 +131,7 @@ def _install_bub_command() -> str: f"chmod 600 {REMOTE_CODEX_HOME}/auth.json; " "fi; " f"SETUPTOOLS_SCM_PRETEND_VERSION={shlex.quote(POWERCONTEXT_VERSION)} {_tool_environment()} " - f"{uv} tool install --force " - f"--overrides {REMOTE_SOURCE_OVERRIDE} --with {REMOTE_SOURCE}/integrations/bub " + f"{uv} tool install --force {plugin}" f"{shlex.quote(f'bub=={BUB_VERSION}')}" ) diff --git a/e2e/bub/src/powercontext_e2e/hosts.py b/e2e/bub/src/powercontext_e2e/hosts.py index e1137c0b4b..0024159f76 100644 --- a/e2e/bub/src/powercontext_e2e/hosts.py +++ b/e2e/bub/src/powercontext_e2e/hosts.py @@ -20,7 +20,7 @@ from harbor.models.trial.config import AgentConfig, ServiceVolumeConfig -from .catalog import E2ETask, MemoryEvaluationSpec +from .catalog import ContinuationEvaluationSpec, E2ETask, MemoryEvaluationSpec from .harbor_agent import BUB_ACP_SERVER_VERSION, BUB_VERSION, REMOTE_CODEX_AUTH from .settings import bub_environment, codex_auth_path, powercontext_bub_environment @@ -44,10 +44,15 @@ def agent_config( self, task: E2ETask, *, - scope_id: str, + scope_id: str | None, invocation_scopes: tuple[str, ...] | None, ) -> AgentConfig: - """Bind the host to one job Scope, or to one Scope per agent invocation when given.""" + """Configure the host agent for one Harbor job. + + With ``scope_id``, the host runs with its PowerContext integration bound to that Scope, or to one Scope per + agent invocation when ``invocation_scopes`` is given. Without it, the host runs with no PowerContext + integration installed. + """ class BubHost: @@ -79,11 +84,10 @@ def agent_config( self, task: E2ETask, *, - scope_id: str, + scope_id: str | None, invocation_scopes: tuple[str, ...] | None, ) -> AgentConfig: - evaluation = task.evaluation - env = powercontext_bub_environment() + env = powercontext_bub_environment() if scope_id is not None else {} if task.execution.model: env.update(bub_environment()) else: @@ -93,22 +97,22 @@ def agent_config( "BUB_MAX_STEPS": str(task.execution.max_steps), "BUB_MAX_TOKENS": str(task.execution.max_tokens), "CODEX_HOME": "/installed-agent/codex", - "POWERCONTEXT_BUB_CAPTURE_CHECKPOINT_EVERY": str( - evaluation.checkpoint_every_events if isinstance(evaluation, MemoryEvaluationSpec) else 5 - ), - "POWERCONTEXT_BUB_CAPTURE_EVENTS": str( - evaluation.capture_events if isinstance(evaluation, MemoryEvaluationSpec) else False - ).lower(), - "POWERCONTEXT_BUB_CAPTURE_LOG": "/logs/agent/powercontext-capture.jsonl", - "POWERCONTEXT_BUB_CAPTURE_MAX_BYTES": str( - evaluation.max_event_bytes if isinstance(evaluation, MemoryEvaluationSpec) else 8192 - ), - "POWERCONTEXT_BUB_SCOPE_ID": scope_id, }) kwargs: dict[str, Any] = {} - if invocation_scopes is not None: - env.pop("POWERCONTEXT_BUB_SCOPE_ID") - kwargs["invocation_scopes"] = invocation_scopes + if scope_id is None: + kwargs["powercontext"] = False + else: + capture_events, checkpoint_every, max_bytes = _capture_settings(task) + env.update({ + "POWERCONTEXT_BUB_CAPTURE_CHECKPOINT_EVERY": str(checkpoint_every), + "POWERCONTEXT_BUB_CAPTURE_EVENTS": str(capture_events).lower(), + "POWERCONTEXT_BUB_CAPTURE_LOG": "/logs/agent/powercontext-capture.jsonl", + "POWERCONTEXT_BUB_CAPTURE_MAX_BYTES": str(max_bytes), + "POWERCONTEXT_BUB_SCOPE_ID": scope_id, + }) + if invocation_scopes is not None: + env.pop("POWERCONTEXT_BUB_SCOPE_ID") + kwargs["invocation_scopes"] = invocation_scopes return AgentConfig( import_path="powercontext_e2e.harbor_agent:PowerContextBubAcpAgent", env=env, @@ -116,6 +120,15 @@ def agent_config( ) +def _capture_settings(task: E2ETask) -> tuple[bool, int, int]: + evaluation = task.evaluation + if isinstance(evaluation, MemoryEvaluationSpec): + return evaluation.capture_events, evaluation.checkpoint_every_events, evaluation.max_event_bytes + # Bub captures nothing automatically by default. The ON arm records every turn so that, like the other hosts' + # integrations, it captures what the user says without relying on the model to call a memory tool. + return isinstance(evaluation, ContinuationEvaluationSpec), 5, 8192 + + _HOSTS: dict[str, HostAdapter] = {"bub": BubHost()} diff --git a/e2e/bub/src/powercontext_e2e/models.py b/e2e/bub/src/powercontext_e2e/models.py index 51da606c3b..a5f90ad108 100644 --- a/e2e/bub/src/powercontext_e2e/models.py +++ b/e2e/bub/src/powercontext_e2e/models.py @@ -90,6 +90,18 @@ class RecallProbeObservation(EvidenceModel): forbidden_context_matched: bool | None = None +class SessionSnapshot(EvidenceModel): + """Server-side state of one Scope after an agent session and the flush that followed it.""" + + session: int = Field(ge=0) + flush_rounds: int = Field(ge=0) + sources: int = Field(ge=0) + memory_pending: int = Field(ge=0) + memory_entries: int = Field(ge=0) + preparations: int = Field(ge=0) + ready_preparations: int = Field(ge=0) + + class HarborTrialObservation(EvidenceModel): job_id: str | None = None trial_name: str | None = None @@ -160,3 +172,59 @@ class EvaluationReport(EvidenceModel): @property def accepted(self) -> bool: return all(bool(result.value) for case in self.cases for result in case.assertions.values()) + + +Arm = Literal["off", "on"] +ArmOutcome = Literal["passed", "failed", "timeout", "error", "integration_failed"] + + +class PairedArmObservation(EvidenceModel): + """One arm of one OFF/ON trial for a continuation workload.""" + + schema_: Literal["powercontext.e2e-paired-arm/v1"] = Field( + default="powercontext.e2e-paired-arm/v1", + alias="schema", + ) + run_id: str + task_id: str + trial: int = Field(ge=1) + arm: Arm + position: int = Field(ge=1, le=2, description="Whether this arm ran first or second within its trial.") + environment: RunEnvironment + scope_id: str | None = None + harbor: HarborTrialObservation + step_rewards: dict[str, float] = Field(default_factory=dict) + outcome: ArmOutcome + errors: tuple[str, ...] = () + sessions: tuple[SessionSnapshot, ...] = () + treatment_failures: tuple[str, ...] = () + + +class ArmSummary(EvidenceModel): + scored: int = Field(ge=0, description="Runs that count toward the success rate: passed, failed, or timed out.") + passed: int = Field(ge=0) + timeouts: int = Field(ge=0) + errors: int = Field(ge=0) + integration_failures: int = Field(ge=0) + + +class PairedSummary(EvidenceModel): + off: ArmSummary + on: ArmSummary + pairs: int = Field(ge=0, description="Trials in which both arms were scored.") + mean_delta: float | None = Field(default=None, description="Mean ON minus OFF score over scored pairs.") + + +class PairedTaskSummary(PairedSummary): + task_id: str + + +class PairedReport(EvidenceModel): + schema_: Literal["powercontext.e2e-paired-report/v1"] = Field( + default="powercontext.e2e-paired-report/v1", + alias="schema", + ) + experiment: str + trials: int = Field(ge=1) + tasks: tuple[PairedTaskSummary, ...] = Field(min_length=1) + total: PairedSummary diff --git a/e2e/bub/src/powercontext_e2e/paired.py b/e2e/bub/src/powercontext_e2e/paired.py new file mode 100644 index 0000000000..589bbad8ee --- /dev/null +++ b/e2e/bub/src/powercontext_e2e/paired.py @@ -0,0 +1,294 @@ +# 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. + +"""Run continuation workloads with PowerContext off and on, and report the paired outcomes.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from statistics import fmean +from typing import TYPE_CHECKING +from uuid import uuid4 + +from harbor.job import Job +from powercontext.http import CreateScopeRequest + +from .catalog import ContinuationEvaluationSpec, E2ETask +from .evidence import redact, write_evidence +from .models import ( + Arm, + ArmOutcome, + ArmSummary, + HarborTrialObservation, + PairedArmObservation, + PairedReport, + PairedSummary, + PairedTaskSummary, + SessionSnapshot, +) +from .report import render_paired_report +from .runner import ( + _harbor_observation, + _job_config, + _load_source_task, + _powercontext_client, + _run_environment, + require_runtime_models, +) +from .sessions import SessionRecorder +from .settings import HarnessSettings + +if TYPE_CHECKING: + from collections.abc import Sequence + from pathlib import Path + + from harbor.models.trial.result import StepResult + from powercontext.client import PowerContextClient + +SCORES: dict[ArmOutcome, int] = {"passed": 1, "failed": 0, "timeout": 0} +AGENT_TIMEOUT = "AgentTimeoutError" + + +class MemoryExtractionUnavailableError(RuntimeError): + """Report a Server that cannot turn captured Sources into Memory, which the ON arm depends on.""" + + def __init__(self) -> None: + super().__init__("The PowerContext Server does not report memory_extraction; the ON arm cannot recall") + + +async def run_paired( + tasks: tuple[E2ETask, ...], + *, + output_dir: Path, + settings: HarnessSettings, + trials: int, +) -> PairedReport: + """Run every task ``trials`` times per arm, alternating which arm goes first, and write the paired report.""" + + if not tasks or trials < 1: + raise ValueError("At least one continuation workload and one trial are required") # noqa: TRY003 + recall_sessions = {task.id: recall_session_index(task, settings) for task in tasks} + require_runtime_models(tasks) + + observations: list[PairedArmObservation] = [] + async with _powercontext_client() as client: + await client.get_readiness() + if not (await client.get_capabilities()).memory_extraction: + raise MemoryExtractionUnavailableError + for task in tasks: + for trial in range(1, trials + 1): + order: tuple[Arm, Arm] = ("off", "on") if trial % 2 else ("on", "off") + for position, arm in enumerate(order, start=1): + arm_dir = output_dir / task.id / f"trial-{trial}" / arm + observation = await _run_arm( + client, + task, + trial=trial, + arm=arm, + position=position, + recall_session=recall_sessions[task.id], + output_dir=arm_dir, + settings=settings, + ) + write_evidence( + arm_dir / "observation.json", + observation.model_dump_json(by_alias=True, indent=2) + "\n", + settings, + ) + observations.append(observation) + + report = summarize(observations, trials=trials) + write_evidence(output_dir / "paired-report.json", report.model_dump_json(by_alias=True, indent=2) + "\n", settings) + write_evidence(output_dir / "report.md", render_paired_report(report), settings) + return report + + +def recall_session_index(task: E2ETask, settings: HarnessSettings) -> int: + """Return the zero-based agent session that answers from earlier sessions.""" + + evaluation = task.evaluation + if not isinstance(evaluation, ContinuationEvaluationSpec): + raise TypeError(f"Workload {task.id!r} is not an OFF/ON continuation workload") # noqa: TRY003 + steps = _load_source_task(task, settings.repository_path()).source_steps + if len(steps) < 2 or steps[-1] != evaluation.recall_step: + raise ValueError( # noqa: TRY003 + f"Workload {task.id!r} must end with its recall step {evaluation.recall_step!r} after an earlier session" + ) + return len(steps) - 1 + + +async def _run_arm( + client: PowerContextClient, + task: E2ETask, + *, + trial: int, + arm: Arm, + position: int, + recall_session: int, + output_dir: Path, + settings: HarnessSettings, +) -> PairedArmObservation: + run_id = f"{task.id}-t{trial}-{arm}-{uuid4().hex[:12]}" + output_dir.mkdir(parents=True, exist_ok=True) + started_at = datetime.now(UTC) + scope_id: str | None = None + recorder: SessionRecorder | None = None + harbor = HarborTrialObservation() + step_results: tuple[StepResult, ...] = () + errors: list[str] = [] + try: + if arm == "on": + scope_id = ( + await client.create_scope( + CreateScopeRequest( + title=f"E2E paired workload: {task.id}", + summary=f"Isolated Scope for the ON arm of {task.id}, trial {trial}, in E2E run {run_id}.", + idempotency_key=f"e2e:{run_id}:{task.id}", + ) + ) + ).scope_id + job = await Job.create(_job_config(task, run_id, scope_id, output_dir, settings)) + if scope_id is not None: + recorder = SessionRecorder(client, scope_id) + job.on_agent_ended(recorder) + harbor, step_results, _ = _harbor_observation(await job.run(), settings) + except Exception as exc: + errors.append(redact(f"{type(exc).__name__}: {exc}", settings)) + + sessions = () if recorder is None else tuple(recorder.snapshots) + treatment = treatment_failures(sessions, recall_session) if arm == "on" else () + exception_types = tuple( + name + for name in ( + *(step.exception_info.exception_type for step in step_results if step.exception_info is not None), + harbor.exception_type, + ) + if name is not None + ) + reward = harbor.rewards.get("reward") + return PairedArmObservation( + run_id=run_id, + task_id=task.id, + trial=trial, + arm=arm, + position=position, + environment=_run_environment(task, started_at, settings), + scope_id=scope_id, + harbor=harbor, + step_rewards={ + step.step_name: float(step.verifier_result.rewards["reward"]) + for step in step_results + if step.verifier_result is not None + and step.verifier_result.rewards + and "reward" in step.verifier_result.rewards + }, + outcome=classify_outcome( + harness_failed=bool(errors), + exception_types=exception_types, + treatment_failures=treatment, + reward=None if reward is None else float(reward), + ), + errors=tuple(errors), + sessions=sessions, + treatment_failures=treatment, + ) + + +def treatment_failures(sessions: Sequence[SessionSnapshot], recall_session: int) -> tuple[str, ...]: + """Explain why an ON run did not receive PowerContext's treatment, or return nothing when it did.""" + + by_session = {snapshot.session: snapshot for snapshot in sessions} + before = by_session.get(recall_session - 1) + recall = by_session.get(recall_session) + if before is None or recall is None: + return ("The Server was not observed after every session",) + failures: list[str] = [] + if before.sources == 0: + failures.append("No Sources were captured before the recall session") + if before.memory_entries == 0: + failures.append("The flush before the recall session created no Memory") + if recall.ready_preparations <= before.ready_preparations: + failures.append("PowerContext supplied no context during the recall session") + return tuple(failures) + + +def classify_outcome( + *, + harness_failed: bool, + exception_types: Sequence[str], + treatment_failures: Sequence[str], + reward: float | None, +) -> ArmOutcome: + """Classify one arm run. + + An agent timeout counts as a failed attempt in both arms. Harness and infrastructure errors, and ON runs that did + not receive PowerContext's treatment, are not measurements of the task, so they are counted separately and left + out of the success rate and the paired difference. + """ + + if harness_failed: + return "error" + if AGENT_TIMEOUT in exception_types: + return "timeout" + if exception_types: + return "error" + if treatment_failures: + return "integration_failed" + return "passed" if reward is not None and reward >= 1 else "failed" + + +def summarize(observations: Sequence[PairedArmObservation], *, trials: int) -> PairedReport: + task_ids = tuple(dict.fromkeys(observation.task_id for observation in observations)) + return PairedReport( + experiment="e2e:paired:" + ",".join(task_ids), + trials=trials, + tasks=tuple( + PairedTaskSummary( + task_id=task_id, + **_summary([o for o in observations if o.task_id == task_id]).model_dump(), + ) + for task_id in task_ids + ), + total=_summary(observations), + ) + + +def _summary(observations: Sequence[PairedArmObservation]) -> PairedSummary: + scores = { + (observation.task_id, observation.trial, observation.arm): SCORES[observation.outcome] + for observation in observations + if observation.outcome in SCORES + } + deltas = [ + scores[task_id, trial, "on"] - scores[task_id, trial, "off"] + for task_id, trial in dict.fromkeys((o.task_id, o.trial) for o in observations) + if (task_id, trial, "on") in scores and (task_id, trial, "off") in scores + ] + return PairedSummary( + off=_arm_summary([o for o in observations if o.arm == "off"]), + on=_arm_summary([o for o in observations if o.arm == "on"]), + pairs=len(deltas), + mean_delta=fmean(deltas) if deltas else None, + ) + + +def _arm_summary(observations: Sequence[PairedArmObservation]) -> ArmSummary: + outcomes = [observation.outcome for observation in observations] + return ArmSummary( + scored=sum(outcome in SCORES for outcome in outcomes), + passed=outcomes.count("passed"), + timeouts=outcomes.count("timeout"), + errors=outcomes.count("error"), + integration_failures=outcomes.count("integration_failed"), + ) diff --git a/e2e/bub/src/powercontext_e2e/report.py b/e2e/bub/src/powercontext_e2e/report.py index bb2b3a3a57..3bbc85cfca 100644 --- a/e2e/bub/src/powercontext_e2e/report.py +++ b/e2e/bub/src/powercontext_e2e/report.py @@ -20,7 +20,7 @@ from marko.element import Element from marko.md_renderer import MarkdownRenderer -from .models import EvaluationReport, TaskObservation +from .models import ArmSummary, EvaluationReport, PairedReport, PairedSummary, TaskObservation def render_report(observation: TaskObservation, report: EvaluationReport) -> str: @@ -68,6 +68,44 @@ def render_evaluation_summary(report: EvaluationReport) -> str: return markdown.render(document) +def render_paired_report(report: PairedReport) -> str: + markdown = Markdown(renderer=MarkdownRenderer) + document = block.Document() + children: list[Element] = [ + *_nodes(markdown, "# PowerContext OFF/ON comparison"), + block.BlankLine(0), + *_nodes( + markdown, + f"Preliminary: {report.trials} trial(s) per arm and no uncertainty estimate. Errors and ON runs that did " + "not receive PowerContext's treatment are counted but left out of success rates and paired differences.", + ), + ] + for title, summary in ( + *((f"`{task.task_id}`", task) for task in report.tasks), + ("All workloads", report.total), + ): + children.extend((block.BlankLine(0), *_nodes(markdown, f"## {title}"), block.BlankLine(0))) + children.extend(_nodes(markdown, _paired_summary_text(summary))) + document.children = children + return markdown.render(document) + + +def _paired_summary_text(summary: PairedSummary) -> str: + delta = "n/a" if summary.mean_delta is None else f"{summary.mean_delta:+.2f}" + return "\n".join(( + f"- OFF: {_arm_text(summary.off)}", + f"- ON: {_arm_text(summary.on)}", + f"- Scored pairs: {summary.pairs}; mean ON minus OFF: {delta}", + )) + + +def _arm_text(arm: ArmSummary) -> str: + return ( + f"{arm.passed}/{arm.scored} passed ({arm.timeouts} timed out); " + f"{arm.errors} error(s), {arm.integration_failures} integration failure(s) not scored" + ) + + def _nodes(markdown: Markdown, source: str) -> list[Element]: return list(markdown.parse(source).children) diff --git a/e2e/bub/src/powercontext_e2e/runner.py b/e2e/bub/src/powercontext_e2e/runner.py index 4c0d38a8a6..3686d50ced 100644 --- a/e2e/bub/src/powercontext_e2e/runner.py +++ b/e2e/bub/src/powercontext_e2e/runner.py @@ -42,7 +42,7 @@ from powercontext.http import CreateScopeRequest, ListMemoryEntriesRequest, PrepareContextRequest from .artifacts import write_artifacts -from .catalog import E2ETask, MemoryEvaluationSpec, OutcomeEvaluationSpec +from .catalog import ContinuationEvaluationSpec, E2ETask, MemoryEvaluationSpec, OutcomeEvaluationSpec from .evaluation import evaluate_observation, matches_forbidden_context from .evidence import fingerprint, load_resolved_instructions, redact, write_evaluation_report, write_evidence from .hosts import host_adapter @@ -151,11 +151,9 @@ async def run_tasks( settings: HarnessSettings, failure_policy: FailurePolicy = "collect-all", ) -> bool: - model_workload_ids = tuple( - task.id for task in tasks if task.execution.model and not host_adapter(task).model_configured() - ) - if model_workload_ids: - raise ModelNotConfiguredError(model_workload_ids) + if continuation_ids := [task.id for task in tasks if isinstance(task.evaluation, ContinuationEvaluationSpec)]: + raise ValueError(f"Run OFF/ON continuation workloads with the paired command: {continuation_ids!r}") # noqa: TRY003 + require_runtime_models(tasks) accepted = True for group in group_tasks(tasks): @@ -168,6 +166,15 @@ async def run_tasks( return accepted +def require_runtime_models(tasks: tuple[E2ETask, ...]) -> None: + """Reject model-backed workloads whose host has no runtime-selected model.""" + + if model_workload_ids := tuple( + task.id for task in tasks if task.execution.model and not host_adapter(task).model_configured() + ): + raise ModelNotConfiguredError(model_workload_ids) + + def group_tasks(tasks: tuple[E2ETask, ...]) -> tuple[ExecutionGroup, ...]: """Group selected tasks only when they share an explicit batch category.""" @@ -408,7 +415,7 @@ def _source_harbor_observation( def _job_config( task: E2ETask, run_id: str, - scope_id: str, + scope_id: str | None, output_dir: Path, settings: HarnessSettings, *, diff --git a/e2e/bub/src/powercontext_e2e/sessions.py b/e2e/bub/src/powercontext_e2e/sessions.py new file mode 100644 index 0000000000..8a649dfe35 --- /dev/null +++ b/e2e/bub/src/powercontext_e2e/sessions.py @@ -0,0 +1,72 @@ +# 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. + +"""Settle a PowerContext Scope between agent sessions and record what the Server observed.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from powercontext.http import FlushMemoryRequest, GetStatsRequest + +from .models import SessionSnapshot + +if TYPE_CHECKING: + from harbor.trial.hooks import TrialHookEvent + from powercontext.client import PowerContextClient + +MAX_FLUSH_ROUNDS = 20 + + +async def settle_session(client: PowerContextClient, scope_id: str, session: int) -> SessionSnapshot: + """Flush captured Sources into Memory, as elapsed time would between real sessions, then snapshot the Scope. + + Host plugins flush on different schedules, so the harness flushes the same way for every host. + """ + + rounds = 0 + while rounds < MAX_FLUSH_ROUNDS: + response = await client.flush_memory(FlushMemoryRequest(scope_id=scope_id)) + rounds += 1 + if response.current_cursor >= response.high_watermark or response.current_cursor <= response.previous_cursor: + break + stats = await client.get_stats( + GetStatsRequest.model_validate({"selection": {"mode": "exact", "scope_ids": [scope_id]}}) + ) + return SessionSnapshot( + session=session, + flush_rounds=rounds, + sources=stats.inventory.sources.total, + memory_pending=stats.inventory.sources.memory_pending, + memory_entries=stats.inventory.memory.entries.total, + preparations=stats.recall.totals.preparations, + ready_preparations=stats.recall.totals.ready_preparations, + ) + + +class SessionRecorder: + """Harbor agent-end hook that settles one Scope after every agent session of a single-trial job. + + Harbor fires the hook after the agent's timed phase, so the flush neither uses the agent's time budget nor + appears in its execution time. + """ + + def __init__(self, client: PowerContextClient, scope_id: str) -> None: + self._client = client + self._scope_id = scope_id + self.snapshots: list[SessionSnapshot] = [] + + async def __call__(self, event: TrialHookEvent) -> None: + del event + self.snapshots.append(await settle_session(self._client, self._scope_id, len(self.snapshots))) diff --git a/e2e/bub/tests/test_harbor_agent.py b/e2e/bub/tests/test_harbor_agent.py index 3aafe4b72d..94fe3b9b85 100644 --- a/e2e/bub/tests/test_harbor_agent.py +++ b/e2e/bub/tests/test_harbor_agent.py @@ -35,3 +35,10 @@ def test_install_bub_command_overrides_release_floor_for_mounted_source() -> Non assert _SOURCE_OVERRIDE.read_text(encoding="utf-8").splitlines()[-1] == ( "powercontext[client] @ file:///opt/powercontext/source" ) + + +def test_off_arm_installs_bub_without_the_powercontext_plugin() -> None: + command = _install_bub_command(powercontext=False) + + assert "integrations/bub" not in command + assert f"bub=={version('bub')}" in command diff --git a/e2e/bub/tests/test_harbor_job_config.py b/e2e/bub/tests/test_harbor_job_config.py index 1ea50a528f..a0ccb70adf 100644 --- a/e2e/bub/tests/test_harbor_job_config.py +++ b/e2e/bub/tests/test_harbor_job_config.py @@ -166,3 +166,21 @@ def test_model_workload_requires_a_runtime_model(tmp_path: Path) -> None: asyncio.run(run_tasks((task,), output_dir=output_dir, settings=HarnessSettings(repository=_REPOSITORY))) assert not output_dir.exists() + + +def test_off_arm_runs_the_host_without_powercontext(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setenv("BUB_MODEL", "provider:model") + monkeypatch.setenv("POWERCONTEXT_BUB_BASE_URL", "http://host-gateway:8000") + task = load_tasks(_REPOSITORY / "e2e" / "bub" / "paired-tasks" / "project-decision-continuation.yaml")[0] + + off = _job_config(task, "run-1", None, tmp_path / "off", HarnessSettings(repository=_REPOSITORY)) + on = _job_config(task, "run-1", "scope-1", tmp_path / "on", HarnessSettings(repository=_REPOSITORY)) + + (off_agent,) = off.agents + (on_agent,) = on.agents + assert not [name for name in off_agent.env if name.startswith("POWERCONTEXT_")] + assert off_agent.kwargs == {"powercontext": False} + assert off_agent.env["BUB_MODEL"] == on_agent.env["BUB_MODEL"] == "provider:model" + assert on_agent.env["POWERCONTEXT_BUB_SCOPE_ID"] == "scope-1" + assert on_agent.env["POWERCONTEXT_BUB_CAPTURE_EVENTS"] == "true" + assert on_agent.kwargs == {} diff --git a/e2e/bub/tests/test_paired.py b/e2e/bub/tests/test_paired.py new file mode 100644 index 0000000000..6c6f1d48b0 --- /dev/null +++ b/e2e/bub/tests/test_paired.py @@ -0,0 +1,250 @@ +# 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. + +"""OFF/ON continuation workloads: grading, treatment checks, outcome rules, and the paired summary.""" + +from __future__ import annotations + +import asyncio +import runpy +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from powercontext_e2e.catalog import load_tasks +from powercontext_e2e.models import HarborTrialObservation, PairedArmObservation, RunEnvironment, SessionSnapshot +from powercontext_e2e.paired import classify_outcome, recall_session_index, summarize, treatment_failures +from powercontext_e2e.runner import run_tasks +from powercontext_e2e.sessions import SessionRecorder, settle_session +from powercontext_e2e.settings import HarnessSettings + +_REPOSITORY = Path(__file__).resolve().parents[3] +_PAIRED_TASKS = load_tasks(_REPOSITORY / "e2e" / "bub" / "paired-tasks") +_HARBOR_TASKS = _REPOSITORY / "e2e" / "bub" / "harbor-tasks" +_SETTINGS = HarnessSettings(repository=_REPOSITORY) + + +def _grade(answer_path: Path, reward_path: Path) -> None: + # run_path does not write bytecode, which would change the Harbor task checksum. + grader = runpy.run_path( + str(_HARBOR_TASKS / "project-decision-continuation" / "steps" / "recall" / "tests" / "grade.py") + ) + grader["main"](answer_path, reward_path) + + +@pytest.mark.parametrize( + ("answer", "reward"), + [ + ("We chose OceanBase with a shard count of 12.", 1), + ("oceanbase, twelve shards", 1), + ("OceanBase with 120 shards.", 0), + ("PostgreSQL with 12 shards.", 0), + ("unknown", 0), + ], +) +def test_recall_grader_requires_every_fact(tmp_path: Path, answer: str, reward: int) -> None: + answer_path = tmp_path / "answer.txt" + answer_path.write_text(answer, encoding="utf-8") + reward_path = tmp_path / "reward.txt" + + _grade(answer_path, reward_path) + + assert reward_path.read_text(encoding="utf-8") == f"{reward}\n" + + +def test_recall_grader_scores_a_missing_answer_as_zero(tmp_path: Path) -> None: + reward_path = tmp_path / "reward.txt" + + _grade(tmp_path / "answer.txt", reward_path) + + assert reward_path.read_text(encoding="utf-8") == "0\n" + + +@pytest.mark.parametrize("task", _PAIRED_TASKS, ids=lambda task: task.id) +def test_continuation_tasks_hide_the_answer_until_the_recall_session(task) -> None: + # Harbor uploads shared tests before every step and leaves them in the container, so an answer key there would + # be readable in the earlier session. + assert recall_session_index(task, _SETTINGS) >= 1 + assert not (_HARBOR_TASKS / task.dataset.task_id / "tests").exists() + + +def test_recall_step_must_be_the_final_session() -> None: + task = _PAIRED_TASKS[0] + task = task.model_copy(update={"evaluation": task.evaluation.model_copy(update={"recall_step": "capture"})}) + + with pytest.raises(ValueError, match="must end with its recall step"): + recall_session_index(task, _SETTINGS) + + +def test_acceptance_rejects_continuation_workloads(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="paired command"): + asyncio.run(run_tasks(_PAIRED_TASKS, output_dir=tmp_path / "out", settings=_SETTINGS)) + + +def _snapshot(session: int, *, sources: int = 1, memory: int = 1, ready: int = 0) -> SessionSnapshot: + return SessionSnapshot( + session=session, + flush_rounds=1, + sources=sources, + memory_pending=0, + memory_entries=memory, + preparations=ready, + ready_preparations=ready, + ) + + +def test_treatment_passes_when_recall_received_context_from_captured_memory() -> None: + assert treatment_failures((_snapshot(0, ready=1), _snapshot(1, ready=2)), recall_session=1) == () + + +@pytest.mark.parametrize( + ("sessions", "failure"), + [ + ((_snapshot(0, sources=0, memory=0), _snapshot(1, ready=1)), "No Sources were captured"), + ((_snapshot(0, memory=0), _snapshot(1, ready=1)), "created no Memory"), + # Context supplied late in the capture session does not show that the recall session received any. + ((_snapshot(0, ready=2), _snapshot(1, ready=2)), "no context during the recall session"), + ((_snapshot(0),), "not observed after every session"), + ], +) +def test_treatment_fails_when_any_link_from_capture_to_recall_is_missing(sessions, failure: str) -> None: + assert any(failure in reason for reason in treatment_failures(sessions, recall_session=1)) + + +@pytest.mark.parametrize( + ("kwargs", "outcome"), + [ + ({"reward": 1.0}, "passed"), + ({"reward": 0.0}, "failed"), + ({"reward": None}, "failed"), + ({"exception_types": ("AgentTimeoutError",)}, "timeout"), + ({"exception_types": ("EnvironmentStartTimeoutError",)}, "error"), + ({"harness_failed": True, "reward": 1.0}, "error"), + ({"treatment_failures": ("no context",), "reward": 1.0}, "integration_failed"), + # A timed-out ON run has no final snapshot; it still counts as a failed attempt, as it would with OFF. + ({"exception_types": ("AgentTimeoutError",), "treatment_failures": ("not observed",)}, "timeout"), + ], +) +def test_outcome_classification(kwargs, outcome: str) -> None: + arguments = {"harness_failed": False, "exception_types": (), "treatment_failures": (), "reward": None} | kwargs + + assert classify_outcome(**arguments) == outcome + + +def _observation(trial: int, arm: str, outcome: str, task_id: str = "task") -> PairedArmObservation: + now = datetime.now(UTC) + return PairedArmObservation( + run_id=f"{task_id}-{trial}-{arm}", + task_id=task_id, + trial=trial, + arm=arm, + position=1, + environment=RunEnvironment( + commit="c", + database="sqlite", + adapter_version="a", + adapter_protocol_version="p", + started_at=now, + finished_at=now, + ), + harbor=HarborTrialObservation(), + outcome=outcome, + ) + + +def test_summary_pairs_only_trials_where_both_arms_were_scored() -> None: + report = summarize( + ( + _observation(1, "off", "failed"), + _observation(1, "on", "passed"), + _observation(2, "off", "timeout"), + _observation(2, "on", "integration_failed"), + _observation(3, "off", "passed"), + _observation(3, "on", "passed"), + ), + trials=3, + ) + + (task,) = report.tasks + assert (task.off.passed, task.off.scored, task.off.timeouts) == (1, 3, 1) + assert (task.on.passed, task.on.scored, task.on.integration_failures) == (2, 2, 1) + assert task.pairs == 2 + assert task.mean_delta == 0.5 + assert (report.total.pairs, report.total.mean_delta) == (2, 0.5) + + +def test_summary_reports_no_difference_without_a_scored_pair() -> None: + report = summarize((_observation(1, "off", "error"), _observation(1, "on", "passed")), trials=1) + + assert report.total.pairs == 0 + assert report.total.mean_delta is None + assert report.total.off.errors == 1 + + +class _FlushingClient: + def __init__(self, cursors: list[tuple[int, int, int]]) -> None: + self._cursors = iter(cursors) + self.flushes = 0 + + async def flush_memory(self, request): + self.flushes += 1 + previous, current, high = next(self._cursors) + return SimpleNamespace(previous_cursor=previous, current_cursor=current, high_watermark=high) + + async def get_stats(self, request): + assert request.selection.root.scope_ids[0].root == "scope-1" + return SimpleNamespace( + inventory=SimpleNamespace( + sources=SimpleNamespace(total=3, memory_pending=0), + memory=SimpleNamespace(entries=SimpleNamespace(total=2)), + ), + recall=SimpleNamespace(totals=SimpleNamespace(preparations=4, ready_preparations=1)), + ) + + +def test_settling_flushes_until_the_scope_is_caught_up() -> None: + client = _FlushingClient([(0, 1, 3), (1, 2, 3), (2, 3, 3)]) + + snapshot = asyncio.run(settle_session(client, "scope-1", session=0)) + + assert client.flushes == 3 + assert snapshot == SessionSnapshot( + session=0, + flush_rounds=3, + sources=3, + memory_pending=0, + memory_entries=2, + preparations=4, + ready_preparations=1, + ) + + +def test_settling_stops_when_a_flush_makes_no_progress() -> None: + client = _FlushingClient([(0, 1, 3), (1, 1, 3), (1, 2, 3)]) + + snapshot = asyncio.run(settle_session(client, "scope-1", session=1)) + + assert client.flushes == 2 + assert snapshot.flush_rounds == 2 + + +def test_session_recorder_numbers_sessions_in_the_order_harbor_ends_them() -> None: + recorder = SessionRecorder(_FlushingClient([(0, 1, 1), (1, 2, 2)]), "scope-1") + + asyncio.run(recorder(None)) + asyncio.run(recorder(None)) + + assert [snapshot.session for snapshot in recorder.snapshots] == [0, 1] From d32d9b11e21d9017df6c61ea6cf1e3f4557c1118 Mon Sep 17 00:00:00 2001 From: Fengzdadi <453788063@qq.com> Date: Sat, 26 Sep 2026 15:27:19 -0400 Subject: [PATCH 2/5] fix(e2e): keep agent outcomes when settling a paired Scope fails Harbor awaits the agent-end hook in a finally block, so an exception from the flush or statistics read replaced the agent's own outcome: a timed-out ON session became an unscored error. Record settle failures as treatment failures instead of raising. Skip the flush after the final session, where only the statistics snapshot is read, and document the Client and Bub timeouts the flush needs in the paired run example. Co-Authored-By: Claude Opus 5.5 --- e2e/bub/README.md | 17 ++++++++++---- e2e/bub/src/powercontext_e2e/paired.py | 13 ++++++++--- e2e/bub/src/powercontext_e2e/sessions.py | 24 ++++++++++++++------ e2e/bub/tests/test_paired.py | 29 +++++++++++++++++++----- 4 files changed, 62 insertions(+), 21 deletions(-) diff --git a/e2e/bub/README.md b/e2e/bub/README.md index 8ca37cfeae..48cd5f507b 100644 --- a/e2e/bub/README.md +++ b/e2e/bub/README.md @@ -157,11 +157,13 @@ this for `--trials` trials. The arm that runs first alternates between trials. memory tool. This is not the plugin's default setting. - Everything else is the same in both arms: image, host version, model, and budget. -After each ON session the harness flushes the Scope, standing in for the time that passes between real sessions. It -repeats the flush until the Scope has processed every captured Source, a flush makes no progress, or 20 rounds pass. -The flush runs from a Harbor agent-end hook after the agent's timed phase, so it does not use the agent's time -budget. Host plugins flush on different schedules, so the harness flushes the same way for every host. The Server's -generation model therefore takes part in the ON arm; the run fails early when the Server does not report +After each ON session the harness records the Scope's Server statistics. When another session follows, it first +flushes the Scope, standing in for the time that passes between real sessions, and repeats the flush until the Scope +has processed every captured Source, a flush makes no progress, or 20 rounds pass. This runs from a Harbor agent-end +hook after the agent's timed phase, so it does not use the agent's time budget. A failed flush or statistics read is +recorded as a treatment failure rather than replacing the agent's own outcome, so a timed-out session still counts as +a timeout. Host plugins flush on different schedules, so the harness flushes the same way for every host. The +Server's generation model therefore takes part in the ON arm; the run fails early when the Server does not report `memory_extraction`. An ON run counts only when Server statistics for its Scope show that Sources were captured and turned into Memory @@ -169,9 +171,14 @@ before the recall session, and that PowerContext supplied context during it. Oth Integration failures and harness or infrastructure errors are reported but left out of success rates and paired differences. An agent timeout counts as a failed attempt in either arm. +The harness Client waits for each flush, which runs the Server's generation model, so raise its 10-second default +timeout; the Bub plugin also flushes during a session. + ```bash export POWERCONTEXT_CLIENT_SERVER_URL=http://127.0.0.1:8000 +export POWERCONTEXT_CLIENT_TIMEOUT=150 export POWERCONTEXT_BUB_BASE_URL=http://host-gateway:8000 +export POWERCONTEXT_BUB_TIMEOUT=150 export POWERCONTEXT_BUB_TRUST_TRANSPORT_SECURITY=true export BUB_MODEL=openrouter:openai/gpt-5.4 export BUB_API_KEY="$OPENROUTER_API_KEY" diff --git a/e2e/bub/src/powercontext_e2e/paired.py b/e2e/bub/src/powercontext_e2e/paired.py index 589bbad8ee..86342c3535 100644 --- a/e2e/bub/src/powercontext_e2e/paired.py +++ b/e2e/bub/src/powercontext_e2e/paired.py @@ -160,14 +160,21 @@ async def _run_arm( ).scope_id job = await Job.create(_job_config(task, run_id, scope_id, output_dir, settings)) if scope_id is not None: - recorder = SessionRecorder(client, scope_id) + recorder = SessionRecorder(client, scope_id, final_session=recall_session) job.on_agent_ended(recorder) harbor, step_results, _ = _harbor_observation(await job.run(), settings) except Exception as exc: errors.append(redact(f"{type(exc).__name__}: {exc}", settings)) sessions = () if recorder is None else tuple(recorder.snapshots) - treatment = treatment_failures(sessions, recall_session) if arm == "on" else () + treatment = ( + ( + *(redact(failure, settings) for failure in (recorder.failures if recorder is not None else ())), + *treatment_failures(sessions, recall_session), + ) + if arm == "on" + else () + ) exception_types = tuple( name for name in ( @@ -256,7 +263,7 @@ def summarize(observations: Sequence[PairedArmObservation], *, trials: int) -> P tasks=tuple( PairedTaskSummary( task_id=task_id, - **_summary([o for o in observations if o.task_id == task_id]).model_dump(), + **dict(_summary([o for o in observations if o.task_id == task_id])), ) for task_id in task_ids ), diff --git a/e2e/bub/src/powercontext_e2e/sessions.py b/e2e/bub/src/powercontext_e2e/sessions.py index 8a649dfe35..3c4c036dd1 100644 --- a/e2e/bub/src/powercontext_e2e/sessions.py +++ b/e2e/bub/src/powercontext_e2e/sessions.py @@ -29,14 +29,15 @@ MAX_FLUSH_ROUNDS = 20 -async def settle_session(client: PowerContextClient, scope_id: str, session: int) -> SessionSnapshot: - """Flush captured Sources into Memory, as elapsed time would between real sessions, then snapshot the Scope. +async def settle_session(client: PowerContextClient, scope_id: str, session: int, *, flush: bool) -> SessionSnapshot: + """Snapshot the Scope after one session, first flushing captured Sources into Memory when a session follows. - Host plugins flush on different schedules, so the harness flushes the same way for every host. + The flush stands in for the time that passes between real sessions. Host plugins flush on different schedules, so + the harness flushes the same way for every host. """ rounds = 0 - while rounds < MAX_FLUSH_ROUNDS: + while flush and rounds < MAX_FLUSH_ROUNDS: response = await client.flush_memory(FlushMemoryRequest(scope_id=scope_id)) rounds += 1 if response.current_cursor >= response.high_watermark or response.current_cursor <= response.previous_cursor: @@ -59,14 +60,23 @@ class SessionRecorder: """Harbor agent-end hook that settles one Scope after every agent session of a single-trial job. Harbor fires the hook after the agent's timed phase, so the flush neither uses the agent's time budget nor - appears in its execution time. + appears in its execution time. Harbor awaits the hook in a ``finally`` block, where an exception would replace the + agent's own, such as a timeout, so failures are recorded instead of raised. """ - def __init__(self, client: PowerContextClient, scope_id: str) -> None: + def __init__(self, client: PowerContextClient, scope_id: str, *, final_session: int) -> None: self._client = client self._scope_id = scope_id + self._final_session = final_session self.snapshots: list[SessionSnapshot] = [] + self.failures: list[str] = [] async def __call__(self, event: TrialHookEvent) -> None: del event - self.snapshots.append(await settle_session(self._client, self._scope_id, len(self.snapshots))) + session = len(self.snapshots) + len(self.failures) + try: + snapshot = await settle_session(self._client, self._scope_id, session, flush=session < self._final_session) + except Exception as exc: + self.failures.append(f"Settling the Scope after session {session} failed: {type(exc).__name__}: {exc}") + else: + self.snapshots.append(snapshot) diff --git a/e2e/bub/tests/test_paired.py b/e2e/bub/tests/test_paired.py index 6c6f1d48b0..bfac6f2725 100644 --- a/e2e/bub/tests/test_paired.py +++ b/e2e/bub/tests/test_paired.py @@ -195,12 +195,15 @@ def test_summary_reports_no_difference_without_a_scored_pair() -> None: class _FlushingClient: - def __init__(self, cursors: list[tuple[int, int, int]]) -> None: + def __init__(self, cursors: list[tuple[int, int, int]], *, failing_flushes: frozenset[int] = frozenset()) -> None: self._cursors = iter(cursors) + self._failing_flushes = failing_flushes self.flushes = 0 async def flush_memory(self, request): self.flushes += 1 + if self.flushes in self._failing_flushes: + raise TimeoutError("flush timed out") # noqa: TRY003 previous, current, high = next(self._cursors) return SimpleNamespace(previous_cursor=previous, current_cursor=current, high_watermark=high) @@ -218,7 +221,7 @@ async def get_stats(self, request): def test_settling_flushes_until_the_scope_is_caught_up() -> None: client = _FlushingClient([(0, 1, 3), (1, 2, 3), (2, 3, 3)]) - snapshot = asyncio.run(settle_session(client, "scope-1", session=0)) + snapshot = asyncio.run(settle_session(client, "scope-1", session=0, flush=True)) assert client.flushes == 3 assert snapshot == SessionSnapshot( @@ -235,16 +238,30 @@ def test_settling_flushes_until_the_scope_is_caught_up() -> None: def test_settling_stops_when_a_flush_makes_no_progress() -> None: client = _FlushingClient([(0, 1, 3), (1, 1, 3), (1, 2, 3)]) - snapshot = asyncio.run(settle_session(client, "scope-1", session=1)) + snapshot = asyncio.run(settle_session(client, "scope-1", session=1, flush=True)) assert client.flushes == 2 assert snapshot.flush_rounds == 2 -def test_session_recorder_numbers_sessions_in_the_order_harbor_ends_them() -> None: - recorder = SessionRecorder(_FlushingClient([(0, 1, 1), (1, 2, 2)]), "scope-1") +def test_recorder_flushes_before_later_sessions_but_only_snapshots_the_final_one() -> None: + client = _FlushingClient([(0, 1, 1)]) + recorder = SessionRecorder(client, "scope-1", final_session=1) asyncio.run(recorder(None)) asyncio.run(recorder(None)) - assert [snapshot.session for snapshot in recorder.snapshots] == [0, 1] + assert client.flushes == 1 + assert [(snapshot.session, snapshot.flush_rounds) for snapshot in recorder.snapshots] == [(0, 1), (1, 0)] + + +def test_recorder_records_a_failed_settle_instead_of_raising() -> None: + # Harbor awaits the hook in a finally block, where raising would replace a timed-out agent's own exception. + recorder = SessionRecorder(_FlushingClient([], failing_flushes=frozenset({1})), "scope-1", final_session=1) + + asyncio.run(recorder(None)) + asyncio.run(recorder(None)) + + assert recorder.failures == ["Settling the Scope after session 0 failed: TimeoutError: flush timed out"] + assert [snapshot.session for snapshot in recorder.snapshots] == [1] + assert any("not observed" in reason for reason in treatment_failures(recorder.snapshots, recall_session=1)) From c138cca0877770eae4a0a375d675859b9bab11f4 Mon Sep 17 00:00:00 2001 From: Fengzdadi <453788063@qq.com> Date: Sat, 26 Sep 2026 16:17:26 -0400 Subject: [PATCH 3/5] fix(e2e): count ON runs that PowerContext cannot help as attempts The treatment check required a flush to create Memory and the recall session to receive context. Both are PowerContext's own behavior under the treatment, so an ON run in which PowerContext kept or returned nothing was excluded as an integration failure instead of being scored, which inflates the ON success rate. Gate only on the integration: Sources captured before the recall session and a context request during it. The snapshots still record Memory and ready context. Co-Authored-By: Claude Opus 5.5 --- e2e/bub/README.md | 6 ++++-- e2e/bub/src/powercontext_e2e/paired.py | 13 ++++++++----- e2e/bub/tests/test_paired.py | 25 +++++++++++++++++-------- 3 files changed, 29 insertions(+), 15 deletions(-) diff --git a/e2e/bub/README.md b/e2e/bub/README.md index 48cd5f507b..2dbf720aaf 100644 --- a/e2e/bub/README.md +++ b/e2e/bub/README.md @@ -166,8 +166,10 @@ a timeout. Host plugins flush on different schedules, so the harness flushes the Server's generation model therefore takes part in the ON arm; the run fails early when the Server does not report `memory_extraction`. -An ON run counts only when Server statistics for its Scope show that Sources were captured and turned into Memory -before the recall session, and that PowerContext supplied context during it. Otherwise it is an integration failure. +An ON run counts only when Server statistics for its Scope show that Sources were captured before the recall session +and that the integration asked PowerContext for context during it. Otherwise it is an integration failure. Whether a +flush creates Memory and whether recall returns content are PowerContext's own behavior, so the snapshots record them +but a run that gets nothing useful still counts as an ON attempt. Integration failures and harness or infrastructure errors are reported but left out of success rates and paired differences. An agent timeout counts as a failed attempt in either arm. diff --git a/e2e/bub/src/powercontext_e2e/paired.py b/e2e/bub/src/powercontext_e2e/paired.py index 86342c3535..e08de1ed08 100644 --- a/e2e/bub/src/powercontext_e2e/paired.py +++ b/e2e/bub/src/powercontext_e2e/paired.py @@ -213,7 +213,12 @@ async def _run_arm( def treatment_failures(sessions: Sequence[SessionSnapshot], recall_session: int) -> tuple[str, ...]: - """Explain why an ON run did not receive PowerContext's treatment, or return nothing when it did.""" + """Explain why an ON run did not receive PowerContext's treatment, or return nothing when it did. + + The treatment is the integration capturing earlier sessions and asking PowerContext for context during the recall + session. Whether a flush creates Memory and whether recall returns content are PowerContext's own behavior under + that treatment, so they are recorded in the snapshots but do not decide whether a run counts. + """ by_session = {snapshot.session: snapshot for snapshot in sessions} before = by_session.get(recall_session - 1) @@ -223,10 +228,8 @@ def treatment_failures(sessions: Sequence[SessionSnapshot], recall_session: int) failures: list[str] = [] if before.sources == 0: failures.append("No Sources were captured before the recall session") - if before.memory_entries == 0: - failures.append("The flush before the recall session created no Memory") - if recall.ready_preparations <= before.ready_preparations: - failures.append("PowerContext supplied no context during the recall session") + if recall.preparations <= before.preparations: + failures.append("PowerContext was not asked for context during the recall session") return tuple(failures) diff --git a/e2e/bub/tests/test_paired.py b/e2e/bub/tests/test_paired.py index bfac6f2725..27ee2656b7 100644 --- a/e2e/bub/tests/test_paired.py +++ b/e2e/bub/tests/test_paired.py @@ -94,33 +94,42 @@ def test_acceptance_rejects_continuation_workloads(tmp_path: Path) -> None: asyncio.run(run_tasks(_PAIRED_TASKS, output_dir=tmp_path / "out", settings=_SETTINGS)) -def _snapshot(session: int, *, sources: int = 1, memory: int = 1, ready: int = 0) -> SessionSnapshot: +def _snapshot(session: int, *, sources: int = 1, memory: int = 1, asked: int = 0, ready: int = 0) -> SessionSnapshot: return SessionSnapshot( session=session, flush_rounds=1, sources=sources, memory_pending=0, memory_entries=memory, - preparations=ready, + preparations=asked, ready_preparations=ready, ) def test_treatment_passes_when_recall_received_context_from_captured_memory() -> None: - assert treatment_failures((_snapshot(0, ready=1), _snapshot(1, ready=2)), recall_session=1) == () + sessions = (_snapshot(0, asked=1, ready=1), _snapshot(1, asked=2, ready=2)) + + assert treatment_failures(sessions, recall_session=1) == () + + +def test_treatment_passes_when_powercontext_keeps_or_returns_nothing() -> None: + # The integration captured the earlier session and asked during recall; an empty answer is a scored ON failure, + # not an excluded run. + sessions = (_snapshot(0, memory=0, asked=4), _snapshot(1, memory=0, asked=8)) + + assert treatment_failures(sessions, recall_session=1) == () @pytest.mark.parametrize( ("sessions", "failure"), [ - ((_snapshot(0, sources=0, memory=0), _snapshot(1, ready=1)), "No Sources were captured"), - ((_snapshot(0, memory=0), _snapshot(1, ready=1)), "created no Memory"), - # Context supplied late in the capture session does not show that the recall session received any. - ((_snapshot(0, ready=2), _snapshot(1, ready=2)), "no context during the recall session"), + ((_snapshot(0, sources=0, memory=0), _snapshot(1, asked=1)), "No Sources were captured"), + # Requests made during the capture session do not show that the recall session asked. + ((_snapshot(0, asked=4, ready=2), _snapshot(1, asked=4, ready=2)), "not asked for context"), ((_snapshot(0),), "not observed after every session"), ], ) -def test_treatment_fails_when_any_link_from_capture_to_recall_is_missing(sessions, failure: str) -> None: +def test_treatment_fails_when_the_integration_did_not_capture_or_ask(sessions, failure: str) -> None: assert any(failure in reason for reason in treatment_failures(sessions, recall_session=1)) From cbe61e687c4054b8c8c911760717c42aa8dc52ef Mon Sep 17 00:00:00 2001 From: Fengzdadi <453788063@qq.com> Date: Sat, 26 Sep 2026 18:19:38 -0400 Subject: [PATCH 4/5] fix(e2e): keep workload answers out of the agent container The harness mounted the whole repository into the task container so the agent could install the local PowerContext integration. The repository also holds every workload's answer key and the benchmark datasets, and an agent searching its container found a LoCoMo answer in the dataset file during a real OFF run. Mount only what installation needs: the powercontext package sources, and through the host adapter, the Bub integration and its overrides file. Container paths are unchanged. Co-Authored-By: Claude Opus 5.5 --- e2e/bub/README.md | 4 ++- e2e/bub/src/powercontext_e2e/hosts.py | 45 ++++++++++++++++--------- e2e/bub/src/powercontext_e2e/runner.py | 15 ++++----- e2e/bub/tests/test_harbor_job_config.py | 20 +++++++++++ 4 files changed, 59 insertions(+), 25 deletions(-) diff --git a/e2e/bub/README.md b/e2e/bub/README.md index 2dbf720aaf..fd38f15fff 100644 --- a/e2e/bub/README.md +++ b/e2e/bub/README.md @@ -239,7 +239,9 @@ from that container. In the fixed nested-container harness, `host-gateway` addre proxy exposed there can be passed as `http://host-gateway:`. The typed setting is also treated as a secret when evidence is written. -Agent setup uses Bub's supported installation path: `uv tool install` installs Bub with the local PowerContext plugin, +The agent container sees only the repository files that installation needs: the `powercontext` package and the +host integration. Workload files, answer keys, and benchmark data stay on the host, because the agent can search its +container. Agent setup uses Bub's supported installation path: `uv tool install` installs Bub with the local PowerContext plugin, then `bub install bub-acp-server` adds the ACP server to the same environment. Harbor uploads and runs its native ACP client. The Terminal-Bench task keeps its original image, setup, verifier, and isolation boundary. The harness ignores dataset CPU and memory limits because it evaluates Memory behavior rather than benchmark resource compliance. This diff --git a/e2e/bub/src/powercontext_e2e/hosts.py b/e2e/bub/src/powercontext_e2e/hosts.py index 0024159f76..2ea39a341e 100644 --- a/e2e/bub/src/powercontext_e2e/hosts.py +++ b/e2e/bub/src/powercontext_e2e/hosts.py @@ -16,12 +16,14 @@ from __future__ import annotations +from collections.abc import Iterable +from pathlib import Path from typing import Any, Protocol from harbor.models.trial.config import AgentConfig, ServiceVolumeConfig from .catalog import ContinuationEvaluationSpec, E2ETask, MemoryEvaluationSpec -from .harbor_agent import BUB_ACP_SERVER_VERSION, BUB_VERSION, REMOTE_CODEX_AUTH +from .harbor_agent import BUB_ACP_SERVER_VERSION, BUB_VERSION, REMOTE_CODEX_AUTH, REMOTE_SOURCE from .settings import bub_environment, codex_auth_path, powercontext_bub_environment @@ -37,8 +39,8 @@ def model_configured(self) -> bool: def agent_model(self) -> str | None: """Return the runtime-selected model recorded in evidence.""" - def mounts(self, task: E2ETask) -> list[ServiceVolumeConfig]: - """Return host-owned bind mounts for the task container.""" + def mounts(self, task: E2ETask, repository: Path) -> list[ServiceVolumeConfig]: + """Return host-owned bind mounts for the task container, such as the integration sources it installs.""" def agent_config( self, @@ -67,18 +69,11 @@ def model_configured(self) -> bool: def agent_model(self) -> str | None: return bub_environment().get("BUB_MODEL") - def mounts(self, task: E2ETask) -> list[ServiceVolumeConfig]: - if not task.execution.model or not (auth_path := codex_auth_path()).is_file(): - return [] - return [ - { - "type": "bind", - "source": str(auth_path), - "target": REMOTE_CODEX_AUTH, - "read_only": True, - "bind": {"create_host_path": False}, - } - ] + def mounts(self, task: E2ETask, repository: Path) -> list[ServiceVolumeConfig]: + mounts = source_mounts(repository, ("integrations/bub", "e2e/bub/source-overrides.txt")) + if task.execution.model and (auth_path := codex_auth_path()).is_file(): + mounts.append(read_only_bind(auth_path, REMOTE_CODEX_AUTH)) + return mounts def agent_config( self, @@ -129,6 +124,26 @@ def _capture_settings(task: E2ETask) -> tuple[bool, int, int]: return isinstance(evaluation, ContinuationEvaluationSpec), 5, 8192 +def source_mounts(repository: Path, paths: Iterable[str]) -> list[ServiceVolumeConfig]: + """Mount selected repository paths read-only at the same relative place under the agent's source directory. + + Only what an installation needs is mounted: the whole repository would also expose workload answer keys and + benchmark data to the agent. + """ + + return [read_only_bind(repository / path, f"{REMOTE_SOURCE}/{path}") for path in paths] + + +def read_only_bind(source: Path, target: str) -> ServiceVolumeConfig: + return { + "type": "bind", + "source": str(source), + "target": target, + "read_only": True, + "bind": {"create_host_path": False}, + } + + _HOSTS: dict[str, HostAdapter] = {"bub": BubHost()} diff --git a/e2e/bub/src/powercontext_e2e/runner.py b/e2e/bub/src/powercontext_e2e/runner.py index 3686d50ced..0043d0b49a 100644 --- a/e2e/bub/src/powercontext_e2e/runner.py +++ b/e2e/bub/src/powercontext_e2e/runner.py @@ -45,7 +45,7 @@ from .catalog import ContinuationEvaluationSpec, E2ETask, MemoryEvaluationSpec, OutcomeEvaluationSpec from .evaluation import evaluate_observation, matches_forbidden_context from .evidence import fingerprint, load_resolved_instructions, redact, write_evaluation_report, write_evidence -from .hosts import host_adapter +from .hosts import host_adapter, source_mounts from .models import ( CaptureRecord, EvaluationReport, @@ -67,6 +67,9 @@ TaskStatus = Literal["completed", "failed", "skipped"] BATCH_CATEGORY_PREFIX = "batch:" BATCH_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]*$") +# Host integrations install against the local powercontext package, so the agent container gets that package's +# sources and nothing else from the repository. +POWERCONTEXT_PACKAGE_PATHS = ("pyproject.toml", "README.md", "LICENSE", "src") class TaskArtifacts(NamedTuple): @@ -425,14 +428,8 @@ def _job_config( host = host_adapter(task) repository = settings.repository_path() mounts: list[ServiceVolumeConfig] = [ - { - "type": "bind", - "source": str(repository), - "target": "/opt/powercontext/source", - "read_only": True, - "bind": {"create_host_path": False}, - }, - *host.mounts(task), + *source_mounts(repository, POWERCONTEXT_PACKAGE_PATHS), + *host.mounts(task, repository), ] agent = host.agent_config( task, diff --git a/e2e/bub/tests/test_harbor_job_config.py b/e2e/bub/tests/test_harbor_job_config.py index a0ccb70adf..637c1d6ea9 100644 --- a/e2e/bub/tests/test_harbor_job_config.py +++ b/e2e/bub/tests/test_harbor_job_config.py @@ -184,3 +184,23 @@ def test_off_arm_runs_the_host_without_powercontext(monkeypatch, tmp_path: Path) assert on_agent.env["POWERCONTEXT_BUB_SCOPE_ID"] == "scope-1" assert on_agent.env["POWERCONTEXT_BUB_CAPTURE_EVENTS"] == "true" assert on_agent.kwargs == {} + + +@pytest.mark.parametrize( + "manifest", + ["paired-tasks/project-decision-continuation.yaml", "tasks/acceptance-01-project-database-decision.yaml"], +) +def test_agent_container_cannot_read_workload_answers(tmp_path: Path, manifest: str) -> None: + # The agent can search its container, so no mount may expose task files, answer keys, or benchmark data. + task = load_tasks(_REPOSITORY / "e2e" / "bub" / manifest)[0] + protected = [_REPOSITORY / "e2e" / "bub" / name for name in ("harbor-tasks", "paired-tasks", "tasks")] + protected.append(_REPOSITORY / "benchmark") + + sources = [Path(mount["source"]) for mount in _config(task, tmp_path).environment.mounts] + + assert not [ + (source, path) + for source in sources + for path in protected + if path.is_relative_to(source) or source.is_relative_to(path) + ] From 4d5cac2a1ebd627ca9c18f5add78b97dfe0577ab Mon Sep 17 00:00:00 2001 From: Fengzdadi <453788063@qq.com> Date: Sat, 26 Sep 2026 23:35:31 -0400 Subject: [PATCH 5/5] fix(e2e): grade asserted recall values and score the recall step The recall grader matched keywords, so "OceanBase with 24 shards, not 12" and "PostgreSQL rather than OceanBase, with 12 shards" both scored as successful recall. The recall step now asks for a JSON object with only a database and a shard count, and the grader checks those values; contradictory, hedged, unknown, and malformed answers score 0, as do repeated or extra keys that could hide a contradiction. An arm's reward came from Harbor's trial reward, which averages step rewards unless a task opts into the final-step strategy, so an unfinished capture-session chore could fail a correct recall. The arm is now scored by the declared recall step's own reward, and a task that sets min_reward on an earlier step is rejected, because Harbor would skip the recall step when that step falls short. Co-Authored-By: Claude Opus 5.5 --- e2e/bub/README.md | 24 ++--- .../steps/recall/instruction.md | 3 +- .../steps/recall/tests/grade.py | 44 ++++++++-- .../steps/recall/tests/test.sh | 2 +- .../project-decision-continuation.yaml | 2 +- e2e/bub/src/powercontext_e2e/paired.py | 88 ++++++++++++++----- e2e/bub/tests/test_paired.py | 71 +++++++++++++-- 7 files changed, 183 insertions(+), 51 deletions(-) diff --git a/e2e/bub/README.md b/e2e/bub/README.md index fd38f15fff..9690cf8b05 100644 --- a/e2e/bub/README.md +++ b/e2e/bub/README.md @@ -144,9 +144,12 @@ independent ACP session and Bub tape. A continuation workload is a Harbor multi-step task written in plain language, so any agent host can run it. An earlier session mentions a fact only in the conversation, next to an unrelated small job. The final recall session -asks for that fact and has the agent write its answer to a file. The recall step's own tests grade the answer, and -the answer key lives only there, because Harbor leaves every uploaded test directory in the container for later -steps. The task reward is the final step's reward. +asks for that fact and has the agent write its answer to a file as structured values, so the grader checks what the +answer asserts rather than keywords that a contradictory or hedged answer could also contain. The recall step's own +tests grade the answer, and the answer key lives only there, because Harbor leaves every uploaded test directory in +the container for later steps. The recall step's own reward decides the run whatever the task's multi-step reward +strategy; earlier steps' rewards are recorded for diagnosis only. A task may not set `min_reward` on an earlier +step, because Harbor would then skip the recall step when that step's unrelated job falls short. The `paired` command runs each selected workload with PowerContext off and on, in separate containers, and repeats this for `--trials` trials. The arm that runs first alternates between trials. @@ -239,13 +242,14 @@ from that container. In the fixed nested-container harness, `host-gateway` addre proxy exposed there can be passed as `http://host-gateway:`. The typed setting is also treated as a secret when evidence is written. -The agent container sees only the repository files that installation needs: the `powercontext` package and the -host integration. Workload files, answer keys, and benchmark data stay on the host, because the agent can search its -container. Agent setup uses Bub's supported installation path: `uv tool install` installs Bub with the local PowerContext plugin, -then `bub install bub-acp-server` adds the ACP server to the same environment. Harbor uploads and runs its native ACP -client. The Terminal-Bench task keeps its original image, setup, verifier, and isolation boundary. The harness ignores -dataset CPU and memory limits because it evaluates Memory behavior rather than benchmark resource compliance. This -also keeps the fixed harness usable in nested container runtimes that cannot create additional cgroups. +The agent container sees only the repository files that installation needs: the `powercontext` package and the host +integration. Workload files, answer keys, and benchmark data stay on the host, because the agent can search its +container. Agent setup uses Bub's supported installation path: `uv tool install` installs Bub with the local +PowerContext plugin, then `bub install bub-acp-server` adds the ACP server to the same environment. Harbor uploads and +runs its native ACP client. The Terminal-Bench task keeps its original image, setup, verifier, and isolation boundary. +The harness ignores dataset CPU and memory limits because it evaluates Memory behavior rather than benchmark resource +compliance. This also keeps the fixed harness usable in nested container runtimes that cannot create additional +cgroups. Long-horizon acceptance requires observable Memory behavior: diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md index 5181d47f69..25ca8108a6 100644 --- a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md @@ -1,2 +1,3 @@ Which database did we decide to use for the ticket service, and what shard count did we agree on? Write your answer -as one short sentence to /workspace/answer.txt. If you do not know, write "unknown". +to /workspace/answer.json as a JSON object with only two keys, a `database` string and a `shard_count` integer, for +example `{"database": "", "shard_count": }`. Use null for any value you do not know. diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py index 9aac7346fa..097ba1e711 100644 --- a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py @@ -14,27 +14,59 @@ """Grade the recall answer without depending on the agent host. -This file is uploaded only with the recall step's tests, so no earlier session can read the expected answer. +The answer is structured, so the grader checks the values the agent asserts rather than matching keywords that a +contradictory or hedged answer could also contain. This file is uploaded only with the recall step's tests, so no +earlier session can read the expected answer. """ from __future__ import annotations -import re +import json import sys import unicodedata from pathlib import Path -# Every group must match; any alternative inside a group is enough. -REQUIRED = (("oceanbase",), ("12", "twelve")) +EXPECTED_DATABASE = "oceanbase" +EXPECTED_SHARD_COUNT = 12 +ANSWER_KEYS = {"database", "shard_count"} + + +class DuplicateKeyError(ValueError): + """A repeated key would let a later value silently override a contradictory earlier one.""" def score(answer: str) -> int: - text = unicodedata.normalize("NFC", answer.casefold()) + try: + payload = json.loads(answer, object_pairs_hook=_unique_keys) + except ValueError: + return 0 + # Extra keys could carry a hedge or an alternative that the checked fields do not show. + if not isinstance(payload, dict) or set(payload) != ANSWER_KEYS: + return 0 + database = payload.get("database") + shard_count = payload.get("shard_count") return int( - all(any(re.search(rf"\b{re.escape(term)}\b", text) for term in group) for group in REQUIRED), + isinstance(database, str) + and unicodedata.normalize("NFKC", database).strip().casefold() == EXPECTED_DATABASE + and _integer(shard_count) == EXPECTED_SHARD_COUNT ) +def _unique_keys(pairs: list[tuple[str, object]]) -> dict[str, object]: + keys = [key for key, _ in pairs] + if len(keys) != len(set(keys)): + raise DuplicateKeyError + return dict(pairs) + + +def _integer(value: object) -> int | None: + if isinstance(value, int) and not isinstance(value, bool): + return value + if isinstance(value, str) and value.strip().isdecimal(): + return int(value.strip()) + return None + + def main(answer_path: Path, reward_path: Path) -> None: answer = answer_path.read_text(encoding="utf-8", errors="replace") if answer_path.is_file() else "" reward_path.write_text(f"{score(answer)}\n", encoding="utf-8") diff --git a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh index 2fe882fd4a..58fe7aa152 100644 --- a/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh +++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/test.sh @@ -15,4 +15,4 @@ set -eu -python3 /tests/grade.py /workspace/answer.txt /logs/verifier/reward.txt +python3 /tests/grade.py /workspace/answer.json /logs/verifier/reward.txt diff --git a/e2e/bub/paired-tasks/project-decision-continuation.yaml b/e2e/bub/paired-tasks/project-decision-continuation.yaml index 4156574250..0b4298b1d6 100644 --- a/e2e/bub/paired-tasks/project-decision-continuation.yaml +++ b/e2e/bub/paired-tasks/project-decision-continuation.yaml @@ -20,7 +20,7 @@ categories: dataset: path: e2e/bub/harbor-tasks task_id: project-decision-continuation - checksum: c517d77f6fdaf4b7675357b3fd826fc42e24caf2604deb4258223711a1652812 + checksum: 87429c6ac133a99343d72e8fc614fb2f18d773a83c318097ec085a610bb3315e execution: type: bub model: true diff --git a/e2e/bub/src/powercontext_e2e/paired.py b/e2e/bub/src/powercontext_e2e/paired.py index e08de1ed08..c0fbcb751d 100644 --- a/e2e/bub/src/powercontext_e2e/paired.py +++ b/e2e/bub/src/powercontext_e2e/paired.py @@ -79,6 +79,7 @@ async def run_paired( if not tasks or trials < 1: raise ValueError("At least one continuation workload and one trial are required") # noqa: TRY003 recall_sessions = {task.id: recall_session_index(task, settings) for task in tasks} + recall_steps = {task.id: _continuation(task).recall_step for task in tasks} require_runtime_models(tasks) observations: list[PairedArmObservation] = [] @@ -98,6 +99,7 @@ async def run_paired( arm=arm, position=position, recall_session=recall_sessions[task.id], + recall_step=recall_steps[task.id], output_dir=arm_dir, settings=settings, ) @@ -117,17 +119,28 @@ async def run_paired( def recall_session_index(task: E2ETask, settings: HarnessSettings) -> int: """Return the zero-based agent session that answers from earlier sessions.""" - evaluation = task.evaluation - if not isinstance(evaluation, ContinuationEvaluationSpec): - raise TypeError(f"Workload {task.id!r} is not an OFF/ON continuation workload") # noqa: TRY003 - steps = _load_source_task(task, settings.repository_path()).source_steps + evaluation = _continuation(task) + source = _load_source_task(task, settings.repository_path()) + steps = source.source_steps if len(steps) < 2 or steps[-1] != evaluation.recall_step: raise ValueError( # noqa: TRY003 f"Workload {task.id!r} must end with its recall step {evaluation.recall_step!r} after an earlier session" ) + # Harbor skips the remaining steps when a step scores below its min_reward, so an earlier session's unrelated + # job could keep the recall session from running at all. + if gated := [step.name for step in (source.harbor_task.config.steps or ())[:-1] if step.min_reward is not None]: + raise ValueError( # noqa: TRY003 + f"Workload {task.id!r} sets min_reward on {gated!r}, which could stop its recall step from running" + ) return len(steps) - 1 +def _continuation(task: E2ETask) -> ContinuationEvaluationSpec: + if not isinstance(task.evaluation, ContinuationEvaluationSpec): + raise TypeError(f"Workload {task.id!r} is not an OFF/ON continuation workload") # noqa: TRY003 + return task.evaluation + + async def _run_arm( client: PowerContextClient, task: E2ETask, @@ -136,6 +149,7 @@ async def _run_arm( arm: Arm, position: int, recall_session: int, + recall_step: str, output_dir: Path, settings: HarnessSettings, ) -> PairedArmObservation: @@ -175,15 +189,6 @@ async def _run_arm( if arm == "on" else () ) - exception_types = tuple( - name - for name in ( - *(step.exception_info.exception_type for step in step_results if step.exception_info is not None), - harbor.exception_type, - ) - if name is not None - ) - reward = harbor.rewards.get("reward") return PairedArmObservation( run_id=run_id, task_id=task.id, @@ -193,18 +198,13 @@ async def _run_arm( environment=_run_environment(task, started_at, settings), scope_id=scope_id, harbor=harbor, - step_rewards={ - step.step_name: float(step.verifier_result.rewards["reward"]) - for step in step_results - if step.verifier_result is not None - and step.verifier_result.rewards - and "reward" in step.verifier_result.rewards - }, - outcome=classify_outcome( + step_rewards=step_rewards(step_results), + outcome=arm_outcome( + step_results, + harbor, + recall_step=recall_step, harness_failed=bool(errors), - exception_types=exception_types, treatment_failures=treatment, - reward=None if reward is None else float(reward), ), errors=tuple(errors), sessions=sessions, @@ -212,6 +212,48 @@ async def _run_arm( ) +def arm_outcome( + step_results: Sequence[StepResult], + harbor: HarborTrialObservation, + *, + recall_step: str, + harness_failed: bool, + treatment_failures: Sequence[str], +) -> ArmOutcome: + """Classify one arm from Harbor's results, scoring it by the recall step's own reward.""" + + exception_types = tuple( + name + for name in ( + *(step.exception_info.exception_type for step in step_results if step.exception_info is not None), + harbor.exception_type, + ) + if name is not None + ) + return classify_outcome( + harness_failed=harness_failed, + exception_types=exception_types, + treatment_failures=treatment_failures, + reward=step_rewards(step_results).get(recall_step), + ) + + +def step_rewards(step_results: Sequence[StepResult]) -> dict[str, float]: + """Return each step's own reward. + + The recall step's reward decides an arm. Harbor's trial reward follows the task's multi-step strategy and can + average in earlier steps, whose small jobs are unrelated to recall. + """ + + return { + step.step_name: float(step.verifier_result.rewards["reward"]) + for step in step_results + if step.verifier_result is not None + and step.verifier_result.rewards + and "reward" in step.verifier_result.rewards + } + + def treatment_failures(sessions: Sequence[SessionSnapshot], recall_session: int) -> tuple[str, ...]: """Explain why an ON run did not receive PowerContext's treatment, or return nothing when it did. diff --git a/e2e/bub/tests/test_paired.py b/e2e/bub/tests/test_paired.py index 27ee2656b7..6e69b0e3d6 100644 --- a/e2e/bub/tests/test_paired.py +++ b/e2e/bub/tests/test_paired.py @@ -18,15 +18,25 @@ import asyncio import runpy +import shutil from datetime import UTC, datetime from pathlib import Path from types import SimpleNamespace import pytest +from harbor.models.task.task import Task as HarborTask +from harbor.models.trial.result import StepResult +from harbor.models.verifier.result import VerifierResult from powercontext_e2e.catalog import load_tasks from powercontext_e2e.models import HarborTrialObservation, PairedArmObservation, RunEnvironment, SessionSnapshot -from powercontext_e2e.paired import classify_outcome, recall_session_index, summarize, treatment_failures +from powercontext_e2e.paired import ( + arm_outcome, + classify_outcome, + recall_session_index, + summarize, + treatment_failures, +) from powercontext_e2e.runner import run_tasks from powercontext_e2e.sessions import SessionRecorder, settle_session from powercontext_e2e.settings import HarnessSettings @@ -48,15 +58,26 @@ def _grade(answer_path: Path, reward_path: Path) -> None: @pytest.mark.parametrize( ("answer", "reward"), [ - ("We chose OceanBase with a shard count of 12.", 1), - ("oceanbase, twelve shards", 1), - ("OceanBase with 120 shards.", 0), - ("PostgreSQL with 12 shards.", 0), - ("unknown", 0), + ('{"database": "OceanBase", "shard_count": 12}', 1), + ('{"database": " oceanbase ", "shard_count": "12"}', 1), + # Contradictory answers name the right fact somewhere but assert another value. + ('{"database": "OceanBase", "shard_count": 24}', 0), + ('{"database": "PostgreSQL", "shard_count": 12}', 0), + ("We chose OceanBase with 24 shards, not 12.", 0), + ("We chose PostgreSQL rather than OceanBase, with 12 shards.", 0), + # Uncertain answers do not assert the decision. + ('{"database": null, "shard_count": null}', 0), + ('{"database": "OceanBase?", "shard_count": 12}', 0), + ('{"database": "maybe OceanBase", "shard_count": 12}', 0), + ('{"database": "OceanBase", "shard_count": "about 12"}', 0), + ('["OceanBase", 12]', 0), + # A repeated key or an extra field could hide a contradiction from the checked values. + ('{"database": "PostgreSQL", "database": "OceanBase", "shard_count": 12}', 0), + ('{"database": "OceanBase", "shard_count": 12, "note": "or PostgreSQL with 24"}', 0), ], ) -def test_recall_grader_requires_every_fact(tmp_path: Path, answer: str, reward: int) -> None: - answer_path = tmp_path / "answer.txt" +def test_recall_grader_checks_the_asserted_values(tmp_path: Path, answer: str, reward: int) -> None: + answer_path = tmp_path / "answer.json" answer_path.write_text(answer, encoding="utf-8") reward_path = tmp_path / "reward.txt" @@ -68,7 +89,7 @@ def test_recall_grader_requires_every_fact(tmp_path: Path, answer: str, reward: def test_recall_grader_scores_a_missing_answer_as_zero(tmp_path: Path) -> None: reward_path = tmp_path / "reward.txt" - _grade(tmp_path / "answer.txt", reward_path) + _grade(tmp_path / "answer.json", reward_path) assert reward_path.read_text(encoding="utf-8") == "0\n" @@ -133,6 +154,38 @@ def test_treatment_fails_when_the_integration_did_not_capture_or_ask(sessions, f assert any(failure in reason for reason in treatment_failures(sessions, recall_session=1)) +def test_the_recall_step_reward_decides_the_arm_whatever_harbor_averaged() -> None: + # Harbor averages step rewards unless a task opts into its final-step strategy; an unfinished capture-session + # chore must not turn a correct recall into a failure. + steps = ( + StepResult(step_name="capture", verifier_result=VerifierResult(rewards={"reward": 0})), + StepResult(step_name="recall", verifier_result=VerifierResult(rewards={"reward": 1})), + ) + averaged = HarborTrialObservation(rewards={"reward": 0.5}) + + outcome = arm_outcome(steps, averaged, recall_step="recall", harness_failed=False, treatment_failures=()) + + assert outcome == "passed" + + +def test_continuation_tasks_cannot_gate_the_recall_step_behind_an_earlier_reward(tmp_path: Path) -> None: + # Harbor skips the remaining steps when a step scores below its min_reward. + task = next(task for task in _PAIRED_TASKS if task.id == "project-decision-continuation") + task_dir = tmp_path / "e2e" / "bub" / "harbor-tasks" / task.dataset.task_id + shutil.copytree(_HARBOR_TASKS / task.dataset.task_id, task_dir) + config = task_dir / "task.toml" + config.write_text( + config.read_text(encoding="utf-8").replace('name = "capture"', 'name = "capture"\nmin_reward = 1.0'), + encoding="utf-8", + ) + task = task.model_copy( + update={"dataset": task.dataset.model_copy(update={"checksum": HarborTask(task_dir).checksum})} + ) + + with pytest.raises(ValueError, match="min_reward"): + recall_session_index(task, HarnessSettings(repository=tmp_path)) + + @pytest.mark.parametrize( ("kwargs", "outcome"), [