diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
index 4e55b1a2..1afdf305 100644
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -43,6 +43,13 @@ jobs:
           uv pip install solc-select
           uv run solc-select install 0.8.29
           uv run solc-select use 0.8.29
+          # certoraRun resolves Certora-convention names like `solc8.29` as
+          # literal executables; solc-select only ships the `solc` shim, so
+          # fetch the static binary under that exact name.
+          sudo curl -fsSL https://github.com/ethereum/solidity/releases/download/v0.8.29/solc-static-linux \
+            -o /usr/local/bin/solc8.29
+          sudo chmod +x /usr/local/bin/solc8.29
+          solc8.29 --version
 
       - name: async runners
         run: |
diff --git a/composer/bind.py b/composer/bind.py
index 3235bd78..919e21c5 100644
--- a/composer/bind.py
+++ b/composer/bind.py
@@ -32,3 +32,11 @@
         os.environ.get("COMPOSER_RECORD_OUT"),
         no_thinking=bool(os.environ.get("COMPOSER_RECORD_NO_THINKING")),
     )
+
+if (_resp := os.environ.get("COMPOSER_RESPONSE_TAPE")):
+    # Scripted human replies for console HITL interrupts (commit approvals,
+    # requirement relaxations). Independent of — and replayed alongside —
+    # COMPOSER_TEST_TAPE; the tape module exposes ``install_response_tape``.
+    import importlib
+    _rmod = importlib.import_module(f"composer.testing.ui_harness_{_resp}")
+    _rmod.install_response_tape()
diff --git a/composer/io/graph_runner.py b/composer/io/graph_runner.py
index 4db36b8e..a57a29f3 100644
--- a/composer/io/graph_runner.py
+++ b/composer/io/graph_runner.py
@@ -85,6 +85,7 @@ async def run_graph[H, S: StateLike, I: StateLike, C: StateLike | None](
                 curr_input = graph_input
                 graph_input = None
                 interrupted = False
+                interrupt_data: H | None = None
                 async for (ty, payload) in graph.astream(
                     curr_input, config=curr_config, context=ctxt, stream_mode=["checkpoints", "updates", "custom"]
                 ):
@@ -105,35 +106,28 @@ async def run_graph[H, S: StateLike, I: StateLike, C: StateLike | None](
                             assert human_handler is not None
                             if "configurable" in curr_config and "checkpoint_id" in curr_config["configurable"]:
                                 del curr_config["configurable"]["checkpoint_id"]
-                            interrupt_data = cast(H, payload["__interrupt__"][0].value)
-                            curr_state = cast(S, (await graph.aget_state({"configurable": {"thread_id": tid}})).values)
-                            human_response = await human_handler(interrupt_data, curr_state)
-                            graph_input = Command(resume=human_response)
-                            interrupted = True
-                            break
-                        elif ty == "custom":
-                            event_sink(
-                                CustomUpdate(payload, thread_id=tid, checkpoint_id=curr_checkpoint) # pyright: ignore[reportPossiblyUnboundVariable]
-                            )
-                        else:
-                            assert ty == "updates"
-                            if "__interrupt__" in payload:
-                                assert human_handler is not None
-                                if "configurable" in curr_config and "checkpoint_id" in curr_config["configurable"]:
-                                    del curr_config["configurable"]["checkpoint_id"]
+                            # Record the interrupt but keep draining the stream:
+                            # the interrupt checkpoint may not be committed when
+                            # this update is yielded, and the resume below targets
+                            # the thread's latest checkpoint. Breaking out here
+                            # races that write — a resume against the stale
+                            # checkpoint replays the previous node (duplicated LLM
+                            # turns / re-fired interrupts / dropped state updates).
+                            if not interrupted:
                                 interrupt_data = cast(H, payload["__interrupt__"][0].value)
-                                curr_state = cast(S, (await graph.aget_state({"configurable": {"thread_id": tid}})).values)
-                                human_response = await human_handler(interrupt_data, curr_state)
-                                graph_input = Command(resume=human_response)
                                 interrupted = True
-                                break
-                            event_sink(
-                                StateUpdate(
-                                    payload, thread_id=tid
-                                )
+                            continue
+                        event_sink(
+                            StateUpdate(
+                                payload, thread_id=tid
                             )
-                    if interrupted:
-                        continue
+                        )
+                if interrupted:
+                    assert human_handler is not None
+                    curr_state = cast(S, (await graph.aget_state({"configurable": {"thread_id": tid}})).values)
+                    human_response = await human_handler(cast(H, interrupt_data), curr_state)
+                    graph_input = Command(resume=human_response)
+                    continue
 
                 result_state = (await graph.aget_state({"configurable": {"thread_id": tid}})).values
                 return cast(S, result_state)
diff --git a/composer/natreq/extractor.py b/composer/natreq/extractor.py
index b13cac94..0be34a5f 100644
--- a/composer/natreq/extractor.py
+++ b/composer/natreq/extractor.py
@@ -31,6 +31,11 @@
 from composer.io.context import with_handler, run_graph
 from composer.io.event_handler import NullEventHandler
 from composer.ui.tool_display import tool_display
+from composer.diagnostics.timing import set_current_task_id
+
+# Requirements extraction runs as its own ``run_graph`` (no ``run_task`` scope);
+# set a task_id so the harness tape can address its LLM calls.
+REQUIREMENTS_TASK_ID = "requirements"
 
 
 @dataclass
@@ -172,6 +177,7 @@ async def get_requirements(
         graph_input = ExtractionInput(input=input_text, memory=None, did_read=False)
 
         async with with_handler(io, NullEventHandler()):  # type: ignore[arg-type]
-            final_state = await run_graph(built, ExtractionContext(rag_db=db), graph_input, config, description="Requirements extraction")
+            with set_current_task_id(REQUIREMENTS_TASK_ID):
+                final_state = await run_graph(built, ExtractionContext(rag_db=db), graph_input, config, description="Requirements extraction")  
         assert "reqs" in final_state
         return ExtractionResult(reqs=final_state["reqs"], thread_id=thread_id)
diff --git a/composer/prover/agentic_analyzer.py b/composer/prover/agentic_analyzer.py
index f09a1dca..39531a2e 100644
--- a/composer/prover/agentic_analyzer.py
+++ b/composer/prover/agentic_analyzer.py
@@ -51,8 +51,10 @@
 from composer.prover.ptypes import RuleResult, RulePath, AnalyzedDiagnosis
 from composer.prover.report_store import ReportStore
 from composer.spec.graph_builder import bind_standard, run_to_completion
-from composer.spec.util import uniq_thread_id
+from composer.spec.util import uniq_thread_id, string_hash
 from composer.templates.loader import load_jinja_template
+from composer.diagnostics.timing import set_current_task_id
+from composer.prover.cex_task_ids import cex_rule_task_id, cex_aggregator_task_id
 from composer.tools.thinking import RoughDraftState, get_rough_draft_tools
 
 
@@ -479,8 +481,17 @@ async def _process_rule(
                 for i, entry in enumerate(scratchpad)
             ]
 
+        async def _process_rule_addressed(
+            rule_group: FailingRule,
+        ) -> list[_PerRuleRootCause]:
+            # Distinct task_id lane per concurrent per-rule analysis (keyed by the
+            # prover tool_call_id + rule name) so the harness tape can address the
+            # gathered sub-agents; in production this just scopes timing per rule.
+            with set_current_task_id(cex_rule_task_id(tool_call_id, rule_group.rule_name)):
+                return await _process_rule(rule_group)
+
         per_rule = await asyncio.gather(
-            *(_process_rule(rg) for rg in failing_rules)
+            *(_process_rule_addressed(rg) for rg in failing_rules)
         )
         # Flatten preserving rule order (gather preserves input order).
         all_causes: list[_PerRuleRootCause] = [
@@ -501,9 +512,10 @@ async def _process_rule(
         # Cross-rule aggregator. Sees the per-rule root causes (already
         # within-rule-clustered) and partitions them by cross-rule
         # equivalence.
-        agg = await self._run_aggregator(
-            all_causes, report_fs=report_fs, tool_call_id=tool_call_id,
-        )
+        with set_current_task_id(cex_aggregator_task_id(tool_call_id)):
+            agg = await self._run_aggregator(
+                all_causes, report_fs=report_fs, tool_call_id=tool_call_id,
+            )
 
         # Validator (see ``_aggregator_validator``) already enforced that
         # every cause_index is in range, no duplicates, every input
@@ -514,7 +526,10 @@ async def _process_rule(
 
         for partition in agg.partitions:
             attributed: list[RulePath] = []
-            report_key = uuid.uuid4().hex
+            # Content-addressed rather than a uuid: stable across identical
+            # re-runs and reconstructible by the harness tape (which scripts the
+            # diagnosis text), while staying opaque to the author.
+            report_key = string_hash(partition.diagnosis)
             for i in partition.cause_indices:
                 for attr in all_causes[i].attributed_rule_paths:
                     attributed.append(attr)
diff --git a/composer/prover/cex_task_ids.py b/composer/prover/cex_task_ids.py
new file mode 100644
index 00000000..bc6edf59
--- /dev/null
+++ b/composer/prover/cex_task_ids.py
@@ -0,0 +1,23 @@
+"""Task-ids for the agentic CEX analyzer's sub-agents.
+
+The codegen pipeline runs through ``run_graph`` without a ``run_task`` scope, and
+the per-rule CEX analyzers fan out under ``asyncio.gather`` (see
+``AgenticCexHandler.analyze``). Concurrent sub-agents that share one task_id
+can't be routed deterministically by the harness tape, so each is scoped to its
+own task_id lane via ``set_current_task_id``.
+
+The keys fold in the prover ``tool_call_id`` so analyses from distinct prover
+runs land in distinct lanes (otherwise repeated runs of the same rule collide).
+Centralized here so the harness tape reconstructs the same lane keys — the tape
+controls the ``tool_call_id`` it scripts, so the derivation stays deterministic.
+"""
+
+
+def cex_rule_task_id(tool_call_id: str, rule_name: str) -> str:
+    """Lane for the per-rule CEX analyzer of ``rule_name`` under one prover run."""
+    return f"cex-{tool_call_id}-{rule_name}"
+
+
+def cex_aggregator_task_id(tool_call_id: str) -> str:
+    """Lane for the cross-rule CEX aggregator of one prover run."""
+    return f"cex-agg-{tool_call_id}"
diff --git a/composer/spec/cex_remediation.py b/composer/spec/cex_remediation.py
index 3328fdc9..17cd5797 100644
--- a/composer/spec/cex_remediation.py
+++ b/composer/spec/cex_remediation.py
@@ -28,7 +28,6 @@
 from typing_extensions import TypedDict
 from dataclasses import dataclass
 import difflib
-import uuid
 
 from pydantic import BaseModel, Field
 
@@ -44,7 +43,7 @@
 from composer.core.state import AIComposerState
 from composer.input.files import Document
 from composer.spec.graph_builder import bind_standard, run_to_completion
-from composer.spec.util import uniq_thread_id
+from composer.spec.util import uniq_thread_id, string_hash
 from composer.tools.thinking import RoughDraftState, get_rough_draft_tools
 from composer.ui.tool_display import tool_display
 from composer.prover.report_store import ReportStore
@@ -430,7 +429,9 @@ async def run(self) -> str:
             # apply_remediation_proposal can fetch it without the
             # codegen author having to re-emit (or paraphrase) the spec
             # text. The codegen author only ever sees the diff + the key.
-            proposal_key = uuid.uuid4().hex
+            # Content-addressed (not a uuid): deterministic across identical
+            # re-runs and reconstructible by the harness tape, still opaque.
+            proposal_key = string_hash(result.proposed_cvl)
             await deps.proposals.record(proposal_key, result.proposed_cvl)
             return _render_remediation(
                 result,
diff --git a/composer/spec/code_explorer.py b/composer/spec/code_explorer.py
index fd4906da..1d6744a6 100644
--- a/composer/spec/code_explorer.py
+++ b/composer/spec/code_explorer.py
@@ -67,9 +67,7 @@ def _code_explorer_graph(
         sys_prompt
     ).with_initial_prompt(
         "Answer the following question about the source code"
-    ).compile_async(
-        checkpointer=InMemorySaver()
-    )
+    ).compile_async()
 
 class _ExploreCodeCommon(BaseModel):
     """
diff --git a/composer/testing/harness_tape.py b/composer/testing/harness_tape.py
index 55773688..ef92bf73 100644
--- a/composer/testing/harness_tape.py
+++ b/composer/testing/harness_tape.py
@@ -1,4 +1,4 @@
-from typing import Any, Callable, Sequence, override, cast
+from typing import Any, Callable, Iterator, Sequence, override, cast
 import random
 import asyncio
 
@@ -118,6 +118,52 @@ async def ainvoke(
         return cast(AIMessage, lane[i])
 
 
+class _DummyUploader:
+    """A ``FileUploader`` stand-in that never touches a Files API: every input is
+    read from disk and returned as an in-memory text document. Installed under the
+    harness so a taped run does no real uploads — the codegen path uploads the spec
+    + interface via ``upload_text_file_if_needed``, which would otherwise hit the
+    live Files API."""
+
+    async def upload_text_file_if_needed(self, file_path: Any) -> Any:
+        return self._inline(file_path)
+
+    async def upload_file_if_needed(self, file_path: Any) -> Any:
+        return self._inline(file_path)
+
+    async def get_document(self, path: Any) -> Any:
+        import os
+        return self._inline(path) if os.path.isfile(str(path)) else None
+
+    @staticmethod
+    def _inline(path: Any) -> Any:
+        import os
+        from pathlib import Path
+        from composer.input.files import InMemoryTextFile
+        p = str(path)
+        return InMemoryTextFile(
+            basename=os.path.basename(p),
+            string_contents=Path(p).read_text(encoding="utf-8"),
+            provider="anthropic"
+        )
+
+
+# The currently-installed tape state for this process. The seam patches below are
+# installed once per process as stable dispatcher functions that read these slots
+# at call time; re-installing a tape (a later test in the same xdist worker) only
+# swaps the slot. Rebinding the module attributes per install instead would strand
+# any module that imported a patched name by value under the earlier test's tape.
+_active_fake: Any = None
+_active_responses: Iterator[str] | None = None
+_llm_seams_patched = False
+_prompt_seam_patched = False
+
+
+def _current_fake() -> Any:
+    assert _active_fake is not None, "no harness tape installed"
+    return _active_fake
+
+
 def install_fake_llm(fake: Any) -> None:
     """Route every LLM-construction path in the pipeline to ``fake``.
 
@@ -129,11 +175,17 @@ def install_fake_llm(fake: Any) -> None:
     and short-circuits ``get_provider_for`` before it tries to ``_lookup`` a fake
     model name.
 
-    Patches the ``registry`` / ``services`` source bindings, so it must run BEFORE
-    the entry path imports ``get_provider_for`` by name (``composer/bind.py`` is
-    that hook). Eager-import callers (the integration test) additionally rebind
-    their own ``get_provider_for`` reference to the patched one.
+    The first install in a process must run BEFORE the entry path imports
+    ``get_provider_for`` by name (``composer/bind.py`` is that hook). Eager-import
+    callers (the integration tests) additionally rebind their own
+    ``get_provider_for`` reference to the patched one. Later installs just swap
+    the active fake.
     """
+    global _active_fake, _llm_seams_patched
+    _active_fake = fake
+    if _llm_seams_patched:
+        return
+
     import composer.llm.registry as registry
     import composer.workflow.services as services
 
@@ -141,7 +193,7 @@ class _FakeProvider:
         provider : ProviderKind = "anthropic"
 
         def builder_for(self, *, cache_level: Any = None, disable_thinking: bool = False) -> Any:
-            return fake
+            return _current_fake()
 
     fp = _FakeProvider()
 
@@ -153,5 +205,51 @@ def _fake_get_provider_for(
         return fp
 
     registry.get_provider_for = _fake_get_provider_for
-    services.create_llm = lambda args: fake
-    services.create_llm_base = lambda args: fake
+    registry.uploader_for = lambda _provider: _DummyUploader()
+    services.create_llm = lambda args: _current_fake()
+    services.create_llm_base = lambda args: _current_fake()
+    _llm_seams_patched = True
+
+
+def install_fake_responses(responses: list[str]) -> None:
+    """Replay scripted human replies for console HITL interrupts.
+
+    Patches ``composer.ui.prompt.prompt_input`` (and the binding
+    ``composer.ui.console`` imported it under) to return each ``responses`` entry
+    in call order, applying the call's own ``filter`` as a sanity check. Raises if
+    the tape is exhausted or a scripted reply fails the prompt's filter. Install
+    before the entry path imports ``prompt_input`` by name (``composer/bind.py`` is
+    that hook). Replayed alongside ``install_fake_llm``.
+
+    Covers only the console HITL path; the autoprove interactive-refinement
+    conversation uses a different input path and is not handled here.
+    """
+    global _active_responses, _prompt_seam_patched
+    _active_responses = iter(responses)
+    if _prompt_seam_patched:
+        return
+
+    import composer.ui.prompt as prompt_mod
+
+    def _fake_prompt_input(
+        prompt_str: str,
+        debug_thunk: Callable[[], None],
+        filter: Callable[[str], str | None] | None = None,
+    ) -> str:
+        assert _active_responses is not None, "no response tape installed"
+        try:
+            resp = next(_active_responses)
+        except StopIteration:
+            raise RuntimeError(
+                f"response tape exhausted — no scripted reply for HITL prompt: {prompt_str!r}"
+            )
+        if filter is not None and (rejection := filter(resp)) is not None:
+            raise RuntimeError(
+                f"scripted response {resp!r} rejected for prompt {prompt_str!r}: {rejection}"
+            )
+        return resp
+
+    prompt_mod.prompt_input = _fake_prompt_input
+    import composer.ui.console as console_mod
+    console_mod.prompt_input = _fake_prompt_input
+    _prompt_seam_patched = True
diff --git a/composer/testing/ui_harness_codegen_capped_vault.py b/composer/testing/ui_harness_codegen_capped_vault.py
new file mode 100644
index 00000000..8d79929d
--- /dev/null
+++ b/composer/testing/ui_harness_codegen_capped_vault.py
@@ -0,0 +1,662 @@
+"""Fake-LLM end-to-end harness tape for the **codegen** pipeline.
+
+Replays a scripted run of ``console-codegen`` against the
+``test_scenarios/codegen_capped_vault`` scenario with zero real LLM calls;
+everything else (solc, the Certora prover, Postgres, the VFS, the working-spec
+machinery) runs for real. Install with ``COMPOSER_TEST_TAPE=codegen_capped_vault``.
+
+Scenario shape
+--------------
+A ``CappedVault`` with a per-account cap. The author's first draft is faithful
+but omits the ``totalDeposited`` update; the spec's ``depositRaisesBalance`` rule
+is over-strong (a zero-value deposit is a legal no-op). So the first prover run
+fails BOTH rules, for different reasons:
+
+  - ``depositIncreasesTotal``  — implementation bug → fixed by editing Solidity.
+  - ``depositRaisesBalance``   — spec bug → fixed spec-side via ``cex_remediation``.
+
+The two failures drive two PARALLEL per-rule CEX analyses, which is the whole
+point of the addressing work: each runs in its own task_id lane
+(``cex_rule_task_id(<prover tc id>, <rule>)``), with the cross-rule aggregator in
+``cex_aggregator_task_id(<prover tc id>)``. We PIN the run-1 ``certora_prover``
+tool_call_id to ``cvprun1`` so those lane keys are derivable here.
+
+The system doc states two contradictory requirements (always-accept-deposits vs
+enforce-the-cap). The requirements extractor emits both; the judge finds R2
+satisfied and R1 violated; the author relaxes R1 (it contradicts R2) and delivers.
+
+Lanes / task_ids
+----------------
+    requirements                          : requirements extraction agent
+    codegen                               : the main author + every SEQUENTIAL
+                                            sub-agent it spawns (cex_remediation,
+                                            its summary_critic, the requirements
+                                            judge) — all inherit the codegen
+                                            task_id, so they interleave here in
+                                            call order.
+    cex-cvprun1-depositIncreasesTotal     : per-rule CEX analyzer (rule A)
+    cex-cvprun1-depositRaisesBalance      : per-rule CEX analyzer (rule B)
+    cex-agg-cvprun1                       : cross-rule CEX aggregator
+
+Human-in-the-loop (replayed via ``COMPOSER_RESPONSE_TAPE``, not the LLM tape)
+----------------------------------------------------------------------------
+Two author turns trigger console HITL interrupts; the fake LLM only supplies AI
+turns, so ``install_response_tape`` scripts the human replies (``_HUMAN_RESPONSES``),
+consumed in call order:
+  - ``commit_working_spec``            → ``ACCEPTED`` (promotes the remediated rule B
+                                         to the master spec so the final master
+                                         prover run stamps the ``prover`` validation).
+  - ``requirement_relaxation_request`` → ``ACCEPTED`` (skips R1).
+
+Verify loop
+-----------
+Prover verdicts are real, so this is a record→verify→iterate artifact like the
+autoprove tapes. Spots to confirm on the first live run are flagged inline:
+the exact prover verdicts, the order ``group_failing`` yields the two rules (which
+fixes which cause_index the aggregator partitions reference), and whether the
+post-skip ``requirements_evaluation`` (re-run) is what stamps the reqs validation.
+"""
+
+from typing import Any
+import uuid
+
+from langchain_core.messages import AIMessage, BaseMessage
+from langchain_core.messages.tool import ToolCall
+
+from composer.testing.harness_tape import (
+    HarnessFakeLLM, install_fake_llm, install_fake_responses,
+)
+from composer.spec.util import string_hash
+from composer.workflow.executor import CODEGEN_TASK_ID, RESUME_COMMENTARY_TASK_ID
+from composer.natreq.extractor import REQUIREMENTS_TASK_ID
+from composer.prover.cex_task_ids import cex_rule_task_id, cex_aggregator_task_id
+
+
+# The pinned tool_call_id of the run-1 certora_prover call, so the per-rule /
+# aggregator CEX lane keys are derivable (the handler keys lanes off it).
+_PROVER_TC = "cvprun1"
+
+
+def _tc(name: str, *, _id: str | None = None, **args: Any) -> ToolCall:
+    """Tool-call dict. ``_id`` pins the id (needed where downstream addressing
+    derives from the tool_call_id, e.g. the run-1 prover call); otherwise random."""
+    return {
+        "id": _id if _id is not None else f"toolu_{uuid.uuid4().hex[:20]}",
+        "name": name,
+        "args": args,
+        "type": "tool_call",
+    }
+
+
+def _ai(text: str = "", *tool_calls: ToolCall) -> AIMessage:
+    """A tape entry: optional text plus zero or more tool_calls. A turn with no
+    tool_calls ends the agent loop (returns to output_key extraction)."""
+    content: list[str | dict] = []
+    if text:
+        content.append(text)
+    content.extend(
+        {"type": "tool_use", "id": t["id"], "name": t["name"], "input": t["args"]}
+        for t in tool_calls
+    )
+    return AIMessage(content=content, tool_calls=list(tool_calls))
+
+
+# ---------------------------------------------------------------------------
+# Requirements (must match VERBATIM across: the `reqs` result, the judge's
+# per-requirement `requirement` field, and the relaxation. judge_res_checker
+# compares the judge's text to the extracted reqs exactly.)
+# ---------------------------------------------------------------------------
+
+_R1 = "A deposit must always succeed and must never be rejected."
+_R2 = (
+    "An account's balance must never exceed the cap of 1000; any deposit that "
+    "would exceed the cap must be rejected."
+)
+
+
+# ---------------------------------------------------------------------------
+# Source + spec artifacts. Real tools gatekeep these: solc compiles the impl,
+# the Certora prover proves/CEXes the spec.
+# ---------------------------------------------------------------------------
+
+# First draft: faithful, but deposit() omits `totalDeposited += amount`.
+_BUGGY_IMPL = """\
+// SPDX-License-Identifier: UNLICENSED
+pragma solidity ^0.8.29;
+
+contract CappedVault {
+    uint256 public constant CAP = 1000;
+    mapping(address => uint256) public balance;
+    uint256 public totalDeposited;
+
+    function deposit(uint256 amount) external {
+        require(balance[msg.sender] + amount <= CAP);
+        balance[msg.sender] += amount;
+    }
+
+    function withdraw(uint256 amount) external {
+        require(balance[msg.sender] >= amount);
+        balance[msg.sender] -= amount;
+        totalDeposited -= amount;
+    }
+}
+"""
+
+# The implementation fix for depositIncreasesTotal — add the missing update.
+# (edit_file old/new; the `+= amount;` line is unique to deposit.)
+_IMPL_FIX_OLD = """\
+        balance[msg.sender] += amount;
+    }"""
+_IMPL_FIX_NEW = """\
+        balance[msg.sender] += amount;
+        totalDeposited += amount;
+    }"""
+
+# The remediated spec cex_remediation proposes for depositRaisesBalance: identical
+# to the master spec except the assertion is guarded on `amount > 0`. Staged as
+# the working spec, then committed to master.
+_REMEDIATED_SPEC = """\
+methods {
+    function balance(address) external returns (uint256) envfree;
+    function totalDeposited() external returns (uint256) envfree;
+    function CAP() external returns (uint256) envfree;
+    function deposit(uint256) external;
+    function withdraw(uint256) external;
+}
+
+rule depositIncreasesTotal(uint256 amount) {
+    env e;
+    mathint before = totalDeposited();
+    deposit(e, amount);
+    assert to_mathint(totalDeposited()) == before + amount,
+        "a successful deposit must increase totalDeposited by exactly amount";
+}
+
+rule depositRaisesBalance(uint256 amount) {
+    env e;
+    address caller = e.msg.sender;
+    mathint before = balance(caller);
+    deposit(e, amount);
+    assert amount > 0 => to_mathint(balance(caller)) > before,
+        "a positive deposit must raise the caller's balance";
+}
+"""
+
+
+# ---------------------------------------------------------------------------
+# CEX diagnoses. The aggregator partition's `diagnosis` text is what the handler
+# content-addresses into the report_key (string_hash(partition.diagnosis)). So the
+# author's cex_remediation(report_key=...) key is derived from _DIAG_B here, and
+# the staged-proposal key from _REMEDIATED_SPEC — both match production exactly
+# now that report_key / proposal_key are content-addressed.
+# ---------------------------------------------------------------------------
+
+_DIAG_A = (
+    "deposit() updates the caller's balance but never updates the totalDeposited "
+    "accumulator, so the running total stays put while balances grow. "
+    "depositIncreasesTotal requires totalDeposited to rise by exactly the deposited "
+    "amount, so any nonzero deposit violates it. Implementation defect, not a spec "
+    "problem: the function is missing `totalDeposited += amount`."
+)
+_DIAG_B = (
+    "depositRaisesBalance asserts the caller's balance strictly increases after any "
+    "successful deposit, but a zero-value deposit is a legal no-op that leaves the "
+    "balance unchanged (the counterexample deposits amount == 0). The rule is "
+    "over-strong as written; a strict increase should only be required for amount > 0."
+)
+
+_KEY_A = string_hash(_DIAG_A)
+_KEY_B = string_hash(_DIAG_B)
+_PROPOSAL_KEY = string_hash(_REMEDIATED_SPEC)
+
+
+# ---------------------------------------------------------------------------
+# Lane: requirements extraction
+# ---------------------------------------------------------------------------
+# Tools: memory, reqs (result), human_in_the_loop, cvl_manual_search, rough draft.
+# _extraction_res_checker gates `reqs` behind read_rough_draft (did_read).
+
+_REQUIREMENTS_TAPE: list[BaseMessage] = [
+    _ai(
+        "Reading the system doc + spec to pull out the stated requirements.",
+        _tc(
+            "write_rough_draft",
+            rough_draft=(
+                "The system doc states two requirements that pull against each "
+                "other: (1) deposits must always succeed, (2) a per-account cap of "
+                "1000 must be enforced, rejecting over-cap deposits. Extract both "
+                "verbatim; the downstream judge + relaxation resolve the conflict."
+            ),
+        ),
+    ),
+    _ai("Reading the draft back before emitting.", _tc("read_rough_draft")),
+    _ai("Requirements extracted.", _tc("result", value=[_R1, _R2])),
+]
+
+
+# ---------------------------------------------------------------------------
+# Lane: per-rule CEX analyzer — depositIncreasesTotal (rule A)
+# ---------------------------------------------------------------------------
+# Tools: rough draft + report-scoped fs tools. Validator gates `result` behind
+# read_rough_draft. result payload = _PerCexCommitWrapper{commit}.
+
+_CEX_RULE_A_TAPE: list[BaseMessage] = [
+    _ai(
+        "Analyzing the depositIncreasesTotal counterexample.",
+        _tc(
+            "write_rough_draft",
+            rough_draft=(
+                "CEX deposits a nonzero amount; balance updates but totalDeposited "
+                "is unchanged in the post-state. The source's deposit() has no "
+                "totalDeposited write. Root cause: missing accumulator update — an "
+                "implementation defect."
+            ),
+        ),
+    ),
+    _ai("Reading the draft.", _tc("read_rough_draft")),
+    _ai(
+        "Committing the root cause.",
+        _tc(
+            "result",
+            commit={
+                "decision": "new",
+                "text": _DIAG_A,
+                "evidence": (
+                    "Post-state totalDeposited equals its pre-state value while "
+                    "balance[caller] increased by the deposited amount; deposit() "
+                    "contains no write to totalDeposited."
+                ),
+            },
+        ),
+    ),
+]
+
+
+# ---------------------------------------------------------------------------
+# Lane: per-rule CEX analyzer — depositRaisesBalance (rule B)
+# ---------------------------------------------------------------------------
+
+_CEX_RULE_B_TAPE: list[BaseMessage] = [
+    _ai(
+        "Analyzing the depositRaisesBalance counterexample.",
+        _tc(
+            "write_rough_draft",
+            rough_draft=(
+                "CEX has amount == 0. deposit(0) is a legal no-op, so the strict "
+                "`balance > before` assertion fails. The implementation is fine; the "
+                "rule over-constrains the zero case."
+            ),
+        ),
+    ),
+    _ai("Reading the draft.", _tc("read_rough_draft")),
+    _ai(
+        "Committing the root cause.",
+        _tc(
+            "result",
+            commit={
+                "decision": "new",
+                "text": _DIAG_B,
+                "evidence": (
+                    "Counterexample operand amount == 0; balance[caller] is identical "
+                    "in pre- and post-state, so the strict-increase assertion fails."
+                ),
+            },
+        ),
+    ),
+]
+
+
+# ---------------------------------------------------------------------------
+# Lane: cross-rule CEX aggregator
+# ---------------------------------------------------------------------------
+# Sees the two per-rule root causes (indices 0 and 1, in group_failing order) and
+# partitions them. The two causes are distinct, so two singleton partitions. The
+# partition `diagnosis` texts are content-addressed into the report_keys.
+#
+# VERIFY: cause_indices below assume group_failing yields depositIncreasesTotal
+# first (index 0) and depositRaisesBalance second (index 1). If the live order is
+# reversed, swap the two cause_indices. Either way the run won't error (the keys
+# resolve by diagnosis text), but the rendered per-rule attribution depends on it.
+
+_CEX_AGGREGATOR_TAPE: list[BaseMessage] = [
+    _ai(
+        "Partitioning the two root causes.",
+        _tc(
+            "write_rough_draft",
+            rough_draft=(
+                "Two unrelated causes: (0) a missing totalDeposited update — an "
+                "implementation defect; (1) an over-strong spec assertion on the "
+                "zero-deposit case. No cross-rule equivalence; one partition each."
+            ),
+        ),
+    ),
+    _ai("Reading the draft.", _tc("read_rough_draft")),
+    _ai(
+        "Two distinct diagnoses.",
+        _tc(
+            "result",
+            partitions=[
+                {"diagnosis": _DIAG_A, "cause_indices": [0]},
+                {"diagnosis": _DIAG_B, "cause_indices": [1]},
+            ],
+        ),
+    ),
+]
+
+
+# ---------------------------------------------------------------------------
+# Lane: codegen — the main author plus every sequential sub-agent it spawns
+# (cex_remediation, that remediator's summary_critic, the requirements judge),
+# all in call order.
+# ---------------------------------------------------------------------------
+
+_CODEGEN_TAPE: list[BaseMessage] = [
+    # T1 — initial (buggy) draft.
+    _ai(
+        "Drafting the CappedVault implementation from the interface + spec.",
+        _tc("put_file", files={"src/CappedVault.sol": _BUGGY_IMPL}),
+    ),
+
+    # T2 — first prover run (all rules). PINNED id so the CEX lanes are derivable.
+    # Triggers the two per-rule analyzers + aggregator (lanes above).
+    _ai(
+        "Running the prover on the first draft.",
+        _tc(
+            "certora_prover",
+            _id=_PROVER_TC,
+            source_files=["src/CappedVault.sol"],
+            target_contract="CappedVault",
+            compiler_version="solc8.29",
+            loop_iter=1,
+            rule=None,
+            use_working_spec=False,
+        ),
+    ),
+
+    # T3 — fix the implementation defect (depositIncreasesTotal).
+    _ai(
+        "The CEX report shows deposit() never updates totalDeposited — adding it.",
+        _tc(
+            "edit_file",
+            path="src/CappedVault.sol",
+            old_string=_IMPL_FIX_OLD,
+            new_string=_IMPL_FIX_NEW,
+        ),
+    ),
+
+    # T4 — the other failure is a spec bug; route it to cex_remediation. Author is
+    # blocked from editing the spec inline; report_key is the content-addressed key
+    # for diagnosis B (derived above).
+    _ai(
+        "The depositRaisesBalance failure is a spec over-constraint, not a code "
+        "bug. Handing it to cex_remediation.",
+        _tc("cex_remediation", report_key=_KEY_B, target_spec_path="rules.spec"),
+    ),
+
+    # --- remediation sub-agent (runs inside the codegen lane) ---
+    # RM1 — remediator runs its proposal past the critic before delivering.
+    _ai(
+        "Proposing a guarded assertion; checking it with the summary critic first.",
+        _tc(
+            "summary_critic",
+            proposed_cvl=_REMEDIATED_SPEC,
+            proposed_addendum=None,
+            target_call="deposit",
+            rule_under_repair="depositRaisesBalance",
+            cex_diagnosis=_DIAG_B,
+        ),
+    ),
+    # --- summary_critic sub-agent (rough-draft gated) ---
+    _ai(
+        "Reviewing the proposed guard for soundness + faithfulness.",
+        _tc(
+            "write_rough_draft",
+            rough_draft=(
+                "`amount > 0 => balance' > balance` is sound: a zero deposit is a "
+                "legal no-op per the system doc, and for positive deposits the cap "
+                "bounds the add so balance strictly increases. Faithful to the doc. "
+                "Verdict: sound."
+            ),
+        ),
+    ),
+    _ai("Reading the draft.", _tc("read_rough_draft")),
+    _ai(
+        "Critic verdict.",
+        _tc("result", sound=True, issues=[], suggested_direction=None),
+    ),
+    # RM2 — remediator delivers the proposal (stored under _PROPOSAL_KEY).
+    _ai(
+        "Critic approved; delivering the remediation.",
+        _tc(
+            "result",
+            proposed_cvl=_REMEDIATED_SPEC,
+            rationale=(
+                "Root cause B (over-strong assertion): guard the strict-increase "
+                "assertion on `amount > 0`, leaving the zero-deposit no-op out of "
+                "scope. No implementation change required."
+            ),
+            addendum=None,
+        ),
+    ),
+
+    # T5 — stage the proposal as the working spec.
+    _ai(
+        "Staging the remediated spec as my working copy.",
+        _tc("apply_remediation_proposal", proposal_key=_PROPOSAL_KEY),
+    ),
+
+    # T6 — re-run against the WORKING spec: fixed impl + guarded rule B → both pass.
+    # (Working-spec run does not stamp the prover validation; that needs master.)
+    _ai(
+        "Re-running the prover against the working spec.",
+        _tc(
+            "certora_prover",
+            source_files=["src/CappedVault.sol"],
+            target_contract="CappedVault",
+            compiler_version="solc8.29",
+            loop_iter=1,
+            rule=None,
+            use_working_spec=True,
+        ),
+    ),
+
+    # T7 — promote the working spec to master (HITL: driver approves).
+    _ai(
+        "Both rules verify; committing the spec fix to master.",
+        _tc(
+            "commit_working_spec",
+            explanation=(
+                "Guards depositRaisesBalance on `amount > 0` so a zero-value deposit "
+                "(a legal no-op) no longer counts as a violation. Behavior-preserving "
+                "for positive deposits."
+            ),
+        ),
+    ),
+
+    # T8 — final MASTER-spec run (rule=None, use_working_spec=False): stamps prover.
+    _ai(
+        "Verifying the committed master spec.",
+        _tc(
+            "certora_prover",
+            source_files=["src/CappedVault.sol"],
+            target_contract="CappedVault",
+            compiler_version="solc8.29",
+            loop_iter=1,
+            rule=None,
+            use_working_spec=False,
+        ),
+    ),
+
+    # T9 — evaluate the requirements (spawns the judge sub-agent).
+    _ai(
+        "Code + spec verify. Evaluating the requirements.",
+        _tc("requirements_evaluation"),
+    ),
+    # --- judge sub-agent: no rough draft (memory unset → no did_read gate). Each
+    # requirement text must match the extracted reqs verbatim, by 1-based number.
+    _ai(
+        "Requirements verdict.",
+        _tc(
+            "result",
+            judgement_result=[
+                {
+                    "classification": "VIOLATED",
+                    "requirement": _R1,
+                    "requirement_number": 1,
+                    "commentary": (
+                        "deposit() reverts once the cap would be exceeded, so a "
+                        "deposit is not always accepted."
+                    ),
+                },
+                {
+                    "classification": "SATISFIED",
+                    "requirement": _R2,
+                    "requirement_number": 2,
+                    "commentary": "The cap require() rejects any over-cap deposit.",
+                },
+            ],
+        ),
+    ),
+
+    # T10 — R1 is violated only because it contradicts R2 (the enforced cap). Relax
+    # it (HITL: driver replies ACCEPTED → skipped_reqs = {1}).
+    _ai(
+        "R1 (always accept) directly contradicts R2 (enforce the cap); the cap is "
+        "the real requirement. Requesting to relax R1.",
+        _tc(
+            "requirement_relaxation_request",
+            context="The cap (R2) is enforced, which necessarily makes some deposits revert.",
+            req_number=1,
+            req_text=_R1,
+            judgment="VIOLATED: deposit() reverts once the cap would be exceeded.",
+            explanation=(
+                "R1 (deposits always succeed) is mutually exclusive with R2's enforced "
+                "cap; R2 is the real requirement, so R1 should be relaxed."
+            ),
+        ),
+    ),
+
+    # T11 — re-evaluate now that R1 is skipped; this is what should stamp the reqs
+    # validation (R1 ignored, R2 satisfied). VERIFY this is the stamping path.
+    _ai(
+        "Re-evaluating with R1 relaxed.",
+        _tc("requirements_evaluation"),
+    ),
+    _ai(
+        "Requirements verdict (R1 relaxed).",
+        _tc(
+            "result",
+            judgement_result=[
+                {
+                    "classification": "VIOLATED",
+                    "requirement": _R1,
+                    "requirement_number": 1,
+                    "commentary": "Still violated, but relaxed as contradictory with R2.",
+                },
+                {
+                    "classification": "SATISFIED",
+                    "requirement": _R2,
+                    "requirement_number": 2,
+                    "commentary": "The cap require() rejects any over-cap deposit.",
+                },
+            ],
+        ),
+    ),
+
+    # T12 — deliver. check_completion requires the prover (T8) + reqs validations.
+    _ai(
+        "All rules verify and the requirements are resolved. Delivering.",
+        _tc(
+            "result",
+            source=["src/CappedVault.sol"],
+            comments=(
+                "CappedVault implements per-account deposits with a 1000 cap. The "
+                "totalDeposited accumulator update was added after the prover flagged "
+                "depositIncreasesTotal. depositRaisesBalance was over-strong on the "
+                "zero-deposit case and was remediated spec-side to guard on amount > 0. "
+                "Requirement R1 (always accept deposits) was relaxed: it contradicts "
+                "the enforced cap (R2)."
+            ),
+        ),
+    ),
+
+]
+
+
+
+
+# ---------------------------------------------------------------------------
+# Lane map + install
+# ---------------------------------------------------------------------------
+
+# Post-run resume-commentary lane. ``create_resume_commentary`` runs after the
+# main graph as a structured-output call: ``with_structured_output(ResumeCommentary)``
+# → ``bind_tools`` (no-op'd by the fake) piped through ``PydanticToolsParser``, which
+# matches a tool call named after the schema class. So the scripted response is a
+# ``ResumeCommentary`` tool call. Its own lane keeps the agent from consuming it.
+_RESUME_COMMENTARY_TAPE: list[BaseMessage] = [
+    _ai(
+        "",
+        _tc(
+            "ResumeCommentary",
+            commentary=(
+                "Implemented CappedVault: added the missing totalDeposited update, "
+                "remediated depositRaisesBalance to guard on amount > 0, and relaxed "
+                "the contradictory deposits-always-succeed requirement."
+            ),
+            interface_path="src/CappedVault.sol",
+        ),
+    ),
+]
+
+
+_CODEGEN_LANES: dict[str, list[BaseMessage]] = {
+    REQUIREMENTS_TASK_ID: _REQUIREMENTS_TAPE,
+    CODEGEN_TASK_ID: _CODEGEN_TAPE,
+    RESUME_COMMENTARY_TASK_ID: _RESUME_COMMENTARY_TAPE,
+    cex_rule_task_id(_PROVER_TC, "depositIncreasesTotal"): _CEX_RULE_A_TAPE,
+    cex_rule_task_id(_PROVER_TC, "depositRaisesBalance"): _CEX_RULE_B_TAPE,
+    cex_aggregator_task_id(_PROVER_TC): _CEX_AGGREGATOR_TAPE,
+}
+
+
+def get_codegen_capped_vault_llm(with_delay: bool = True) -> HarnessFakeLLM:
+    """A fresh fake LLM loaded with the codegen tape (independent lane cursors)."""
+    return HarnessFakeLLM(lanes=_CODEGEN_LANES, with_human_delay=with_delay)
+
+
+def install_harness_tape(with_delay: bool = True) -> HarnessFakeLLM:
+    """Route the codegen pipeline's models to the fake LLM + disable the
+    agent-index cache. ``composer/bind.py`` calls this when
+    ``COMPOSER_TEST_TAPE=codegen_capped_vault`` is set, before the entry path
+    imports ``get_provider_for`` by name (so the patch lands first)."""
+    fake = get_codegen_capped_vault_llm(with_delay)
+    import composer.spec.agent_index as a_ind
+    a_ind._UNSAFE_DISABLE_CACHE = True
+    install_fake_llm(fake)
+    return fake
+
+
+# ---------------------------------------------------------------------------
+# Console HITL replies — replayed via COMPOSER_RESPONSE_TAPE, alongside the LLM
+# tape. Consumed in call order: the commit_working_spec approval (T7) then the R1
+# relaxation (T10). Both interrupts accept on a leading "ACCEPTED".
+# ---------------------------------------------------------------------------
+
+_HUMAN_RESPONSES = [
+    "ACCEPTED",
+    "ACCEPTED",
+]
+
+
+def install_response_tape() -> None:
+    """Replay this scenario's console HITL replies. ``composer/bind.py`` calls
+    this when ``COMPOSER_RESPONSE_TAPE=codegen_capped_vault`` is set."""
+    install_fake_responses(_HUMAN_RESPONSES)
+
+
+__all__ = [
+    "get_codegen_capped_vault_llm",
+    "install_harness_tape",
+    "install_response_tape",
+]
diff --git a/composer/tools/relaxation.py b/composer/tools/relaxation.py
index 2297f604..1cb8e109 100644
--- a/composer/tools/relaxation.py
+++ b/composer/tools/relaxation.py
@@ -1,7 +1,6 @@
 from composer.core.state import AIComposerState
 from composer.human.types import RequirementRelaxationType
 from graphcore.tools.human import human_interaction_tool
-from composer.ui.tool_display import tool_display
 
 def _maybe_relax(s: AIComposerState, q: RequirementRelaxationType, resp: str) -> dict:
     if resp.startswith("ACCEPTED"):
diff --git a/composer/workflow/executor.py b/composer/workflow/executor.py
index a1a6d7c3..d565bbde 100644
--- a/composer/workflow/executor.py
+++ b/composer/workflow/executor.py
@@ -8,7 +8,6 @@
 from langchain_core.language_models.chat_models import BaseChatModel
 from langchain_core.tools import BaseTool
 
-
 from graphcore.graph import Builder
 from graphcore.tools.vfs import VFSState, VFSAccessor
 
@@ -22,7 +21,7 @@
 
 from composer.llm.provider import ModelProvider
 
-from composer.kb.knowledge_base import DefaultEmbedder, kb_tools as make_kb_tools
+from composer.kb.knowledge_base import DefaultEmbedder, kb_tools as make_kb_tools, DEFAULT_KB_NS as _KB_NS
 from composer.workflow.factories import get_vfs_tools, get_memory_ns
 from composer.workflow.services import standard_connections, IndexedConnections
 from composer.workflow.types import PromptParams, WorkflowSuccess, WorkflowFailure, WorkflowCrash, WorkflowResult
@@ -42,6 +41,7 @@
 from composer.templates.loader import load_jinja_template
 from composer.io.protocol import CodeGenIOHandler, WorkflowPurpose
 from composer.io.context import with_handler, run_graph
+from composer.diagnostics.timing import set_current_task_id
 from composer.ui.codegen_events import CodeGenEventHandler
 from composer.core.state import AIComposerInput, AIComposerExtra
 from composer.prover.agentic_analyzer import AgenticCexHandler
@@ -57,7 +57,16 @@
 from composer.workflow.summarization import SummaryGeneration
 
 
-_KB_NS = ("cvl",)
+# Codegen runs through ``run_graph`` rather than ``run_task``, so it has no
+# task_id of its own. Set one explicitly around the main loop so the harness
+# tape can address codegen LLM calls (and so timing attributes to a real task).
+CODEGEN_TASK_ID = "codegen"
+
+# create_resume_commentary runs *after* the main graph returns — a bare
+# structured-output call, not part of the agent loop. Its own task_id keeps it in
+# a dedicated harness lane so the agent can't consume its scripted response (and a
+# mismatch surfaces as a clean codegen-lane exhaustion rather than here).
+RESUME_COMMENTARY_TASK_ID = "resume-commentary"
 
 
 class _ExecutorOptions(WorkflowOptions, ModelOptionsBase, Protocol):
@@ -414,13 +423,15 @@ async def _run_codegen(
 
     try:
         async with with_handler(handler, CodeGenEventHandler(handler)):
-            final_state = await run_graph(workflow_exec, work_context, flow_input, config, description="Code generation")
+            with set_current_task_id(CODEGEN_TASK_ID):
+                final_state = await run_graph(workflow_exec, work_context, flow_input, config, description="Code generation")
 
         result = final_state.get("generated_code", None)
         if result is None:
             return WorkflowFailure()
 
-        res_commentary = await create_resume_commentary(final_state, llm=llm.builder_for())
+        with set_current_task_id(RESUME_COMMENTARY_TASK_ID):
+            res_commentary = await create_resume_commentary(final_state, llm=llm.builder_for())
         await audit_store.register_complete(
             thread_id, materializer.iterate(final_state), res_commentary.interface_path, res_commentary.commentary
         )
diff --git a/composer/workflow/meta.py b/composer/workflow/meta.py
index 514b9f13..7ade2969 100644
--- a/composer/workflow/meta.py
+++ b/composer/workflow/meta.py
@@ -21,7 +21,7 @@ class ResumeCommentary(BaseModel):
     interface_path: str = Field(description="The path of the interface file on the VFS")
 
 async def create_resume_commentary(state: AIComposerState, llm: BaseChatModel) -> ResumeCommentary:
-    llm = llm.copy(update={"thinking": None})
+    llm = llm.model_copy(update={"thinking": None})
     bound = llm.with_structured_output(ResumeCommentary)
     messages = state["messages"].copy()
 
diff --git a/pyproject.toml b/pyproject.toml
index b5374060..f999a88f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -76,7 +76,7 @@ cpu = ["torch"]
 cuda = ["torch"]
 
 [dependency-groups]
-test = ["pytest>=9.0", "pytest-asyncio>=1.3", "testcontainers>=4.0"]
+test = ["pytest>=9.0", "pytest-asyncio>=1.3", "testcontainers>=4.0", "pytest-xdist"]
 ci = ["pyright"]
 ragbuild = [
     # duplicated! no way to prevent this, says only language where this nonsense is required
diff --git a/test_scenarios/codegen_capped_vault/ICappedVault.sol b/test_scenarios/codegen_capped_vault/ICappedVault.sol
new file mode 100644
index 00000000..bfa90b4e
--- /dev/null
+++ b/test_scenarios/codegen_capped_vault/ICappedVault.sol
@@ -0,0 +1,21 @@
+// SPDX-License-Identifier: UNLICENSED
+pragma solidity ^0.8.29;
+
+/// Per-account deposit vault with a fixed cap. The implementation you generate
+/// must expose exactly this surface.
+interface ICappedVault {
+    /// Deposit `amount` into the caller's balance.
+    function deposit(uint256 amount) external;
+
+    /// Withdraw `amount` from the caller's balance.
+    function withdraw(uint256 amount) external;
+
+    /// The caller-visible balance of `account`.
+    function balance(address account) external view returns (uint256);
+
+    /// The running total of all deposited funds.
+    function totalDeposited() external view returns (uint256);
+
+    /// The per-account balance cap.
+    function CAP() external view returns (uint256);
+}
diff --git a/test_scenarios/codegen_capped_vault/system.md b/test_scenarios/codegen_capped_vault/system.md
new file mode 100644
index 00000000..c9a2e0bd
--- /dev/null
+++ b/test_scenarios/codegen_capped_vault/system.md
@@ -0,0 +1,21 @@
+# CappedVault
+
+A minimal on-chain vault. Each account may deposit funds into its own balance
+and withdraw them later. The vault tracks a running total of all funds held.
+
+## Behavior
+
+- `deposit(amount)` adds `amount` to the caller's balance and to the global
+  `totalDeposited` running total.
+- `withdraw(amount)` removes `amount` from the caller's balance (and from
+  `totalDeposited`), reverting if the caller's balance is insufficient.
+- `balance(account)`, `totalDeposited()`, and `CAP()` are read-only views.
+
+## Requirements
+
+1. **Deposits must always succeed.** A user calling `deposit` must never have
+   their deposit rejected; the vault must always accept incoming funds.
+
+2. **No account may ever exceed the cap.** Each account's balance must never
+   exceed `CAP` (1000). Any deposit that would push the caller's balance above
+   the cap must be rejected.
diff --git a/test_scenarios/codegen_capped_vault/vault.spec b/test_scenarios/codegen_capped_vault/vault.spec
new file mode 100644
index 00000000..c9a93b8a
--- /dev/null
+++ b/test_scenarios/codegen_capped_vault/vault.spec
@@ -0,0 +1,36 @@
+// CVL specification for CappedVault.
+//
+// Two rules. Both VIOLATED against a faithful first draft for *different*
+// reasons, which drives the two parallel CEX analyses in the codegen harness:
+//
+//   - depositIncreasesTotal — conceptually correct; fails only because the
+//     first-draft implementation forgets to update `totalDeposited`. Fixed by
+//     editing the Solidity.
+//   - depositRaisesBalance — over-strong as written (a zero-value deposit is a
+//     legal no-op, so `>` is wrong). Fixed spec-side via cex_remediation, which
+//     guards the assertion on `amount > 0`.
+
+methods {
+    function balance(address) external returns (uint256) envfree;
+    function totalDeposited() external returns (uint256) envfree;
+    function CAP() external returns (uint256) envfree;
+    function deposit(uint256) external;
+    function withdraw(uint256) external;
+}
+
+rule depositIncreasesTotal(uint256 amount) {
+    env e;
+    mathint before = totalDeposited();
+    deposit(e, amount);
+    assert to_mathint(totalDeposited()) == before + amount,
+        "a successful deposit must increase totalDeposited by exactly amount";
+}
+
+rule depositRaisesBalance(uint256 amount) {
+    env e;
+    address caller = e.msg.sender;
+    mathint before = balance(caller);
+    deposit(e, amount);
+    assert to_mathint(balance(caller)) > before,
+        "a successful deposit must raise the caller's balance";
+}
diff --git a/tests/conftest.py b/tests/conftest.py
index 1ff25f3b..47ae1819 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -18,6 +18,7 @@
 import numpy as np
 import psycopg
 import pytest
+from pytest import MonkeyPatch
 import pytest_asyncio
 from numpy import ndarray
 
@@ -26,7 +27,7 @@
 
 from psycopg.rows import dict_row
 from psycopg.connection_async import AsyncConnection
-from psycopg.sql import SQL, Identifier
+from psycopg.sql import SQL, Identifier, Literal
 from psycopg_pool.pool_async import AsyncConnectionPool as PGAsyncPool
 
 import composer.diagnostics.timing as timing_mod
@@ -136,6 +137,108 @@ def qna_factory() -> Callable[[], QnATransformer]:
 # Testcontainers: Postgres + indexed store
 # =========================================================================
 
+import pathlib
+import shutil
+
+@dataclass
+class ScenarioProvider:
+    _test_roots: pathlib.Path
+    _work_dir: pathlib.Path
+
+    def by_name(self, nm: str) -> pathlib.Path:
+        """Return a private copy of the scenario, made under this test's tmp dir.
+
+        The pipeline runs in-place — it materializes fixed-name spec/conf files
+        and dumps artifacts into the scenario directory — so tests sharing the
+        checked-in directory race under xdist and pollute the working tree.
+        """
+        src = self._test_roots / nm
+        assert src.is_dir(), f"{nm} is not a valid scenario name"
+        dst = self._work_dir / nm
+        shutil.copytree(
+            src,
+            dst,
+            ignore=shutil.ignore_patterns(".certora_internal", ".CertoraProverLiteReports", "emv-*"),
+        )
+        return dst
+
+
+@pytest.fixture
+def scenario_provider(tmp_path: pathlib.Path) -> ScenarioProvider:
+    scenario_dir = pathlib.Path(__file__).parent.parent / "test_scenarios"
+    return ScenarioProvider(scenario_dir, tmp_path)
+
+def _db_url(pg: "PostgresContainer", database: str) -> str:
+    return (
+        f"postgresql://{pg.username}:{pg.password}"
+        f"@{pg.get_container_host_ip()}:{pg.get_exposed_port(5432)}/{database}"
+    )
+
+_RAG_DB = "rag_db"
+# DBs that hold pgvector embeddings and need the extension (the store role's DB
+# + the RAG DB); checkpoint/memory are plain.
+_VECTOR_DBS = ("langgraph_store_db", _RAG_DB)
+
+
+# graphcore's Postgres memory backend doesn't self-create its schema; this mirrors
+# the memories_fs DDL in graphcore/tests/conftest.py (keep in sync if that moves).
+_MEMORIES_DDL = """
+CREATE TABLE IF NOT EXISTS memories_fs(
+    namespace TEXT NOT NULL,
+    entry_name TEXT NOT NULL,
+    full_path TEXT,
+    parent_path TEXT,
+    is_directory BOOL NOT NULL,
+    contents TEXT,
+    FOREIGN KEY(parent_path, namespace) REFERENCES memories_fs(full_path, namespace) ON DELETE CASCADE,
+    UNIQUE (namespace, full_path),
+    UNIQUE (namespace, parent_path, entry_name),
+    CHECK (parent_path is NOT NULL OR (full_path = '/memories' AND is_directory AND entry_name = 'memories')),
+    CHECK (parent_path is NULL OR (full_path = concat(parent_path, '/', entry_name))),
+    CHECK (contents IS NOT NULL != is_directory)
+);
+CREATE INDEX IF NOT EXISTS memories_namespace_path ON memories_fs(namespace, full_path text_pattern_ops);
+"""
+
+@dataclass
+class LanggraphDBSetup:
+    rag_db: str
+
+@pytest_asyncio.fixture(scope="session")
+async def langgraph_db() -> AsyncIterator[LanggraphDBSetup | None]:
+    if not _HAS_TESTCONTAINERS:
+        pytest.skip("No pgcontainers")
+    with PostgresContainer("pgvector/pgvector:pg16") as pg_container:
+        import composer.workflow.services as services
+        admin_url = pg_container.get_connection_url(driver=None)
+        with psycopg.connect(admin_url, autocommit=True) as admin:
+            for cfg in services._DATABASE_CONFIGS.values():
+                admin.execute(SQL("CREATE ROLE {} LOGIN PASSWORD {}").format(
+                    Identifier(cfg["user"]), Literal(cfg["password"])))
+                admin.execute(SQL("CREATE DATABASE {} OWNER {}").format(
+                    Identifier(cfg["database"]), Identifier(cfg["user"])))
+            admin.execute(SQL("CREATE DATABASE {}").format(Identifier(_RAG_DB)))
+        # pgvector must be installed by a superuser, so do it on the admin connection.
+        for db in _VECTOR_DBS:
+            with psycopg.connect(_db_url(pg_container, db), autocommit=True) as conn:
+                conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
+
+        # The memory backend doesn't self-create its schema (the checkpointer/store do,
+        # via .setup()), so create memories_fs as the memory role (its DB owner).
+        mem = services._DATABASE_CONFIGS["memory"]
+        mem_url = (
+            f"postgresql://{mem['user']}:{mem['password']}"
+            f"@{pg_container.get_container_host_ip()}:{pg_container.get_exposed_port(5432)}/{mem['database']}"
+        )
+        with psycopg.connect(mem_url, autocommit=True) as conn:
+            conn.execute(_MEMORIES_DDL)
+
+        with MonkeyPatch().context() as mp:
+            mp.setenv("CERTORA_AI_COMPOSER_PGHOST", pg_container.get_container_host_ip())
+            mp.setenv("CERTORA_AI_COMPOSER_PGPORT", str(pg_container.get_exposed_port(5432)))
+            yield LanggraphDBSetup(_db_url(pg_container, _RAG_DB))
+
+
 
 @pytest.fixture(scope="session")
 def pg_container() -> Iterator["PostgresContainer | None"]:
diff --git a/tests/test_autoprove_integration.py b/tests/test_autoprove_integration.py
index 65190777..6493ffaf 100644
--- a/tests/test_autoprove_integration.py
+++ b/tests/test_autoprove_integration.py
@@ -15,13 +15,10 @@
 import json
 from pathlib import Path
 from types import SimpleNamespace
-from typing import cast, TYPE_CHECKING
+from typing import cast
 
-import psycopg
 import pytest
-from psycopg.sql import SQL, Identifier, Literal
 
-import composer.workflow.services as services
 from composer.diagnostics.timing import RunSummary
 from composer.spec.source.autoprove_common import autoprove_executor, AutoProveArgs, AutoProvePhase
 from composer.spec.source.autosetup import SetupSuccess
@@ -29,19 +26,12 @@
 from composer.testing.ui_harness_autoprove_Counter import install_harness_tape
 from composer.pipeline.ptypes import CorePhases, SystemAnalysisSpec
 
-if TYPE_CHECKING:
-    from testcontainers.postgres import PostgresContainer
-
 
 from tests.conftest import needs_postgres, MockSentenceTransformer
 
 pytestmark = [pytest.mark.expensive, needs_postgres, pytest.mark.asyncio]
 
-_SCENARIO = Path(__file__).parent.parent / "test_scenarios" / "autoprove_counter"
-_RAG_DB = "rag_db"
-# DBs that hold pgvector embeddings and need the extension (the store role's DB
-# + the RAG DB); checkpoint/memory are plain.
-_VECTOR_DBS = ("langgraph_store_db", _RAG_DB)
+_SCENARIO_NAME = "autoprove_counter"
 
 # The config AutoSetup produced for Counter on a local run (its outputs aren't
 # checked in). DummyERC20Impl is dropped — Counter is standalone and the mock
@@ -62,60 +52,35 @@
 _SUMMARIES_REL = "specs/summaries/Counter_base_summaries.spec"
 
 
-async def _fake_autosetup_phase(*_args, **_kwargs) -> SetupSuccess:
-    """Stand in for the AutoSetup subprocess, which makes its LLM
-    calls that we, in autoprover land, aren't going to start taping.
-    It is also not an intersting unit of test for this workflow, so just use the
-    trivial, precomputed setups.
-    Writes out the (trivial, no-op) summaries spec the generated CVL imports
-    against, then returns the config AutoSetup would have produced for Counter."""
-    summaries = _SCENARIO / "certora" / _SUMMARIES_REL
-    summaries.parent.mkdir(parents=True, exist_ok=True)
-    summaries.write_text(
-        "// Auto-generated base summaries for Counter\n// No summaries needed for Counter\n"
-    )
-    return SetupSuccess(
-        prover_config=dict(_COUNTER_PROVER_CONFIG),
-        summaries_path=_SUMMARIES_REL,
-        user_types=[],
-    )
-
-# graphcore's Postgres memory backend doesn't self-create its schema; this mirrors
-# the memories_fs DDL in graphcore/tests/conftest.py (keep in sync if that moves).
-_MEMORIES_DDL = """
-CREATE TABLE IF NOT EXISTS memories_fs(
-    namespace TEXT NOT NULL,
-    entry_name TEXT NOT NULL,
-    full_path TEXT,
-    parent_path TEXT,
-    is_directory BOOL NOT NULL,
-    contents TEXT,
-    FOREIGN KEY(parent_path, namespace) REFERENCES memories_fs(full_path, namespace) ON DELETE CASCADE,
-    UNIQUE (namespace, full_path),
-    UNIQUE (namespace, parent_path, entry_name),
-    CHECK (parent_path is NOT NULL OR (full_path = '/memories' AND is_directory AND entry_name = 'memories')),
-    CHECK (parent_path is NULL OR (full_path = concat(parent_path, '/', entry_name))),
-    CHECK (contents IS NOT NULL != is_directory)
-);
-CREATE INDEX IF NOT EXISTS memories_namespace_path ON memories_fs(namespace, full_path text_pattern_ops);
-"""
-
+def _fake_autosetup_phase(scenario_path: Path):
 
-def _db_url(pg: "PostgresContainer", database: str) -> str:
-    return (
-        f"postgresql://{pg.username}:{pg.password}"
-        f"@{pg.get_container_host_ip()}:{pg.get_exposed_port(5432)}/{database}"
-    )
+    async def _fake_autosetup_phase_impl(*_args, **_kwargs) -> SetupSuccess:
+        """Stand in for the AutoSetup subprocess, which makes its LLM
+        calls that we, in autoprover land, aren't going to start taping.
+        It is also not an intersting unit of test for this workflow, so just use the
+        trivial, precomputed setups.
+        Writes out the (trivial, no-op) summaries spec the generated CVL imports
+        against, then returns the config AutoSetup would have produced for Counter."""
+        summaries = scenario_path / "certora" / _SUMMARIES_REL
+        summaries.parent.mkdir(parents=True, exist_ok=True)
+        summaries.write_text(
+            "// Auto-generated base summaries for Counter\n// No summaries needed for Counter\n"
+        )
+        return SetupSuccess(
+            prover_config=dict(_COUNTER_PROVER_CONFIG),
+            summaries_path=_SUMMARIES_REL,
+            user_types=[],
+        )
+    return _fake_autosetup_phase_impl
 
 
-def _make_args(rag_conn: str, system_doc: str | None = str(_SCENARIO / "system.md")) -> AutoProveArgs:
+def _make_args(rag_conn: str, scenario_dir: Path, system_doc: str | None) -> AutoProveArgs:
     """Hand-built ``AutoProveArgs`` (the CLI path builds this via argparse).
 
-    ``system_doc`` defaults to the scenario's ``system.md``; pass ``None`` to exercise
-    the auto-discovery path."""
+    Pass ``system_doc=None`` to exercise the Design Doc Discovery path."""
     return cast(AutoProveArgs, SimpleNamespace(
-        project_root=str(_SCENARIO),
-        main_contract=f"{_SCENARIO / "src/Counter.sol"}:Counter",
+        project_root=str(scenario_dir),
+        main_contract=f"{scenario_dir / "src/Counter.sol"}:Counter",
         system_doc=system_doc,
         max_concurrent=4,
         cache_ns=None,
@@ -137,62 +102,14 @@ def _make_args(rag_conn: str, system_doc: str | None = str(_SCENARIO / "system.m
         interleaved_thinking=False,
     ))
 
-@pytest.fixture(scope="session")
-def provisioned_rag_url(pg_container: "PostgresContainer | None") -> str:
-    """Stand up the fixed-name roles / databases / extensions / schema the autoprove
-    pipeline connects to, once per session; return the RAG DB url.
-
-    These database names *and* login roles are hardcoded in
-    ``services._DATABASE_CONFIGS`` — only host/port are overridable (redirected
-    per-test in ``_install_mocks``). The pipeline therefore always talks to these same
-    databases: there is nothing per-test to make unique, so they are created once,
-    unconditionally. No other fixture creates these fixed names, so there is nothing to
-    collide with (unlike ``get_test_database``'s per-invocation ``test_store_<uuid>``).
-
-    Per-test isolation comes from namespacing, not fresh databases: each run uses a
-    unique ``thread_id`` (checkpoint/store) and ``memory_ns``, with caching off, so the
-    two integration tests don't step on each other despite sharing these DBs."""
-    if pg_container is None:
-        pytest.skip("No pgcontainers")
-    # Roles + databases matching services._DATABASE_CONFIGS (a login role owning its
-    # own DB), so the real connection-string path works unpatched.
-    admin_url = pg_container.get_connection_url(driver=None)
-    with psycopg.connect(admin_url, autocommit=True) as admin:
-        for cfg in services._DATABASE_CONFIGS.values():
-            admin.execute(SQL("CREATE ROLE {} LOGIN PASSWORD {}").format(
-                Identifier(cfg["user"]), Literal(cfg["password"])))
-            admin.execute(SQL("CREATE DATABASE {} OWNER {}").format(
-                Identifier(cfg["database"]), Identifier(cfg["user"])))
-        admin.execute(SQL("CREATE DATABASE {}").format(Identifier(_RAG_DB)))
-    # pgvector must be installed by a superuser, so do it on the admin connection.
-    for db in _VECTOR_DBS:
-        with psycopg.connect(_db_url(pg_container, db), autocommit=True) as conn:
-            conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
-
-    # The memory backend doesn't self-create its schema (the checkpointer/store do,
-    # via .setup()), so create memories_fs as the memory role (its DB owner), not the
-    # superuser — the backend connects as that role and must own the table.
-    mem = services._DATABASE_CONFIGS["memory"]
-    mem_url = (
-        f"postgresql://{mem['user']}:{mem['password']}"
-        f"@{pg_container.get_container_host_ip()}:{pg_container.get_exposed_port(5432)}/{mem['database']}"
-    )
-    with psycopg.connect(mem_url, autocommit=True) as conn:
-        conn.execute(_MEMORIES_DDL)
-    return _db_url(pg_container, _RAG_DB)
-
-
-def _install_mocks(pg_container: "PostgresContainer", monkeypatch) -> None:
-    """Function-scoped connection redirection + LLM/AutoSetup mocks (undone per test by
-    ``monkeypatch``). The databases themselves are stood up once by the
-    ``provisioned_rag_url`` session fixture."""
-    # Only host/port need redirecting — the role creds already match the configs.
-    monkeypatch.setenv("CERTORA_AI_COMPOSER_PGHOST", pg_container.get_container_host_ip())
-    monkeypatch.setenv("CERTORA_AI_COMPOSER_PGPORT", str(pg_container.get_exposed_port(5432)))
 
+def _install_mocks(monkeypatch, scenario_dir: Path) -> None:
+    """LLM / AutoSetup / embedding mocks (undone per test by ``monkeypatch``). The
+    databases themselves — and the host/port connection redirection — are handled once
+    per session by the ``langgraph_db`` fixture."""
     # Mock only the LLM (Counter tape) + disable the agent-index cache.
     install_harness_tape(with_delay=False)
-    # autoprove_common imported `get_provider_for` by name, so install_harness_tape's
+    # pipeline.cli imported `get_provider_for` by name, so install_harness_tape's
     # patch of registry.get_provider_for doesn't reach that binding — rebind it here.
     import composer.llm.registry as registry
     monkeypatch.setattr(
@@ -208,7 +125,7 @@ def _install_mocks(pg_container: "PostgresContainer", monkeypatch) -> None:
     # canned Counter SetupSuccess. Patch the name the pipeline imported, not the
     # definition in harness.py.
     monkeypatch.setattr(
-        "composer.spec.source.pipeline.run_autosetup_phase", _fake_autosetup_phase
+        "composer.spec.source.pipeline.run_autosetup_phase", _fake_autosetup_phase(scenario_dir)
     )
     # The report phase is best-effort and absorbs failures (grouping degrades to a
     # fallback bucket; the outer guard logs-and-continues). Flip both into re-raise
@@ -218,38 +135,37 @@ def _install_mocks(pg_container: "PostgresContainer", monkeypatch) -> None:
     )
 
 
-def _read_job_info() -> dict:
+def _read_job_info(scenario_dir: Path) -> dict:
     """The always-written run manifest the entry point's ``finally`` dumps."""
-    return json.loads((_SCENARIO / "certora" / "ap_report" / "job_info.json").read_text())
+    return json.loads((scenario_dir / "certora" / "ap_report" / "job_info.json").read_text())
 
 
-async def test_autoprove_counter_runs_end_to_end(
-    pg_container: "PostgresContainer", provisioned_rag_url: str, monkeypatch
-):
-    assert _SCENARIO.is_dir(), _SCENARIO
-    _install_mocks(pg_container, monkeypatch)
+async def test_autoprove_counter_runs_end_to_end(scenario_provider, langgraph_db, monkeypatch):
+    scenario_dir = scenario_provider.by_name(_SCENARIO_NAME)
+    _install_mocks(monkeypatch, scenario_dir)
     monkeypatch.setenv("AUTOPROVER_USER_ID", "e2e-user")
     # Run the whole pipeline. Pass == it completes without raising.
     summary = RunSummary()
-    async with autoprove_executor(_make_args(provisioned_rag_url), summary) as run:
+    async with autoprove_executor(
+        _make_args(langgraph_db.rag_db, scenario_dir, str(scenario_dir / "system.md")),
+        summary,
+    ) as run:
         await run(AutoProveConsoleHandler().make_handler)
 
     # The finally dumped the run manifest to ap_report, carrying this run's identity
     # and its (non-empty, since the pipeline ran the prover) usage totals.
-    job_info = _read_job_info()
+    job_info = _read_job_info(scenario_dir)
     assert job_info["user_id"] == "e2e-user"
     assert job_info["run_id"] == summary.run_id
     assert "token_usage" in job_info and "prover_usage" in job_info
 
 
-async def test_autoprove_dumps_job_info_when_pipeline_crashes(
-    pg_container: "PostgresContainer", provisioned_rag_url: str, monkeypatch
-):
+async def test_autoprove_dumps_job_info_when_pipeline_crashes(scenario_provider, langgraph_db, monkeypatch):
     """The core guarantee: job_info.json is written even when the run crashes. Patch the
     pipeline to blow up after the executor is set up; the entry-point ``finally`` must
     still land the manifest (with this run's identity) in ap_report."""
-    assert _SCENARIO.is_dir(), _SCENARIO
-    _install_mocks(pg_container, monkeypatch)
+    scenario_dir = scenario_provider.by_name(_SCENARIO_NAME)
+    _install_mocks(monkeypatch, scenario_dir)
     monkeypatch.setenv("AUTOPROVER_USER_ID", "crash-user")
 
     async def _boom(*_args, **_kwargs):
@@ -285,22 +201,26 @@ async def prepare_system(
 
     summary = RunSummary()
     with pytest.raises(RuntimeError, match="pipeline exploded"):
-        async with autoprove_executor(_make_args(provisioned_rag_url), summary) as run:
+        async with autoprove_executor(
+            _make_args(langgraph_db.rag_db, scenario_dir, str(scenario_dir / "system.md")),
+            summary,
+        ) as run:
             await run(AutoProveConsoleHandler().make_handler)
 
-    job_info = _read_job_info()
+    job_info = _read_job_info(scenario_dir)
     assert job_info["user_id"] == "crash-user"
     assert job_info["run_id"] == summary.run_id
 
 
-async def test_autoprove_counter_no_doc_runs_end_to_end(
-    pg_container: "PostgresContainer", provisioned_rag_url: str, monkeypatch
-):
+async def test_autoprove_counter_no_doc_runs_end_to_end(scenario_provider, langgraph_db, monkeypatch):
     """Same pipeline, design doc OMITTED: the Design Doc Discovery phase runs the
     finder (its tape lane selects ``system.md``) and the run completes via the
     discovered doc."""
-    assert _SCENARIO.is_dir(), _SCENARIO
-    _install_mocks(pg_container, monkeypatch)
+    scenario_dir = scenario_provider.by_name(_SCENARIO_NAME)
+    _install_mocks(monkeypatch, scenario_dir)
     summary = RunSummary()
-    async with autoprove_executor(_make_args(provisioned_rag_url, system_doc=None), summary) as run:
+    async with autoprove_executor(
+        _make_args(langgraph_db.rag_db, scenario_dir, None),
+        summary,
+    ) as run:
         await run(AutoProveConsoleHandler().make_handler)
diff --git a/tests/test_codegen_integration.py b/tests/test_codegen_integration.py
new file mode 100644
index 00000000..65dc11fd
--- /dev/null
+++ b/tests/test_codegen_integration.py
@@ -0,0 +1,117 @@
+"""End-to-end integration test for the codegen pipeline.
+
+The LLM is mocked — the hand-authored CappedVault tape, installed via
+``install_harness_tape`` (which also disables the agent-index cache) — and the
+console HITL replies are scripted via ``install_response_tape``. Everything else
+runs for real: Postgres (checkpoint / store / memory) in a testcontainer, solc,
+the VFS + working-spec machinery, and the live Certora cloud prover.
+
+The scenario (``test_scenarios/codegen_capped_vault``) is built so the first
+draft fails BOTH spec rules for different reasons — an implementation bug
+(``depositIncreasesTotal``) and an over-strong assertion (``depositRaisesBalance``)
+— which drives the two parallel per-rule CEX analyses, the cross-rule aggregator,
+the spec-side ``cex_remediation`` sub-agent, the working-spec commit, and a
+contradictory-requirement relaxation. Prover verdicts are real, so given the
+deterministic tape + fixed spec/code the run is reasonably deterministic.
+Pass/fail is simply: the workflow reaches ``WorkflowSuccess``.
+
+Marked ``expensive`` (live cloud prover + containers + the embedding model load)
+and skipped without testcontainers. Run with ``-m expensive``.
+"""
+from pathlib import Path
+from types import SimpleNamespace
+from typing import cast
+
+import pytest
+
+import composer.llm.registry as registry
+from composer.input.parsing import upload_input
+from composer.input.types import CommandLineArgs
+from composer.workflow.executor import execute_ai_composer_workflow
+from composer.workflow.types import WorkflowSuccess
+from composer.ui.console import ConsoleHandler
+from composer.ui.tool_display import tool_context
+from composer.testing.ui_harness_codegen_capped_vault import (
+    install_harness_tape,
+    install_response_tape,
+)
+
+from tests.conftest import needs_postgres, MockSentenceTransformer
+
+pytestmark = [pytest.mark.expensive, needs_postgres, pytest.mark.asyncio]
+
+_SCENARIO_NAME = "codegen_capped_vault"
+
+
+def _make_args(rag_conn: str, scenario_dir: Path) -> CommandLineArgs:
+    """Hand-built args (the CLI path builds this via argparse). The three input
+    documents are the scenario's spec / interface / system doc."""
+    return cast(CommandLineArgs, SimpleNamespace(
+        spec_file=str(scenario_dir / "vault.spec"),
+        interface_file=str(scenario_dir / "ICappedVault.sol"),
+        system_doc=str(scenario_dir / "system.md"),
+        rag_db=rag_conn,
+        thread_id=None,
+        checkpoint_id=None,
+        recursion_limit=100,
+        prover_capture_output=True,
+        prover_keep_folders=False,
+        local_prover=False,
+        debug_prompt_override=None,
+        skip_reqs=False,
+        # Model-config fields: only read through ``get_provider_for``, which the
+        # tape patches to ignore them, so the values are inert — present to
+        # satisfy the args surface.
+        model="fake-model",
+        tokens=128_000,
+        thinking_tokens=2048,
+        memory_tool=False,
+        interleaved_thinking=False,
+    ))
+
+
+def _install_mocks(monkeypatch) -> None:
+    """LLM / HITL / embedding mocks. The databases themselves — and the host/port
+    connection redirection — are handled once per session by ``langgraph_db``."""
+    # The CappedVault tape (patches registry.get_provider_for + uploader_for and
+    # disables the agent-index cache) plus the scripted console HITL replies.
+    install_harness_tape(with_delay=False)
+    install_response_tape()
+    # Swap the real sentence-transformer for the deterministic mock: no model
+    # download (and no "Sentence transformers not available" throw), and nothing
+    # in this run depends on real embeddings (index cache disabled by the tape,
+    # RAG DB empty). Each module imported ``get_model`` by name, so patch every
+    # binding the run reaches: the executor (aliased ``get_rag_model``) and the
+    # requirements extractor.
+    monkeypatch.setattr(
+        "composer.workflow.executor.get_rag_model", MockSentenceTransformer
+    )
+    monkeypatch.setattr(
+        "composer.natreq.extractor.get_model", MockSentenceTransformer
+    )
+
+
+async def test_codegen_capped_vault_runs_end_to_end(scenario_provider, langgraph_db, monkeypatch):
+    scenario_dir = scenario_provider.by_name(_SCENARIO_NAME)
+    _install_mocks(monkeypatch)
+
+    args = _make_args(langgraph_db.rag_db, scenario_dir)
+    # Go through the registry module (not a name imported at module load) so these
+    # pick up the tape's patched ``get_provider_for`` / ``uploader_for``.
+    llm = registry.get_provider_for(options=args)
+    input_data = await upload_input(registry.uploader_for(llm.provider), args)
+
+    # Run the whole workflow. Pass == it reaches WorkflowSuccess: the impl bug was
+    # fixed, the spec over-constraint remediated + committed, and the contradictory
+    # requirement relaxed, so the final master prover + requirements validations
+    # both stamp and check_completion lets the author deliver.
+    handler = ConsoleHandler(capture_prover_output=True)
+    with tool_context():
+        result = await execute_ai_composer_workflow(
+            handler=handler,
+            llm=llm,
+            input=input_data,
+            workflow_options=args,
+        )
+
+    assert isinstance(result, WorkflowSuccess)
diff --git a/uv.lock b/uv.lock
index c312c76b..7f31e03f 100644
--- a/uv.lock
+++ b/uv.lock
@@ -140,6 +140,7 @@ ragbuild = [
 test = [
     { name = "pytest" },
     { name = "pytest-asyncio" },
+    { name = "pytest-xdist" },
     { name = "testcontainers" },
 ]
 
@@ -201,6 +202,7 @@ ragbuild = [
 test = [
     { name = "pytest", specifier = ">=9.0" },
     { name = "pytest-asyncio", specifier = ">=1.3" },
+    { name = "pytest-xdist" },
     { name = "testcontainers", specifier = ">=4.0" },
 ]
 
@@ -1278,6 +1280,15 @@ wheels = [
     { url = "https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.8.0/en_core_web_sm-3.8.0-py3-none-any.whl", hash = "sha256:1932429db727d4bff3deed6b34cfc05df17794f4a52eeb26cf8928f7c1a0fb85" },
 ]
 
+[[package]]
+name = "execnet"
+version = "2.1.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/bf/89/780e11f9588d9e7128a3f87788354c7946a9cbb1401ad38a48c4db9a4f07/execnet-2.1.2.tar.gz", hash = "sha256:63d83bfdd9a23e35b9c6a3261412324f964c2ec8dcd8d3c6916ee9373e0befcd", size = 166622, upload-time = "2025-11-12T09:56:37.75Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ab/84/02fc1827e8cdded4aa65baef11296a9bbe595c474f0d6d758af082d849fd/execnet-2.1.2-py3-none-any.whl", hash = "sha256:67fba928dd5a544b783f6056f449e5e3931a5c378b128bc18501f7ea79e296ec", size = 40708, upload-time = "2025-11-12T09:56:36.333Z" },
+]
+
 [[package]]
 name = "filelock"
 version = "3.25.2"
@@ -3875,6 +3886,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
 ]
 
+[[package]]
+name = "pytest-xdist"
+version = "3.8.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "execnet" },
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/78/b4/439b179d1ff526791eb921115fca8e44e596a13efeda518b9d845a619450/pytest_xdist-3.8.0.tar.gz", hash = "sha256:7e578125ec9bc6050861aa93f2d59f1d8d085595d6551c2c90b6f4fad8d3a9f1", size = 88069, upload-time = "2025-07-01T13:30:59.346Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ca/31/d4e37e9e550c2b92a9cbc2e4d0b7420a27224968580b5a447f420847c975/pytest_xdist-3.8.0-py3-none-any.whl", hash = "sha256:202ca578cfeb7370784a8c33d6d05bc6e13b4f25b5053c30a152269fd10f0b88", size = 46396, upload-time = "2025-07-01T13:30:56.632Z" },
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"