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 31192aa..66e697f 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: Planner corrections complete; re-review pending +- Status: Second planner corrections complete; re-review pending ## Scope @@ -68,6 +68,15 @@ prioritizes child reservations so a running parent cannot deadlock while waiting on its own child. The child enters the priority queue before idle reclamation releases capacity. +- Planner re-review of `3de85d6` confirmed the original three findings closed, + then reproduced two further supported failure paths: a child reservation + could wait forever when no parent was idle during the one-time reclaim but a + spawning/running parent became reclaimable later; and reclaim/worker-stop + failure could leave an unowned process lease. The pool now wakes pending + reclaimers on `ready`/`idle` state transitions without polling. Scheduler + 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`. - 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 @@ -101,7 +110,10 @@ parent reclamation; parallel sibling preservation; chain skip/abort rules; parent generation cancellation with no orphan and zero leaked permits; recursive child rejection; parent/child same-project write-lease queuing; - and unknown schema/version raw-payload suppression. + 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. - 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 @@ -117,18 +129,17 @@ 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, 106 passed and 1 staged-only +- All cumulative Pi tests passed: 22 files, 110 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. -- First full-suite run, executed concurrently with the build, passed 2208/2209 - and hit the known Windows temporary JSON `rename` `EPERM` in the unchanged - conversation store. The failing test passed in isolation, then the serial - full-suite rerun passed. After planner corrections the final serial full - suite passed: 202 files, 2209 passed and 1 staged-only skipped. +- 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. - 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 @@ -153,7 +164,9 @@ Future impact: any final Main composition must pass one `PiProcessBudget` to both the parent pool and child scheduler, preserve child-priority budget reservations ahead of normal parent-start waiters, and wire the scheduler's - capacity reclaimer to `PiWorkerPool.reclaimIdleWorker`. 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. + 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. diff --git a/electron/coding-runtime/pi/subagent.ts b/electron/coding-runtime/pi/subagent.ts index d421be4..ecb4a71 100644 --- a/electron/coding-runtime/pi/subagent.ts +++ b/electron/coding-runtime/pi/subagent.ts @@ -58,7 +58,7 @@ export interface PiSubagentDispatchOptions { export interface PiSubagentSchedulerOptions { openChild(input: PiSubagentChildOpenInput): Promise; processBudget: PiProcessBudget; - reclaimProcessCapacity?(): Promise; + reclaimProcessCapacity?(signal?: AbortSignal): Promise; createId?: (kind: 'dispatch' | 'task') => string; } @@ -216,7 +216,7 @@ export class PiSubagentScheduler { private readonly openChild: PiSubagentSchedulerOptions['openChild']; private readonly processBudget: PiProcessBudget; private readonly childPermits: FifoSemaphore; - private readonly reclaimProcessCapacity: (() => Promise) | undefined; + private readonly reclaimProcessCapacity: ((signal?: AbortSignal) => Promise) | undefined; private readonly createId: NonNullable; private readonly dispatches = new Map(); private readonly parentDispatches = new Map>(); @@ -353,12 +353,7 @@ export class PiSubagentScheduler { let child: PiSubagentChild | undefined; try { releaseChild = await this.childPermits.acquire(record.controller.signal); - const budgetWasFull = this.processBudget.activeCount >= this.processBudget.maxProcesses; - const processLeaseFlight = this.processBudget.acquire(record.controller.signal, 'child'); - if (budgetWasFull && this.reclaimProcessCapacity) { - await this.reclaimProcessCapacity(); - } - processLease = await processLeaseFlight; + processLease = await this.acquireProcessLease(record.controller.signal); if (record.controller.signal.aborted) throw new PiSubagentChildError('SUBAGENT_ABORTED'); projected.status = 'running'; this.emit(details, onUpdate); @@ -395,6 +390,33 @@ export class PiSubagentScheduler { } } + private async acquireProcessLease(signal: AbortSignal): Promise { + const reservation = new AbortController(); + const abortReservation = () => reservation.abort(); + signal.addEventListener('abort', abortReservation, { once: true }); + if (signal.aborted) reservation.abort(); + const budgetWasFull = this.processBudget.activeCount >= this.processBudget.maxProcesses; + const processLeaseFlight = this.processBudget.acquire(reservation.signal, 'child'); + try { + if (budgetWasFull && this.reclaimProcessCapacity) { + const outcome = await Promise.race([ + processLeaseFlight.then((lease) => ({ lease })), + this.reclaimProcessCapacity(reservation.signal).then(() => null), + ]); + if (outcome) return outcome.lease; + } + return await processLeaseFlight; + } catch (error) { + reservation.abort(); + const orphanedLease = await processLeaseFlight.catch(() => undefined); + orphanedLease?.release(); + throw error; + } finally { + signal.removeEventListener('abort', abortReservation); + reservation.abort(); + } + } + private markRemaining( details: SubagentDetailsV1, from: number, diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index 1fc4d85..3ac7614 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -230,6 +230,7 @@ export class PiWorkerPool { private readonly activeRuns = new Map(); private readonly generations = new Map(); private readonly listeners = new Set<(event: PiWorkerPoolEvent) => void>(); + private readonly reclaimWaiters = new Set<() => void>(); private commandSequence = 0; private runningCount = 0; private useSequence = 0; @@ -274,6 +275,7 @@ export class PiWorkerPool { .then(async ({ opened: { worker, session }, lease }) => { const record = this.createRecord(conversation, worker, session, generation, revision, lease); this.workers.set(conversation.conversationId, record); + this.notifyReclaimableWorker(); const state = this.publicState(record); await this.trimIdleWorkers(); return state; @@ -328,6 +330,7 @@ export class PiWorkerPool { lease, ); this.workers.set(conversation.conversationId, record); + this.notifyReclaimableWorker(); await this.trimIdleWorkers(); return this.publicState(record); })().finally(() => { @@ -344,13 +347,19 @@ export class PiWorkerPool { return record ? this.publicState(record) : null; } - async reclaimIdleWorker(): Promise { - const record = [...this.workers.values()] - .filter((candidate) => candidate.state === 'ready' || candidate.state === 'idle') - .sort((left, right) => left.lastUsed - right.lastUsed)[0]; - if (!record) return false; - await this.evict(record); - return true; + async reclaimIdleWorker(signal?: AbortSignal): Promise { + 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()] + .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; + continue; + } + await this.waitForReclaimableWorker(signal); + } } subscribe(listener: (event: PiWorkerPoolEvent) => void): () => void { @@ -406,6 +415,7 @@ export class PiWorkerPool { const record = this.workers.get(conversationId); if (record && record.state !== 'crashed') { record.state = 'idle'; + this.notifyReclaimableWorker(); try { this.revisions.settleRun(record.revisionWorkerId); } catch { @@ -605,6 +615,7 @@ export class PiWorkerPool { if (current && this.workers.get(run.conversationId) === current && current.state !== 'crashed') { current.state = 'idle'; + this.notifyReclaimableWorker(); if (beganRun) this.revisions.settleRun(current.revisionWorkerId); void this.trimIdleWorkers(); } @@ -619,6 +630,7 @@ export class PiWorkerPool { this.activeRuns.delete(conversationId); this.runningCount -= 1; record.state = 'idle'; + this.notifyReclaimableWorker(); const revisionAction = this.revisions.settleRun(record.revisionWorkerId); if (!this.shuttingDown && (record.reconfigureAfterSettled || revisionAction.action === 'rebuild-after-settled')) { @@ -716,6 +728,7 @@ export class PiWorkerPool { ? 'running' : 'ready'; this.workers.set(conversationId, replacement); + if (replacement.state === 'ready') this.notifyReclaimableWorker(); this.emit({ type: 'worker.replaced', conversationId, @@ -796,15 +809,16 @@ export class PiWorkerPool { await Promise.all(idle.slice(0, excess).map((record) => this.evict(record))); } - private async evict(record: WorkerRecord): Promise { + private async evict(record: WorkerRecord): Promise { const conversationId = record.conversation.conversationId; - if (this.workers.get(conversationId) !== record) return; + if (this.workers.get(conversationId) !== record) return false; record.unsubscribeEvent(); record.unsubscribeInvalidation(); this.cancelGenerationResources(record); this.revisions.removeWorker(record.revisionWorkerId); this.workers.delete(conversationId); await this.stopAndRelease(record); + return true; } private launchWaitingRuns(): void { @@ -902,9 +916,39 @@ export class PiWorkerPool { } private async stopAndRelease(record: WorkerRecord): Promise { - await record.worker.stop(); - record.processLease?.release(); - record.processLease = null; + try { + await record.worker.stop(); + } finally { + record.processLease?.release(); + record.processLease = null; + } + } + + private waitForReclaimableWorker(signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const cleanup = () => { + this.reclaimWaiters.delete(wake); + signal?.removeEventListener('abort', cancel); + this.shutdownController.signal.removeEventListener('abort', cancel); + }; + const wake = () => { + cleanup(); + resolve(); + }; + const cancel = () => { + cleanup(); + reject(new Error( + this.shuttingDown ? 'Pi worker pool is shutting down' : 'Pi idle worker reclaim cancelled', + )); + }; + this.reclaimWaiters.add(wake); + signal?.addEventListener('abort', cancel, { once: true }); + this.shutdownController.signal.addEventListener('abort', cancel, { once: true }); + }); + } + + private notifyReclaimableWorker(): void { + for (const wake of [...this.reclaimWaiters]) wake(); } private emit(event: PiWorkerPoolEvent): void { diff --git a/tests/unit/pi-subagent.test.ts b/tests/unit/pi-subagent.test.ts index 4b5f37b..36d95e9 100644 --- a/tests/unit/pi-subagent.test.ts +++ b/tests/unit/pi-subagent.test.ts @@ -8,7 +8,7 @@ import { type PiSubagentChild, type PiSubagentChildOpenInput, } from '../../electron/coding-runtime/pi/subagent'; -import { PiProcessBudget } from '../../electron/coding-runtime/pi/worker-pool'; +import { PiProcessBudget, PiWorkerPool } from '../../electron/coding-runtime/pi/worker-pool'; function parent(runId = 'run-a') { return { @@ -251,4 +251,140 @@ describe('Pi subagent scheduler', () => { expect(processBudget.activeCount).toBe(0); await scheduler.close(); }); + + it('reclaims a parent that becomes idle after child capacity is already reserved', async () => { + const parentOpenGate = deferred(); + const processBudget = new PiProcessBudget(1); + let parentStopped = false; + let childRan = false; + const pool = new PiWorkerPool({ + processBudget, + maxIdle: 1, + openWorker: async () => { + await parentOpenGate.promise; + return { + worker: { + id: 'delayed-parent', + generation: 1, + async request() { + return { type: 'response' as const, id: 'parent', success: true as const }; + }, + async send() {}, + subscribe() { return () => undefined; }, + subscribeInvalidation() { return () => undefined; }, + async stop() { + parentStopped = true; + return { mode: 'stdin-close' as const, code: 0, signal: null }; + }, + }, + session: { piSessionId: 'parent-session', sessionKey: 'parent-key' }, + }; + }, + }); + const preparingParent = pool.prepare({ + conversationId: 'delayed-parent', + projectId: 'project-a', + agentId: 'agent-a', + title: 'Delayed parent', + model: { + model: { accountId: 'account-a', modelId: 'model-a', thinkingLevel: 'medium' }, + modelResolution: 'resolved', + }, + }); + await expect.poll(() => processBudget.activeCount).toBe(1); + + const scheduler = new PiSubagentScheduler({ + processBudget, + reclaimProcessCapacity: (signal) => pool.reclaimIdleWorker(signal), + openChild: async (input) => ({ + id: input.taskId, + async run() { + childRan = true; + return { summary: 'done' }; + }, + async stop() {}, + }), + }); + const child = scheduler.dispatch({ + ...parent('delayed-idle'), + request: { mode: 'single', tasks: [task('agent-a')] }, + }); + await expect.poll(() => processBudget.waitingCount).toBe(1); + + parentOpenGate.resolve(); + await preparingParent; + await expect(child).resolves.toMatchObject({ + details: { tasks: [{ status: 'complete' }] }, + }); + expect(parentStopped).toBe(true); + expect(childRan).toBe(true); + expect(processBudget.activeCount).toBe(0); + expect(processBudget.waitingCount).toBe(0); + await scheduler.close(); + await pool.shutdown(); + }); + + it('cancels a queued child reservation when the capacity reclaimer rejects', async () => { + const processBudget = new PiProcessBudget(1); + const parentLease = await processBudget.acquire(); + const scheduler = new PiSubagentScheduler({ + processBudget, + reclaimProcessCapacity: async () => { + throw new Error('reclaim failed'); + }, + openChild: async () => { + throw new Error('child must not open'); + }, + }); + + await expect(scheduler.dispatch({ + ...parent('reclaim-rejected'), + request: { mode: 'single', tasks: [task('agent-a')] }, + })).resolves.toMatchObject({ + details: { tasks: [{ status: 'error', errorCode: 'SUBAGENT_CHILD_FAILED' }] }, + }); + expect(processBudget.activeCount).toBe(1); + expect(processBudget.waitingCount).toBe(0); + parentLease.release(); + expect(processBudget.activeCount).toBe(0); + expect(processBudget.waitingCount).toBe(0); + await scheduler.close(); + }); + + it('cancels both capacity waits when the parent aborts before an idle worker exists', async () => { + const processBudget = new PiProcessBudget(1); + const parentLease = await processBudget.acquire(); + let reclaimerCancelled = false; + const scheduler = new PiSubagentScheduler({ + processBudget, + reclaimProcessCapacity: async (signal) => await new Promise((_resolve, reject) => { + const cancel = () => { + reclaimerCancelled = true; + reject(new Error('reclaim cancelled')); + }; + if (signal?.aborted) cancel(); + else signal?.addEventListener('abort', cancel, { once: true }); + }), + openChild: async () => { + throw new Error('child must not open'); + }, + }); + const controller = new AbortController(); + const child = scheduler.dispatch({ + ...parent('reclaim-aborted'), + request: { mode: 'single', tasks: [task('agent-a')] }, + }, { signal: controller.signal }); + await expect.poll(() => processBudget.waitingCount).toBe(1); + + controller.abort(); + await expect(child).resolves.toMatchObject({ + details: { tasks: [{ status: 'aborted', errorCode: 'SUBAGENT_ABORTED' }] }, + }); + expect(reclaimerCancelled).toBe(true); + expect(processBudget.activeCount).toBe(1); + expect(processBudget.waitingCount).toBe(0); + parentLease.release(); + expect(processBudget.activeCount).toBe(0); + await scheduler.close(); + }); }); diff --git a/tests/unit/pi-worker-pool.test.ts b/tests/unit/pi-worker-pool.test.ts index 4228b50..ddd9b81 100644 --- a/tests/unit/pi-worker-pool.test.ts +++ b/tests/unit/pi-worker-pool.test.ts @@ -227,6 +227,33 @@ describe('Pi worker pool', () => { expect(processBudget.activeCount).toBe(0); }); + it('releases the process lease even when stopping an idle worker fails', async () => { + const processBudget = new PiProcessBudget(1); + const pool = new PiWorkerPool({ + processBudget, + maxIdle: 1, + openWorker: async ({ conversation: input }) => { + const worker = new FakeWorker(`worker-${input.conversationId}`); + worker.stop = async () => { + worker.stopped = true; + throw new Error('stop failed'); + }; + return { + worker, + session: { + piSessionId: `session-${input.conversationId}`, + sessionKey: `key-${input.conversationId}`, + }, + }; + }, + }); + await pool.prepare(conversation('conversation-stop-failure')); + + await expect(pool.dispose('conversation-stop-failure')).rejects.toThrow('stop failed'); + 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({