From 1727b75f3008e9fd36597dd99fa6c49272dc2f16 Mon Sep 17 00:00:00 2001 From: brother7 <7brother7@gmail.com> Date: Sun, 23 Aug 2026 13:53:28 +0800 Subject: [PATCH] fix: reclaim queued Pi parent capacity --- .../20260823-pi-child-workers-5c8e2a71.md | 39 +++++-- electron/coding-runtime/pi/worker-pool.ts | 49 +++++++-- tests/unit/pi-worker-pool.test.ts | 100 ++++++++++++++++++ 3 files changed, 171 insertions(+), 17 deletions(-) diff --git a/.project-docs/30-worklog/tasks/20260823-pi-child-workers-5c8e2a71.md b/.project-docs/30-worklog/tasks/20260823-pi-child-workers-5c8e2a71.md index 66e697f..d49ea3a 100644 --- a/.project-docs/30-worklog/tasks/20260823-pi-child-workers-5c8e2a71.md +++ b/.project-docs/30-worklog/tasks/20260823-pi-child-workers-5c8e2a71.md @@ -8,7 +8,7 @@ - Worktree: D:\Datas\OthersProjects\makelore-pi-child-workers-5c8e2a71 - Base commit: b806c78139aa11e338c680d4c3fa92e076901aa2 - Owner: codex -- Status: Second planner corrections complete; re-review pending +- Status: Third planner correction complete; final re-review pending ## Scope @@ -77,6 +77,16 @@ reservations race direct capacity against this cancellable event-driven reclaim and always settle/release on reclaim rejection, parent abort, or any other early exit. Parent `stop()` now releases its process lease in `finally`. +- Planner re-review of `a10b98e` confirmed both lease-lifecycle findings closed, + then reproduced the production-capacity shape `4 running + 4 queued`: queued + parent workers retained all eight process leases, leaving no ready/idle + worker for four children to reclaim. The pool now prefers ready/idle + eviction, then suspends the oldest queued parent process while retaining its + top-level FIFO entry, Conversation state, and session binding. When its turn + arrives, the same queue entry reopens that session as a new worker generation + before sending the original prompt. This closes the deadlock without + dropping queued work, reordering it, adding another queue, or exceeding the + shared eight-process cap. - Added the managed ephemeral child opener. It resolves only enabled, unarchived project Agents from `.niancode/project.json`, materializes their exact model/prompt/skills, keeps credentials in the child environment and @@ -113,7 +123,10 @@ unknown schema/version raw-payload suppression; first reclaim finding no idle followed by delayed parent readiness and automatic child execution; reclaim rejection/abort with zero queued lease; and stop rejection with zero - active/waiting lease. + active/waiting lease. A production-scale regression also covers four running + plus four queued parents dispatching four parallel children: all four queued + workers suspend, all children complete, and the queued parents then resume + FIFO with unchanged session bindings and generation 2. - Locked real Pi 0.84.2 workspace smoke passed for both the parent worker and an ephemeral read-only child launched through Electron Node with the child extension role and `--no-session`. A probe extension reads Pi's actual @@ -129,17 +142,18 @@ authenticated Main bridge; the active-tool process probes and bridge execute test jointly cover visibility and tool invocation without an external Provider. -- All cumulative Pi tests passed: 22 files, 110 passed and 1 staged-only +- All cumulative Pi tests passed: 22 files, 111 passed and 1 staged-only skipped; the staged-only command passed separately as described above. - `pnpm run typecheck`: passed. - `pnpm run lint:check`: passed with 0 errors and 6 pre-existing frontend warnings outside this task. - `pnpm run build:vite`: passed for Renderer, Main, Preload, and utility worker bundles; existing chunk-size/dynamic-import warnings remain. -- The final full-suite first pass hit the known Windows temporary JSON `rename` - `EPERM` in the unchanged conversation store. The failing runtime test passed - 1/1 in isolation, and the single full-suite rerun passed: 202 files, 2213 - passed and 1 staged-only skipped. +- An earlier full-suite pass hit the known Windows temporary JSON `rename` + `EPERM` in the unchanged conversation store; the failing runtime test passed + 1/1 in isolation and the rerun passed. After the queued-parent correction, + the final full suite passed on its first run: 202 files, 2214 passed and 1 + staged-only skipped. - Real external Provider validation remains **Explicitly Waived / Accepted Risk** with `realTurnVerified=false`. Provider concurrency, credential isolation, protocol compatibility, and image-path risk are accepted rather @@ -166,7 +180,10 @@ reservations ahead of normal parent-start waiters, and wire the scheduler's cancellable capacity reclaimer to `PiWorkerPool.reclaimIdleWorker`. The pool must notify an already waiting child when a spawning/running parent becomes - `ready`/`idle`, and stop/reclaim failure must never retain a lease. Semantic - conflicts: none with the accepted PI runtime specification; this makes its - parent/child cap executable. Human confirmation required: no, unless - integration changes the accepted process-cap policy. + `ready`/`idle`, and stop/reclaim failure must never retain a lease. If the + cap consists only of running and queued parents, it must suspend queued + parent processes in FIFO-safe/LRU order, retain their queue entries and + session bindings, and reopen them as a new generation when scheduled. + Semantic conflicts: none with the accepted PI runtime specification; this + makes its parent/child cap executable. Human confirmation required: no, + unless integration changes the accepted process-cap policy. diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index 3ac7614..2ea2d3d 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -351,11 +351,20 @@ export class PiWorkerPool { while (true) { if (signal?.aborted) throw new Error('Pi idle worker reclaim cancelled'); if (this.shuttingDown) throw new Error('Pi worker pool is shutting down'); - const record = [...this.workers.values()] + const idleRecord = [...this.workers.values()] .filter((candidate) => candidate.state === 'ready' || candidate.state === 'idle') .sort((left, right) => left.lastUsed - right.lastUsed)[0]; - if (record) { - if (await this.evict(record)) return true; + if (idleRecord) { + if (await this.evict(idleRecord)) return true; + continue; + } + const queuedRecord = [...this.workers.values()] + .filter((candidate) => candidate.state === 'queued' + && candidate.processLease !== null + && !candidate.processStopFlight) + .sort((left, right) => left.lastUsed - right.lastUsed)[0]; + if (queuedRecord) { + if (await this.suspendQueuedWorker(queuedRecord)) return true; continue; } await this.waitForReclaimableWorker(signal); @@ -501,6 +510,7 @@ export class PiWorkerPool { this.waitingRuns.push(pending); pending.queuedAt = this.now(); record.state = 'queued'; + this.notifyReclaimableWorker(); return { queuePosition: this.waitingRuns.length, accepted }; } @@ -550,7 +560,7 @@ export class PiWorkerPool { this.cancelGenerationResources(record); this.revisions.removeWorker(record.revisionWorkerId); if (this.workers.get(conversationId) === record) this.workers.delete(conversationId); - await this.stopAndRelease(record); + await this.ensureStoppedAndReleased(record); } private async performShutdown(): Promise { @@ -569,7 +579,7 @@ export class PiWorkerPool { record.unsubscribeInvalidation(); this.cancelGenerationResources(record); this.revisions.removeWorker(record.revisionWorkerId); - await this.stopAndRelease(record); + await this.ensureStoppedAndReleased(record); })); } @@ -651,6 +661,10 @@ export class PiWorkerPool { if (!current) throw new Error('Conversation worker is no longer available'); if (current !== record) return await this.ensureFresh(current); if (record.rebuildFlight) return await record.rebuildFlight; + if (!record.processLease) { + record.rebuildFlight = this.beginRebuild(record, this.revisions.current); + return await record.rebuildFlight; + } const action = this.revisions.beforePrompt(record.revisionWorkerId); if (action.action !== 'rebuild-before-prompt') return record; record.rebuildFlight = this.beginRebuild(record, action.revision); @@ -817,7 +831,23 @@ export class PiWorkerPool { this.cancelGenerationResources(record); this.revisions.removeWorker(record.revisionWorkerId); this.workers.delete(conversationId); - await this.stopAndRelease(record); + await this.ensureStoppedAndReleased(record); + return true; + } + + private async suspendQueuedWorker(record: WorkerRecord): Promise { + const conversationId = record.conversation.conversationId; + if (this.workers.get(conversationId) !== record + || record.state !== 'queued' + || !record.processLease + || record.processStopFlight) { + return false; + } + record.unsubscribeEvent(); + record.unsubscribeInvalidation(); + this.cancelGenerationResources(record); + record.processStopFlight = this.stopAndRelease(record); + await record.processStopFlight; return true; } @@ -924,6 +954,13 @@ export class PiWorkerPool { } } + private async ensureStoppedAndReleased(record: WorkerRecord): Promise { + if (!record.processStopFlight) { + record.processStopFlight = this.stopAndRelease(record); + } + await record.processStopFlight; + } + private waitForReclaimableWorker(signal?: AbortSignal): Promise { return new Promise((resolve, reject) => { const cleanup = () => { diff --git a/tests/unit/pi-worker-pool.test.ts b/tests/unit/pi-worker-pool.test.ts index ddd9b81..a370db8 100644 --- a/tests/unit/pi-worker-pool.test.ts +++ b/tests/unit/pi-worker-pool.test.ts @@ -12,6 +12,7 @@ import type { PiProcessError } from '../../electron/coding-runtime/pi/process-er import { PiProcessError as PiProcessFailure } from '../../electron/coding-runtime/pi/process-errors'; 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'; const MODEL = { model: { @@ -254,6 +255,105 @@ describe('Pi worker pool', () => { expect(processBudget.waitingCount).toBe(0); }); + it('suspends and later resumes the oldest queued parent to make child capacity', async () => { + const processBudget = new PiProcessBudget(8); + const workers = new Map(); + const pool = new PiWorkerPool({ + processBudget, + maxRunning: 4, + maxIdle: 8, + 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}`, + }, + }; + }, + }); + const ids = Array.from({ length: 8 }, (_, index) => `conversation-${index + 1}`); + for (const id of ids) await pool.prepare(conversation(id)); + const tickets = ids.map((conversationId, index) => pool.startTopLevel({ + conversationId, + runId: `run-${index + 1}`, + command: { type: 'prompt', message: conversationId }, + })); + for (const ticket of tickets) void ticket.accepted.catch(() => undefined); + await expect.poll(() => ids.map((id) => workers.get(id)?.[0]?.requests.length ?? 0)) + .toEqual([1, 1, 1, 1, 0, 0, 0, 0]); + expect(ids.slice(4).map((id) => pool.getState(id)?.state)) + .toEqual(['queued', 'queued', 'queued', 'queued']); + expect(processBudget.activeCount).toBe(8); + + const scheduler = new PiSubagentScheduler({ + processBudget, + reclaimProcessCapacity: (signal) => pool.reclaimIdleWorker(signal), + openChild: async (input) => ({ + id: input.taskId, + async run() { return { summary: 'child complete' }; }, + async stop() {}, + }), + }); + await expect(scheduler.dispatch({ + conversationId: ids[0]!, + workerGeneration: 1, + runId: 'run-child', + projectId: 'project-a', + request: { + mode: 'parallel', + tasks: Array.from({ length: 4 }, (_, index) => ({ + agentId: `agent-${index + 1}`, + task: `inspect ${index + 1}`, + toolProfile: 'read-only', + })), + }, + })).resolves.toMatchObject({ + details: { + tasks: Array.from({ length: 4 }, () => ({ status: 'complete' })), + }, + }); + + const firstQueuedId = ids[4]!; + expect(ids.slice(4).map((id) => workers.get(id)?.[0]?.stopped)) + .toEqual([true, true, true, true]); + expect(pool.getState(firstQueuedId)).toMatchObject({ + state: 'queued', + generation: 1, + session: { + piSessionId: `session-${firstQueuedId}`, + sessionKey: `key-${firstQueuedId}`, + }, + }); + expect(processBudget.activeCount).toBe(4); + expect(processBudget.waitingCount).toBe(0); + + for (let index = 0; index < 4; index += 1) { + const queuedId = ids[index + 4]!; + workers.get(ids[index]!)?.[0]?.emit({ type: 'agent_settled' }); + await expect.poll(() => workers.get(queuedId)?.length).toBe(2); + await expect.poll(() => workers.get(queuedId)?.[1]?.requests.length).toBe(1); + await expect(tickets[index + 4]!.accepted).resolves.toMatchObject({ success: true }); + } + expect(pool.getState(firstQueuedId)).toMatchObject({ + state: 'running', + generation: 2, + session: { + piSessionId: `session-${firstQueuedId}`, + sessionKey: `key-${firstQueuedId}`, + }, + }); + expect(ids.slice(4).map((id) => workers.get(id)?.length)).toEqual([2, 2, 2, 2]); + expect(processBudget.activeCount).toBe(8); + + await scheduler.close(); + await pool.shutdown(); + expect(processBudget.activeCount).toBe(0); + expect(processBudget.waitingCount).toBe(0); + }); + it('never evicts a running worker when the warm-idle LRU exceeds its cap', async () => { const workers = new Map(); const pool = new PiWorkerPool({