mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-05 06:24:37 +00:00
fix(cli): buffer Pi prompts until RPC startup ready (#1146)
* 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
HeavyGee
@@ -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, []);
|
||||
|
||||
@@ -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': {
|
||||
|
||||
+46
-10
@@ -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');
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user