test(s33): pin the concurrent-equals-sequential contract — RED before the executor exists
Commits the determinism contract RED, mirroring S3.1 (b05747aRED ->6618f67GREEN). The wave executor does not exist yet, so `concurrency=3` still runs the sequential path — and the test says so precisely rather than passing vacuously: max in-flight was 1: concurrency=3 never overlapped two projects call sequence=[FV42 x4, RV13 x4, BRU x4] That blocked call sequence is the proof the probe works. A test that had gone GREEN here would have been asserting something the sequential path already satisfies — the Spike B call-counter tautology repeating — and the plan's escalation rule would have stopped execution. The probe returns a coroutine yielding N times via asyncio.sleep(0) (a pure scheduler yield, no wall clock) before delegating to the canonical body, with N read from the prompt blob since client_factory's argument is the ROLE, not the project id. The stream path is delegated untouched, and the subclass chain is preserved so BudgetMiddleware still engages. Three self-checks make it non-vacuous by construction: max-in-flight > 1, completion order != submission order, and >= 3 distinct verdict ids. Self-check 1 asserts non-identity rather than an exact reversal — asyncio.sleep(0) is a scheduler yield, not an ordering primitive, so demanding a specific permutation would only add a way to fail for a reason that is not the contract. The docstring also corrects what the repetitions guard: NOT hash-seed nondeterminism (PYTHONHASHSEED is fixed per interpreter, so every in-process rep shares one seed), but scheduler nondeterminism under real concurrency. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LQapztREtC2mkr5oU811pr
This commit is contained in:
parent
c15165ca77
commit
5d0bb4c981
1 changed files with 225 additions and 1 deletions
|
|
@ -15,9 +15,26 @@ fail-fast; the determinism contract and its probe arrive in Step 2.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import Any
|
||||
|
||||
from portfolio_optimiser.run import _waves, run_portfolio
|
||||
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.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:
|
||||
|
|
@ -59,3 +76,210 @@ async def test_concurrency_below_one_fails_fast_before_any_project_loads() -> No
|
|||
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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue