From a572ed6d8f8a096ec28480edc8badeb88f6278c6 Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Fri, 9 Oct 2026 21:57:41 +0900
Subject: [PATCH 01/10] fix(server): dispatch a lone idle request without the
 coalesce window

An idle worker dispatches a request that stays alone immediately instead
of holding it for idle_coalesce_ms. A one-item arrival inside that window
is still the tail of a staggered burst and coalesces, and several items
already pending when a batch is selected still fuse. The idle 1ms poll is
now a wake-up event that submit() sets.

The staggered-burst test changed because #373 deliberately stops charging
the idle window to a request that arrives alone, and a burst that is
already queued still fuses.

Refs #373
---
 .../src/sie_server/config/engine.py           |  15 +--
 .../sie_server/core/worker/model_worker.py    |  84 +++++++++----
 .../src/sie_server/core/worker/types.py       |  17 ++-
 .../sie_server/tests/core/test_worker_core.py | 114 ++++++++++++++----
 .../tests/core/test_worker_drain.py           |  16 ++-
 5 files changed, 179 insertions(+), 67 deletions(-)

diff --git a/packages/sie_server/src/sie_server/config/engine.py b/packages/sie_server/src/sie_server/config/engine.py
index ac3aede46..1be488b1a 100644
--- a/packages/sie_server/src/sie_server/config/engine.py
+++ b/packages/sie_server/src/sie_server/config/engine.py
@@ -303,13 +303,14 @@ class EngineConfig(BaseSettings):
         Field(
             ge=0.0,
             description=(
-                "Accumulation window for the FIRST batch after a worker was "
-                "idle (#2874). An idle worker used to dispatch immediately "
-                "with whatever was pending, shredding bursty arrivals into a "
-                "train of small serialized forwards. With this window the "
-                "batcher coalesces until arrivals have been quiet for this "
-                "long (bounded by max_batch_wait_ms), so a lone request waits "
-                "at most this window. 0 restores immediate idle dispatch."
+                "Tail window for a staggered burst that reaches an idle "
+                "worker (#2874, #373). A request that arrives alone is "
+                "dispatched immediately and is not held for this long. One "
+                "later item within this gap of the previous submit coalesces "
+                "with what follows it, and several items already pending "
+                "still form one batch. Bounded by max_batch_wait_ms. 0 "
+                "dispatches every idle arrival immediately, which shreds a "
+                "staggered burst."
             ),
         ),
     ] = 3.0
