feat(s33): wave executor with per-project snapshot and deterministic merge barrier

Replaces run_portfolio's sequential loop with a wave loop over _waves(ids, k):
each wave takes a per-project snapshot of the shared store, runs the wave under
one asyncio.gather in a single event loop, then crosses a merge barrier that
folds each project's NEW verdicts back in wave-submission order. Step 2's
contract goes GREEN; runs stays in project_ids order because gather resolves in
argument order, not completion order.

TWO PLAN CORRECTIONS, both found by the RED-first test rather than by reading:

1. The plan specified sorting the merged verdicts on `project_id`. Measured, that
   produces a deterministic order which is the WRONG one: lexicographic gives
   BRU/FV42/RV13 while the sequential pass gives FV42/RV13/BRU. It satisfies
   "deterministic" while breaking "identical to concurrency=1" — and the second is
   the actual contract. The merge preserves submission order instead.

2. The plan named the barrier's sort as the load-bearing seam. It is not — with
   per-project snapshots the wave list is never reordered by completion, so a
   sorted() there would re-sort an already-ordered list and read as a guard while
   guarding nothing. The SNAPSHOT is the half that carries the load. Rather than
   ship a decorative sort, both halves were measured (scratchpad-restore, never
   git checkout):

     detach _wave_snapshot  -> RED (store lands in completion order)
     detach merge ordering  -> RED (reversed wave order diverges)

   Both restored byte-identical (sha 42b01d46).

The snapshot carries `retriever` across deliberately: dropping it would silently
downgrade a caller-owned store's S3.1 semantic-retrieval opt-in mid-pass.

Also strengthens the scripted-client consolidation guard, which the probe broke by
being a legitimate third _inner_get_response def-site. It pinned a literal count
of 2 — the wrong shape: it failed on any new legitimate subclass while still
passing if someone pasted a duplicated body into an already-listed file. It now
pins the property (registered sites, scripted-lineage overrides must delegate via
super(), foreign-lineage doubles must genuinely be foreign). Verified load-bearing:
removing both delegation sites turns it RED.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQapztREtC2mkr5oU811pr

src/portfolio_optimiser/run.py

@@ -25,6 +25,7 @@ durable learned verdict captured out-of-band in the VerdictStore (D7-portable).
 
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Callable, Sequence
 from dataclasses import dataclass, replace
 from decimal import ROUND_HALF_UP, Decimal
