feat(s33): collect-and-continue error policy via RunFailure slots

One project's exception no longer cancels its siblings: the wave's gather
runs with return_exceptions=True and each raised project becomes a frozen
RunFailure(project_id, error, error_type) in the defaulted
PortfolioResult.failures, while every completed project keeps its full
RunResult. RunFailure is a distinct type rather than an error field on
RunResult (six required non-defaulted fields -> dummies would be fabricated
provenance); the deviation from the spec's wording is stated in both
docstrings. TaskGroup is rejected: it cancels siblings on first exception.

snapshots reaches _merge_wave unfiltered, and results are paired back to pids
by POSITION (gather resolves in argument order) so a mid-wave failure cannot
disturb store order or misattribute the failure.

Detach points measured, not asserted:
  drop return_exceptions=True      -> RED (both new tests)
  reorder snapshots before merge   -> RED (+ Session 1's determinism test)
  pair results by sorted(), not position -> RED (failure misattributed)
  filter failed members before merge -> GREEN, measured

The last one corrects the plan: its carried-forward requirement implied
filtering before the merge was the hazard. Filtering preserves relative order,
which is all _merge_wave consumes, so the variant is undetectable AND harmless.
The docstring now names the reorder as the detach point and records the
filter asymmetry, rather than claiming a detach point that does not exist.

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

tests/test_portfolio_concurrent_loadbearing.py

@@ -23,7 +23,7 @@ import pytest
 from conftest import _PORTFOLIO_DEFAULT_REPLY, _ProjectAwareUsageChatClient
 from test_portfolio import REPLIES
 
-from portfolio_optimiser.run import PortfolioResult, _waves, run_portfolio
+from portfolio_optimiser.run import PortfolioResult, RunFailure, _waves, run_portfolio
 from portfolio_optimiser.verdicts import VerdictStore
 
 # The shipped 3-project fixture, in submission order (mirrors tests/test_portfolio.py:57).
@@ -312,3 +312,191 @@ async def test_runs_follow_project_ids_not_completion_order() -> None:
         f"runs came back in {probe.completion_order()} (completion) rather than "
         f"{_PORTFOLIO_IDS} (submission) order"
     )
