fix(server): suppress Pi interruption stream error (#2311)

* Handle Pi aborted responses after interruption

* fix(server): suppress late Pi interruption terminal response

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();
 

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<string, unknown> {
   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<void> {
     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<void> {
@@ -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;