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 d49ea3a..a159295 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: Third planner correction complete; final re-review pending +- Status: Fourth planner correction complete; final re-review pending ## Scope @@ -87,6 +87,13 @@ 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. +- Planner re-review of `1727b75` confirmed the 4+4 deadlock closed, then + reproduced a supported stop/dispatch overlap: a queued worker could be + selected while its suspension `stop()` was still pending, so `ensureFresh` + saw the not-yet-released old lease and sent the prompt to a stopping + generation. `ensureFresh` now awaits any existing `processStopFlight`, + rechecks record ownership/rebuild state, and only then reopens a lease-less + worker. The old generation never receives the queued prompt. - 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 @@ -126,7 +133,10 @@ 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. + FIFO with unchanged session bindings and generation 2. Their `stop()` calls + are deliberately held behind a gate while an earlier parent settles; the + test confirms no prompt reaches the stopping generation before the gate and + the original prompt reaches only the reopened generation. - 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 @@ -149,11 +159,10 @@ warnings outside this task. - `pnpm run build:vite`: passed for Renderer, Main, Preload, and utility worker bundles; existing chunk-size/dynamic-import warnings remain. -- 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. +- The final full-suite first pass hit the known Windows temporary JSON `rename` + `EPERM` in the unchanged conversation store while the other 2213 tests + passed. The failing runtime test passed 1/1 in isolation and the single full + rerun passed: 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 @@ -183,7 +192,9 @@ `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. + session bindings, and reopen them as a new generation when scheduled. A + queued run selected while suspension is still stopping must await that stop + flight and recheck ownership before it can reopen or send the prompt. 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 2ea2d3d..cdce787 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -657,10 +657,18 @@ export class PiWorkerPool { } private async ensureFresh(record: WorkerRecord): Promise { - const current = this.workers.get(record.conversation.conversationId); + const conversationId = record.conversation.conversationId; + let current = this.workers.get(conversationId); 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.processStopFlight) { + await record.processStopFlight; + current = this.workers.get(conversationId); + 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; diff --git a/tests/unit/pi-worker-pool.test.ts b/tests/unit/pi-worker-pool.test.ts index a370db8..17f633f 100644 --- a/tests/unit/pi-worker-pool.test.ts +++ b/tests/unit/pi-worker-pool.test.ts @@ -258,12 +258,22 @@ describe('Pi worker pool', () => { it('suspends and later resumes the oldest queued parent to make child capacity', async () => { const processBudget = new PiProcessBudget(8); const workers = new Map(); + const queuedStopGate = deferred(); + let stoppingQueuedWorkers = 0; const pool = new PiWorkerPool({ processBudget, maxRunning: 4, maxIdle: 8, openWorker: async ({ conversation: input, generation, existingSession }) => { const worker = new FakeWorker(`worker-${input.conversationId}-${generation}`); + if (generation === 1 && Number(input.conversationId.split('-').at(-1)) > 4) { + worker.stop = async () => { + worker.stopped = true; + stoppingQueuedWorkers += 1; + await queuedStopGate.promise; + return { mode: 'stdin-close' as const, code: 0, signal: null }; + }; + } workers.set(input.conversationId, [...(workers.get(input.conversationId) ?? []), worker]); return { worker, @@ -297,7 +307,7 @@ describe('Pi worker pool', () => { async stop() {}, }), }); - await expect(scheduler.dispatch({ + const children = scheduler.dispatch({ conversationId: ids[0]!, workerGeneration: 1, runId: 'run-child', @@ -310,27 +320,47 @@ describe('Pi worker pool', () => { toolProfile: 'read-only', })), }, - })).resolves.toMatchObject({ - details: { - tasks: Array.from({ length: 4 }, () => ({ status: 'complete' })), - }, }); + await expect.poll(() => stoppingQueuedWorkers).toBe(4); + expect(processBudget.activeCount).toBe(8); + expect(processBudget.waitingCount).toBe(4); const firstQueuedId = ids[4]!; - expect(ids.slice(4).map((id) => workers.get(id)?.[0]?.stopped)) - .toEqual([true, true, true, true]); + let firstQueuedAccepted = false; + void tickets[4]!.accepted.then( + () => { firstQueuedAccepted = true; }, + () => undefined, + ); + workers.get(ids[0]!)?.[0]?.emit({ type: 'agent_settled' }); + await new Promise((resolvePromise) => setImmediate(resolvePromise)); + expect(firstQueuedAccepted).toBe(false); + expect(workers.get(ids[4]!)?.length).toBe(1); + expect(workers.get(ids[4]!)?.[0]?.requests).toHaveLength(0); expect(pool.getState(firstQueuedId)).toMatchObject({ - state: 'queued', + state: 'running', generation: 1, session: { piSessionId: `session-${firstQueuedId}`, sessionKey: `key-${firstQueuedId}`, }, }); - expect(processBudget.activeCount).toBe(4); + + queuedStopGate.resolve(); + await expect(children).resolves.toMatchObject({ + details: { + tasks: Array.from({ length: 4 }, () => ({ status: 'complete' })), + }, + }); + + expect(ids.slice(4).map((id) => workers.get(id)?.[0]?.stopped)) + .toEqual([true, true, true, true]); + await expect.poll(() => workers.get(firstQueuedId)?.length).toBe(2); + await expect.poll(() => workers.get(firstQueuedId)?.[1]?.requests.length).toBe(1); + await expect(tickets[4]!.accepted).resolves.toMatchObject({ success: true }); + expect(processBudget.activeCount).toBe(5); expect(processBudget.waitingCount).toBe(0); - for (let index = 0; index < 4; index += 1) { + for (let index = 1; 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);