diff --git a/packages/shared/src/agent/pi-rpc-client.ts b/packages/shared/src/agent/pi-rpc-client.ts index 7190503..d335ef4 100644 --- a/packages/shared/src/agent/pi-rpc-client.ts +++ b/packages/shared/src/agent/pi-rpc-client.ts @@ -576,18 +576,23 @@ export class PiRpcClient { return; } - const pending = this.findPending(event); const control = this.findControl(event); if (control) { clearTimeout(control.timeout); this.pendingControls.delete(control.id); - if (event.type === 'error') { - control.reject(createCodeError('PI_RUNTIME_ERROR', String(event.message ?? 'Pi runtime error.'))); + if (event.type === 'error' || event.success === false) { + control.reject(createCodeError( + 'PI_RUNTIME_ERROR', + String(event.error ?? event.message ?? 'Pi runtime error.') + )); } else { control.resolve(event.data); } + return; } + const pending = this.findPending(event); + if (!pending) return; pending.trace.push({ @@ -684,6 +689,7 @@ export class PiRpcClient { private findPending(event: Record) { const id = typeof event.id === 'string' ? event.id : typeof event.requestId === 'string' ? event.requestId : undefined; if (id && this.pendingPrompts.has(id)) return this.pendingPrompts.get(id); + if (event.type === 'response') return undefined; if (this.pendingPrompts.size === 1) return Array.from(this.pendingPrompts.values())[0]; return undefined; } diff --git a/packages/shared/src/agent/pi-runtime-agent-backend.test.ts b/packages/shared/src/agent/pi-runtime-agent-backend.test.ts index 0425ff5..33b0ccf 100644 --- a/packages/shared/src/agent/pi-runtime-agent-backend.test.ts +++ b/packages/shared/src/agent/pi-runtime-agent-backend.test.ts @@ -249,11 +249,65 @@ describe('PiRpcClient', () => { await stream.abort(); await consume; - expect(events).toContain('message_update'); + expect(events).toEqual(['message_update']); + expect(events).not.toContain('response'); + expect(events).not.toContain('abort'); expect(result).toMatchObject({ aborted: true }); expect((result as { answer: string }).answer).toBe('Partial'); }); + it('ignores stale prompt responses while a prompt is active', async () => { + const client = new PiRpcClient({ + spawnProcess: createSpawn(() => + new FakePiProcess((line, proc) => { + if (line.type !== 'prompt') return; + + proc.writeEvent({ + id: 'stale-prompt-id', + type: 'response', + command: 'prompt', + success: false, + error: 'stale prompt response', + }); + proc.writeEvent({ + id: line.id, + type: 'response', + command: 'prompt', + success: true, + }); + proc.writeEvent({ + id: line.id, + type: 'agent_end', + messages: [{ role: 'assistant', content: [{ type: 'text', text: 'completed' }] }], + }); + }) + ), + }); + + await expect(client.prompt('hello')).resolves.toMatchObject({ answer: 'completed' }); + }); + + it('falls back to the only active prompt for id-less agent events', async () => { + const client = new PiRpcClient({ + spawnProcess: createSpawn(() => + new FakePiProcess((line, proc) => { + if (line.type !== 'prompt') return; + + proc.writeEvent({ + type: 'message_update', + assistantMessageEvent: { type: 'text_delta', delta: 'id-less ' }, + }); + proc.writeEvent({ + type: 'agent_end', + messages: [{ role: 'assistant', content: [{ type: 'text', text: 'event' }] }], + }); + }) + ), + }); + + await expect(client.prompt('hello')).resolves.toMatchObject({ answer: 'event' }); + }); + it('rejects when Pi refuses a prompt before execution', async () => { const client = new PiRpcClient({ spawnProcess: createSpawn(() =>