puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-25 18:52:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(pi): settle autonomous Pi turns triggered by extensions (#3849)

Browse Source 
* 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 <boudra.moha@gmail.com>
This commit is contained in:
Michael Jakl
2026-09-07 08:57:01 +02:00
committed by GitHub
co-authored by Mohamed Boudra
parent b403dea32b
commit c424f82922
2 changed files with 227 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/pi/agent.test.ts
+190
View File
@@ -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<void>();
const turnFinished = Promise.withResolvers<void>();
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<void>();
const turnFinished = Promise.withResolvers<void>();
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();
packages/server/src/server/agent/providers/pi/agent.ts
+37 -43
View File
@@ -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<void> {
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<PiAgentSessionEvent, { type: "agent_end" | "agent_settled" }>;
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({