fix(pi): converge worker failures and thinking state

tests/unit/pi-worker-pool.test.ts

@@ -13,6 +13,10 @@ import { PiProcessError as PiProcessFailure } from '../../electron/coding-runtim
 import type { PiRpcCommand, PiRpcEvent } from '../../electron/coding-runtime/pi/rpc-client';
 import type { PiRuntimeTelemetryEvent } from '../../electron/coding-runtime/pi/telemetry';
 import { PiSubagentScheduler } from '../../electron/coding-runtime/pi/subagent';
+import type {
+  PiWorkerProofFailure,
+  PiWorkerStopReason,
+} from '../../electron/coding-runtime/pi/worker-process';
 
 const MODEL = {
   model: {
@@ -43,6 +47,7 @@ class FakeWorker implements PiConversationWorker {
   readonly generation = 1;
   readonly requests: PiRpcCommand[] = [];
   stopped = false;
+  readonly stopReasons: PiWorkerStopReason[] = [];
   private readonly eventListeners = new Set<(event: PiRpcEvent) => void>();
   private readonly invalidationListeners = new Set<(error: PiProcessError) => void>();
 
@@ -75,13 +80,50 @@ class FakeWorker implements PiConversationWorker {
     for (const listener of this.invalidationListeners) listener(error);
   }
 
-  async stop() {
+  async injectFailureForProof(failure: PiWorkerProofFailure): Promise<void> {
+    this.invalidate(new PiProcessFailure(
+      failure === 'protocol_invalidation' ? 'PI_RPC_PROTOCOL_ERROR' : 'PI_RPC_EXITED',
+      'injected proof failure',
+    ));
+  }
+
+  async stop(reason: PiWorkerStopReason) {
     this.stopped = true;
+    this.stopReasons.push(reason);
     return { mode: 'stdin-close' as const, code: 0, signal: null };
   }
 }
 
 describe('Pi worker pool', () => {
+  it('injects a proof failure only into the requested current generation', async () => {
+    const workers = new Map<string, FakeWorker>();
+    const pool = new PiWorkerPool({
+      maxIdle: 4,
+      openWorker: async ({ conversation: input }) => {
+        const worker = new FakeWorker(`worker-${input.conversationId}`);
+        workers.set(input.conversationId, worker);
+        return {
+          worker,
+          session: {
+            piSessionId: `session-${input.conversationId}`,
+            sessionKey: `key-${input.conversationId}`,
+          },
+        };
+      },
+    });
+    await Promise.all([
+      pool.prepare(conversation('conversation-target')),
+      pool.prepare(conversation('conversation-other')),
+    ]);
+
+    await expect(pool.injectFailureForProof('conversation-target', 'unexpected_exit'))
+      .resolves.toEqual({ generation: 1 });
+
+    expect(pool.getState('conversation-target')).toMatchObject({ state: 'crashed', generation: 1 });
+    expect(pool.getState('conversation-other')).toMatchObject({ state: 'ready', generation: 1 });
+    expect(workers.get('conversation-other')?.stopped).toBe(false);
+  });
+
   it('single-flights prepare per Conversation and never shares its worker with another Conversation', async () => {
     const gate = deferred();
     const opened: string[] = [];
@@ -418,9 +460,11 @@ describe('Pi worker pool', () => {
 
   it('cleans only the crashed generation and releases its permit for the next Conversation', async () => {
     const workers = new Map<string, FakeWorker>();
+    const processBudget = new PiProcessBudget(8);
     const pool = new PiWorkerPool({
       maxRunning: 2,
       maxIdle: 3,
+      processBudget,
       openWorker: async ({ conversation: input }) => {
         const worker = new FakeWorker(`worker-${input.conversationId}`);
         workers.set(input.conversationId, worker);
@@ -461,6 +505,7 @@ describe('Pi worker pool', () => {
     expect(pool.getState('conversation-a')).toMatchObject({ state: 'crashed', generation: 1 });
     expect(pool.getState('conversation-b')).toMatchObject({ state: 'running', generation: 1 });
     await expect.poll(() => workers.get('conversation-c')!.requests.length).toBe(1);
+    await expect.poll(() => processBudget.activeCount).toBe(2);
     expect(cancelled).not.toContain('other');
   });
 
@@ -468,6 +513,7 @@ describe('Pi worker pool', () => {
     const workers = new Map<string, FakeWorker[]>();
     const revisions: Array<{ conversationId: string; provider: number; resources: number }> = [];
     const telemetry: PiRuntimeTelemetryEvent[] = [];
+    const replacementReasons: string[] = [];
     const pool = new PiWorkerPool({
       maxIdle: 4,
       onTelemetry: (event) => telemetry.push(event),
@@ -484,6 +530,9 @@ describe('Pi worker pool', () => {
         };
       },
     });
+    pool.subscribe((event) => {
+      if (event.type === 'worker.replaced') replacementReasons.push(event.reason);
+    });
     await Promise.all([
       pool.prepare(conversation('conversation-running')),
       pool.prepare(conversation('conversation-idle')),
@@ -520,6 +569,46 @@ describe('Pi worker pool', () => {
       { conversationId: 'conversation-idle', provider: 2, resources: 1 },
       { conversationId: 'conversation-running', provider: 2, resources: 1 },
     ]));
+    expect(replacementReasons).toEqual([
+      'stale_resource_rebuild',
+      'stale_resource_rebuild',
+    ]);
+    expect(workers.get('conversation-idle')![0]!.stopReasons).toEqual([
+      'stale_resource_rebuild',
+    ]);
+    expect(workers.get('conversation-running')![0]!.stopReasons).toEqual([
+      'stale_resource_rebuild',
+    ]);
+  });
+
+  it('records an explicit recover replacement reason without changing the session binding', async () => {
+    const workers: FakeWorker[] = [];
+    const replacements: Array<{ reason: string; generation: number }> = [];
+    const pool = new PiWorkerPool({
+      openWorker: async ({ conversation: input, generation, existingSession }) => {
+        const worker = new FakeWorker(`worker-${generation}`);
+        workers.push(worker);
+        return {
+          worker,
+          session: existingSession ?? {
+            piSessionId: `session-${input.conversationId}`,
+            sessionKey: `key-${input.conversationId}`,
+          },
+        };
+      },
+    });
+    pool.subscribe((event) => {
+      if (event.type === 'worker.replaced') {
+        replacements.push({ reason: event.reason, generation: event.generation });
+      }
+    });
+
+    const before = await pool.prepare(conversation('conversation-recover'));
+    const recovered = await pool.recover('conversation-recover');
+
+    expect(recovered.session).toEqual(before.session);
+    expect(replacements).toEqual([{ reason: 'recover', generation: 2 }]);
+    expect(workers[0]!.stopReasons).toEqual(['recover']);
   });
 
   it('re-applies the idle LRU after a running stale worker rebuilds on settle', async () => {