Handle each pi stdout line in its own event-loop turn

plugins/provider-pi/src/bridge/bridge.round2.test.ts

@@ -152,6 +152,45 @@ it("a steer consumed by the run is reported accepted and named in the reply", as
   expect(harness.messages.some((m) => m.method === "error")).toBe(false);
 }, 90_000);
 
+it("a steer's ack precedes the event pi wrote in the same chunk as the prompt response", async () => {
+  // Pi answers the steer's prompt and the resumed run writes its next event
+  // back to back, so one pipe read can carry both. The bridge's order must
+  // not depend on that chunking: the ack, emitted after the request's
+  // continuation, goes out before the event, as a line-at-a-time read has it
+  // (the parity replay of pi/compaction pins the recorded order).
+  vi.stubEnv("FAKE_PI_BATCH_STEER_REPLY", "1");
+  const threadId = "thr_r2_steer_batch";
+  await harness.startThread(threadId);
+  turnStart(threadId, "/hold", "creq_ab23456789");
+  await harness.waitForDelta(threadId, (d) => d.kind === "turn.open");
+  const steer = await harness.request((nextId += 1), "turn/steer", {
+    threadId,
+    providerThreadId: threadId,
+    expectedTurnId: "turn-1",
+    clientRequestId: "creq_cd23456789",
+    input: [{ type: "text", text: "take the left path", mentions: [] }],
+    options: FULL_PERMISSION_OPTIONS,
+  });
+  expect(steer.result).toMatchObject({ threadId });
+  await harness.waitForDelta(threadId, (d) => d.kind === "turn.boundary");
+  const deltas = harness.deltasOf(threadId);
+  const queued = deltas.findIndex((d) => queueUpdateSteering(d)?.includes("take the left path") === true);
+  const accepted = deltas.findIndex((d) => d.kind === "input.accepted" && d.clientRequestId === "creq_cd23456789");
+  const consumed = deltas.findIndex((d, index) => index > queued && queueUpdateSteering(d)?.length === 0);
+  expect(queued).toBeGreaterThan(-1);
+  expect(accepted).toBeGreaterThan(queued);
+  expect(consumed).toBeGreaterThan(accepted);
+}, 90_000);
+
+/** The steering queue an `unhandled` delta carries when it wraps pi's `queue_update`. */
+function queueUpdateSteering(delta: Record<string, unknown>): unknown[] | null {
+  if (delta.kind !== "unhandled") return null;
+  const raw = delta.raw as { params?: { message?: { type?: unknown; steering?: unknown } } } | undefined;
+  const message = raw?.params?.message;
+  if (message?.type !== "queue_update" || !Array.isArray(message.steering)) return null;
+  return message.steering;
+}
+
 it("a steer still queued when the run ends is reported dropped through the delivery barrier, not a timer", async () => {
   vi.stubEnv("FAKE_PI_DROP_STEER_AT_END", "1");
   const threadId = "thr_r2_steer_drop";

plugins/provider-pi/src/bridge/fake-pi-rpc.mjs

@@ -51,8 +51,10 @@
  *   FAKE_PI_NO_SESSION_START=1 never emits session_start to the extension (so
  *   no `ready`); FAKE_PI_DROP_STEER_AT_END=1 ends a run with a queued steer
  *   still queued; FAKE_PI_STREAMING_AFTER_END=1 reports isStreaming after a
- *   run ended (a continuation pi is still finishing). The prompt `/die` exits
- *   the process mid-run without answering.
+ *   run ended (a continuation pi is still finishing);
+ *   FAKE_PI_BATCH_STEER_REPLY=1 writes a steer's `prompt` response and the
+ *   resumed run's first event in one stdout write (one read on the bridge's
+ *   side). The prompt `/die` exits the process mid-run without answering.
  * - `prompt` with `streamingBehavior: "steer"` during a `/hold` run is queued
  *   (`queue_update.steering`), consumed when the run resumes, and the run's
  *   reply names it.
@@ -185,8 +187,21 @@ const followUp = [];
 const steering = [];
 let endedWithStreamingFlag = false;
 
+/** A line held back to go out in one write with the next one. */
+let heldLine = null;
+
 function send(message) {
-  process.stdout.write(`${JSON.stringify(message)}\n`);
+  const line = `${JSON.stringify(message)}\n`;
+  if (heldLine === null) {
+    process.stdout.write(line);
+    return;
+  }
+  // One write, so the bridge reads both lines in one chunk.
+  process.stdout.write(`${heldLine}${line}`);
+  heldLine = null;
+}
+function holdUntilNextSend(message) {
+  heldLine = `${JSON.stringify(message)}\n`;
 }
 function respond(id, command, data) {
   send({ id, type: "response", command, success: true, ...(data === undefined ? {} : { data }) });
@@ -451,7 +466,12 @@ async function handle(command) {
         // it at its end.
         steering.push(command.message);
         queueUpdate();
-        respond(id, "prompt");
+        if (process.env.FAKE_PI_BATCH_STEER_REPLY === "1" && holdAbort) {
+          // The response goes out with the resumed run's first event.
+          holdUntilNextSend({ id, type: "response", command: "prompt", success: true });
+        } else {
+          respond(id, "prompt");
+        }
         if (holdAbort) {
           holdAbort("steer");
         }

plugins/provider-pi/src/bridge/rpc-child.ts

@@ -135,6 +135,16 @@ export class PiRpcChild {
   private readonly channelWriter: Writable | null;
   private readonly channelRecorder: ChannelRecorder | null;
   private killEscalation: ReturnType<typeof setTimeout> | null = null;
+  /**
+   * Pi's stdout lines, handled one per event-loop turn. One read can carry a
+   * response and the events pi wrote after it; handled in one synchronous
+   * loop, the next event's delivery would run before the request's
+   * continuation (a steer's ack after `await request("prompt")`), so the
+   * bridge's order would depend on pipe chunking. Yielding between lines
+   * lets the continuation finish first, the order a line-at-a-time read has.
+   */
+  private readonly stdoutLines: string[] = [];
+  private stdoutDraining = false;
 
   constructor(private readonly args: SpawnPiRpcChildArgs) {
     const launch = resolvePiLaunch(process.env);
@@ -159,7 +169,7 @@ export class PiRpcChild {
     if (stdout) {
       experimental_readBoundedLines({
         input: stdout,
-        onLine: (line) => this.handleStdoutLine(line),
+        onLine: (line) => this.queueStdoutLine(line),
         onOverflow: (bytes) => {
           process.stderr.write(`pi bridge: dropped a ${bytes}-byte stdout line\n`);
         },
@@ -328,6 +338,27 @@ export class PiRpcChild {
     stdin.write(line);
   }
 
+  private queueStdoutLine(line: string): void {
+    this.stdoutLines.push(line);
+    if (!this.stdoutDraining) {
+      this.stdoutDraining = true;
+      this.drainStdoutLine();
+    }
+  }
+
+  private drainStdoutLine(): void {
+    const line = this.stdoutLines.shift();
+    if (line === undefined) {
+      this.stdoutDraining = false;
+      return;
+    }
+    try {
+      this.handleStdoutLine(line);
+    } finally {
+      setImmediate(() => this.drainStdoutLine());
+    }
+  }
+
   private handleStdoutLine(line: string): void {
     const trimmed = line.trim();
     if (!trimmed) {