From ab240700753ef99058463040ec7482632b7f3264 Mon Sep 17 00:00:00 2001 From: Matt Cowger Date: Wed, 29 Jul 2026 03:41:09 -0700 Subject: [PATCH] fix(server): suppress Pi interruption stream error (#2311) * Handle Pi aborted responses after interruption * fix(server): suppress late Pi interruption terminal response --- .../server/agent/providers/pi/agent.test.ts | 56 ++++++++++++++++++ .../src/server/agent/providers/pi/agent.ts | 59 ++++++++++++++++++- 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index daf25470c..ee918354f 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -707,6 +707,62 @@ describe("PiRpcAgentSession", () => { ); }); + test("treats Pi's aborted terminal response as cancellation after an interrupt", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.abort = async () => { + fakeSession.finishTurn({ + role: "assistant", + provider: "openai-responses", + model: "gpt-5.6-terra", + responseId: "resp-aborted", + stopReason: "aborted", + errorMessage: "OpenAI Responses stream ended before a terminal response event", + content: [], + }); + }; + + const { turnId } = await session.startTurn("stop this turn"); + await session.interrupt(); + + await expect(events.nextTurnCancellation()).resolves.toEqual({ + type: "turn_canceled", + provider: "pi", + reason: "interrupted", + turnId, + }); + }); + + test("suppresses late aborted terminal response arriving after interrupt resolves", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.abort = async () => {}; + + const { turnId } = await session.startTurn("stop this turn"); + await session.interrupt(); + + await expect(events.nextTurnCancellation()).resolves.toEqual({ + type: "turn_canceled", + provider: "pi", + reason: "interrupted", + turnId, + }); + + fakeSession.finishTurn({ + role: "assistant", + provider: "openai-responses", + model: "gpt-5.6-terra", + responseId: "resp-aborted", + stopReason: "aborted", + errorMessage: "OpenAI Responses stream ended before a terminal response event", + content: [], + }); + + expect( + (events as unknown as { events: AgentStreamEvent[] }).events.map((e) => e.type), + ).not.toContain("turn_failed"); + }); + test("adds Pi assistant context to generic provider finish errors", async () => { const { pi, session, events } = await createSession(); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 66237340d..f08c68e28 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -849,6 +849,11 @@ function latestPiErrorMessage(messages: PiAgentMessage[]): string | null { return formatPiErrorMessage(latestAssistant); } +function isPiAbortedTerminalResponse(messages: PiAgentMessage[]): boolean { + const latestAssistant = messages.findLast((message) => message.role === "assistant"); + return latestAssistant?.stopReason?.toLowerCase() === "aborted"; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -1241,6 +1246,11 @@ export class PiRpcAgentSession implements AgentSession { private state: PiSessionState; private readonly currentModeId: string | null; private closed = false; + // Pi reports an aborted OpenAI Responses stream before the abort RPC resolves. + // Keep the turn active until that RPC acknowledges the user-requested cancellation. + private interruptingTurnId: string | null = null; + private lastInterruptedTurnId: string | null = null; + private interruptedTerminalError: { turnId: string; error: string } | null = null; constructor(options: PiRpcAgentSessionOptions) { this.runtimeSession = options.runtimeSession; @@ -1290,6 +1300,7 @@ export class PiRpcAgentSession implements AgentSession { const payload = convertPromptInput(prompt, { model: this.state.model }); const turnId = randomUUID(); this.activeTurnId = turnId; + this.lastInterruptedTurnId = null; this.activeClientMessageId = options?.clientMessageId ?? null; this.activeAssistantMessageId = null; this.activeTurnStarted = false; @@ -1434,7 +1445,33 @@ export class PiRpcAgentSession implements AgentSession { async interrupt(): Promise { const turnId = this.activeTurnId; - await this.runtimeSession.abort(); + if (turnId) { + this.interruptingTurnId = turnId; + this.lastInterruptedTurnId = turnId; + } + try { + await this.runtimeSession.abort(); + } catch (error) { + if (this.interruptingTurnId === turnId) { + this.interruptingTurnId = null; + } + if (this.interruptedTerminalError?.turnId === turnId) { + const terminalError = this.interruptedTerminalError; + this.interruptedTerminalError = null; + this.activeTurnId = null; + this.activeClientMessageId = null; + this.activeTurnStarted = false; + this.activeAssistantMessageId = null; + this.clearNoTurnBuffers(); + this.emit({ + type: "turn_failed", + provider: this.provider, + turnId, + error: terminalError.error, + }); + } + throw error; + } if (turnId && this.activeTurnId === turnId) { this.activeTurnId = null; this.activeClientMessageId = null; @@ -1448,6 +1485,12 @@ export class PiRpcAgentSession implements AgentSession { turnId, }); } + if (this.interruptingTurnId === turnId) { + this.interruptingTurnId = null; + } + if (this.interruptedTerminalError?.turnId === turnId) { + this.interruptedTerminalError = null; + } } async revertConversation(input: { messageId: string }): Promise { @@ -2246,6 +2289,20 @@ export class PiRpcAgentSession implements AgentSession { } private completeTurn(turnId: string | undefined, messages: PiAgentMessage[]): void { + if (turnId && this.interruptingTurnId === turnId && isPiAbortedTerminalResponse(messages)) { + this.interruptedTerminalError = { + turnId, + error: latestPiErrorMessage(messages) ?? "Pi turn failed", + }; + return; + } + if ( + isPiAbortedTerminalResponse(messages) && + (turnId === this.lastInterruptedTurnId || (!turnId && this.lastInterruptedTurnId !== null)) + ) { + this.lastInterruptedTurnId = null; + return; + } this.activeTurnId = null; this.activeClientMessageId = null; this.activeAssistantMessageId = null;