fix(pi): retain ownership for uncertain mutations

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

@@ -10,7 +10,12 @@ import {
 } from '../../electron/coding-runtime/pi/worker-pool';
 import type { PiProcessError } from '../../electron/coding-runtime/pi/process-errors';
 import { PiProcessError as PiProcessFailure } from '../../electron/coding-runtime/pi/process-errors';
-import type { PiRpcCommand, PiRpcEvent } from '../../electron/coding-runtime/pi/rpc-client';
+import type {
+  PiRpcCommand,
+  PiRpcEvent,
+  PiRpcRequestOptions,
+  PiRpcResponse,
+} 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 {
@@ -50,14 +55,61 @@ class FakeWorker implements PiConversationWorker {
   readonly stopReasons: PiWorkerStopReason[] = [];
   private readonly eventListeners = new Set<(event: PiRpcEvent) => void>();
   private readonly invalidationListeners = new Set<(error: PiProcessError) => void>();
+  private timeoutType: string | null = null;
+  private pendingType: string | null = null;
+  private lateResult: ((result: {
+    response?: PiRpcResponse;
+    error?: PiProcessError;
+  }) => void) | undefined;
 
   constructor(readonly id: string) {}
 
-  async request() {
-    this.requests.push(arguments[0] as PiRpcCommand);
+  async request<T = unknown>(command: PiRpcCommand, options?: PiRpcRequestOptions) {
+    this.requests.push(command);
+    if (this.pendingType === command.type) {
+      this.pendingType = null;
+      return await new Promise<PiRpcResponse<T>>((_resolve, reject) => {
+        options?.signal?.addEventListener('abort', () => reject(new PiProcessFailure(
+          'PI_RPC_ABORTED',
+          `fake ${command.type} was aborted after authoritative settlement`,
+        )), { once: true });
+      });
+    }
+    if (this.timeoutType === command.type) {
+      this.timeoutType = null;
+      this.lateResult = (options as PiRpcRequestOptions & {
+        onLateResult?(result: {
+          response?: PiRpcResponse;
+          error?: PiProcessError;
+        }): void;
+      } | undefined)?.onLateResult;
+      throw new PiProcessFailure('PI_RPC_TIMEOUT', `fake ${command.type} confirmation timeout`);
+    }
     return { type: 'response' as const, id: 'fake', success: true };
   }
 
+  timeoutNext(type: string): void {
+    this.timeoutType = type;
+  }
+
+  pendNext(type: string): void {
+    this.pendingType = type;
+  }
+
+  completeLateSuccess(): void {
+    this.lateResult?.({
+      response: { type: 'response', id: 'fake-late', success: true },
+    });
+    this.lateResult = undefined;
+  }
+
+  completeLateFailure(): void {
+    this.lateResult?.({
+      error: new PiProcessFailure('PI_RPC_RESPONSE_ERROR', 'fake late rejection'),
+    });
+    this.lateResult = undefined;
+  }
+
   async send(command: PiRpcCommand): Promise<void> {
     this.requests.push(command);
   }
@@ -95,6 +147,108 @@ class FakeWorker implements PiConversationWorker {
 }
 
 describe('Pi worker pool', () => {
+  it('retains top-level ownership after a mutation confirmation timeout', async () => {
+    const workers = new Map<string, FakeWorker>();
+    const pool = new PiWorkerPool({
+      maxRunning: 2,
+      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-sibling')),
+    ]);
+    workers.get('conversation-target')!.timeoutNext('prompt');
+
+    const target = pool.startTopLevel({
+      conversationId: 'conversation-target',
+      runId: 'run-target',
+      command: { type: 'prompt', message: 'slow preflight' },
+    });
+    await expect(target.accepted).rejects.toMatchObject({ code: 'PI_RPC_TIMEOUT' });
+
+    expect(pool.getActiveRun('conversation-target')).toMatchObject({ runId: 'run-target' });
+    expect(pool.getResilienceProofDiagnostics().runs).toEqual({ active: 1, waiting: 0 });
+    expect(() => pool.startTopLevel({
+      conversationId: 'conversation-target',
+      runId: 'run-overlap',
+      command: { type: 'compact' },
+    })).toThrow('Conversation already has a top-level run');
+
+    const sibling = pool.startTopLevel({
+      conversationId: 'conversation-sibling',
+      runId: 'run-sibling',
+      command: { type: 'prompt', message: 'independent sibling' },
+    });
+    await expect(sibling.accepted).resolves.toMatchObject({ success: true });
+    workers.get('conversation-sibling')!.emit({ type: 'agent_settled' });
+    workers.get('conversation-target')!.completeLateSuccess();
+    expect(pool.getActiveRun('conversation-target')).toMatchObject({ runId: 'run-target' });
+
+    workers.get('conversation-target')!.emit({ type: 'agent_settled' });
+    expect(pool.getActiveRun('conversation-target')).toBeNull();
+    expect(pool.getResilienceProofDiagnostics().runs).toEqual({ active: 0, waiting: 0 });
+  });
+
+  it('releases uncertain top-level ownership after a late explicit failure', async () => {
+    const worker = new FakeWorker('worker-target');
+    const pool = new PiWorkerPool({
+      maxIdle: 2,
+      openWorker: async () => ({
+        worker,
+        session: { piSessionId: 'session-target', sessionKey: 'key-target' },
+      }),
+    });
+    await pool.prepare(conversation('conversation-target'));
+    worker.timeoutNext('compact');
+    const ticket = pool.startTopLevel({
+      conversationId: 'conversation-target',
+      runId: 'run-compact',
+      command: { type: 'compact' },
+    });
+    await expect(ticket.accepted).rejects.toMatchObject({ code: 'PI_RPC_TIMEOUT' });
+    expect(pool.getActiveRun('conversation-target')).toMatchObject({ runId: 'run-compact' });
+
+    worker.completeLateFailure();
+    await expect.poll(() => pool.getActiveRun('conversation-target')).toBeNull();
+    expect(pool.getResilienceProofDiagnostics().runs).toEqual({ active: 0, waiting: 0 });
+  });
+
+  it('treats agent settlement as authoritative when the RPC confirmation is still pending', async () => {
+    const worker = new FakeWorker('worker-target');
+    const pool = new PiWorkerPool({
+      maxIdle: 2,
+      openWorker: async () => ({
+        worker,
+        session: { piSessionId: 'session-target', sessionKey: 'key-target' },
+      }),
+    });
+    await pool.prepare(conversation('conversation-target'));
+    worker.pendNext('prompt');
+    const ticket = pool.startTopLevel({
+      conversationId: 'conversation-target',
+      runId: 'run-settled-first',
+      command: { type: 'prompt', message: 'settle before response' },
+    });
+    await expect.poll(() => worker.requests.some(({ type }) => type === 'prompt')).toBe(true);
+
+    worker.emit({ type: 'agent_settled' });
+
+    await expect(ticket.accepted).resolves.toMatchObject({ success: true });
+    expect(pool.getActiveRun('conversation-target')).toBeNull();
+    expect(pool.getResilienceProofDiagnostics().runs).toEqual({ active: 0, waiting: 0 });
+  });
+
   it('injects a proof failure only into the requested current generation', async () => {
     const workers = new Map<string, FakeWorker>();
     const pool = new PiWorkerPool({