From 3de85d61d7517c74b9086528a2eebdbe9becf6b3 Mon Sep 17 00:00:00 2001 From: brother7 <7brother7@gmail.com> Date: Sun, 23 Aug 2026 13:14:03 +0800 Subject: [PATCH] fix: close Pi subagent review gaps --- .../20260823-pi-child-workers-5c8e2a71.md | 47 +++++-- electron/coding-runtime/pi/subagent.ts | 16 +-- electron/coding-runtime/pi/worker-pool.ts | 17 ++- electron/coding-runtime/pi/worker-process.ts | 4 +- package.json | 1 + scripts/run-pi-subagent-packaged-smoke.mjs | 43 +++++++ tests/unit/pi-rpc-foundation.test.ts | 2 +- tests/unit/pi-subagent.test.ts | 17 ++- tests/unit/pi-worker-process-real.test.ts | 121 +++++++++++++++++- 9 files changed, 234 insertions(+), 34 deletions(-) create mode 100644 scripts/run-pi-subagent-packaged-smoke.mjs 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 bd1acf4..31192aa 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: Implementation complete; planner review pending +- Status: Planner corrections complete; re-review pending ## Scope @@ -62,6 +62,12 @@ a parent/run dispatch registry, shared `PiProcessBudget` acquisition, idle parent reclamation when the eight-process budget is full, and deterministic abort/error/skipped projection without raw child errors. +- Planner review of `f1c7cd8` found that a previously queued parent process + budget waiter could take a reclaimed idle permit before the child. The + corrected budget keeps normal waiters FIFO, keeps child waiters FIFO, and + 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. - 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 @@ -72,6 +78,10 @@ retain only the shared mutation-lease hooks required by coding tools. The authenticated loopback bridge streams only NDJSON `subagent.v1` details and rejects recursive child dispatch and stale identities. +- Planner review of `f1c7cd8` also found that Pi 0.84.2 treats `--tools` as a + strict whitelist across built-in and extension tools. The managed parent + default now explicitly includes `subagent`; child read-only/coding lists + remain exact and exclude both `subagent` and `ask_user`. - Connected subagent dispatches to PI-070 generation resource cancellation so parent abort/crash/recover/dispose and bridge disconnect stop unfinished children and release child/process permits. Coding children join the same @@ -92,12 +102,23 @@ 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. -- Locked real Pi 0.84.2 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`: 2/2 tests passed. This smoke performs - `get_state` and clean stdin shutdown only; it does not call an external +- 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 + active-tool list at `session_start`: the parent includes `subagent`, while + the child is exactly `read,grep,find,ls`. +- `pnpm run test:pi-subagent:packaged` passed 3/3. The dedicated runner uses + PI-030's production bundler to create and validate a temporary staged + production closure/manifest, then starts the staged Pi CLI as an ephemeral + read-only child, verifies `get_state`, the exact active tools and child + extension role, and clean stdin shutdown. The normal suite skips only this + staging case so it does not perform a production `npm ci` on every unit run. +- The real materialized extension bundle test executes `subagent` through the + 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 tests. +- All cumulative Pi tests passed: 22 files, 106 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. @@ -106,7 +127,8 @@ - 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: 202 files, 2209 tests. + full-suite rerun passed. After planner corrections the final serial full + suite passed: 202 files, 2209 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 @@ -129,8 +151,9 @@ the running parents wait on children. Evidence: the focused full-budget scheduler regression test and the shared-budget implementation in PI-080. Future impact: any final Main composition must pass one `PiProcessBudget` to - both the parent pool and child scheduler 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. + 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. diff --git a/electron/coding-runtime/pi/subagent.ts b/electron/coding-runtime/pi/subagent.ts index f7456e0..d421be4 100644 --- a/electron/coding-runtime/pi/subagent.ts +++ b/electron/coding-runtime/pi/subagent.ts @@ -353,8 +353,12 @@ export class PiSubagentScheduler { let child: PiSubagentChild | undefined; try { releaseChild = await this.childPermits.acquire(record.controller.signal); - await this.reclaimIdleCapacity(record.controller.signal); - processLease = await this.processBudget.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; if (record.controller.signal.aborted) throw new PiSubagentChildError('SUBAGENT_ABORTED'); projected.status = 'running'; this.emit(details, onUpdate); @@ -391,14 +395,6 @@ export class PiSubagentScheduler { } } - private async reclaimIdleCapacity(signal: AbortSignal): Promise { - while (!signal.aborted - && this.processBudget.activeCount >= this.processBudget.maxProcesses - && this.reclaimProcessCapacity) { - if (!await this.reclaimProcessCapacity()) break; - } - } - 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 e04490b..1fc4d85 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -116,6 +116,7 @@ export interface PiProcessLease { interface PiProcessBudgetWaiter { signal?: AbortSignal; + priority: 'normal' | 'child'; resolve(lease: PiProcessLease): void; reject(error: Error): void; onAbort?: () => void; @@ -134,11 +135,19 @@ export class PiProcessBudget { get activeCount(): number { return this.active; } get waitingCount(): number { return this.waiters.length; } - acquire(signal?: AbortSignal): Promise { + acquire( + signal?: AbortSignal, + priority: 'normal' | 'child' = 'normal', + ): Promise { if (signal?.aborted) return Promise.reject(new Error('Pi process budget acquisition cancelled')); if (this.active < this.maxProcesses) return Promise.resolve(this.issueLease()); return new Promise((resolve, reject) => { - const waiter: PiProcessBudgetWaiter = { resolve, reject, ...(signal ? { signal } : {}) }; + const waiter: PiProcessBudgetWaiter = { + resolve, + reject, + priority, + ...(signal ? { signal } : {}), + }; if (signal) { waiter.onAbort = () => { const index = this.waiters.indexOf(waiter); @@ -166,7 +175,9 @@ export class PiProcessBudget { private advance(): void { while (this.active < this.maxProcesses && this.waiters.length > 0) { - const waiter = this.waiters.shift() as PiProcessBudgetWaiter; + const childIndex = this.waiters.findIndex(({ priority }) => priority === 'child'); + const [waiter] = this.waiters.splice(childIndex >= 0 ? childIndex : 0, 1); + if (!waiter) return; if (waiter.onAbort && waiter.signal) { waiter.signal.removeEventListener('abort', waiter.onAbort); } diff --git a/electron/coding-runtime/pi/worker-process.ts b/electron/coding-runtime/pi/worker-process.ts index ac49879..4bb52a8 100644 --- a/electron/coding-runtime/pi/worker-process.ts +++ b/electron/coding-runtime/pi/worker-process.ts @@ -65,7 +65,9 @@ export type PiWorkerProcessOptions = { export function buildPiRpcArgs( sessionDir: string, additionalArgs: readonly string[] = [], - tools: readonly string[] = ['read', 'bash', 'edit', 'write', 'grep', 'find', 'ls', 'ask_user'], + tools: readonly string[] = [ + 'read', 'bash', 'edit', 'write', 'grep', 'find', 'ls', 'ask_user', 'subagent', + ], ): string[] { return [ '--mode', 'rpc', diff --git a/package.json b/package.json index 57cfa72..0e66245 100644 --- a/package.json +++ b/package.json @@ -46,6 +46,7 @@ "lint:check": "eslint .", "typecheck": "tsc --noEmit", "test": "vitest run", + "test:pi-subagent:packaged": "node scripts/run-pi-subagent-packaged-smoke.mjs", "test:contract:ai-hardware": "vitest run --config vitest.contract.config.ts", "test:electron:windows": "node scripts/run-electron-vitest.mjs", "verify:electron:release": "node scripts/run-electron-vitest.mjs --verify-release-runtime", diff --git a/scripts/run-pi-subagent-packaged-smoke.mjs b/scripts/run-pi-subagent-packaged-smoke.mjs new file mode 100644 index 0000000..2fcd0b4 --- /dev/null +++ b/scripts/run-pi-subagent-packaged-smoke.mjs @@ -0,0 +1,43 @@ +import { spawn } from 'node:child_process'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; + +import { bundlePiRuntime } from './bundle-pi-runtime.mjs'; +import { defaultPiBundleTarget } from './lib/pi-runtime-bundle.mjs'; + +function runVitest(runtimeRoot) { + return new Promise((resolvePromise, reject) => { + const child = spawn(process.execPath, [ + resolve('node_modules/vitest/vitest.mjs'), + 'run', + 'tests/unit/pi-worker-process-real.test.ts', + ], { + cwd: process.cwd(), + env: { ...process.env, MAKELORE_PI_STAGED_RUNTIME_ROOT: runtimeRoot }, + stdio: 'inherit', + windowsHide: true, + }); + child.once('error', reject); + child.once('exit', (code, signal) => { + if (code === 0) resolvePromise(); + else { + reject(new Error( + `Packaged subagent smoke failed with code ${code ?? 'null'} signal ${signal ?? 'none'}`, + )); + } + }); + }); +} + +const outputRoot = await mkdtemp(join(tmpdir(), 'makelore-pi-subagent-package-')); +try { + const [bundle] = await bundlePiRuntime({ + outputRoot, + targets: [defaultPiBundleTarget()], + }); + if (!bundle) throw new Error('Pi runtime bundler returned no staged runtime'); + await runVitest(bundle.destination); +} finally { + await rm(outputRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }); +} diff --git a/tests/unit/pi-rpc-foundation.test.ts b/tests/unit/pi-rpc-foundation.test.ts index 1dd6d7f..733427e 100644 --- a/tests/unit/pi-rpc-foundation.test.ts +++ b/tests/unit/pi-rpc-foundation.test.ts @@ -167,7 +167,7 @@ describe('Pi worker process', () => { '--no-context-files', '--no-approve', '--tools', - 'read,bash,edit,write,grep,find,ls,ask_user', + 'read,bash,edit,write,grep,find,ls,ask_user,subagent', '--model', 'model-a', ]); expect(buildPiRpcArgs('sessions', ['--no-session'], ['read', 'grep', 'find', 'ls'])) diff --git a/tests/unit/pi-subagent.test.ts b/tests/unit/pi-subagent.test.ts index ac6c0ba..4b5f37b 100644 --- a/tests/unit/pi-subagent.test.ts +++ b/tests/unit/pi-subagent.test.ts @@ -213,7 +213,13 @@ describe('Pi subagent scheduler', () => { const processBudget = new PiProcessBudget(2); const firstParent = await processBudget.acquire(); const idleParent = await processBudget.acquire(); + let waitingParentSettled = false; + const waitingParent = processBudget.acquire().then((lease) => { + waitingParentSettled = true; + return lease; + }); let reclaimed = 0; + let childRanBeforeWaitingParent = false; const scheduler = new PiSubagentScheduler({ processBudget, reclaimProcessCapacity: async () => { @@ -223,7 +229,10 @@ describe('Pi subagent scheduler', () => { }, openChild: async (input) => ({ id: input.taskId, - async run() { return { summary: 'done' }; }, + async run() { + childRanBeforeWaitingParent = !waitingParentSettled; + return { summary: 'done' }; + }, async stop() {}, }), }); @@ -234,8 +243,12 @@ describe('Pi subagent scheduler', () => { details: { tasks: [{ status: 'complete' }] }, }); expect(reclaimed).toBe(1); - expect(processBudget.activeCount).toBe(1); + expect(childRanBeforeWaitingParent).toBe(true); + const waitingParentLease = await waitingParent; + expect(processBudget.activeCount).toBe(2); + waitingParentLease.release(); firstParent.release(); + expect(processBudget.activeCount).toBe(0); await scheduler.close(); }); }); diff --git a/tests/unit/pi-worker-process-real.test.ts b/tests/unit/pi-worker-process-real.test.ts index a5af691..d28320d 100644 --- a/tests/unit/pi-worker-process-real.test.ts +++ b/tests/unit/pi-worker-process-real.test.ts @@ -2,14 +2,49 @@ import { afterEach, describe, expect, it } from 'vitest'; import { realpathSync } from 'node:fs'; -import { mkdir, mkdtemp, readFile, rm } from 'node:fs/promises'; +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; import { createRequire } from 'node:module'; import { tmpdir } from 'node:os'; import { join, resolve } from 'node:path'; import { PiWorkerProcess } from '../../electron/coding-runtime/pi/worker-process'; import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; +import { + PI_RUNTIME_MANIFEST, + PI_RUNTIME_VERSION, +} from '../../scripts/lib/pi-runtime-bundle.mjs'; const scratchRoots: string[] = []; +const packagedRuntimeRoot = process.env.MAKELORE_PI_STAGED_RUNTIME_ROOT; + +async function materializeActiveToolsProbe(root: string): Promise<{ + extensionPath: string; + resultPath: string; +}> { + const extensionPath = join(root, 'active-tools-probe.mjs'); + const resultPath = join(root, 'active-tools.json'); + await writeFile(extensionPath, ` +import { writeFile } from 'node:fs/promises'; +export default function activeToolsProbe(pi) { + pi.on('session_start', async () => { + await writeFile(process.env.MAKELORE_PI_ACTIVE_TOOLS_FILE, JSON.stringify(pi.getActiveTools())); + }); +} +`.trimStart()); + return { extensionPath, resultPath }; +} + +async function readActiveTools(resultPath: string): Promise { + let lastError: unknown; + for (let attempt = 0; attempt < 80; attempt += 1) { + try { + return JSON.parse(await readFile(resultPath, 'utf8')) as string[]; + } catch (error) { + lastError = error; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + } + throw lastError; +} afterEach(async () => { await Promise.all(scratchRoots.splice(0).map((root) => rm(root, { @@ -48,14 +83,18 @@ describe('locked Pi worker process smoke', () => { extensionsDir: join(root, 'extensions'), }); await extensionHost.bindRun('real-conversation', 1, 'real-run'); + const probe = await materializeActiveToolsProbe(root); const worker = await new PiWorkerProcess({ executablePath: electronExecutable, cliPath: join(packageRoot, 'dist', 'cli.js'), cwd, configDir, sessionDir, - additionalArgs: ['--extension', extension.extensionPath], - env: extension.env, + additionalArgs: [ + '--extension', extension.extensionPath, + '--extension', probe.extensionPath, + ], + env: { ...extension.env, MAKELORE_PI_ACTIVE_TOOLS_FILE: probe.resultPath }, sensitiveValues: extension.sensitiveValues, commandTimeoutMs: 5_000, }).start(); @@ -66,6 +105,7 @@ describe('locked Pi worker process smoke', () => { success: true, }); expect(worker.stderrDiagnostic).not.toContain('Failed to load extension'); + expect(await readActiveTools(probe.resultPath)).toContain('subagent'); await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); } finally { await worker.stop().catch(() => undefined); @@ -98,6 +138,7 @@ describe('locked Pi worker process smoke', () => { role: 'child', runId: 'real-parent-run', }); + const probe = await materializeActiveToolsProbe(root); const worker = await new PiWorkerProcess({ executablePath: electronExecutable, cliPath: join(packageRoot, 'dist', 'cli.js'), @@ -105,8 +146,12 @@ describe('locked Pi worker process smoke', () => { configDir, sessionDir, tools: ['read', 'grep', 'find', 'ls'], - additionalArgs: ['--extension', extension.extensionPath, '--no-session'], - env: extension.env, + additionalArgs: [ + '--extension', extension.extensionPath, + '--extension', probe.extensionPath, + '--no-session', + ], + env: { ...extension.env, MAKELORE_PI_ACTIVE_TOOLS_FILE: probe.resultPath }, sensitiveValues: extension.sensitiveValues, commandTimeoutMs: 5_000, }).start(); @@ -115,6 +160,7 @@ describe('locked Pi worker process smoke', () => { type: 'response', command: 'get_state', success: true, }); expect(worker.stderrDiagnostic).not.toContain('Failed to load extension'); + expect(await readActiveTools(probe.resultPath)).toEqual(['read', 'grep', 'find', 'ls']); await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); } finally { await worker.stop().catch(() => undefined); @@ -122,4 +168,69 @@ describe('locked Pi worker process smoke', () => { await extensionHost.close(); } }, 15_000); + + it.skipIf(!packagedRuntimeRoot)( + 'starts the staged production-closure runtime as an ephemeral read-only child', + async () => { + const requireFromProject = createRequire(resolve('package.json')); + const electronExecutable = requireFromProject('electron') as string; + const packageRoot = realpathSync(packagedRuntimeRoot as string); + const manifest = JSON.parse(await readFile(join(packageRoot, PI_RUNTIME_MANIFEST), 'utf8')) as { + runtime?: { version?: string; cliEntry?: string }; + productionPackages?: string[]; + }; + expect(manifest.runtime).toMatchObject({ + version: PI_RUNTIME_VERSION, + cliEntry: 'dist/cli.js', + }); + expect(manifest.productionPackages?.length).toBeGreaterThan(0); + + const root = await mkdtemp(join(tmpdir(), 'makelore-pi-staged-child-')); + scratchRoots.push(root); + const configDir = join(root, 'config'); + const sessionDir = join(root, 'sessions'); + const cwd = join(root, 'project'); + await Promise.all([mkdir(configDir), mkdir(sessionDir), mkdir(cwd)]); + const extensionHost = new PiManagedExtensionHost(); + const extension = await extensionHost.registerWorker({ + conversationId: 'staged-parent', + generation: 1, + projectId: 'staged-project', + extensionsDir: join(root, 'extensions'), + role: 'child', + runId: 'staged-run', + }); + const probe = await materializeActiveToolsProbe(root); + const worker = new PiWorkerProcess({ + executablePath: electronExecutable, + cliPath: join(packageRoot, 'dist', 'cli.js'), + cwd, + configDir, + sessionDir, + tools: ['read', 'grep', 'find', 'ls'], + additionalArgs: [ + '--extension', extension.extensionPath, + '--extension', probe.extensionPath, + '--no-session', + ], + env: { ...extension.env, MAKELORE_PI_ACTIVE_TOOLS_FILE: probe.resultPath }, + sensitiveValues: extension.sensitiveValues, + commandTimeoutMs: 5_000, + }); + try { + await worker.start(); + await expect(worker.request({ type: 'get_state' })).resolves.toMatchObject({ + type: 'response', command: 'get_state', success: true, + }); + expect(await readActiveTools(probe.resultPath)).toEqual(['read', 'grep', 'find', 'ls']); + expect(worker.stderrDiagnostic).not.toContain('Failed to load extension'); + await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); + } finally { + await worker.stop().catch(() => undefined); + await extension.dispose(); + await extensionHost.close(); + } + }, + 15_000, + ); });