From e79aeffffab16c4c5570971e24d1a20997be681d Mon Sep 17 00:00:00 2001 From: brother7 <7brother7@gmail.com> Date: Sat, 22 Aug 2026 23:42:22 +0800 Subject: [PATCH] fix: close Pi worker rebuild races --- .../tasks/20260822-pi-worker-pool-8a4e2c91.md | 5 +- electron/coding-runtime/pi/worker-pool.ts | 19 ++++- tests/unit/pi-worker-pool.test.ts | 78 +++++++++++++++++++ 3 files changed, 99 insertions(+), 3 deletions(-) diff --git a/.project-docs/30-worklog/tasks/20260822-pi-worker-pool-8a4e2c91.md b/.project-docs/30-worklog/tasks/20260822-pi-worker-pool-8a4e2c91.md index 43ee361..b65cacd 100644 --- a/.project-docs/30-worklog/tasks/20260822-pi-worker-pool-8a4e2c91.md +++ b/.project-docs/30-worklog/tasks/20260822-pi-worker-pool-8a4e2c91.md @@ -29,6 +29,7 @@ ## Outcome +- Planner audit of commit `9f55eec` found and the follow-up fixes close two reproducible acceptance gaps: stale rebuild now re-runs idle LRU, and concurrent `recover()` reuses an existing rebuild flight rather than spawning an unowned second replacement. - Implemented the PI-050 runtime/pool/registry seam with one persistent Pi worker session per Conversation, generation-scoped resources, cap-4 fair top-level scheduling, idle LRU, and a shared default-eight process budget for later child workers. - Wired the PI-040 managed resource, credential projection, revision, and single-auth-refresh contracts into worker open/reopen. Cross-account model changes rebuild only the target worker so the new Provider credential remains generation-isolated; active runs settle before that rebuild. - Added targeted recover/dispose/fork/settings/queue/compact/abort orchestration, new-generation `get_state/get_entries` fetch plus snapshot replacement, privacy-safe milestone correlation, and bounded pool shutdown. PI-060 retains ownership of projecting the returned session tree into transcript nodes. @@ -38,10 +39,10 @@ ## Verification -- Final focused suite: 7 files / 34 tests passed, covering prepare/binding single-flight, cap-4 FIFO permits, default-eight shared process budget, running-safe idle LRU, generation resource cleanup, stale revisions, in-flight shutdown, RPC acceptance/rejection, model/thinking target isolation, one-refresh auth recovery, recover/fork/dispose, managed credentials/resources, and a two-real-child-process abort-isolation integration. +- Final focused suite: 7 files / 36 tests passed, covering prepare/binding single-flight, cap-4 FIFO permits, default-eight shared process budget, running-safe idle LRU including post-rebuild trimming, generation resource cleanup, stale revisions, rebuild/recover single-flight, in-flight shutdown, RPC acceptance/rejection, model/thinking target isolation, one-refresh auth recovery, recover/fork/dispose, managed credentials/resources, and a two-real-child-process abort-isolation integration. - `corepack pnpm run typecheck`: passed. - `corepack pnpm run lint:check`: passed with 0 errors and 6 pre-existing warnings outside PI-050 files. -- `corepack pnpm test`: 193 files / 2169 tests passed. +- `corepack pnpm test`: 193 files / 2171 tests passed after the planner-audit fixes. - `corepack pnpm run build:vite`: passed for Renderer, Main, Preload, and release utility output; existing dynamic-import and chunk-size warnings remain unchanged. - `corepack pnpm run test:electron:windows`: 1 file / 3 tests passed. No Host API or Renderer behavior is wired in PI-050, so there was no applicable user-visible Playwright spec to add or run. - `node scripts/probe-pi-provider-contracts.mjs --timeout-ms 30000`: all four local provider-shaped contracts passed with distinct sessions, overlapping two-worker image turns, target-only abort, environment credential references, and clean stdin-close. The report explicitly retained `realTurnVerified=false` and `realProviderDecision=explicitly-waived-accepted-risk`. diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index c202ce7..5ca2c98 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -470,9 +470,25 @@ export class PiWorkerPool { if (this.shuttingDown) throw new Error('Pi worker pool is shutting down'); const pendingPrepare = this.prepareFlights.get(conversationId); if (pendingPrepare) await pendingPrepare; - const record = this.workers.get(conversationId); + let record = this.workers.get(conversationId); if (!record) throw new Error('Conversation worker is not prepared'); this.cancelConversationRuns(conversationId, new Error('Conversation worker is recovering')); + while (record.rebuildFlight) { + const existingFlight = record.rebuildFlight; + try { + return this.publicState(await existingFlight); + } catch (error) { + if (this.shuttingDown) throw error; + const current = this.workers.get(conversationId); + if (current && current !== record) { + record = current; + continue; + } + if (record.rebuildFlight !== existingFlight) continue; + record.rebuildFlight = undefined; + break; + } + } record.rebuildFlight = this.beginRebuild(record, this.revisions.current); return this.publicState(await record.rebuildFlight); } @@ -673,6 +689,7 @@ export class PiWorkerPool { generation, state: this.publicState(replacement), }); + await this.trimIdleWorkers(); return replacement; } catch (error) { record.state = 'crashed'; diff --git a/tests/unit/pi-worker-pool.test.ts b/tests/unit/pi-worker-pool.test.ts index 9b6ee83..15088a1 100644 --- a/tests/unit/pi-worker-pool.test.ts +++ b/tests/unit/pi-worker-pool.test.ts @@ -354,6 +354,84 @@ describe('Pi worker pool', () => { ])); }); + it('re-applies the idle LRU after a running stale worker rebuilds on settle', async () => { + const workers = new Map(); + const pool = new PiWorkerPool({ + maxIdle: 1, + openWorker: async ({ conversation: input, generation, existingSession }) => { + const worker = new FakeWorker(`worker-${input.conversationId}-${generation}`); + workers.set(input.conversationId, [...(workers.get(input.conversationId) ?? []), worker]); + return { + worker, + session: existingSession ?? { + piSessionId: `session-${input.conversationId}`, + sessionKey: `key-${input.conversationId}`, + }, + }; + }, + }); + await pool.prepare(conversation('conversation-running')); + const running = pool.startTopLevel({ + conversationId: 'conversation-running', + runId: 'run-running', + command: { type: 'prompt', message: 'running' }, + }); + await running.accepted; + await pool.prepare(conversation('conversation-idle')); + pool.markProviderStale(); + + workers.get('conversation-running')![0]!.emit({ type: 'agent_settled' }); + await expect.poll(() => workers.get('conversation-running')?.length).toBe(2); + await expect.poll(() => workers.get('conversation-idle')![0]!.stopped).toBe(true); + expect(pool.getState('conversation-idle')).toBeNull(); + expect(pool.getState('conversation-running')).toMatchObject({ state: 'ready', generation: 2 }); + }); + + it('reuses an in-flight rebuild for recover and leaves no unowned replacement worker', async () => { + const generationTwoGate = deferred(); + const workers: FakeWorker[] = []; + const openedGenerations: number[] = []; + const pool = new PiWorkerPool({ + maxIdle: 2, + openWorker: async ({ conversation: input, generation, existingSession }) => { + openedGenerations.push(generation); + const worker = new FakeWorker(`worker-${input.conversationId}-${generation}`); + workers.push(worker); + if (generation === 2) await generationTwoGate.promise; + return { + worker, + session: existingSession ?? { + piSessionId: `session-${input.conversationId}`, + sessionKey: `key-${input.conversationId}`, + }, + }; + }, + }); + await pool.prepare(conversation('conversation-a')); + pool.markProviderStale(); + const ticket = pool.startTopLevel({ + conversationId: 'conversation-a', + runId: 'run-a', + command: { type: 'prompt', message: 'do not replay' }, + }); + const ticketOutcome = ticket.accepted.then( + () => 'resolved', + (error: unknown) => error instanceof Error ? error.message : String(error), + ); + await expect.poll(() => openedGenerations).toEqual([1, 2]); + + const recovered = pool.recover('conversation-a'); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(openedGenerations).toEqual([1, 2]); + generationTwoGate.resolve(); + + await expect(recovered).resolves.toMatchObject({ generation: 2, state: 'ready' }); + expect(await ticketOutcome).toMatch(/recovering|cancelled/); + await pool.shutdown(); + expect(workers).toHaveLength(2); + expect(workers.every((worker) => worker.stopped)).toBe(true); + }); + it('rejects queued work and stops every parent worker during app shutdown', async () => { const workers: FakeWorker[] = []; const pool = new PiWorkerPool({