diff --git a/packages/sie_server/src/sie_server/core/worker/model_worker.py b/packages/sie_server/src/sie_server/core/worker/model_worker.py
index 09bdc8270..d10c76cdd 100644
--- a/packages/sie_server/src/sie_server/core/worker/model_worker.py
+++ b/packages/sie_server/src/sie_server/core/worker/model_worker.py
@@ -285,6 +285,14 @@ def __init__(
         self._running = False
         self._stopping = False  # True when graceful stop has begun
         self._process_task: asyncio.Task[None] | None = None
+        # Wakes ``_get_next_batch_fcfs`` when a direct submit lands, instead of
+        # polling every 1 ms. ``stop()`` sets it too so an idle loop unblocks.
+        self._work_available = asyncio.Event()
+        # Previous and latest direct-submit times (monotonic seconds). 0 means
+        # no arrival yet. The gap decides whether an idle one-item arrival is
+        # the tail of a staggered burst (#373) or a request that stays alone.
+        self._prev_arrival_at = 0.0
+        self._last_arrival_at = 0.0
         self._stats = WorkerStats()
 
         # Batches currently between "extracted from a batcher" and
@@ -406,6 +414,7 @@ async def stop(self) -> None:
             return
 
         self._running = False
+        self._work_available.set()
 
         try:
             if self._process_task is not None:
@@ -983,6 +992,11 @@ async def _submit_to_batcher(
         """
         batcher = self._get_batcher(lora)
         await batcher.submit_many([(item, metadata) for item in prepared_items])
+        now = time.monotonic()
+        if self._last_arrival_at > 0.0:
+            self._prev_arrival_at = self._last_arrival_at
+        self._last_arrival_at = now
+        self._work_available.set()
         self._ensure_process_loop_started()
         return metadata.future
 
@@ -1082,19 +1096,19 @@ async def _get_next_batch_fcfs(
         Selects the batcher whose first pending request has waited the longest.
         This ensures fairness across LoRAs - no LoRA starves even with low traffic.
 
-        When the worker was idle (had to poll for requests), a short
-        accumulation window (``idle_coalesce_ms``, #2874) caps the coalesce
-        wait instead of dispatching immediately: bursty arrivals at an idle
-        worker fuse into one batch rather than degenerating into a train of
-        small serialized forwards, while a lone request only waits the small
-        cap once arrivals stop. Setting ``idle_coalesce_ms=0`` restores the
-        legacy immediate dispatch.
+        When the worker was idle, a request that is still alone is dispatched
+        immediately (#373). It is not held for ``idle_coalesce_ms``. One new
+        item whose previous arrival is inside that window is the tail of a
+        staggered burst and still coalesces, so the first of the burst may run
+        alone and the rest fuse behind it. Several items already pending when
+        a batch is selected form one batch. ``idle_coalesce_ms=0`` disables
+        the tail window and dispatches every idle arrival immediately.
 
         Args:
             was_idle: Whether the worker was idle before this call. When True,
-                dispatches after the capped idle accumulation window (or
-                immediately when the window is 0). When False, uses the
-                normal timeout/coalesce mechanism to accumulate a proper
+                a lone request with no recent peer dispatches immediately;
+                a staggered tail uses the capped idle window. When False, uses
+                the normal timeout/coalesce mechanism to accumulate a proper
                 batch.
 
         Returns:
@@ -1103,6 +1117,9 @@ async def _get_next_batch_fcfs(
             (was truly idle).
         """
         while True:
+            if not self._running:
+                return None, FormattedBatch(items=[], metadata=[], total_cost=0), was_idle
+
             oldest_lora: str | None = None
             oldest_time: float = float("inf")
 
@@ -1117,21 +1134,42 @@ async def _get_next_batch_fcfs(
             if oldest_lora is not None or (oldest_lora is None and self._batchers[None].pending_count > 0):
                 # Found a batcher with pending items - get batch from it
                 selected_lora = oldest_lora if oldest_lora is not None else None
+                selected = self._batchers[selected_lora]
                 idle_window_ms = self._config.idle_coalesce_ms
-                if was_idle and idle_window_ms > 0:
-                    batch = await self._batchers[selected_lora].get_batch(coalesce_cap_ms=idle_window_ms)
+                # A request that stays alone must not sleep the idle window.
+                # Holding is only for a one-item tail after a recent peer, or
+                # for several items already queued (they fuse, and the cap
+                # can still catch a straggler). ``idle_coalesce_ms=0`` skips
+                # both and dispatches immediately.
+                lone_without_peer = (
+                    was_idle
+                    and idle_window_ms > 0
+                    and selected.pending_count == 1
+                    and not self._follows_recent_arrival(idle_window_ms)
+                )
+                if was_idle and idle_window_ms > 0 and not lone_without_peer:
+                    batch = await selected.get_batch(coalesce_cap_ms=idle_window_ms)
                 else:
-                    batch = await self._batchers[selected_lora].get_batch(immediate=was_idle)
+                    batch = await selected.get_batch(immediate=was_idle)
                 return selected_lora, batch, was_idle
 
-            # No batchers have pending items - worker is idle
+            # No batchers have pending items - worker is idle until submit().
             was_idle = True
-            await asyncio.sleep(0.001)
+            await self._work_available.wait()
+            self._work_available.clear()
 
-            # Check if we should stop
-            if not self._running:
-                # Return empty batch to exit gracefully
-                return None, FormattedBatch(items=[], metadata=[], total_cost=0), was_idle
+    def _follows_recent_arrival(self, idle_window_ms: float) -> bool:
+        """True when the latest direct submit is inside the idle tail window.
+
+        ``_prev_arrival_at`` is the arrival before the one that woke this
+        selection. A first arrival, or one that shows up after the window has
+        gone quiet, is not a tail and must dispatch without waiting (#373).
+        """
+        previous = self._prev_arrival_at
+        latest = self._last_arrival_at
+        if previous <= 0.0 or latest < previous:
+            return False
+        return (latest - previous) * 1000.0 <= idle_window_ms
 
     async def _drain_active_batcher(self, active_lora: str | None) -> bool:
         """Continuous-batching drain of ``active_lora``'s batcher.
@@ -1186,10 +1224,10 @@ async def _process_loop(self) -> None:
             await self._process_loop_concurrent()
             return
 
-        # Track idle state across iterations. When idle, the next batch is
-        # dispatched immediately (low-concurrency optimization). When busy
-        # (just finished inference + drain), we let BatchFormer's
-        # timeout/coalesce mechanism accumulate a proper batch.
+        # Track idle state across iterations. An idle worker dispatches a
+        # request that stays alone immediately (#373); a staggered tail still
+        # uses the idle window. After a busy batch, BatchFormer's
+        # timeout/coalesce mechanism accumulates the next batch.
         was_idle = True
 
         while self._running:
diff --git a/packages/sie_server/src/sie_server/core/worker/types.py b/packages/sie_server/src/sie_server/core/worker/types.py
index a40fe0eb6..34c7fae00 100644
--- a/packages/sie_server/src/sie_server/core/worker/types.py
+++ b/packages/sie_server/src/sie_server/core/worker/types.py
@@ -199,15 +199,14 @@ class WorkerConfig:
     # instead of being shredded into several half-full ones.
     coalesce_ms: float = 15.0
     coalesce_ratio: float = 0.5
-    # Idle-dispatch accumulation window (#2874). An idle worker used to
-    # dispatch IMMEDIATELY with whatever was pending, so a burst arriving at
-    # an idle worker degenerated into a train of small serialized forwards
-    # (many batch-of-1 dispatches instead of one fused batch). With this
-    # window, the first batch after idleness coalesces arrivals until the
-    # queue has been quiet for ``idle_coalesce_ms`` (or the batch fills, or
-    # ``max_batch_wait_ms`` elapses since the first request). A lone request
-    # therefore waits at most this window — single-digit ms — before
-    # dispatch, and ``0`` restores the legacy immediate dispatch.
+    # Idle-dispatch tail window (#2874, #373). A request that arrives alone
+    # is dispatched immediately — it is not held for this long. One later
+    # item that arrives within this gap of the previous submit is the tail
+    # of a staggered burst and coalesces (the head of that burst may already
+    # have run alone). Several items already pending when a batch is
+    # selected still form one batch. ``0`` disables the tail window and
+    # dispatches every idle arrival immediately, which shreds a staggered
+    # burst into one forward per request.
     idle_coalesce_ms: float = 3.0
     max_queue_size: int = 1000  # Maximum pending items in queue (0 = unlimited)
     instrumentation: bool = False
diff --git a/packages/sie_server/tests/core/test_worker_core.py b/packages/sie_server/tests/core/test_worker_core.py
index d85682e22..09afdee6d 100644
--- a/packages/sie_server/tests/core/test_worker_core.py
+++ b/packages/sie_server/tests/core/test_worker_core.py
@@ -76,8 +76,8 @@ def test_defaults(self) -> None:
         assert config.max_batch_wait_ms == 15
         assert config.coalesce_ms == 15.0
         assert config.coalesce_ratio == 0.5
-        # Idle accumulation window (#2874): single-digit ms by default so an
-        # idle worker fuses bursts without materially delaying lone requests.
+        # Tail window for a staggered burst (#2874, #373). Default stays
+        # non-zero; a lone request is not charged it.
         assert config.idle_coalesce_ms == 3.0
 
     def test_custom_values(self) -> None:
@@ -586,11 +586,13 @@ async def test_idle_dispatch_no_wait(self, mock_adapter: MagicMock, tokenized_it
 
     @pytest.mark.asyncio
     async def test_staggered_burst_fuses_with_idle_window(self, mock_adapter: MagicMock) -> None:
-        """#2874: a burst arriving at an IDLE worker fuses into one batch.
+        """#373: a request that arrives alone is not charged the idle window.
 
-        Before the idle accumulation window, an idle worker dispatched
-        immediately with whatever was pending, so staggered arrivals became a
-        train of serialized batch-of-1 forwards (the was_idle degeneracy).
+        The head of a time-staggered burst runs immediately, even when
+        ``idle_coalesce_ms`` is far larger than the gap. A later arrival
+        inside the window is the tail and still coalesces with what follows
+        it. A burst that is already queued still fuses
+        (``test_already_queued_burst_fuses``).
         """
         call_sizes: list[int] = []
 
@@ -607,28 +609,91 @@ def counting_encode(items, output_types, **kwargs):
             max_batch_tokens=1000,
             max_batch_requests=64,
             max_batch_wait_ms=500,
-            coalesce_ms=200,
-            coalesce_ratio=0.5,
-            idle_coalesce_ms=60,
+            coalesce_ms=400,
+            coalesce_ratio=1.0,
+            idle_coalesce_ms=200,
         )
         worker = ModelWorker(mock_adapter, config)
         await worker.start()
 
         try:
+            started = time.monotonic()
+            first = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 0")],
+                ["dense"],
+            )
+            await asyncio.wait_for(first, timeout=2.0)
+            head_ms = (time.monotonic() - started) * 1000
+            # Must not have slept the 200ms window. 50ms leaves room for CI noise.
+            assert head_ms < 50, f"lone head waited {head_ms:.1f}ms"
+            assert call_sizes == [1], f"expected the head to run alone, got: {call_sizes}"
+
             futures = []
-            for i in range(4):
+            for i in range(1, 4):
                 future = await worker.submit(
                     [make_text_item([1, 2], 0)],
                     [Item(text=f"text {i}")],
                     ["dense"],
                 )
                 futures.append(future)
-                # Stagger arrivals well inside the 60ms idle window.
+                # Gaps well inside the 200ms tail window.
                 await asyncio.sleep(0.005)
 
             await asyncio.gather(*futures)
 
-            # All four staggered requests fused into a single forward.
+            assert sum(call_sizes) == 4
+            assert call_sizes == [1, 3], f"expected the tail to fuse, got: {call_sizes}"
+
+        finally:
+            await worker.stop()
+
+    @pytest.mark.asyncio
+    async def test_already_queued_burst_fuses(self, mock_adapter: MagicMock) -> None:
+        """Several items already pending when an idle worker selects still fuse.
+
+        Queued before the process loop runs, so selection sees the whole
+        burst rather than a request that arrived alone (#373).
+        """
+        call_sizes: list[int] = []
+
+        def counting_encode(items, output_types, **kwargs):
+            call_sizes.append(len(items))
+            return EncodeOutput(
+                dense=np.array([[0.1, 0.2, 0.3]] * len(items)),
+                batch_size=len(items),
+            )
+
+        mock_adapter.encode.side_effect = counting_encode
+
+        config = WorkerConfig(
+            max_batch_tokens=1000,
+            max_batch_requests=64,
+            max_batch_wait_ms=500,
+            coalesce_ms=400,
+            coalesce_ratio=1.0,
+            idle_coalesce_ms=200,
+        )
+        worker = ModelWorker(mock_adapter, config)
+        await worker.start()
+        start_loop = worker._ensure_process_loop_started
+        worker._ensure_process_loop_started = lambda: None  # type: ignore[method-assign]
+
+        try:
+            futures = []
+            for i in range(4):
+                future = await worker.submit(
+                    [make_text_item([1, 2], 0)],
+                    [Item(text=f"text {i}")],
+                    ["dense"],
+                )
+                futures.append(future)
+            assert worker.pending_count == 4
+
+            del worker._ensure_process_loop_started
+            start_loop()
+            await asyncio.gather(*futures)
+
             assert sum(call_sizes) == 4
             assert call_sizes == [4], f"expected one fused batch, got: {call_sizes}"
 
@@ -637,11 +702,11 @@ def counting_encode(items, output_types, **kwargs):
 
     @pytest.mark.asyncio
     async def test_staggered_burst_shreds_without_idle_window(self, mock_adapter: MagicMock) -> None:
-        """Contrast for the test above: ``idle_coalesce_ms=0`` restores the
-        legacy immediate idle dispatch, and the same staggered burst shreds
-        into multiple small forwards. This is the mutation the idle window
-        exists to kill — if the window stops being applied, the fusing test
-        above degrades into THIS shape and fails.
+        """``idle_coalesce_ms=0`` does not coalesce a staggered tail.
+
+        Every idle arrival dispatches immediately, so the same stagger as
+        ``test_staggered_burst_fuses_with_idle_window`` cannot land in one
+        forward. This is the escape hatch, not the default.
         """
         call_sizes: list[int] = []
 
@@ -687,9 +752,12 @@ def counting_encode(items, output_types, **kwargs):
             await worker.stop()
 
     @pytest.mark.asyncio
-    async def test_lone_request_waits_only_the_idle_window(self, mock_adapter: MagicMock) -> None:
-        """A lone request must not wait the busy-path coalesce/batch windows —
-        only the small idle accumulation window bounds its dispatch latency.
+    async def test_lone_request_does_not_wait_the_idle_window(self, mock_adapter: MagicMock) -> None:
+        """A request that stays alone is not held for ``idle_coalesce_ms`` (#373).
+
+        The busy-path coalesce and batch windows are also not charged. The
+        bound is loose on purpose: a hard 1ms assert flakes under CI noise,
+        but it must finish far below the 200ms idle window.
         """
         mock_adapter.encode.return_value = EncodeOutput(
             dense=np.array([[0.1, 0.2, 0.3]]),
@@ -702,7 +770,7 @@ async def test_lone_request_waits_only_the_idle_window(self, mock_adapter: Magic
             max_batch_wait_ms=500,
             coalesce_ms=400,
             coalesce_ratio=1.0,
-            idle_coalesce_ms=5,
+            idle_coalesce_ms=200,
         )
         worker = ModelWorker(mock_adapter, config)
         await worker.start()
@@ -717,9 +785,7 @@ async def test_lone_request_waits_only_the_idle_window(self, mock_adapter: Magic
             await asyncio.wait_for(future, timeout=2.0)
             elapsed_ms = (time.monotonic() - start) * 1000
 
-            # ~5ms idle window + mock inference; far below the 400ms coalesce
-            # and 500ms batch windows.
-            assert elapsed_ms < 200
+            assert elapsed_ms < 50, f"lone request waited {elapsed_ms:.1f}ms"
 
         finally:
             await worker.stop()
diff --git a/packages/sie_server/tests/core/test_worker_drain.py b/packages/sie_server/tests/core/test_worker_drain.py
index 320b0aca1..71548f56e 100644
--- a/packages/sie_server/tests/core/test_worker_drain.py
+++ b/packages/sie_server/tests/core/test_worker_drain.py
@@ -46,9 +46,10 @@ def _parked_config() -> WorkerConfig:
     """Config whose batcher never yields a batch during a test.
 
     Every dispatch trigger (cost limit, request limit, first-request timeout,
-    coalesce window, idle accumulation window) is pushed far out of reach, so
-    submitted items stay queued in the ``BatchFormer`` for the whole test and
-    the drain is the only thing that can complete them. This is the state a
+    coalesce window, idle tail window) is pushed far out of reach.
+    ``_submit`` seeds a prior arrival so a one-item submit takes the tail
+    path (#373) and stays queued in the ``BatchFormer`` for the whole test.
+    The drain is the only thing that can complete it. This is the state a
     real worker is in whenever it is evicted with a backlog.
     """
     return WorkerConfig(
@@ -128,7 +129,14 @@ async def _submit(
     count: int = 1,
     options: dict[str, Any] | None = None,
 ) -> asyncio.Future[WorkerResult]:
-    """Queue one request of ``count`` items and let the process loop park."""
+    """Queue one request of ``count`` items and let the process loop park.
+
+    A lone idle request is dispatched without waiting ``idle_coalesce_ms``
+    (#373). Seeding a just-prior arrival makes this submit a staggered tail
+    (or, when ``count > 1``, several items already queued) so the parked
+    window still holds it inside ``get_batch``.
+    """
+    worker._last_arrival_at = time.monotonic()
     future = await worker.submit(
         [_item(i) for i in range(count)],
         [Item(text=f"hello-{i}") for i in range(count)],

From 8dfac7bedc9b796bb3fa7bd793558d4904de4696 Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Sat, 10 Oct 2026 05:24:37 +0900
Subject: [PATCH 02/10] fix(server): treat a multi-item idle request as one
 arrival

Refs #373
---
 .../src/sie_server/config/engine.py           |  18 +-
 .../sie_server/src/sie_server/core/batcher.py |  15 +-
 .../sie_server/core/worker/model_worker.py    |  74 +++-----
 .../src/sie_server/core/worker/types.py       |  15 +-
 .../sie_server/tests/core/test_worker_core.py | 172 ++++++++++++++++--
 .../tests/core/test_worker_drain.py           |  37 ++--
 6 files changed, 235 insertions(+), 96 deletions(-)

diff --git a/packages/sie_server/src/sie_server/config/engine.py b/packages/sie_server/src/sie_server/config/engine.py
index 1be488b1a..0acab55d6 100644
--- a/packages/sie_server/src/sie_server/config/engine.py
+++ b/packages/sie_server/src/sie_server/config/engine.py
@@ -303,14 +303,16 @@ class EngineConfig(BaseSettings):
         Field(
             ge=0.0,
             description=(
-                "Tail window for a staggered burst that reaches an idle "
-                "worker (#2874, #373). A request that arrives alone is "
-                "dispatched immediately and is not held for this long. One "
-                "later item within this gap of the previous submit coalesces "
-                "with what follows it, and several items already pending "
-                "still form one batch. Bounded by max_batch_wait_ms. 0 "
-                "dispatches every idle arrival immediately, which shreds a "
-                "staggered burst."
+                "Tail window for a burst that reaches an idle worker "
+                "(#2874, #373). One request pending on a batcher is "
+                "dispatched immediately, however many items it carries, and "
+                "is not held for this long. A follow-up waits only while "
+                "another request on that same batcher is still pending. A "
+                "finished request does not hold the next one, and another "
+                "LoRA's arrival does not hold this batcher. Several requests "
+                "already pending still form one batch. Bounded by "
+                "max_batch_wait_ms. 0 dispatches every idle arrival "
+                "immediately."
             ),
         ),
     ] = 3.0
diff --git a/packages/sie_server/src/sie_server/core/batcher.py b/packages/sie_server/src/sie_server/core/batcher.py
index c4a5cf4ca..f49c1b2f9 100644
--- a/packages/sie_server/src/sie_server/core/batcher.py
+++ b/packages/sie_server/src/sie_server/core/batcher.py
@@ -237,9 +237,22 @@ def config(self) -> BatchConfig:
 
     @property
     def pending_count(self) -> int:
-        """Return number of pending requests."""
+        """Return number of pending items.
+
+        One submitted request occupies one entry per item.
+        """
         return len(self._pending)
 
+    @property
+    def pending_request_count(self) -> int:
+        """Return number of distinct pending requests.
+
+        ``submit_many`` appends one entry per item and reuses one metadata
+        object, so ``pending_count`` counts items. Callers that mean "one
+        request" (idle dispatch, #373) use this.
+        """
+        return len({id(request.metadata) for request in self._pending})
+
     @property
     def pending_cost(self) -> int:
         """Return total cost in pending requests."""
diff --git a/packages/sie_server/src/sie_server/core/worker/model_worker.py b/packages/sie_server/src/sie_server/core/worker/model_worker.py
index d10c76cdd..9c6d9778f 100644
--- a/packages/sie_server/src/sie_server/core/worker/model_worker.py
+++ b/packages/sie_server/src/sie_server/core/worker/model_worker.py
@@ -285,14 +285,8 @@ def __init__(
         self._running = False
         self._stopping = False  # True when graceful stop has begun
         self._process_task: asyncio.Task[None] | None = None
-        # Wakes ``_get_next_batch_fcfs`` when a direct submit lands, instead of
-        # polling every 1 ms. ``stop()`` sets it too so an idle loop unblocks.
+        # ``stop()`` sets this too so an idle ``_get_next_batch_fcfs`` unblocks.
         self._work_available = asyncio.Event()
-        # Previous and latest direct-submit times (monotonic seconds). 0 means
-        # no arrival yet. The gap decides whether an idle one-item arrival is
-        # the tail of a staggered burst (#373) or a request that stays alone.
-        self._prev_arrival_at = 0.0
-        self._last_arrival_at = 0.0
         self._stats = WorkerStats()
 
         # Batches currently between "extracted from a batcher" and
@@ -992,10 +986,6 @@ async def _submit_to_batcher(
         """
         batcher = self._get_batcher(lora)
         await batcher.submit_many([(item, metadata) for item in prepared_items])
-        now = time.monotonic()
-        if self._last_arrival_at > 0.0:
-            self._prev_arrival_at = self._last_arrival_at
-        self._last_arrival_at = now
         self._work_available.set()
         self._ensure_process_loop_started()
         return metadata.future
@@ -1096,20 +1086,22 @@ async def _get_next_batch_fcfs(
         Selects the batcher whose first pending request has waited the longest.
         This ensures fairness across LoRAs - no LoRA starves even with low traffic.
 
-        When the worker was idle, a request that is still alone is dispatched
-        immediately (#373). It is not held for ``idle_coalesce_ms``. One new
-        item whose previous arrival is inside that window is the tail of a
-        staggered burst and still coalesces, so the first of the burst may run
-        alone and the rest fuse behind it. Several items already pending when
-        a batch is selected form one batch. ``idle_coalesce_ms=0`` disables
-        the tail window and dispatches every idle arrival immediately.
+        When the worker was idle, one pending request is dispatched immediately
+        (#373), however many items it carries. It is not held for
+        ``idle_coalesce_ms``. A follow-up is a tail only while another request
+        on that same batcher is still pending (arrived and not yet dispatched);
+        the idle window can then catch a straggler. A finished request is no
+        longer pending, so it does not hold the next one, and another LoRA's
+        arrival does not hold this batcher. Several requests already pending
+        when a batch is selected form one batch. ``idle_coalesce_ms=0``
+        disables the tail window and dispatches every idle arrival immediately.
 
         Args:
             was_idle: Whether the worker was idle before this call. When True,
-                a lone request with no recent peer dispatches immediately;
-                a staggered tail uses the capped idle window. When False, uses
-                the normal timeout/coalesce mechanism to accumulate a proper
-                batch.
+                one pending request dispatches immediately and two or more on
+                the selected batcher use the capped idle window. When False,
+                uses the normal timeout/coalesce mechanism to accumulate a
+                proper batch.
 
         Returns:
             Tuple of (lora_name, batch, was_idle) where lora_name is None for
@@ -1136,18 +1128,11 @@ async def _get_next_batch_fcfs(
                 selected_lora = oldest_lora if oldest_lora is not None else None
                 selected = self._batchers[selected_lora]
                 idle_window_ms = self._config.idle_coalesce_ms
-                # A request that stays alone must not sleep the idle window.
-                # Holding is only for a one-item tail after a recent peer, or
-                # for several items already queued (they fuse, and the cap
-                # can still catch a straggler). ``idle_coalesce_ms=0`` skips
-                # both and dispatches immediately.
-                lone_without_peer = (
-                    was_idle
-                    and idle_window_ms > 0
-                    and selected.pending_count == 1
-                    and not self._follows_recent_arrival(idle_window_ms)
-                )
-                if was_idle and idle_window_ms > 0 and not lone_without_peer:
+                # Alone is one request, not one item: ``submit_many`` queues
+                # an entry per item. A second request still sitting on this
+                # batcher is the only tail. ``idle_coalesce_ms=0`` skips it.
+                alone = selected.pending_request_count == 1
+                if was_idle and idle_window_ms > 0 and not alone:
                     batch = await selected.get_batch(coalesce_cap_ms=idle_window_ms)
                 else:
                     batch = await selected.get_batch(immediate=was_idle)
@@ -1158,19 +1143,6 @@ async def _get_next_batch_fcfs(
             await self._work_available.wait()
             self._work_available.clear()
 
-    def _follows_recent_arrival(self, idle_window_ms: float) -> bool:
-        """True when the latest direct submit is inside the idle tail window.
-
-        ``_prev_arrival_at`` is the arrival before the one that woke this
-        selection. A first arrival, or one that shows up after the window has
-        gone quiet, is not a tail and must dispatch without waiting (#373).
-        """
-        previous = self._prev_arrival_at
-        latest = self._last_arrival_at
-        if previous <= 0.0 or latest < previous:
-            return False
-        return (latest - previous) * 1000.0 <= idle_window_ms
-
     async def _drain_active_batcher(self, active_lora: str | None) -> bool:
         """Continuous-batching drain of ``active_lora``'s batcher.
 
@@ -1224,10 +1196,10 @@ async def _process_loop(self) -> None:
             await self._process_loop_concurrent()
             return
 
-        # Track idle state across iterations. An idle worker dispatches a
-        # request that stays alone immediately (#373); a staggered tail still
-        # uses the idle window. After a busy batch, BatchFormer's
-        # timeout/coalesce mechanism accumulates the next batch.
+        # Track idle state across iterations. An idle worker dispatches one
+        # pending request immediately (#373). Two or more still pending on
+        # the selected batcher use the idle window. After a busy batch,
+        # BatchFormer's timeout/coalesce mechanism accumulates the next batch.
         was_idle = True
 
         while self._running:
diff --git a/packages/sie_server/src/sie_server/core/worker/types.py b/packages/sie_server/src/sie_server/core/worker/types.py
index 34c7fae00..359623d99 100644
--- a/packages/sie_server/src/sie_server/core/worker/types.py
+++ b/packages/sie_server/src/sie_server/core/worker/types.py
@@ -199,14 +199,13 @@ class WorkerConfig:
     # instead of being shredded into several half-full ones.
     coalesce_ms: float = 15.0
     coalesce_ratio: float = 0.5
-    # Idle-dispatch tail window (#2874, #373). A request that arrives alone
-    # is dispatched immediately — it is not held for this long. One later
-    # item that arrives within this gap of the previous submit is the tail
-    # of a staggered burst and coalesces (the head of that burst may already
-    # have run alone). Several items already pending when a batch is
-    # selected still form one batch. ``0`` disables the tail window and
-    # dispatches every idle arrival immediately, which shreds a staggered
-    # burst into one forward per request.
+    # Idle-dispatch tail window (#2874, #373). One request pending on an
+    # idle batcher is dispatched immediately, however many items it carries,
+    # and is not held for this long. A follow-up waits only while another
+    # request on that same batcher is still pending. A finished request does
+    # not hold the next one, and another LoRA's arrival does not hold this
+    # batcher. Several requests already pending when a batch is selected
+    # still form one batch. ``0`` dispatches every idle arrival immediately.
     idle_coalesce_ms: float = 3.0
     max_queue_size: int = 1000  # Maximum pending items in queue (0 = unlimited)
     instrumentation: bool = False
diff --git a/packages/sie_server/tests/core/test_worker_core.py b/packages/sie_server/tests/core/test_worker_core.py
index 09afdee6d..4a48f0d64 100644
--- a/packages/sie_server/tests/core/test_worker_core.py
+++ b/packages/sie_server/tests/core/test_worker_core.py
@@ -586,13 +586,12 @@ async def test_idle_dispatch_no_wait(self, mock_adapter: MagicMock, tokenized_it
 
     @pytest.mark.asyncio
     async def test_staggered_burst_fuses_with_idle_window(self, mock_adapter: MagicMock) -> None:
-        """#373: a request that arrives alone is not charged the idle window.
+        """#373: the head runs alone, and a still-pending tail fuses behind it.
 
-        The head of a time-staggered burst runs immediately, even when
-        ``idle_coalesce_ms`` is far larger than the gap. A later arrival
-        inside the window is the tail and still coalesces with what follows
-        it. A burst that is already queued still fuses
-        (``test_already_queued_burst_fuses``).
+        The head is not charged ``idle_coalesce_ms``. A finished request does
+        not hold the next one. Two requests queued on the same batcher are a
+        tail, so the idle window holds them and a third arrival inside the
+        window joins that forward: two forwards, not one per request.
         """
         call_sizes: list[int] = []
 
@@ -629,18 +628,28 @@ def counting_encode(items, output_types, **kwargs):
             assert head_ms < 50, f"lone head waited {head_ms:.1f}ms"
             assert call_sizes == [1], f"expected the head to run alone, got: {call_sizes}"
 
-            futures = []
-            for i in range(1, 4):
-                future = await worker.submit(
-                    [make_text_item([1, 2], 0)],
-                    [Item(text=f"text {i}")],
-                    ["dense"],
-                )
-                futures.append(future)
-                # Gaps well inside the 200ms tail window.
-                await asyncio.sleep(0.005)
+            # The head has finished, so it must not hold these. Queue two
+            # before the loop selects, then one more inside the 200ms window.
+            second = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 1")],
+                ["dense"],
+            )
+            third = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 2")],
+                ["dense"],
+            )
+            assert worker.pending_count == 2
+            await asyncio.sleep(0.01)
+            assert worker.pending_count == 2, "tail was dispatched instead of held"
+            fourth = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 3")],
+                ["dense"],
+            )
 
-            await asyncio.gather(*futures)
+            await asyncio.gather(second, third, fourth)
 
             assert sum(call_sizes) == 4
             assert call_sizes == [1, 3], f"expected the tail to fuse, got: {call_sizes}"
@@ -790,6 +799,135 @@ async def test_lone_request_does_not_wait_the_idle_window(self, mock_adapter: Ma
         finally:
             await worker.stop()
 
+    @pytest.mark.asyncio
+    async def test_multi_item_request_does_not_wait_the_idle_window(self, mock_adapter: MagicMock) -> None:
+        """One submit of two items is one request and skips ``idle_coalesce_ms`` (#373)."""
+        call_sizes: list[int] = []
+
+        def counting_encode(items, output_types, **kwargs):
+            call_sizes.append(len(items))
+            return EncodeOutput(
+                dense=np.array([[0.1, 0.2, 0.3]] * len(items)),
+                batch_size=len(items),
+            )
+
+        mock_adapter.encode.side_effect = counting_encode
+
+        config = WorkerConfig(
+            max_batch_tokens=1000,
+            max_batch_requests=64,
+            max_batch_wait_ms=500,
+            coalesce_ms=400,
+            coalesce_ratio=1.0,
+            idle_coalesce_ms=200,
+        )
+        worker = ModelWorker(mock_adapter, config)
+        await worker.start()
+
+        try:
+            start = time.monotonic()
+            future = await worker.submit(
+                [make_text_item([1, 2], 0), make_text_item([3, 4], 1)],
+                [Item(text="hello-0"), Item(text="hello-1")],
+                ["dense"],
+            )
+            await asyncio.wait_for(future, timeout=2.0)
+            elapsed_ms = (time.monotonic() - start) * 1000
+
+            assert elapsed_ms < 50, f"multi-item request waited {elapsed_ms:.1f}ms"
+            assert call_sizes == [2]
+
+        finally:
+            await worker.stop()
+
+    @pytest.mark.asyncio
+    async def test_sequential_request_does_not_wait_the_idle_window(self, mock_adapter: MagicMock) -> None:
+        """A finished request does not make the next submit wait (#373).
+
+        The second request is sent as soon as the first has been awaited, well
+        inside ``idle_coalesce_ms``. It is not a tail: nothing is still pending
+        on this batcher.
+        """
+        mock_adapter.encode.return_value = EncodeOutput(
+            dense=np.array([[0.1, 0.2, 0.3]]),
+            batch_size=1,
+        )
+
+        config = WorkerConfig(
+            max_batch_tokens=1000,
+            max_batch_requests=64,
+            max_batch_wait_ms=500,
+            coalesce_ms=400,
+            coalesce_ratio=1.0,
+            idle_coalesce_ms=200,
+        )
+        worker = ModelWorker(mock_adapter, config)
+        await worker.start()
+
+        try:
+            first = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="first")],
+                ["dense"],
+            )
+            await asyncio.wait_for(first, timeout=2.0)
+
+            start = time.monotonic()
+            second = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="second")],
+                ["dense"],
+            )
+            await asyncio.wait_for(second, timeout=2.0)
+            elapsed_ms = (time.monotonic() - start) * 1000
+
+            assert elapsed_ms < 50, f"sequential request waited {elapsed_ms:.1f}ms"
+
+        finally:
+            await worker.stop()
+
+    @pytest.mark.asyncio
+    async def test_other_lora_arrival_does_not_hold_this_batcher(self, mock_adapter: MagicMock) -> None:
+        """A different LoRA's arrival does not make this batcher wait (#373)."""
+        mock_adapter.encode.return_value = EncodeOutput(
+            dense=np.array([[0.1, 0.2, 0.3]]),
+            batch_size=1,
+        )
+
+        config = WorkerConfig(
+            max_batch_tokens=1000,
+            max_batch_requests=64,
+            max_batch_wait_ms=500,
+            coalesce_ms=400,
+            coalesce_ratio=1.0,
+            idle_coalesce_ms=200,
+        )
+        worker = ModelWorker(mock_adapter, config)
+        await worker.start()
+
+        try:
+            other = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="lora")],
+                ["dense"],
+                options={"lora": "adapter-a"},
+            )
+            await asyncio.wait_for(other, timeout=2.0)
+
+            start = time.monotonic()
+            base = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="base")],
+                ["dense"],
+            )
+            await asyncio.wait_for(base, timeout=2.0)
+            elapsed_ms = (time.monotonic() - start) * 1000
+
+            assert elapsed_ms < 50, f"other LoRA held this batcher for {elapsed_ms:.1f}ms"
+
+        finally:
+            await worker.stop()
+
     @pytest.mark.asyncio
     async def test_concurrent_requests_still_batch(self, mock_adapter: MagicMock) -> None:
         """Concurrent requests are still batched together when worker is busy."""
diff --git a/packages/sie_server/tests/core/test_worker_drain.py b/packages/sie_server/tests/core/test_worker_drain.py
index 71548f56e..c9c109ba3 100644
--- a/packages/sie_server/tests/core/test_worker_drain.py
+++ b/packages/sie_server/tests/core/test_worker_drain.py
@@ -46,9 +46,9 @@ def _parked_config() -> WorkerConfig:
     """Config whose batcher never yields a batch during a test.
 
     Every dispatch trigger (cost limit, request limit, first-request timeout,
-    coalesce window, idle tail window) is pushed far out of reach.
-    ``_submit`` seeds a prior arrival so a one-item submit takes the tail
-    path (#373) and stays queued in the ``BatchFormer`` for the whole test.
+    coalesce window, idle tail window) is pushed far out of reach. A lone
+    idle request is still dispatched immediately (#373), so ``_submit`` queues
+    a second request on the same batcher and the tail window holds the burst.
     The drain is the only thing that can complete it. This is the state a
     real worker is in whenever it is evicted with a backlog.
     """
@@ -123,6 +123,12 @@ def _item(index: int = 0) -> TextPreparedItem:
     return make_text_item([1, 2, 3, 4, 5], index)
 
 
+def _retrieve_exception(future: asyncio.Future[Any]) -> None:
+    """Retrieve a parked partner's error so a drained future does not warn."""
+    if not future.cancelled():
+        future.exception()
+
+
 async def _submit(
     worker: ModelWorker,
     *,
@@ -131,18 +137,24 @@ async def _submit(
 ) -> asyncio.Future[WorkerResult]:
     """Queue one request of ``count`` items and let the process loop park.
 
-    A lone idle request is dispatched without waiting ``idle_coalesce_ms``
-    (#373). Seeding a just-prior arrival makes this submit a staggered tail
-    (or, when ``count > 1``, several items already queued) so the parked
-    window still holds it inside ``get_batch``.
+    A lone idle request, including one submit of several items, is dispatched
+    without waiting ``idle_coalesce_ms`` (#373). A second request on the same
+    batcher is submitted through the public API before the loop selects, so
+    the burst is a tail and the parked window holds both inside ``get_batch``.
     """
-    worker._last_arrival_at = time.monotonic()
     future = await worker.submit(
         [_item(i) for i in range(count)],
         [Item(text=f"hello-{i}") for i in range(count)],
         ["dense"],
         options=options,
     )
+    partner = await worker.submit(
+        [_item(count)],
+        [Item(text="park")],
+        ["dense"],
+        options=options,
+    )
+    partner.add_done_callback(_retrieve_exception)
     # Give the process loop a chance to start and block on the batcher, so
     # the drain is exercised against a live (not merely un-started) loop.
     await asyncio.sleep(0)
@@ -157,7 +169,8 @@ async def test_queued_request_fails_with_retryable_error(self, mock_adapter: Mag
         worker = ModelWorker(mock_adapter, _parked_config(), model_name="test-model")
         await worker.start()
         future = await _submit(worker)
-        assert worker.pending_count == 1
+        # The request under test plus the one-item request that holds the window.
+        assert worker.pending_count == 2
         assert not future.done()
 
         await worker.stop()
@@ -178,7 +191,8 @@ async def test_no_item_is_left_pending_in_any_batcher(self, mock_adapter: MagicM
         lora_a = await _submit(worker, options={"lora": "adapter-a"})
         lora_b = await _submit(worker, options={"lora": "adapter-b"})
         assert set(worker._batchers) == {None, "adapter-a", "adapter-b"}
-        assert worker.pending_count == 3
+        # Each batcher has the request under test and the request that holds it.
+        assert worker.pending_count == 6
 
         await worker.stop()
 
@@ -199,7 +213,8 @@ async def test_multi_item_request_fails_its_single_future_once(self, mock_adapte
         worker = ModelWorker(mock_adapter, _parked_config(), model_name="test-model")
         await worker.start()
         future = await _submit(worker, count=4)
-        assert worker.pending_count == 4
+        # Four items share the future under test; one more item holds the window.
+        assert worker.pending_count == 5
 
         await worker.stop()
 

From de5cb672198e0a29d1858bcd9d3e5d7e355a47eb Mon Sep 17 00:00:00 2001
From: YeonwooSung <neos960518@gmail.com>
Date: Sat, 10 Oct 2026 23:21:42 +0900
Subject: [PATCH 03/10] test(server): contrast idle-window fusion with the
 queued-tail burst

The same already-queued tail shreds when idle_coalesce_ms is 0.
The GPU before/after measurement is still outstanding.

Refs #373
---
 .../sie_server/core/worker/model_worker.py    |  8 +--
 .../sie_server/tests/core/test_worker_core.py | 63 +++++++++++++------
 2 files changed, 47 insertions(+), 24 deletions(-)

diff --git a/packages/sie_server/src/sie_server/core/worker/model_worker.py b/packages/sie_server/src/sie_server/core/worker/model_worker.py
index 9c6d9778f..188971616 100644
--- a/packages/sie_server/src/sie_server/core/worker/model_worker.py
+++ b/packages/sie_server/src/sie_server/core/worker/model_worker.py
@@ -1129,10 +1129,10 @@ async def _get_next_batch_fcfs(
                 selected = self._batchers[selected_lora]
                 idle_window_ms = self._config.idle_coalesce_ms
                 # Alone is one request, not one item: ``submit_many`` queues
-                # an entry per item. A second request still sitting on this
-                # batcher is the only tail. ``idle_coalesce_ms=0`` skips it.
-                alone = selected.pending_request_count == 1
-                if was_idle and idle_window_ms > 0 and not alone:
+                # an entry per item. The distinct-request count is only the
+                # idle tail check. A busy worker, and ``idle_coalesce_ms=0``,
+                # dispatch without consulting it.
+                if was_idle and idle_window_ms > 0 and selected.pending_request_count > 1:
                     batch = await selected.get_batch(coalesce_cap_ms=idle_window_ms)
                 else:
                     batch = await selected.get_batch(immediate=was_idle)
diff --git a/packages/sie_server/tests/core/test_worker_core.py b/packages/sie_server/tests/core/test_worker_core.py
index 4a48f0d64..33b9ec6f6 100644
--- a/packages/sie_server/tests/core/test_worker_core.py
+++ b/packages/sie_server/tests/core/test_worker_core.py
@@ -711,11 +711,14 @@ def counting_encode(items, output_types, **kwargs):
 
     @pytest.mark.asyncio
     async def test_staggered_burst_shreds_without_idle_window(self, mock_adapter: MagicMock) -> None:
-        """``idle_coalesce_ms=0`` does not coalesce a staggered tail.
-
-        Every idle arrival dispatches immediately, so the same stagger as
-        ``test_staggered_burst_fuses_with_idle_window`` cannot land in one
-        forward. This is the escape hatch, not the default.
+        """``idle_coalesce_ms=0`` does not hold the queued tail the window fuses.
+
+        Same shape as ``test_staggered_burst_fuses_with_idle_window``: the
+        head runs alone, two requests are already queued before the loop
+        selects, and one more arrives 10ms later. With the window off that
+        pair is dispatched immediately, so the later arrival does not join
+        it (``[1, 2, 1]``). With the window on the same tail is held and
+        fuses (``[1, 3]``). This is the escape hatch, not the default.
         """
         call_sizes: list[int] = []
 
@@ -732,30 +735,50 @@ def counting_encode(items, output_types, **kwargs):
             max_batch_tokens=1000,
             max_batch_requests=64,
             max_batch_wait_ms=500,
-            coalesce_ms=200,
-            coalesce_ratio=0.5,
+            coalesce_ms=400,
+            coalesce_ratio=1.0,
             idle_coalesce_ms=0,
         )
         worker = ModelWorker(mock_adapter, config)
         await worker.start()
 
         try:
-            futures = []
-            for i in range(4):
-                future = await worker.submit(
-                    [make_text_item([1, 2], 0)],
-                    [Item(text=f"text {i}")],
-                    ["dense"],
-                )
-                futures.append(future)
-                await asyncio.sleep(0.005)
+            started = time.monotonic()
+            first = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 0")],
+                ["dense"],
+            )
+            await asyncio.wait_for(first, timeout=2.0)
+            head_ms = (time.monotonic() - started) * 1000
+            assert head_ms < 50, f"lone head waited {head_ms:.1f}ms"
+            assert call_sizes == [1], f"expected the head to run alone, got: {call_sizes}"
 
-            await asyncio.gather(*futures)
+            # Same already-queued tail as the fusing test. The window is 0,
+            # so the pair must leave before the later arrival instead of being held.
+            second = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 1")],
+                ["dense"],
+            )
+            third = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 2")],
+                ["dense"],
+            )
+            assert worker.pending_count == 2
+            await asyncio.sleep(0.01)
+            assert worker.pending_count == 0, "tail was held instead of dispatched"
+            fourth = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="text 3")],
+                ["dense"],
+            )
+
+            await asyncio.gather(second, third, fourth)
 
             assert sum(call_sizes) == 4
-            # The first arrival dispatches alone (immediate idle dispatch), so
-            # the burst cannot land in a single forward.
-            assert len(call_sizes) >= 2, f"expected shredded batches without the window, got: {call_sizes}"
+            assert call_sizes == [1, 2, 1], f"expected the queued tail to shred, got: {call_sizes}"
 
         finally:
             await worker.stop()

From f556494c8b70a6b25cb1d27acdadf1040d337218 Mon Sep 17 00:00:00 2001
From: svonava <svonava@gmail.com>
Date: Fri, 16 Oct 2026 14:17:40 -0700
Subject: [PATCH 04/10] fix(worker): admit runnable requests before idle
 dispatch

---
 .../sie_server/core/worker/model_worker.py    | 11 +++-
 .../sie_server/tests/core/test_worker_core.py | 57 +++++++++++++++++++
 2 files changed, 67 insertions(+), 1 deletion(-)

diff --git a/packages/sie_server/src/sie_server/core/worker/model_worker.py b/packages/sie_server/src/sie_server/core/worker/model_worker.py
index 96f660f4f..c9964bda6 100644
--- a/packages/sie_server/src/sie_server/core/worker/model_worker.py
+++ b/packages/sie_server/src/sie_server/core/worker/model_worker.py
@@ -1093,7 +1093,9 @@ async def _get_next_batch_fcfs(
 
         When the worker was idle, one pending request is dispatched immediately
         (#373), however many items it carries. It is not held for
-        ``idle_coalesce_ms``. A follow-up is a tail only while another request
+        ``idle_coalesce_ms``. A lone head yields one event-loop turn so already
+        runnable submissions can join it, without waiting for a timer. A
+        follow-up is a tail only while another request
         on that same batcher is still pending (arrived and not yet dispatched);
         the idle window can then catch a straggler. A finished request is no
         longer pending, so it does not hold the next one, and another LoRA's
@@ -1133,6 +1135,13 @@ async def _get_next_batch_fcfs(
                 selected_lora = oldest_lora if oldest_lora is not None else None
                 selected = self._batchers[selected_lora]
                 idle_window_ms = self._config.idle_coalesce_ms
+                if was_idle and idle_window_ms > 0 and selected.pending_request_count == 1:
+                    # Let submissions already runnable on this loop join the
+                    # head before deciding whether it is a lone request. This
+                    # yields once without waiting for a timer or later arrival.
+                    await asyncio.sleep(0)
+                    if not self._running:
+                        return None, FormattedBatch(items=[], metadata=[], total_cost=0), was_idle
                 # Alone is one request, not one item: ``submit_many`` queues
                 # an entry per item. The distinct-request count is only the
                 # idle tail check. A busy worker, and ``idle_coalesce_ms=0``,
diff --git a/packages/sie_server/tests/core/test_worker_core.py b/packages/sie_server/tests/core/test_worker_core.py
index 33b9ec6f6..498bbfca1 100644
--- a/packages/sie_server/tests/core/test_worker_core.py
+++ b/packages/sie_server/tests/core/test_worker_core.py
@@ -657,6 +657,63 @@ def counting_encode(items, output_types, **kwargs):
         finally:
             await worker.stop()
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("warmup", [False, True])
+    async def test_ready_submissions_join_the_idle_head(self, mock_adapter: MagicMock, warmup: bool) -> None:
+        """A runnable submit joins the head without adding a fixed idle delay."""
+        call_sizes: list[int] = []
+
+        def counting_encode(items, output_types, **kwargs):
+            call_sizes.append(len(items))
+            return EncodeOutput(
+                dense=np.array([[0.1, 0.2, 0.3]] * len(items)),
+                batch_size=len(items),
+            )
+
+        mock_adapter.encode.side_effect = counting_encode
+        worker = ModelWorker(
+            mock_adapter,
+            WorkerConfig(
+                max_batch_tokens=1000,
+                max_batch_requests=64,
+                max_batch_wait_ms=500,
+                coalesce_ms=400,
+                coalesce_ratio=1.0,
+                idle_coalesce_ms=3,
+            ),
+        )
+        await worker.start()
+
+        async def submit_ready():
+            return await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="second")],
+                ["dense"],
+            )
+
+        try:
+            if warmup:
+                warmed = await worker.submit(
+                    [make_text_item([1, 2], 0)],
+                    [Item(text="warmup")],
+                    ["dense"],
+                )
+                await asyncio.wait_for(warmed, timeout=2.0)
+                call_sizes.clear()
+
+            first = await worker.submit(
+                [make_text_item([1, 2], 0)],
+                [Item(text="first")],
+                ["dense"],
+            )
+            # The process loop is already runnable ahead of this submit. Both
+            # are ready before this task suspends; no timed arrival is required.
+            second = await asyncio.create_task(submit_ready())
+            await asyncio.wait_for(asyncio.gather(first, second), timeout=2.0)
+            assert call_sizes == [2]
+        finally:
+            await worker.stop()
+
     @pytest.mark.asyncio
     async def test_already_queued_burst_fuses(self, mock_adapter: MagicMock) -> None:
         """Several items already pending when an idle worker selects still fuse.

From 7ad3294a37eee17f2239fc5e10c4d2c51b965342 Mon Sep 17 00:00:00 2001
From: svonava <svonava@gmail.com>
Date: Fri, 16 Oct 2026 11:29:18 -0700
Subject: [PATCH 05/10] test(server): add paired GPU idle-worker benchmark

---
 tools/benchmarks/idle_worker/.gitignore  |   2 +
 tools/benchmarks/idle_worker/README.md   |  57 +++++
 tools/benchmarks/idle_worker/__init__.py |   0
 tools/benchmarks/idle_worker/prepare.py  |  49 ++++
 tools/benchmarks/idle_worker/run.py      | 287 +++++++++++++++++++++++
 tools/benchmarks/idle_worker/server.py   |  50 ++++
 tools/benchmarks/idle_worker/test_run.py |  41 ++++
 7 files changed, 486 insertions(+)
 create mode 100644 tools/benchmarks/idle_worker/.gitignore
 create mode 100644 tools/benchmarks/idle_worker/README.md
 create mode 100644 tools/benchmarks/idle_worker/__init__.py
 create mode 100644 tools/benchmarks/idle_worker/prepare.py
 create mode 100644 tools/benchmarks/idle_worker/run.py
 create mode 100644 tools/benchmarks/idle_worker/server.py
 create mode 100644 tools/benchmarks/idle_worker/test_run.py

diff --git a/tools/benchmarks/idle_worker/.gitignore b/tools/benchmarks/idle_worker/.gitignore
new file mode 100644
index 000000000..c36187138
--- /dev/null
+++ b/tools/benchmarks/idle_worker/.gitignore
@@ -0,0 +1,2 @@
+_inputs/
+results/
diff --git a/tools/benchmarks/idle_worker/README.md b/tools/benchmarks/idle_worker/README.md
new file mode 100644
index 000000000..ba6fa0789
--- /dev/null
+++ b/tools/benchmarks/idle_worker/README.md
@@ -0,0 +1,57 @@
+# Idle-worker scheduling comparison
+
+This benchmark compares two exact SIE source revisions on one NVIDIA GPU. It
+uses `sie_sdk.SIEClient` against the local Python worker HTTP endpoint, records
+the actual `X-Queue-Time` headers, and observes worker batch sizes. It targets
+the Python worker scheduler changed in [PR #658](https://github.com/superlinked/sie/pull/658).
+It does not measure the queue-only gateway/sidecar path, whose preformed batches
+have a different scheduling owner.
+
+The default workload uses the catalog-pinned `knowledgator/gliclass-large-v1.0`
+checkpoint, four synthetic support requests, and five fixed classification
+labels. Both arms use their checked-in default model profile and scheduling
+configuration. The archive includes the server source, model catalog and bundle
+definitions. Preparation refuses different dependency locks.
+
+From an initialized checkout:
+
+```sh
+mise exec -- python tools/benchmarks/idle_worker/prepare.py \
+  --baseline b0c26b02d21dbd146646c41b2b4f42627a0c9ebd \
+  --candidate 0649b2834f9c6c075f386516705d3179675575d4
+```
+
+On the GPU machine, with the same locked workspace and prepared `_inputs/`:
+
+```sh
+mise exec -- python tools/benchmarks/idle_worker/run.py \
+  --output /tmp/sie-idle-worker-results
+```
+
+The default order is baseline/candidate/candidate/baseline/baseline/candidate.
+Each fresh server warms request concurrency 1, 8 and 32, then measures:
+
+- 100 sequential requests, including every queue-time sample;
+- four simultaneous HTTP bursts each at concurrency 8 and 32;
+- 12 seconds of closed-loop traffic each at concurrency 8 and 32, including
+  the final outstanding requests in elapsed time.
+
+HTTP bursts can reach the worker at different times: distinguish an already
+queued burst from the PR's intended first-request/tail split. The batch trace
+records item IDs so the observed grouping is inspectable. A wrapper records a
+batch only after normal dispatch completes; it does not replace scheduling or
+inference. The existing worker unit tests cover deterministic queued-burst
+formation.
+
+Each completed phase is written before the next starts. Batch records are
+flushed as JSONL after each dispatch, so SIGTERM during server shutdown cannot
+discard them. If a trace is missing, request timings remain available and the
+trial is marked `timing_complete_trace_missing`.
+
+Use an otherwise idle GPU and reserve enough CPU for the client and server.
+The server uses four Torch CPU threads. Keep the GPU allocation, dependencies,
+input texts and warmup identical for both arms; compare all repeated trials
+before interpreting a small throughput change. The output includes source and
+archive hashes, runtime versions, GPU identity, raw sequential headers, HTTP
+latencies, batch traces, and each server's log. Generated snapshots and results
+are ignored by Git.
diff --git a/tools/benchmarks/idle_worker/__init__.py b/tools/benchmarks/idle_worker/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tools/benchmarks/idle_worker/prepare.py b/tools/benchmarks/idle_worker/prepare.py
new file mode 100644
index 000000000..025aba197
--- /dev/null
+++ b/tools/benchmarks/idle_worker/prepare.py
@@ -0,0 +1,49 @@
+"""Snapshot two Git revisions for a same-machine idle-worker comparison."""
+
+import argparse
+import hashlib
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+INPUTS = Path(__file__).parent / "_inputs"
+PATHS = ("packages/sie_server/src", "packages/sie_server/models", "packages/sie_server/bundles")
+GIT = shutil.which("git")
+
+
+def git(*args: str) -> bytes:
+    if GIT is None:
+        raise RuntimeError("git is required")
+    return subprocess.check_output([GIT, *args], cwd=ROOT)  # noqa: S603 — local Git revisions, no shell.
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--baseline", required=True)
+    parser.add_argument("--candidate", required=True)
+    args = parser.parse_args()
+    INPUTS.mkdir(exist_ok=True)
+    manifest = {}
+    locks = []
+    for arm, revision in (("baseline", args.baseline), ("candidate", args.candidate)):
+        commit = git("rev-parse", f"{revision}^{{commit}}").decode().strip()
+        archive = git("archive", "--format=tar", commit, *PATHS)
+        (INPUTS / f"{arm}.tar").write_bytes(archive)
+        lock = git("show", f"{commit}:uv.lock")
+        locks.append(lock)
+        manifest[arm] = {
+            "commit": commit,
+            "archive_sha256": hashlib.sha256(archive).hexdigest(),
+            "uv_lock_sha256": hashlib.sha256(lock).hexdigest(),
+        }
+    if locks[0] != locks[1] or locks[1] != (ROOT / "uv.lock").read_bytes():
+        raise SystemExit("Both revisions and the execution workspace must use the same uv.lock")
+    manifest["source_diff"] = git("diff", "--stat", args.baseline, args.candidate, "--", *PATHS).decode()
+    (INPUTS / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
+    print(json.dumps(manifest, indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tools/benchmarks/idle_worker/run.py b/tools/benchmarks/idle_worker/run.py
new file mode 100644
index 000000000..828922f2f
--- /dev/null
+++ b/tools/benchmarks/idle_worker/run.py
@@ -0,0 +1,287 @@
+"""Compare idle-worker HTTP latency and throughput on one GPU via SIEClient."""
+
+import argparse
+import hashlib
+import importlib.metadata
+import itertools
+import json
+import os
+import platform
+import shutil
+import socket
+import statistics
+import subprocess
+import sys
+import tarfile
+import tempfile
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from http import HTTPStatus
+from pathlib import Path
+
+import httpx
+from sie_sdk import SIEClient
+from sie_sdk.types import Item
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+LABELS = ["billing", "technical support", "account access", "sales", "other"]
+TEXTS = [
+    "The payment for my annual subscription was charged twice yesterday. Please check the invoice and refund the duplicate charge. I need the account to stay active while your team reviews the transaction.",
+    "Our data export has stopped working after the latest update. The dashboard shows the job as complete, but the downloaded file contains only the column headings. We tried again from another browser with the same result. Please help us restore the missing records before the report is due tomorrow.",
+    "I cannot log in to my account because the verification message never arrives. I checked the spam folder and confirmed that the email address is correct. My colleagues can still access their accounts. Please let me know how to recover access without creating a second account.",
+    "We are evaluating the service for a team of twenty analysts and would like information about the available subscription plans. We expect to process several thousand documents each week. Could you explain how usage is measured and whether the service supports a shared workspace for our team? We would also like to schedule a short demonstration next week.",
+]
+
+
+def distribution(values: list[float]) -> dict:
+    ordered = sorted(values)
+    return {
+        "n": len(values),
+        "median": statistics.median(values),
+        "p95": ordered[min(len(values) - 1, int(len(values) * 0.95))],
+        "max": max(values),
+    }
+
+
+def timing_sample(response: httpx.Response) -> dict[str, float] | None:
+    if not response.request.url.path.startswith("/v1/extract/") or response.status_code != HTTPStatus.OK:
+        return None
+    return {
+        key: float(response.headers[header])
+        for key, header in (
+            ("queue_ms", "X-Queue-Time"),
+            ("inference_ms", "X-Inference-Time"),
+            ("total_ms", "X-Total-Time"),
+        )
+    }
+
+
+def measure(
+    client: SIEClient, model: str, phase: str, concurrency: int, *, count: int = 0, duration: float = 0
+) -> dict:
+    barrier = threading.Barrier(concurrency)
+    counter = itertools.count()
+    start = 0.0
+    latencies: list[float] = []
+
+    def lane() -> None:
+        barrier.wait()
+        while True:
+            index = next(counter)
+            if (count and index >= count) or (duration and time.perf_counter() - start >= duration):
+                return
+            before = time.perf_counter()
+            result = client.extract(
+                model,
+                Item(text=TEXTS[index % len(TEXTS)], id=f"{phase}:{index}"),
+                labels=LABELS,
+                wait_for_capacity=False,
+                max_oom_retries=0,
+            )
+            if result.get("error"):
+                raise RuntimeError(result["error"])
+            if not result.get("classifications"):
+                raise RuntimeError("No classifications returned")
+            latencies.append((time.perf_counter() - before) * 1000)
+
+    with ThreadPoolExecutor(max_workers=concurrency) as executor:
+        start = time.perf_counter()
+        futures = [executor.submit(lane) for _ in range(concurrency)]
+        for future in futures:
+            future.result()
+    elapsed = time.perf_counter() - start
+    return {
+        "phase": phase,
+        "concurrency": concurrency,
+        "elapsed_s": elapsed,
+        "requests": len(latencies),
+        "requests_per_s": len(latencies) / elapsed,
+        "latency_ms": distribution(latencies),
+    }
+
+
+def wait_for_server(process: subprocess.Popen, port: int) -> None:
+    deadline = time.monotonic() + 300
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            raise RuntimeError(f"Server exited with {process.returncode}")
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=1):
+                return
+        except OSError:
+            time.sleep(0.25)
+    raise TimeoutError("Server did not bind within 300 seconds")
+
+
+def trial(args: argparse.Namespace, arm: str, trial_index: int, source: Path, output: Path) -> dict:
+    name = f"{trial_index:02d}-{arm}"
+    samples: list[dict] = []
+
+    def capture(response: httpx.Response) -> None:
+        if (sample := timing_sample(response)) is not None:
+            samples.append(sample)
+
+    environment = os.environ.copy()
+    environment.update(
+        {
+            "PYTHONPATH": str(source / "packages/sie_server/src") + os.pathsep + environment.get("PYTHONPATH", ""),
+            "SIE_BUNDLE": "default",
+            "SIE_TELEMETRY_DISABLED": "true",
+            "SIE_TORCH_NUM_THREADS": "4",
+            "TOKENIZERS_PARALLELISM": "false",
+            "PYTHONUNBUFFERED": "1",
+            "SIE_LOG_LEVEL": "warning",
+        }
+    )
+    trace = output / f"{name}-batches.jsonl"
+    argv = [
+        sys.executable,
+        str(HERE / "server.py"),
+        "--model",
+        args.model,
+        "--models-dir",
+        str(source / "packages/sie_server/models"),
+        "--trace",
+        str(trace),
+        "--port",
+        str(args.port),
+        "--device",
+        args.device,
+    ]
+    phases = []
+    result = {"arm": arm, "trial": trial_index, "phases": phases, "status": "running"}
+
+    def record_phase(phase: dict) -> None:
+        phases.append(phase)
+        (output / f"{name}.json").write_text(json.dumps(result, indent=2) + "\n")
+
+    with (output / f"{name}-server.log").open("w") as log:
+        process = subprocess.Popen(argv, env=environment, stdout=log, stderr=subprocess.STDOUT)  # noqa: S603 — fixed local server, no shell.
+        try:
+            wait_for_server(process, args.port)
+            base_url = f"http://127.0.0.1:{args.port}"
+            transport = httpx.Client(
+                base_url=base_url,
+                timeout=300,
+                event_hooks={"response": [capture]},
+                limits=httpx.Limits(max_connections=64, max_keepalive_connections=64),
+            )
+            with SIEClient(base_url, timeout_s=300, http_client=transport) as client:
+                for concurrency in (1, 8, 32):
+                    measure(client, args.model, f"warmup{concurrency}", concurrency, count=64)
+                samples.clear()
+                sequential = measure(client, args.model, "sequential", 1, count=args.sequential)
+                if len(samples) != args.sequential:
+                    raise RuntimeError(f"Expected {args.sequential} successful timing samples, got {len(samples)}")
+                sequential["queue_ms"] = distribution([sample["queue_ms"] for sample in samples])
+                sequential["inference_ms"] = distribution([sample["inference_ms"] for sample in samples])
+                sequential["queue_under_1ms"] = sum(sample["queue_ms"] < 1 for sample in samples)
+                sequential["samples"] = samples.copy()
+                record_phase(sequential)
+                for concurrency in (8, 32):
+                    for burst in range(4):
+                        time.sleep(0.05)
+                        record_phase(
+                            measure(client, args.model, f"burst{concurrency}-{burst}", concurrency, count=concurrency)
+                        )
+                    measure(client, args.model, f"warmthroughput{concurrency}", concurrency, count=128)
+                    record_phase(
+                        measure(client, args.model, f"throughput{concurrency}", concurrency, duration=args.duration)
+                    )
+        finally:
+            process.terminate()
+            try:
+                process.wait(timeout=40)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+    result.update(status="complete", server_returncode=process.returncode)
+    if trace.exists():
+        batches = [json.loads(line) for line in trace.read_text().splitlines()]
+        for phase in phases:
+            prefix = phase["phase"] + ":"
+            phase["worker_batch_sizes"] = [
+                record["size"] for record in batches if any(i.startswith(prefix) for i in record["request_ids"])
+            ]
+    else:
+        result["status"] = "timing_complete_trace_missing"
+    (output / f"{name}.json").write_text(json.dumps(result, indent=2) + "\n")
+    return result
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--model", default="knowledgator/gliclass-large-v1.0")
+    parser.add_argument("--output", type=Path, default=HERE / "results")
+    parser.add_argument("--duration", type=float, default=12)
+    parser.add_argument("--sequential", type=int, default=100)
+    parser.add_argument("--order", default="baseline,candidate,candidate,baseline,baseline,candidate")
+    parser.add_argument("--port", type=int, default=8373)
+    parser.add_argument("--device", default="cuda")
+    args = parser.parse_args()
+    if args.duration <= 0 or args.sequential < 1:
+        parser.error("duration and sequential must be positive")
+    order = args.order.split(",")
+    if not order or any(arm not in {"baseline", "candidate"} for arm in order):
+        parser.error("order must contain baseline/candidate arms")
+    args.output.mkdir(parents=True, exist_ok=True)
+    manifest = json.loads((HERE / "_inputs/manifest.json").read_text())
+    nvidia_smi = shutil.which("nvidia-smi")
+    if nvidia_smi is None:
+        raise RuntimeError("nvidia-smi is required")
+    environment = {
+        "python": platform.python_version(),
+        "packages": {
+            name: importlib.metadata.version(name)
+            for name in ("torch", "transformers", "gliclass", "gliformer", "httpx", "uvicorn")
+        },
+        "nvidia_smi": subprocess.check_output(  # noqa: S603 — fixed GPU metadata query, no shell.
+            [nvidia_smi, "--query-gpu=name,uuid,driver_version,memory.total", "--format=csv,noheader"], text=True
+        ).strip(),
+        "cpu_count": os.cpu_count(),
+        "torch_threads": 4,
+        "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES", ""),
+    }
+    result = {
+        "manifest": manifest,
+        "environment": environment,
+        "model": args.model,
+        "texts": TEXTS,
+        "labels": LABELS,
+        "trials": [],
+    }
+    with tempfile.TemporaryDirectory(prefix="sie-idle-worker-") as temporary:
+        roots = {}
+        for arm in set(order):
+            archive = HERE / f"_inputs/{arm}.tar"
+            if hashlib.sha256(archive.read_bytes()).hexdigest() != manifest[arm]["archive_sha256"]:
+                raise RuntimeError(f"{arm} archive hash mismatch")
+            roots[arm] = Path(temporary) / arm
+            roots[arm].mkdir()
+            with tarfile.open(archive) as snapshot:
+                snapshot.extractall(roots[arm], filter="data")
+        for index, arm in enumerate(order):
+            print(f"IDLE_WORKER_START trial={index} arm={arm}", flush=True)
+            current = trial(args, arm, index, roots[arm], args.output)
+            result["trials"].append(current)
+            (args.output / "results.json").write_text(json.dumps(result, indent=2) + "\n")
+            print(
+                json.dumps(
+                    {
+                        "arm": arm,
+                        "trial": index,
+                        "phases": [
+                            {k: v for k, v in p.items() if k not in {"samples", "worker_batch_sizes"}}
+                            for p in current["phases"]
+                        ],
+                    }
+                ),
+                flush=True,
+            )
+    print("IDLE_WORKER_COMPLETE", flush=True)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tools/benchmarks/idle_worker/server.py b/tools/benchmarks/idle_worker/server.py
new file mode 100644
index 000000000..68f3cfaf7
--- /dev/null
+++ b/tools/benchmarks/idle_worker/server.py
@@ -0,0 +1,50 @@
+"""Local worker HTTP server with observational batch tracing for the benchmark."""
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+
+import uvicorn
+from sie_server.app.app_factory import AppFactory
+from sie_server.app.app_state_config import AppStateConfig
+from sie_server.core.batcher import FormattedBatch, HasCost
+from sie_server.core.worker import ModelWorker, model_worker
+from sie_server.core.worker.types import RequestMetadata
+
+parser = argparse.ArgumentParser(description=__doc__)
+parser.add_argument("--model", required=True)
+parser.add_argument("--models-dir", type=Path, required=True)
+parser.add_argument("--trace", type=Path, required=True)
+parser.add_argument("--port", type=int, default=8373)
+parser.add_argument("--device", default="cuda")
+args = parser.parse_args()
+trace_stream = args.trace.open("w")
+original_process_batch = ModelWorker._process_batch
+
+
+async def traced_process_batch(
+    self: ModelWorker, batch: FormattedBatch[HasCost, RequestMetadata], *, engine_queue_owned: bool = False
+) -> None:
+    await original_process_batch(self, batch, engine_queue_owned=engine_queue_owned)
+    ids = sorted({str(item.id) for metadata in batch.metadata for item in metadata.items})
+    trace_stream.write(json.dumps({"size": batch.size, "request_ids": ids}) + "\n")
+    trace_stream.flush()
+
+
+ModelWorker._process_batch = traced_process_batch  # ty: ignore[invalid-assignment]
+source = Path(model_worker.__file__)
+print(
+    json.dumps({"worker_path": str(source), "worker_sha256": hashlib.sha256(source.read_bytes()).hexdigest()}),
+    flush=True,
+)
+app = AppFactory.create_app(
+    AppStateConfig(
+        models_dir=args.models_dir,
+        device=args.device,
+        model_filter=[args.model],
+        preload_models=[args.model],
+        remote_serving=False,
+    )
+)
+uvicorn.run(app, host="127.0.0.1", port=args.port, log_level="warning", access_log=False, loop="asyncio")
diff --git a/tools/benchmarks/idle_worker/test_run.py b/tools/benchmarks/idle_worker/test_run.py
new file mode 100644
index 000000000..b5e49518b
--- /dev/null
+++ b/tools/benchmarks/idle_worker/test_run.py
@@ -0,0 +1,41 @@
+"""The timing probe follows the SDK's real extraction transport path."""
+
+import httpx
+from sie_sdk import SIEClient
+from sie_sdk._msgpack import packb
+from sie_sdk.types import Item
+
+from tools.benchmarks.idle_worker.run import timing_sample
+
+
+def test_timing_hook_follows_sdk_extract_route() -> None:
+    samples = []
+
+    def response_for(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            request=request,
+            headers={
+                "X-Queue-Time": "3.81",
+                "X-Inference-Time": "20.0",
+                "X-Total-Time": "24.0",
+                "Content-Type": "application/msgpack",
+            },
+            content=packb(
+                {
+                    "model": "knowledgator/gliclass-large-v1.0",
+                        "items": [
+                        {"id": "probe", "entities": [], "classifications": [{"label": "billing", "score": 0.9}]}
+                    ],
+                }
+            ),
+        )
+
+    transport = httpx.Client(
+        base_url="http://127.0.0.1:8373",
+        transport=httpx.MockTransport(response_for),
+        event_hooks={"response": [lambda response: samples.append(timing_sample(response))]},
+    )
+    with SIEClient("http://127.0.0.1:8373", http_client=transport) as client:
+        client.extract("knowledgator/gliclass-large-v1.0", Item(text="Refund please", id="probe"), labels=["billing"])
+    assert samples == [{"queue_ms": 3.81, "inference_ms": 20.0, "total_ms": 24.0}]

From 7fc9c798015447db8c2d75224a1dfa705fb14eb3 Mon Sep 17 00:00:00 2001
From: svonava <svonava@gmail.com>
Date: Fri, 16 Oct 2026 14:40:29 -0700
Subject: [PATCH 06/10] test(benchmarks): make idle-worker smoke directly
 runnable

---
 tools/benchmarks/idle_worker/README.md               | 2 +-
 tools/benchmarks/idle_worker/{ => tests}/test_run.py | 4 +---
 2 files changed, 2 insertions(+), 4 deletions(-)
 rename tools/benchmarks/idle_worker/{ => tests}/test_run.py (88%)

diff --git a/tools/benchmarks/idle_worker/README.md b/tools/benchmarks/idle_worker/README.md
index ba6fa0789..8a16596ed 100644
--- a/tools/benchmarks/idle_worker/README.md
+++ b/tools/benchmarks/idle_worker/README.md
@@ -18,7 +18,7 @@ From an initialized checkout:
 ```sh
 mise exec -- python tools/benchmarks/idle_worker/prepare.py \
   --baseline b0c26b02d21dbd146646c41b2b4f42627a0c9ebd \
-  --candidate 0649b2834f9c6c075f386516705d3179675575d4
+  --candidate f556494c8b70a6b25cb1d27acdadf1040d337218
 ```
 
 On the GPU machine, with the same locked workspace and prepared `_inputs/`:
diff --git a/tools/benchmarks/idle_worker/test_run.py b/tools/benchmarks/idle_worker/tests/test_run.py
similarity index 88%
rename from tools/benchmarks/idle_worker/test_run.py
rename to tools/benchmarks/idle_worker/tests/test_run.py
index b5e49518b..f839e7277 100644
--- a/tools/benchmarks/idle_worker/test_run.py
+++ b/tools/benchmarks/idle_worker/tests/test_run.py
@@ -24,9 +24,7 @@ def response_for(request: httpx.Request) -> httpx.Response:
             content=packb(
                 {
                     "model": "knowledgator/gliclass-large-v1.0",
-                        "items": [
-                        {"id": "probe", "entities": [], "classifications": [{"label": "billing", "score": 0.9}]}
-                    ],
+                    "items": [{"id": "probe", "entities": [], "classifications": [{"label": "billing", "score": 0.9}]}],
                 }
             ),
         )

From 8845eab5274ad571e0270c9d8c842035b5b5a426 Mon Sep 17 00:00:00 2001
From: svonava <svonava@gmail.com>
Date: Fri, 16 Oct 2026 15:01:27 -0700
Subject: [PATCH 07/10] fix(benchmarks): preserve idle-worker trial provenance

---
 tools/benchmarks/idle_worker/README.md        |  9 +-
 tools/benchmarks/idle_worker/prepare.py       |  4 +-
 tools/benchmarks/idle_worker/run.py           | 21 +++-
 .../idle_worker/tests/test_prepare.py         | 59 ++++++++++++
 .../benchmarks/idle_worker/tests/test_run.py  | 96 ++++++++++++++++++-
 5 files changed, 182 insertions(+), 7 deletions(-)
 create mode 100644 tools/benchmarks/idle_worker/tests/test_prepare.py

diff --git a/tools/benchmarks/idle_worker/README.md b/tools/benchmarks/idle_worker/README.md
index 8a16596ed..42d36105f 100644
--- a/tools/benchmarks/idle_worker/README.md
+++ b/tools/benchmarks/idle_worker/README.md
@@ -43,10 +43,15 @@ batch only after normal dispatch completes; it does not replace scheduling or
 inference. The existing worker unit tests cover deterministic queued-burst
 formation.
 
-Each completed phase is written before the next starts. Batch records are
-flushed as JSONL after each dispatch, so SIGTERM during server shutdown cannot
+Each completed phase is written before the next starts. If a later phase fails,
+the trial is marked `failed` and its completed measurements remain available.
+Trial JSON is replaced atomically so an incomplete write preserves the last
+completed phase. Batch records are flushed as JSONL after each dispatch, so
+SIGTERM during server shutdown cannot
 discard them. If a trace is missing, request timings remain available and the
 trial is marked `timing_complete_trace_missing`.
+Reported p95 values use nearest rank: sort the samples and select rank
+`ceil(0.95 * n)`, counting from one.
 
 Use an otherwise idle GPU and reserve enough CPU for the client and server.
 The server uses four Torch CPU threads. Keep the GPU allocation, dependencies,
diff --git a/tools/benchmarks/idle_worker/prepare.py b/tools/benchmarks/idle_worker/prepare.py
index 025aba197..af2b71970 100644
--- a/tools/benchmarks/idle_worker/prepare.py
+++ b/tools/benchmarks/idle_worker/prepare.py
@@ -40,7 +40,9 @@ def main() -> None:
         }
     if locks[0] != locks[1] or locks[1] != (ROOT / "uv.lock").read_bytes():
         raise SystemExit("Both revisions and the execution workspace must use the same uv.lock")
-    manifest["source_diff"] = git("diff", "--stat", args.baseline, args.candidate, "--", *PATHS).decode()
+    manifest["source_diff"] = git(
+        "diff", "--stat", manifest["baseline"]["commit"], manifest["candidate"]["commit"], "--", *PATHS
+    ).decode()
     (INPUTS / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
     print(json.dumps(manifest, indent=2))
 
diff --git a/tools/benchmarks/idle_worker/run.py b/tools/benchmarks/idle_worker/run.py
index 828922f2f..ebd3bf9ab 100644
--- a/tools/benchmarks/idle_worker/run.py
+++ b/tools/benchmarks/idle_worker/run.py
@@ -5,6 +5,7 @@
 import importlib.metadata
 import itertools
 import json
+import math
 import os
 import platform
 import shutil
@@ -36,11 +37,12 @@
 
 
 def distribution(values: list[float]) -> dict:
+    """Summarize samples using the nearest-rank definition of p95."""
     ordered = sorted(values)
     return {
         "n": len(values),
         "median": statistics.median(values),
-        "p95": ordered[min(len(values) - 1, int(len(values) * 0.95))],
+        "p95": ordered[math.ceil(len(values) * 0.95) - 1],
         "max": max(values),
     }
 
@@ -152,10 +154,16 @@ def capture(response: httpx.Response) -> None:
     ]
     phases = []
     result = {"arm": arm, "trial": trial_index, "phases": phases, "status": "running"}
+    result_path = output / f"{name}.json"
+
+    def persist_result() -> None:
+        pending = result_path.with_suffix(".json.tmp")
+        pending.write_text(json.dumps(result, indent=2) + "\n")
+        pending.replace(result_path)
 
     def record_phase(phase: dict) -> None:
         phases.append(phase)
-        (output / f"{name}.json").write_text(json.dumps(result, indent=2) + "\n")
+        persist_result()
 
     with (output / f"{name}-server.log").open("w") as log:
         process = subprocess.Popen(argv, env=environment, stdout=log, stderr=subprocess.STDOUT)  # noqa: S603 — fixed local server, no shell.
@@ -190,6 +198,13 @@ def record_phase(phase: dict) -> None:
                     record_phase(
                         measure(client, args.model, f"throughput{concurrency}", concurrency, duration=args.duration)
                     )
+        except BaseException as error:
+            result["status"] = "failed"
+            try:
+                persist_result()
+            except OSError as persistence_error:
+                error.add_note(f"Could not persist failed trial: {persistence_error}")
+            raise
         finally:
             process.terminate()
             try:
@@ -207,7 +222,7 @@ def record_phase(phase: dict) -> None:
             ]
     else:
         result["status"] = "timing_complete_trace_missing"
-    (output / f"{name}.json").write_text(json.dumps(result, indent=2) + "\n")
+    persist_result()
     return result
 
 
diff --git a/tools/benchmarks/idle_worker/tests/test_prepare.py b/tools/benchmarks/idle_worker/tests/test_prepare.py
new file mode 100644
index 000000000..f5a01add2
--- /dev/null
+++ b/tools/benchmarks/idle_worker/tests/test_prepare.py
@@ -0,0 +1,59 @@
+"""Snapshot provenance must stay stable even if an input branch moves."""
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from tools.benchmarks.idle_worker import prepare
+
+
+def test_diff_describes_snapshots_when_input_branch_moves(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(prepare, "ROOT", tmp_path)
+    monkeypatch.setattr(prepare, "INPUTS", tmp_path / "_inputs")
+    git = prepare.git
+    git("init", "--quiet")
+    (tmp_path / "uv.lock").write_text("same lock\n")
+    for directory in prepare.PATHS:
+        target = tmp_path / directory
+        target.mkdir(parents=True)
+        (target / "probe.txt").write_text("baseline\n")
+
+    def commit() -> str:
+        git("add", ".")
+        git(
+            "-c",
+            "user.name=Benchmark Test",
+            "-c",
+            "user.email=benchmark@example.test",
+            "commit",
+            "--quiet",
+            "-m",
+            "snapshot",
+        )
+        return git("rev-parse", "HEAD").decode().strip()
+
+    baseline = commit()
+    (tmp_path / prepare.PATHS[0] / "probe.txt").write_text("candidate\n")
+    candidate = commit()
+    git("branch", "baseline", baseline)
+    git("branch", "candidate", candidate)
+    expected_diff = git("diff", "--stat", baseline, candidate, "--", *prepare.PATHS).decode()
+    (tmp_path / prepare.PATHS[0] / "later.txt").write_text("not in the archived candidate\n")
+    moved_candidate = commit()
+
+    def git_with_moving_branch(*args: str) -> bytes:
+        if args[0] == "diff":
+            git("update-ref", "refs/heads/candidate", moved_candidate)
+        return git(*args)
+
+    monkeypatch.setattr(prepare, "git", git_with_moving_branch)
+    monkeypatch.setattr(sys, "argv", ["prepare.py", "--baseline", "baseline", "--candidate", "candidate"])
+    prepare.main()
+
+    manifest = json.loads((prepare.INPUTS / "manifest.json").read_text())
+    assert manifest["baseline"]["commit"] == baseline
+    assert manifest["candidate"]["commit"] == candidate
+    assert git("rev-parse", "candidate").decode().strip() == moved_candidate
+    assert manifest["source_diff"] == expected_diff
diff --git a/tools/benchmarks/idle_worker/tests/test_run.py b/tools/benchmarks/idle_worker/tests/test_run.py
index f839e7277..332577212 100644
--- a/tools/benchmarks/idle_worker/tests/test_run.py
+++ b/tools/benchmarks/idle_worker/tests/test_run.py
@@ -1,11 +1,17 @@
 """The timing probe follows the SDK's real extraction transport path."""
 
+import argparse
+import json
+from pathlib import Path
+
 import httpx
+import pytest
 from sie_sdk import SIEClient
 from sie_sdk._msgpack import packb
 from sie_sdk.types import Item
 
-from tools.benchmarks.idle_worker.run import timing_sample
+from tools.benchmarks.idle_worker import run
+from tools.benchmarks.idle_worker.run import distribution, timing_sample
 
 
 def test_timing_hook_follows_sdk_extract_route() -> None:
@@ -37,3 +43,91 @@ def response_for(request: httpx.Request) -> httpx.Response:
     with SIEClient("http://127.0.0.1:8373", http_client=transport) as client:
         client.extract("knowledgator/gliclass-large-v1.0", Item(text="Refund please", id="probe"), labels=["billing"])
     assert samples == [{"queue_ms": 3.81, "inference_ms": 20.0, "total_ms": 24.0}]
+
+
+@pytest.mark.parametrize(("count", "p95"), [(1, 1), (20, 19), (100, 95), (101, 96), (300, 285)])
+def test_distribution_uses_nearest_rank_p95(count: int, p95: int) -> None:
+    assert distribution(list(reversed(range(1, count + 1))))["p95"] == p95
+
+
+@pytest.mark.parametrize("failure_type", [RuntimeError, KeyboardInterrupt])
+@pytest.mark.parametrize("persistence_fails", [False, True])
+def test_failed_trial_preserves_completed_phases_and_stops_server(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure_type: type[BaseException], persistence_fails: bool
+) -> None:
+    stopped = []
+
+    class Process:
+        returncode = None
+
+        def terminate(self) -> None:
+            stopped.append("terminate")
+
+        def wait(self, timeout: int) -> None:
+            stopped.append(timeout)
+
+    monkeypatch.setattr(run.subprocess, "Popen", lambda *args, **kwargs: Process())
+    monkeypatch.setattr(run, "wait_for_server", lambda *args: None)
+    real_client = httpx.Client
+
+    def response_for(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            request=request,
+            headers={
+                "X-Queue-Time": "0.12",
+                "X-Inference-Time": "20",
+                "X-Total-Time": "21",
+                "Content-Type": "application/msgpack",
+            },
+            content=packb(
+                {
+                    "model": "probe",
+                    "items": [
+                        {"id": "sequential:0", "entities": [], "classifications": [{"label": "billing", "score": 0.9}]}
+                    ],
+                }
+            ),
+        )
+
+    monkeypatch.setattr(
+        run.httpx,
+        "Client",
+        lambda *args, **kwargs: real_client(*args, **kwargs, transport=httpx.MockTransport(response_for)),
+    )
+    failure = failure_type("burst failed after sequential completed")
+    trace = tmp_path / "00-candidate-batches.jsonl"
+    trace_record = '{"request_ids": ["sequential:0"], "size": 1}\n'
+
+    def measure(client: SIEClient, model: str, phase: str, concurrency: int, **kwargs) -> dict:
+        if phase.startswith("warmup"):
+            return {}
+        if phase == "sequential":
+            client.extract(model, Item(text="Refund please", id="sequential:0"), labels=["billing"])
+            trace.write_text(trace_record)
+            return {"phase": phase, "requests": 1}
+        raise failure
+
+    monkeypatch.setattr(run, "measure", measure)
+    real_write_text = Path.write_text
+
+    def write_text(path: Path, data: str, *args, **kwargs) -> int:
+        if persistence_fails and '"status": "failed"' in data:
+            real_write_text(path, "incomplete")
+            raise OSError("disk unavailable while recording failure")
+        return real_write_text(path, data, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "write_text", write_text)
+    args = argparse.Namespace(model="probe", port=8373, device="cpu", sequential=1, duration=0.01)
+    with pytest.raises(failure_type) as caught:
+        run.trial(args, "candidate", 0, tmp_path, tmp_path)
+
+    assert caught.value is failure
+    assert stopped == ["terminate", 40]
+    result = json.loads((tmp_path / "00-candidate.json").read_text())
+    assert result["status"] == ("running" if persistence_fails else "failed")
+    assert result["phases"][0]["phase"] == "sequential"
+    assert result["phases"][0]["samples"] == [{"queue_ms": 0.12, "inference_ms": 20.0, "total_ms": 21.0}]
+    assert trace.read_text() == trace_record
+    if persistence_fails:
+        assert any("disk unavailable" in note for note in caught.value.__notes__)

From a7c024f6f53cce0f8e43825845c5ad6965c9c792 Mon Sep 17 00:00:00 2001
From: svonava <svonava@gmail.com>
Date: Sun, 18 Oct 2026 06:44:46 -0700
Subject: [PATCH 08/10] fix(server): let ready callbacks enqueue idle
 submissions

---
 .../sie_server/core/worker/model_worker.py    |  33 +++++-
 .../sie_server/tests/core/test_worker_core.py | 110 ++++++++++++++++++
 2 files changed, 137 insertions(+), 6 deletions(-)

diff --git a/packages/sie_server/src/sie_server/core/worker/model_worker.py b/packages/sie_server/src/sie_server/core/worker/model_worker.py
index c9964bda6..75ca3cf2c 100644
--- a/packages/sie_server/src/sie_server/core/worker/model_worker.py
+++ b/packages/sie_server/src/sie_server/core/worker/model_worker.py
@@ -123,6 +123,27 @@ def _dispatch_width(adapter: object) -> int:
     return max(1, width)
 
 
+async def _yield_ready_callbacks() -> None:
+    """Let ready producer callbacks schedule submissions before the idle recheck.
+
+    Completing a Future queues this task behind work those callbacks create.
+    This is one scheduling checkpoint, not a drain of all incoming work: later
+    network arrivals or unfinished preparation may still miss the batch.
+    """
+    loop = asyncio.get_running_loop()
+    ready: asyncio.Future[None] = loop.create_future()
+
+    def wake() -> None:
+        if not ready.done():
+            ready.set_result(None)
+
+    handle = loop.call_soon(wake)
+    try:
+        await ready
+    finally:
+        handle.cancel()
+
+
 class ModelWorker:
     """Worker that batches and processes inference requests for a single model.
 
@@ -1093,8 +1114,9 @@ async def _get_next_batch_fcfs(
 
         When the worker was idle, one pending request is dispatched immediately
         (#373), however many items it carries. It is not held for
-        ``idle_coalesce_ms``. A lone head yields one event-loop turn so already
-        runnable submissions can join it, without waiting for a timer. A
+        ``idle_coalesce_ms``. A lone head yields one callback generation so
+        ready producers and the submissions they schedule can join it, without
+        waiting for a timer. Later arrivals may still miss this batch. A
         follow-up is a tail only while another request
         on that same batcher is still pending (arrived and not yet dispatched);
         the idle window can then catch a straggler. A finished request is no
@@ -1136,10 +1158,9 @@ async def _get_next_batch_fcfs(
                 selected = self._batchers[selected_lora]
                 idle_window_ms = self._config.idle_coalesce_ms
                 if was_idle and idle_window_ms > 0 and selected.pending_request_count == 1:
-                    # Let submissions already runnable on this loop join the
-                    # head before deciding whether it is a lone request. This
-                    # yields once without waiting for a timer or later arrival.
-                    await asyncio.sleep(0)
+                    # A ready transport callback may create the submission task.
+                    # Give that task one opportunity before the lone-head check.
+                    await _yield_ready_callbacks()
                     if not self._running:
                         return None, FormattedBatch(items=[], metadata=[], total_cost=0), was_idle
                 # Alone is one request, not one item: ``submit_many`` queues
diff --git a/packages/sie_server/tests/core/test_worker_core.py b/packages/sie_server/tests/core/test_worker_core.py
index 498bbfca1..3d5fc25d9 100644
--- a/packages/sie_server/tests/core/test_worker_core.py
+++ b/packages/sie_server/tests/core/test_worker_core.py
@@ -16,6 +16,54 @@
 from sie_server.types.inputs import Item
 
 
+@pytest.fixture(params=["asyncio", "uvloop"])
+def ready_loop_factory(request: pytest.FixtureRequest):
+    if request.param == "uvloop":
+        return pytest.importorskip("uvloop").new_event_loop
+    return asyncio.new_event_loop
+
+
+@pytest.mark.parametrize("stage", ["before_start", "before_resolve", "after_resolve_before_resume"])
+def test_ready_callback_checkpoint_cancellation(
+    stage: str, monkeypatch: pytest.MonkeyPatch, ready_loop_factory
+) -> None:
+    async def check() -> None:
+        """Cancellation cannot leave an owned callback or resolve a cancelled Future."""
+        loop = asyncio.get_running_loop()
+        call_soon = loop.call_soon
+        handles: list[asyncio.Handle] = []
+        errors: list[dict] = []
+        old_handler = loop.get_exception_handler()
+
+        def record_callback(callback, *args, context=None):
+            handle = call_soon(callback, *args, context=context)
+            if getattr(callback, "__module__", None) == model_worker_module.__name__:
+                handles.append(handle)
+            return handle
+
+        monkeypatch.setattr(loop, "call_soon", record_callback)
+        loop.set_exception_handler(lambda _loop, context: errors.append(context))
+        try:
+            task = asyncio.create_task(model_worker_module._yield_ready_callbacks())
+            if stage == "before_start":
+                task.cancel()
+            elif stage == "before_resolve":
+                loop.call_soon(task.cancel)
+            else:
+                loop.call_soon(lambda: loop.call_soon(task.cancel))
+            with pytest.raises(asyncio.CancelledError):
+                await task
+            await asyncio.sleep(0)
+            assert len(handles) == (0 if stage == "before_start" else 1)
+            assert all(handle.cancelled() for handle in handles)
+            assert errors == []
+        finally:
+            loop.set_exception_handler(old_handler)
+
+    with asyncio.Runner(loop_factory=ready_loop_factory) as runner:
+        runner.run(check())
+
+
 class TestRequestMetadata:
     """Tests for RequestMetadata dataclass."""
 
@@ -714,6 +762,68 @@ async def submit_ready():
         finally:
             await worker.stop()
 
+    @pytest.mark.parametrize("warmup", [False, True])
+    def test_ready_producer_callbacks_join_the_idle_head(
+        self, mock_adapter: MagicMock, warmup: bool, ready_loop_factory
+    ) -> None:
+        async def check() -> None:
+            """Ready transport-like callbacks create request tasks behind the worker.
+
+            A bare sleep(0) resumes the worker before those tasks. The checkpoint
+            lets their synchronous preparation and submit run without a timed wait.
+            """
+            call_sizes: list[int] = []
+            producer_callbacks: list[int] = []
+            producers_seen_at_dispatch: list[int] = []
+
+            def counting_encode(items, output_types, **kwargs):
+                call_sizes.append(len(items))
+                producers_seen_at_dispatch.append(len(producer_callbacks))
+                return EncodeOutput(dense=np.array([[0.1, 0.2, 0.3]] * len(items)), batch_size=len(items))
+
+            mock_adapter.encode.side_effect = counting_encode
+            worker = ModelWorker(
+                mock_adapter,
+                WorkerConfig(max_batch_tokens=1000, max_batch_requests=64, idle_coalesce_ms=3),
+            )
+            await worker.start()
+            tasks: list[asyncio.Task[None]] = []
+            submitted: list[asyncio.Future[WorkerResult]] = []
+            all_submitted = asyncio.Event()
+
+            async def submit_tail(index: int) -> None:
+                submitted.append(
+                    await worker.submit([make_text_item([1, 2], 0)], [Item(text=f"tail {index}")], ["dense"])
+                )
+                if len(submitted) == 7:
+                    all_submitted.set()
+
+            def produce(index: int) -> None:
+                producer_callbacks.append(index)
+                tasks.append(asyncio.create_task(submit_tail(index)))
+
+            try:
+                if warmup:
+                    warmed = await worker.submit([make_text_item([1, 2], 0)], [Item(text="warmup")], ["dense"])
+                    await asyncio.wait_for(warmed, timeout=2)
+                    call_sizes.clear()
+                    producers_seen_at_dispatch.clear()
+                first = await worker.submit([make_text_item([1, 2], 0)], [Item(text="head")], ["dense"])
+                loop = asyncio.get_running_loop()
+                for index in range(7):
+                    loop.call_soon(produce, index)
+                await asyncio.wait_for(all_submitted.wait(), timeout=2)
+                await asyncio.wait_for(asyncio.gather(first, *submitted), timeout=2)
+                assert producer_callbacks == list(range(7))
+                assert producers_seen_at_dispatch == [7]
+                assert call_sizes == [8]
+            finally:
+                await asyncio.gather(*tasks, return_exceptions=True)
+                await worker.stop()
+
+        with asyncio.Runner(loop_factory=ready_loop_factory) as runner:
+            runner.run(check())
+
     @pytest.mark.asyncio
     async def test_already_queued_burst_fuses(self, mock_adapter: MagicMock) -> None:
         """Several items already pending when an idle worker selects still fuse.

From 1661625e64d67024e8471a75fcde474b03c46bd3 Mon Sep 17 00:00:00 2001
From: svonava <svonava@gmail.com>
Date: Sun, 18 Oct 2026 08:56:14 -0700
Subject: [PATCH 09/10] docs(benchmarks): report ready-callback idle-worker
 comparison

---
 .../idle_worker/ready_generation_result.md    | 102 ++++++++++++++++++
 1 file changed, 102 insertions(+)
 create mode 100644 tools/benchmarks/idle_worker/ready_generation_result.md

diff --git a/tools/benchmarks/idle_worker/ready_generation_result.md b/tools/benchmarks/idle_worker/ready_generation_result.md
new file mode 100644
index 000000000..588ab12fe
--- /dev/null
+++ b/tools/benchmarks/idle_worker/ready_generation_result.md
@@ -0,0 +1,102 @@
+The new fixed three-arm L4 comparison completed all 18 trials. It keeps the sequential latency benefit and improves C8 burst fusion over the prior `sleep(0)` implementation, but it does not meet the issue's strict one-batch requirement for every HTTP burst. The result remains `completed_valid: true`, `qualification_passed: false`, with the existing human review hold unchanged.
+
+The exact source arms were A: original timed coalescing `b0c26b02d21dbd146646c41b2b4f42627a0c9ebd`; B: current PR658 `8845eab5274ad571e0270c9d8c842035b5b5a426`; C: ready-callback candidate `a7c024f6f53cce0f8e43825845c5ad6965c9c792`. C preserves the PR history and changes only the lone-idle scheduling checkpoint plus its regression tests. A callback completes a Future so the waiting worker resumes after tasks created by already-ready producer callbacks. It is a finite scheduling opportunity; later network bodies or unfinished preprocessing can still miss it.
+
+All arms used the same single NVIDIA L4, 8 CPU cores, 32 GiB memory, pinned `knowledgator/gliclass-large-v1.0` revision `ae8a18644e84cfda53325bd91c7941f2f6fd0ccb`, and actual local SIE HTTP API through `sie_sdk.SIEClient`. The observed model parameters were FP16 on CUDA. The default profile retained its existing bucketed CUDA graphs, threshold 0, batch-token limit 16384, and `idle_coalesce_ms=3`. Each fresh server used asyncio, four Torch intra-op threads and one inter-op thread. Python 3.12.12, Torch 2.9.1+cu129, Transformers 4.57.6, GLiClass 0.1.17, gliformer 0.1.2, httpx 0.28.1, and uvicorn 0.46.0 were admitted exactly. CPU scheduling/cancellation controls also cover uvloop; this GPU experiment did not use uvloop.
+
+The six fixed triplets were ABC, CBA, BCA, ACB, CAB, BAC. Every trial warmed C1/C8/C32 with 64 calls each, measured 100 sequential calls, four idle bursts each at C8 and C32, and two 12-second closed-loop throughput phases, each preceded by 128 warmup calls. Throughput includes the full drain after the admission cutoff. All 119,432 SDK calls, including warmups, completed with one HTTP 200 response each. Full request/response bytes, SDK results, timing headers, and request-to-worker-batch joins were retained; no trial, request, pair, or outlier was excluded.
+
+| Observation | A: timed coalescing | B: `sleep(0)` | C: ready callback |
+|---|---:|---:|---:|
+| Sequential queue samples below 1 ms | 0/600 | 600/600 | 600/600 |
+| Per-trial queue median range | 3.395–3.440 ms | 0.090–0.100 ms | 0.100–0.110 ms |
+| Per-trial queue nearest-rank p95 range | 3.84–4.27 ms | 0.11–0.13 ms | 0.13–0.17 ms |
+| Maximum sequential queue time | 4.44 ms | 0.31 ms | 0.25 ms |
+| Per-trial SDK latency p95 range | 14.469–15.779 ms | 10.840–11.373 ms | 10.987–12.363 ms |
+| C8 bursts forming one worker batch | 24/24 | 3/24 | 18/24 |
+| C32 bursts forming one worker batch | 0/24 | 0/24 | 0/24 |
+
+C's six remaining C8 bursts split `[1, 7]`. At C32, the original baseline also split every burst; this does not waive the issue's one-batch criterion. The full ordered pattern inventory is below.
+
+C/A throughput geometric ratios were 1.01095 at C8 (pointwise 95% interval 0.99637–1.02574) and 1.00010 at C32 (0.97286–1.02811). C/B ratios were 0.99319 (0.97069–1.01621) and 0.98980 (0.97423–1.00561). These are six paired fresh-process observations on one shared device/session. There is no predefined equivalence margin, so intervals containing 1 do not establish "unchanged within noise" or statistical equivalence. All twelve directions, individual ratios, and paired log dispersions are retained below.
+
+All responses retained the same classification label sets, ordered label lists, winning labels, and other nonnumeric output structure. Sequential outputs matched the fixed first-A reference exactly. The largest direct score difference over all phases was 0.00390625 in each arm, and the same maximum range occurred within each arm's own repeats. Those differences remain visible; no output tolerance or task-quality claim is introduced.
+
+The earlier six-trial yield result and its C8 split failure remain unchanged (`results.json` SHA256 `0c89a0b4693cffe29225415eace97e6551c2b426ca4eb99693b5da75c347ed4b`). The first attempt at this new three-arm protocol ended before inference because the cache verifier compared a lexical mount alias against a canonical root. Its failure was retained; run2 changed only that verified path-handling defect and attempt identities, after CPU alias/escape/retarget controls. No measured workload, output policy, trial order, or timing bound was changed in response to performance results.
+
+The experiment includes identical output capture and batch tracing in every arm. It shares one device, image, and session across the balanced process order, so thermal, cache, JIT, and process variation remain possible. It measures the Python worker's HTTP path, not gateway/sidecar-only scheduling. The method does not predict arbitrary later network arrivals. The measured candidate has not yet been integrated with newer main; any later integration needs explicit source mapping and fresh CI/review.
+
+The complete archive was verified before analysis. The provider was stopped and independently observed with zero containers and no tasks, every owned local process was reaped, and all experiment leases were released. No further GPU attempt is proposed.
+
+Evidence SHA256 values: frozen method `fcf1aa6eac0542065c12f93c952d60b6b5d2aa7413455b2e3a97fa3c8437a8c2`; source-input manifest `398f68d9d62639a389fdb3b07850f19120a7eeda14d6edcaecf723e291147c35`; unchanged analyzer `b323ceedeb056f27bf40729ebd1749b8629a3b29bc4887ad866a83a55c859c08`; complete analysis `51e0d08bb724d08a5b270b3ca051ff917462c5b3cbed709553bbbba52a2c2c1f`; terminal receipt `cb99a4ee8e090a7a0fd7de8c0db41549b547bbce92fd1bc8ef332eb86b9d5998`; full 470-member archive `9933b24ff54bc85d6c617ba053aaf27a34ada8bc2ba69370f3d3dc70027fa9ed` (33,591,363 compressed bytes; 332,647,329 raw bytes). These identify retained artifacts; they are not a claim that private storage locations are publicly accessible. The complete method and result tables follow; the checked-in two-arm harness and its reproducibility limits are described at the end.
+
+All 18 trials, in execution order (nearest-rank p95; block/trial numbers below start at zero).
+
+| Trial | Block | Arm | Queue median / p95 (ms) | SDK p95 (ms) | C8 requests/s | C32 requests/s | C8 fused | C32 fused |
+|---:|---:|:---:|---:|---:|---:|---:|---:|---:|
+| 0 | 0 | A | 3.410 / 4.18 | 14.969 | 209.678 | 284.718 | 4/4 | 0/4 |
+| 1 | 0 | B | 0.090 / 0.11 | 10.840 | 206.797 | 288.443 | 0/4 | 0/4 |
+| 2 | 0 | C | 0.110 / 0.13 | 11.222 | 211.718 | 291.785 | 3/4 | 0/4 |
+| 3 | 1 | C | 0.110 / 0.13 | 11.296 | 209.282 | 275.702 | 3/4 | 0/4 |
+| 4 | 1 | B | 0.090 / 0.12 | 11.148 | 208.874 | 279.815 | 0/4 | 0/4 |
+| 5 | 1 | A | 3.440 / 3.84 | 14.469 | 210.761 | 282.735 | 4/4 | 0/4 |
+| 6 | 2 | B | 0.100 / 0.13 | 11.296 | 215.243 | 282.947 | 2/4 | 0/4 |
+| 7 | 2 | C | 0.100 / 0.13 | 10.987 | 209.845 | 274.439 | 2/4 | 0/4 |
+| 8 | 2 | A | 3.410 / 4.23 | 15.779 | 202.622 | 284.480 | 4/4 | 0/4 |
+| 9 | 3 | A | 3.395 / 4.27 | 15.429 | 200.816 | 276.213 | 4/4 | 0/4 |
+| 10 | 3 | C | 0.110 / 0.14 | 11.683 | 201.679 | 280.336 | 4/4 | 0/4 |
+| 11 | 3 | B | 0.100 / 0.13 | 11.373 | 208.513 | 282.631 | 0/4 | 0/4 |
+| 12 | 4 | C | 0.110 / 0.17 | 12.363 | 211.302 | 280.297 | 4/4 | 0/4 |
+| 13 | 4 | A | 3.430 / 4.13 | 14.901 | 209.259 | 281.434 | 4/4 | 0/4 |
+| 14 | 4 | B | 0.090 / 0.12 | 10.985 | 214.833 | 286.102 | 0/4 | 0/4 |
+| 15 | 5 | B | 0.100 / 0.12 | 11.329 | 210.389 | 283.979 | 1/4 | 0/4 |
+| 16 | 5 | A | 3.435 / 4.26 | 14.853 | 209.371 | 276.820 | 4/4 | 0/4 |
+| 17 | 5 | C | 0.105 / 0.13 | 11.233 | 212.252 | 284.243 | 2/4 | 0/4 |
+
+All ratio directions use six process pairs. Intervals are pointwise, two-sided 95% t intervals on paired log ratios (df = 5), without an equivalence margin.
+
+| Throughput ratio | Six block ratios | Geometric mean | 95% interval | Paired log SD |
+|---|---|---:|---|---:|
+| throughput32:A/B | 0.98709, 1.01044, 1.00542, 0.97729, 0.98369, 0.97479 | 0.98969 | 0.97433–1.00530 | 0.014912 |
+| throughput32:A/C | 0.97578, 1.02551, 1.03659, 0.98529, 1.00406, 0.97389 | 0.99990 | 0.97266–1.02790 | 0.026316 |
+| throughput32:B/A | 1.01308, 0.98967, 0.99461, 1.02323, 1.01658, 1.02586 | 1.01041 | 0.99473–1.02635 | 0.014912 |
+| throughput32:B/C | 0.98855, 1.01492, 1.03100, 1.00818, 1.02071, 0.99907 | 1.01031 | 0.99442–1.02645 | 0.015102 |
+| throughput32:C/A | 1.02482, 0.97513, 0.96470, 1.01493, 0.99596, 1.02681 | 1.00010 | 0.97286–1.02811 | 0.026316 |
+| throughput32:C/B | 1.01159, 0.98530, 0.96993, 0.99188, 0.97971, 1.00093 | 0.98980 | 0.97423–1.00561 | 0.015102 |
+| throughput8:A/B | 1.01393, 1.00904, 0.94136, 0.96309, 0.97406, 0.99516 | 0.98243 | 0.95311–1.01266 | 0.028872 |
+| throughput8:A/C | 0.99036, 1.00707, 0.96558, 0.99572, 0.99033, 0.98643 | 0.98917 | 0.97490–1.00365 | 0.013844 |
+| throughput8:B/A | 0.98626, 0.99104, 1.06229, 1.03833, 1.02664, 1.00486 | 1.01788 | 0.98750–1.04920 | 0.028872 |
+| throughput8:B/C | 0.97676, 0.99805, 1.02572, 1.03389, 1.01671, 0.99122 | 1.00686 | 0.98405–1.03020 | 0.021838 |
+| throughput8:C/A | 1.00973, 0.99298, 1.03565, 1.00430, 1.00976, 1.01376 | 1.01095 | 0.99637–1.02574 | 0.013844 |
+| throughput8:C/B | 1.02380, 1.00196, 0.97492, 0.96722, 0.98357, 1.00885 | 0.99319 | 0.97069–1.01621 | 0.021838 |
+
+Every measured idle-burst pattern (ordered worker batch sizes; 24 bursts per arm and concurrency).
+
+| Arm | Concurrency | Pattern counts |
+|:---:|---:|---|
+| A | 8 | `[8]` × 24 |
+| A | 32 | `[19, 13]` × 1; `[21, 11]` × 1; `[25, 7]` × 3; `[26, 6]` × 6; `[27, 5]` × 6; `[28, 4]` × 4; `[29, 3]` × 2; `[30, 2]` × 1 |
+| B | 8 | `[1, 7]` × 21; `[8]` × 3 |
+| B | 32 | `[1, 15, 16]` × 2; `[1, 31]` × 15; `[25, 7]` × 1; `[27, 5]` × 3; `[28, 4]` × 3 |
+| C | 8 | `[1, 7]` × 6; `[8]` × 18 |
+| C | 32 | `[1, 11, 20]` × 1; `[1, 12, 19]` × 1; `[1, 23, 8]` × 1; `[1, 31]` × 2; `[22, 10]` × 1; `[24, 8]` × 1; `[26, 6]` × 6; `[27, 5]` × 4; `[28, 4]` × 6; `[31, 1]` × 1 |
+
+The public serving source and original two-arm workload are in this repository. The existing `tools/benchmarks/idle_worker/README.md` documents source preparation and execution on a caller-owned GPU. Once the measured candidate commit is published, a fresh original-versus-candidate comparison can use:
+
+```sh
+mise exec -- python tools/benchmarks/idle_worker/prepare.py \
+  --baseline b0c26b02d21dbd146646c41b2b4f42627a0c9ebd \
+  --candidate a7c024f6f53cce0f8e43825845c5ad6965c9c792
+mise exec -- python tools/benchmarks/idle_worker/run.py \
+  --output /tmp/sie-idle-worker-results
+```
+
+These commands run the existing two-arm benchmark, not the completed three-arm experiment verbatim. The 18-trial experiment retained additional full request/response bytes, source/runtime/cache admission, exact process ownership, failure persistence, and a separate full-row analyzer. Those additions are not yet packaged as a public one-command reproducer. Its fixed method is specified above so that the difference is visible; do not treat a new run of the older harness as a hash-identical replay or pool it with this result.
+
+For a faithful independent implementation of the three-arm method, preserve all six listed triplets on the same device, fresh actual server/SDK instances, the exact environment/profile and fixed public checkpoint, all warmup/count/duration/drain rules, 50 ms before each idle burst, and all original raw output/batch observations. The full source archive path inventory consisted of server source, SDK source, model catalog, bundle definitions, and `uv.lock`; the same lock SHA256 was `c1286da62744012fe54f07cd7ec740d1d1a69db0e2db75c8a0a0a8007ac3a397`. The A/B/C archive SHA256 values were `0926ef65a420c437f9f5f2147a44f0b6ed07e0cd51cf7a376a2bb99eccee2257`, `6e8ba5a26bd3c2f223cd2bb05351e53307c0180d645a4541678a9fb8fded29d0`, and `e2380107510882bf8c273d6139e79d63412a31327fe4365a45f335656512928f`.
+
+The fixed active-run ceiling was 1800 seconds, with 180 seconds active plus at most 30 seconds cleanup for each admitted fresh trial. The experiment stopped on the first failure, required enough remaining budget for a complete trial and cleanup, and did not append blocks, shorten windows, or omit outliers. Independent process/phase identity and every complete request/wire/batch join are necessary to reproduce the integrity assessment; timing similarity alone is insufficient.
+
+The seven required public snapshot files totaled 1,763,619,203 bytes. The checkpoint `model.safetensors` was 1,752,499,444 bytes with SHA256 `9d6bafde8a1a43bf5563fd6296f8934752d23462de9113b0b8ac56ecc7cdddec`; the required-file manifest SHA256 was `c003d0e73df9bb7d23337c6d6a54312dc07e2639704e47981b15b158812ab053`. Full file content, revision, and resolved cache paths were checked before model work. A cache or environment failure is an incomplete attempt, not a reason to substitute inputs silently.
+
+Independent reviewers recomputed all 119,432 request joins, 288 phases, 19,221 post-forward worker batches, every ratio and output diagnostic. Worker batch records identify SIE dispatch grouping; they do not establish physical CUDA tensor shapes or the cause of a particular HTTP split. The result-review SHA256 is `9a00f78392a47f60f5eff6956fa0037dc9468d1d31e19efdf2109905c2c28b08`; the independent terminal/archive review SHA256 is `725bbd0d6f842ccceb4416175dcb3718af4998182e03c2dde496290306a5d8b2`.

From 49d6ea415ebbd2b6c7d63b656eccf4c6ead15b85 Mon Sep 17 00:00:00 2001
From: svonava <svonava@gmail.com>
Date: Mon, 19 Oct 2026 09:19:21 -0700
Subject: [PATCH 10/10] docs: address idle worker benchmark review

---
 tools/benchmarks/idle_worker/README.md        |   2 +-
 .../idle_worker/ready_generation_result.md    | 102 ------------------
 2 files changed, 1 insertion(+), 103 deletions(-)
 delete mode 100644 tools/benchmarks/idle_worker/ready_generation_result.md

diff --git a/tools/benchmarks/idle_worker/README.md b/tools/benchmarks/idle_worker/README.md
index 42d36105f..cc9cbce2f 100644
--- a/tools/benchmarks/idle_worker/README.md
+++ b/tools/benchmarks/idle_worker/README.md
@@ -18,7 +18,7 @@ From an initialized checkout:
 ```sh
 mise exec -- python tools/benchmarks/idle_worker/prepare.py \
   --baseline b0c26b02d21dbd146646c41b2b4f42627a0c9ebd \
-  --candidate f556494c8b70a6b25cb1d27acdadf1040d337218
+  --candidate a7c024f6f53cce0f8e43825845c5ad6965c9c792
 ```
 
 On the GPU machine, with the same locked workspace and prepared `_inputs/`:
diff --git a/tools/benchmarks/idle_worker/ready_generation_result.md b/tools/benchmarks/idle_worker/ready_generation_result.md
deleted file mode 100644
index 588ab12fe..000000000
--- a/tools/benchmarks/idle_worker/ready_generation_result.md
+++ /dev/null
@@ -1,102 +0,0 @@
-The new fixed three-arm L4 comparison completed all 18 trials. It keeps the sequential latency benefit and improves C8 burst fusion over the prior `sleep(0)` implementation, but it does not meet the issue's strict one-batch requirement for every HTTP burst. The result remains `completed_valid: true`, `qualification_passed: false`, with the existing human review hold unchanged.
-
-The exact source arms were A: original timed coalescing `b0c26b02d21dbd146646c41b2b4f42627a0c9ebd`; B: current PR658 `8845eab5274ad571e0270c9d8c842035b5b5a426`; C: ready-callback candidate `a7c024f6f53cce0f8e43825845c5ad6965c9c792`. C preserves the PR history and changes only the lone-idle scheduling checkpoint plus its regression tests. A callback completes a Future so the waiting worker resumes after tasks created by already-ready producer callbacks. It is a finite scheduling opportunity; later network bodies or unfinished preprocessing can still miss it.
-
-All arms used the same single NVIDIA L4, 8 CPU cores, 32 GiB memory, pinned `knowledgator/gliclass-large-v1.0` revision `ae8a18644e84cfda53325bd91c7941f2f6fd0ccb`, and actual local SIE HTTP API through `sie_sdk.SIEClient`. The observed model parameters were FP16 on CUDA. The default profile retained its existing bucketed CUDA graphs, threshold 0, batch-token limit 16384, and `idle_coalesce_ms=3`. Each fresh server used asyncio, four Torch intra-op threads and one inter-op thread. Python 3.12.12, Torch 2.9.1+cu129, Transformers 4.57.6, GLiClass 0.1.17, gliformer 0.1.2, httpx 0.28.1, and uvicorn 0.46.0 were admitted exactly. CPU scheduling/cancellation controls also cover uvloop; this GPU experiment did not use uvloop.
-
-The six fixed triplets were ABC, CBA, BCA, ACB, CAB, BAC. Every trial warmed C1/C8/C32 with 64 calls each, measured 100 sequential calls, four idle bursts each at C8 and C32, and two 12-second closed-loop throughput phases, each preceded by 128 warmup calls. Throughput includes the full drain after the admission cutoff. All 119,432 SDK calls, including warmups, completed with one HTTP 200 response each. Full request/response bytes, SDK results, timing headers, and request-to-worker-batch joins were retained; no trial, request, pair, or outlier was excluded.
-
-| Observation | A: timed coalescing | B: `sleep(0)` | C: ready callback |
-|---|---:|---:|---:|
-| Sequential queue samples below 1 ms | 0/600 | 600/600 | 600/600 |
-| Per-trial queue median range | 3.395–3.440 ms | 0.090–0.100 ms | 0.100–0.110 ms |
-| Per-trial queue nearest-rank p95 range | 3.84–4.27 ms | 0.11–0.13 ms | 0.13–0.17 ms |
-| Maximum sequential queue time | 4.44 ms | 0.31 ms | 0.25 ms |
-| Per-trial SDK latency p95 range | 14.469–15.779 ms | 10.840–11.373 ms | 10.987–12.363 ms |
-| C8 bursts forming one worker batch | 24/24 | 3/24 | 18/24 |
-| C32 bursts forming one worker batch | 0/24 | 0/24 | 0/24 |
-
-C's six remaining C8 bursts split `[1, 7]`. At C32, the original baseline also split every burst; this does not waive the issue's one-batch criterion. The full ordered pattern inventory is below.
-
-C/A throughput geometric ratios were 1.01095 at C8 (pointwise 95% interval 0.99637–1.02574) and 1.00010 at C32 (0.97286–1.02811). C/B ratios were 0.99319 (0.97069–1.01621) and 0.98980 (0.97423–1.00561). These are six paired fresh-process observations on one shared device/session. There is no predefined equivalence margin, so intervals containing 1 do not establish "unchanged within noise" or statistical equivalence. All twelve directions, individual ratios, and paired log dispersions are retained below.
-
-All responses retained the same classification label sets, ordered label lists, winning labels, and other nonnumeric output structure. Sequential outputs matched the fixed first-A reference exactly. The largest direct score difference over all phases was 0.00390625 in each arm, and the same maximum range occurred within each arm's own repeats. Those differences remain visible; no output tolerance or task-quality claim is introduced.
-
-The earlier six-trial yield result and its C8 split failure remain unchanged (`results.json` SHA256 `0c89a0b4693cffe29225415eace97e6551c2b426ca4eb99693b5da75c347ed4b`). The first attempt at this new three-arm protocol ended before inference because the cache verifier compared a lexical mount alias against a canonical root. Its failure was retained; run2 changed only that verified path-handling defect and attempt identities, after CPU alias/escape/retarget controls. No measured workload, output policy, trial order, or timing bound was changed in response to performance results.
-
-The experiment includes identical output capture and batch tracing in every arm. It shares one device, image, and session across the balanced process order, so thermal, cache, JIT, and process variation remain possible. It measures the Python worker's HTTP path, not gateway/sidecar-only scheduling. The method does not predict arbitrary later network arrivals. The measured candidate has not yet been integrated with newer main; any later integration needs explicit source mapping and fresh CI/review.
-
-The complete archive was verified before analysis. The provider was stopped and independently observed with zero containers and no tasks, every owned local process was reaped, and all experiment leases were released. No further GPU attempt is proposed.
-
-Evidence SHA256 values: frozen method `fcf1aa6eac0542065c12f93c952d60b6b5d2aa7413455b2e3a97fa3c8437a8c2`; source-input manifest `398f68d9d62639a389fdb3b07850f19120a7eeda14d6edcaecf723e291147c35`; unchanged analyzer `b323ceedeb056f27bf40729ebd1749b8629a3b29bc4887ad866a83a55c859c08`; complete analysis `51e0d08bb724d08a5b270b3ca051ff917462c5b3cbed709553bbbba52a2c2c1f`; terminal receipt `cb99a4ee8e090a7a0fd7de8c0db41549b547bbce92fd1bc8ef332eb86b9d5998`; full 470-member archive `9933b24ff54bc85d6c617ba053aaf27a34ada8bc2ba69370f3d3dc70027fa9ed` (33,591,363 compressed bytes; 332,647,329 raw bytes). These identify retained artifacts; they are not a claim that private storage locations are publicly accessible. The complete method and result tables follow; the checked-in two-arm harness and its reproducibility limits are described at the end.
-
-All 18 trials, in execution order (nearest-rank p95; block/trial numbers below start at zero).
-
-| Trial | Block | Arm | Queue median / p95 (ms) | SDK p95 (ms) | C8 requests/s | C32 requests/s | C8 fused | C32 fused |
-|---:|---:|:---:|---:|---:|---:|---:|---:|---:|
-| 0 | 0 | A | 3.410 / 4.18 | 14.969 | 209.678 | 284.718 | 4/4 | 0/4 |
-| 1 | 0 | B | 0.090 / 0.11 | 10.840 | 206.797 | 288.443 | 0/4 | 0/4 |
-| 2 | 0 | C | 0.110 / 0.13 | 11.222 | 211.718 | 291.785 | 3/4 | 0/4 |
-| 3 | 1 | C | 0.110 / 0.13 | 11.296 | 209.282 | 275.702 | 3/4 | 0/4 |
-| 4 | 1 | B | 0.090 / 0.12 | 11.148 | 208.874 | 279.815 | 0/4 | 0/4 |
-| 5 | 1 | A | 3.440 / 3.84 | 14.469 | 210.761 | 282.735 | 4/4 | 0/4 |
-| 6 | 2 | B | 0.100 / 0.13 | 11.296 | 215.243 | 282.947 | 2/4 | 0/4 |
-| 7 | 2 | C | 0.100 / 0.13 | 10.987 | 209.845 | 274.439 | 2/4 | 0/4 |
-| 8 | 2 | A | 3.410 / 4.23 | 15.779 | 202.622 | 284.480 | 4/4 | 0/4 |
-| 9 | 3 | A | 3.395 / 4.27 | 15.429 | 200.816 | 276.213 | 4/4 | 0/4 |
-| 10 | 3 | C | 0.110 / 0.14 | 11.683 | 201.679 | 280.336 | 4/4 | 0/4 |
-| 11 | 3 | B | 0.100 / 0.13 | 11.373 | 208.513 | 282.631 | 0/4 | 0/4 |
-| 12 | 4 | C | 0.110 / 0.17 | 12.363 | 211.302 | 280.297 | 4/4 | 0/4 |
-| 13 | 4 | A | 3.430 / 4.13 | 14.901 | 209.259 | 281.434 | 4/4 | 0/4 |
-| 14 | 4 | B | 0.090 / 0.12 | 10.985 | 214.833 | 286.102 | 0/4 | 0/4 |
-| 15 | 5 | B | 0.100 / 0.12 | 11.329 | 210.389 | 283.979 | 1/4 | 0/4 |
-| 16 | 5 | A | 3.435 / 4.26 | 14.853 | 209.371 | 276.820 | 4/4 | 0/4 |
-| 17 | 5 | C | 0.105 / 0.13 | 11.233 | 212.252 | 284.243 | 2/4 | 0/4 |
-
-All ratio directions use six process pairs. Intervals are pointwise, two-sided 95% t intervals on paired log ratios (df = 5), without an equivalence margin.
-
-| Throughput ratio | Six block ratios | Geometric mean | 95% interval | Paired log SD |
-|---|---|---:|---|---:|
-| throughput32:A/B | 0.98709, 1.01044, 1.00542, 0.97729, 0.98369, 0.97479 | 0.98969 | 0.97433–1.00530 | 0.014912 |
-| throughput32:A/C | 0.97578, 1.02551, 1.03659, 0.98529, 1.00406, 0.97389 | 0.99990 | 0.97266–1.02790 | 0.026316 |
-| throughput32:B/A | 1.01308, 0.98967, 0.99461, 1.02323, 1.01658, 1.02586 | 1.01041 | 0.99473–1.02635 | 0.014912 |
-| throughput32:B/C | 0.98855, 1.01492, 1.03100, 1.00818, 1.02071, 0.99907 | 1.01031 | 0.99442–1.02645 | 0.015102 |
-| throughput32:C/A | 1.02482, 0.97513, 0.96470, 1.01493, 0.99596, 1.02681 | 1.00010 | 0.97286–1.02811 | 0.026316 |
-| throughput32:C/B | 1.01159, 0.98530, 0.96993, 0.99188, 0.97971, 1.00093 | 0.98980 | 0.97423–1.00561 | 0.015102 |
-| throughput8:A/B | 1.01393, 1.00904, 0.94136, 0.96309, 0.97406, 0.99516 | 0.98243 | 0.95311–1.01266 | 0.028872 |
-| throughput8:A/C | 0.99036, 1.00707, 0.96558, 0.99572, 0.99033, 0.98643 | 0.98917 | 0.97490–1.00365 | 0.013844 |
-| throughput8:B/A | 0.98626, 0.99104, 1.06229, 1.03833, 1.02664, 1.00486 | 1.01788 | 0.98750–1.04920 | 0.028872 |
-| throughput8:B/C | 0.97676, 0.99805, 1.02572, 1.03389, 1.01671, 0.99122 | 1.00686 | 0.98405–1.03020 | 0.021838 |
-| throughput8:C/A | 1.00973, 0.99298, 1.03565, 1.00430, 1.00976, 1.01376 | 1.01095 | 0.99637–1.02574 | 0.013844 |
-| throughput8:C/B | 1.02380, 1.00196, 0.97492, 0.96722, 0.98357, 1.00885 | 0.99319 | 0.97069–1.01621 | 0.021838 |
-
-Every measured idle-burst pattern (ordered worker batch sizes; 24 bursts per arm and concurrency).
-
-| Arm | Concurrency | Pattern counts |
-|:---:|---:|---|
-| A | 8 | `[8]` × 24 |
-| A | 32 | `[19, 13]` × 1; `[21, 11]` × 1; `[25, 7]` × 3; `[26, 6]` × 6; `[27, 5]` × 6; `[28, 4]` × 4; `[29, 3]` × 2; `[30, 2]` × 1 |
-| B | 8 | `[1, 7]` × 21; `[8]` × 3 |
-| B | 32 | `[1, 15, 16]` × 2; `[1, 31]` × 15; `[25, 7]` × 1; `[27, 5]` × 3; `[28, 4]` × 3 |
-| C | 8 | `[1, 7]` × 6; `[8]` × 18 |
-| C | 32 | `[1, 11, 20]` × 1; `[1, 12, 19]` × 1; `[1, 23, 8]` × 1; `[1, 31]` × 2; `[22, 10]` × 1; `[24, 8]` × 1; `[26, 6]` × 6; `[27, 5]` × 4; `[28, 4]` × 6; `[31, 1]` × 1 |
-
-The public serving source and original two-arm workload are in this repository. The existing `tools/benchmarks/idle_worker/README.md` documents source preparation and execution on a caller-owned GPU. Once the measured candidate commit is published, a fresh original-versus-candidate comparison can use:
-
-```sh
-mise exec -- python tools/benchmarks/idle_worker/prepare.py \
-  --baseline b0c26b02d21dbd146646c41b2b4f42627a0c9ebd \
-  --candidate a7c024f6f53cce0f8e43825845c5ad6965c9c792
-mise exec -- python tools/benchmarks/idle_worker/run.py \
-  --output /tmp/sie-idle-worker-results
-```
-
-These commands run the existing two-arm benchmark, not the completed three-arm experiment verbatim. The 18-trial experiment retained additional full request/response bytes, source/runtime/cache admission, exact process ownership, failure persistence, and a separate full-row analyzer. Those additions are not yet packaged as a public one-command reproducer. Its fixed method is specified above so that the difference is visible; do not treat a new run of the older harness as a hash-identical replay or pool it with this result.
-
-For a faithful independent implementation of the three-arm method, preserve all six listed triplets on the same device, fresh actual server/SDK instances, the exact environment/profile and fixed public checkpoint, all warmup/count/duration/drain rules, 50 ms before each idle burst, and all original raw output/batch observations. The full source archive path inventory consisted of server source, SDK source, model catalog, bundle definitions, and `uv.lock`; the same lock SHA256 was `c1286da62744012fe54f07cd7ec740d1d1a69db0e2db75c8a0a0a8007ac3a397`. The A/B/C archive SHA256 values were `0926ef65a420c437f9f5f2147a44f0b6ed07e0cd51cf7a376a2bb99eccee2257`, `6e8ba5a26bd3c2f223cd2bb05351e53307c0180d645a4541678a9fb8fded29d0`, and `e2380107510882bf8c273d6139e79d63412a31327fe4365a45f335656512928f`.
-
-The fixed active-run ceiling was 1800 seconds, with 180 seconds active plus at most 30 seconds cleanup for each admitted fresh trial. The experiment stopped on the first failure, required enough remaining budget for a complete trial and cleanup, and did not append blocks, shorten windows, or omit outliers. Independent process/phase identity and every complete request/wire/batch join are necessary to reproduce the integrity assessment; timing similarity alone is insufficient.
-
-The seven required public snapshot files totaled 1,763,619,203 bytes. The checkpoint `model.safetensors` was 1,752,499,444 bytes with SHA256 `9d6bafde8a1a43bf5563fd6296f8934752d23462de9113b0b8ac56ecc7cdddec`; the required-file manifest SHA256 was `c003d0e73df9bb7d23337c6d6a54312dc07e2639704e47981b15b158812ab053`. Full file content, revision, and resolved cache paths were checked before model work. A cache or environment failure is an incomplete attempt, not a reason to substitute inputs silently.
-
-Independent reviewers recomputed all 119,432 request joins, 288 phases, 19,221 post-forward worker batches, every ratio and output diagnostic. Worker batch records identify SIE dispatch grouping; they do not establish physical CUDA tensor shapes or the cause of a particular HTTP split. The result-review SHA256 is `9a00f78392a47f60f5eff6956fa0037dc9468d1d31e19efdf2109905c2c28b08`; the independent terminal/archive review SHA256 is `725bbd0d6f842ccceb4416175dcb3718af4998182e03c2dde496290306a5d8b2`.