diff --git a/e2e/bub/README.md b/e2e/bub/README.md index 294c3f50f..610084344 100644 --- a/e2e/bub/README.md +++ b/e2e/bub/README.md @@ -65,6 +65,8 @@ evaluation: The dataset can be a local Harbor dataset path or a registry dataset name and version. `execution` selects the adapter and its budget. `model` declares only whether the workload requires a model. The runtime selects the model, provider, endpoint, and credentials. `evaluation` declares only externally observable Memory behavior. +Two or more compatible selected tasks with the same `batch:` category share one run-local Harbor task and +container. Their scopes, evidence, and evaluation remain independent; selecting one task uses the normal path. Runtime configuration keeps the native ownership of each component. The harness Client reads `POWERCONTEXT_CLIENT_*`, the Bub adapter forwards native `BUB_*` settings for model-backed workloads, and the @@ -76,12 +78,10 @@ The built-in manifests are: | ID | Dataset | Categories | Purpose | | --- | --- | --- | --- | -| `locomo-multihop-football` | local Harbor multi-step task | `acceptance`, `sample` | Pinned LoCoMo-derived sample (multi-hop) | -| `locomo-open-pastries` | local Harbor multi-step task | `acceptance`, `sample` | Pinned LoCoMo-derived sample (open-domain listing) | -| `locomo-support-group` | local Harbor multi-step task | `acceptance`, `sample` | Pinned LoCoMo-derived sample | -| `locomo-temporal-banker` | local Harbor multi-step task | `acceptance`, `sample` | Pinned LoCoMo-derived sample (temporal) | -| `project-database-decision` | local Harbor multi-step task | `acceptance`, `sample`, `smoke` | Durable project decision | +| `locomo-*` (four manifests) | one local Harbor task each | `acceptance`, `sample`, `batch:locomo` | Pinned LoCoMo-derived cases | +| `project-database-decision` | local Harbor multi-step task | `acceptance`, `sample`, `smoke`, `batch:acceptance` | Durable project decision | | `terminal-bench-db-wal-recovery` | `terminal-bench@2.0` | `long-horizon`, `terminal-bench` | Long-running capture and recall | +| `failure-policy-*` | local Harbor tasks | `acceptance`, `fixture`, `batch:acceptance` | Collect-all and fail-fast behavior | ## Run acceptance workloads @@ -93,6 +93,10 @@ export POWERCONTEXT_BUB_BASE_URL=http://host-gateway:8000 make harness-acceptance ``` +The default run executes all acceptance tasks with collect-all, then reruns `batch:acceptance` with fail-fast. LoCoMo +stays in its own `batch:locomo`; the database decision and failure-policy tasks share `batch:acceptance`. Explicit +selection runs only the selected workloads with the requested failure policy. + Selection uses the `acceptance` command's repeatable `--id` and `--category` options: ```bash @@ -121,12 +125,10 @@ Each selected workload writes the same layout: harbor-jobs/ ``` -`replay.json` is a self-contained Pydantic observation. Its workload's `execution.type` identifies the `bub` adapter, -and the remaining fields record the pre-execution Memory baseline and the instructions resolved by Harbor's ACP -runner. -`eval-report.json` uses -`powercontext.e2e-evaluation/v1`. `report.md` is rendered from the report model with Marko. Native Harbor and ACP -evidence remains under `harbor-jobs/`. +Shared runs write the same v1 files per source task under `batch-/tasks//`, plus one aggregate +evaluation and report at `batch-/`. `collect-all` reports every failed task; `fail-fast` stops only that shared +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. ## Long-horizon task diff --git a/e2e/bub/harbor-tasks/locomo-multihop-football/steps/recall/tests/test.sh b/e2e/bub/harbor-tasks/failure-policy-followup/environment/Dockerfile similarity index 91% rename from e2e/bub/harbor-tasks/locomo-multihop-football/steps/recall/tests/test.sh rename to e2e/bub/harbor-tasks/failure-policy-followup/environment/Dockerfile index ea6268777..6da75edc5 100644 --- a/e2e/bub/harbor-tasks/locomo-multihop-football/steps/recall/tests/test.sh +++ b/e2e/bub/harbor-tasks/failure-policy-followup/environment/Dockerfile @@ -1,4 +1,3 @@ -#!/bin/sh # Copyright (c) 2026 OceanBase. # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,6 +12,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -set -eu +FROM python:3.12-slim-bookworm -echo 1 > /logs/verifier/reward.txt +WORKDIR /workspace diff --git a/e2e/bub/harbor-tasks/failure-policy-followup/steps/run/instruction.md b/e2e/bub/harbor-tasks/failure-policy-followup/steps/run/instruction.md new file mode 100644 index 000000000..ea0891f86 --- /dev/null +++ b/e2e/bub/harbor-tasks/failure-policy-followup/steps/run/instruction.md @@ -0,0 +1 @@ +,touch /workspace/failure-policy-followup-ran diff --git a/e2e/bub/harbor-tasks/failure-policy-followup/steps/run/tests/test.sh b/e2e/bub/harbor-tasks/failure-policy-followup/steps/run/tests/test.sh new file mode 100644 index 000000000..1e78473bd --- /dev/null +++ b/e2e/bub/harbor-tasks/failure-policy-followup/steps/run/tests/test.sh @@ -0,0 +1,22 @@ +#!/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 + +if [ -e /workspace/failure-policy-followup-ran ]; then + echo 1 > /logs/verifier/reward.txt +else + echo 0 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/harbor-tasks/failure-policy-followup/task.toml b/e2e/bub/harbor-tasks/failure-policy-followup/task.toml new file mode 100644 index 000000000..d92b5fae6 --- /dev/null +++ b/e2e/bub/harbor-tasks/failure-policy-followup/task.toml @@ -0,0 +1,28 @@ +# 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 = 300.0 + +[verifier] +timeout_sec = 60.0 + +[environment] +build_timeout_sec = 300.0 + +[[steps]] +name = "run" diff --git a/e2e/bub/harbor-tasks/locomo-open-pastries/steps/capture/tests/test.sh b/e2e/bub/harbor-tasks/failure-policy-timeout/environment/Dockerfile similarity index 91% rename from e2e/bub/harbor-tasks/locomo-open-pastries/steps/capture/tests/test.sh rename to e2e/bub/harbor-tasks/failure-policy-timeout/environment/Dockerfile index ea6268777..6da75edc5 100644 --- a/e2e/bub/harbor-tasks/locomo-open-pastries/steps/capture/tests/test.sh +++ b/e2e/bub/harbor-tasks/failure-policy-timeout/environment/Dockerfile @@ -1,4 +1,3 @@ -#!/bin/sh # Copyright (c) 2026 OceanBase. # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,6 +12,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -set -eu +FROM python:3.12-slim-bookworm -echo 1 > /logs/verifier/reward.txt +WORKDIR /workspace diff --git a/e2e/bub/harbor-tasks/failure-policy-timeout/steps/run/instruction.md b/e2e/bub/harbor-tasks/failure-policy-timeout/steps/run/instruction.md new file mode 100644 index 000000000..618e7092f --- /dev/null +++ b/e2e/bub/harbor-tasks/failure-policy-timeout/steps/run/instruction.md @@ -0,0 +1 @@ +,sleep 5 diff --git a/e2e/bub/harbor-tasks/locomo-multihop-football/steps/capture/tests/test.sh b/e2e/bub/harbor-tasks/failure-policy-timeout/steps/run/tests/test.sh similarity index 94% rename from e2e/bub/harbor-tasks/locomo-multihop-football/steps/capture/tests/test.sh rename to e2e/bub/harbor-tasks/failure-policy-timeout/steps/run/tests/test.sh index ea6268777..27384a095 100644 --- a/e2e/bub/harbor-tasks/locomo-multihop-football/steps/capture/tests/test.sh +++ b/e2e/bub/harbor-tasks/failure-policy-timeout/steps/run/tests/test.sh @@ -15,4 +15,4 @@ set -eu -echo 1 > /logs/verifier/reward.txt +echo 0 > /logs/verifier/reward.txt diff --git a/e2e/bub/harbor-tasks/failure-policy-timeout/task.toml b/e2e/bub/harbor-tasks/failure-policy-timeout/task.toml new file mode 100644 index 000000000..ec9c46a60 --- /dev/null +++ b/e2e/bub/harbor-tasks/failure-policy-timeout/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 = 300.0 + +[verifier] +timeout_sec = 60.0 + +[environment] +build_timeout_sec = 300.0 + +[[steps]] +name = "run" + +[steps.agent] +timeout_sec = 1.0 diff --git a/e2e/bub/harbor-tasks/locomo-open-pastries/steps/recall/tests/test.sh b/e2e/bub/harbor-tasks/locomo-multihop-football/tests/test.sh old mode 100644 new mode 100755 similarity index 81% rename from e2e/bub/harbor-tasks/locomo-open-pastries/steps/recall/tests/test.sh rename to e2e/bub/harbor-tasks/locomo-multihop-football/tests/test.sh index ea6268777..fa405bd81 --- a/e2e/bub/harbor-tasks/locomo-open-pastries/steps/recall/tests/test.sh +++ b/e2e/bub/harbor-tasks/locomo-multihop-football/tests/test.sh @@ -15,4 +15,8 @@ set -eu -echo 1 > /logs/verifier/reward.txt +if [ -e /logs/agent/powercontext-step-failed ]; then + echo 0 > /logs/verifier/reward.txt +else + echo 1 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/harbor-tasks/locomo-open-pastries/tests/test.sh b/e2e/bub/harbor-tasks/locomo-open-pastries/tests/test.sh new file mode 100755 index 000000000..fa405bd81 --- /dev/null +++ b/e2e/bub/harbor-tasks/locomo-open-pastries/tests/test.sh @@ -0,0 +1,22 @@ +#!/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 + +if [ -e /logs/agent/powercontext-step-failed ]; then + echo 0 > /logs/verifier/reward.txt +else + echo 1 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/harbor-tasks/locomo-support-group/steps/capture/tests/test.sh b/e2e/bub/harbor-tasks/locomo-support-group/steps/capture/tests/test.sh deleted file mode 100644 index ea6268777..000000000 --- a/e2e/bub/harbor-tasks/locomo-support-group/steps/capture/tests/test.sh +++ /dev/null @@ -1,18 +0,0 @@ -#!/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 - -echo 1 > /logs/verifier/reward.txt diff --git a/e2e/bub/harbor-tasks/locomo-support-group/steps/recall/tests/test.sh b/e2e/bub/harbor-tasks/locomo-support-group/steps/recall/tests/test.sh deleted file mode 100644 index ea6268777..000000000 --- a/e2e/bub/harbor-tasks/locomo-support-group/steps/recall/tests/test.sh +++ /dev/null @@ -1,18 +0,0 @@ -#!/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 - -echo 1 > /logs/verifier/reward.txt diff --git a/e2e/bub/harbor-tasks/locomo-support-group/tests/test.sh b/e2e/bub/harbor-tasks/locomo-support-group/tests/test.sh new file mode 100755 index 000000000..fa405bd81 --- /dev/null +++ b/e2e/bub/harbor-tasks/locomo-support-group/tests/test.sh @@ -0,0 +1,22 @@ +#!/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 + +if [ -e /logs/agent/powercontext-step-failed ]; then + echo 0 > /logs/verifier/reward.txt +else + echo 1 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/harbor-tasks/locomo-temporal-banker/steps/capture/tests/test.sh b/e2e/bub/harbor-tasks/locomo-temporal-banker/steps/capture/tests/test.sh deleted file mode 100644 index ea6268777..000000000 --- a/e2e/bub/harbor-tasks/locomo-temporal-banker/steps/capture/tests/test.sh +++ /dev/null @@ -1,18 +0,0 @@ -#!/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 - -echo 1 > /logs/verifier/reward.txt diff --git a/e2e/bub/harbor-tasks/locomo-temporal-banker/steps/recall/tests/test.sh b/e2e/bub/harbor-tasks/locomo-temporal-banker/steps/recall/tests/test.sh deleted file mode 100644 index ea6268777..000000000 --- a/e2e/bub/harbor-tasks/locomo-temporal-banker/steps/recall/tests/test.sh +++ /dev/null @@ -1,18 +0,0 @@ -#!/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 - -echo 1 > /logs/verifier/reward.txt diff --git a/e2e/bub/harbor-tasks/locomo-temporal-banker/tests/test.sh b/e2e/bub/harbor-tasks/locomo-temporal-banker/tests/test.sh new file mode 100755 index 000000000..fa405bd81 --- /dev/null +++ b/e2e/bub/harbor-tasks/locomo-temporal-banker/tests/test.sh @@ -0,0 +1,22 @@ +#!/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 + +if [ -e /logs/agent/powercontext-step-failed ]; then + echo 0 > /logs/verifier/reward.txt +else + echo 1 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/harbor-tasks/project-database-decision/steps/capture/tests/test.sh b/e2e/bub/harbor-tasks/project-database-decision/steps/capture/tests/test.sh index ea6268777..fa405bd81 100644 --- a/e2e/bub/harbor-tasks/project-database-decision/steps/capture/tests/test.sh +++ b/e2e/bub/harbor-tasks/project-database-decision/steps/capture/tests/test.sh @@ -15,4 +15,8 @@ set -eu -echo 1 > /logs/verifier/reward.txt +if [ -e /logs/agent/powercontext-step-failed ]; then + echo 0 > /logs/verifier/reward.txt +else + echo 1 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/harbor-tasks/project-database-decision/steps/recall/tests/test.sh b/e2e/bub/harbor-tasks/project-database-decision/steps/recall/tests/test.sh index ea6268777..fa405bd81 100644 --- a/e2e/bub/harbor-tasks/project-database-decision/steps/recall/tests/test.sh +++ b/e2e/bub/harbor-tasks/project-database-decision/steps/recall/tests/test.sh @@ -15,4 +15,8 @@ set -eu -echo 1 > /logs/verifier/reward.txt +if [ -e /logs/agent/powercontext-step-failed ]; then + echo 0 > /logs/verifier/reward.txt +else + echo 1 > /logs/verifier/reward.txt +fi diff --git a/e2e/bub/src/powercontext_e2e/__main__.py b/e2e/bub/src/powercontext_e2e/__main__.py index a78ef043a..529683f95 100644 --- a/e2e/bub/src/powercontext_e2e/__main__.py +++ b/e2e/bub/src/powercontext_e2e/__main__.py @@ -20,11 +20,39 @@ import asyncio import sys from pathlib import Path +from typing import TYPE_CHECKING from loguru import logger from .settings import HarnessSettings +if TYPE_CHECKING: + from .catalog import E2ETask + + +async def _run_default_acceptance( + tasks: tuple[E2ETask, ...], + *, + output_dir: Path, + settings: HarnessSettings, +) -> bool: + from .catalog import select_tasks + from .runner import run_tasks + + collect_all_passed = await run_tasks( + select_tasks(tasks, categories=("acceptance",)), + output_dir=output_dir, + settings=settings, + failure_policy="collect-all", + ) + fail_fast_passed = await run_tasks( + select_tasks(tasks, categories=("batch:acceptance",)), + output_dir=output_dir / "fail-fast", + settings=settings, + failure_policy="fail-fast", + ) + return collect_all_passed and fail_fast_passed + def main() -> None: logger.remove() @@ -49,6 +77,12 @@ def main() -> None: help="Select one category; repeat to select more than one.", ) acceptance_parser.add_argument("--output", type=Path, required=True) + acceptance_parser.add_argument( + "--failure-policy", + choices=("collect-all", "fail-fast"), + default="collect-all", + help="Continue through case failures or stop the Harbor trial at the first failed step.", + ) rescore_parser = subparsers.add_parser("rescore") rescore_parser.add_argument("replay", type=Path) @@ -66,11 +100,22 @@ def main() -> None: tasks = load_tasks(args.manifest) selected = select_tasks(tasks, ids=tuple(args.id), categories=tuple(args.category)) - passed = asyncio.run( - run_tasks( - selected, - output_dir=args.output, - settings=settings, + passed = ( + asyncio.run( + run_tasks( + selected, + output_dir=args.output, + settings=settings, + failure_policy=args.failure_policy, + ) + ) + if args.id or args.category + else asyncio.run( + _run_default_acceptance( + tasks, + output_dir=args.output, + settings=settings, + ) ) ) raise SystemExit(0 if passed else 1) diff --git a/e2e/bub/src/powercontext_e2e/catalog.py b/e2e/bub/src/powercontext_e2e/catalog.py index 613277304..8236f6be4 100644 --- a/e2e/bub/src/powercontext_e2e/catalog.py +++ b/e2e/bub/src/powercontext_e2e/catalog.py @@ -94,6 +94,15 @@ def require_unique_probe_ids(self) -> MemoryEvaluationSpec: return self +class ExpectedExecutionSpec(CatalogModel): + collect_all: Literal["completed", "failed", "skipped"] = Field(alias="collect-all") + fail_fast: Literal["completed", "failed", "skipped"] = Field(alias="fail-fast") + + +class OutcomeEvaluationSpec(CatalogModel): + expected_execution: ExpectedExecutionSpec + + class E2ETask(CatalogModel): schema_: Literal["powercontext.e2e-task/v1"] = Field(alias="schema") id: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]*$") @@ -101,7 +110,7 @@ class E2ETask(CatalogModel): provenance: Provenance | None = None dataset: HarborDatasetSpec execution: BubExecutionSpec - evaluation: MemoryEvaluationSpec + evaluation: MemoryEvaluationSpec | OutcomeEvaluationSpec class TaskSelectionError(ValueError): diff --git a/e2e/bub/src/powercontext_e2e/evaluation.py b/e2e/bub/src/powercontext_e2e/evaluation.py index 6ad1a71d1..9a285969a 100644 --- a/e2e/bub/src/powercontext_e2e/evaluation.py +++ b/e2e/bub/src/powercontext_e2e/evaluation.py @@ -17,7 +17,9 @@ from __future__ import annotations from pathlib import Path +from typing import Literal +from .catalog import MemoryEvaluationSpec, OutcomeEvaluationSpec from .models import ( CaseEvaluation, EvaluationReport, @@ -27,6 +29,65 @@ ) CAPTURE_EVENTS = frozenset({"user_prompt", "llm_result", "tool_result"}) +FailurePolicy = Literal["fail-fast", "collect-all"] + + +def evaluate_observation( + observation: TaskObservation, + *, + experiment: str, + failure_policy: FailurePolicy, +) -> EvaluationReport: + evaluation = observation.task.evaluation + if isinstance(evaluation, OutcomeEvaluationSpec): + return _evaluate_execution( + observation, + evaluation, + experiment=experiment, + failure_policy=failure_policy, + ) + return MemoryEvaluator.evaluate(observation, experiment=experiment) + + +def _evaluate_execution( + observation: TaskObservation, + evaluation: OutcomeEvaluationSpec, + *, + experiment: str, + failure_policy: FailurePolicy, +) -> EvaluationReport: + expected = ( + evaluation.expected_execution.collect_all + if failure_policy == "collect-all" + else evaluation.expected_execution.fail_fast + ) + actual = observation.status + expected_reward = {"completed": 1, "failed": 0}.get(expected) + actual_reward = observation.harbor.rewards.get("reward") + reward_matches = expected == "skipped" or actual_reward == expected_reward + return EvaluationReport( + experiment=experiment, + cases=( + CaseEvaluation( + name=observation.task.id, + assertions={ + "execution_outcome": EvaluationValue( + value=actual == expected and reward_matches, + reason=( + f"Expected {expected!r}; observed {actual!r}." + if expected == "skipped" + else ( + f"Expected {expected!r} with Harbor reward {expected_reward!r}; " + f"observed {actual!r} with Harbor reward {actual_reward!r}." + ) + ), + ) + }, + labels={"task_outcome": EvaluationValue(value=actual)}, + attributes=_attributes(observation), + ), + ), + ) class MemoryEvaluator: @@ -35,7 +96,27 @@ class MemoryEvaluator: @staticmethod def evaluate(observation: TaskObservation, *, experiment: str) -> EvaluationReport: task = observation.task + attributes = _attributes(observation) + if observation.status == "skipped": + return EvaluationReport( + experiment=experiment, + cases=( + CaseEvaluation( + name=task.id, + assertions={ + "execution_completed": EvaluationValue( + value=False, + reason="Skipped after an earlier task stopped the shared Harbor trial.", + ) + }, + labels={"task_outcome": EvaluationValue(value="skipped")}, + attributes=attributes, + ), + ), + ) evaluation = task.evaluation + if not isinstance(evaluation, MemoryEvaluationSpec): + raise TypeError("MemoryEvaluator requires a Memory evaluation task") # noqa: TRY003 eligible_records = [record for record in observation.capture_records if record.event in CAPTURE_EVENTS] captured_records = [record for record in eligible_records if record.status == "captured"] capture_coverage = len(captured_records) / len(eligible_records) if eligible_records else 0.0 @@ -87,17 +168,9 @@ def evaluate(observation: TaskObservation, *, experiment: str) -> EvaluationRepo any(fragment.casefold() in entry.text.casefold() for entry in observation.memory_after.entries) for fragment in evaluation.expected_memory ) + observed_checksum = observation.harbor.source_task_checksum or observation.harbor.task_checksum thresholds = evaluation.thresholds - attributes = { - "commit": observation.environment.commit, - "database": observation.environment.database, - "dataset": task.dataset.name or str(task.dataset.path), - "execution_adapter": task.execution.type, - "harbor_task_id": task.dataset.task_id, - "run_id": observation.run_id, - "workload_id": task.id, - } metrics = { "capture_events": len(eligible_records), "captured_sources": len(captured_records), @@ -115,8 +188,8 @@ def evaluate(observation: TaskObservation, *, experiment: str) -> EvaluationRepo reason=None if observation.status == "completed" else "; ".join(observation.errors), ), "task_provenance_matches": EvaluationValue( - value=observation.harbor.task_checksum == task.dataset.checksum, - reason=f"Expected {task.dataset.checksum!r}; observed {observation.harbor.task_checksum!r}.", + value=observed_checksum == task.dataset.checksum, + reason=(f"Expected {task.dataset.checksum!r}; observed {observed_checksum!r}."), ), "native_acp_evidence_recorded": EvaluationValue( value=not missing_native_artifacts, @@ -171,7 +244,7 @@ def evaluate(observation: TaskObservation, *, experiment: str) -> EvaluationRepo for name, reward in sorted(observation.harbor.rewards.items()) }, } - labels = {"task_outcome": EvaluationValue(value=_task_outcome(observation.harbor))} + labels = {"task_outcome": EvaluationValue(value=_task_outcome(observation.harbor, observation.status))} return EvaluationReport( experiment=experiment, cases=( @@ -192,9 +265,26 @@ def _contains_fragments(value: str, expected: tuple[str, ...]) -> bool: return all(fragment.casefold() in folded for fragment in expected) -def _task_outcome(harbor: HarborTrialObservation) -> str: +def _attributes(observation: TaskObservation) -> dict[str, str]: + task = observation.task + return { + "commit": observation.environment.commit, + "database": observation.environment.database, + "dataset": task.dataset.name or str(task.dataset.path), + "execution_adapter": task.execution.type, + "harbor_task_id": task.dataset.task_id, + "run_id": observation.run_id, + "workload_id": task.id, + } + + +def _task_outcome(harbor: HarborTrialObservation, status: Literal["completed", "failed", "skipped"]) -> str: + if status == "skipped": + return "skipped" if harbor.exception_type is not None: return f"error:{harbor.exception_type}" + if status == "failed": + return "not_passed" if not harbor.rewards: return "unscored" return "passed" if all(float(reward) >= 1 for reward in harbor.rewards.values()) else "not_passed" diff --git a/e2e/bub/src/powercontext_e2e/harbor_agent.py b/e2e/bub/src/powercontext_e2e/harbor_agent.py index 3a9babe0b..4453aac09 100644 --- a/e2e/bub/src/powercontext_e2e/harbor_agent.py +++ b/e2e/bub/src/powercontext_e2e/harbor_agent.py @@ -23,6 +23,7 @@ from harbor.agents.installed import acp as harbor_acp from harbor.environments.base import BaseEnvironment +from harbor.models.agent.context import AgentContext AGENT_ID = "powercontext-bub-acp" REMOTE_BIN_DIR = "/installed-agent/bin" @@ -35,12 +36,15 @@ BUB_VERSION = version("bub") POWERCONTEXT_VERSION = version("powercontext") BUB_ACP_SERVER_VERSION = "0.0.2" +STEP_FAILURE_MARKER = "/logs/agent/powercontext-step-failed" class PowerContextBubAcpAgent(harbor_acp.AcpAgent): """Install Bub through its supported uv tool and plugin commands.""" def __init__(self, **kwargs: Any) -> None: + self._invocation_scopes = tuple(kwargs.pop("invocation_scopes", ())) + self._step_index = 0 super().__init__( registry_entry={ "id": AGENT_ID, @@ -55,6 +59,25 @@ def __init__(self, **kwargs: Any) -> None: **kwargs, ) + @override + async def run(self, instruction: str, environment: BaseEnvironment, context: AgentContext) -> None: + try: + await environment.exec(command=f"touch {STEP_FAILURE_MARKER}") + if not self._invocation_scopes: + await super().run(instruction, environment, context) + else: + if self._step_index >= len(self._invocation_scopes): + invocation = self._step_index + 1 + raise RuntimeError( # noqa: TRY003 + f"No E2E scope configured for agent invocation {invocation}" + ) + scope_id = self._invocation_scopes[self._step_index] + with environment.scoped_exec_env({"POWERCONTEXT_BUB_SCOPE_ID": scope_id}): + await super().run(instruction, environment, context) + await environment.exec(command=f"rm -f {STEP_FAILURE_MARKER}") + finally: + self._step_index += 1 + @override async def install(self, environment: BaseEnvironment) -> None: await self.exec_as_root( diff --git a/e2e/bub/src/powercontext_e2e/models.py b/e2e/bub/src/powercontext_e2e/models.py index df0709ebe..dc18f8bbc 100644 --- a/e2e/bub/src/powercontext_e2e/models.py +++ b/e2e/bub/src/powercontext_e2e/models.py @@ -94,6 +94,7 @@ class HarborTrialObservation(EvidenceModel): trial_name: str | None = None trial_uri: str | None = None task_checksum: str | None = None + source_task_checksum: str | None = None rewards: dict[str, float | int] = Field(default_factory=dict) exception_type: str | None = None exception_message: str | None = None @@ -122,7 +123,7 @@ class TaskObservation(EvidenceModel): run_id: str environment: RunEnvironment task: E2ETask - status: Literal["completed", "failed"] + status: Literal["completed", "failed", "skipped"] errors: tuple[str, ...] = () harbor: HarborTrialObservation capture_records: tuple[CaptureRecord, ...] = () diff --git a/e2e/bub/src/powercontext_e2e/report.py b/e2e/bub/src/powercontext_e2e/report.py index 51c80c0de..bb2b3a3a5 100644 --- a/e2e/bub/src/powercontext_e2e/report.py +++ b/e2e/bub/src/powercontext_e2e/report.py @@ -27,7 +27,7 @@ def render_report(observation: TaskObservation, report: EvaluationReport) -> str markdown = Markdown(renderer=MarkdownRenderer) document = block.Document() children: list[Element] = [] - children.extend(_nodes(markdown, "# PowerContext end-to-end Memory evaluation")) + children.extend(_nodes(markdown, "# PowerContext end-to-end evaluation")) children.append(block.BlankLine(0)) children.extend( _nodes( @@ -50,13 +50,24 @@ def render_report(observation: TaskObservation, report: EvaluationReport) -> str ) children.extend(_nodes(markdown, reward_lines)) children.append(block.BlankLine(0)) - children.extend(_nodes(markdown, "## Memory evaluation")) + children.extend(_nodes(markdown, "## Evaluation")) children.append(block.BlankLine(0)) children.extend(_nodes(markdown, f"```text\n{_evaluation_text(report)}\n```")) document.children = children return markdown.render(document) +def render_evaluation_summary(report: EvaluationReport) -> str: + markdown = Markdown(renderer=MarkdownRenderer) + document = block.Document() + document.children = [ + *_nodes(markdown, "# PowerContext end-to-end evaluation"), + block.BlankLine(0), + *_nodes(markdown, f"```text\n{_evaluation_text(report)}\n```"), + ] + return markdown.render(document) + + 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 1f4675947..c4d930de3 100644 --- a/e2e/bub/src/powercontext_e2e/runner.py +++ b/e2e/bub/src/powercontext_e2e/runner.py @@ -16,31 +16,39 @@ from __future__ import annotations -from contextlib import suppress +import re +import shutil +from collections.abc import Mapping from datetime import UTC, datetime -from pathlib import Path -from typing import Any +from pathlib import Path, PurePosixPath +from typing import Any, Literal, NamedTuple from urllib.parse import unquote, urlparse from uuid import uuid4 +from dirhash import dirhash +from harbor.environments.definition import environment_content_hash from harbor.job import Job from harbor.models.environment_type import EnvironmentType from harbor.models.job.config import DatasetConfig, JobConfig +from harbor.models.task.config import MultiStepRewardStrategy, TaskConfig +from harbor.models.task.paths import TaskPaths +from harbor.models.task.task import Task as HarborTask from harbor.models.trial.config import AgentConfig, EnvironmentConfig, ResourceMode, ServiceVolumeConfig +from harbor.models.trial.config import TaskConfig as HarborTrialTaskConfig +from harbor.models.trial.paths import TrialPaths +from harbor.models.trial.result import StepResult from powercontext.client import PowerContextClient from powercontext.client.settings import ClientSettings -from powercontext.http import ( - ListMemoryEntriesRequest, - PrepareContextRequest, -) +from powercontext.http import ListMemoryEntriesRequest, PrepareContextRequest from .artifacts import write_artifacts -from .catalog import E2ETask -from .evaluation import MemoryEvaluator -from .evidence import fingerprint, load_resolved_instructions, redact +from .catalog import E2ETask, MemoryEvaluationSpec, OutcomeEvaluationSpec +from .evaluation import evaluate_observation +from .evidence import fingerprint, load_resolved_instructions, redact, write_evaluation_report, write_evidence from .harbor_agent import BUB_ACP_SERVER_VERSION, BUB_VERSION from .models import ( CaptureRecord, + EvaluationReport, HarborTrialObservation, MemoryEntrySnapshot, MemorySnapshot, @@ -52,6 +60,7 @@ SourceReferenceSnapshot, TaskObservation, ) +from .report import render_evaluation_summary from .settings import ( HarnessSettings, ModelNotConfiguredError, @@ -60,17 +69,85 @@ powercontext_bub_environment, ) +FailurePolicy = Literal["fail-fast", "collect-all"] +TaskStatus = Literal["completed", "failed", "skipped"] +BATCH_CATEGORY_PREFIX = "batch:" +BATCH_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]*$") -async def evaluate_task( - task: E2ETask, + +class TaskArtifacts(NamedTuple): + capture_records: tuple[CaptureRecord, ...] = () + native_artifacts: tuple[NativeArtifact, ...] = () + resolved_instructions: tuple[ResolvedInstruction, ...] = () + + +class ExecutionGroup(NamedTuple): + tasks: tuple[E2ETask, ...] + batch: str | None + + @property + def output_id(self) -> str: + return self.tasks[0].id if self.batch is None else f"batch-{self.batch}" + + +class SourceTask(NamedTuple): + task: E2ETask + harbor_task: HarborTask + source_steps: tuple[str, ...] + + @property + def runtime_steps(self) -> tuple[str, ...]: + return tuple(f"{self.task.id}-{step}" for step in self.source_steps) + + +class PreparedRuntime(NamedTuple): + task_config: HarborTrialTaskConfig + sources: tuple[SourceTask, ...] + + +class TaskRun(NamedTuple): + task: E2ETask + run_id: str + scope_id: str + started_at: datetime + memory_before: MemorySnapshot + + +async def evaluate_tasks( + tasks: tuple[E2ETask, ...], *, output_dir: Path, settings: HarnessSettings, + failure_policy: FailurePolicy = "collect-all", ) -> bool: - observation = await run_task(task, output_dir=output_dir, settings=settings) - report = MemoryEvaluator.evaluate(observation, experiment=f"e2e:{task.id}") - write_artifacts(observation, report, output_dir, settings=settings) - return report.accepted + observations = await run_task_group( + tasks, + output_dir=output_dir, + settings=settings, + failure_policy=failure_policy, + ) + reports = tuple( + evaluate_observation( + observation, + experiment=f"e2e:{observation.task.id}", + failure_policy=failure_policy, + ) + for observation in observations + ) + if len(observations) == 1: + write_artifacts(observations[0], reports[0], output_dir, settings=settings) + return reports[0].accepted + + for observation, report in zip(observations, reports, strict=True): + write_artifacts(observation, report, output_dir / "tasks" / observation.task.id, settings=settings) + aggregate = EvaluationReport( + experiment=f"e2e:batch:{_task_batch(tasks[0])}", + cases=tuple(case for report in reports for case in report.cases), + ) + output_dir.mkdir(parents=True, exist_ok=True) + write_evaluation_report(output_dir / "eval-report.json", report=aggregate, settings=settings) + write_evidence(output_dir / "report.md", render_evaluation_summary(aggregate), settings) + return aggregate.accepted async def run_tasks( @@ -78,101 +155,257 @@ async def run_tasks( *, output_dir: Path, settings: HarnessSettings, + failure_policy: FailurePolicy = "collect-all", ) -> bool: model_workload_ids = tuple(task.id for task in tasks if task.execution.model) if model_workload_ids and "BUB_MODEL" not in bub_environment(): raise ModelNotConfiguredError(model_workload_ids) accepted = True - for task in tasks: - task_accepted = await evaluate_task(task, output_dir=output_dir / task.id, settings=settings) - accepted = task_accepted and accepted + for group in group_tasks(tasks): + accepted &= await evaluate_tasks( + group.tasks, + output_dir=output_dir / group.output_id, + settings=settings, + failure_policy=failure_policy, + ) return accepted -async def run_task(task: E2ETask, *, output_dir: Path, settings: HarnessSettings) -> TaskObservation: - started_at = datetime.now(UTC) - run_id = f"{task.id}-{uuid4().hex[:12]}" - scope_id = f"e2e:{run_id}" - errors: list[str] = [] - capture_records: tuple[CaptureRecord, ...] = () - native_artifacts: tuple[NativeArtifact, ...] = () - resolved_instructions: tuple[ResolvedInstruction, ...] = () - harbor_observation = HarborTrialObservation() - memory_before = MemorySnapshot() - memory_after = MemorySnapshot() - probes: tuple[RecallProbeObservation, ...] = () - client_settings = ClientSettings() - client_token = None if client_settings.api_token is None else client_settings.api_token.get_secret_value() - - async with PowerContextClient( - client_settings.server_url, - token=client_token, - timeout=client_settings.timeout, - ) as client: +def group_tasks(tasks: tuple[E2ETask, ...]) -> tuple[ExecutionGroup, ...]: + """Group selected tasks only when they share an explicit batch category.""" + + groups: list[ExecutionGroup] = [] + positions: dict[str, int] = {} + for task in tasks: + batch = _task_batch(task) + if batch is None: + groups.append(ExecutionGroup((task,), None)) + elif batch not in positions: + positions[batch] = len(groups) + groups.append(ExecutionGroup((task,), batch)) + else: + index = positions[batch] + groups[index] = ExecutionGroup((*groups[index].tasks, task), batch) + return tuple(ExecutionGroup(group.tasks, group.batch if len(group.tasks) > 1 else None) for group in groups) + + +def _task_batch(task: E2ETask) -> str | None: + batches = tuple( + category.removeprefix(BATCH_CATEGORY_PREFIX) + for category in task.categories + if category.startswith(BATCH_CATEGORY_PREFIX) + ) + if len(batches) > 1 or any(BATCH_NAME_PATTERN.fullmatch(batch) is None for batch in batches): + raise ValueError(f"Task {task.id!r} must declare at most one valid batch category") # noqa: TRY003 + return batches[0] if batches else None + + +async def run_task_group( + tasks: tuple[E2ETask, ...], + *, + output_dir: Path, + settings: HarnessSettings, + failure_policy: FailurePolicy = "collect-all", +) -> tuple[TaskObservation, ...]: + if not tasks: + raise ValueError("At least one E2E task is required") # noqa: TRY003 + + runtime = ( + prepare_runtime_task(tasks, output_dir=output_dir, settings=settings, failure_policy=failure_policy) + if len(tasks) > 1 + else None + ) + run_id = _run_id(tasks) + scopes = {task.id: f"e2e:{run_id}:{task.id}" if runtime else f"e2e:{run_id}" for task in tasks} + runs = {task.id: TaskRun(task, run_id, scopes[task.id], datetime.now(UTC), MemorySnapshot()) for task in tasks} + invocation_scopes = ( + tuple(scopes[source.task.id] for source in runtime.sources for _ in source.runtime_steps) if runtime else () + ) + job_config = _job_config( + tasks[0], + run_id, + scopes[tasks[0].id], + output_dir, + settings, + runtime=runtime, + invocation_scopes=invocation_scopes, + ) + harbor = HarborTrialObservation() + step_results: tuple[StepResult, ...] = () + trial_dir: Path | None = None + execution_errors: list[str] = [] + + async with _powercontext_client() as client: try: - await client.get_readiness() - memory_before = await memory_snapshot(client, scope_id) + memory_tasks = tuple(task for task in tasks if isinstance(task.evaluation, MemoryEvaluationSpec)) + if memory_tasks: + await client.get_readiness() + for task in memory_tasks: + run = runs[task.id] + runs[task.id] = run._replace(memory_before=await memory_snapshot(client, run.scope_id)) output_dir.mkdir(parents=True, exist_ok=True) - job = await Job.create(_job_config(task, run_id, scope_id, output_dir, settings)) - result = await job.run() - harbor_observation, trial_dir = _harbor_observation(result, settings) - if harbor_observation.exception_type is not None: - errors.append( - f"{harbor_observation.exception_type}: {harbor_observation.exception_message or ''}".strip() - ) - if trial_dir is not None: - capture_records = _load_capture_records(trial_dir) - native_artifacts = _native_artifacts(trial_dir, task.execution.native_artifact_names) - resolved_instructions = load_resolved_instructions(trial_dir, settings) - - memory_after = await memory_snapshot(client, scope_id) - probe_observations: list[RecallProbeObservation] = [] - for probe in task.evaluation.probes: - probe_observations.append( - RecallProbeObservation( - id=probe.id, - query=probe.query, - prepared_context=await prepared_context(client, scope_id, probe.query), - ) + harbor, step_results, trial_dir = _harbor_observation(await (await Job.create(job_config)).run(), settings) + if harbor.exception_type is not None: + execution_errors.append(f"{harbor.exception_type}: {harbor.exception_message or ''}".strip()) + except Exception as exc: + execution_errors.append(redact(f"{type(exc).__name__}: {exc}", settings)) + + if runtime is None: + artifacts = _collect_task_artifacts(trial_dir, tasks[0], settings, errors=execution_errors) + reward_failed = _outcome_reward_failed(tasks[0], harbor.rewards) + status: TaskStatus = "failed" if execution_errors or reward_failed else "completed" + return ( + await _finalize_task( + client, + runs[tasks[0].id], + status, + tuple(execution_errors), + harbor, + artifacts, + settings, + ), + ) + + error_owner = next( + ( + source.task.id + for source in reversed(runtime.sources) + if any(step.step_name in source.runtime_steps for step in step_results) + ), + runtime.sources[0].task.id, + ) + observations: list[TaskObservation] = [] + for source in runtime.sources: + owned = tuple(step for step in step_results if step.step_name in source.runtime_steps) + missing = [step for step in source.runtime_steps if step not in {result.step_name for result in owned}] + verifier = owned[-1].verifier_result if owned else None + reward_failed = _outcome_reward_failed( + source.task, + None if verifier is None else verifier.rewards, + ) + status: TaskStatus = ( + "skipped" + if not owned and source.task.id != error_owner + else "failed" + if missing + or any(step.exception_info is not None for step in owned) + or reward_failed + or (source.task.id == error_owner and bool(execution_errors)) + else "completed" + ) + errors = list(execution_errors) if source.task.id == error_owner else [] + if missing and owned: + errors.append(f"Harbor did not execute steps: {missing!r}") + artifacts = _collect_task_artifacts( + trial_dir, + source.task, + settings, + errors=errors, + step_names=source.runtime_steps if status != "skipped" else (), + whole_trial=False, + ) + if errors: + status = "failed" + observations.append( + await _finalize_task( + client, + runs[source.task.id], + status, + tuple(errors), + _source_harbor_observation(harbor, source, status, owned, settings), + artifacts, + settings, ) - probes = tuple(probe_observations) + ) + return tuple(observations) + + +def _run_id(tasks: tuple[E2ETask, ...]) -> str: + name = tasks[0].id if len(tasks) == 1 else f"batch-{_task_batch(tasks[0])}" + return f"{name}-{uuid4().hex[:12]}" + + +def _outcome_reward_failed(task: E2ETask, rewards: Mapping[str, float | int] | None) -> bool: + reward = None if rewards is None else rewards.get("reward") + return isinstance(task.evaluation, OutcomeEvaluationSpec) and reward is not None and float(reward) < 1 + + +def _powercontext_client() -> PowerContextClient: + settings = ClientSettings() + token = None if settings.api_token is None else settings.api_token.get_secret_value() + return PowerContextClient(settings.server_url, token=token, timeout=settings.timeout) + + +async def _finalize_task( + client: PowerContextClient, + run: TaskRun, + status: TaskStatus, + errors: tuple[str, ...], + harbor: HarborTrialObservation, + artifacts: TaskArtifacts, + settings: HarnessSettings, +) -> TaskObservation: + final_errors = list(errors) + memory_after = MemorySnapshot() + probes: tuple[RecallProbeObservation, ...] = () + if status != "skipped" and isinstance(run.task.evaluation, MemoryEvaluationSpec): + try: + memory_after = await memory_snapshot(client, run.scope_id) + probes = await _prepared_probes(client, run.task.evaluation, run.scope_id) except Exception as exc: - errors.append(redact(f"{type(exc).__name__}: {exc}", settings)) - with suppress(Exception): - memory_after = await memory_snapshot(client, scope_id) + final_errors.append(redact(f"{type(exc).__name__}: {exc}", settings)) + if final_errors and status == "completed": + status = "failed" return TaskObservation( - run_id=run_id, - environment=RunEnvironment( - commit=settings.commit_id(), - database=settings.database, - adapter_version=BUB_VERSION, - adapter_protocol_version=BUB_ACP_SERVER_VERSION, - agent_model=bub_environment().get("BUB_MODEL") if task.execution.model else None, - started_at=started_at, - finished_at=datetime.now(UTC), - ), - task=task, - status="completed" if not errors else "failed", - errors=tuple(errors), - harbor=harbor_observation, - capture_records=capture_records, - native_artifacts=native_artifacts, - resolved_instructions=resolved_instructions, - memory_before=memory_before, + run_id=run.run_id, + environment=_run_environment(run.task, run.started_at, settings), + task=run.task, + status=status, + errors=tuple(final_errors), + harbor=harbor, + capture_records=artifacts.capture_records, + native_artifacts=artifacts.native_artifacts, + resolved_instructions=artifacts.resolved_instructions, + memory_before=run.memory_before, memory_after=memory_after, probes=probes, ) +def _source_harbor_observation( + shared: HarborTrialObservation, + source: SourceTask, + status: TaskStatus, + steps: tuple[StepResult, ...], + settings: HarnessSettings, +) -> HarborTrialObservation: + step_exception = next((step.exception_info for step in steps if step.exception_info is not None), None) + verifier = steps[-1].verifier_result if steps else None + updates: dict[str, Any] = { + "source_task_checksum": source.harbor_task.checksum, + "rewards": dict(verifier.rewards or {}) if verifier is not None else {}, + "exception_type": None if step_exception is None else step_exception.exception_type, + "exception_message": ( + None if step_exception is None else redact(step_exception.exception_message or "", settings) + ), + } + if status == "failed" and step_exception is None and shared.exception_type is not None: + updates.update(exception_type=shared.exception_type, exception_message=shared.exception_message) + return shared.model_copy(update=updates) + + def _job_config( task: E2ETask, run_id: str, scope_id: str, output_dir: Path, settings: HarnessSettings, + *, + runtime: PreparedRuntime | None = None, + invocation_scopes: tuple[str, ...] = (), ) -> JobConfig: repository = settings.repository_path() mounts: list[ServiceVolumeConfig] = [ @@ -193,6 +426,7 @@ def _job_config( "bind": {"create_host_path": False}, }) + evaluation = task.evaluation agent_env = powercontext_bub_environment() if task.execution.model: agent_env.update(bub_environment()) @@ -203,10 +437,16 @@ def _job_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(task.evaluation.checkpoint_every_events), - "POWERCONTEXT_BUB_CAPTURE_EVENTS": str(task.evaluation.capture_events).lower(), + "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(task.evaluation.max_event_bytes), + "POWERCONTEXT_BUB_CAPTURE_MAX_BYTES": str( + evaluation.max_event_bytes if isinstance(evaluation, MemoryEvaluationSpec) else 8192 + ), "POWERCONTEXT_BUB_SCOPE_ID": scope_id, }) if settings.agent_proxy_url is not None: @@ -219,6 +459,10 @@ def _job_config( "https_proxy": proxy_url, "no_proxy": "127.0.0.1,localhost,host-gateway,powercontext", }) + agent_kwargs: dict[str, Any] = {} + if runtime is not None: + agent_env.pop("POWERCONTEXT_BUB_SCOPE_ID") + agent_kwargs["invocation_scopes"] = invocation_scopes return JobConfig( job_name=run_id, @@ -238,9 +482,11 @@ def _job_config( AgentConfig( import_path="powercontext_e2e.harbor_agent:PowerContextBubAcpAgent", env=agent_env, + kwargs=agent_kwargs, ) ], - datasets=[_dataset_config(task, repository)], + datasets=[] if runtime else [_dataset_config(task, repository)], + tasks=[runtime.task_config] if runtime else [], ) @@ -251,6 +497,125 @@ def _dataset_config(task: E2ETask, repository: Path) -> DatasetConfig: return DatasetConfig(name=dataset.name, version=dataset.version, task_names=[dataset.task_id]) +def prepare_runtime_task( + tasks: tuple[E2ETask, ...], + *, + output_dir: Path, + settings: HarnessSettings, + failure_policy: FailurePolicy, +) -> PreparedRuntime: + """Assemble compatible source tasks into one run-local Harbor task.""" + + sources = _validate_batch_compatibility(tasks, settings) + batch = _task_batch(tasks[0]) + if len(tasks) < 2 or batch is None: + raise ValueError("Runtime aggregation requires at least two tasks in one explicit batch") # noqa: TRY003 + + runtime_paths = TaskPaths(output_dir / "harbor-runtime-dataset" / f"batch-{batch}-{uuid4().hex[:12]}") + runtime_paths.task_dir.mkdir(parents=True) + first_paths = sources[0].harbor_task.paths + for source_dir, target_dir in ( + (first_paths.environment_dir, runtime_paths.environment_dir), + (first_paths.tests_dir, runtime_paths.tests_dir), + ): + if source_dir.exists(): + shutil.copytree(source_dir, target_dir) + runtime_paths.steps_dir.mkdir() + for source in sources: + for source_step, runtime_step in zip(source.source_steps, source.runtime_steps, strict=True): + shutil.copytree(source.harbor_task.paths.step_dir(source_step), runtime_paths.step_dir(runtime_step)) + + config = _runtime_task_config(sources, failure_policy) + runtime_paths.config_path.write_text(config.model_dump_toml(), encoding="utf-8") + HarborTask(runtime_paths.task_dir) + return PreparedRuntime(HarborTrialTaskConfig(path=runtime_paths.task_dir), sources) + + +def _validate_batch_compatibility(tasks: tuple[E2ETask, ...], settings: HarnessSettings) -> tuple[SourceTask, ...]: + batch = _task_batch(tasks[0]) + if batch is None or any(_task_batch(task) != batch for task in tasks): + raise ValueError("Runtime aggregation requires one explicit shared batch") # noqa: TRY003 + if any(task.dataset.path is None for task in tasks): + raise ValueError(f"Batch {batch!r} requires local Harbor datasets") # noqa: TRY003 + if len({task.id for task in tasks}) != len(tasks): + raise ValueError(f"Batch {batch!r} task IDs must be unique") # noqa: TRY003 + + repository = settings.repository_path() + sources = tuple(_load_source_task(task, repository) for task in tasks) + if any( + source.harbor_task.config.multi_step_reward_strategy is not MultiStepRewardStrategy.FINAL for source in sources + ): + raise ValueError(f"Batch {batch!r} requires Harbor's final multi-step reward strategy") # noqa: TRY003 + first_profile = _runtime_profile(sources[0]) + for source in sources[1:]: + profile = _runtime_profile(source) + if incompatible := [name for name, value in first_profile.items() if profile[name] != value]: + raise ValueError( # noqa: TRY003 + f"Source task {source.task.id!r} has incompatible batch settings: {incompatible!r}" + ) + return sources + + +def _load_source_task(task: E2ETask, repository: Path) -> SourceTask: + dataset_path = task.dataset.path + if dataset_path is None: + raise ValueError(f"Source task {task.id!r} does not use a local Harbor dataset") # noqa: TRY003 + task_dir = repository / dataset_path / task.dataset.task_id + try: + harbor_task = HarborTask(task_dir) + except Exception as exc: + raise ValueError(f"Source task {task.id!r} cannot be loaded from {task_dir}") from exc # noqa: TRY003 + if harbor_task.checksum != task.dataset.checksum: + raise ValueError(f"Source task {task.id!r} checksum changed") # noqa: TRY003 + return SourceTask(task, harbor_task, _task_layout(task, harbor_task)) + + +def _runtime_profile(source: SourceTask) -> dict[str, Any]: + task = source.task + paths = source.harbor_task.paths + evaluation = task.evaluation + return { + "dataset": task.dataset.model_dump(mode="json", exclude={"task_id", "checksum"}), + "execution": task.execution.model_dump(mode="json"), + "capture": ( + evaluation.capture_events, + evaluation.checkpoint_every_events, + evaluation.max_event_bytes, + ) + if isinstance(evaluation, MemoryEvaluationSpec) + else (False, 5, 8192), + "harbor": source.harbor_task.config.model_dump(mode="json", exclude={"steps"}), + "environment": environment_content_hash( + paths.environment_dir, + docker_image=source.harbor_task.config.environment.docker_image, + ), + "tests": dirhash(paths.tests_dir, "sha256") if paths.tests_dir.is_dir() else None, + } + + +def _runtime_task_config(sources: tuple[SourceTask, ...], failure_policy: FailurePolicy) -> TaskConfig: + min_reward = 1.0 if failure_policy == "fail-fast" else None + steps = [ + step.model_copy(update={"name": runtime_name, "min_reward": min_reward}) + for source in sources + for step, runtime_name in zip(source.harbor_task.config.steps or (), source.runtime_steps, strict=True) + ] + return sources[0].harbor_task.config.model_copy(update={"steps": steps}) + + +def _task_layout(task: E2ETask, harbor_task: HarborTask) -> tuple[str, ...]: + steps = tuple(step.name for step in harbor_task.config.steps or ()) + paths = tuple(PurePosixPath(step) for step in steps) + if not steps or len(steps) != len(set(steps)): + raise ValueError(f"Source task {task.id!r} must have unique Harbor steps") # noqa: TRY003 + if any( + "\\" in step or path.is_absolute() or len(path.parts) != 1 or path.parts[0] == ".." or path.as_posix() != step + for step, path in zip(steps, paths, strict=True) + ): + raise ValueError(f"Source task {task.id!r} step names must be single path components") # noqa: TRY003 + return steps + + async def memory_snapshot(client: PowerContextClient, scope_id: str) -> MemorySnapshot: response = await client.list_memory_entries(ListMemoryEntriesRequest(scope_id=scope_id)) return MemorySnapshot( @@ -277,9 +642,48 @@ async def prepared_context(client: PowerContextClient, scope_id: str, query: str return PreparedContextSnapshot(status=prepared.status.value, content=prepared.content or "") -def _harbor_observation(result: Any, settings: HarnessSettings) -> tuple[HarborTrialObservation, Path | None]: +async def _prepared_probes( + client: PowerContextClient, + evaluation: MemoryEvaluationSpec, + scope_id: str, +) -> tuple[RecallProbeObservation, ...]: + probes = [] + for probe in evaluation.probes: + probes.append( + RecallProbeObservation( + id=probe.id, + query=probe.query, + prepared_context=await prepared_context(client, scope_id, probe.query), + ) + ) + return tuple(probes) + + +def _run_environment(task: E2ETask, started_at: datetime, settings: HarnessSettings) -> RunEnvironment: + return RunEnvironment( + commit=settings.commit_id(), + database=settings.database, + adapter_version=BUB_VERSION, + adapter_protocol_version=BUB_ACP_SERVER_VERSION, + agent_model=bub_environment().get("BUB_MODEL") if task.execution.model else None, + started_at=started_at, + finished_at=datetime.now(UTC), + ) + + +def _harbor_observation( + result: Any, settings: HarnessSettings +) -> tuple[HarborTrialObservation, tuple[StepResult, ...], Path | None]: if not result.trial_results: - return HarborTrialObservation(job_id=str(result.id)), None + return ( + HarborTrialObservation( + job_id=str(result.id), + exception_type="HarborJobError", + exception_message="Harbor job returned no trial results.", + ), + (), + None, + ) trial = result.trial_results[0] rewards = trial.verifier_result.rewards if trial.verifier_result is not None else {} exception = trial.exception_info @@ -295,6 +699,7 @@ def _harbor_observation(result: Any, settings: HarnessSettings) -> tuple[HarborT started_at=trial.started_at, finished_at=trial.finished_at, ), + tuple(trial.step_results or ()), _trial_dir(trial.trial_uri), ) @@ -304,9 +709,49 @@ def _trial_dir(trial_uri: str) -> Path | None: return Path(unquote(parsed.path)) if parsed.scheme == "file" else None -def _load_capture_records(trial_dir: Path) -> tuple[CaptureRecord, ...]: +def _task_artifacts( + trial_dir: Path | None, + task: E2ETask, + settings: HarnessSettings, + *, + step_names: tuple[str, ...] = (), + whole_trial: bool = True, +) -> TaskArtifacts: + if trial_dir is None: + return TaskArtifacts() + roots = (trial_dir,) if whole_trial else tuple(TrialPaths(trial_dir).step_dir(name) for name in step_names) + prefixes = tuple(f"steps/{name}/" for name in step_names) + instructions = load_resolved_instructions(trial_dir, settings) + return TaskArtifacts( + tuple(record for root in roots for record in _load_capture_records(root)), + tuple( + artifact + for root in roots + for artifact in _native_artifacts(root, task.execution.native_artifact_names, relative_to=trial_dir) + ), + tuple(instruction for instruction in instructions if whole_trial or instruction.artifact.startswith(prefixes)), + ) + + +def _collect_task_artifacts( + trial_dir: Path | None, + task: E2ETask, + settings: HarnessSettings, + *, + errors: list[str], + step_names: tuple[str, ...] = (), + whole_trial: bool = True, +) -> TaskArtifacts: + try: + return _task_artifacts(trial_dir, task, settings, step_names=step_names, whole_trial=whole_trial) + except Exception as exc: + errors.append(redact(f"{type(exc).__name__}: {exc}", settings)) + return TaskArtifacts() + + +def _load_capture_records(root: Path) -> tuple[CaptureRecord, ...]: records: list[CaptureRecord] = [] - for path in sorted(trial_dir.rglob("powercontext-capture.jsonl")): + for path in sorted(root.rglob("powercontext-capture.jsonl")): records.extend( CaptureRecord.model_validate_json(line) for line in path.read_text(encoding="utf-8").splitlines() @@ -315,7 +760,9 @@ def _load_capture_records(trial_dir: Path) -> tuple[CaptureRecord, ...]: return tuple(records) -def _native_artifacts(trial_dir: Path, names: frozenset[str]) -> tuple[NativeArtifact, ...]: +def _native_artifacts( + root: Path, names: frozenset[str], *, relative_to: Path | None = None +) -> tuple[NativeArtifact, ...]: return tuple( - fingerprint(path, relative_to=trial_dir) for path in sorted(trial_dir.rglob("*")) if path.name in names + fingerprint(path, relative_to=relative_to or root) for path in sorted(root.rglob("*")) if path.name in names ) diff --git a/e2e/bub/tasks/project-database-decision.yaml b/e2e/bub/tasks/acceptance-01-project-database-decision.yaml similarity index 92% rename from e2e/bub/tasks/project-database-decision.yaml rename to e2e/bub/tasks/acceptance-01-project-database-decision.yaml index 6c1adacdf..d8d729fbc 100644 --- a/e2e/bub/tasks/project-database-decision.yaml +++ b/e2e/bub/tasks/acceptance-01-project-database-decision.yaml @@ -18,10 +18,11 @@ categories: - acceptance - sample - smoke + - batch:acceptance dataset: path: e2e/bub/harbor-tasks task_id: project-database-decision - checksum: 8c1f9d11581c4cea3f7df56fd0ddfc03ce07f6ab29a86cde6a14c52a349264f9 + checksum: 4d0e45489408dbece683ab91b0d77419bb90a202d1c66c1209e69315ad97e799 execution: type: bub model: false diff --git a/e2e/bub/tasks/acceptance-02-failure-policy-timeout.yaml b/e2e/bub/tasks/acceptance-02-failure-policy-timeout.yaml new file mode 100644 index 000000000..bb59a0876 --- /dev/null +++ b/e2e/bub/tasks/acceptance-02-failure-policy-timeout.yaml @@ -0,0 +1,33 @@ +# 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: failure-policy-timeout +categories: + - acceptance + - fixture + - batch:acceptance +dataset: + path: e2e/bub/harbor-tasks + task_id: failure-policy-timeout + checksum: d0f01c9a9f1b331ff3904ddc864c4aa5ec673fa2179d9947b853827fb54714d4 +execution: + type: bub + model: false + max_steps: 10 + max_tokens: 4096 +evaluation: + expected_execution: + collect-all: failed + fail-fast: failed diff --git a/e2e/bub/tasks/acceptance-03-failure-policy-followup.yaml b/e2e/bub/tasks/acceptance-03-failure-policy-followup.yaml new file mode 100644 index 000000000..2a3f78f62 --- /dev/null +++ b/e2e/bub/tasks/acceptance-03-failure-policy-followup.yaml @@ -0,0 +1,33 @@ +# 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: failure-policy-followup +categories: + - acceptance + - fixture + - batch:acceptance +dataset: + path: e2e/bub/harbor-tasks + task_id: failure-policy-followup + checksum: 350d4719cac43cd99e93ea7d92ca6bcb651e1e5f1805a041a1abbf58af5216fc +execution: + type: bub + model: false + max_steps: 10 + max_tokens: 4096 +evaluation: + expected_execution: + collect-all: completed + fail-fast: skipped diff --git a/e2e/bub/tasks/locomo-multihop-football.yaml b/e2e/bub/tasks/locomo-multihop-football.yaml index 5fb5f4305..0e54cce07 100644 --- a/e2e/bub/tasks/locomo-multihop-football.yaml +++ b/e2e/bub/tasks/locomo-multihop-football.yaml @@ -17,6 +17,7 @@ id: locomo-multihop-football categories: - acceptance - sample + - batch:locomo provenance: source: benchmark/locomo/dataset/locomo10.json revision: 4448275ea2c5cd0af5774d80aea7b05b5a16e1b996caf8554ca3d762a301ae84 @@ -28,7 +29,7 @@ provenance: dataset: path: e2e/bub/harbor-tasks task_id: locomo-multihop-football - checksum: 1b3697582140d2b2adb82adcd00cc33a4203da4284f55da3da124f5ee98320db + checksum: d2700bd79b6f5e7d96bb0ab802c414a4a90fd652eb2612036a99aebd2b25f9e4 execution: type: bub model: false diff --git a/e2e/bub/tasks/locomo-open-pastries.yaml b/e2e/bub/tasks/locomo-open-pastries.yaml index 256864298..1c48cab1a 100644 --- a/e2e/bub/tasks/locomo-open-pastries.yaml +++ b/e2e/bub/tasks/locomo-open-pastries.yaml @@ -17,6 +17,7 @@ id: locomo-open-pastries categories: - acceptance - sample + - batch:locomo provenance: source: benchmark/locomo/dataset/locomo10.json revision: 4448275ea2c5cd0af5774d80aea7b05b5a16e1b996caf8554ca3d762a301ae84 @@ -28,7 +29,7 @@ provenance: dataset: path: e2e/bub/harbor-tasks task_id: locomo-open-pastries - checksum: c1c4df25435b4faba60fada92891a3423cd4826236c04daf84e7ee721e6dd094 + checksum: e5578c5f860c439230a9e67399ee61f9303ddf95831b08d78a40385c7f964831 execution: type: bub model: false diff --git a/e2e/bub/tasks/locomo-support-group.yaml b/e2e/bub/tasks/locomo-support-group.yaml index 8f9dd5e23..3ad7f23f0 100644 --- a/e2e/bub/tasks/locomo-support-group.yaml +++ b/e2e/bub/tasks/locomo-support-group.yaml @@ -17,6 +17,7 @@ id: locomo-support-group categories: - acceptance - sample + - batch:locomo provenance: source: benchmark/locomo/dataset/locomo10.json revision: 4448275ea2c5cd0af5774d80aea7b05b5a16e1b996caf8554ca3d762a301ae84 @@ -28,7 +29,7 @@ provenance: dataset: path: e2e/bub/harbor-tasks task_id: locomo-support-group - checksum: 01fec08696a68bf4b6c70529a8d8cf864fe8e7e2a3915a263a653bfe380a5b69 + checksum: f1bdfa62f960e3a749b84c19e3cb7661bb3ab0d4ba049547efdbcfb870db18fa execution: type: bub model: false diff --git a/e2e/bub/tasks/locomo-temporal-banker.yaml b/e2e/bub/tasks/locomo-temporal-banker.yaml index 73afa602e..98b46dd69 100644 --- a/e2e/bub/tasks/locomo-temporal-banker.yaml +++ b/e2e/bub/tasks/locomo-temporal-banker.yaml @@ -17,6 +17,7 @@ id: locomo-temporal-banker categories: - acceptance - sample + - batch:locomo provenance: source: benchmark/locomo/dataset/locomo10.json revision: 4448275ea2c5cd0af5774d80aea7b05b5a16e1b996caf8554ca3d762a301ae84 @@ -28,7 +29,7 @@ provenance: dataset: path: e2e/bub/harbor-tasks task_id: locomo-temporal-banker - checksum: 6b7577995e466e35964a904b8c6dee83fc54e9b2818a8ee236b079126b6e0afe + checksum: f9982061c303d10de5b44f02152e6a3c9abab5e5d76f60fed76f561a603cc363 execution: type: bub model: false diff --git a/e2e/bub/tests/test_workload_catalog.py b/e2e/bub/tests/test_workload_catalog.py index a355294e5..3ec4ffdc2 100644 --- a/e2e/bub/tests/test_workload_catalog.py +++ b/e2e/bub/tests/test_workload_catalog.py @@ -16,7 +16,10 @@ from pathlib import Path +import pytest + from powercontext_e2e.catalog import load_tasks, select_tasks +from powercontext_e2e.runner import group_tasks def test_workloads_can_be_selected_by_multiple_ids_or_category() -> None: @@ -24,11 +27,13 @@ def test_workloads_can_be_selected_by_multiple_ids_or_category() -> None: tasks = load_tasks(repository / "e2e" / "bub" / "tasks") assert [task.id for task in tasks] == [ + "project-database-decision", + "failure-policy-timeout", + "failure-policy-followup", "locomo-multihop-football", "locomo-open-pastries", "locomo-support-group", "locomo-temporal-banker", - "project-database-decision", "terminal-bench-db-wal-recovery", ] assert {task.execution.type for task in tasks} == {"bub"} @@ -45,9 +50,24 @@ def test_workloads_can_be_selected_by_multiple_ids_or_category() -> None: "terminal-bench-db-wal-recovery", ] assert [task.id for task in acceptance] == [ + "project-database-decision", + "failure-policy-timeout", + "failure-policy-followup", "locomo-multihop-football", "locomo-open-pastries", "locomo-support-group", "locomo-temporal-banker", - "project-database-decision", ] + assert [group.tasks for group in group_tasks(acceptance)] == [ + tuple(task for task in acceptance if "batch:acceptance" in task.categories), + tuple(task for task in acceptance if "batch:locomo" in task.categories), + ] + + +def test_batch_category_cannot_escape_the_runtime_dataset() -> None: + repository = Path(__file__).resolve().parents[3] + task = load_tasks(repository / "e2e" / "bub" / "tasks" / "locomo-support-group.yaml")[0] + task = task.model_copy(update={"categories": (*task.categories[:-1], "batch:x/../../evil")}) + + with pytest.raises(ValueError, match="valid batch category"): + group_tasks((task,))