test(portfolio): gate the wave handler's catch width and the failed-project ledger (v/t/s)

Three items on one seam — what a FAILED project does to the wave loop — plus the
snapshot copy they sit next to.

(v) The catch is BaseException, not Exception, and that width was ungated. The
existing collect-and-continue test raises RuntimeError, so it stays green when
the handler is narrowed: measured, the whole of test_portfolio_concurrent_
loadbearing.py (13 tests) passes under the narrowing. asyncio.CancelledError is
the one realistic vector that separates the two — probed first, gather(
return_exceptions=True) COLLECTS it, while KeyboardInterrupt propagates
regardless and could never be helped by a wider catch. Narrowed, a cancelled
member is cast into runs as a fake RunResult and the pass dies in _aggregate,
pointing away from its cause. RED measured.

(t) sum_token_usage excludes a failed project's spend, and that is the honest
answer, not a bug: a run that died before producing a stamp has no provenance,
and inventing one is the fabrication RunFailure exists to avoid. What needed
gating is that those tokens still reach the ledger the global cap is enforced
against — otherwise a repeatedly-failing project burns budget while the meter
reads clean. Pins meter.spent as the pass's real cost, sum_token_usage as the
completed-run subtotal, and their difference as exactly the failed spend. RED
measured against the likely "fix" (sourcing sum_token_usage from the meter),
which is wrong because a seeded meter also carries EARLIER passes' spend; 21
existing budget/portfolio tests stay green under it.

(s) _wave_snapshot uses dataclasses.replace, so a field added later is carried
without touching the function. Not cosmetic: measured, dropping retriever by
hand-enumerating left all 585 tests green — the Step-2 coverage its docstring
credited no longer existed, so the S3.1 retriever seam could be downgraded
mid-pass in silence. Now gated by a property test derived from
dataclasses.fields (not a field count, the shape rejected earlier). The explicit
verdicts copy is retained and separately gated: replace(store) alone shares the
caller's list and takes the byte-identical determinism test RED.

strict=True on the zip is documented as deliberately untested — measured green
when dropped, since gather is built from exactly snapshots, so a test could only
go red by manufacturing a mismatch and would exercise zip rather than this pass.

The new double is registered in the S2.5 consolidation guard's delegating-
overrides list rather than the guard being weakened; it already delegates via
super()._inner_get_response, which test_delegating_overrides_call_super now
enforces on it.

583 -> 586 tests.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MbgTCEZma764i1rHTrzceU

src/portfolio_optimiser/run.py

@@ -622,13 +622,23 @@ def _wave_snapshot(store: VerdictStore) -> VerdictStore:
     dropped it would silently downgrade a caller-owned store's semantic retrieval to the
     structural default mid-pass.
 
-    **Honesty boundary: this copies exactly two fields because ``VerdictStore`` HAS exactly two.**
-    A third field added later would be silently dropped here — the same defect class as the
-    ``retriever`` omission this function was first written with, which the Step-2 contract test
-    caught. It is left as a documented hazard rather than a guard: an assertion on the field count
-    would go red on every benign addition to ``VerdictStore``, which trains people to edit the
-    guard rather than think about the snapshot — a worse outcome than the line you are reading."""
-    return VerdictStore(verdicts=list(store.verdicts), retriever=store.retriever)
+    **Field-complete by construction.** ``dataclasses.replace`` carries over every field
+    ``VerdictStore`` declares and overrides only ``verdicts``, so a field added later is copied
+    without this function being touched. The hand-enumerated version this replaced could silently
+    drop one — the exact defect it was first written with (the ``retriever`` omission), which its
+    own docstring then recorded as a standing hazard. Deriving the copy from the dataclass removes
+    the hazard instead of documenting it, and does so without the field-count assertion that idea
+    was rejected for: there is nothing left to keep in sync.
+
+    That docstring credited a Step-2 contract test with catching the omission. MEASURED while this
+    change was made: no such coverage remained — reinstating the hand-enumerated form left the
+    whole suite green, so the S3.1 retriever seam could be downgraded mid-pass in silence. The gate
+    is now ``test_wave_snapshot_carries_every_field_except_the_copied_verdicts``.
+
+    ``verdicts`` is still listed explicitly, and must be: ``replace`` copies field REFERENCES, so
+    omitting it would hand back a store sharing the caller's list — the very race this snapshot
+    exists to remove, reintroduced by the call that looks tidiest."""
+    return replace(store, verdicts=list(store.verdicts))
 
 
 def _merge_wave(store: VerdictStore, wave: Sequence[tuple[str, VerdictStore]]) -> None:
@@ -921,7 +931,20 @@ async def run_portfolio(
         # ``gather`` resolves in ARGUMENT order, not completion order, and waves follow
         # ``project_ids`` — so ``runs`` stays in caller order however the schedule interleaved, and
         # position is a sound key for pairing each result back to the pid that produced it.
+        #
+        # ``strict=True`` is FUTURE-PROOFING and is deliberately untested — measured, not assumed:
+        # dropping it leaves the whole suite green, because ``gather`` is constructed from exactly
+        # ``snapshots``, so the two lengths cannot diverge today. A test could only go red by
+        # manufacturing a mismatch, which would exercise ``zip`` rather than this pass. It earns its
+        # place against a later edit that filters or extends one sequence without the other — then a
+        # silent truncation would mis-attribute every result after the gap, and this fails instead.
         for (pid, _snapshot), outcome in zip(snapshots, wave_results, strict=True):
+            # ``BaseException``, NOT ``Exception``, and the width is load-bearing: a member raising
+            # ``asyncio.CancelledError`` (a BaseException since 3.8) is COLLECTED by
+            # ``return_exceptions=True`` and must land in ``failures``. Narrowed to ``Exception`` it
+            # would fall to the ``cast`` below and put a live exception object into ``runs``, which
+            # dies later in ``_aggregate`` — a crash pointing away from its cause. Gated by
+            # ``tests/test_portfolio_failure_accounting_loadbearing.py``.
             if isinstance(outcome, BaseException):
                 failures.append(
                     RunFailure(

tests/test_portfolio_concurrent_loadbearing.py

@@ -27,6 +27,7 @@ from __future__ import annotations
 
 import ast
 import asyncio
+import dataclasses
 from collections.abc import Mapping, Sequence
 from pathlib import Path
 from typing import Any
@@ -43,10 +44,17 @@ from test_portfolio_learning_loadbearing import (
 )
 
 from portfolio_optimiser import ledger as ledger_mod
+from portfolio_optimiser import semretrieval
 from portfolio_optimiser import verdicts as verdicts_mod
 from portfolio_optimiser.contracts import GoalConfig, GoalContract
 from portfolio_optimiser.ledger import SavingsLedger
-from portfolio_optimiser.run import PortfolioResult, RunFailure, _waves, run_portfolio
+from portfolio_optimiser.run import (
+    PortfolioResult,
+    RunFailure,
+    _wave_snapshot,
+    _waves,
+    run_portfolio,
+)
 from portfolio_optimiser.verdicts import VerdictStore
 
 # The shipped 3-project fixture, in submission order (mirrors tests/test_portfolio.py:57).
@@ -223,6 +231,43 @@ def _aggregate_fields(result: PortfolioResult) -> tuple[Any, ...]:
     )
 
 
+def test_wave_snapshot_carries_every_field_except_the_copied_verdicts() -> None:
+    """The snapshot copies ``verdicts`` and carries EVERY other ``VerdictStore`` field over.
+
+    **Detach point: hand-enumerate the fields (``VerdictStore(verdicts=list(store.verdicts))``)
+    -> this test goes RED.** MEASURED, and the measurement is why this test exists: before it was
+    written, dropping ``retriever`` in exactly that way left the whole suite GREEN (585 passed). No
+    test set ``retriever`` at all, so the S3.1 opt-in seam could be silently downgraded to the
+    structural default mid-pass — a caller-owned semantic retriever would stop being used from the
+    first wave onward, and nothing would say so. ``_wave_snapshot``'s docstring had claimed this
+    was caught by a Step-2 contract test; that coverage no longer existed.
+
+    This asserts the PROPERTY (every declared field survives), not a field count — the shape that
+    was rejected for going red on every benign addition to ``VerdictStore``. Deriving the expected
+    set from ``dataclasses.fields`` means a field added later is covered without editing this test,
+    and ``replace`` in the implementation means it is carried without editing the snapshot either.
+    """
+    # A real ``Retriever``, not a bare sentinel: the field is typed, and the assertion below is
+    # about identity surviving the copy, which a genuine instance demonstrates without pretence.
+    marker = semretrieval.StructuralRetriever(verdicts_mod.similarity)
+    store = VerdictStore(verdicts=[], retriever=marker)
+
+    snapshot = _wave_snapshot(store)
+
+    carried = [f.name for f in dataclasses.fields(VerdictStore) if f.name != "verdicts"]
+    assert carried, "VerdictStore declares no field beyond ``verdicts`` — this test guards nothing"
+    for name in carried:
+        assert getattr(snapshot, name) is getattr(store, name), (
+            f"``{name}`` was not carried into the wave snapshot — a project in the wave would run "
+            f"against a store whose {name} silently differs from the one the caller configured"
+        )
+
+    # The half that must NOT be carried by reference: the verdict list is copied, so a project
+    # appending to its snapshot cannot touch the shared store. Guarded here too, because the
+    # tidiest wrong fix for the above (``replace(store)``) breaks exactly this.
+    assert snapshot.verdicts is not store.verdicts
+
+
 async def test_concurrent_pass_is_byte_identical_to_sequential() -> None:
     """S3.3's core contract: ``concurrency=3`` produces the SAME store-verdict id SEQUENCE as
     ``concurrency=1``, under a completion order that genuinely differs from submission order.

tests/test_portfolio_failure_accounting_loadbearing.py

@@ -0,0 +1,242 @@
+"""S3.3 load-bearing — what a FAILED project does to the wave handler and to the token ledger.
+
+Two properties on the same seam (``run.py`` wave-result handling), sharing one fixture because they
+share one code path: a project that raises mid-wave.
+
+1. **The wave handler catches ``BaseException``, not ``Exception``.** The existing collect-and-
+   continue test (``test_one_project_failure_does_not_cancel_its_siblings``) raises ``RuntimeError``,
+   which is an ``Exception`` — so it stays GREEN if the handler is narrowed, and cannot gate the
+   width of the catch. ``asyncio.CancelledError`` derives from ``BaseException`` (3.8+) and is the
+   one realistic vector that separates the two.
+
+   MEASURED, not assumed, before this test was written: ``asyncio.gather(return_exceptions=True)``
+   COLLECTS a ``CancelledError`` raised by a member coroutine into the result list, so the
+   ``BaseException`` branch is genuinely reachable. ``KeyboardInterrupt`` is NOT a usable vector —
+   asyncio special-cases it and propagates out of ``gather`` regardless of ``return_exceptions``,
+   so widening the catch could never help there. That measurement is why this file tests exactly
+   one BaseException subclass and does not pretend to cover the rest.
+
+2. **A failed project's tokens still count against the GLOBAL cap.** ``PortfolioResult.
+   sum_token_usage`` sums ``provenance.token_usage`` over ``runs``, and a failed project has no
+   ``RunResult`` — so the aggregate necessarily UNDER-reports what the pass actually spent. That is
+   the honest answer rather than a bug to be patched: a run that died before producing a stamp has
+   no provenance, and inventing one would fabricate exactly the thing this repo's provenance rules
+   exist to prevent (the same reasoning that made ``RunFailure`` a distinct type).
+
+   What must NOT be true is that those tokens vanish from the ledger the cap is enforced against.
+   If they did, a project that fails repeatedly could burn unbounded budget while the meter read
+   clean. This pins the split: ``meter.spent`` is the pass's real cost, ``sum_token_usage`` is the
+   completed-run subtotal, and the difference is exactly the failed project's spend.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Mapping, Sequence
+
+from portfolio_optimiser.budget import PortfolioBudget, PortfolioMeter
+from portfolio_optimiser.run import RunFailure, run_portfolio
+from portfolio_optimiser.simulation import ScriptedChatClient
+
+_PORTFOLIO_IDS = ["FV42-GSV-E1", "RV13-RAS-TP", "BRU-LAKS-REHAB"]
+
+# The MIDDLE project. Middle is deliberate (mirroring the Step-4 test): a failure at either end can
+# be dropped by a truncation bug and still leave the survivors in the right relative order.
+_FAILING_PID = "RV13-RAS-TP"
+
+_DEFAULT_REPLY = (
+    '{"measure":"Reduce scope","affected_items":'
+    '[{"code":"01.1","quantity":1,"unit_cost":100000}],"claimed_saving_nok":20000}'
+)
+
+REPLIES = {
+    "FV42-GSV-E1": (
+        '{"measure":"Reduce scope","affected_items":['
+        '{"code":"05.2","quantity":4300,"unit_cost":215},'
+        '{"code":"03.1","quantity":1800,"unit_cost":310}],"claimed_saving_nok":200000}'
+    ),
+    "RV13-RAS-TP": (
+        '{"measure":"Material substitution","affected_items":['
+        '{"code":"88.2","quantity":180,"unit_cost":4200}],"claimed_saving_nok":130000}'
+    ),
+    "BRU-LAKS-REHAB": (
+        '{"measure":"Reduce scope","affected_items":['
+        '{"code":"05.2","quantity":4300,"unit_cost":215},'
+        '{"code":"07.4","quantity":2400,"unit_cost":690}],"claimed_saving_nok":210000}'
+    ),
+}
+
+# Measured in tests/test_portfolio_budget_loadbearing.py: 4 chat calls x ``tokens`` per reply, so a
+# completed run costs a flat 32 tokens at tokens=8.
+_TOKENS_PER_REPLY = 8
+_PER_RUN_SPEND = 32
+_SURVIVOR_COUNT = 2
+
+
+class _FailAfterNCallsClient(ScriptedChatClient):
+    """Project-aware scripted client that raises a chosen exception for ONE project, after that
+    project has already completed ``fail_after`` calls.
+
+    ``fail_after`` is what makes property 2 measurable: at 0 the project dies having spent nothing,
+    which would make the ledger assertion trivially equal. The counter is SHARED across the clients
+    the factory produces, because ``client_factory`` is called per ROLE and a run uses several — a
+    per-instance counter would restart on every role and never reach the threshold.
+
+    The failure is raised from inside a coroutine, not synchronously: ``_inner_get_response`` is a
+    sync method RETURNING an awaitable, so raising synchronously would blow up during ``gather``'s
+    argument construction — before any concurrency exists — and would test a different thing than a
+    mid-flight failure.
+    """
+
+    def __init__(
+        self,
+        replies: dict[str, str],
+        *,
+        failing_pid: str,
+        error: BaseException,
+        counter: dict[str, int],
+        fail_after: int = 0,
+    ) -> None:
+        table = dict(replies)
+
+        def _select(blob: str, _role: str) -> str:
+            return next((r for pid, r in table.items() if pid in blob), _DEFAULT_REPLY)
+
+        super().__init__(
+            reply_selector=_select,
+            default_reply=_DEFAULT_REPLY,
+            tokens_per_reply=_TOKENS_PER_REPLY,
+        )
+        self._failing_pid = failing_pid
+        self._error = error
+        self._counter = counter
+        self._fail_after = fail_after
+
+    def _inner_get_response(
+        self,
+        *,
+        messages: Sequence[Any],
+        options: Mapping[str, Any],
+        stream: bool = False,
+        **kwargs: Any,
+    ) -> Any:
+        blob = " ".join(getattr(m, "text", "") or "" for m in messages)
+        if not stream and self._failing_pid in blob:
+            seen = self._counter.get("n", 0)
+            if seen >= self._fail_after:
+                error = self._error
+
+                async def _boom() -> Any:
+                    raise error
+
+                return _boom()
+            self._counter["n"] = seen + 1
+        return super()._inner_get_response(
+            messages=messages, options=options, stream=stream, **kwargs
+        )
+
+
+def _factory(error: BaseException, *, fail_after: int = 0) -> Any:
+    counter: dict[str, int] = {}
+
+    def factory(_role: str) -> Any:
+        return _FailAfterNCallsClient(
+            REPLIES,
+            failing_pid=_FAILING_PID,
+            error=error,
+            counter=counter,
+            fail_after=fail_after,
+        )
+
+    return factory
+
+
+async def test_a_cancelled_project_is_collected_not_propagated(fresh_store) -> None:
+    """RED 1: a wave member raising ``asyncio.CancelledError`` is recorded as a ``RunFailure``
+    while its siblings complete — the pass does not raise and does not lose their results.
+
+    **Detach point: narrow ``run.py``'s ``isinstance(outcome, BaseException)`` to ``Exception``
+    -> this test goes RED (measured).** ``CancelledError`` then misses the failure branch and falls
+    through to ``cast(RunResult, outcome)``, which puts a live exception object into ``runs``; the
+    pass dies in ``_aggregate`` reaching for ``.provenance`` on it. Note the failure mode is a
+    crash in a DIFFERENT function than the defect — collect-and-continue is negated for cancelled
+    members, and the traceback points away from the cause. That is precisely why the width of this
+    catch is worth pinning rather than leaving to the ``RuntimeError`` test.
+
+    Why ``CancelledError`` specifically is documented in this module's docstring: it is the only
+    ``BaseException`` subclass ``gather(return_exceptions=True)`` actually hands back.
+    """
+    result = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        store=fresh_store,
+        client_factory=_factory(asyncio.CancelledError("synthetic cancellation")),
+    )
+
+    # Reaching this line at all is half the contract: the pass RETURNED rather than propagating.
+    assert len(result.failures) == 1, (
+        f"expected exactly one failure slot, got {result.failures} — a CancelledError that is not "
+        "collected either propagates out of run_portfolio or is cast into runs as a fake RunResult"
+    )
+    failure = result.failures[0]
+    assert isinstance(failure, RunFailure)
+    assert failure.project_id == _FAILING_PID
+    assert failure.error_type == "CancelledError"
+
+    # The siblings survived and are real RunResults — the assertion that catches an exception
+    # object smuggled into ``runs`` by a narrowed catch.
+    assert len(result.runs) == _SURVIVOR_COUNT
+    assert [r.verdict.proposal_features.claimed_saving_nok for r in result.runs] == [
+        200_000.0,
+        210_000.0,
+    ]
+    assert result.validated_count + result.rejected_count == _SURVIVOR_COUNT
+
+
+async def test_a_failed_projects_tokens_still_count_against_the_global_cap(fresh_store) -> None:
+    """RED 2: tokens spent by a project that later failed remain in the ledger the global cap is
+    enforced against, even though they are absent from ``sum_token_usage``.
+
+    The split is the point, and both halves are asserted:
+
+    - ``meter.spent`` — the pass's REAL cost — includes the failed project's partial spend. If it
+      did not, a project failing on every attempt would burn budget invisibly and the S3.4 cap
+      would not bound the pass at all.
+    - ``sum_token_usage`` — the completed-run subtotal — excludes it, because a run that never
+      produced provenance has no honest token figure to contribute. Pinned so that a later
+      "fix" cannot quietly fabricate one to make the two numbers agree.
+
+    The difference between them is asserted to be EXACTLY the failed project's spend, so this stays
+    a statement about accounting rather than a loose inequality that would hold for many wrong
+    reasons.
+    """
+    # Two calls complete for the failing project before it dies, so its partial spend is non-zero
+    # and known: without that, both sides of the ledger assertion would be equal by accident.
+    fail_after = 2
+    partial_spend = fail_after * _TOKENS_PER_REPLY
+
+    meter = PortfolioMeter(PortfolioBudget(max_total_tokens=10_000, max_tokens_per_run=1_000))
+    result = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        store=fresh_store,
+        client_factory=_factory(RuntimeError("synthetic backend failure"), fail_after=fail_after),
+        portfolio_meter=meter,
+    )
+
+    assert len(result.runs) == _SURVIVOR_COUNT
+    assert len(result.failures) == 1
+    assert result.failures[0].project_id == _FAILING_PID
+
+    completed_spend = _SURVIVOR_COUNT * _PER_RUN_SPEND
+    assert result.sum_token_usage == completed_spend, (
+        "sum_token_usage must total the COMPLETED runs' provenance and nothing else; a failed "
+        "project has no provenance to contribute and must not be given a fabricated one"
+    )
+    assert meter.spent == completed_spend + partial_spend, (
+        f"the global ledger recorded {meter.spent}, expected {completed_spend + partial_spend} — "
+        f"the failed project's {partial_spend} tokens were really spent, so a cap that does not "
+        "see them does not bound the pass"
+    )
+    # The under-report is real, bounded, and exactly the failed project's spend.
+    assert meter.spent - result.sum_token_usage == partial_spend

tests/test_scripted_client_consolidation.py

@@ -31,6 +31,10 @@ _DELEGATING_OVERRIDES = [
     # S3.3 ordering probe: yields to the event loop N times, then delegates. It cannot live in the
     # reply-selector seam, which the canonical calls synchronously and so can never await.
     "tests/test_portfolio_concurrent_loadbearing.py",
+    # S3.3 failure-accounting probe: RAISES for one project (after N completed calls), otherwise
+    # delegates. Like the ordering probe it cannot live in the reply-selector seam — that seam
+    # returns a reply string, and this double's whole subject is the absence of one.
+    "tests/test_portfolio_failure_accounting_loadbearing.py",
 ]
 
 # Doubles in a DIFFERENT lineage (``spikes._harness.FakeChatClient``). There is no canonical