+
+
+# --------------------------------------------------------------------------------------------
+# Step 4 — collect-and-continue: one project's failure must not cancel its siblings, and must
+# not disturb the order the surviving projects merge in.
+# --------------------------------------------------------------------------------------------
+
+# The MIDDLE project of the wave. Middle is deliberate: a failure at either end can be dropped by
+# a truncation bug and still leave the survivors in the right relative order, so an end position
+# would let the ordering assertion below pass for the wrong reason.
+_FAILING_PID = "RV13-RAS-TP"
+
+
+class _FailingProbeClient(_OrderProbeClient):
+    """The order probe, plus a synthetic backend failure for exactly ONE project.
+
+    The failure is keyed on the prompt blob for the same reason N is (see ``_OrderProbeClient``):
+    ``client_factory`` receives the ROLE, not the project id, so per-project behaviour can only be
+    selected from the prompt. It is raised from inside a coroutine rather than synchronously,
+    because the canonical ``_inner_get_response`` is a sync method RETURNING an awaitable — raising
+    synchronously would blow up at ``asyncio.gather`` argument-construction time, before any
+    concurrency exists, and would therefore test a different thing than a mid-flight failure.
+
+    The stream path is delegated untouched, exactly as in the base probe."""
+
+    def __init__(
+        self,
+        replies: dict[str, str],
+        *,
+        default_reply: str,
+        recorder: _Recorder,
+        failing_pid: str,
+    ) -> None:
+        super().__init__(replies, default_reply=default_reply, recorder=recorder)
+        self._failing_pid = failing_pid
+
+    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:
+            recorder = self._recorder
+            failing_pid = self._failing_pid
+
+            async def _boom() -> Any:
+                recorder.enter(failing_pid)
+                try:
+                    raise RuntimeError(f"synthetic backend failure for {failing_pid}")
+                finally:
+                    recorder.leave(failing_pid)
+
+            return _boom()
+        return super()._inner_get_response(
+            messages=messages, options=options, stream=stream, **kwargs
+        )
+
+
+async def _failing_pass(k: int, recorder: _Recorder) -> tuple[list[str], PortfolioResult]:
+    """A pass over the same 3-project fixture where the MIDDLE project's backend raises."""
+    store = VerdictStore(verdicts=[])
+
+    def factory(_role: str) -> Any:
+        return _FailingProbeClient(
+            REPLIES,
+            default_reply=_PORTFOLIO_DEFAULT_REPLY,
+            recorder=recorder,
+            failing_pid=_FAILING_PID,
+        )
+
+    result = await run_portfolio(
+        _PORTFOLIO_IDS, "local", store=store, client_factory=factory, concurrency=k
+    )
+    return [v.id for v in result.store.verdicts], result
+
+
+async def test_one_project_failure_does_not_cancel_its_siblings() -> None:
+    """Collect-and-continue (SC3): one project's exception is captured as a ``RunFailure`` slot
+    while every sibling in the same wave completes normally, and the pass itself does not raise.
+
+    **Detach point 1: drop ``return_exceptions=True`` from the wave's ``asyncio.gather`` -> this
+    test goes RED (measured).** Without it the first exception propagates out of ``gather``,
+    ``run_portfolio`` raises, and the surviving projects' results are discarded — the caller loses
+    two completed runs because a third failed.
+
+    **Detach point 2: pair the gathered results back to project ids by anything other than POSITION
+    -> this test goes RED (measured with ``sorted(snapshots)``).** ``gather`` resolves in argument
+    order, which is the only thing that makes positional pairing sound; a pairing keyed on a sorted
+    or completion-derived sequence attributes the failure to an innocent project, and the assertion
+    on ``failure.project_id`` is what catches it.
+
+    ``asyncio.TaskGroup`` is rejected for the same reason and is NOT an implementation detail: it
+    cancels its siblings on first exception, which is collect-and-continue's exact negation.
+
+    **Why a separate ``RunFailure`` rather than the spec's literal "``RunResult`` slot with an error
+    field".** ``RunResult`` is frozen with six required non-defaulted fields (``run.py:88-93``); a
+    run that never reached generation has no honest value for ``provenance``, ``verdict`` or
+    ``outcome``, and inventing them would put fabricated provenance into the aggregate — the one
+    thing this repo's provenance rules exist to prevent. The deviation is stated in the plan's
+    Step 4 and in ``PortfolioResult``'s docstring, not hidden."""
+    recorder = _Recorder()
+    store_ids, result = await _failing_pass(3, recorder)
+
+    # The pass returned rather than raising — that is half the contract, and it is asserted by
+    # having reached this line at all.
+    assert len(result.runs) == 2, (
+        f"expected the two healthy projects to survive, got {len(result.runs)} runs — a sibling "
+        f"was cancelled by {_FAILING_PID}'s failure"
+    )
+    assert len(result.failures) == 1, f"expected exactly one failure slot, got {result.failures}"
+
+    failure = result.failures[0]
+    assert isinstance(failure, RunFailure)
+    assert failure.project_id == _FAILING_PID, (
+        f"failure was attributed to {failure.project_id!r}, not {_FAILING_PID!r} — the result "
+        "list and the wave's submission list have drifted out of alignment"
+    )
+    assert failure.error_type == "RuntimeError"
+    assert "synthetic backend failure" in failure.error
+
+    # The survivors are the two healthy projects, in SUBMISSION order — 200k is FV42 (submitted
+    # first), 210k is BRU (submitted last). ``measure_type`` is not a discriminator here: FV42 and
+    # BRU deliberately share "Reduce scope".
+    assert [r.verdict.proposal_features.claimed_saving_nok for r in result.runs] == [
+        200_000.0,
+        210_000.0,
+    ]
+    # The aggregate counts only what actually ran: a failure is neither a validation nor a
+    # rejection, so it must not inflate either partition.
+    assert result.validated_count + result.rejected_count == 2
+    assert len(store_ids) == 2, (
+        f"the failed project contributed a verdict to the store ({store_ids}) — a run that never "
+        "produced an outcome must not leave one behind"
+    )
+
+
+async def test_store_order_survives_a_mid_wave_failure() -> None:
+    """The merge barrier still sees SUBMISSION order when a wave member failed.
+
+    **Detach point: REORDER the wave list on its way to ``_merge_wave`` -> this test goes RED
+    (measured with ``reversed(snapshots)``, which also takes Session 1's determinism test red).**
+    ``_merge_wave`` derives store order from the order ``snapshots`` arrives in and from nothing
+    else, so that sequence is the contract.
+
+    This is the trap Session 1 could not have caught, and said so: its determinism test contains no
+    failing project, so it cannot exercise a ``return_exceptions=True`` handler at all.
+
+    **What this test does NOT catch, measured rather than assumed.** The plan's carried-forward
+    requirement was "filter when building ``runs``/``failures``, never before the merge". Filtering
+    the failed members out of ``snapshots`` before the barrier was measured GREEN — filtering
+    preserves RELATIVE order, and relative order is all ``_merge_wave`` consumes, so the variant is
+    both undetectable here and harmless in fact. The executor still passes ``snapshots`` unfiltered,
+    but the honest reason is defensive rather than tested: it removes the place where the
+    reorder-shaped mistake would be written. Recording this asymmetry is the point — a docstring
+    that claimed "filter -> RED" would name a detach point that does not exist, which is precisely
+    the green-but-dead defect this repo's method is built to prevent.
+
+    The two probe self-checks are repeated here rather than inherited, because without them this
+    assertion is satisfied by a sequential pass, where submission and completion order coincide and
+    no ordering bug can be observed at all."""
+    recorder = _Recorder()
+    store_ids, result = await _failing_pass(3, recorder)
+
+    assert recorder.max_in_flight > 1, (
+        f"max in-flight was {recorder.max_in_flight}: the wave never overlapped two projects, so "
+        f"the ordering assertion below holds trivially. call sequence={recorder.entries}"
+    )
+    observed = recorder.completion_order()
+    assert observed != _PORTFOLIO_IDS, (
+        f"completion order {observed} equals submission order {_PORTFOLIO_IDS}: nothing reordered, "
+        "so this test would stay green with the barrier's ordering discipline removed"
+    )
+
+    # The contract: the store's verdict SEQUENCE is the surviving projects in submission order —
+    # which is exactly the order ``runs`` carries. Compared as sequences, never sets.
+    assert store_ids == [r.verdict.id for r in result.runs], (
+        f"store order {store_ids} diverged from submission order "
+        f"{[r.verdict.id for r in result.runs]} once a wave member failed (completion order was "
+        f"{observed})"
+    )
+    assert len(set(store_ids)) == 2, (
+        f"the two survivors did not mint distinct verdict ids ({store_ids}), so there is no "
+        "ordering left for this test to pin"
+    )