fix(pi): warn when Stop clears queued steer messages

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

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<PiAgentSessionEvent, { type: "message_end" }>,
     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;