fix(pi): preserve recovery before prompt acknowledgement

Own intentional-close lifecycle reporting at Pi event publication so pending RPC rejections and process exits cannot consume unfinished work. Preserve cleanup output and errors from open sessions.

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

@@ -2202,6 +2202,29 @@ describe("PiRpcAgentSession", () => {
     },
   );
 
+  test("closing Pi before prompt acknowledgement preserves the unfinished turn", async () => {
+    const { pi, session, events } = await createSession();
+    pi.latestSession().holdNextPrompt();
+    await session.startTurn("keep working");
+    expect(pi.latestSession().prompts).toHaveLength(1);
+    const before = events.turnLifecycleEvents();
+
+    await session.close();
+
+    expect(events.turnLifecycleEvents()).toEqual(before);
+  });
+
+  test("a rejected prompt still fails an open Pi session", async () => {
+    const { pi, session, events } = await createSession();
+    pi.latestSession().holdNextPrompt();
+    await session.startTurn("keep working");
+
+    await pi.latestSession().failHeldPrompt(new Error("Prompt rejected"));
+
+    await expect(events.nextTurnFailure()).resolves.toMatchObject({ error: "Prompt rejected" });
+    await session.close();
+  });
+
   test("fails the active turn when the Pi process exits mid-turn", async () => {
     const { pi, session, events } = await createSession();
 

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

@@ -1729,6 +1729,17 @@ export class PiRpcAgentSession implements AgentSession {
   }
 
   private publish(event: AgentStreamEvent): void {
+    // Runtime release preserves unfinished work for restart recovery. Cleanup
+    // may still emit timeline updates, but must not settle or restart its turn.
+    if (
+      this.closed &&
+      (event.type === "turn_started" ||
+        event.type === "turn_completed" ||
+        event.type === "turn_canceled" ||
+        event.type === "turn_failed")
+    ) {
+      return;
+    }
     for (const subscriber of this.subscribers) subscriber(event);
   }
 
@@ -2204,16 +2215,12 @@ export class PiRpcAgentSession implements AgentSession {
     this.pendingSettledMessages = null;
     this.pendingSteerSubmissions.length = 0;
     this.clearNoTurnBuffers();
-    // Releasing the runtime is not a failed turn: restart recovery still owns
-    // unfinished work after the session has been intentionally closed.
-    if (!this.closed) {
-      this.emit({
-        type: "turn_failed",
-        provider: this.provider,
-        turnId,
-        error,
-      });
-    }
+    this.emit({
+      type: "turn_failed",
+      provider: this.provider,
+      turnId,
+      error,
+    });
   }
 
   private handleSessionEvent(event: PiAgentSessionEvent): void {

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

@@ -427,6 +427,9 @@ export class FakePiSession implements PiRuntimeSession {
 
   async close(): Promise<void> {
     this.closeCalls += 1;
+    if (this.activeHeldPrompt) {
+      await this.failHeldPrompt(new Error("Pi RPC session is closed"));
+    }
     this.emit({
       type: "process_exit",
       error: "Pi RPC process exited with code 0 and signal null",