"""S3.3 concurrent fan-out — load-bearing seams for the D-D wave model. ``run_portfolio`` fans out over independent projects. S3.3 lets it run up to ``k`` projects concurrently WITHOUT spending the property the whole suite rests on: determinism. The design rests on one measured fact (``verdicts.py:303-304``): the only order-sensitive shared state in the pass is ``VerdictStore.verdicts`` — ONE list with ONE append site. Retrieval ranking is already order-independent (``verdicts.py:274-279`` ranks on ``(-similarity, id)``). So a per-wave snapshot plus a merge barrier that sorts each wave's new verdicts on ``project_id`` restores byte-identity exactly — nothing wider is needed, and nothing narrower suffices. This file grows across the plan's six steps. Step 1 covers the partitioning helper and the fail-fast; the determinism contract and its probe arrive in Step 2. """ from __future__ import annotations import ast import asyncio from collections.abc import Mapping, Sequence from pathlib import Path from typing import Any from unittest import mock import pytest from conftest import _PORTFOLIO_DEFAULT_REPLY, _ProjectAwareUsageChatClient from test_ledger import _prefilled from test_portfolio import REPLIES from portfolio_optimiser import ledger as ledger_mod 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.verdicts import VerdictStore # The shipped 3-project fixture, in submission order (mirrors tests/test_portfolio.py:57). # ``REPLIES`` is reused from there deliberately: its three proposals are already verified to # validate AND to mint three DISTINCT verdict ids, which is the precondition self-check 3 asserts. _PORTFOLIO_IDS = ["FV42-GSV-E1", "RV13-RAS-TP", "BRU-LAKS-REHAB"] # Scheduler yields per project, DESCENDING by submission index: the FIRST-submitted project yields # most and therefore tends to finish LAST. This is what makes completion order depart from # submission order — the precondition for the merge barrier to be load-bearing at all. _YIELDS_BY_PID = {pid: (len(_PORTFOLIO_IDS) - i) * 4 for i, pid in enumerate(_PORTFOLIO_IDS)} def test_waves_single_wave_when_k_covers_every_project() -> None: """k >= len(ids) is one wave — the whole portfolio runs concurrently.""" assert _waves(["a", "b", "c"], 3) == [["a", "b", "c"]] assert _waves(["a", "b", "c"], 99) == [["a", "b", "c"]] def test_waves_partitions_into_consecutive_chunks() -> None: """A k that does not divide the portfolio leaves a short final wave — never a dropped id.""" assert _waves(["a", "b", "c"], 2) == [["a", "b"], ["c"]] assert _waves(["a", "b", "c", "d"], 2) == [["a", "b"], ["c", "d"]] def test_waves_k1_is_the_sequential_path() -> None: """k=1 degenerates to one project per wave — the pre-S3.3 execution order, by construction. This is what makes Step 1 behaviour-preserving: with ``concurrency=1`` every wave holds a single project, so the wave loop visits ids in exactly the order the old ``for pid in ids`` loop did.""" assert _waves(["a", "b", "c"], 1) == [["a"], ["b"], ["c"]] def test_waves_preserves_caller_order_and_loses_nothing() -> None: """Flattening any partition reproduces the input exactly — order preserved, no id dropped or duplicated. The wave model may change WHEN a project runs, never WHICH projects run.""" ids = ["delta", "alpha", "charlie", "bravo", "echo"] for k in range(1, len(ids) + 2): flattened = [pid for wave in _waves(ids, k) for pid in wave] assert flattened == ids, f"k={k} did not preserve caller order" assert _waves([], 3) == [] async def test_concurrency_below_one_fails_fast_before_any_project_loads() -> None: """k < 1 raises ``ValueError`` with an explanatory message, mirroring the fail-fast idiom at ``run.py:547-553``. The raise precedes project loading, so the caller gets the error rather than a silently-empty pass — CLAUDE.md's "stoppkriterier + budsjett-tak påkrevd ved oppstart, fail-fast, aldri ubegrenset loop".""" for bad in (0, -1): with pytest.raises(ValueError, match="concurrency must be >= 1"): await run_portfolio(concurrency=bad) # -------------------------------------------------------------------------------------------- # Step 2 — the determinism contract, and the probe that makes it non-vacuous. # -------------------------------------------------------------------------------------------- class _Recorder: """Observes the pass at the ONE seam a test may touch: the chat client. Tracks, per client call, which project it belonged to and when it entered/left the awaited body. That yields three things no counter or timing could: genuine max-in-flight (two projects are concurrent iff their calls overlap between entry and exit), per-project completion order (the order of each project's LAST exit), and the raw call sequence.""" def __init__(self) -> None: self.entries: list[str] = [] self.exits: list[str] = [] self._active: set[str] = set() self.max_in_flight = 0 def enter(self, pid: str) -> None: self.entries.append(pid) self._active.add(pid) self.max_in_flight = max(self.max_in_flight, len(self._active)) def leave(self, pid: str) -> None: self.exits.append(pid) self._active.discard(pid) def completion_order(self) -> list[str]: """Projects ordered by their LAST client call — the closest observable proxy for the order in which the projects actually finished.""" order: list[str] = [] for pid in self.exits: if pid in order: order.remove(pid) order.append(pid) return order class _OrderProbeClient(_ProjectAwareUsageChatClient): """Perturbs completion order DETERMINISTICALLY, without a wall-clock sleep. ``_inner_get_response`` (``simulation.py:113-120``) is a SYNCHRONOUS method declared to return ``Awaitable[ChatResponse] | ResponseStream[...]``. The probe therefore returns a coroutine that yields to the event loop N times before delegating to the canonical body. ``asyncio.sleep(0)`` is a pure scheduler yield, not a timed wait, so the perturbation is deterministic rather than timing-dependent — and it cannot make a sequential pass interleave, which is exactly why this probe can tell the two apart. N is derived from the PROMPT BLOB, never from a constructor argument: ``client_factory`` is ``Callable[[str], BaseChatClient]`` whose argument is the ROLE, not the project id — which is precisely why ``_ProjectAwareUsageChatClient`` scans the prompt in the first place. The stream path is delegated UNTOUCHED. The canonical returns a coroutine when ``stream=False`` and a stream-shaped object when ``stream=True``; wrapping both in one ``async def`` would hand the framework the wrong shape. The subclass chain (probe -> ``_ProjectAwareUsageChatClient`` -> ``ScriptedChatClient`` -> ``OpenAIChatCompletionClient``) is preserved so ``BudgetMiddleware`` still engages.""" def __init__(self, replies: dict[str, str], *, default_reply: str, recorder: _Recorder) -> None: super().__init__(replies, default_reply=default_reply) self._recorder = recorder def _inner_get_response( self, *, messages: Sequence[Any], options: Mapping[str, Any], stream: bool = False, **kwargs: Any, ) -> Any: if stream: return super()._inner_get_response( messages=messages, options=options, stream=True, **kwargs ) blob = " ".join(getattr(m, "text", "") or "" for m in messages) pid = next((p for p in _YIELDS_BY_PID if p in blob), "") yields = _YIELDS_BY_PID.get(pid, 0) recorder = self._recorder async def _probed() -> Any: recorder.enter(pid) try: for _ in range(yields): await asyncio.sleep(0) return await super(_OrderProbeClient, self)._inner_get_response( messages=messages, options=options, stream=False, **kwargs ) finally: recorder.leave(pid) return _probed() async def _pass(k: int, recorder: _Recorder) -> tuple[list[str], PortfolioResult]: """One portfolio pass at concurrency ``k`` on a FRESH store, returning the store's verdict-id SEQUENCE (order is the contract — never a set, never ``sorted(...)``) and the aggregate.""" store = VerdictStore(verdicts=[]) def factory(_role: str) -> Any: return _OrderProbeClient(REPLIES, default_reply=_PORTFOLIO_DEFAULT_REPLY, recorder=recorder) result = await run_portfolio( _PORTFOLIO_IDS, "local", store=store, client_factory=factory, concurrency=k ) return [v.id for v in result.store.verdicts], result def _aggregate_fields(result: PortfolioResult) -> tuple[Any, ...]: """The aggregate's comparable scalar surface, plus the ORDER of ``runs`` — which must follow ``project_ids``, never completion order.""" return ( result.validated_count, result.rejected_count, result.sum_claimed_saving_nok, result.sum_token_usage, result.stopped_early, tuple(r.verdict.id for r in result.runs), ) 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. **Detach point: remove the ``sorted(...)`` from ``_merge_wave`` -> this test goes RED.** That is the seam. Without the sort, each wave's verdicts land in completion order, which the probe has made differ from submission order, and the two id sequences diverge. **Probe mechanism.** ``_OrderProbeClient`` returns a coroutine that yields to the loop N times (``asyncio.sleep(0)``, a pure scheduler yield — no wall clock) before delegating to the canonical body, with N keyed on the project found in the prompt blob and DESCENDING by submission index. See the class docstring for why N cannot be a constructor argument and why the stream path is delegated untouched. **Three probe self-checks, and why the test is worthless without them.** Each one closes a way this assertion could pass while proving nothing: 1. *The probe actually reordered.* If the yields had no effect, completion order would equal submission order, the barrier would never be exercised, and the equality would hold with the sort deleted. Asserted as completion order != submission order — NOT as an exact permutation: ``asyncio.sleep(0)`` is a scheduler yield, not an ordering primitive, and the total interleaving depends on every ``await`` along each ``run_project`` path, not only the probe's. Any non-identity permutation exercises the barrier, so demanding a specific one would only add a way for the test to fail for a reason that is not the contract. 2. *Concurrency actually happened.* A ``concurrency=3`` that silently degraded to sequential would make both sides run the identical path — trivially equal, and the sneakiest failure mode of the set. Asserted as max-in-flight > 1, measured from genuine overlap between a client call's entry and exit, not from a counter. 3. *There is something to order.* Ids hash candidate features; if the fixture collapsed to one id under first-write-wins there would be no ordering to preserve. Asserted as >= 3 distinct ids. **SC2 has a strong half and a weak half, and this test does not pretend otherwise.** Equality of STORE CONTENT is a real test of the barrier. Equality of the AGGREGATE is nearly free on this fixture: retrieval ranks on ``(-similarity, id)`` (order-independent, ``verdicts.py:274-279``) and the Step-1 ExpeL fold is ``bundle_dir``-gated with no reference project setting it. Reporting only "byte identical" would be half-vacuous — the same defect class as the Spike B call-counter tautology. Step 6 adds the bundle-backed case where the difference IS observable. **What the repetitions do and do not guard.** They do NOT guard hash-seed nondeterminism: ``PYTHONHASHSEED`` is fixed for the life of the interpreter, so every in-process repetition shares one seed and the second rep can never disagree with the first on that axis. What they DO guard is scheduler nondeterminism under real concurrency — ``asyncio.gather`` gives no byte-identical interleaving guarantee across repetitions, so a barrier that happened to work once is caught here. Claiming otherwise would be the over-read the S3.1 review corrected.""" probe = _Recorder() concurrent_ids, concurrent = await _pass(3, probe) # Self-check 2 — concurrency actually happened. assert probe.max_in_flight > 1, ( f"max in-flight was {probe.max_in_flight}: concurrency=3 never overlapped two projects, so " f"this pass ran the sequential path and the equality below proves nothing. " f"call sequence={probe.entries}" ) # Self-check 1 — the probe actually reordered (non-identity, not an exact permutation). observed = probe.completion_order() assert observed != _PORTFOLIO_IDS, ( f"completion order {observed} equals submission order {_PORTFOLIO_IDS}: the probe did not " "reorder anything, so the merge barrier was never exercised and this test would stay " "green with the sort deleted" ) # Self-check 3 — there is something to order. assert len(set(concurrent_ids)) >= 3, ( f"only {len(set(concurrent_ids))} distinct verdict ids ({concurrent_ids}): the fixture " "collapsed under first-write-wins, so no ordering exists to preserve" ) # The contract itself, repeated to catch a barrier that works only by scheduling luck. for rep in range(20): sequential_ids, sequential = await _pass(1, _Recorder()) again_ids, again = await _pass(3, _Recorder()) assert again_ids == sequential_ids, ( f"rep {rep}: concurrent store order {again_ids} != sequential {sequential_ids} — the " f"merge barrier is not restoring deterministic order (probe reordered to {observed})" ) assert _aggregate_fields(again) == _aggregate_fields(sequential), ( f"rep {rep}: aggregate diverged between k=3 and k=1 — note this is the WEAK half of " "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" ) # -------------------------------------------------------------------------------------------- # 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" ) # -------------------------------------------------------------------------------------------- # Step 5 — goal-stop at wave boundaries, and the two structural guards (single loop, one writer). # -------------------------------------------------------------------------------------------- async def _goal_pass( k: int, *, ledger: SavingsLedger, goals: GoalConfig, recorder: _Recorder ) -> PortfolioResult: """A pass at concurrency ``k`` under a goal contract, on a FRESH store.""" def factory(_role: str) -> Any: return _OrderProbeClient(REPLIES, default_reply=_PORTFOLIO_DEFAULT_REPLY, recorder=recorder) return await run_portfolio( _PORTFOLIO_IDS, "local", store=VerdictStore(verdicts=[]), client_factory=factory, ledger=ledger, goals=goals, concurrency=k, ) def _ran(result: PortfolioResult) -> list[str]: """The project ids that actually RAN, in ``runs`` order.""" return [r.outcome.proposal.project_id for r in result.runs] async def test_goal_stop_membership_is_identical_across_concurrency() -> None: """WHICH projects run under a goal contract must not depend on ``concurrency`` (SC6 under D-D). The wave-model reading of the existing ``break``/``continue`` semantics, now pinned rather than only described: goal checks run at WAVE ASSEMBLY, never mid-wave. A HARD per-project goal removes that pid from its wave before the wave starts; a HARD portfolio goal stops the pass, and the in-flight wave still crosses its merge barrier before the loop exits. Checking mid-wave instead would make membership depend on which project's round-trips finished first — the exact nondeterminism S3.3 exists to avoid. **Detach point: check the goal at WAVE granularity instead of per member -> this test goes RED (measured by keying the per-project lookup on ``wave_ids[0]``, which leaves the skipped pid in the wave at k=3 while k=1 is unaffected, since at k=1 the wave IS the member).** **What is NOT the detach point, corrected against measurement rather than assumed.** The intuitive claim — "check mid-wave and membership becomes completion-order dependent" — is false here, and stating it would have been an over-read. ``_goal_limit_if_reached`` reads only the ledger, the goal contract and the baseline, and all three are INVARIANT for the duration of a pass (C3, pinned by the one-writer test below). A check moved later in the wave therefore reads exactly the same values and reaches exactly the same decision. Determinism of membership comes from the ledger being static, NOT from where the check sits; what wave-assembly placement actually buys is that a skipped project is never STARTED, so no work and no store write happen for a project the goal excluded. **The two halves are not equally strong, and this test does not pretend otherwise.** The per-project skip is the informative half: two projects still run, concurrently, and the skipped pid must be absent at both ``k``. The portfolio hard stop is the weak half — because the ledger is static during a pass (C3, and Step 5's one-writer guard below turns that from prose into a test), a reached portfolio goal is observed on the very first check, so BOTH ``k`` yield an empty ``runs`` and the equality is nearly free. It is asserted anyway because a regression that let a hard stop leak one wave's worth of runs would show up here first.""" # Strong half — a per-project HARD goal skips exactly that pid, at every k. per_project = GoalConfig(per_project={"RV13-RAS-TP": GoalContract(absolute_ore=1000)}) probe = _Recorder() concurrent = await _goal_pass( 3, ledger=_prefilled("RV13-RAS-TP", 1000), goals=per_project, recorder=probe ) sequential = await _goal_pass( 1, ledger=_prefilled("RV13-RAS-TP", 1000), goals=per_project, recorder=_Recorder() ) assert probe.max_in_flight > 1, ( f"max in-flight was {probe.max_in_flight}: the surviving two projects never overlapped, so " "membership under concurrency is untested here" ) assert _ran(concurrent) == _ran(sequential) == ["FV42-GSV-E1", "BRU-LAKS-REHAB"], ( f"membership diverged with k: k=3 ran {_ran(concurrent)}, k=1 ran {_ran(sequential)}" ) assert concurrent.stopped_early is False and sequential.stopped_early is False, ( "a per-project skip is not a pass-stop" ) # Weak half — a HARD portfolio goal already reached stops the pass identically at every k. portfolio = GoalConfig(portfolio=GoalContract(absolute_ore=100)) stopped_concurrent = await _goal_pass( 3, ledger=_prefilled("FV42-GSV-E1", 100), goals=portfolio, recorder=_Recorder() ) stopped_sequential = await _goal_pass( 1, ledger=_prefilled("FV42-GSV-E1", 100), goals=portfolio, recorder=_Recorder() ) assert _ran(stopped_concurrent) == _ran(stopped_sequential) == [] assert stopped_concurrent.stopped_early is stopped_sequential.stopped_early is True assert stopped_concurrent.stop_reason == stopped_sequential.stop_reason, ( "the goal-stop DECISION itself must not depend on k — frozen-dataclass equality" ) def test_no_thread_or_process_path_exists_under_src() -> None: """NG1, structural: S3.3's concurrency is ONE asyncio event loop and nothing else. A ratchet, green today, in the style of ``test_okf.py::test_okf_is_maf_free`` and AST-based for the same reason — a raw substring scan would trip on prose that merely NAMES the forbidden mechanism, which this very docstring does. Why it is worth a test rather than a convention: MAF's thread-safety is undocumented, and Spike B (``a2dff21``) measured workflow state bleeding across reused instances. A ``run_in_executor`` or ``asyncio.to_thread`` hop introduced later would move project runs onto threads where none of S3.3's reasoning holds — the wave snapshot argument is about coroutines interleaving at ``await`` points, not about parallel memory access — and it would do so without any existing test noticing. ``asyncio.to_thread`` is included deliberately: it is the modern, innocuous- looking spelling of exactly the hop this guard forbids.""" src_dir = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" forbidden_modules = {"threading", "_thread", "multiprocessing", "concurrent"} forbidden_calls = {"run_in_executor", "to_thread", "ThreadPoolExecutor", "ProcessPoolExecutor"} offences: list[str] = [] for path in sorted(src_dir.rglob("*.py")): tree = ast.parse(path.read_text(encoding="utf-8")) for node in ast.walk(tree): if isinstance(node, ast.Import): offences += [ f"{path.name}: import {a.name}" for a in node.names if a.name.split(".")[0] in forbidden_modules ] elif isinstance(node, ast.ImportFrom): if (node.module or "").split(".")[0] in forbidden_modules: offences.append(f"{path.name}: from {node.module} import ...") elif isinstance(node, ast.Call): name = getattr(node.func, "attr", None) or getattr(node.func, "id", None) if name in forbidden_calls: offences.append(f"{path.name}: {name}(...)") assert offences == [], ( f"a thread/process execution path exists under src/ ({offences}) — S3.3's determinism " "argument covers coroutines on one loop only (NG1)" ) async def test_concurrent_pass_does_not_write_on_the_run_path() -> None: """The one-writer rule, turned from a prose invariant into a measurement (C3, assumption 2). The whole goal-stop design rests on the ledger being STATIC during a pass: ``_goal_limit_if_ reached`` reads an ACCUMULATED sum that earlier, out-of-band HITL realizations produced, and it reads it while projects are running. If a run realized savings mid-pass, the reads would race the writes and goal membership would stop being deterministic — so assumption 2 is load-bearing for the test above, not merely tidy. Likewise ``promote_verdict``: the wiki is the CURATED layer behind the fail-closed Step-8 gate. A concurrent pass writing to it would push raw agent output into next run's navigation context without any human/persona approval — self-contamination, at k-times the rate. Both are asserted by TRIPWIRE rather than by grep: the module attribute is replaced, so any call reaching it through any path fails the test, including one added later in a module this file never mentions. ``write_verdict`` is deliberately NOT covered here — it is guarded by the Steg-7 role-split tests, and duplicating it would imply this test knows something about the outbox that it does not.""" calls: list[str] = [] def _tripwire(name: str) -> Any: def _fail(*_args: Any, **_kwargs: Any) -> Any: calls.append(name) raise AssertionError(f"{name} was called on the run path during a concurrent pass") return _fail with ( mock.patch.object(verdicts_mod, "promote_verdict", _tripwire("promote_verdict")), mock.patch.object(ledger_mod, "realize", _tripwire("realize")), ): _, result = await _pass(3, _Recorder()) assert calls == [], f"one-writer rule violated during a concurrent pass: {calls}" assert len(result.runs) == 3, ( "the pass must still have completed — an empty pass proves nothing" )