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..9690cf8b05 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,71 @@ evaluation and report at `batch-<name>/`. `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 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.
+
+- 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 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 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.
+
+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"
+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
+<output>/
+  paired-report.json
+  report.md
+  <workload-id>/trial-<n>/<off|on>/
+    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,
@@ -176,11 +242,14 @@ from that container. In the fixed nested-container harness, `host-gateway` addre
 proxy exposed there can be passed as `http://host-gateway:<port>`. 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,
-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/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..25ca8108a6
--- /dev/null
+++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/instruction.md
@@ -0,0 +1,3 @@
+Which database did we decide to use for the ticket service, and what shard count did we agree on? Write your answer
+to /workspace/answer.json as a JSON object with only two keys, a `database` string and a `shard_count` integer, for
+example `{"database": "<name>", "shard_count": <number>}`. 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
new file mode 100644
index 0000000000..097ba1e711
--- /dev/null
+++ b/e2e/bub/harbor-tasks/project-decision-continuation/steps/recall/tests/grade.py
@@ -0,0 +1,76 @@
+# 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.
+
+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 json
+import sys
+import unicodedata
+from pathlib import Path
+
+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:
+    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(
+        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")
+
+
+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..58fe7aa152
--- /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.json /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..0b4298b1d6
--- /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: 87429c6ac133a99343d72e8fc614fb2f18d773a83c318097ec085a610bb3315e
+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..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 E2ETask, MemoryEvaluationSpec
-from .harbor_agent import BUB_ACP_SERVER_VERSION, BUB_VERSION, REMOTE_CODEX_AUTH
+from .catalog import ContinuationEvaluationSpec, E2ETask, MemoryEvaluationSpec
+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,17 +39,22 @@ 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,
         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:
@@ -62,28 +69,20 @@ 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,
         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 +92,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 +115,35 @@ 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
+
+
+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/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..c0fbcb751d
--- /dev/null
+++ b/e2e/bub/src/powercontext_e2e/paired.py
@@ -0,0 +1,346 @@
+# 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}
+    recall_steps = {task.id: _continuation(task).recall_step 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],
+                        recall_step=recall_steps[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 = _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,
+    *,
+    trial: int,
+    arm: Arm,
+    position: int,
+    recall_session: int,
+    recall_step: str,
+    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, 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 = (
+        (
+            *(redact(failure, settings) for failure in (recorder.failures if recorder is not None else ())),
+            *treatment_failures(sessions, recall_session),
+        )
+        if arm == "on"
+        else ()
+    )
+    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_rewards(step_results),
+        outcome=arm_outcome(
+            step_results,
+            harbor,
+            recall_step=recall_step,
+            harness_failed=bool(errors),
+            treatment_failures=treatment,
+        ),
+        errors=tuple(errors),
+        sessions=sessions,
+        treatment_failures=treatment,
+    )
+
+
+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.
+
+    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)
+    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 recall.preparations <= before.preparations:
+        failures.append("PowerContext was not asked for 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,
+                **dict(_summary([o for o in observations if o.task_id == task_id])),
+            )
+            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..0043d0b49a 100644
--- a/e2e/bub/src/powercontext_e2e/runner.py
+++ b/e2e/bub/src/powercontext_e2e/runner.py
@@ -42,10 +42,10 @@
 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
+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):
@@ -151,11 +154,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 +169,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 +418,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,
     *,
@@ -418,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/src/powercontext_e2e/sessions.py b/e2e/bub/src/powercontext_e2e/sessions.py
new file mode 100644
index 0000000000..3c4c036dd1
--- /dev/null
+++ b/e2e/bub/src/powercontext_e2e/sessions.py
@@ -0,0 +1,82 @@
+# 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, *, flush: bool) -> SessionSnapshot:
+    """Snapshot the Scope after one session, first flushing captured Sources into Memory when a session follows.
+
+    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 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:
+            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. 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, *, 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
+        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_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..637c1d6ea9 100644
--- a/e2e/bub/tests/test_harbor_job_config.py
+++ b/e2e/bub/tests/test_harbor_job_config.py
@@ -166,3 +166,41 @@ 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 == {}
+
+
+@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)
+    ]
diff --git a/e2e/bub/tests/test_paired.py b/e2e/bub/tests/test_paired.py
new file mode 100644
index 0000000000..6e69b0e3d6
--- /dev/null
+++ b/e2e/bub/tests/test_paired.py
@@ -0,0 +1,329 @@
+# 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
+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 (
+    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
+
+_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"),
+    [
+        ('{"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_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"
+
+    _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.json", 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, asked: int = 0, ready: int = 0) -> SessionSnapshot:
+    return SessionSnapshot(
+        session=session,
+        flush_rounds=1,
+        sources=sources,
+        memory_pending=0,
+        memory_entries=memory,
+        preparations=asked,
+        ready_preparations=ready,
+    )
+
+
+def test_treatment_passes_when_recall_received_context_from_captured_memory() -> None:
+    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, 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_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))
+
+
+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"),
+    [
+        ({"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]], *, 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)
+
+    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, flush=True))
+
+    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, flush=True))
+
+    assert client.flushes == 2
+    assert snapshot.flush_rounds == 2
+
+
+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 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))