From 7ca4e71fa0ff327d11c639caf3543fad168c239c Mon Sep 17 00:00:00 2001 From: HeavyGee <133152184+heavygee@users.noreply.github.com> Date: Fri, 24 Jul 2026 03:52:06 +0100 Subject: [PATCH] fix(cli): buffer Pi prompts until RPC startup ready (#1146) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(cli): buffer Pi prompts until RPC startup ready A prompt POSTed immediately after spawn (a supported handoff pattern used by hapi-ping-peer and intake scripts) could reach `pi --mode rpc` before its `new_session`/`get_state` startup finished, wedging the turn: `agent_start` then silence, no tool calls. The socket goes `active` (spawn success) well before Pi's session is initialized, so `active` is not a safe ready signal for Pi. Gate outbound prompt/steer sends behind a startup ready gate on PiSession: `runWhenReady()` delivers immediately once ready, else buffers FIFO; `markReady()` fires on the first `get_state` response (the signal that persists `metadata.piSessionId`, which working callers already wait for) and drains the buffer in order. A 30s unref'd fallback timer force-drains if `get_state` never lands, degrading to prior send-anyway behaviour rather than swallowing the message forever. Fixes #1143 Co-authored-by: Cursor * fix(cli): honor cancel-queued-message for buffered Pi prompts Addresses the MAJOR review finding on the startup ready-buffer: while a prompt is held behind runWhenReady, the hub can send cancel-queued-message for its localId. Pi registered no onCancelQueuedMessage handler, so ApiSessionClient acked removed:false, the hub marked the row invoked, yet the buffered closure still drained on get_state and fired the cancelled prompt. Carry the localId with each buffered send and add PiSession.cancelBufferedMessage, then register apiSession.onCancelQueuedMessage so a cancel drops the still-buffered prompt (returns true) instead of sending it. Once drained to Pi it cannot be recalled — returns false, matching the other agents' queue.cancelByLocalId best-effort semantics. Co-authored-by: Cursor --------- Co-authored-by: Cursor --- cli/src/pi/loop.test.ts | 33 +++++++++ cli/src/pi/loop.ts | 7 ++ cli/src/pi/runPi.ts | 56 +++++++++++++--- cli/src/pi/session.test.ts | 133 +++++++++++++++++++++++++++++++++++++ cli/src/pi/session.ts | 57 ++++++++++++++++ 5 files changed, 276 insertions(+), 10 deletions(-) create mode 100644 cli/src/pi/session.test.ts diff --git a/cli/src/pi/loop.test.ts b/cli/src/pi/loop.test.ts index 626abcd0..e611a313 100644 --- a/cli/src/pi/loop.test.ts +++ b/cli/src/pi/loop.test.ts @@ -241,6 +241,39 @@ describe('wireTransportEvents', () => { expect(session.client.updateMetadata).toHaveBeenCalledWith(expect.any(Function)); }); + it('marks session ready on get_state response (drains buffered sends) — issue #1143', () => { + const transport = createMockTransport(); + wireTransportEvents(transport, session, []); + + // A prompt buffered before Pi finished startup must not run yet. + const buffered = vi.fn(); + session.runWhenReady(buffered); + expect(buffered).not.toHaveBeenCalled(); + expect(session.isReady).toBe(false); + + emitEvent({ + type: 'response', + command: 'get_state', + success: true, + data: { sessionId: 'pi-session-ready' }, + }); + + // get_state landing is the ready signal — buffered work drains. + expect(session.isReady).toBe(true); + expect(buffered).toHaveBeenCalledTimes(1); + }); + + it('marks session ready on get_state even when sessionId is absent', () => { + // Robustness: readiness must not hinge on Pi always echoing sessionId, + // otherwise a missing field would buffer prompts forever. + const transport = createMockTransport(); + wireTransportEvents(transport, session, []); + + emitEvent({ type: 'response', command: 'get_state', success: true, data: {} }); + + expect(session.isReady).toBe(true); + }); + it('handles error response — sends session event', () => { const transport = createMockTransport(); wireTransportEvents(transport, session, []); diff --git a/cli/src/pi/loop.ts b/cli/src/pi/loop.ts index d4f48256..58841aea 100644 --- a/cli/src/pi/loop.ts +++ b/cli/src/pi/loop.ts @@ -154,6 +154,13 @@ function handleResponse( switch (command) { case 'get_state': { handleGetState(response.data, session); + // Pi has finished startup init (this is the response that persists + // metadata.piSessionId — the signal working callers already wait + // for). Release any prompts buffered during the spawn window so they + // reach an initialized Pi session instead of wedging (issue #1143). + // markReady is idempotent; a missing sessionId still flips ready so + // buffered prompts are never swallowed forever. + session.markReady(); break; } case 'set_model': { diff --git a/cli/src/pi/runPi.ts b/cli/src/pi/runPi.ts index f3417483..cdfa7990 100644 --- a/cli/src/pi/runPi.ts +++ b/cli/src/pi/runPi.ts @@ -14,6 +14,11 @@ import type { SlashCommandsResponse } from '@hapi/protocol/apiTypes'; import type { ListPiModelsResponse } from '@hapi/protocol/apiTypes'; import { RPC_METHODS } from '@hapi/protocol/rpcMethods'; +// Grace period before force-draining prompts buffered during Pi startup when no +// get_state response arrives. Comfortably above the 10s Pi RPC timeout so a slow +// but healthy startup still flips ready via get_state first (issue #1143). +const PI_READY_FALLBACK_MS = 30_000; + export async function runPi(opts: { startedBy?: 'runner' | 'terminal'; startingMode?: 'local' | 'remote'; @@ -312,18 +317,37 @@ export async function runPi(opts: { // --- User message handler --- apiSession.onUserMessage((message, localId) => { const formattedText = formatMessageWithAttachments(message.content.text, message.content.attachments); - if (piSession.piIsStreaming) { - // Steer does not start a new turn, so the localId would never be - // drained by turn_start. Mark it consumed immediately so it does - // not poison the FIFO for the next real prompt. - transport.send({ type: 'steer', message: formattedText }); - if (localId) piSession.emitMessagesConsumed([localId]); - } else { - if (localId) pendingLocalIds.push(localId); - transport.send({ type: 'prompt', message: formattedText }); - } + // Gate the send behind Pi startup readiness. A prompt POSTed immediately + // after spawn (supported handoff pattern — hapi-ping-peer, intake + // scripts) would otherwise reach Pi before new_session/get_state finish + // and wedge the turn. runWhenReady delivers now if ready, else buffers + // FIFO until the first get_state response drains it (issue #1143). + // piIsStreaming is evaluated at delivery time so a message buffered + // before startup still takes the prompt path. + piSession.runWhenReady(() => { + if (piSession.piIsStreaming) { + // Steer does not start a new turn, so the localId would never be + // drained by turn_start. Mark it consumed immediately so it does + // not poison the FIFO for the next real prompt. + transport.send({ type: 'steer', message: formattedText }); + if (localId) piSession.emitMessagesConsumed([localId]); + } else { + if (localId) pendingLocalIds.push(localId); + transport.send({ type: 'prompt', message: formattedText }); + } + }, localId); }); + // --- Cancel-queued-message handler --- + // A prompt buffered during the startup window (runWhenReady) can be cancelled + // by the hub before it drains. Without this, ApiSessionClient acks + // removed:false, the hub marks the row invoked, yet the closure would still + // fire the cancelled prompt on get_state. Dropping it from the buffer keeps + // the queued-message cancel contract intact (issue #1143 review — MAJOR). + // Once sent to Pi it cannot be recalled — return false (best-effort), which + // matches the other agents' queue.cancelByLocalId semantics. + apiSession.onCancelQueuedMessage((localId) => piSession.cancelBufferedMessage(localId)); + // --- Abort handler --- // Only cancel the current turn, keep session alive for next prompt. // Pi's `abort` command cancels the active turn but the process stays in RPC mode. @@ -351,6 +375,17 @@ export async function runPi(opts: { // --- Run --- let crashed = false; + // Fallback: if Pi never returns get_state (never flips ready), force-drain + // buffered prompts after a grace period rather than swallowing them forever. + // This degrades to pre-fix behaviour (send anyway) instead of something + // worse. markReady is idempotent, so a real get_state that lands first wins. + const readyFallback = setTimeout(() => { + if (!piSession.isReady) { + logger.debug('[pi] get_state ready signal not seen within grace — draining buffered messages'); + piSession.markReady(); + } + }, PI_READY_FALLBACK_MS); + readyFallback.unref?.(); try { transport.start(); transport.send({ type: 'new_session' }); @@ -395,6 +430,7 @@ export async function runPi(opts: { lifecycle.setSessionEndReason('error'); logger.debug('[pi] Loop error:', error); } finally { + clearTimeout(readyFallback); if (!crashed && !lifecycle.hasExplicitSessionEndReason()) { lifecycle.setSessionEndReason('completed'); } diff --git a/cli/src/pi/session.test.ts b/cli/src/pi/session.test.ts new file mode 100644 index 00000000..073b3acc --- /dev/null +++ b/cli/src/pi/session.test.ts @@ -0,0 +1,133 @@ +import { describe, it, expect, vi } from 'vitest'; +import { PiSession } from './session'; + +vi.mock('@/ui/logger', () => ({ + logger: { + debug: vi.fn(), + warn: vi.fn(), + info: vi.fn(), + }, +})); + +function createMockSession(): PiSession { + return new PiSession({ + api: {} as any, + client: { + keepAlive: vi.fn(), + updateMetadata: vi.fn(), + sendAgentMessage: vi.fn(), + emitMessagesConsumed: vi.fn(), + sendSessionEvent: vi.fn(), + } as any, + path: '/tmp/test', + logPath: '/tmp/test.log', + startedBy: 'terminal', + startingMode: 'local', + }); +} + +// --- Ready gate + outbound buffer (Pi RPC ready-race, issue #1143) --- +// +// A prompt POSTed immediately after spawn used to be sent to Pi before +// `new_session`/`get_state` finished, wedging the turn (agent_start then +// silence). runWhenReady buffers such sends until markReady() (fired when Pi's +// get_state response lands), then drains them FIFO. + +describe('PiSession ready gate', () => { + it('starts not ready', () => { + const session = createMockSession(); + expect(session.isReady).toBe(false); + }); + + it('buffers work until markReady, then drains FIFO', () => { + const session = createMockSession(); + const order: number[] = []; + + session.runWhenReady(() => order.push(1)); + session.runWhenReady(() => order.push(2)); + session.runWhenReady(() => order.push(3)); + + // Nothing runs before ready. + expect(order).toEqual([]); + + session.markReady(); + + // Drained in the order they were enqueued. + expect(order).toEqual([1, 2, 3]); + expect(session.isReady).toBe(true); + }); + + it('runs work immediately once ready', () => { + const session = createMockSession(); + session.markReady(); + + const fn = vi.fn(); + session.runWhenReady(fn); + + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('markReady is idempotent — does not re-run drained work', () => { + const session = createMockSession(); + const fn = vi.fn(); + session.runWhenReady(fn); + + session.markReady(); + session.markReady(); + + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('preserves FIFO across mixed buffered + post-ready enqueues', () => { + const session = createMockSession(); + const order: string[] = []; + + session.runWhenReady(() => order.push('buffered-1')); + session.runWhenReady(() => order.push('buffered-2')); + session.markReady(); + session.runWhenReady(() => order.push('live-3')); + + expect(order).toEqual(['buffered-1', 'buffered-2', 'live-3']); + }); +}); + +// --- cancel-queued-message contract (issue #1143 review — MAJOR) --- +// +// A prompt buffered during the startup window can be cancelled by the hub +// before it drains. cancelBufferedMessage must drop it (so it never fires) and +// report removed:true; anything already drained or unknown reports false so the +// hub keeps the row as invoked (best-effort, like the other agents). + +describe('PiSession cancelBufferedMessage', () => { + it('drops a buffered send by localId so it never drains', () => { + const session = createMockSession(); + const fired: string[] = []; + + session.runWhenReady(() => fired.push('keep-1'), 'id-1'); + session.runWhenReady(() => fired.push('cancel-2'), 'id-2'); + session.runWhenReady(() => fired.push('keep-3'), 'id-3'); + + expect(session.cancelBufferedMessage('id-2')).toBe(true); + + session.markReady(); + + // Cancelled prompt never fired; FIFO preserved for survivors. + expect(fired).toEqual(['keep-1', 'keep-3']); + }); + + it('returns false when the localId is not buffered', () => { + const session = createMockSession(); + session.runWhenReady(() => {}, 'id-1'); + + expect(session.cancelBufferedMessage('unknown')).toBe(false); + }); + + it('returns false after the message has already drained', () => { + const session = createMockSession(); + session.runWhenReady(() => {}, 'id-1'); + session.markReady(); + + // Already sent to Pi — cannot be recalled. + expect(session.cancelBufferedMessage('id-1')).toBe(false); + }); +}); diff --git a/cli/src/pi/session.ts b/cli/src/pi/session.ts index e5740fe3..3c024759 100644 --- a/cli/src/pi/session.ts +++ b/cli/src/pi/session.ts @@ -46,6 +46,18 @@ export class PiSession { // RPC resolver — initialized by wireTransportEvents, session-scoped rpcResolver: PiRpcResolver | null = null; + // Startup ready gate (issue #1143). Pi's socket goes `active` (spawn success) + // before `pi --mode rpc` finishes `new_session`/`get_state`, so a prompt sent + // in that window reaches Pi before its session is initialized and wedges + // (agent_start, then silence). Outbound sends that assume a live Pi session + // are queued via runWhenReady() and drained FIFO once markReady() fires (on + // the first get_state response). + private piReady = false; + // Buffered sends carry their localId so a cancel-queued-message that arrives + // while a prompt is still held (before drain) can drop it instead of firing + // a cancelled prompt on markReady (issue #1143 review — MAJOR). + private readyQueue: Array<{ localId?: string; fn: () => void }> = []; + private keepAliveInterval: NodeJS.Timeout | null = null; constructor(opts: { @@ -76,6 +88,51 @@ export class PiSession { this.currentThinkingLevel = undefined; } + /** True once Pi RPC startup has completed and buffered sends have drained. */ + get isReady(): boolean { + return this.piReady; + } + + /** + * Run `fn` now if Pi startup is ready, else buffer it FIFO until markReady(). + * Used to gate outbound prompt/steer sends so they never reach Pi before its + * session is initialized (issue #1143). Pass the message `localId` so a + * cancel-queued-message can drop it while still buffered. + */ + runWhenReady(fn: () => void, localId?: string): void { + if (this.piReady) { + fn(); + return; + } + this.readyQueue.push({ localId, fn }); + } + + /** + * Drop a still-buffered send by localId (cancel-queued-message contract). + * Returns true if it was buffered and removed (so the hub un-queues the row), + * false if it was already drained/sent to Pi or never buffered (best-effort, + * mirrors the other agents' queue.cancelByLocalId semantics). + */ + cancelBufferedMessage(localId: string): boolean { + const idx = this.readyQueue.findIndex((item) => item.localId === localId); + if (idx === -1) return false; + this.readyQueue.splice(idx, 1); + return true; + } + + /** + * Signal that Pi RPC startup is complete (first get_state response). + * Drains buffered sends in enqueue order. Idempotent — later get_state + * responses (or the startup fallback timer) are no-ops. + */ + markReady(): void { + if (this.piReady) return; + this.piReady = true; + const queued = this.readyQueue; + this.readyQueue = []; + for (const { fn } of queued) fn(); + } + startKeepAlive(): void { this.pushKeepAlive(); this.keepAliveInterval = setInterval(() => this.pushKeepAlive(), 2000);