fix(pi): queue prompts until manual compaction finishes (#7848)

* fix(pi): wait for manual compaction before sending prompts

* test(pi): simplify compaction regressions and stabilize live smoke

packages/server/src/server/agent/providers/pi/agent.test.ts

@@ -3063,6 +3063,81 @@ describe("PiRpcAgentClient", () => {
     ]);
   });
 
+  test.each([
+    { outcome: "successful", error: null },
+    { outcome: "failed", error: new Error("summarizer failed") },
+  ])("waits for $outcome manual compaction before sending a new prompt", async ({ error }) => {
+    const { pi, session, events } = await createSession();
+    onTestFinished(() => session.close());
+    const fakeSession = pi.latestSession();
+    fakeSession.compactError = error;
+    const compactionFinished = Promise.withResolvers<void>();
+    fakeSession.compactCompletion = compactionFinished.promise;
+    const handler = session.tryHandleOutOfBand("/compact");
+    const compaction = handler!.run({ emit: () => {} });
+
+    const { turnId } = await session.startTurn("continue after compaction");
+    await flushTurnScheduling();
+    expect(fakeSession.compactRequests).toEqual([{}]);
+    expect(fakeSession.prompts).toEqual([]);
+
+    compactionFinished.resolve();
+    await compaction;
+    await flushTurnScheduling();
+    expect(fakeSession.prompts).toEqual([{ message: "continue after compaction", imageCount: 0 }]);
+
+    fakeSession.finishTurn();
+    await events.nextTurnCompletion();
+    expect(events.turnLifecycleEvents()).toContainEqual({ type: "turn_completed", turnId });
+    expect(events.eventTypes()).not.toContain("turn_failed");
+  });
+
+  test.each(["interrupt", "close"] as const)(
+    "does not send a prompt waiting on compaction after %s",
+    async (action) => {
+      const { pi, session } = await createSession();
+      onTestFinished(() => session.close());
+      const fakeSession = pi.latestSession();
+      const compactionFinished = Promise.withResolvers<void>();
+      fakeSession.compactCompletion = compactionFinished.promise;
+      const handler = session.tryHandleOutOfBand("/compact");
+      const compaction = handler!.run({ emit: () => {} });
+
+      await session.startTurn("do not send this prompt");
+      await session[action]();
+      compactionFinished.resolve();
+      await compaction;
+      await flushTurnScheduling();
+
+      expect(fakeSession.prompts).toEqual([]);
+    },
+  );
+
+  test("does not send a prompt when compaction finishes during interruption", async () => {
+    const { pi, session, events } = await createSession();
+    onTestFinished(() => session.close());
+    const fakeSession = pi.latestSession();
+    const compactionFinished = Promise.withResolvers<void>();
+    fakeSession.compactCompletion = compactionFinished.promise;
+    const clearQueueFinished = Promise.withResolvers<void>();
+    fakeSession.clearQueueCompletion = clearQueueFinished.promise;
+    const handler = session.tryHandleOutOfBand("/compact");
+    const compaction = handler!.run({ emit: () => {} });
+    const { turnId } = await session.startTurn("do not send this prompt");
+
+    const interruption = session.interrupt();
+    compactionFinished.resolve();
+    await compaction;
+    await flushTurnScheduling();
+    expect(fakeSession.controlRequests).toEqual(["clear_queue"]);
+    expect(fakeSession.prompts).toEqual([]);
+
+    clearQueueFinished.resolve();
+    await interruption;
+    expect(events.turnLifecycleEvents()).toEqual([{ type: "turn_canceled", turnId }]);
+    expect(fakeSession.prompts).toEqual([]);
+  });
+
   test("closes Pi compact loading marker when RPC rejects after compaction starts", async () => {
     const { pi, session } = await createSession();
     const fakeSession = pi.latestSession();

packages/server/src/server/agent/providers/pi/agent.ts

@@ -1189,6 +1189,7 @@ export class PiRpcAgentSession implements AgentSession {
   private readonly contextUserEntries: PiCapturedEntry[] = [];
   private readonly treeUserEntriesById = new Map<string, PiCapturedEntry>();
   private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
+  private compactionPromise: Promise<void> | null = null;
   private outOfBandCompactionEmit: ((event: AgentStreamEvent) => void) | null = null;
   private outOfBandCompactionStarted = false;
   private outOfBandCompactionCompleted = false;
@@ -1296,6 +1297,15 @@ export class PiRpcAgentSession implements AgentSession {
 
     void (async () => {
       try {
+        if (this.compactionPromise) {
+          // Pi RPC rejects prompts until the manual compaction request settles.
+          await this.compactionPromise.catch(() => undefined);
+          const turnCanceled =
+            this.closed || this.activeTurnId !== turnId || this.interruptingTurn?.turnId === turnId;
+          if (turnCanceled) {
+            return;
+          }
+        }
         const ack = await this.runtimeSession.prompt(payload.text, payload.images);
         this.activePromptRequestId = ack.requestId ?? null;
         const correlatedResult = ack.requestId
@@ -1804,7 +1814,8 @@ export class PiRpcAgentSession implements AgentSession {
     this.outOfBandCompactionStarted = false;
     this.outOfBandCompactionCompleted = false;
     try {
-      await this.runtimeSession.compact(customInstructions);
+      this.compactionPromise = this.runtimeSession.compact(customInstructions);
+      await this.compactionPromise;
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
       if (
@@ -1830,6 +1841,7 @@ export class PiRpcAgentSession implements AgentSession {
         },
       });
     } finally {
+      this.compactionPromise = null;
       if (this.outOfBandCompactionEmit === emit && !this.outOfBandCompactionStarted) {
         this.outOfBandCompactionEmit = null;
         this.outOfBandCompactionStarted = false;

packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts

@@ -108,6 +108,7 @@ export class FakePiSession implements PiRuntimeSession {
   steerError: Error | null = null;
   readonly controlRequests: string[] = [];
   clearQueueError: Error | null = null;
+  clearQueueCompletion: Promise<void> | null = null;
   readonly compactRequests: Array<{ customInstructions?: string }> = [];
   readonly setAutoCompactionRequests: boolean[] = [];
   readonly subagentSubscriptionRequests: FakePiSubagentSubscriptionLevel[] = [];
@@ -140,6 +141,7 @@ export class FakePiSession implements PiRuntimeSession {
   subagentSubscriptionError: Error | null = null;
   setSessionNameError: Error | null = null;
   compactError: Error | null = null;
+  compactCompletion: Promise<void> | null = null;
   emitCompactEnd = true;
   getStateError: Error | null = null;
   getSessionStatsError: Error | null = null;
@@ -233,6 +235,7 @@ export class FakePiSession implements PiRuntimeSession {
 
   async clearQueue(): Promise<void> {
     this.controlRequests.push("clear_queue");
+    await this.clearQueueCompletion;
     if (this.clearQueueError) {
       throw this.clearQueueError;
     }
@@ -241,6 +244,7 @@ export class FakePiSession implements PiRuntimeSession {
   async compact(customInstructions?: string): Promise<void> {
     this.compactRequests.push(customInstructions === undefined ? {} : { customInstructions });
     this.emit({ type: "compaction_start", reason: "manual" });
+    await this.compactCompletion;
     if (this.emitCompactEnd) {
       this.emit({ type: "compaction_end", reason: "manual" });
     }

packages/server/src/server/daemon-e2e/pi.real.e2e.test.ts

@@ -219,7 +219,7 @@ test(
 );
 
 test(
-  "real Pi daemon executes manual compact out-of-band instead of prompt text",
+  "real Pi daemon accepts a prompt after manual compact",
   async () => {
     const cwd = tmpCwd("pi-manual-compact-");
 
@@ -234,20 +234,24 @@ test(
           cwd,
           title: "pi-manual-compact",
           provider: "pi",
-          model: PI_COMPACTION_TEST_MODEL,
+          model: PI_REAL_TEST_MODEL,
         });
 
         await client.sendMessage(agent.id, "Reply exactly: compact-ready");
         const finish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS);
-        expect(finish.status).toBe("idle");
+        expect(finish.status, JSON.stringify(await fetchCanonicalTimeline(client, agent.id))).toBe(
+          "idle",
+        );
 
         await client.sendMessage(agent.id, "/compact summarize this e2e run");
-
         await waitForTimelineItem(
           client,
           agent.id,
           (item) => item.type === "compaction" && item.status === "completed",
         );
+        await client.sendMessage(agent.id, "Reply exactly: after-compaction");
+        const followUpFinish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS);
+        expect(followUpFinish.status).toBe("idle");
 
         const items = await fetchCanonicalTimeline(client, agent.id);
         expect(items).toEqual(
@@ -263,9 +267,12 @@ test(
             }),
           ]),
         );
-        expect(
-          items.some((item) => item.type === "user_message" && item.text.includes("/compact")),
-        ).toBe(false);
+        expect(items).toContainEqual(
+          expect.objectContaining({
+            type: "assistant_message",
+            text: "after-compaction",
+          }),
+        );
         expect(
           items
             .filter((item) => item.type === "assistant_message")