wangxuming/makelore
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: close Pi subagent review gaps

Browse Source
This commit is contained in:
brother7
2026-08-23 13:14:03 +08:00
parent f1c7cd8ad4
commit 3de85d61d7
9 changed files with 234 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
.project-docs/30-worklog/tasks/20260823-pi-child-workers-5c8e2a71.md
View File

@@ -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.

16
electron/coding-runtime/pi/subagent.ts
View File

@@ -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<void> {
while (!signal.aborted
&& this.processBudget.activeCount >= this.processBudget.maxProcesses
&& this.reclaimProcessCapacity) {
if (!await this.reclaimProcessCapacity()) break;
}
}
private markRemaining(
details: SubagentDetailsV1,
from: number,

17
electron/coding-runtime/pi/worker-pool.ts
View File

@@ -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<PiProcessLease> {
acquire(
signal?: AbortSignal,
priority: 'normal' | 'child' = 'normal',
): Promise<PiProcessLease> {
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<PiProcessLease>((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);
}

4
electron/coding-runtime/pi/worker-process.ts
View File

@@ -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',

1
package.json
View File

@@ -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",

43
scripts/run-pi-subagent-packaged-smoke.mjs Normal file
View File

@@ -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 });
}

2
tests/unit/pi-rpc-foundation.test.ts
View File

@@ -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']))

17
tests/unit/pi-subagent.test.ts
View File

@@ -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();
});
});

121
tests/unit/pi-worker-process-real.test.ts
View File

@@ -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<string[]> {
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,
);
});