From f1c7cd8ad4fbf8d81f608779a3523b7e2358d34f Mon Sep 17 00:00:00 2001 From: brother7 <7brother7@gmail.com> Date: Sun, 23 Aug 2026 12:40:22 +0800 Subject: [PATCH] feat: add Pi subagent scheduler and child runtime --- .../20260823-pi-child-workers-5c8e2a71.md | 136 ++++++ .../coding-runtime/conversation-reducer.ts | 15 +- electron/coding-runtime/pi/event-projector.ts | 81 +++- electron/coding-runtime/pi/extension-host.ts | 126 +++++- .../pi/extensions/makelore-runtime.ts | 93 +++- electron/coding-runtime/pi/runtime.ts | 11 + .../coding-runtime/pi/session-projector.ts | 23 + electron/coding-runtime/pi/subagent-child.ts | 227 ++++++++++ electron/coding-runtime/pi/subagent.ts | 428 ++++++++++++++++++ electron/coding-runtime/pi/worker-pool.ts | 9 + electron/coding-runtime/pi/worker-process.ts | 10 +- electron/coding-runtime/subagent-protocol.ts | 82 ++++ tests/unit/pi-event-projector.test.ts | 46 ++ tests/unit/pi-extension-bundle.test.ts | 88 +++- tests/unit/pi-extension-host.test.ts | 152 ++++++- tests/unit/pi-managed-worker-opener.test.ts | 2 +- tests/unit/pi-rpc-foundation.test.ts | 14 + tests/unit/pi-session-projector.test.ts | 51 +++ tests/unit/pi-subagent-child.test.ts | 155 +++++++ tests/unit/pi-subagent.test.ts | 241 ++++++++++ tests/unit/pi-worker-process-real.test.ts | 49 ++ 21 files changed, 1986 insertions(+), 53 deletions(-) create mode 100644 .project-docs/30-worklog/tasks/20260823-pi-child-workers-5c8e2a71.md create mode 100644 electron/coding-runtime/pi/subagent-child.ts create mode 100644 electron/coding-runtime/pi/subagent.ts create mode 100644 electron/coding-runtime/subagent-protocol.ts create mode 100644 tests/unit/pi-subagent-child.test.ts create mode 100644 tests/unit/pi-subagent.test.ts 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 new file mode 100644 index 0000000..bd1acf4 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260823-pi-child-workers-5c8e2a71.md @@ -0,0 +1,136 @@ +# Task: Implement PI-080 child workers and subagent runtime + +## Identity + +- Task ID: 20260823-pi-child-workers-5c8e2a71 +- Mode: Feature +- Branch: codex/20260823-pi-child-workers-5c8e2a71-pi-child-workers +- Worktree: D:\Datas\OthersProjects\makelore-pi-child-workers-5c8e2a71 +- Base commit: b806c78139aa11e338c680d4c3fa92e076901aa2 +- Owner: codex +- Status: Implementation complete; planner review pending + +## Scope + +- Implement `PI-080 — Subagent scheduler and nested protocol` on top of the + cumulative PI-070 baseline `b806c78139aa11e338c680d4c3fa92e076901aa2`. +- Own the Main-private child scheduler/supervisor, managed ephemeral child Pi + opener, parent-child lifecycle registry, global child concurrency limit, and + strict `subagent.v1` validation/projection. +- Extend the single managed Makelore Pi extension and its authenticated Main + bridge only as needed to dispatch `single`, `parallel`, and `chain` child + work and stream stable product details back to the parent tool. +- Add fake scheduler integration coverage, real Pi child process smoke, + process/semaphore leak coverage, bridge identity/abort coverage, and + same-project parent/child write-lease coverage. Renderer/Host product routes + and final nested UI remain owned by later tickets. + +## Intent And Constraints + +- Follow planner task `20260822-pi-runtime-spec-b6e2c9a4`, ticket `PI-080`, + Spec `UX-041` through `UX-044`, `EXT-001` through `EXT-006`, and section 9.4. +- A dispatch accepts exactly one mode, contains at most eight bounded tasks, + and resolves only through Makelore-owned `subagent.v1`; unknown schema or raw + Pi/example details never enter product state or diagnostics. +- Every child owns an independent ephemeral Pi process/context and uses an + enabled project Agent's managed prompt, skills, model, credentials, and one + exact `read-only` or `coding` tool profile. Child sessions are not persisted + as recoverable user Conversations and children cannot recursively dispatch + more subagents. +- Reuse the same application `PiProcessBudget` as parent workers for the locked + parent/child soft cap of eight and enforce one global FIFO maximum of four + running children across all parents. Do not add a competing process counter. +- Reuse PI-070's single project mutation lease. A coding child loads the same + managed extension lease hooks as its parent; read-only children receive no + mutation tools. Do not introduce child worktrees or a second mutation lock. +- Parent abort/crash/recover/dispose, stale generation/run identity, bridge + disconnect, and host shutdown cancel unfinished children and release every + child/process permit. Parallel failures preserve sibling results; chain + failure marks the remaining tasks skipped. +- Keep Provider credentials in the child environment/redaction set only. Real + external Provider validation remains **Explicitly Waived / Accepted Risk** + with `realTurnVerified=false`; macOS x64/arm64 remains deferred to mandatory + PI-150. Neither is a Pass. +- Keep the change surgical: no dual runtime, generic plugin/permission system, + project/user Pi discovery, Renderer wire changes, Host product routes, + OpenCode refactor, or canonical project-memory edits. + +## Outcome + +- Added the Main-private `PiSubagentScheduler` with strict one-to-eight task + validation, single/parallel/chain execution, one FIFO four-child semaphore, + 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. +- 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 + redaction set, selects the exact read-only or coding tools, uses + `--no-session`, and returns bounded public summary/usage data. +- Upgraded the single materialized Makelore extension bundle to v2. Parent + workers expose `ask_user` and `subagent`; child workers expose neither and + 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. +- 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 + project write lease as their parent. +- Added vendor-neutral strict `subagent.v1` projection for live events, durable + session hydration, and reducer validation. Unknown versions render one fixed + unavailable message and never copy their raw payload into product state. +- `PiConversationRuntime` now configures the scheduler/bridge against the + existing pool generation tracker. Final product cutover and nested Renderer + presentation remain outside this ticket. + +## Verification + +- Focused scheduler, managed-child, bridge, bundle, projector, session, process + and contract tests passed, including: 1/4/8 task coverage and ninth-task + rejection; two-parent global child concurrency of four; full-budget idle + 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. +- 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 + Provider. +- All cumulative Pi tests passed: 22 files, 106 tests. +- `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: 202 files, 2209 tests. +- 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 + than marked Pass. macOS x64/arm64 validation was not run and remains deferred + to mandatory PI-150; it is not marked Pass. + +## Follow-ups + +- PI-130 owns the final nested Renderer execution graph and user-facing child + error/abort presentation. +- PI-150 owns final packaged provider-shaped subagent smoke and mandatory + macOS x64/arm64 validation. + +## Promotion Candidates + +- Target canonical document: Pi runtime process/concurrency architecture. + Proposal: record that child budget acquisition must reclaim an idle parent + worker when the shared eight-process budget is full; otherwise the supported + shape of four running parents plus four warm-idle parents can deadlock while + 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. diff --git a/electron/coding-runtime/conversation-reducer.ts b/electron/coding-runtime/conversation-reducer.ts index 3531e0f..c64e4e7 100644 --- a/electron/coding-runtime/conversation-reducer.ts +++ b/electron/coding-runtime/conversation-reducer.ts @@ -14,6 +14,7 @@ import type { KnownToolDetails, PublicWorkerState, } from './contracts'; +import { projectSubagentDetailsV1 } from './subagent-protocol'; export type ConversationInvalidationCode = | 'snapshot-required' @@ -161,19 +162,7 @@ function isKnownToolDetails(value: unknown): value is KnownToolDetails { }); } if (record.schema === 'subagent.v1') { - return isNonEmptyString(record.dispatchId) - && ['single', 'parallel', 'chain'].includes(String(record.mode)) - && Array.isArray(record.tasks) - && record.tasks.every((task) => { - const item = asRecord(task); - return item !== null - && isNonEmptyString(item.taskId) - && isNonEmptyString(item.agentId) - && ['read-only', 'coding'].includes(String(item.toolProfile)) - && ['queued', 'running', 'complete', 'error', 'aborted', 'skipped'] - .includes(String(item.status)) - && (item.usage === undefined || isUsage(item.usage)); - }); + return projectSubagentDetailsV1(record) !== undefined; } return false; } diff --git a/electron/coding-runtime/pi/event-projector.ts b/electron/coding-runtime/pi/event-projector.ts index fee40d8..60490f2 100644 --- a/electron/coding-runtime/pi/event-projector.ts +++ b/electron/coding-runtime/pi/event-projector.ts @@ -7,6 +7,7 @@ import type { PublicUsage, } from '../contracts'; import type { PiRpcEvent } from './rpc-client'; +import { subagentDetailsOfResult } from '../subagent-protocol'; export interface PiEventProjectorOptions { createId(): string; @@ -89,6 +90,49 @@ function outputBlocks(toolId: string, value: unknown): ConversationContentBlock[ }); } +function projectToolResult( + snapshot: ConversationSnapshot, + tool: ConversationToolNode, + status: ConversationToolNode['status'], + value: unknown, +): ConversationPatch[] { + const details = subagentDetailsOfResult(value); + const result = asRecord(value); + const unknownSubagentDetails = tool.toolName === 'subagent' + && result?.details !== undefined + && !details; + const output = unknownSubagentDetails + ? [{ + kind: 'text' as const, + id: `${tool.id}:output:unavailable`, + text: 'Subagent details are unavailable for this version.', + status: 'complete' as const, + }] + : outputBlocks(tool.id, value); + const patches: ConversationPatch[] = [{ + op: 'tool.upsert', + node: { + ...tool, + status, + output, + ...(details ? { details } : {}), + }, + }]; + const runId = snapshot.run.runId; + if (details && runId) { + patches.push({ + op: 'subagent.upsert', + node: { + kind: 'subagent', + id: `subagent:${details.dispatchId}`, + runId, + details, + }, + }); + } + return patches; +} + function currentTool(snapshot: ConversationSnapshot, toolCallId: string): ConversationToolNode | undefined { return snapshot.nodes.find( (node): node is ConversationToolNode => node.kind === 'tool' @@ -613,28 +657,19 @@ export class PiEventProjector { if (typeof event.toolCallId !== 'string') return []; const tool = currentTool(snapshot, event.toolCallId); if (!tool) return []; - return [{ - op: 'tool.upsert', - node: { - ...tool, - status: 'running', - output: outputBlocks(tool.id, event.partialResult), - }, - }]; + return projectToolResult(snapshot, tool, 'running', event.partialResult); } if (event.type === 'tool_execution_end') { if (typeof event.toolCallId !== 'string') return []; const tool = currentTool(snapshot, event.toolCallId); if (!tool) return []; - return [{ - op: 'tool.upsert', - node: { - ...tool, - status: event.isError === true ? 'error' : 'complete', - output: outputBlocks(tool.id, event.result), - }, - }]; + return projectToolResult( + snapshot, + tool, + event.isError === true ? 'error' : 'complete', + event.result, + ); } if (event.type === 'message_end') { @@ -642,14 +677,12 @@ export class PiEventProjector { if (message?.role === 'toolResult' && typeof message.toolCallId === 'string') { const tool = currentTool(snapshot, message.toolCallId); if (!tool) return []; - return [{ - op: 'tool.upsert', - node: { - ...tool, - status: message.isError === true ? 'error' : 'complete', - output: outputBlocks(tool.id, message), - }, - }]; + return projectToolResult( + snapshot, + tool, + message.isError === true ? 'error' : 'complete', + message, + ); } if (message?.role !== 'assistant') return []; const current = currentAssistant(snapshot); diff --git a/electron/coding-runtime/pi/extension-host.ts b/electron/coding-runtime/pi/extension-host.ts index 83a713f..b354b1f 100644 --- a/electron/coding-runtime/pi/extension-host.ts +++ b/electron/coding-runtime/pi/extension-host.ts @@ -8,6 +8,11 @@ import { PiProjectWriteLeaseCoordinator, type PiProjectWriteLease, } from './write-lease'; +import type { PiGenerationResourceInput } from './worker-pool'; +import { + parsePiSubagentDispatchRequest, + type PiSubagentScheduler, +} from './subagent'; const MAX_REQUEST_BYTES = 64 * 1024; @@ -16,6 +21,7 @@ interface WorkerRegistrationRecord { conversationId: string; generation: number; projectId: string; + role: 'parent' | 'child'; contextFile: string; runId: string | null; leases: Map; @@ -34,9 +40,11 @@ export interface RegisterPiExtensionWorkerInput { generation: number; projectId: string; extensionsDir: string; + role?: 'parent' | 'child'; + runId?: string; } -interface BridgeRequest { +interface LeaseBridgeRequest { action: 'lease.acquire' | 'lease.release'; conversationId: string; workerGeneration: number; @@ -45,17 +53,35 @@ interface BridgeRequest { leaseId?: string; } +interface SubagentBridgeRequest { + action: 'subagent.dispatch'; + conversationId: string; + workerGeneration: number; + runId: string; + resourceId: string; + request: unknown; +} + +type BridgeRequest = LeaseBridgeRequest | SubagentBridgeRequest; + +export interface PiExtensionSubagentBridge { + scheduler: PiSubagentScheduler; + trackGenerationResource?(input: PiGenerationResourceInput): () => void; +} + function recordValue(value: unknown): value is Record { return value !== null && typeof value === 'object' && !Array.isArray(value); } function bridgeRequest(value: unknown): value is BridgeRequest { if (!recordValue(value)) return false; - return (value.action === 'lease.acquire' || value.action === 'lease.release') - && typeof value.conversationId === 'string' + const common = typeof value.conversationId === 'string' && Number.isSafeInteger(value.workerGeneration) && typeof value.runId === 'string' - && typeof value.resourceId === 'string' + && typeof value.resourceId === 'string'; + if (!common) return false; + if (value.action === 'subagent.dispatch') return 'request' in value; + return (value.action === 'lease.acquire' || value.action === 'lease.release') && (value.leaseId === undefined || typeof value.leaseId === 'string'); } @@ -64,6 +90,7 @@ export class PiManagedExtensionHost { private readonly registrations = new Map(); private readonly runBindings = new Map(); private readonly requestFlights = new Set>(); + private subagentBridge: PiExtensionSubagentBridge | undefined; private server: Server | null = null; private bridgeUrl: string | null = null; private startFlight: Promise | null = null; @@ -73,6 +100,10 @@ export class PiManagedExtensionHost { this.leases = leases; } + configureSubagents(bridge: PiExtensionSubagentBridge): void { + this.subagentBridge = bridge; + } + async registerWorker(input: RegisterPiExtensionWorkerInput): Promise { if (!Number.isSafeInteger(input.generation) || input.generation <= 0) { throw new Error('Worker generation must be a positive safe integer'); @@ -80,6 +111,10 @@ export class PiManagedExtensionHost { const bridgeUrl = await this.start(); await mkdir(input.extensionsDir, { recursive: true }); const extensionPath = await materializeMakelorePiExtension(input.extensionsDir); + const role = input.role ?? 'parent'; + if (role === 'child' && !input.runId?.trim()) { + throw new Error('Child extension registration requires a parent run id'); + } const token = randomBytes(32).toString('base64url'); const contextFile = path.join(input.extensionsDir, `worker-${randomUUID()}.json`); const record: WorkerRegistrationRecord = { @@ -87,8 +122,11 @@ export class PiManagedExtensionHost { conversationId: input.conversationId, generation: input.generation, projectId: input.projectId, + role, contextFile, - runId: this.runBindings.get(input.conversationId) ?? null, + runId: role === 'child' + ? input.runId as string + : this.runBindings.get(input.conversationId) ?? null, leases: new Map(), waiters: new Map(), }; @@ -101,6 +139,7 @@ export class PiManagedExtensionHost { MAKELORE_PI_BRIDGE_URL: bridgeUrl, MAKELORE_PI_WORKER_TOKEN: token, MAKELORE_PI_CONTEXT_FILE: contextFile, + MAKELORE_PI_WORKER_ROLE: role, }, sensitiveValues: [token], dispose: async () => { @@ -206,6 +245,10 @@ export class PiManagedExtensionHost { this.respond(response, 400, { error: 'Bridge resource id is required' }); return; } + if (value.action === 'subagent.dispatch') { + await this.dispatchSubagents(request, response, record, value); + return; + } if (value.action === 'lease.release') { const lease = record.leases.get(value.resourceId); if (!lease || !value.leaseId || lease.id !== value.leaseId) { @@ -251,6 +294,75 @@ export class PiManagedExtensionHost { } } + private async dispatchSubagents( + request: IncomingMessage, + response: ServerResponse, + record: WorkerRegistrationRecord, + value: SubagentBridgeRequest, + ): Promise { + if (record.role !== 'parent') { + this.respond(response, 403, { error: 'Child workers cannot dispatch subagents' }); + return; + } + if (!this.subagentBridge) { + this.respond(response, 503, { error: 'Subagent scheduler is unavailable' }); + return; + } + try { + parsePiSubagentDispatchRequest(value.request); + } catch { + this.respond(response, 400, { error: 'Subagent dispatch request is invalid' }); + return; + } + if (record.waiters.has(value.resourceId) || record.leases.has(value.resourceId)) { + this.respond(response, 409, { error: 'Bridge resource already exists' }); + return; + } + const controller = new AbortController(); + record.waiters.set(value.resourceId, controller); + const cancel = () => controller.abort(); + request.once('aborted', cancel); + response.once('close', cancel); + let untrack = () => undefined; + try { + untrack = this.subagentBridge.trackGenerationResource?.({ + conversationId: record.conversationId, + kind: 'child', + id: value.resourceId, + cancel, + }) ?? untrack; + response.writeHead(200, { + 'content-type': 'application/x-ndjson; charset=utf-8', + ...(this.closing ? { connection: 'close' } : {}), + }); + const result = await this.subagentBridge.scheduler.dispatch({ + conversationId: record.conversationId, + workerGeneration: record.generation, + runId: value.runId, + projectId: record.projectId, + request: value.request, + }, { + signal: controller.signal, + onUpdate: (details) => { + if (!response.writableEnded && !response.destroyed) { + response.write(`${JSON.stringify({ details })}\n`); + } + }, + }); + if (!response.writableEnded && !response.destroyed) { + response.end(`${JSON.stringify({ details: result.details, done: true })}\n`); + } + } catch { + if (!response.headersSent) this.respond(response, 400, { error: 'Subagent dispatch failed' }); + else if (!response.writableEnded && !response.destroyed) response.end(); + } finally { + request.removeListener('aborted', cancel); + response.removeListener('close', cancel); + record.waiters.delete(value.resourceId); + untrack(); + } + } + private readBody(request: IncomingMessage): Promise { return new Promise((resolve, reject) => { const chunks: Buffer[] = []; @@ -286,7 +398,8 @@ export class PiManagedExtensionHost { private findWorker(conversationId: string, generation: number): WorkerRegistrationRecord | undefined { return [...this.registrations.values()].find((record) => ( - record.conversationId === conversationId && record.generation === generation + record.role === 'parent' + && record.conversationId === conversationId && record.generation === generation )); } @@ -294,6 +407,7 @@ export class PiManagedExtensionHost { await atomicWriteJson(record.contextFile, { conversationId: record.conversationId, workerGeneration: record.generation, + role: record.role, ...(record.runId ? { runId: record.runId } : {}), }); } diff --git a/electron/coding-runtime/pi/extensions/makelore-runtime.ts b/electron/coding-runtime/pi/extensions/makelore-runtime.ts index 488aa0e..48fea0c 100644 --- a/electron/coding-runtime/pi/extensions/makelore-runtime.ts +++ b/electron/coding-runtime/pi/extensions/makelore-runtime.ts @@ -1,13 +1,14 @@ import path from 'node:path'; import { atomicWriteText } from '../../../coding-projects/atomic-json'; -export const MAKELORE_PI_EXTENSION_VERSION = 1; +export const MAKELORE_PI_EXTENSION_VERSION = 2; export const MAKELORE_PI_EXTENSION_FILENAME = `makelore-runtime-v${MAKELORE_PI_EXTENSION_VERSION}.mjs`; const BUNDLE_SOURCE = String.raw` import { readFile } from 'node:fs/promises'; const MUTATION_TOOLS = new Set(['bash', 'edit', 'write']); +const WORKER_ROLE = process.env.MAKELORE_PI_WORKER_ROLE || 'parent'; const leases = new Map(); async function runtimeContext() { @@ -32,6 +33,48 @@ async function bridge(action, body, signal) { return result; } +async function bridgeStream(action, body, signal, onUpdate) { + const context = await runtimeContext(); + const response = await fetch(process.env.MAKELORE_PI_BRIDGE_URL, { + method: 'POST', + headers: { + authorization: 'Bearer ' + process.env.MAKELORE_PI_WORKER_TOKEN, + 'content-type': 'application/json', + }, + body: JSON.stringify({ ...context, action, ...body }), + signal, + }); + if (!response.ok) { + const result = await response.json().catch(() => ({})); + throw new Error(result.error || 'Makelore runtime bridge rejected the request'); + } + if (!response.body) throw new Error('Makelore runtime bridge returned no stream'); + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffered = ''; + let details; + while (true) { + const { value, done } = await reader.read(); + buffered += decoder.decode(value || new Uint8Array(), { stream: !done }); + let newline = buffered.indexOf('\n'); + while (newline >= 0) { + const line = buffered.slice(0, newline); + buffered = buffered.slice(newline + 1); + if (line) { + const item = JSON.parse(line); + if (item.details?.schema === 'subagent.v1') { + details = item.details; + onUpdate?.({ content: [], details }); + } + } + newline = buffered.indexOf('\n'); + } + if (done) break; + } + if (!details) throw new Error('Makelore subagent stream returned no details'); + return details; +} + async function releaseLease(toolCallId) { const leaseId = leases.get(toolCallId); if (!leaseId) return; @@ -44,7 +87,7 @@ async function releaseAll() { } export default function makeloreRuntime(pi) { - pi.registerTool({ + if (WORKER_ROLE === 'parent') pi.registerTool({ name: 'ask_user', label: 'Ask user', description: 'Ask the user for a selection, confirmation, short input, or editor text.', @@ -77,6 +120,52 @@ export default function makeloreRuntime(pi) { }, }); + if (WORKER_ROLE === 'parent') pi.registerTool({ + name: 'subagent', + label: 'Subagent', + description: 'Dispatch one or more managed project Agents in single, parallel, or chain mode.', + parameters: { + type: 'object', + additionalProperties: false, + required: ['mode', 'tasks'], + properties: { + mode: { type: 'string', enum: ['single', 'parallel', 'chain'] }, + tasks: { + type: 'array', minItems: 1, maxItems: 8, + items: { + type: 'object', + additionalProperties: false, + required: ['agentId', 'task', 'toolProfile'], + properties: { + agentId: { type: 'string' }, + task: { type: 'string' }, + toolProfile: { type: 'string', enum: ['read-only', 'coding'] }, + }, + }, + }, + }, + }, + async execute(toolCallId, params, signal, onUpdate) { + const details = await bridgeStream( + 'subagent.dispatch', + { resourceId: toolCallId, request: params }, + signal, + onUpdate, + ); + const complete = details.tasks.filter((task) => task.status === 'complete').length; + const failed = details.tasks.length - complete; + return { + content: [{ + type: 'text', + text: failed === 0 + ? complete + ' subagent task(s) completed' + : complete + ' completed; ' + failed + ' did not complete', + }], + details, + }; + }, + }); + pi.on('tool_call', async (event, ctx) => { if (!MUTATION_TOOLS.has(event.toolName)) return; ctx.ui.setStatus('makelore.write-lease', '等待项目写入'); diff --git a/electron/coding-runtime/pi/runtime.ts b/electron/coding-runtime/pi/runtime.ts index 8573870..578a80d 100644 --- a/electron/coding-runtime/pi/runtime.ts +++ b/electron/coding-runtime/pi/runtime.ts @@ -75,6 +75,7 @@ import { projectPiSessionSnapshot, } from './session-projector'; import { PiManagedExtensionHost } from './extension-host'; +import type { PiSubagentScheduler } from './subagent'; import { PiInteractionStore, type PiInteractionResponse, @@ -98,6 +99,7 @@ export interface PiConversationRuntimeOptions { isAuthenticationError?(error: unknown): boolean; refreshCredential?(accountId: string): Promise; extensionHost?: PiManagedExtensionHost; + subagentScheduler?: PiSubagentScheduler; getDraftRevision?(conversationId: string): number; knownExtensionWidgetKeys?: readonly string[]; onExtensionUiProjection?(projection: PiExtensionUiProjection): void; @@ -480,6 +482,15 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.isAuthenticationError = options.isAuthenticationError; this.refreshCredential = options.refreshCredential; this.extensionHost = options.extensionHost; + if (options.subagentScheduler && !this.extensionHost) { + throw new Error('Subagent scheduler requires the managed extension host'); + } + if (options.subagentScheduler) { + this.extensionHost?.configureSubagents({ + scheduler: options.subagentScheduler, + trackGenerationResource: (input) => this.pool.trackGenerationResource(input), + }); + } this.interactions = new PiInteractionStore(this.pool, (interaction) => { this.emit(interaction.conversationId, { op: 'interaction.upsert', interaction }, interaction.runId); }); diff --git a/electron/coding-runtime/pi/session-projector.ts b/electron/coding-runtime/pi/session-projector.ts index f71143a..fc5b2ae 100644 --- a/electron/coding-runtime/pi/session-projector.ts +++ b/electron/coding-runtime/pi/session-projector.ts @@ -12,6 +12,7 @@ import type { PiImageProjectionInput, PiProjectedAttachment, } from './event-projector'; +import { subagentDetailsOfResult } from '../subagent-protocol'; export interface PiSessionSnapshotInput { snapshot: ConversationSnapshot; @@ -184,6 +185,7 @@ async function projectEntries( ): Promise { const nodes: ConversationNode[] = []; const tools = new Map(); + const subagentIds = new Set(); for (const entry of path) { if (entry.type === 'compaction') { nodes.push({ @@ -211,6 +213,27 @@ async function projectEntries( 'output', ); tool.status = message.isError === true ? 'error' : 'complete'; + const details = subagentDetailsOfResult(message); + if (details) { + tool.details = details; + const id = `subagent:${details.dispatchId}`; + if (!subagentIds.has(id)) { + subagentIds.add(id); + nodes.push({ + kind: 'subagent', + id, + runId: input.snapshot.run.runId ?? `session:${entry.id}`, + details, + }); + } + } else if (tool.toolName === 'subagent' && message.details !== undefined) { + tool.output = [{ + kind: 'text', + id: `${tool.id}:output:unavailable`, + text: 'Subagent details are unavailable for this version.', + status: 'complete', + }]; + } continue; } if (message.role !== 'user' && message.role !== 'assistant') continue; diff --git a/electron/coding-runtime/pi/subagent-child.ts b/electron/coding-runtime/pi/subagent-child.ts new file mode 100644 index 0000000..8091a4f --- /dev/null +++ b/electron/coding-runtime/pi/subagent-child.ts @@ -0,0 +1,227 @@ +import type { ModelSummary, ProviderAccount } from '../../shared/providers/types'; +import { readCodingProjectConfigV2 } from '../../coding-projects/project-config'; +import type { CodingProjectStore } from '../../coding-projects/project-store'; +import type { PublicUsage } from '../contracts'; +import type { PiManagedInputRevision } from './managed-input-revision'; +import { + buildPiProviderCatalog, + buildPiWorkerCredentialProjection, + selectPiProviderModel, + writePiProviderCatalog, +} from './provider-config'; +import { + buildPiManagedInputArgs, + ensurePiManagedPaths, + materializePiAgentResources, +} from './resource-loader'; +import type { + PiRpcCommand, + PiRpcEvent, + PiRpcRequestOptions, + PiRpcResponse, +} from './rpc-client'; +import type { + PiSubagentChild, + PiSubagentChildOpenInput, + PiSubagentChildResult, +} from './subagent'; +import { PiSubagentChildError } from './subagent'; +import { PiWorkerProcess, type PiWorkerProcessOptions, type PiWorkerStopResult } from './worker-process'; +import type { PiProcessError } from './process-errors'; +import type { PiManagedExtensionHost } from './extension-host'; + +const READ_ONLY_TOOLS = ['read', 'grep', 'find', 'ls'] as const; +const CODING_TOOLS = ['read', 'bash', 'edit', 'write', 'grep', 'find', 'ls'] as const; + +export interface PiSubagentProcessAdapter { + start(): Promise; + request( + command: PiRpcCommand, + options?: PiRpcRequestOptions, + ): Promise>; + subscribe(listener: (event: PiRpcEvent) => void): () => void; + subscribeInvalidation(listener: (error: PiProcessError) => void): () => void; + stop(): Promise; +} + +export interface PiManagedSubagentChildOpenerOptions { + projectStore: CodingProjectStore; + executablePath: string; + cliPath: string; + userDataDir: string; + bundledSkillsDir: string; + extensionHost: PiManagedExtensionHost; + loadProviderInput(): Promise<{ accounts: ProviderAccount[]; modelSummaries: ModelSummary[] }>; + resolveCredential(account: ProviderAccount): Promise; + getRevision(): PiManagedInputRevision; + getLocalProxyCredential?(): Promise; + createProcess?(options: PiWorkerProcessOptions): PiSubagentProcessAdapter; +} + +function recordValue(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null; +} + +function usageOf(value: unknown): PublicUsage | undefined { + const usage = recordValue(value); + if (!usage || typeof usage.input !== 'number' || typeof usage.output !== 'number') return undefined; + return { + inputTokens: usage.input, + outputTokens: usage.output, + ...(typeof usage.cacheRead === 'number' ? { cacheReadTokens: usage.cacheRead } : {}), + ...(typeof usage.cacheWrite === 'number' ? { cacheWriteTokens: usage.cacheWrite } : {}), + }; +} + +class ManagedPiSubagentChild implements PiSubagentChild { + private ran = false; + private stopped = false; + + constructor( + readonly id: string, + private readonly process: PiSubagentProcessAdapter, + private readonly disposeExtension: () => Promise, + ) {} + + async run(prompt: string, signal: AbortSignal): Promise { + if (this.ran) throw new PiSubagentChildError('SUBAGENT_ALREADY_RAN'); + this.ran = true; + if (signal.aborted) throw new PiSubagentChildError('SUBAGENT_ABORTED'); + let usage: PublicUsage | undefined; + let settle!: () => void; + let fail!: (error: PiSubagentChildError) => void; + const settled = new Promise((resolve, reject) => { + settle = resolve; + fail = reject; + }); + const unsubscribeEvents = this.process.subscribe((event) => { + if (event.type === 'message_end') { + const message = recordValue(event.message); + if (message?.role === 'assistant') usage = usageOf(message.usage) ?? usage; + } + if (event.type === 'agent_settled') settle(); + }); + const unsubscribeInvalidation = this.process.subscribeInvalidation(() => { + fail(new PiSubagentChildError('SUBAGENT_CHILD_CRASHED')); + }); + const abort = () => { + void this.process.request({ type: 'abort' }).catch(() => undefined); + fail(new PiSubagentChildError('SUBAGENT_ABORTED')); + }; + signal.addEventListener('abort', abort, { once: true }); + try { + const accepted = await this.process.request({ type: 'prompt', message: prompt }); + if (!accepted.success) throw new PiSubagentChildError('SUBAGENT_PROMPT_REJECTED'); + await settled; + if (signal.aborted) throw new PiSubagentChildError('SUBAGENT_ABORTED'); + const response = await this.process.request<{ text?: unknown }>({ + type: 'get_last_assistant_text', + }, { retry: 'read-only-once' }); + if (!response.success) throw new PiSubagentChildError('SUBAGENT_RESULT_UNAVAILABLE'); + const summary = typeof response.data?.text === 'string' + ? response.data.text + : ''; + return { + summary, + ...(usage ? { usage } : {}), + }; + } finally { + signal.removeEventListener('abort', abort); + unsubscribeEvents(); + unsubscribeInvalidation(); + } + } + + async stop(): Promise { + if (this.stopped) return; + this.stopped = true; + try { + await this.process.stop(); + } finally { + await this.disposeExtension(); + } + } +} + +export function createPiManagedSubagentChildOpener( + options: PiManagedSubagentChildOpenerOptions, +): (input: PiSubagentChildOpenInput) => Promise { + const createProcess = options.createProcess ?? ((processOptions) => new PiWorkerProcess(processOptions)); + return async (input) => { + const project = (await options.projectStore.listProjects()) + .find((candidate) => candidate.id === input.projectId); + if (!project) throw new PiSubagentChildError('SUBAGENT_PROJECT_UNAVAILABLE'); + const configRead = await readCodingProjectConfigV2(project.path); + if (configRead.status !== 'valid') { + throw new PiSubagentChildError('SUBAGENT_PROJECT_CONFIG_UNAVAILABLE'); + } + const agent = configRead.config.agents.find((candidate) => ( + candidate.id === input.agentId && candidate.enabled && candidate.archivedAt === null + )); + if (!agent) throw new PiSubagentChildError('SUBAGENT_AGENT_UNAVAILABLE'); + if (!agent.model || agent.modelResolution !== 'resolved') { + throw new PiSubagentChildError('SUBAGENT_MODEL_REQUIRED'); + } + + const providerInput = await options.loadProviderInput(); + const catalog = buildPiProviderCatalog(providerInput); + const selection = selectPiProviderModel(catalog, agent.model); + const account = providerInput.accounts.find((candidate) => candidate.id === selection.accountId); + const descriptor = catalog.descriptors.find((candidate) => candidate.accountId === selection.accountId); + if (!account || !descriptor) throw new PiSubagentChildError('SUBAGENT_MODEL_UNAVAILABLE'); + const managedPaths = await ensurePiManagedPaths(options.userDataDir); + await writePiProviderCatalog(managedPaths.modelsFile, catalog, agent.model); + const resources = await materializePiAgentResources({ + userDataDir: options.userDataDir, + projectId: input.projectId, + agentId: agent.id, + prompt: agent.prompt, + skillIds: agent.skillIds, + bundledSkillsDir: options.bundledSkillsDir, + revision: options.getRevision(), + }); + const credential = await buildPiWorkerCredentialProjection({ + account, + descriptor, + resolveCredential: options.resolveCredential, + ...(options.getLocalProxyCredential + ? { localProxyCredential: await options.getLocalProxyCredential() } + : {}), + }); + const extension = await options.extensionHost.registerWorker({ + conversationId: input.conversationId, + generation: input.workerGeneration, + projectId: input.projectId, + extensionsDir: managedPaths.extensionsDir, + role: 'child', + runId: input.runId, + }); + const process = createProcess({ + executablePath: options.executablePath, + cliPath: options.cliPath, + cwd: project.path, + configDir: resources.paths.configDir, + sessionDir: resources.projectSessionsDir, + tools: input.toolProfile === 'coding' ? CODING_TOOLS : READ_ONLY_TOOLS, + additionalArgs: [ + ...buildPiManagedInputArgs(selection, resources), + '--extension', extension.extensionPath, + '--no-session', + ], + env: { ...credential.env, ...extension.env }, + sensitiveValues: [...credential.sensitiveValues, ...extension.sensitiveValues], + }); + try { + await process.start(); + const ready = await process.request({ type: 'get_state' }, { retry: 'read-only-once' }); + if (!ready.success) throw new PiSubagentChildError('SUBAGENT_CHILD_START_FAILED'); + return new ManagedPiSubagentChild(input.taskId, process, extension.dispose); + } catch (error) { + await process.stop().catch(() => undefined); + await extension.dispose(); + throw error; + } + }; +} diff --git a/electron/coding-runtime/pi/subagent.ts b/electron/coding-runtime/pi/subagent.ts new file mode 100644 index 0000000..f7456e0 --- /dev/null +++ b/electron/coding-runtime/pi/subagent.ts @@ -0,0 +1,428 @@ +import { randomUUID } from 'node:crypto'; + +import type { PublicUsage, SubagentDetailsV1 } from '../contracts'; +import { PiProcessBudget, type PiProcessLease } from './worker-pool'; + +export type PiSubagentMode = SubagentDetailsV1['mode']; +export type PiSubagentToolProfile = SubagentDetailsV1['tasks'][number]['toolProfile']; + +export interface PiSubagentTaskRequest { + agentId: string; + task: string; + toolProfile: PiSubagentToolProfile; +} + +export interface PiSubagentDispatchRequest { + mode: PiSubagentMode; + tasks: PiSubagentTaskRequest[]; +} + +export interface PiSubagentParentIdentity { + conversationId: string; + workerGeneration: number; + runId: string; + projectId: string; +} + +export interface PiSubagentChildOpenInput extends PiSubagentParentIdentity { + dispatchId: string; + taskId: string; + agentId: string; + toolProfile: PiSubagentToolProfile; +} + +export interface PiSubagentChildResult { + summary: string; + usage?: PublicUsage; +} + +export interface PiSubagentChild { + readonly id: string; + run( + prompt: string, + signal: AbortSignal, + onProgress?: (summary: string) => void, + ): Promise; + stop(): Promise; +} + +export interface PiSubagentDispatchResult { + details: SubagentDetailsV1; +} + +export interface PiSubagentDispatchOptions { + signal?: AbortSignal; + onUpdate?: (details: SubagentDetailsV1) => void; +} + +export interface PiSubagentSchedulerOptions { + openChild(input: PiSubagentChildOpenInput): Promise; + processBudget: PiProcessBudget; + reclaimProcessCapacity?(): Promise; + createId?: (kind: 'dispatch' | 'task') => string; +} + +interface DispatchRecord { + identity: PiSubagentParentIdentity; + controller: AbortController; + children: Set; + flight: Promise; +} + +interface SemaphoreWaiter { + signal: AbortSignal; + resolve(release: () => void): void; + reject(error: Error): void; + abort(): void; +} + +const MAX_TASKS_PER_DISPATCH = 8; +const MAX_AGENT_ID_LENGTH = 128; +const MAX_TASK_LENGTH = 6_000; +const MAX_SUMMARY_LENGTH = 4_000; + +export class PiSubagentChildError extends Error { + constructor(readonly code: string) { + super(code); + this.name = 'PiSubagentChildError'; + } +} + +class FifoSemaphore { + private readonly waiters: SemaphoreWaiter[] = []; + private active = 0; + + constructor(readonly maximum: number) { + if (!Number.isSafeInteger(maximum) || maximum <= 0) { + throw new Error('Subagent concurrency must be a positive safe integer'); + } + } + + acquire(signal: AbortSignal): Promise<() => void> { + if (signal.aborted) return Promise.reject(new PiSubagentChildError('SUBAGENT_ABORTED')); + if (this.active < this.maximum) return Promise.resolve(this.issuePermit()); + return new Promise<() => void>((resolve, reject) => { + const waiter: SemaphoreWaiter = { + signal, + resolve, + reject, + abort: () => { + const index = this.waiters.indexOf(waiter); + if (index >= 0) this.waiters.splice(index, 1); + reject(new PiSubagentChildError('SUBAGENT_ABORTED')); + }, + }; + signal.addEventListener('abort', waiter.abort, { once: true }); + this.waiters.push(waiter); + }); + } + + private issuePermit(): () => void { + this.active += 1; + let released = false; + return () => { + if (released) return; + released = true; + this.active -= 1; + this.advance(); + }; + } + + private advance(): void { + while (this.active < this.maximum && this.waiters.length > 0) { + const waiter = this.waiters.shift() as SemaphoreWaiter; + waiter.signal.removeEventListener('abort', waiter.abort); + if (waiter.signal.aborted) { + waiter.reject(new PiSubagentChildError('SUBAGENT_ABORTED')); + continue; + } + waiter.resolve(this.issuePermit()); + } + } +} + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null; +} + +function boundedText(value: unknown, label: string, maximum: number): string { + if (typeof value !== 'string') throw new Error(`${label} must be a string`); + const normalized = value.trim(); + if (!normalized) throw new Error(`${label} must not be empty`); + if (normalized.length > maximum) throw new Error(`${label} is too long`); + return normalized; +} + +export function parsePiSubagentDispatchRequest(value: unknown): PiSubagentDispatchRequest { + const record = asRecord(value); + if (!record || !['single', 'parallel', 'chain'].includes(String(record.mode))) { + throw new Error('Subagent dispatch mode is invalid'); + } + if (!Array.isArray(record.tasks) || record.tasks.length === 0) { + throw new Error('Subagent dispatch requires at least one task'); + } + if (record.tasks.length > MAX_TASKS_PER_DISPATCH) { + throw new Error('Subagent dispatch accepts at most 8 tasks'); + } + if (record.mode === 'single' && record.tasks.length !== 1) { + throw new Error('Single subagent dispatch requires exactly one task'); + } + const tasks = record.tasks.map((candidate, index) => { + const task = asRecord(candidate); + if (!task) throw new Error(`Subagent task ${index + 1} is invalid`); + if (task.toolProfile !== 'read-only' && task.toolProfile !== 'coding') { + throw new Error(`Subagent task ${index + 1} tool profile is invalid`); + } + return { + agentId: boundedText(task.agentId, `Subagent task ${index + 1} agent id`, MAX_AGENT_ID_LENGTH), + task: boundedText(task.task, `Subagent task ${index + 1} prompt`, MAX_TASK_LENGTH), + toolProfile: task.toolProfile, + }; + }); + return { mode: record.mode as PiSubagentMode, tasks }; +} + +function safeSummary(value: string): string { + return value.length <= MAX_SUMMARY_LENGTH ? value : `${value.slice(0, MAX_SUMMARY_LENGTH - 1)}…`; +} + +function safeUsage(value: PublicUsage | undefined): PublicUsage | undefined { + if (!value) return undefined; + const fields = [ + value.inputTokens, + value.outputTokens, + value.cacheReadTokens, + value.cacheWriteTokens, + ].filter((field) => field !== undefined); + if (fields.some((field) => !Number.isFinite(field) || field < 0)) return undefined; + return structuredClone(value); +} + +function publicErrorCode(error: unknown, aborted: boolean): string { + if (aborted) return 'SUBAGENT_ABORTED'; + if (error instanceof PiSubagentChildError && /^[A-Z][A-Z0-9_]{0,63}$/.test(error.code)) { + return error.code; + } + return 'SUBAGENT_CHILD_FAILED'; +} + +function parentKey(identity: PiSubagentParentIdentity): string { + return `${identity.conversationId}:${identity.workerGeneration}:${identity.runId}`; +} + +export class PiSubagentScheduler { + private readonly openChild: PiSubagentSchedulerOptions['openChild']; + private readonly processBudget: PiProcessBudget; + private readonly childPermits: FifoSemaphore; + private readonly reclaimProcessCapacity: (() => Promise) | undefined; + private readonly createId: NonNullable; + private readonly dispatches = new Map(); + private readonly parentDispatches = new Map>(); + private closing = false; + + constructor(options: PiSubagentSchedulerOptions) { + this.openChild = options.openChild; + this.processBudget = options.processBudget; + this.childPermits = new FifoSemaphore(4); + this.reclaimProcessCapacity = options.reclaimProcessCapacity; + this.createId = options.createId ?? ((kind) => `${kind}-${randomUUID()}`); + } + + dispatch( + input: PiSubagentParentIdentity & { request: unknown }, + options: PiSubagentDispatchOptions = {}, + ): Promise { + if (this.closing) return Promise.reject(new Error('Subagent scheduler is shutting down')); + const request = parsePiSubagentDispatchRequest(input.request); + const dispatchId = this.createId('dispatch'); + if (this.dispatches.has(dispatchId)) { + return Promise.reject(new Error(`Duplicate subagent dispatch id: ${dispatchId}`)); + } + const identity: PiSubagentParentIdentity = { + conversationId: input.conversationId, + workerGeneration: input.workerGeneration, + runId: input.runId, + projectId: input.projectId, + }; + const controller = new AbortController(); + const externalAbort = () => controller.abort(); + options.signal?.addEventListener('abort', externalAbort, { once: true }); + if (options.signal?.aborted) controller.abort(); + const tasks: SubagentDetailsV1['tasks'] = request.tasks.map((task) => ({ + taskId: this.createId('task'), + agentId: task.agentId, + toolProfile: task.toolProfile, + status: 'queued', + })); + const details: SubagentDetailsV1 = { + schema: 'subagent.v1', + dispatchId, + mode: request.mode, + tasks, + }; + const record = { + identity, + controller, + children: new Set(), + } as DispatchRecord; + const flight = this.runDispatch(record, request, details, options.onUpdate) + .finally(() => { + options.signal?.removeEventListener('abort', externalAbort); + this.dispatches.delete(dispatchId); + const key = parentKey(identity); + const ids = this.parentDispatches.get(key); + ids?.delete(dispatchId); + if (ids?.size === 0) this.parentDispatches.delete(key); + }); + record.flight = flight; + this.dispatches.set(dispatchId, record); + const key = parentKey(identity); + const ids = this.parentDispatches.get(key) ?? new Set(); + ids.add(dispatchId); + this.parentDispatches.set(key, ids); + this.emit(details, options.onUpdate); + return flight; + } + + abortParent(identity: PiSubagentParentIdentity): void { + for (const dispatchId of this.parentDispatches.get(parentKey(identity)) ?? []) { + this.dispatches.get(dispatchId)?.controller.abort(); + } + } + + async close(): Promise { + if (this.closing) { + await Promise.allSettled([...this.dispatches.values()].map((record) => record.flight)); + return; + } + this.closing = true; + for (const record of this.dispatches.values()) record.controller.abort(); + await Promise.allSettled([...this.dispatches.values()].map((record) => record.flight)); + } + + private async runDispatch( + record: DispatchRecord, + request: PiSubagentDispatchRequest, + details: SubagentDetailsV1, + onUpdate: PiSubagentDispatchOptions['onUpdate'], + ): Promise { + if (request.mode === 'chain') { + let previous = ''; + for (let index = 0; index < request.tasks.length; index += 1) { + const task = request.tasks[index] as PiSubagentTaskRequest; + if (record.controller.signal.aborted) { + this.markRemaining(details, index, 'aborted', onUpdate); + break; + } + const prompt = task.task.split('{previous}').join(previous); + await this.runTask(record, task, details, index, prompt, onUpdate); + const projected = details.tasks[index]; + if (projected?.status !== 'complete') { + this.markRemaining( + details, + index + 1, + record.controller.signal.aborted ? 'aborted' : 'skipped', + onUpdate, + ); + break; + } + previous = projected.summary ?? ''; + } + } else { + await Promise.all(request.tasks.map((task, index) => ( + this.runTask(record, task, details, index, task.task, onUpdate) + ))); + } + return { details: structuredClone(details) }; + } + + private async runTask( + record: DispatchRecord, + task: PiSubagentTaskRequest, + details: SubagentDetailsV1, + index: number, + prompt: string, + onUpdate: PiSubagentDispatchOptions['onUpdate'], + ): Promise { + const projected = details.tasks[index]; + if (!projected) return; + let releaseChild: (() => void) | undefined; + let processLease: PiProcessLease | undefined; + 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); + if (record.controller.signal.aborted) throw new PiSubagentChildError('SUBAGENT_ABORTED'); + projected.status = 'running'; + this.emit(details, onUpdate); + child = await this.openChild({ + ...record.identity, + dispatchId: details.dispatchId, + taskId: projected.taskId, + agentId: task.agentId, + toolProfile: task.toolProfile, + }); + record.children.add(child); + const result = await child.run(prompt, record.controller.signal, (summary) => { + projected.summary = safeSummary(summary); + this.emit(details, onUpdate); + }); + if (record.controller.signal.aborted) throw new PiSubagentChildError('SUBAGENT_ABORTED'); + projected.status = 'complete'; + projected.summary = safeSummary(result.summary); + const usage = safeUsage(result.usage); + if (usage) projected.usage = usage; + } catch (error) { + const aborted = record.controller.signal.aborted + || (error instanceof PiSubagentChildError && error.code === 'SUBAGENT_ABORTED'); + projected.status = aborted ? 'aborted' : 'error'; + projected.errorCode = publicErrorCode(error, aborted); + } finally { + if (child) { + record.children.delete(child); + await child.stop().catch(() => undefined); + } + processLease?.release(); + releaseChild?.(); + this.emit(details, onUpdate); + } + } + + 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, + status: 'aborted' | 'skipped', + onUpdate: PiSubagentDispatchOptions['onUpdate'], + ): void { + for (let index = from; index < details.tasks.length; index += 1) { + const task = details.tasks[index]; + if (!task || task.status !== 'queued') continue; + task.status = status; + if (status === 'aborted') task.errorCode = 'SUBAGENT_ABORTED'; + } + this.emit(details, onUpdate); + } + + private emit( + details: SubagentDetailsV1, + onUpdate: PiSubagentDispatchOptions['onUpdate'], + ): void { + if (!onUpdate) return; + try { + onUpdate(structuredClone(details)); + } catch { + // UI projection observers must not affect child lifecycle. + } + } +} diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index 02c2c43..e04490b 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -333,6 +333,15 @@ 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; + } + subscribe(listener: (event: PiWorkerPoolEvent) => void): () => void { this.listeners.add(listener); return () => this.listeners.delete(listener); diff --git a/electron/coding-runtime/pi/worker-process.ts b/electron/coding-runtime/pi/worker-process.ts index 818e039..ac49879 100644 --- a/electron/coding-runtime/pi/worker-process.ts +++ b/electron/coding-runtime/pi/worker-process.ts @@ -52,6 +52,7 @@ export type PiWorkerProcessOptions = { cwd: string; configDir: string; sessionDir: string; + tools?: readonly string[]; additionalArgs?: readonly string[]; env?: NodeJS.ProcessEnv; sensitiveValues?: readonly string[]; @@ -64,6 +65,7 @@ export type PiWorkerProcessOptions = { export function buildPiRpcArgs( sessionDir: string, additionalArgs: readonly string[] = [], + tools: readonly string[] = ['read', 'bash', 'edit', 'write', 'grep', 'find', 'ls', 'ask_user'], ): string[] { return [ '--mode', 'rpc', @@ -75,7 +77,7 @@ export function buildPiRpcArgs( '--no-themes', '--no-context-files', '--no-approve', - '--tools', 'read,bash,edit,write,grep,find,ls,ask_user', + '--tools', tools.join(','), ...additionalArgs, ]; } @@ -217,7 +219,11 @@ export class PiWorkerProcess { async start(): Promise { if (this.child) throw new Error('Pi worker process already started'); const generation = this.generationValue; - const args = buildPiRpcArgs(this.options.sessionDir, this.options.additionalArgs); + const args = buildPiRpcArgs( + this.options.sessionDir, + this.options.additionalArgs, + this.options.tools, + ); assertSensitiveValuesAbsentFromArgs(args, this.options.sensitiveValues); const child = spawn( this.options.executablePath, diff --git a/electron/coding-runtime/subagent-protocol.ts b/electron/coding-runtime/subagent-protocol.ts new file mode 100644 index 0000000..de1af59 --- /dev/null +++ b/electron/coding-runtime/subagent-protocol.ts @@ -0,0 +1,82 @@ +import type { PublicUsage, SubagentDetailsV1 } from './contracts'; + +const STATUSES = new Set(['queued', 'running', 'complete', 'error', 'aborted', 'skipped']); + +function recordValue(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null; +} + +function publicUsage(value: unknown): PublicUsage | undefined { + const usage = recordValue(value); + if (!usage + || typeof usage.inputTokens !== 'number' + || typeof usage.outputTokens !== 'number' + || !Number.isFinite(usage.inputTokens) + || !Number.isFinite(usage.outputTokens) + || usage.inputTokens < 0 + || usage.outputTokens < 0 + || (usage.cacheReadTokens !== undefined + && (typeof usage.cacheReadTokens !== 'number' + || !Number.isFinite(usage.cacheReadTokens) + || usage.cacheReadTokens < 0)) + || (usage.cacheWriteTokens !== undefined + && (typeof usage.cacheWriteTokens !== 'number' + || !Number.isFinite(usage.cacheWriteTokens) + || usage.cacheWriteTokens < 0))) { + return undefined; + } + return { + inputTokens: usage.inputTokens, + outputTokens: usage.outputTokens, + ...(typeof usage.cacheReadTokens === 'number' ? { cacheReadTokens: usage.cacheReadTokens } : {}), + ...(typeof usage.cacheWriteTokens === 'number' ? { cacheWriteTokens: usage.cacheWriteTokens } : {}), + }; +} + +export function projectSubagentDetailsV1(value: unknown): SubagentDetailsV1 | undefined { + const record = recordValue(value); + if (!record + || record.schema !== 'subagent.v1' + || typeof record.dispatchId !== 'string' + || !record.dispatchId.trim() + || !['single', 'parallel', 'chain'].includes(String(record.mode)) + || !Array.isArray(record.tasks) + || record.tasks.length === 0 + || record.tasks.length > 8) return undefined; + const tasks: SubagentDetailsV1['tasks'] = []; + for (const value of record.tasks) { + const task = recordValue(value); + if (!task + || typeof task.taskId !== 'string' + || !task.taskId.trim() + || typeof task.agentId !== 'string' + || !task.agentId.trim() + || (task.toolProfile !== 'read-only' && task.toolProfile !== 'coding') + || !STATUSES.has(String(task.status)) + || (task.summary !== undefined && typeof task.summary !== 'string') + || (task.errorCode !== undefined && typeof task.errorCode !== 'string')) return undefined; + const usage = task.usage === undefined ? undefined : publicUsage(task.usage); + if (task.usage !== undefined && !usage) return undefined; + tasks.push({ + taskId: task.taskId, + agentId: task.agentId, + toolProfile: task.toolProfile, + status: task.status as SubagentDetailsV1['tasks'][number]['status'], + ...(typeof task.summary === 'string' ? { summary: task.summary.slice(0, 4_000) } : {}), + ...(typeof task.errorCode === 'string' ? { errorCode: task.errorCode.slice(0, 64) } : {}), + ...(usage ? { usage } : {}), + }); + } + return { + schema: 'subagent.v1', + dispatchId: record.dispatchId, + mode: record.mode as SubagentDetailsV1['mode'], + tasks, + }; +} + +export function subagentDetailsOfResult(value: unknown): SubagentDetailsV1 | undefined { + return projectSubagentDetailsV1(recordValue(value)?.details); +} diff --git a/tests/unit/pi-event-projector.test.ts b/tests/unit/pi-event-projector.test.ts index a8c323d..5a0d04f 100644 --- a/tests/unit/pi-event-projector.test.ts +++ b/tests/unit/pi-event-projector.test.ts @@ -58,6 +58,52 @@ function apply( } describe('Pi event projector', () => { + it('projects only subagent.v1 details and never exposes unknown raw details', async () => { + const projector = new PiEventProjector({ createId: () => 'unused' }); + let snapshot = emptySnapshot(); + snapshot.nodes.push({ + kind: 'tool', id: 'tool-subagent', toolCallId: 'call-subagent', toolName: 'subagent', + title: 'subagent', inputText: '{}', status: 'running', output: [], + }); + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'tool_execution_update', + toolCallId: 'call-subagent', + partialResult: { + content: [], + details: { + schema: 'subagent.v1', dispatchId: 'dispatch-a', mode: 'parallel', + tasks: [{ + taskId: 'task-a', agentId: 'agent-a', toolProfile: 'read-only', + status: 'complete', summary: 'done', usage: { inputTokens: 3, outputTokens: 5 }, + }], + }, + }, + })); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'subagent', id: 'subagent:dispatch-a', runId: 'run-a', + details: expect.objectContaining({ schema: 'subagent.v1' }), + })); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'tool', id: 'tool-subagent', + details: expect.objectContaining({ dispatchId: 'dispatch-a' }), + })); + + snapshot.nodes.push({ + kind: 'tool', id: 'tool-unknown', toolCallId: 'call-unknown', toolName: 'subagent', + title: 'subagent', inputText: '{}', status: 'running', output: [], + }); + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'tool_execution_end', + toolCallId: 'call-unknown', isError: false, + result: { details: { schema: 'subagent.v2', raw: 'RAW_SECRET_DETAILS' } }, + })); + expect(JSON.stringify(snapshot)).not.toContain('RAW_SECRET_DETAILS'); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'tool', id: 'tool-unknown', + output: [expect.objectContaining({ text: 'Subagent details are unavailable for this version.' })], + })); + }); + it('keeps one assistant UI identity while content-index deltas become an authoritative message', async () => { const projector = new PiEventProjector({ createId: () => 'assistant-ui-a', diff --git a/tests/unit/pi-extension-bundle.test.ts b/tests/unit/pi-extension-bundle.test.ts index cb7c6ac..87d1f8d 100644 --- a/tests/unit/pi-extension-bundle.test.ts +++ b/tests/unit/pi-extension-bundle.test.ts @@ -6,8 +6,14 @@ import path from 'node:path'; import { pathToFileURL } from 'node:url'; import { afterEach, describe, expect, it } from 'vitest'; import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; +import { PiSubagentScheduler } from '../../electron/coding-runtime/pi/subagent'; +import { PiProcessBudget } from '../../electron/coding-runtime/pi/worker-pool'; type ExtensionHandler = (...arguments_: unknown[]) => Promise | unknown; +type ExtensionTool = { + name: string; + execute?: (...arguments_: unknown[]) => Promise; +}; const roots: string[] = []; const hosts: PiManagedExtensionHost[] = []; @@ -22,6 +28,15 @@ describe('Makelore Pi extension bundle', () => { const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-bundle-')); roots.push(root); const host = new PiManagedExtensionHost(); + const scheduler = new PiSubagentScheduler({ + processBudget: new PiProcessBudget(8), + openChild: async (input) => ({ + id: input.taskId, + async run() { return { summary: `done ${input.agentId}` }; }, + async stop() {}, + }), + }); + host.configureSubagents({ scheduler }); hosts.push(host); const extensionWorker = await host.registerWorker({ conversationId: 'conversation-a1', generation: 1, projectId: 'project-a', extensionsDir: root, @@ -38,22 +53,44 @@ describe('Makelore Pi extension bundle', () => { bridge: process.env.MAKELORE_PI_BRIDGE_URL, token: process.env.MAKELORE_PI_WORKER_TOKEN, context: process.env.MAKELORE_PI_CONTEXT_FILE, + role: process.env.MAKELORE_PI_WORKER_ROLE, }; Object.assign(process.env, extensionWorker.env); try { const module = await import(/* @vite-ignore */ pathToFileURL(extensionWorker.extensionPath).href) as { default(factory: { - registerTool(tool: { name: string }): void; + registerTool(tool: ExtensionTool): void; on(event: string, handler: ExtensionHandler): void; }): void; }; const handlers = new Map(); - const tools: string[] = []; + const tools = new Map(); module.default({ - registerTool: (tool) => tools.push(tool.name), + registerTool: (tool) => tools.set(tool.name, tool), on: (event, handler) => handlers.set(event, handler), }); - expect(tools).toEqual(['ask_user']); + expect([...tools.keys()]).toEqual(['ask_user', 'subagent']); + + const updates: unknown[] = []; + const subagentResult = await tools.get('subagent')?.execute?.( + 'subagent-1', + { + mode: 'single', + tasks: [{ agentId: 'agent-a', task: 'Inspect', toolProfile: 'read-only' }], + }, + new AbortController().signal, + (update: unknown) => updates.push(update), + ); + expect(subagentResult).toMatchObject({ + details: { + schema: 'subagent.v1', mode: 'single', + tasks: [{ agentId: 'agent-a', status: 'complete', summary: 'done agent-a' }], + }, + }); + expect(updates.length).toBeGreaterThan(0); + expect(updates.every((update) => ( + (update as { details?: { schema?: string } }).details?.schema === 'subagent.v1' + ))).toBe(true); const statuses: Array = []; const context = { @@ -84,12 +121,55 @@ describe('Makelore Pi extension bundle', () => { await handlers.get('tool_result')?.({ toolCallId: 'write-1' }); expect((await waiting).status).toBe(200); } finally { + await scheduler.close(); if (previousEnvironment.bridge === undefined) delete process.env.MAKELORE_PI_BRIDGE_URL; else process.env.MAKELORE_PI_BRIDGE_URL = previousEnvironment.bridge; if (previousEnvironment.token === undefined) delete process.env.MAKELORE_PI_WORKER_TOKEN; else process.env.MAKELORE_PI_WORKER_TOKEN = previousEnvironment.token; if (previousEnvironment.context === undefined) delete process.env.MAKELORE_PI_CONTEXT_FILE; else process.env.MAKELORE_PI_CONTEXT_FILE = previousEnvironment.context; + if (previousEnvironment.role === undefined) delete process.env.MAKELORE_PI_WORKER_ROLE; + else process.env.MAKELORE_PI_WORKER_ROLE = previousEnvironment.role; + } + }); + + it('does not expose parent-only tools from a child process', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-child-bundle-')); + roots.push(root); + const host = new PiManagedExtensionHost(); + hosts.push(host); + const child = await host.registerWorker({ + conversationId: 'conversation-child', generation: 1, projectId: 'project-a', + extensionsDir: root, role: 'child', runId: 'run-parent', + }); + const previous = { + bridge: process.env.MAKELORE_PI_BRIDGE_URL, + token: process.env.MAKELORE_PI_WORKER_TOKEN, + context: process.env.MAKELORE_PI_CONTEXT_FILE, + role: process.env.MAKELORE_PI_WORKER_ROLE, + }; + Object.assign(process.env, child.env); + try { + const module = await import( + /* @vite-ignore */ `${pathToFileURL(child.extensionPath).href}?child=${Date.now()}` + ) as { + default(factory: { + registerTool(tool: ExtensionTool): void; + on(event: string, handler: ExtensionHandler): void; + }): void; + }; + const tools: string[] = []; + module.default({ registerTool: (tool) => tools.push(tool.name), on: () => undefined }); + expect(tools).toEqual([]); + } finally { + for (const [key, value] of Object.entries(previous)) { + const environmentKey = key === 'bridge' ? 'MAKELORE_PI_BRIDGE_URL' + : key === 'token' ? 'MAKELORE_PI_WORKER_TOKEN' + : key === 'context' ? 'MAKELORE_PI_CONTEXT_FILE' + : 'MAKELORE_PI_WORKER_ROLE'; + if (value === undefined) delete process.env[environmentKey]; + else process.env[environmentKey] = value; + } } }); }); diff --git a/tests/unit/pi-extension-host.test.ts b/tests/unit/pi-extension-host.test.ts index e27fc93..76dd789 100644 --- a/tests/unit/pi-extension-host.test.ts +++ b/tests/unit/pi-extension-host.test.ts @@ -6,6 +6,11 @@ import path from 'node:path'; import { afterEach, describe, expect, it } from 'vitest'; import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; import { PiProjectWriteLeaseCoordinator } from '../../electron/coding-runtime/pi/write-lease'; +import { + PiSubagentChildError, + PiSubagentScheduler, +} from '../../electron/coding-runtime/pi/subagent'; +import { PiProcessBudget } from '../../electron/coding-runtime/pi/worker-pool'; const roots: string[] = []; const hosts: PiManagedExtensionHost[] = []; @@ -81,7 +86,7 @@ describe('managed Pi extension bridge', () => { 'utf8', )) as Record; expect(context).toEqual({ - conversationId: 'conversation-a', workerGeneration: 2, runId: 'run-a', + conversationId: 'conversation-a', workerGeneration: 2, role: 'parent', runId: 'run-a', }); await first.dispose(); const response = await post(replacement, { @@ -91,6 +96,117 @@ describe('managed Pi extension bridge', () => { expect(response.status).toBe(200); }); + it('streams stable subagent details and rejects recursive child dispatch', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-subagent-')); + roots.push(root); + const scheduler = new PiSubagentScheduler({ + processBudget: new PiProcessBudget(8), + openChild: async (input) => ({ + id: input.taskId, + async run() { return { summary: `done ${input.agentId}` }; }, + async stop() {}, + }), + }); + const host = new PiManagedExtensionHost(); + const tracked: Array<{ kind: string; id: string }> = []; + let untracked = 0; + host.configureSubagents({ + scheduler, + trackGenerationResource: (input) => { + tracked.push({ kind: input.kind, id: input.id }); + return () => { untracked += 1; }; + }, + }); + hosts.push(host); + const parent = await host.registerWorker({ + conversationId: 'conversation-a', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const child = await host.registerWorker({ + conversationId: 'conversation-a', generation: 1, projectId: 'project-a', extensionsDir: root, + role: 'child', runId: 'run-a', + }); + await host.bindRun('conversation-a', 1, 'run-a'); + const request = { + action: 'subagent.dispatch', conversationId: 'conversation-a', workerGeneration: 1, + runId: 'run-a', resourceId: 'subagent-tool', + request: { + mode: 'single', + tasks: [{ agentId: 'agent-a', task: 'Inspect', toolProfile: 'read-only' }], + }, + }; + const response = await post(parent, request); + expect(response.status).toBe(200); + const lines = (await response.text()).trim().split('\n').map((line) => JSON.parse(line)); + expect(lines.at(-1)).toMatchObject({ + done: true, + details: { + schema: 'subagent.v1', mode: 'single', + tasks: [{ agentId: 'agent-a', status: 'complete', summary: 'done agent-a' }], + }, + }); + expect(tracked).toEqual([{ kind: 'child', id: 'subagent-tool' }]); + expect(untracked).toBe(1); + expect((await post(child, request)).status).toBe(403); + await scheduler.close(); + }); + + it('propagates generation cancellation without an orphan child or leaked budget', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-child-cancel-')); + roots.push(root); + const processBudget = new PiProcessBudget(8); + let stopped = 0; + const scheduler = new PiSubagentScheduler({ + processBudget, + openChild: async (input) => ({ + id: input.taskId, + async run(_prompt, signal) { + await new Promise((_resolve, reject) => { + const abort = () => reject(new PiSubagentChildError('SUBAGENT_ABORTED')); + if (signal.aborted) abort(); + else signal.addEventListener('abort', abort, { once: true }); + }); + return { summary: 'unreachable' }; + }, + async stop() { stopped += 1; }, + }), + }); + let cancelGeneration: (() => void) | undefined; + let untracked = 0; + const host = new PiManagedExtensionHost(); + host.configureSubagents({ + scheduler, + trackGenerationResource: (input) => { + cancelGeneration = input.cancel; + return () => { untracked += 1; }; + }, + }); + hosts.push(host); + const parent = await host.registerWorker({ + conversationId: 'conversation-a', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + await host.bindRun('conversation-a', 1, 'run-a'); + const flight = post(parent, { + action: 'subagent.dispatch', conversationId: 'conversation-a', workerGeneration: 1, + runId: 'run-a', resourceId: 'subagent-tool', + request: { + mode: 'single', + tasks: [{ agentId: 'agent-a', task: 'Wait', toolProfile: 'read-only' }], + }, + }); + await expect.poll(() => processBudget.activeCount).toBe(1); + cancelGeneration?.(); + const response = await flight; + const lines = (await response.text()).trim().split('\n').map((line) => JSON.parse(line)); + expect(lines.at(-1)).toMatchObject({ + done: true, + details: { tasks: [{ status: 'aborted', errorCode: 'SUBAGENT_ABORTED' }] }, + }); + expect(stopped).toBe(1); + expect(processBudget.activeCount).toBe(0); + expect(untracked).toBe(1); + await scheduler.close(); + }); + it('validates worker identity and enforces project-scoped leases over loopback HTTP', async () => { const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-')); roots.push(root); @@ -152,4 +268,38 @@ describe('managed Pi extension bridge', () => { }); expect(currentWorker.status).toBe(200); }); + + it('joins a coding child to the same project write lease as its parent', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-child-lease-')); + roots.push(root); + const leases = new PiProjectWriteLeaseCoordinator(); + const host = new PiManagedExtensionHost(leases); + hosts.push(host); + const parent = await host.registerWorker({ + conversationId: 'conversation-a', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const child = await host.registerWorker({ + conversationId: 'conversation-a', generation: 1, projectId: 'project-a', extensionsDir: root, + role: 'child', runId: 'run-a', + }); + await host.bindRun('conversation-a', 1, 'run-a'); + const identity = { conversationId: 'conversation-a', workerGeneration: 1, runId: 'run-a' }; + const held = await post(parent, { + ...identity, action: 'lease.acquire', resourceId: 'parent-write', + }); + const parentLease = await held.json() as { leaseId: string }; + let childSettled = false; + const waiting = post(child, { + ...identity, action: 'lease.acquire', resourceId: 'child-write', + }).then((response) => { + childSettled = true; + return response; + }); + await expect.poll(() => leases.waitingCount('project-a')).toBe(1); + expect(childSettled).toBe(false); + expect((await post(parent, { + ...identity, action: 'lease.release', resourceId: 'parent-write', leaseId: parentLease.leaseId, + })).status).toBe(200); + expect((await waiting).status).toBe(200); + }); }); diff --git a/tests/unit/pi-managed-worker-opener.test.ts b/tests/unit/pi-managed-worker-opener.test.ts index ea93a26..c6d11f5 100644 --- a/tests/unit/pi-managed-worker-opener.test.ts +++ b/tests/unit/pi-managed-worker-opener.test.ts @@ -161,7 +161,7 @@ describe('managed Pi worker opener', () => { expect(argv).toContain('grilling'); expect(argv).toContain('--session-id'); expect(argv).toContain('--extension'); - expect(argv).toContain('makelore-runtime-v1.mjs'); + expect(argv).toContain('makelore-runtime-v2.mjs'); expect(options.additionalArgs?.filter((argument) => argument === '--extension')).toHaveLength(1); expect(argv).not.toContain('PRIVATE MANAGED PROMPT'); expect(argv).not.toContain('provider-secret-value'); diff --git a/tests/unit/pi-rpc-foundation.test.ts b/tests/unit/pi-rpc-foundation.test.ts index c39cdae..1dd6d7f 100644 --- a/tests/unit/pi-rpc-foundation.test.ts +++ b/tests/unit/pi-rpc-foundation.test.ts @@ -170,6 +170,20 @@ describe('Pi worker process', () => { 'read,bash,edit,write,grep,find,ls,ask_user', '--model', 'model-a', ]); + expect(buildPiRpcArgs('sessions', ['--no-session'], ['read', 'grep', 'find', 'ls'])) + .toEqual([ + '--mode', 'rpc', + '--offline', + '--session-dir', 'sessions', + '--no-extensions', + '--no-skills', + '--no-prompt-templates', + '--no-themes', + '--no-context-files', + '--no-approve', + '--tools', 'read,grep,find,ls', + '--no-session', + ]); }); it('correlates out-of-order responses, dispatches events, and reassembles partial lines', async () => { diff --git a/tests/unit/pi-session-projector.test.ts b/tests/unit/pi-session-projector.test.ts index 62b52ef..46244fc 100644 --- a/tests/unit/pi-session-projector.test.ts +++ b/tests/unit/pi-session-projector.test.ts @@ -33,6 +33,57 @@ function baseSnapshot(): ConversationSnapshot { const zeroCost = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }; describe('Pi session projector', () => { + it('hydrates stable subagent nodes and bounds unknown detail versions', async () => { + const snapshot = await projectPiSessionSnapshot({ + snapshot: baseSnapshot(), + workerGeneration: 1, + state: { sessionId: 'session-a', isStreaming: false, isCompacting: false }, + entries: { + leafId: 'entry-unknown-result', + entries: [ + { + type: 'message', id: 'entry-assistant', parentId: null, + message: { + role: 'assistant', stopReason: 'toolUse', usage: { input: 1, output: 1 }, + content: [ + { type: 'toolCall', id: 'call-known', name: 'subagent', arguments: {} }, + { type: 'toolCall', id: 'call-unknown', name: 'subagent', arguments: {} }, + ], + }, + }, + { + type: 'message', id: 'entry-known-result', parentId: 'entry-assistant', + message: { + role: 'toolResult', toolCallId: 'call-known', toolName: 'subagent', content: [], + details: { + schema: 'subagent.v1', dispatchId: 'dispatch-a', mode: 'single', + tasks: [{ + taskId: 'task-a', agentId: 'agent-a', toolProfile: 'coding', status: 'complete', + }], + }, + }, + }, + { + type: 'message', id: 'entry-unknown-result', parentId: 'entry-known-result', + message: { + role: 'toolResult', toolCallId: 'call-unknown', toolName: 'subagent', content: [], + details: { schema: 'subagent.v9', raw: 'RAW_SESSION_SECRET' }, + }, + }, + ], + }, + }); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'subagent', id: 'subagent:dispatch-a', + details: expect.objectContaining({ schema: 'subagent.v1' }), + })); + expect(JSON.stringify(snapshot)).not.toContain('RAW_SESSION_SECRET'); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'tool', toolCallId: 'call-unknown', + output: [expect.objectContaining({ text: 'Subagent details are unavailable for this version.' })], + })); + }); + it('hydrates only the authoritative active leaf path', async () => { const snapshot = await projectPiSessionSnapshot({ snapshot: baseSnapshot(), diff --git a/tests/unit/pi-subagent-child.test.ts b/tests/unit/pi-subagent-child.test.ts new file mode 100644 index 0000000..b919d25 --- /dev/null +++ b/tests/unit/pi-subagent-child.test.ts @@ -0,0 +1,155 @@ +// @vitest-environment node + +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { createCodingProjectAgent } from '../../electron/coding-projects/project-config'; +import { + createCodingProjectStore, + createLocalCodingProject, + createMemoryCodingProjectStorage, +} from '../../electron/coding-projects/project-store'; +import type { ProviderAccount } from '../../electron/shared/providers/types'; +import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; +import type { PiProcessError } from '../../electron/coding-runtime/pi/process-errors'; +import type { + PiRpcCommand, + PiRpcEvent, + PiRpcRequestOptions, + PiRpcResponse, +} from '../../electron/coding-runtime/pi/rpc-client'; +import { + createPiManagedSubagentChildOpener, + type PiSubagentProcessAdapter, +} from '../../electron/coding-runtime/pi/subagent-child'; +import type { PiWorkerProcessOptions } from '../../electron/coding-runtime/pi/worker-process'; + +const roots: string[] = []; +const NOW = '2026-08-23T00:00:00.000Z'; + +class FakeChildProcess implements PiSubagentProcessAdapter { + private readonly listeners = new Set<(event: PiRpcEvent) => void>(); + private readonly invalidationListeners = new Set<(error: PiProcessError) => void>(); + stopped = false; + + async start() { return this; } + + async request( + command: PiRpcCommand, + _options?: PiRpcRequestOptions, + ): Promise> { + if (command.type === 'prompt') { + queueMicrotask(() => { + for (const listener of this.listeners) { + listener({ + type: 'message_end', + message: { role: 'assistant', usage: { input: 7, output: 11, cacheRead: 2 } }, + }); + listener({ type: 'agent_settled' }); + } + }); + } + if (command.type === 'get_last_assistant_text') { + return { + type: 'response', id: 'summary', success: true, + data: { text: 'managed child summary' } as T, + }; + } + return { type: 'response', id: command.type, success: true, data: {} as T }; + } + + subscribe(listener: (event: PiRpcEvent) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + subscribeInvalidation(listener: (error: PiProcessError) => void): () => void { + this.invalidationListeners.add(listener); + return () => this.invalidationListeners.delete(listener); + } + + async stop() { + this.stopped = true; + return { mode: 'stdin-close' as const, code: 0, signal: null }; + } +} + +afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +describe('managed Pi subagent child opener', () => { + it('opens an ephemeral managed Agent with the exact tool profile and public result', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-subagent-child-')); + roots.push(root); + const projectPath = path.join(root, 'project'); + const projectStore = createCodingProjectStore(createMemoryCodingProjectStorage(), { + createId: () => 'project-a', now: () => NOW, + }); + await createLocalCodingProject({ projectPath, now: NOW }, projectStore); + await createCodingProjectAgent(projectPath, { + id: 'agent-a', avatarId: 'avatar-01', roleName: 'Reviewer', name: 'Agent A', + model: { accountId: 'account-a', modelId: 'model-a', thinkingLevel: 'high' }, + modelResolution: 'resolved', + responsibility: { mission: 'Review', owns: [], boundaries: [], collaborators: [], principles: [] }, + prompt: 'PRIVATE CHILD PROMPT', skillIds: ['grilling'], + }, { now: NOW }); + const account: ProviderAccount = { + id: 'account-a', vendorId: 'custom', label: 'Account A', authMode: 'api_key', + apiProtocol: 'openai-completions', baseUrl: 'https://provider.example/v1', model: 'model-a', + enabled: true, isDefault: true, createdAt: NOW, updatedAt: NOW, + }; + const host = new PiManagedExtensionHost(); + const processOptions: PiWorkerProcessOptions[] = []; + const processes: FakeChildProcess[] = []; + const opener = createPiManagedSubagentChildOpener({ + projectStore, + executablePath: 'electron.exe', + cliPath: 'pi-cli.js', + userDataDir: path.join(root, 'user-data'), + bundledSkillsDir: path.resolve('resources/coding-skills'), + extensionHost: host, + loadProviderInput: async () => ({ accounts: [account], modelSummaries: [] }), + resolveCredential: async () => 'provider-secret', + getRevision: () => ({ provider: 3, resources: 4 }), + createProcess: (options) => { + processOptions.push(options); + const child = new FakeChildProcess(); + processes.push(child); + return child; + }, + }); + const identity = { + conversationId: 'conversation-a', workerGeneration: 2, runId: 'run-a', projectId: 'project-a', + dispatchId: 'dispatch-a', agentId: 'agent-a', + }; + const readOnly = await opener({ + ...identity, taskId: 'task-read', toolProfile: 'read-only', + }); + const coding = await opener({ + ...identity, taskId: 'task-code', toolProfile: 'coding', + }); + expect(processOptions.map(({ tools }) => tools)).toEqual([ + ['read', 'grep', 'find', 'ls'], + ['read', 'bash', 'edit', 'write', 'grep', 'find', 'ls'], + ]); + for (const options of processOptions) { + expect(options.additionalArgs).toContain('--no-session'); + expect(options.additionalArgs).not.toContain('--session-id'); + expect(options.additionalArgs?.some((argument) => argument.includes('grilling'))).toBe(true); + expect(JSON.stringify(options.additionalArgs)).not.toContain('PRIVATE CHILD PROMPT'); + expect(JSON.stringify(options.additionalArgs)).not.toContain('provider-secret'); + expect(Object.values(options.env ?? {})).toContain('provider-secret'); + expect(options.env?.MAKELORE_PI_WORKER_ROLE).toBe('child'); + } + await expect(readOnly.run('Inspect', new AbortController().signal)).resolves.toEqual({ + summary: 'managed child summary', + usage: { inputTokens: 7, outputTokens: 11, cacheReadTokens: 2 }, + }); + await readOnly.stop(); + await coding.stop(); + expect(processes.every(({ stopped }) => stopped)).toBe(true); + await host.close(); + }); +}); diff --git a/tests/unit/pi-subagent.test.ts b/tests/unit/pi-subagent.test.ts new file mode 100644 index 0000000..ac6c0ba --- /dev/null +++ b/tests/unit/pi-subagent.test.ts @@ -0,0 +1,241 @@ +// @vitest-environment node + +import { describe, expect, it } from 'vitest'; +import { + PiSubagentChildError, + PiSubagentScheduler, + parsePiSubagentDispatchRequest, + type PiSubagentChild, + type PiSubagentChildOpenInput, +} from '../../electron/coding-runtime/pi/subagent'; +import { PiProcessBudget } from '../../electron/coding-runtime/pi/worker-pool'; + +function parent(runId = 'run-a') { + return { + conversationId: 'conversation-a', + workerGeneration: 1, + runId, + projectId: 'project-a', + }; +} + +function task(agentId: string, toolProfile: 'read-only' | 'coding' = 'read-only') { + return { agentId, task: `Inspect ${agentId}`, toolProfile }; +} + +function deferred() { + let resolve!: () => void; + const promise = new Promise((done) => { resolve = done; }); + return { promise, resolve }; +} + +describe('Pi subagent scheduler', () => { + it('validates one bounded dispatch mode and rejects a ninth task', () => { + expect(parsePiSubagentDispatchRequest({ mode: 'single', tasks: [task('one')] })).toEqual({ + mode: 'single', tasks: [task('one')], + }); + expect(parsePiSubagentDispatchRequest({ + mode: 'parallel', tasks: Array.from({ length: 8 }, (_, index) => task(`agent-${index}`)), + }).tasks).toHaveLength(8); + expect(() => parsePiSubagentDispatchRequest({ + mode: 'parallel', tasks: Array.from({ length: 9 }, (_, index) => task(`agent-${index}`)), + })).toThrowError('Subagent dispatch accepts at most 8 tasks'); + expect(() => parsePiSubagentDispatchRequest({ + mode: 'single', tasks: [task('one'), task('two')], + })).toThrowError('Single subagent dispatch requires exactly one task'); + expect(() => parsePiSubagentDispatchRequest({ mode: 'unknown', tasks: [task('one')] })) + .toThrowError('Subagent dispatch mode is invalid'); + }); + + it('shares one FIFO four-child cap across two parents and releases the process budget', async () => { + const gate = deferred(); + const processBudget = new PiProcessBudget(8); + let running = 0; + let maxRunning = 0; + const opened: PiSubagentChildOpenInput[] = []; + const scheduler = new PiSubagentScheduler({ + processBudget, + createId: (() => { + let id = 0; + return (kind) => `${kind}-${++id}`; + })(), + openChild: async (input) => { + opened.push(structuredClone(input)); + return { + id: input.taskId, + async run() { + running += 1; + maxRunning = Math.max(maxRunning, running); + await gate.promise; + running -= 1; + return { + summary: `done ${input.agentId}`, + usage: { inputTokens: 1, outputTokens: 2 }, + }; + }, + async stop() {}, + } satisfies PiSubagentChild; + }, + }); + const request = { + mode: 'parallel' as const, + tasks: Array.from({ length: 4 }, (_, index) => task(`agent-${index}`)), + }; + const left = scheduler.dispatch({ ...parent('run-left'), request }); + const right = scheduler.dispatch({ ...parent('run-right'), request }); + await expect.poll(() => running).toBe(4); + expect(maxRunning).toBe(4); + expect(processBudget.activeCount).toBe(4); + gate.resolve(); + const [leftResult, rightResult] = await Promise.all([left, right]); + expect(leftResult.details.tasks).toHaveLength(4); + expect(rightResult.details.tasks).toHaveLength(4); + expect(opened).toHaveLength(8); + expect(maxRunning).toBe(4); + expect(processBudget.activeCount).toBe(0); + await scheduler.close(); + }); + + it('preserves parallel sibling results and stops a chain at its first failure', async () => { + const scheduler = new PiSubagentScheduler({ + processBudget: new PiProcessBudget(8), + createId: (() => { + let id = 0; + return (kind) => `${kind}-${++id}`; + })(), + openChild: async (input) => ({ + id: input.taskId, + async run(prompt) { + if (input.agentId === 'broken') { + throw new PiSubagentChildError('SUBAGENT_CHILD_FAILED'); + } + return { + summary: `${input.agentId}:${prompt}`, + usage: { inputTokens: 3, outputTokens: 5 }, + }; + }, + async stop() {}, + }), + }); + const parallel = await scheduler.dispatch({ + ...parent('parallel'), + request: { + mode: 'parallel', + tasks: [task('left'), task('broken'), task('right')], + }, + }); + expect(parallel.details.tasks).toEqual(expect.arrayContaining([ + expect.objectContaining({ agentId: 'left', status: 'complete', summary: 'left:Inspect left' }), + expect.objectContaining({ agentId: 'broken', status: 'error', errorCode: 'SUBAGENT_CHILD_FAILED' }), + expect.objectContaining({ agentId: 'right', status: 'complete', usage: { inputTokens: 3, outputTokens: 5 } }), + ])); + + const chain = await scheduler.dispatch({ + ...parent('chain'), + request: { + mode: 'chain', + tasks: [ + { agentId: 'first', task: 'first', toolProfile: 'read-only' }, + { agentId: 'broken', task: 'review {previous}', toolProfile: 'read-only' }, + { agentId: 'never', task: 'never', toolProfile: 'coding' }, + ], + }, + }); + expect(chain.details.tasks.map(({ status }) => status)).toEqual(['complete', 'error', 'skipped']); + expect(chain.details.tasks[1]).toMatchObject({ errorCode: 'SUBAGENT_CHILD_FAILED' }); + await scheduler.close(); + }); + + it('aborts every unfinished child with the parent and leaves no permit or process lease', async () => { + const processBudget = new PiProcessBudget(8); + const stopped: string[] = []; + const scheduler = new PiSubagentScheduler({ + processBudget, + openChild: async (input) => ({ + id: input.taskId, + async run(_prompt, signal) { + await new Promise((_resolve, reject) => { + const abort = () => reject(new PiSubagentChildError('SUBAGENT_ABORTED')); + if (signal.aborted) abort(); + else signal.addEventListener('abort', abort, { once: true }); + }); + return { summary: 'unreachable' }; + }, + async stop() { stopped.push(input.taskId); }, + }), + }); + const controller = new AbortController(); + const flight = scheduler.dispatch({ + ...parent('abort'), + request: { mode: 'parallel', tasks: [task('left'), task('right', 'coding')] }, + }, { signal: controller.signal }); + await expect.poll(() => processBudget.activeCount).toBe(2); + controller.abort(); + const result = await flight; + expect(result.details.tasks.map(({ status }) => status)).toEqual(['aborted', 'aborted']); + expect(stopped).toHaveLength(2); + expect(processBudget.activeCount).toBe(0); + await scheduler.close(); + }); + + it('marks the rest of an in-flight chain aborted when its parent aborts', async () => { + const processBudget = new PiProcessBudget(8); + const scheduler = new PiSubagentScheduler({ + processBudget, + openChild: async (input) => ({ + id: input.taskId, + async run(_prompt, signal) { + await new Promise((_resolve, reject) => { + const abort = () => reject(new PiSubagentChildError('SUBAGENT_ABORTED')); + if (signal.aborted) abort(); + else signal.addEventListener('abort', abort, { once: true }); + }); + return { summary: 'unreachable' }; + }, + async stop() {}, + }), + }); + const controller = new AbortController(); + const flight = scheduler.dispatch({ + ...parent('abort-chain'), + request: { mode: 'chain', tasks: [task('one'), task('two'), task('three')] }, + }, { signal: controller.signal }); + await expect.poll(() => processBudget.activeCount).toBe(1); + controller.abort(); + const result = await flight; + expect(result.details.tasks.map(({ status }) => status)) + .toEqual(['aborted', 'aborted', 'aborted']); + expect(processBudget.activeCount).toBe(0); + await scheduler.close(); + }); + + it('reclaims an idle parent process instead of deadlocking on a full global budget', async () => { + const processBudget = new PiProcessBudget(2); + const firstParent = await processBudget.acquire(); + const idleParent = await processBudget.acquire(); + let reclaimed = 0; + const scheduler = new PiSubagentScheduler({ + processBudget, + reclaimProcessCapacity: async () => { + reclaimed += 1; + idleParent.release(); + return true; + }, + openChild: async (input) => ({ + id: input.taskId, + async run() { return { summary: 'done' }; }, + async stop() {}, + }), + }); + await expect(scheduler.dispatch({ + ...parent('full-budget'), + request: { mode: 'single', tasks: [task('agent-a')] }, + })).resolves.toMatchObject({ + details: { tasks: [{ status: 'complete' }] }, + }); + expect(reclaimed).toBe(1); + expect(processBudget.activeCount).toBe(1); + firstParent.release(); + await scheduler.close(); + }); +}); diff --git a/tests/unit/pi-worker-process-real.test.ts b/tests/unit/pi-worker-process-real.test.ts index daa513a..a5af691 100644 --- a/tests/unit/pi-worker-process-real.test.ts +++ b/tests/unit/pi-worker-process-real.test.ts @@ -73,4 +73,53 @@ describe('locked Pi worker process smoke', () => { await extensionHost.close(); } }, 15_000); + + it('starts a real ephemeral read-only child with the child extension role', async () => { + const requireFromProject = createRequire(resolve('package.json')); + const electronExecutable = requireFromProject('electron') as string; + const packageRoot = realpathSync(resolve( + 'node_modules', + '@earendil-works', + 'pi-coding-agent', + )); + const root = await mkdtemp(join(tmpdir(), 'makelore-pi-real-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: 'real-parent', + generation: 1, + projectId: 'real-project', + extensionsDir: join(root, 'extensions'), + role: 'child', + runId: 'real-parent-run', + }); + const worker = await new PiWorkerProcess({ + executablePath: electronExecutable, + cliPath: join(packageRoot, 'dist', 'cli.js'), + cwd, + configDir, + sessionDir, + tools: ['read', 'grep', 'find', 'ls'], + additionalArgs: ['--extension', extension.extensionPath, '--no-session'], + env: extension.env, + sensitiveValues: extension.sensitiveValues, + commandTimeoutMs: 5_000, + }).start(); + try { + await expect(worker.request({ type: 'get_state' })).resolves.toMatchObject({ + type: 'response', command: 'get_state', success: true, + }); + 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); });