diff --git a/plugins/provider-pi/src/bridge/bridge.round2.test.ts b/plugins/provider-pi/src/bridge/bridge.round2.test.ts index c75e405375..3a3634079f 100644 --- a/plugins/provider-pi/src/bridge/bridge.round2.test.ts +++ b/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): 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"; diff --git a/plugins/provider-pi/src/bridge/fake-pi-rpc.mjs b/plugins/provider-pi/src/bridge/fake-pi-rpc.mjs index 8bd294ed29..8735ad9989 100644 --- a/plugins/provider-pi/src/bridge/fake-pi-rpc.mjs +++ b/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"); } diff --git a/plugins/provider-pi/src/bridge/rpc-child.ts b/plugins/provider-pi/src/bridge/rpc-child.ts index db54ab2304..fcf2581c0e 100644 --- a/plugins/provider-pi/src/bridge/rpc-child.ts +++ b/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 | 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) {