test(integration): fix seven CircleCI integration tests that race or assume a worker spread (#47001)

* test(integration): expect the tool search call replayed beside the function call for Anthropic targets

* test(integration): wait for the team spend flush before killing a worker

* test(integration): poll for the S3 attempt the batch sinks write concurrently

* test(integration): queue the terminal object and its sibling into the same s3_v2 flush

* test(integration): time the long Bedrock version against an equally long unversioned name

* test(integration): pin each half of a burst to one worker instead of hoping the kernel spreads it

* test(integration): assert parsed SSE events for live pings instead of transport chunk boundaries

tests/integration/_support/wire.py

@@ -6,6 +6,7 @@ import time
 from collections.abc import Callable, Generator, Iterator, Mapping
 from contextlib import contextmanager
 from dataclasses import dataclass
+from functools import partial
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from queue import SimpleQueue
 from types import MappingProxyType
@@ -60,6 +61,7 @@ class Reply:
     abort_after: int | None = None
     gate_after_first: threading.Event | None = None
     pause_between_chunks: float = 0
+    heartbeat: bytes | None = None
     headers: Mapping[str, str] = MappingProxyType({})
     drop_connection: bool = False
 
@@ -82,10 +84,14 @@ class Wire:
         return self.connected.qsize() - self.closed.qsize()
 
 
-def _await_release(gate: threading.Event, closing: threading.Event) -> bool:
+def _await_release(
+    gate: threading.Event, closing: threading.Event, beat: Callable[[], None] | None = None, every: float = 0.05
+) -> bool:
     while not closing.is_set():
-        if gate.wait(timeout=0.05):
+        if gate.wait(timeout=every):
             return True
+        if beat is not None:
+            beat()
     return gate.is_set()
 
 
@@ -117,6 +123,10 @@ def wire_server(
             super().finish()
             closed.put(f"{self.client_address[0]}:{self.client_address[1]}")
 
+        def write_chunk(self, chunk: bytes) -> None:
+            self.wfile.write(b"%x\r\n%s\r\n" % (len(chunk), chunk))
+            self.wfile.flush()
+
         def respond(self) -> None:
             headers: Final = {name.lower(): value for name, value in self.headers.items()}
             try:
@@ -156,12 +166,20 @@ def wire_server(
                     for index, chunk in enumerate(reply.chunks):
                         if reply.abort_after == index:
                             break
-                        self.wfile.write(b"%x\r\n%s\r\n" % (len(chunk), chunk))
-                        self.wfile.flush()
+                        self.write_chunk(chunk)
                         if (
                             index == 0
                             and reply.gate_after_first is not None
-                            and not _await_release(reply.gate_after_first, closing)
+                            and not (
+                                _await_release(reply.gate_after_first, closing)
+                                if reply.heartbeat is None
+                                else _await_release(
+                                    reply.gate_after_first,
+                                    closing,
+                                    partial(self.write_chunk, reply.heartbeat),
+                                    reply.pause_between_chunks,
+                                )
+                            )
                         ):
                             break
                         if reply.pause_between_chunks and index + 1 < len(reply.chunks):

tests/integration/observability/test_batch_line_items_cold_storage.py

@@ -817,7 +817,7 @@ def test_a_failing_bucket_still_bills_the_batch_and_the_other_sink_keeps_its_lin
     assert [record["custom_id"] for record in _records(gcs_body)] == [custom_id]
     assert spend > 0
     s3_name: Final = _object_name(f"{S3_PREFIX}/", batch.batch_id)
-    assert rig.s3.tried(s3_name) >= 1
+    eventually(lambda: rig.s3.tried(s3_name), lambda tries: tries >= 1, POLLER_SECONDS)
     assert s3_name not in rig.s3.objects()
 
 

tests/integration/observability/test_s3_v2_upload_fanout.py

@@ -420,7 +420,16 @@ def test_s3_v2_object_rejected_with_a_bucket_wide_code_is_delivered_once_the_fau
 def test_s3_v2_terminal_object_is_put_once_and_dropped_by_default(gateway: Gateway, tmp_path: Path) -> None:
     marker: Final = "s3toolarge" + uuid.uuid4().hex[:8]
     sink: Final = RejectingS3Sink(reject_marker=f"{marker}-huge", reject_status=400, reject_code="EntityTooLarge")
-    with wire_server(_chat_reply) as provider, wire_server(sink.respond) as bucket:
+    flushing: Final = threading.Event()
+    release: Final = threading.Event()
+
+    def respond(request: Request) -> Reply:
+        if f"{marker}-held".encode() in request.body:
+            flushing.set()
+            assert release.wait(timeout=60), "the held flush was never released"
+        return sink.respond(request)
+
+    with wire_server(_chat_reply) as provider, wire_server(respond) as bucket:
         config: Final = _s3_config(tmp_path, bucket.url, {"s3_batch_file_upload": False})
         with (
             owned_proxy(gateway, tmp_path, {"DEFAULT_S3_FLUSH_INTERVAL_SECONDS": "2"}, config=config) as candidate,
@@ -428,11 +437,16 @@ def test_s3_v2_terminal_object_is_put_once_and_dropped_by_default(gateway: Gatew
         ):
             model: Final = scenario.model(api_base=provider.url + "/v1", api_key="synthetic-provider-key")
             key: Final = scenario.key(models=[model])
+            _send(candidate, model, key, f"{marker}-held")
+            eventually(flushing.is_set, bool, seconds=30)
             _send(candidate, model, key, f"{marker}-huge")
-            _send_and_wait_until_landed(candidate, model, key, sink, f"{marker}-sibling")
+            _send(candidate, model, key, f"{marker}-sibling")
+            release.set()
+            eventually(sink.landed_ids, lambda landed: f"{marker}-sibling" in landed, seconds=60)
             _send_and_wait_until_landed(candidate, model, key, sink, f"{marker}-second-flush")
             _send_and_wait_until_landed(candidate, model, key, sink, f"{marker}-third-flush")
-    assert sum(1 for r in provider.drain() if r.method == "POST") == 4
+    assert sum(1 for r in provider.drain() if r.method == "POST") == 5
+    assert sink.landed_ids() == {f"{marker}-{name}" for name in ("held", "sibling", "second-flush", "third-flush")}
     assert sink.rejected_attempts == 1, (
         f"an EntityTooLarge object was PUT {sink.rejected_attempts} times next to delivered siblings; "
         "with the default s3_drop_on_terminal_error it must be attempted once and dropped"

tests/integration/providers/test_bedrock_runtime_chat_completions_sad_wire.py

@@ -27,6 +27,7 @@ BAD_KEY: Final = "sk-synthetic-bad-key"
 NATIVE_TARGET: Final = "/openai/v1/chat/completions"
 CONVERSE_TARGET: Final = f"/model/{GPT}/converse"
 LONG_VERSION_GPT: Final = "openai.gpt-" + "1" * 30000
+LONG_UNVERSIONED_GPT: Final = "openai.gpt-x" + "1" * 30000
 PNG_DATA_URL: Final = (
     "data:image/png;base64,"
     "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg=="
@@ -258,11 +259,12 @@ def test_thirty_thousand_digit_version_is_classified_quickly_and_served_by_conve
     marker: Final = uuid.uuid4().hex
     control_marker: Final = uuid.uuid4().hex
     with wire_server(respond) as wire, gateway.scenario() as scenario, ThreadPoolExecutor(max_workers=1) as pool:
-        control_model: Final = _deployment(scenario, wire)
+        control_model: Final = _deployment(scenario, wire, model=f"bedrock/{LONG_UNVERSIONED_GPT}")
         model: Final = _deployment(scenario, wire, model=f"bedrock/{LONG_VERSION_GPT}")
         control: Final = timed(lambda: _chat(gateway, control_model, control_marker))
         assert _content(control.value) == answer(control_marker), control.value.text
-        _only_request(wire, control_marker)
+        control_request: Final = _only_request(wire, control_marker)
+        assert target_of(control_request) == f"/model/{LONG_UNVERSIONED_GPT}/converse", control_request.target
         liveliness: Final = pool.submit(_timed_liveliness, gateway)
         served: Final = timed(lambda: _chat(gateway, model, marker))
         response: Final = served.value

tests/integration/providers/test_openai_dialect_prompt_cache_breakpoint_owned_proxy.py

@@ -2,8 +2,8 @@ import asyncio
 import re
 import threading
 from collections.abc import Callable, Iterator
-from concurrent.futures import ThreadPoolExecutor
-from dataclasses import dataclass
+from concurrent.futures import Future, ThreadPoolExecutor
+from dataclasses import dataclass, replace
 from pathlib import Path
 from queue import SimpleQueue
 from types import MappingProxyType
@@ -15,7 +15,7 @@ import psutil
 import pytest
 from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment
 from integration._support.database import read_rows
-from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
+from integration._support.process import OwnedProxy, WorkerHealthcheck, graceful_stop_seconds, owned_proxy_process
 from integration._support.responses_vendor import newest_marker
 from integration._support.wire import Reply, Request, Wire, wire_server
 from integration.providers._cache_control_marks_support import owned_config
@@ -27,6 +27,7 @@ from integration.providers._mantle_gpt_prompt_cache_support import (
     SYSTEM_POINT,
     TOKEN,
     Outcome,
+    Step,
     assert_answered,
     assert_burst_landed,
     assert_wire,
@@ -60,7 +61,6 @@ _CELL_TIMEOUT: Final = int(2 * graceful_stop_seconds() + 120)
 _RESTART_TIMEOUT: Final = int(4 * graceful_stop_seconds() + 240)
 _HELD_BURST: Final = 20
 _SPREAD_BURST: Final = 12
-_SPREAD_ATTEMPTS: Final = 20
 _MANTLE_NAME: Final = "mantle-gpt-cache-owned"
 
 
@@ -122,7 +122,9 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
     release.set()
     held: Final[SimpleQueue[str]] = SimpleQueue()
     with gateway_from_environment() as environment, wire_server(_peer(release, held)) as wire:
-        with owned_proxy_process(environment, directory, _overrides(wire), workers=2) as owned:
+        with owned_proxy_process(
+            environment, directory, _overrides(wire), workers=2, healthcheck=WorkerHealthcheck.LENIENT
+        ) as owned:
             eventually(lambda: len(_started_workers(owned.log)), lambda count: count == 2, seconds=60)
             wire.drain()
             yield _Rig(owned.gateway, wire, owned, release, held)
@@ -204,38 +206,48 @@ def test_b11_foundry_gpt6_chat_stays_on_the_plain_chat_wire(rig: _Rig) -> None:
         success_row(name, marker)
 
 
-def _spread_burst(rig: _Rig, name: str, workers: tuple[int, ...]) -> dict[int, int]:
-    plan: Final = plan_burst(_SPREAD_BURST)
-    _drain(rig.held)
-    rig.wire.drain()
-    rig.release.clear()
-    with ThreadPoolExecutor(max_workers=_SPREAD_BURST) as pool:
-        futures: Final = tuple(
-            pool.submit(send, rig.gateway, endpoint, request_body(endpoint, name, prompt_text(marker), stream=stream))
-            for endpoint, stream, marker in plan
-        )
-        eventually(rig.held.qsize, lambda size: size == _SPREAD_BURST, seconds=60)
-        held_by: Final = {pid: _open_peer_connections(pid, rig.wire.url) for pid in workers}
-        rig.release.set()
-        outcomes: Final = tuple(future.result() for future in futures)
-    assert sum(held_by.values()) == _SPREAD_BURST, held_by
-    assert_burst_landed(
-        rig.wire, name, tuple((marker, outcome) for (_, _, marker), outcome in zip(plan, outcomes)), marked=True
-    )
-    return held_by
-
-
 @pytest.mark.timeout(_CELL_TIMEOUT)
 def test_e6_every_worker_of_a_two_worker_proxy_marks_the_mantle_request(rig: _Rig) -> None:
-    with rig.gateway.scenario() as scenario:
+    plan: Final = plan_burst(_SPREAD_BURST)
+    half: Final = len(plan) // 2
+    with (
+        rig.gateway.scenario() as scenario,
+        httpx.Client(
+            base_url=str(rig.gateway.client.base_url),
+            timeout=60,
+            trust_env=False,
+            limits=httpx.Limits(max_keepalive_connections=0),
+        ) as unpooled,
+        ThreadPoolExecutor(max_workers=len(plan)) as pool,
+    ):
         name: Final = mantle_deployment(rig.gateway, scenario, rig.wire)
-        workers: Final = _live_workers(rig.owned.log)
-        assert len(workers) == 2, workers
-        for _ in range(_SPREAD_ATTEMPTS):
-            if all(count > 0 for count in _spread_burst(rig, name, workers).values()):
-                break
-        else:
-            raise AssertionError("One worker never took a marked request")
+        first, second = sorted(_live_workers(rig.owned.log))
+        gateway: Final = replace(rig.gateway, client=unpooled)
+        _drain(rig.held)
+        rig.wire.drain()
+        rig.release.clear()
+
+        def submit(part: tuple[Step, ...]) -> tuple[Future[Outcome], ...]:
+            return tuple(
+                pool.submit(send, gateway, endpoint, request_body(endpoint, name, prompt_text(marker), stream=stream))
+                for endpoint, stream, marker in part
+            )
+
+        try:
+            with rig.owned.suspended(second):
+                on_first: Final = submit(plan[:half])
+                eventually(rig.held.qsize, lambda size: size == half, seconds=60)
+            with rig.owned.suspended(first):
+                on_second: Final = submit(plan[half:])
+                eventually(rig.held.qsize, lambda size: size == len(plan), seconds=60)
+            held_by: Final = {pid: _open_peer_connections(pid, rig.wire.url) for pid in (first, second)}
+        finally:
+            rig.release.set()
+        outcomes: Final = tuple(future.result() for future in (*on_first, *on_second))
+        assert held_by == {first: half, second: len(plan) - half}, held_by
+        assert_burst_landed(
+            rig.wire, name, tuple((marker, outcome) for (_, _, marker), outcome in zip(plan, outcomes)), marked=True
+        )
 
 
 def _mantle_config(wire: Wire, directory: Path, **litellm_params: JsonValue) -> Path:

tests/integration/providers/test_responses_bridge_tool_search_chaos.py

@@ -529,7 +529,7 @@ def _marked_posts(received: tuple[Request, ...], marker: str) -> tuple[Request,
 
 
 @pytest.mark.timeout(_OWNED_PROXY_CELL_SECONDS)
-def test_a_previous_response_id_turn_replays_the_function_call_without_the_search_call(owned_rig: _OwnedRig) -> None:
+def test_a_previous_response_id_turn_replays_the_search_call_beside_the_function_call(owned_rig: _OwnedRig) -> None:
     marker: Final = uuid.uuid4().hex
     owned_rig.wire.drain()
     proxy: Final = owned_rig.proxy.gateway
@@ -566,11 +566,8 @@ def test_a_previous_response_id_turn_replays_the_function_call_without_the_searc
         _assert_forwarded_with_tool_search(opening)
         _assert_forwarded_with_tool_search(replay)
         messages: Final = rv.ITEMS.validate_python(rv.JSON_OBJECT.validate_json(replay.body)["messages"])
-        assert "srvtoolu_" not in json.dumps(messages) and "tool_search" not in json.dumps(messages), messages
         (assistant,) = [turn for turn in messages if turn["role"] == "assistant"]
-        blocks: Final = rv.ITEMS.validate_python(assistant["content"])
-        tool_uses: Final = [(block["type"], block.get("id")) for block in blocks if block["type"] == "tool_use"]
-        assert tool_uses == [("tool_use", f"toolu_{marker}")], blocks
+        assert assistant["content"] == _message(marker)["content"], assistant
         assert [block["tool_use_id"] for block in _tool_results(messages)] == [f"toolu_{marker}"], messages
 
 

tests/integration/routing/test_messages_stream_retry_budget_sources_owned_proxy.py

@@ -1,7 +1,9 @@
 import json
+import threading
 import uuid
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass
+from itertools import groupby, islice
 from pathlib import Path
 from typing import Final, Literal
 
@@ -12,6 +14,7 @@ from integration._support.anthropic_sse import (
     LIFECYCLE,
     PING,
     Attempts,
+    SseEvent,
     delta_text,
     dropping_reply,
     error_body,
@@ -55,7 +58,9 @@ def _served_id(backend: str, marker: str, attempt: int) -> str:
     return f"msg_{backend}_{marker}_a{attempt}"
 
 
-def _primary_reply(behavior: str, attempt: int, full: tuple[bytes, bytes, bytes, bytes]) -> Reply:
+def _primary_reply(
+    behavior: str, attempt: int, full: tuple[bytes, bytes, bytes, bytes], content_gate: threading.Event
+) -> Reply:
     dropped: Final = dropping_reply(full, abort_after=1)
     match behavior:
         case "drop-once":
@@ -63,7 +68,13 @@ def _primary_reply(behavior: str, attempt: int, full: tuple[bytes, bytes, bytes,
         case "drop-always":
             return dropped
         case "hold-ping":
-            return stream_reply((full[0], PING, full[1] + full[2] + full[3]), pause=0.25)
+            return Reply(
+                content_type="text/event-stream",
+                chunks=(full[0], full[1] + full[2] + full[3]),
+                gate_after_first=content_gate,
+                heartbeat=PING,
+                pause_between_chunks=0.25,
+            )
         case "drop-then-too-long":
             if attempt == 1:
                 return dropped
@@ -81,7 +92,7 @@ def _fallback_reply(behavior: str, full: tuple[bytes, bytes, bytes, bytes]) -> R
     return stream_reply(full)
 
 
-def _respond(attempts: Attempts) -> Callable[[Request], Reply]:
+def _respond(attempts: Attempts, content_gate: threading.Event) -> Callable[[Request], Reply]:
     def respond(request: Request) -> Reply:
         assert (request.method, request.target) == ("POST", "/v1/messages"), request
         assert request.headers["x-api-key"] == _API_KEY, request.headers
@@ -93,11 +104,21 @@ def _respond(attempts: Attempts) -> Callable[[Request], Reply]:
         full: Final = message_stream(_served_id(backend, marker, attempt), backend, _TEXT)
         if backend != _PRIMARY:
             return _fallback_reply(behavior, full)
-        return _primary_reply(behavior, attempt, full)
+        return _primary_reply(behavior, attempt, full, content_gate)
 
     return respond
 
 
+def _stream_body(model: str, behavior: Behavior, marker: str, **extra: JsonValue) -> dict[str, JsonValue]:
+    return {
+        "model": model,
+        "max_tokens": 16,
+        "stream": True,
+        "messages": [{"role": "user", "content": f"{behavior}:{marker}"}],
+        **extra,
+    }
+
+
 def _deployment(name: str, backend: str, **extra: JsonValue) -> dict[str, JsonValue]:
     return {
         "model_name": name,
@@ -137,19 +158,10 @@ def _config(wire: Wire, directory: Path) -> Path:
 class _Rig:
     proxy: Gateway
     attempts: Attempts
+    content_gate: threading.Event
 
     def stream(self, model: str, behavior: Behavior, marker: str, **extra: JsonValue) -> httpx.Response:
-        return self.proxy.request(
-            "POST",
-            "/v1/messages",
-            {
-                "model": model,
-                "max_tokens": 16,
-                "stream": True,
-                "messages": [{"role": "user", "content": f"{behavior}:{marker}"}],
-                **extra,
-            },
-        )
+        return self.proxy.request("POST", "/v1/messages", _stream_body(model, behavior, marker, **extra))
 
     def attempts_on(self, backend: str, marker: str) -> int:
         return self.attempts.count(f"{backend}:{marker}")
@@ -158,10 +170,14 @@ class _Rig:
 @pytest.fixture(scope="module")
 def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
     attempts: Final = Attempts()
+    content_gate: Final = threading.Event()
     directory: Final = tmp_path_factory.mktemp("messages-retry-budget-sources")
-    with gateway_from_environment() as gateway, wire_server(_respond(attempts)) as wire:
+    with gateway_from_environment() as gateway, wire_server(_respond(attempts, content_gate)) as wire:
         with owned_proxy(gateway, directory, {}, config=_config(wire, directory), workers=2) as proxy:
-            yield _Rig(proxy, attempts)
+            try:
+                yield _Rig(proxy, attempts, content_gate)
+            finally:
+                content_gate.set()
 
 
 def _assert_completed(response: httpx.Response, served_id: str) -> None:
@@ -202,12 +218,32 @@ def test_request_num_retries_zero_turns_the_router_budget_off(rig: _Rig) -> None
     assert rig.attempts_on(_PRIMARY, marker) == 1
 
 
+def _live_events(lines: Iterator[str]) -> Iterator[SseEvent]:
+    for has_text, block in groupby(lines, key=bool):
+        if has_text:
+            yield from parse_sse("\n".join(block))
+
+
 def test_lifecycle_frames_are_held_until_content_while_pings_go_out_live(rig: _Rig) -> None:
     marker: Final = uuid.uuid4().hex
-    response: Final = rig.stream(_ROUTER_BUDGET, "hold-ping", marker)
-    assert response.status_code == 200, response.text
-    events: Final = parse_sse(response.text)
-    assert event_types(events) == ("ping", *LIFECYCLE), events
+    rig.content_gate.clear()
+    try:
+        with rig.proxy.client.stream(
+            "POST",
+            "/v1/messages",
+            json=_stream_body(_ROUTER_BUDGET, "hold-ping", marker),
+            headers={"Authorization": f"Bearer {rig.proxy.key}"},
+        ) as response:
+            assert response.status_code == 200, response.read()
+            arriving: Final = _live_events(response.iter_lines())
+            before_content: Final = tuple(islice(arriving, 2))
+            rig.content_gate.set()
+            after_content: Final = tuple(arriving)
+    finally:
+        rig.content_gate.set()
+    assert event_types(before_content) == ("ping", "ping"), before_content
+    events: Final = before_content + after_content
+    assert tuple(kind for kind in event_types(events) if kind != "ping") == LIFECYCLE, events
     assert message_id(events) == _served_id(_PRIMARY, marker, 1), events
     assert delta_text(events) == _TEXT, events
     assert rig.attempts_on(_PRIMARY, marker) == 1

tests/integration/spend/test_disable_end_user_cost_tracking.py

@@ -695,6 +695,7 @@ def test_flag_on_burst_lands_each_row_once_through_sink_outage_and_worker_kill(r
         before_kill: Final = _burst(rig, models, key, calls[: len(_SURFACES) * 2])
         _assert_burst_landed_once(rig, models, before_kill)
         assert not rig.lens.outages.acquire(blocking=False), "the Lens sink outage was never hit"
+        _team_requests_flushed(team, len(before_kill))
         victim: Final = min(rig.proxy.worker_pids())
         replacement: Final = rig.proxy.kill_worker(victim)
         assert replacement != victim, (victim, replacement)

tests/integration/streaming/test_faked_responses_stream_chaos.py

@@ -24,7 +24,7 @@ from integration._support.client import (
     string_value,
 )
 from integration._support.database import read_rows
-from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
+from integration._support.process import OwnedProxy, WorkerHealthcheck, graceful_stop_seconds, owned_proxy_process
 from integration._support.wire import Reply, Request, wire_server
 from integration._support.ports import allocate_port
 from integration.streaming.faked_responses_case import (
@@ -111,7 +111,14 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
     port: Final = allocate_port()
     with (
         gateway_from_environment() as shared,
-        owned_proxy_process(shared, directory, {}, config=_config(port, directory), workers=2) as owned,
+        owned_proxy_process(
+            shared,
+            directory,
+            {},
+            config=_config(port, directory),
+            workers=2,
+            healthcheck=WorkerHealthcheck.LENIENT,
+        ) as owned,
     ):
         yield _Rig(port, owned)
 
@@ -306,9 +313,11 @@ class _Held:
         return _respond(request)
 
 
-async def _held_burst(gateway: Gateway, calls: tuple[_Call, ...], held: _Held) -> asyncio.Task[tuple[_Served, ...]]:
+async def _held_burst(
+    gateway: Gateway, calls: tuple[_Call, ...], held: _Held, *, already_held: int = 0
+) -> asyncio.Task[tuple[_Served, ...]]:
     burst: Final = asyncio.create_task(_burst(gateway, calls, tolerate_transport_errors=True))
-    await asyncio.to_thread(eventually, held.markers.qsize, lambda size: size == len(calls), 60)
+    await asyncio.to_thread(eventually, held.markers.qsize, lambda size: size == already_held + len(calls), 60)
     return burst
 
 
@@ -319,12 +328,18 @@ def _held_by(rig: _Rig, upstream: str) -> Mapping[int, int]:
 async def test_held_mixed_burst_spreads_over_both_workers_and_each_value_arrives_once(rig: _Rig) -> None:
     calls: Final = _calls(30)
     held: Final = _Held(threading.Event(), SimpleQueue())
+    first, second = sorted(rig.proxy.worker_pids())
     with wire_server(held.respond, port=rig.port) as wire:
-        burst: Final = await _held_burst(rig.gateway, calls, held)
-        held_by: Final = _held_by(rig, wire.url)
-        held.release.set()
-        served: Final = await burst
-        assert len(held_by) == 2 and all(held_by.values()) and sum(held_by.values()) == 30, held_by
+        try:
+            with rig.proxy.suspended(second):
+                on_first: Final = await _held_burst(rig.gateway, calls[:15], held)
+            with rig.proxy.suspended(first):
+                on_second: Final = await _held_burst(rig.gateway, calls[15:], held, already_held=15)
+            held_by: Final = _held_by(rig, wire.url)
+        finally:
+            held.release.set()
+        served: Final = await on_first + await on_second
+        assert held_by == {first: 15, second: 15}, held_by
         assert len(served) == 30
         for item in served:
             _assert_each_value_once(item)