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

@@ -2183,6 +2183,48 @@ describe("PiRpcAgentSession", () => {
     ]);
   });
 
+  test.each([false, true])(
+    "closing Pi releases an active turn without publishing a terminal outcome (autonomous=%s)",
+    async (autonomous) => {
+      const { pi, session, events } = await createSession();
+      if (autonomous) {
+        pi.latestSession().emit({ type: "agent_start" });
+        pi.latestSession().emit({ type: "turn_start" });
+      } else {
+        await session.startTurn("keep working");
+      }
+      const before = events.turnLifecycleEvents();
+
+      await session.close();
+
+      expect(pi.latestSession().closeCalls).toBe(1);
+      expect(events.turnLifecycleEvents()).toEqual(before);
+    },
+  );
+
+  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);
   }
 

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

@@ -427,6 +427,13 @@ 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",
+    });
   }
 
   emit(event: PiRuntimeEvent): void {