@@ -556,6 +557,54 @@ def _goal_limit_if_reached(goal: GoalContract, observed_ore: int, baseline_ore:
     return None
 
 
+def _wave_snapshot(store: VerdictStore) -> VerdictStore:
+    """A per-project copy of the shared store's CURRENT verdicts (D-D wave model).
+
+    Two coroutines appending to one list would interleave by completion order, which no barrier
+    could then undo. Giving every project in a wave its own copy removes the race at its source
+    rather than serializing it away with a lock — a lock would order the appends by whoever won,
+    which is exactly the nondeterminism being eliminated.
+
+    ``retriever`` is carried across deliberately: it is the S3.1 opt-in seam, and a snapshot that
+    dropped it would silently downgrade a caller-owned store's semantic retrieval to the
+    structural default mid-pass."""
+    return VerdictStore(verdicts=list(store.verdicts), retriever=store.retriever)
+
+
+def _merge_wave(store: VerdictStore, wave: Sequence[tuple[str, VerdictStore]]) -> None:
+    """The deterministic merge barrier — the seam S3.3 rests on.
+
+    Each project ran against its own snapshot, so its NEW verdicts are the ones absent from the
+    wave-start store. They are merged back in **wave-submission order** — the order the caller
+    listed the projects in — so the shared store's contents depend on the portfolio's membership
+    and never on which project's model round-trips happened to finish first.
+
+    **Submission order, NOT lexicographic project_id.** The plan specified a sort on ``project_id``;
+    the Step-2 contract test measured that it produces a deterministic order which is nevertheless
+    the WRONG one. On the shipped fixture, lexicographic order is BRU/FV42/RV13 while the sequential
+    pass yields FV42/RV13/BRU, so a ``project_id`` sort satisfies "deterministic" while breaking
+    "identical to ``concurrency=1``" — and the second is the actual contract. ``wave`` arrives in
+    submission order, so preserving it is the fix.
+
+    **Detach point: this function's ordering discipline is only half the seam — see
+    ``_wave_snapshot``, which is the half that carries the load.** Because each project writes to
+    its own copy, ``wave`` is never reordered by completion, so iterating it in order is already
+    deterministic. Removing the SNAPSHOT is what turns the shared list back into a race and takes
+    ``test_concurrent_pass_is_byte_identical_to_sequential`` RED. This is recorded plainly rather
+    than dressing the loop in a ``sorted(...)`` that would re-sort an already-ordered list and read
+    as a guard while guarding nothing.
+
+    Safe to apply as-is because ``VerdictStore.add`` is first-write-wins per content-hash id
+    (``verdicts.py:303-304``) — the wave-start verdicts every snapshot carries are re-offered and
+    dropped, so only the new ones land."""
+    seen = {v.id for v in store.verdicts}
+    for _pid, snapshot in wave:
+        for verdict in snapshot.verdicts:
+            if verdict.id not in seen:
+                store.add(verdict)
+                seen.add(verdict.id)
+
+
 def _waves(ids: list[str], k: int) -> list[list[str]]:
     """Partition ``ids`` into consecutive waves of at most ``k``, preserving caller order (D-D).
 
@@ -626,55 +675,73 @@ async def run_portfolio(
     runs: list[RunResult] = []
     stopped_early = False
     stop_reason: GoalReached | None = None
-    for pid in ids:
-        if pid not in projects:
-            raise ValueError(f"unknown project_id: {pid!r}")
-        project = projects[pid]
+    for wave_ids in _waves(ids, concurrency):
+        members: list[str] = []
+        for pid in wave_ids:
+            if pid not in projects:
+                raise ValueError(f"unknown project_id: {pid!r}")
+            project = projects[pid]
 
-        if goals.portfolio is not None:
-            observed = ledger.portfolio_total()
-            limit = _goal_limit_if_reached(goals.portfolio, observed, portfolio_baseline_ore)
-            if limit is not None:
-                if goals.portfolio.mode == "hard":
-                    stopped_early = True
-                    stop_reason = GoalReached("portfolio", None, limit, observed)
-                    break
-                if stop_reason is None:
-                    stop_reason = GoalReached("portfolio", None, limit, observed)  # soft flag
+            if goals.portfolio is not None:
+                observed = ledger.portfolio_total()
+                limit = _goal_limit_if_reached(goals.portfolio, observed, portfolio_baseline_ore)
+                if limit is not None:
+                    if goals.portfolio.mode == "hard":
+                        stopped_early = True
+                        stop_reason = GoalReached("portfolio", None, limit, observed)
+                        break
+                    if stop_reason is None:
+                        stop_reason = GoalReached("portfolio", None, limit, observed)  # soft flag
 
-        per_project_goal = goals.per_project.get(pid)
-        if per_project_goal is not None:
-            observed = ledger.per_project_total(pid)
-            limit = _goal_limit_if_reached(per_project_goal, observed, _to_ore(project.total_cost))
-            if limit is not None:
-                if stop_reason is None:
-                    stop_reason = GoalReached("project", pid, limit, observed)
-                if per_project_goal.mode == "hard":
-                    continue  # skip THIS pid; the rest of the pass proceeds
+            per_project_goal = goals.per_project.get(pid)
+            if per_project_goal is not None:
+                observed = ledger.per_project_total(pid)
+                limit = _goal_limit_if_reached(
+                    per_project_goal, observed, _to_ore(project.total_cost)
+                )
+                if limit is not None:
+                    if stop_reason is None:
+                        stop_reason = GoalReached("project", pid, limit, observed)
+                    if per_project_goal.mode == "hard":
+                        continue  # skip THIS pid; the rest of the pass proceeds
+
+            members.append(pid)
+
+        # Every project in the wave reads the SAME wave-start state and writes only its own copy,
+        # so no two coroutines touch one list. The snapshot is what makes the barrier sufficient.
+        snapshots = [(pid, _wave_snapshot(store)) for pid in members]
 
         # run_portfolio only drives full runs (never dry-run), so the return narrows to RunResult;
         # the cast keeps the widened run_project signature honest without an @overload duplication.
-        result = cast(
-            RunResult,
-            await run_project(
-                pid,
-                profile,
-                docs_dir=project.docs_dir,
-                verdict_input=project.verdict_input,
-                bundle_dir=project.bundle_dir,
-                verdict_dir=project.verdict_dir,
-                dimension=dimension,
-                store=store,
-                client_factory=client_factory,
-                max_rounds=max_rounds,
-                max_tokens=max_tokens,
-                top_k=top_k,
-                semantic_retrieval=semantic_retrieval,
-                embedder=embedder,
-                meter=meter_factory() if meter_factory is not None else None,
-            ),
+        wave_results = await asyncio.gather(
+            *(
+                run_project(
+                    pid,
+                    profile,
+                    docs_dir=projects[pid].docs_dir,
+                    verdict_input=projects[pid].verdict_input,
+                    bundle_dir=projects[pid].bundle_dir,
+                    verdict_dir=projects[pid].verdict_dir,
+                    dimension=dimension,
+                    store=snapshot,
+                    client_factory=client_factory,
+                    max_rounds=max_rounds,
+                    max_tokens=max_tokens,
+                    top_k=top_k,
+                    semantic_retrieval=semantic_retrieval,
+                    embedder=embedder,
+                    meter=meter_factory() if meter_factory is not None else None,
+                )
+                for pid, snapshot in snapshots
+            )
         )
-        runs.append(result)
+        # ``gather`` resolves in ARGUMENT order, not completion order, and waves follow
+        # ``project_ids`` — so ``runs`` stays in caller order however the schedule interleaved.
+        runs.extend(cast(RunResult, r) for r in wave_results)
+        _merge_wave(store, snapshots)
+
+        if stopped_early:
+            break
 
     base = _aggregate(tuple(runs), store)
     if stopped_early or stop_reason is not None:

tests/test_portfolio_concurrent_loadbearing.py

@@ -283,3 +283,32 @@ async def test_concurrent_pass_is_byte_identical_to_sequential() -> None:
             "SC2 (see docstring); a divergence here means something broader than ordering broke"
         )
     assert concurrent_ids == sequential_ids
+
+
+async def test_runs_follow_project_ids_not_completion_order() -> None:
+    """``PortfolioResult.runs`` is ordered by ``project_ids``, never by which project finished
+    first — so a caller indexing ``runs[0]`` gets the project they listed first, at every ``k``.
+
+    This holds because ``asyncio.gather`` resolves in ARGUMENT order rather than completion order,
+    and waves are built in caller order. It is asserted separately from the store-content contract
+    because it survives a different set of mistakes: a barrier bug reorders the STORE while leaving
+    ``runs`` correct, and an executor that appended results as they completed would reorder ``runs``
+    while leaving the store correct. The probe is active, so completion order is confirmed to
+    differ from submission order — without that this assertion would be untested at k>1."""
+    probe = _Recorder()
+    _, result = await _pass(3, probe)
+
+    assert probe.max_in_flight > 1, "the pass ran sequentially — runs ordering is untested at k>1"
+    assert probe.completion_order() != _PORTFOLIO_IDS, (
+        "completion order matched submission order, so this assertion would hold trivially"
+    )
+    # ``claimed_saving_nok`` is the per-project discriminator in REPLIES (200k / 130k / 210k) —
+    # ``measure_type`` is NOT, since FV42 and BRU deliberately share "Reduce scope".
+    assert [r.verdict.proposal_features.claimed_saving_nok for r in result.runs] == [
+        200_000.0,
+        130_000.0,
+        210_000.0,
+    ], (
+        f"runs came back in {probe.completion_order()} (completion) rather than "
+        f"{_PORTFOLIO_IDS} (submission) order"
+    )

tests/test_scripted_client_consolidation.py

@@ -9,6 +9,7 @@ five ``def _inner_get_response`` sites (four scripted + test_step5's own-lineage
 
 from __future__ import annotations
 
+import re
 from pathlib import Path
 
 _ROOT = Path(__file__).resolve().parents[1]
@@ -18,9 +19,35 @@ def _py_files(base: str) -> list[Path]:
     return sorted((_ROOT / base).rglob("*.py"))
 
 
+# Files permitted to define ``_inner_get_response``. The invariant this guard protects is that the
+# scripted BODY is not duplicated — not that the def-site count is frozen. Adding a file here is a
+# deliberate act: a new entry must either be the canonical, or a thin override that DELEGATES to it
+# (which ``test_delegating_overrides_call_super`` below then enforces mechanically).
+_CANONICAL_SITE = "src/portfolio_optimiser/simulation.py"
+
+# Overrides in the SCRIPTED lineage — they subclass ``ScriptedChatClient``, so a body of their own
+# would be a copy of the canonical. They must delegate.
+_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",
+]
+
+# Doubles in a DIFFERENT lineage (``spikes._harness.FakeChatClient``). There is no canonical
+# scripted body above them to delegate to, so the delegation rule does not apply — but the
+# separation is asserted rather than assumed, so a file cannot be parked here to dodge the rule.
+_FOREIGN_LINEAGE = ["tests/test_step5_refine_loadbearing.py"]
+
+
 def test_inner_get_response_collapsed_to_two_sites() -> None:
-    """The four scripted clients collapse to ONE canonical ``_inner_get_response`` (simulation.py);
-    test_step5's own-lineage double is the only other def. So exactly 2 def-sites remain — NOT 5."""
+    """The four scripted clients collapse to ONE canonical ``_inner_get_response``
+    (``simulation.py``). Every other def-site must be a registered, DELEGATING override — never a
+    fourth copy of the body.
+
+    The guard originally pinned a literal count of 2. That made it fail on any new legitimate
+    subclass while still passing if someone pasted a duplicated body into an already-listed file —
+    a count is the wrong shape for the invariant. The list below plus
+    ``test_delegating_overrides_call_super`` pin the property itself."""
     sites = [
         p.relative_to(_ROOT).as_posix()
         for base in ("src", "tests")
@@ -28,10 +55,43 @@ def test_inner_get_response_collapsed_to_two_sites() -> None:
         if p.name != Path(__file__).name  # this guard file references the pattern in prose
         and "def _inner_get_response" in p.read_text(encoding="utf-8")
     ]
-    assert sorted(sites) == [
-        "src/portfolio_optimiser/simulation.py",
-        "tests/test_step5_refine_loadbearing.py",
-    ], f"expected the four scripted bodies collapsed to one canonical + test_step5's, got: {sites}"
+    expected = sorted([_CANONICAL_SITE, *_DELEGATING_OVERRIDES, *_FOREIGN_LINEAGE])
+    assert sorted(sites) == expected, (
+        f"unregistered ``_inner_get_response`` def-site — the scripted body must not be copied. "
+        f"Expected {expected}, got: {sites}"
+    )
+
+
+def test_foreign_lineage_doubles_are_genuinely_foreign() -> None:
+    """A file listed as foreign lineage must NOT subclass the scripted canonical.
+
+    Without this, ``_FOREIGN_LINEAGE`` would be an escape hatch: any scripted-lineage subclass
+    could be moved into that list to skip the delegation rule below."""
+    for site in _FOREIGN_LINEAGE:
+        text = (_ROOT / site).read_text(encoding="utf-8")
+        assert "ScriptedChatClient" not in text, (
+            f"{site} is registered as foreign lineage but references ``ScriptedChatClient`` — if it "
+            "is in the scripted lineage it belongs in _DELEGATING_OVERRIDES and must delegate"
+        )
+
+
+def test_delegating_overrides_call_super() -> None:
+    """Every scripted-lineage override actually DELEGATES to the canonical rather than
+    reimplementing it.
+
+    This is the strength the literal count never had: without it, a file already on the list could
+    grow a full copy of the scripted body and the consolidation would be cosmetic again."""
+    for site in _DELEGATING_OVERRIDES:
+        text = (_ROOT / site).read_text(encoding="utf-8")
+        # Match the delegation ITSELF — ``super()._inner_get_response`` or the explicit
+        # ``super(Cls, self)._inner_get_response`` form a nested function needs. Searching for
+        # "super(" and "_inner_get_response" independently would pass on any file that merely
+        # calls ``super().__init__`` near a def, which is accidental-green, not a guard.
+        assert re.search(r"super\([^)]*\)\._inner_get_response", text), (
+            f"{site} defines ``_inner_get_response`` but never delegates to the canonical via "
+            "``super()._inner_get_response`` — that is a duplicated body, which is exactly what "
+            "this guard exists to prevent"
+        )
 
 
 def test_no_src_imports_tests() -> None: