From b773f0f2e29f628fa09442e8a3f1431c418d1ee7 Mon Sep 17 00:00:00 2001
From: Priya Raman <priya@paperclip.ing>
Date: Thu, 3 Sep 2026 23:59:39 +0000
Subject: [PATCH] test(plugin-worker): remove wall-clock race from duplex
 buffered-replay tests

The two buffered-replay tests used a fixed 60 ms sleep as a barrier before
they attached the data listener. The barrier raced the subprocess start and
the stdio latency. Under load the three data frames did not always buffer
within 60 ms, so the synchronous drain saw a partial buffer and the
synchronous assertion failed.

Replace the fixed sleep with a deterministic barrier. Each test now scripts
an exit after the three data frames. The host reads the worker stdout line
by line, so it buffers every data frame before it reads the exit. The exit
settles the wait, so `await session.wait()` proves the host holds all
pre-bind frames before any listener attaches. The test then attaches the
listener and the drain delivers every buffered chunk. Both tests keep the
non-batch setImmediate buffer-then-drain path and lose no coverage.

Co-authored-by: Paperclip <noreply@paperclip.ing>
---
 .../plugin-worker-manager-duplex.test.ts      | 25 +++++++++++++------
 1 file changed, 18 insertions(+), 7 deletions(-)

diff --git a/server/src/__tests__/plugin-worker-manager-duplex.test.ts b/server/src/__tests__/plugin-worker-manager-duplex.test.ts
index d0c28781d3..45084757ca 100644
--- a/server/src/__tests__/plugin-worker-manager-duplex.test.ts
+++ b/server/src/__tests__/plugin-worker-manager-duplex.test.ts
@@ -130,17 +130,22 @@ describe("plugin worker manager duplex channel route", () => {
       const session = await handle.openDuplexChannel(
         duplexOpenInput({
           data: [{ chunk: "one" }, { chunk: "two" }, { chunk: "three" }],
+          exitCode: 0,
         }),
       );
-      // Wait so the three data notifications arrive and buffer before a listener
-      // attaches. The drain then delivers them in order.
-      await new Promise((resolve) => setTimeout(resolve, 60));
+      // The worker writes the three data notifications and the exit in one
+      // stdout write. The host reads worker stdout line by line, so it buffers
+      // all three data frames before it reads the exit. The exit settles the
+      // wait, so the wait is a deterministic barrier: once it resolves, the
+      // host holds all three frames and no listener has attached yet. This
+      // barrier replaces a fixed sleep, so the test does not race the
+      // subprocess start or the stdio latency.
+      await session.wait();
       const chunks: string[] = [];
       // The session streams raw `Uint8Array` chunks. Decode each one back to
       // text, so the assertion below compares the plain-text payload the
       // fixture directive scripted.
       session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
-      await vi.waitFor(() => expect(chunks.length).toBe(3));
       expect(chunks).toEqual(["one", "two", "three"]);
       await session.close();
     } finally {
@@ -213,11 +218,17 @@ describe("plugin worker manager duplex channel route", () => {
       const session = await handle.openDuplexChannel(
         duplexOpenInput({
           data: [{ chunk: "one" }, { chunk: "boom" }, { chunk: "three" }],
+          exitCode: 0,
         }),
       );
-      // Wait so the three data notifications arrive and buffer before a listener
-      // attaches. The drain then delivers them in order.
-      await new Promise((resolve) => setTimeout(resolve, 60));
+      // The worker writes the three data notifications and the exit in one
+      // stdout write. The host reads worker stdout line by line, so it buffers
+      // all three data frames before it reads the exit. The exit settles the
+      // wait, so the wait is a deterministic barrier for "the host holds every
+      // pre-bind frame and no listener has attached". This barrier replaces a
+      // fixed sleep, so the test does not race the subprocess start or the
+      // stdio latency.
+      await session.wait();
       const chunks: string[] = [];
       // The listener throws on one buffered chunk. The manager catches the throw
       // inside the drain, so it does not escape `onData` and every buffered chunk