From 19247ad6c7906a2d9e7dc96b22d95fdb30b17f18 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Sun, 11 Oct 2026 12:42:31 +0900 Subject: [PATCH] fix(pi): warn when Stop clears queued steer messages --- .../server/agent/providers/pi/agent.test.ts | 153 ++++++++++++++++++ .../src/server/agent/providers/pi/agent.ts | 40 ++++- 2 files changed, 189 insertions(+), 4 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 039d3eaf22..f26568c0c3 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -2490,6 +2490,159 @@ describe("PiRpcAgentSession", () => { }); describe("PiRpcAgentSession steering", () => { + test("reports an unconsumed steer as not delivered when Stop clears the queue", async () => { + const { session, events } = await createSession(); + const { turnId } = await session.startTurn("work"); + await session.steerActiveTurn("Reply STEER_CHECK_7391", { + expectedTurnId: turnId, + clientMessageId: "client-steer-1", + }); + + await session.interrupt(); + + expect(events.timelineItems()).toContainEqual({ + type: "notification", + level: "warning", + message: "Not delivered (stopped): Reply STEER_CHECK_7391", + }); + }); + + test("does not report a consumed steer before its native entry ID arrives", async () => { + const { pi, session, events } = await createSession(); + const { turnId } = await session.startTurn("work"); + await session.steerActiveTurn("already consumed", { + expectedTurnId: turnId, + clientMessageId: "client-steer-1", + }); + pi.latestSession().emit({ + type: "message_end", + message: { role: "user", content: "already consumed" }, + }); + + await session.interrupt(); + + expect(events.timelineItems()).toEqual([]); + }); + + test("tracks native consumption before the steer RPC is acknowledged", async () => { + const { pi, session, events } = await createSession(); + const { turnId } = await session.startTurn("work"); + const native = pi.latestSession(); + native.steer = async (message) => { + native.finishSubmittedUserMessage({ id: "native-entry-1", parentId: null, text: message }); + }; + + await expect( + session.steerActiveTurn("already consumed", { + expectedTurnId: turnId, + clientMessageId: "client-steer-1", + }), + ).resolves.toEqual({ status: "accepted" }); + await session.interrupt(); + + expect(events.timelineItems()).toEqual([ + { + type: "user_message", + text: "already consumed", + messageId: "native-entry-1", + clientMessageId: "client-steer-1", + }, + ]); + }); + + test("keeps identical steers distinct from native message IDs when one is consumed", async () => { + const { pi, session, events } = await createSession(); + const { turnId } = await session.startTurn("work"); + await session.steerActiveTurn("same text", { + expectedTurnId: turnId, + clientMessageId: "client-steer-1", + }); + await session.steerActiveTurn("same text", { + expectedTurnId: turnId, + clientMessageId: "client-steer-2", + }); + pi.latestSession().finishSubmittedUserMessage({ + id: "native-entry-1", + parentId: null, + text: "same text", + }); + + await session.interrupt(); + + expect(events.timelineItems()).toEqual([ + { + type: "user_message", + text: "same text", + messageId: "native-entry-1", + clientMessageId: "client-steer-1", + }, + { + type: "notification", + level: "warning", + message: "Not delivered (stopped): same text", + }, + ]); + }); + + test("retains steer tracking when clear_queue fails so Stop can be retried", async () => { + const { pi, session, events } = await createSession(); + const { turnId } = await session.startTurn("work"); + await session.steerActiveTurn("queued", { + expectedTurnId: turnId, + clientMessageId: "client-steer-1", + }); + const native = pi.latestSession(); + native.clearQueueError = new Error("Queue clear failed"); + + await expect(session.interrupt()).rejects.toThrow("Queue clear failed"); + expect(native.controlRequests).toEqual(["clear_queue"]); + expect(events.timelineItems()).toEqual([]); + + native.clearQueueError = null; + await session.interrupt(); + expect(events.timelineItems()).toEqual([ + { + type: "notification", + level: "warning", + message: "Not delivered (stopped): queued", + }, + ]); + }); + + test("reports an acknowledged queue removal even when abort subsequently fails", async () => { + const { pi, session, events } = await createSession(); + const { turnId } = await session.startTurn("work"); + await session.steerActiveTurn("queued", { expectedTurnId: turnId }); + const native = pi.latestSession(); + native.abort = async () => { + throw new Error("Abort failed"); + }; + + await expect(session.interrupt()).rejects.toThrow("Abort failed"); + native.abort = async () => {}; + await session.interrupt(); + + expect(events.timelineItems()).toEqual([ + { + type: "notification", + level: "warning", + message: "Not delivered (stopped): queued", + }, + ]); + }); + + test("does not claim queue removal when an older Pi lacks clear_queue", async () => { + const { pi, session, events } = await createSession(); + const { turnId } = await session.startTurn("work"); + await session.steerActiveTurn("queued", { expectedTurnId: turnId }); + pi.latestSession().clearQueueError = new Error("Unknown command: clear_queue"); + + await session.interrupt(); + + expect(pi.latestSession().controlRequests).toEqual(["clear_queue", "abort"]); + expect(events.timelineItems()).toEqual([]); + }); + test("steers the active Pi turn and correlates the echoed user message", 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 198d596c9b..b02ccf783c 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -233,6 +233,7 @@ interface StartTurnResult { interface PiPendingSteerSubmission { text: string; clientMessageId: string | null; + consumed: boolean; } // COMPAT(piSteerFallback): added in v0.5.0, remove after 2027-03-01 once the pi floor @@ -1403,9 +1404,18 @@ export class PiRpcAgentSession implements AgentSession { return { status: "unavailable" }; } + const submission: PiPendingSteerSubmission = { + text: payload.text, + clientMessageId: options.clientMessageId ?? null, + consumed: false, + }; + // Pi can consume the message before acknowledging the steer RPC. + this.pendingSteerSubmissions.push(submission); try { await this.runtimeSession.steer(payload.text, payload.images); } catch (error) { + const index = this.pendingSteerSubmissions.indexOf(submission); + if (index >= 0) this.pendingSteerSubmissions.splice(index, 1); if (isPiDefinitiveSteerRejection(error)) { return { status: "unavailable" }; } @@ -1416,10 +1426,6 @@ export class PiRpcAgentSession implements AgentSession { if (this.closed || this.activeTurnId !== options.expectedTurnId) { return { status: "unavailable" }; } - this.pendingSteerSubmissions.push({ - text: payload.text, - clientMessageId: options.clientMessageId ?? null, - }); if (options.clearPendingPermissions) { await this.clearPendingPermissionsForSteer(); } @@ -1542,6 +1548,23 @@ export class PiRpcAgentSession implements AgentSession { try { try { await this.runtimeSession.clearQueue(); + // Only an acknowledged clear_queue proves queued messages were dropped. + for (const submission of this.pendingSteerSubmissions.splice(0)) { + if (submission.consumed) { + this.pendingSteerSubmissions.push(submission); + continue; + } + this.emit({ + type: "timeline", + provider: this.provider, + turnId, + item: { + type: "notification", + level: "warning", + message: `Not delivered (stopped): ${submission.text}`, + }, + }); + } } catch (error) { // COMPAT(piClearQueueFallback): added in v0.5.0, remove after 2027-03-01 once // the pi floor supports clear_queue (added in pi 0.84.4). @@ -2481,6 +2504,15 @@ export class PiRpcAgentSession implements AgentSession { event: Extract, turnId: string | undefined, ): void { + if (event.message.role === "user") { + // Entry IDs arrive through the extension later; consumption is already known. + const text = getUserMessageText(event.message.content); + const submission = this.pendingSteerSubmissions.find( + (pending) => !pending.consumed && pending.text === text, + ); + if (submission) submission.consumed = true; + return; + } if (event.message.role === "assistant") { this.activeAssistantMessageId = null; return;