From c424f82922fcd36aa9cc9e473644bca04417b420 Mon Sep 17 00:00:00 2001 From: Michael Jakl Date: Mon, 7 Sep 2026 08:57:01 +0200 Subject: [PATCH] fix(pi): settle autonomous Pi turns triggered by extensions (#3849) * Settle autonomous Pi turns triggered by extensions Treat Pi's agent lifecycle as authoritative for extension-triggered turns that do not have a Paseo provider turn ID. Keep locally handled extension commands immediate, but defer autonomous completion until agent_settled. * Fail autonomous Pi turns on process exit * Fix cancellation of autonomous Pi turns * Synchronize Pi cancellation regressions with queue clearing * Release Pi interruption state at its terminal boundary --------- Co-authored-by: Mohamed Boudra --- .../server/agent/providers/pi/agent.test.ts | 190 ++++++++++++++++++ .../src/server/agent/providers/pi/agent.ts | 80 ++++---- 2 files changed, 227 insertions(+), 43 deletions(-) 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 631ce28b9f..234fe59cec 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -895,6 +895,41 @@ describe("PiRpcAgentSession", () => { ]); }); + test("settles an autonomous turn triggered by a Pi extension custom message", async () => { + const { pi, events } = await createSession(); + const fakeSession = pi.latestSession(); + + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + fakeSession.emit({ + type: "message_end", + message: { + role: "custom", + content: [{ type: "text", text: "Background process completed" }], + }, + }); + fakeSession.finishAgentRun({ + message: { + role: "assistant", + stopReason: "stop", + content: [{ type: "text", text: "Continuation finished" }], + }, + willRetry: false, + }); + + expect(events.timelineItems()).toEqual([ + { type: "assistant_message", text: "Background process completed" }, + ]); + expect(events.turnLifecycleEvents()).toEqual([{ type: "turn_started", turnId: undefined }]); + + fakeSession.settleTurn(); + + expect(events.turnLifecycleEvents()).toEqual([ + { type: "turn_started", turnId: undefined }, + { type: "turn_completed", turnId: undefined }, + ]); + }); + test("canceling a silent Pi extension command leaves the session usable", async () => { const { pi, session, events } = await createSession(); const fakeSession = pi.latestSession(); @@ -953,6 +988,143 @@ describe("PiRpcAgentSession", () => { }); }); + test.each(["aborted", "error"])( + "canceling autonomous work with Pi stopReason=%s waits for abort and allows a follow-up", + async (stopReason) => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + const abortFinished = Promise.withResolvers(); + const turnFinished = Promise.withResolvers(); + fakeSession.abort = async () => { + fakeSession.finishTurn({ + role: "assistant", + stopReason, + errorMessage: "This operation was aborted", + content: [], + }); + turnFinished.resolve(); + await abortFinished.promise; + }; + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + + const stopping = session.interrupt(); + await turnFinished.promise; + expect(events.turnLifecycleEvents()).toEqual([{ type: "turn_started", turnId: undefined }]); + abortFinished.resolve(); + await stopping; + expect(events.turnLifecycleEvents()).toEqual([ + { type: "turn_started", turnId: undefined }, + { type: "turn_canceled", turnId: undefined }, + ]); + + const { turnId } = await session.startTurn("follow-up"); + fakeSession.finishTurn({ role: "assistant", stopReason: "stop", content: [] }); + expect(events.turnLifecycleEvents().at(-1)).toEqual({ type: "turn_completed", turnId }); + }, + ); + + test("a natural completion during Stop does not cancel the next autonomous run", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + const abortFinished = Promise.withResolvers(); + const turnFinished = Promise.withResolvers(); + fakeSession.abort = async () => { + fakeSession.finishTurn({ role: "assistant", stopReason: "stop", content: [] }); + turnFinished.resolve(); + await abortFinished.promise; + }; + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + const stopping = session.interrupt(); + await turnFinished.promise; + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + abortFinished.resolve(); + await stopping; + fakeSession.finishTurn({ role: "assistant", stopReason: "stop", content: [] }); + + expect(events.turnLifecycleEvents()).toEqual([ + { type: "turn_started", turnId: undefined }, + { type: "turn_completed", turnId: undefined }, + { type: "turn_started", turnId: undefined }, + { type: "turn_completed", turnId: undefined }, + ]); + }); + + test("a completed foreground Stop does not suppress a later autonomous abort", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.abort = async () => { + fakeSession.finishTurn({ role: "assistant", stopReason: "stop", content: [] }); + }; + const { turnId } = await session.startTurn("finish while stopping"); + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + await session.interrupt(); + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + fakeSession.finishTurn({ + role: "assistant", + stopReason: "aborted", + errorMessage: "Autonomous run aborted", + content: [], + }); + + expect(events.turnLifecycleEvents()).toEqual([ + { type: "turn_started", turnId }, + { type: "turn_completed", turnId }, + { type: "turn_started", turnId: undefined }, + { type: "turn_failed", turnId: undefined }, + ]); + }); + + test("Pi process exit during Stop emits one autonomous failure", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.abort = async () => { + fakeSession.finishTurn({ + role: "assistant", + stopReason: "error", + errorMessage: "This operation was aborted", + content: [], + }); + fakeSession.emit({ type: "process_exit", error: "Pi process exited" }); + throw new Error("Pi process exited"); + }; + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + + await expect(session.interrupt()).rejects.toThrow("Pi process exited"); + expect(events.turnLifecycleEvents()).toEqual([ + { type: "turn_started", turnId: undefined }, + { type: "turn_failed", turnId: undefined }, + ]); + await expect(events.nextTurnFailure()).resolves.toMatchObject({ error: "Pi process exited" }); + }); + + test("preserves the autonomous terminal error when abort rejects", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.abort = async () => { + fakeSession.finishTurn({ + role: "assistant", + stopReason: "error", + errorMessage: "Provider disconnected", + content: [], + }); + throw new Error("Abort failed"); + }; + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + + await expect(session.interrupt()).rejects.toThrow("Abort failed"); + await expect(events.nextTurnFailure()).resolves.toMatchObject({ + turnId: undefined, + error: expect.stringContaining("Provider disconnected"), + }); + }); + test("suppresses late aborted terminal response arriving after interrupt resolves", async () => { const { pi, session, events } = await createSession(); const fakeSession = pi.latestSession(); @@ -1431,6 +1603,24 @@ describe("PiRpcAgentSession", () => { }); }); + test("fails an autonomous turn when the Pi process exits before settlement", async () => { + const { pi, events } = await createSession(); + const fakeSession = pi.latestSession(); + + fakeSession.emit({ type: "agent_start" }); + fakeSession.emit({ type: "turn_start" }); + fakeSession.emit({ type: "process_exit", error: "Pi exited" }); + + await expect(events.nextTurnFailure()).resolves.toMatchObject({ + error: "Pi exited", + turnId: undefined, + }); + expect(events.turnLifecycleEvents()).toEqual([ + { type: "turn_started", turnId: undefined }, + { type: "turn_failed", turnId: undefined }, + ]); + }); + test("completes locally handled slash commands when agentInvoked is false", async () => { const { pi, session, events } = await createSession(); const fakeSession = pi.latestSession(); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index b262af0d76..aaf517422d 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -1258,11 +1258,9 @@ export class PiRpcAgentSession implements AgentSession { private readonly logger: Logger; private readonly usagePoller: PiUsagePoller; 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; + // Pi publishes the terminal before acknowledging abort. Autonomous runs have no + // turn ID; retain their errors too until the cancellation request settles. + private interruptingTurn: { turnId: string | undefined; error: string | null } | null = null; constructor(options: PiRpcAgentSessionOptions) { this.runtimeSession = options.runtimeSession; @@ -1329,7 +1327,6 @@ export class PiRpcAgentSession implements AgentSession { const turnId = randomUUID(); this.activeTurnId = turnId; this.usagePoller.startTurn(); - this.lastInterruptedTurnId = null; this.activeClientMessageId = options?.clientMessageId ?? null; this.activeAssistantMessageId = null; this.activeTurnStarted = false; @@ -1537,11 +1534,10 @@ export class PiRpcAgentSession implements AgentSession { } async interrupt(): Promise { - const turnId = this.activeTurnId; - if (turnId) { - this.interruptingTurnId = turnId; - this.lastInterruptedTurnId = turnId; - } + const turnId = this.activeTurnId ?? undefined; + const interruption: typeof this.interruptingTurn = + this.activeTurnId || this.activeTurnStarted ? { turnId, error: null } : null; + this.interruptingTurn = interruption; try { try { await this.runtimeSession.clearQueue(); @@ -1554,12 +1550,11 @@ export class PiRpcAgentSession implements AgentSession { } await this.runtimeSession.abort(); } catch (error) { - if (this.interruptingTurnId === turnId) { - this.interruptingTurnId = null; + const terminalError = this.interruptingTurn === interruption ? interruption?.error : null; + if (this.interruptingTurn === interruption) { + this.interruptingTurn = null; } - if (this.interruptedTerminalError?.turnId === turnId) { - const terminalError = this.interruptedTerminalError; - this.interruptedTerminalError = null; + if (terminalError) { this.usagePoller.stopTurn(); this.activeTurnId = null; this.activeClientMessageId = null; @@ -1573,12 +1568,17 @@ export class PiRpcAgentSession implements AgentSession { type: "turn_failed", provider: this.provider, turnId, - error: terminalError.error, + error: terminalError, }); } throw error; } - if (turnId && this.activeTurnId === turnId) { + if ( + interruption && + this.interruptingTurn === interruption && + (this.activeTurnId || this.activeTurnStarted) && + (this.activeTurnId ?? undefined) === turnId + ) { this.usagePoller.stopTurn(); this.activeTurnId = null; this.activeClientMessageId = null; @@ -1595,11 +1595,8 @@ export class PiRpcAgentSession implements AgentSession { turnId, }); } - if (this.interruptingTurnId === turnId) { - this.interruptingTurnId = null; - } - if (this.interruptedTerminalError?.turnId === turnId) { - this.interruptedTerminalError = null; + if (this.interruptingTurn === interruption) { + this.interruptingTurn = null; } } @@ -2167,10 +2164,11 @@ export class PiRpcAgentSession implements AgentSession { private handleProcessExit(error: string): void { this.rejectAllExtensionResults(new Error(error)); - if (!this.activeTurnId) { + this.interruptingTurn = null; + if (!this.activeTurnId && !this.activeTurnStarted) { return; } - const turnId = this.activeTurnId; + const turnId = this.activeTurnId ?? undefined; this.usagePoller.stopTurn(); this.activeTurnId = null; this.activeClientMessageId = null; @@ -2290,6 +2288,9 @@ export class PiRpcAgentSession implements AgentSession { event: Extract; turnId: string | undefined; }): void { + if (!this.activeTurnId && !this.activeTurnStarted) { + return; + } if (event.type === "agent_end") { // COMPAT(piAgentSettled): added in v0.5.0, remove after 2027-02-21 once the Pi // floor emits agent_settled and willRetry. @@ -2297,14 +2298,10 @@ export class PiRpcAgentSession implements AgentSession { this.completeTurn(turnId, event.messages ?? []); return; } - if (this.activeTurnId) { - this.pendingSettledMessages = event.messages ?? []; - } + this.pendingSettledMessages = event.messages ?? []; return; } - if (this.activeTurnId) { - this.completeTurn(turnId, this.pendingSettledMessages ?? []); - } + this.completeTurn(turnId, this.pendingSettledMessages ?? []); } private handleToolExecutionEnd( @@ -2415,7 +2412,9 @@ export class PiRpcAgentSession implements AgentSession { item: { type: "assistant_message", text }, }); } - this.completeTurn(turnId, []); + if (!this.activeTurnStarted) { + this.completeTurn(turnId, []); + } return; } } @@ -2458,20 +2457,16 @@ 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; - } + const errorMessage = latestPiErrorMessage(messages); if ( - isPiAbortedTerminalResponse(messages) && - (turnId === this.lastInterruptedTurnId || (!turnId && this.lastInterruptedTurnId !== null)) + this.interruptingTurn && + this.interruptingTurn.turnId === turnId && + (errorMessage || isPiAbortedTerminalResponse(messages)) ) { - this.lastInterruptedTurnId = null; + this.interruptingTurn.error = errorMessage ?? "Pi turn failed"; return; } + this.interruptingTurn = null; this.activeTurnId = null; this.activeClientMessageId = null; this.activeAssistantMessageId = null; @@ -2480,7 +2475,6 @@ export class PiRpcAgentSession implements AgentSession { this.pendingSettledMessages = null; this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); - const errorMessage = latestPiErrorMessage(messages); if (typeof errorMessage === "string" && errorMessage.length > 0) { this.usagePoller.stopTurn(); this.emit({