From 9a31dacb2aad92a2deee0cec11fdb162a2916bfc Mon Sep 17 00:00:00 2001 From: brother7 <7brother7@gmail.com> Date: Sat, 22 Aug 2026 20:05:59 +0800 Subject: [PATCH] feat: add Pi process and RPC foundation --- .../tasks/20260822-pi-process-rpc-d4f9a2c1.md | 90 +++++ electron-builder.yml | 29 ++ electron/coding-runtime/pi/process-errors.ts | 38 ++ electron/coding-runtime/pi/rpc-client.ts | 268 +++++++++++++ electron/coding-runtime/pi/rpc-framer.ts | 91 +++++ electron/coding-runtime/pi/worker-process.ts | 356 ++++++++++++++++++ package.json | 7 +- scripts/bundle-pi-runtime.mjs | 123 ++++++ scripts/lib/pi-runtime-bundle.mjs | 301 +++++++++++++++ scripts/probe-pi-packaged-runtime.mjs | 81 ++-- tests/fixtures/fake-pi-rpc-child.mjs | 128 +++++++ tests/unit/pi-rpc-foundation.test.ts | 271 +++++++++++++ tests/unit/pi-runtime-bundle.test.ts | 137 +++++++ tests/unit/pi-worker-process-real.test.ts | 61 +++ 14 files changed, 1955 insertions(+), 26 deletions(-) create mode 100644 .project-docs/30-worklog/tasks/20260822-pi-process-rpc-d4f9a2c1.md create mode 100644 electron/coding-runtime/pi/process-errors.ts create mode 100644 electron/coding-runtime/pi/rpc-client.ts create mode 100644 electron/coding-runtime/pi/rpc-framer.ts create mode 100644 electron/coding-runtime/pi/worker-process.ts create mode 100644 scripts/bundle-pi-runtime.mjs create mode 100644 scripts/lib/pi-runtime-bundle.mjs create mode 100644 tests/fixtures/fake-pi-rpc-child.mjs create mode 100644 tests/unit/pi-rpc-foundation.test.ts create mode 100644 tests/unit/pi-runtime-bundle.test.ts create mode 100644 tests/unit/pi-worker-process-real.test.ts diff --git a/.project-docs/30-worklog/tasks/20260822-pi-process-rpc-d4f9a2c1.md b/.project-docs/30-worklog/tasks/20260822-pi-process-rpc-d4f9a2c1.md new file mode 100644 index 0000000..1600504 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260822-pi-process-rpc-d4f9a2c1.md @@ -0,0 +1,90 @@ +# Task: Implement PI-030 Pi process and RPC foundation + +## Identity + +- Task ID: 20260822-pi-process-rpc-d4f9a2c1 +- Mode: Feature +- Branch: codex/20260822-pi-process-rpc-d4f9a2c1-pi-process-rpc +- Worktree: D:\Datas\OthersProjects\makelore-pi-process-rpc-d4f9a2c1 +- Base commit: b6f693048d0edf831aa8c04d9790f67921ebdab9 +- Owner: codex +- Status: Completed + +## Scope + +- Implement `PI-030 Pi process and RPC foundation` from the merged Pi cutover Spec/ticket graph. +- Add the Main-private Pi worker process, strict LF JSONL framer, correlated RPC client, bounded diagnostics, and stable process/RPC errors under `electron/coding-runtime/pi`. +- Add a programmable fake Pi child fixture and focused tests for chunking, malformed/oversized output, out-of-order response, event dispatch, timeout/abort, retry boundaries, crash, backpressure, stderr redaction, and process-tree cleanup. +- Add the permanent Pi production-closure staging script, deterministic runtime manifest, closure verifier, and initial package-resource wiring. +- Verify the locked real Pi `get_state` path and staged entry load/exit without connecting PI to project CRUD, Host API, Renderer, Provider catalog, worker pool, or schema migration. + +## Intent And Constraints + +- Pi remains pinned to `@earendil-works/pi-coding-agent@0.84.2`; launch through Electron executable plus `ELECTRON_RUN_AS_NODE=1` with explicit RPC, offline, no-approve, and no automatic resource-discovery arguments. +- Stdout is protocol-only; any malformed, invalid UTF-8, oversized, blank/non-JSONL, or trailing partial record invalidates and terminates only that worker. Stderr is bounded, sanitized diagnostic text and never participates in RPC. +- Correlate every command by ID, support out-of-order responses and independent events, settle every pending request on timeout/abort/exit/invalidation, and never auto-retry prompt/steer/follow-up/fork/compact. +- Graceful shutdown closes stdin and waits at most three seconds; forced cleanup must terminate the complete child tree. No promise or process may remain after disposal. +- Convert the PI-000 temporary staging workaround into an explicit production bundler: copy the published package root without its installed dependency tree, remove the ten root devDependencies in the staged copy so npm 11 can consume the production-only shrinkwrap, run the repository-pinned npm with scripts/audit/fund disabled, and verify package identities/assets independent of same-version physical flattening. +- Runtime manifest content must be deterministic and path-independent. Do not add checksum/hash sidecars, compatibility layers, dual RPC/SDK paths, Provider logic, Renderer wire types, or incidental OpenCode refactors. +- Preserve the explicit user waivers: real Provider compatibility/concurrency/credential isolation remains unverified Accepted Risk, not Pass; macOS x64/arm64 execution remains deferred to mandatory PI-150 validation. + +## Project Context Loaded + +- `maintain-project-docs` Concurrent Task Gate and Planning Gate passed in the isolated worktree above at base `b6f6930`. +- The base history contains planning `a8c0806`, PI-000 `2bc423e`, merged baseline `72f7a26`, PI-010 implementation `4deba3a`, and PI-010 completion record `b6f6930`. +- The planner task recomputed the 18-node/31-edge DAG after PI-010 and recommended PI-030 as the unique next implementation ticket, while PI-020 remains Ready but intentionally waits. +- Required project entry/current-state/architecture/domain/evidence/commitment records, the merged Pi Spec/tickets, PI-000 evidence, and PI-010 task record were read before planning. +- Other locally planning task records were inspected read-only. Their scopes concern OpenCode model switching, Canvas, Robot, release, or old reviews; none owns Pi process/RPC/bundling. No semantic or file-write conflict was found. +- The canonical project-positioning file remains a placeholder. Repository guidance, current integrated state, and the merged Pi proposal/ticket graph are the authority for this task. +- `codebase-design` keeps the public worker surface small while framing, correlation, retries, diagnostics, and cleanup remain implementation-private. The user-prohibited parallel-subagent design path is not used. + +## Plan + +1. Implement stable process/RPC errors, the strict LF JSONL framer, and the correlated RPC client with focused unit tests. +2. Implement the Pi worker-process lifecycle and programmable fake child, including target-only invalidation, bounded diagnostics, graceful/forced process-tree cleanup, timeout/abort, and backpressure behavior. +3. Implement permanent production-closure staging, package/version/asset verification, deterministic manifest generation, and initial package/build resource wiring. +4. Run the fake-child matrix, real locked Pi `get_state`, staged-entry load/exit, typecheck, lint, full tests, production compile, and task-documentation gates. +5. Commit the implementation and task record, mark the task ready for integration, and send the result to the architecture/planner task. + +## Outcome + +- Implemented the Main-private Pi process/RPC foundation with stable process errors, strict byte-bounded LF JSONL framing, command correlation, out-of-order response handling, event dispatch, timeout/AbortSignal settlement, read-only-once retry boundaries, and write/backpressure handling. +- Implemented an Electron-Node Pi worker with explicit offline/no-approve/no-auto-resource arguments, generation-scoped invalidation, bounded/redacted stderr and protocol diagnostics, stdin-close graceful shutdown, and full-tree forced cleanup. +- Added a programmable fake child covering response/event/out-of-order/partial/malformed/blank/invalid UTF-8/oversized/trailing partial/timeout/retry/stderr/crash/descendant cases, plus a locked real Pi `get_state` smoke through Electron 43.4.0 / Node 24.18.1. +- Implemented permanent npm 11.6.2 production staging from the published package and production-only shrinkwrap. The staged copy explicitly removes the ten mismatched root devDependencies, installs with `--omit=dev --ignore-scripts`, verifies package identities independently of physical flattening, inventories native/WASM assets, and emits a deterministic path-free manifest. +- Wired host-target staging into build/package commands and product `extraResources`. During the production-shaped artifact probe, confirmed that Electron Builder omits a FileSet root named `node_modules`; corrected all OS mappings to copy runtime root files and `node_modules` through two explicit FileSets. +- The controlled Electron Builder `dir` artifact loads Pi from `resources/pi-runtime`, contains all 130 expected production package identities and all 6 inventoried assets, preserves a session, exits by closing stdin, and passes two-worker local overlap/abort isolation. +- Real Provider account verification remains explicitly waived by user decision: protocol/account compatibility, real concurrency, credential isolation, and provider-specific behavior are Accepted Risk and remain `realTurnVerified=false`, not Pass. macOS x64/arm64 execution remains deferred to PI-150 and is not claimed as Pass. + +## Verification + +- `pnpm install --frozen-lockfile`: passed with the repository-pinned pnpm 10.33.4. +- Focused process/RPC/runtime tests: 4 files, 30 tests passed. +- `pnpm run typecheck`: passed. +- Scoped ESLint for every changed TypeScript/JavaScript implementation and test file: passed. +- `pnpm run lint:check`: passed with 0 errors and 6 pre-existing warnings outside this task's files. +- `pnpm test`: 182 files, 2118 tests passed. The first loaded full-suite run exposed two sub-250ms test-only timing budgets; those were raised while retaining the timeout semantics, then the focused and full suites passed. +- `pnpm run build:vite`: passed for Renderer, Main, Preload, and release utility output; existing Vite dynamic-import and chunk-size warnings remain unchanged. +- `pnpm run bundle:pi-runtime`: passed for `win32-x64`, generating a 130-package/6-asset production closure. +- Repeated permanent staging generated byte-identical manifest content; the manifest contains no local path, timestamp, checksum, or build-directory identity. +- `node scripts/probe-pi-packaged-runtime.mjs --samples 1 --timeout-ms 10000 --report release/pi-runtime-probe-result.json`: passed against the production-shaped `resources/pi-runtime` layout. Packaged closure had zero missing/relocated packages and zero missing assets; cold ready was 610 ms, warm ready 580 ms, and both measured exits were 13 ms by stdin close. +- Forced Windows child-tree cleanup test passed; no macOS execution was performed by explicit user direction. + +## Follow-ups + +- `PI-040` may consume the process foundation after `PI-020` supplies schema v2; it owns Provider catalog, secrets, managed prompt/skill/resource projection, and must not move those concerns into this worker. +- `PI-050` may compose this single-worker primitive into the worker pool/session registry after `PI-010`, `PI-020`, `PI-030`, and `PI-040` are integrated. +- `PI-150` must run final release-artifact validation for Windows/Linux and the explicitly deferred macOS x64/arm64 targets. The real Provider waiver stays Accepted Risk unless the user later requests non-blocking evidence collection. + +## Promotion Candidates + +- Target: Pi runtime packaging architecture and final PI-150 verifier guidance. + Proposal: record that Electron Builder unconditionally skips a FileSet source root named `node_modules`; production Pi resources therefore require one FileSet for the staged root excluding `node_modules` and a second FileSet rooted at staged `node_modules`. + Evidence: the first production-shaped controlled artifact omitted all 130 dependencies and 6 assets; the two-FileSet mapping produced zero missing packages/assets and loaded the real Pi entry from `resources/pi-runtime`. + Future impact: prevents a later release build from silently shipping only the Pi root package while dropping its production dependency closure. + Semantic conflicts: none; this refines the existing `resources/pi-runtime` packaging contract without changing runtime ownership. +- Target: Pi runtime staging architecture. + Proposal: promote npm 11.6.2 plus staged-copy root-devDependency removal and identity-based closure validation as the permanent reproducible handling for the published Pi 0.84.2 package/shrinkwrap mismatch. + Evidence: clean staging, flattening fixture, deterministic manifest comparison, and packaged load/exit all passed. + Future impact: PI-150 can reuse the same builder and verifier rather than recreating the PI-000 temporary workaround. + Semantic conflicts: none; the source package and frozen shrinkwrap remain unmodified. diff --git a/electron-builder.yml b/electron-builder.yml index 41e5e52..2baf850 100644 --- a/electron-builder.yml +++ b/electron-builder.yml @@ -55,6 +55,17 @@ publish: # macOS Configuration mac: extraResources: + - from: build/pi-runtime/darwin-${arch} + to: pi-runtime + filter: + - "**/*" + - "!node_modules{,/**/*}" + # electron-builder always skips a FileSet's root node_modules directory; + # copy its contents from node_modules as a second source root. + - from: build/pi-runtime/darwin-${arch}/node_modules + to: pi-runtime/node_modules + filter: + - "**/*" - from: resources/bin/darwin-${arch} to: bin - from: resources/python/darwin-${arch}/python @@ -110,6 +121,15 @@ win: # toolchain, which is blocked in offline development environments. signAndEditExecutable: false extraResources: + - from: build/pi-runtime/win32-${arch} + to: pi-runtime + filter: + - "**/*" + - "!node_modules{,/**/*}" + - from: build/pi-runtime/win32-${arch}/node_modules + to: pi-runtime/node_modules + filter: + - "**/*" - from: resources/bin/win32-${arch} to: bin - from: resources/python/win32-${arch}/python @@ -138,6 +158,15 @@ nsis: # Linux Configuration linux: extraResources: + - from: build/pi-runtime/linux-${arch} + to: pi-runtime + filter: + - "**/*" + - "!node_modules{,/**/*}" + - from: build/pi-runtime/linux-${arch}/node_modules + to: pi-runtime/node_modules + filter: + - "**/*" - from: resources/bin/linux-${arch} to: bin - from: resources/python/linux-${arch}/python diff --git a/electron/coding-runtime/pi/process-errors.ts b/electron/coding-runtime/pi/process-errors.ts new file mode 100644 index 0000000..fd9ca10 --- /dev/null +++ b/electron/coding-runtime/pi/process-errors.ts @@ -0,0 +1,38 @@ +export type PiProcessErrorCode = + | 'PI_RPC_PROTOCOL_ERROR' + | 'PI_RPC_TIMEOUT' + | 'PI_RPC_ABORTED' + | 'PI_RPC_RESPONSE_ERROR' + | 'PI_RPC_WRITE_FAILED' + | 'PI_RPC_EXITED' + | 'PI_WORKER_START_FAILED' + | 'PI_WORKER_STOPPED' + | 'PI_WORKER_STOP_FAILED'; + +export type PiProcessErrorDetails = { + cause?: unknown; + generation?: number; + diagnostic?: string; +}; + +export class PiProcessError extends Error { + readonly code: PiProcessErrorCode; + readonly generation?: number; + readonly diagnostic?: string; + + constructor( + code: PiProcessErrorCode, + message: string, + details: PiProcessErrorDetails = {}, + ) { + super(message, details.cause === undefined ? undefined : { cause: details.cause }); + this.name = 'PiProcessError'; + this.code = code; + this.generation = details.generation; + this.diagnostic = details.diagnostic; + } +} + +export function isPiProcessError(error: unknown): error is PiProcessError { + return error instanceof PiProcessError; +} diff --git a/electron/coding-runtime/pi/rpc-client.ts b/electron/coding-runtime/pi/rpc-client.ts new file mode 100644 index 0000000..9f2a437 --- /dev/null +++ b/electron/coding-runtime/pi/rpc-client.ts @@ -0,0 +1,268 @@ +import type { Writable } from 'node:stream'; +import { PiProcessError } from './process-errors'; + +export type PiRpcCommand = { + type: string; + [key: string]: unknown; +}; + +export type PiRpcResponse = { + type: 'response'; + id: string; + command?: string; + success: boolean; + data?: T; + error?: string; +}; + +export type PiRpcEvent = Record & { type: string }; +export type PiRpcRetryPolicy = 'none' | 'read-only-once'; + +export type PiRpcRequestOptions = { + signal?: AbortSignal; + timeoutMs?: number; + retry?: PiRpcRetryPolicy; +}; + +type PendingRequest = { + commandType: string; + resolve(response: PiRpcResponse): void; + reject(error: PiProcessError): void; + cancel(): void; +}; + +type PiRpcClientOptions = { + generation: number; + defaultTimeoutMs?: number; + idPrefix?: string; + onEventListenerError?(error: unknown): void; +}; + +const READ_ONLY_COMMANDS = new Set([ + 'get_available_models', + 'get_available_thinking_levels', + 'get_commands', + 'get_entries', + 'get_fork_messages', + 'get_last_assistant_text', + 'get_messages', + 'get_session_stats', + 'get_state', + 'get_tree', +]); + +function recordValue(value: unknown): value is Record { + return Boolean(value) && typeof value === 'object' && !Array.isArray(value); +} + +function responseValue(value: Record): value is PiRpcResponse { + return value.type === 'response' + && typeof value.id === 'string' + && typeof value.success === 'boolean'; +} + +export class PiRpcClient { + private readonly writable: Writable; + private readonly generation: number; + private readonly defaultTimeoutMs: number; + private readonly idPrefix: string; + private readonly onEventListenerError: ((error: unknown) => void) | undefined; + private readonly pending = new Map(); + private readonly retiredIds = new Set(); + private readonly listeners = new Set<(event: PiRpcEvent) => void>(); + private sequence = 0; + private invalidated: PiProcessError | null = null; + + constructor(writable: Writable, options: PiRpcClientOptions) { + if (!Number.isSafeInteger(options.generation) || options.generation <= 0) { + throw new Error('generation must be a positive safe integer'); + } + const defaultTimeoutMs = options.defaultTimeoutMs ?? 10_000; + if (!Number.isSafeInteger(defaultTimeoutMs) || defaultTimeoutMs <= 0) { + throw new Error('defaultTimeoutMs must be a positive safe integer'); + } + this.writable = writable; + this.generation = options.generation; + this.defaultTimeoutMs = defaultTimeoutMs; + this.idPrefix = options.idPrefix ?? 'makelore'; + this.onEventListenerError = options.onEventListenerError; + this.writable.on('error', (error) => { + this.invalidate(new PiProcessError('PI_RPC_WRITE_FAILED', 'Pi RPC stdin failed', { + cause: error, + generation: this.generation, + })); + }); + } + + get pendingCount(): number { + return this.pending.size; + } + + subscribe(listener: (event: PiRpcEvent) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + async request( + command: PiRpcCommand, + options: PiRpcRequestOptions = {}, + ): Promise> { + if (!command.type) throw new Error('Pi RPC command type is required'); + const retry = options.retry ?? 'none'; + if (retry === 'read-only-once' && !READ_ONLY_COMMANDS.has(command.type)) { + throw new Error(`Pi RPC ${command.type} is not a retryable read-only command`); + } + + const attempts = retry === 'read-only-once' ? 2 : 1; + let lastError: unknown; + for (let attempt = 0; attempt < attempts; attempt += 1) { + try { + return await this.requestOnce(command, options); + } catch (error) { + lastError = error; + if (!(error instanceof PiProcessError) || error.code !== 'PI_RPC_TIMEOUT') throw error; + if (this.invalidated || options.signal?.aborted) throw error; + } + } + throw lastError; + } + + accept(value: unknown): void { + if (!recordValue(value) || typeof value.type !== 'string') { + throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi RPC record must be an object with a type'); + } + if (value.type === 'response') { + if (!responseValue(value)) { + throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi RPC response shape is invalid'); + } + if (this.retiredIds.delete(value.id)) return; + const pending = this.pending.get(value.id); + if (!pending) { + throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', `Pi RPC response used unknown id ${value.id}`); + } + pending.cancel(); + this.pending.delete(value.id); + if (value.success) pending.resolve(value); + else { + pending.reject(new PiProcessError( + 'PI_RPC_RESPONSE_ERROR', + `Pi RPC ${value.command ?? pending.commandType} failed: ${value.error ?? 'unknown error'}`, + { generation: this.generation }, + )); + } + return; + } + + for (const listener of this.listeners) { + try { + listener(value as PiRpcEvent); + } catch (error) { + try { + this.onEventListenerError?.(error); + } catch { + // Diagnostics must not affect transport state. + } + } + } + } + + invalidate(error: PiProcessError): void { + if (this.invalidated) return; + this.invalidated = error; + for (const [id, pending] of this.pending) { + pending.cancel(); + this.retire(id); + pending.reject(error); + } + this.pending.clear(); + } + + private async requestOnce( + command: PiRpcCommand, + options: PiRpcRequestOptions, + ): Promise> { + if (this.invalidated) throw this.invalidated; + if (options.signal?.aborted) { + throw new PiProcessError('PI_RPC_ABORTED', `Pi RPC ${command.type} was aborted`, { + generation: this.generation, + }); + } + + const timeoutMs = options.timeoutMs ?? this.defaultTimeoutMs; + if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) { + throw new Error('timeoutMs must be a positive safe integer'); + } + const id = `${this.idPrefix}-${this.generation}-${++this.sequence}`; + + const response = new Promise>((resolve, reject) => { + const timeout = setTimeout(() => { + this.pending.delete(id); + this.retire(id); + reject(new PiProcessError( + 'PI_RPC_TIMEOUT', + `Pi RPC ${command.type} timed out after ${timeoutMs}ms`, + { generation: this.generation }, + )); + }, timeoutMs); + const abort = (): void => { + this.pending.delete(id); + this.retire(id); + clearTimeout(timeout); + reject(new PiProcessError('PI_RPC_ABORTED', `Pi RPC ${command.type} was aborted`, { + generation: this.generation, + })); + }; + options.signal?.addEventListener('abort', abort, { once: true }); + this.pending.set(id, { + commandType: command.type, + resolve: (value) => resolve(value as PiRpcResponse), + reject, + cancel: () => { + clearTimeout(timeout); + options.signal?.removeEventListener('abort', abort); + }, + }); + }); + + const write = this.write(`${JSON.stringify({ ...command, id })}\n`).catch((error) => { + throw new PiProcessError('PI_RPC_WRITE_FAILED', `Could not write Pi RPC ${command.type}`, { + cause: error, + generation: this.generation, + }); + }); + try { + const [, record] = await Promise.all([write, response]); + return record; + } catch (error) { + const pending = this.pending.get(id); + if (pending) { + pending.cancel(); + this.pending.delete(id); + this.retire(id); + pending.reject(error instanceof PiProcessError + ? error + : new PiProcessError('PI_RPC_WRITE_FAILED', `Could not write Pi RPC ${command.type}`, { + cause: error, + generation: this.generation, + })); + } + throw error; + } + } + + private write(line: string): Promise { + return new Promise((resolve, reject) => { + this.writable.write(line, (error?: Error | null) => { + if (error) reject(error); + else resolve(); + }); + }); + } + + private retire(id: string): void { + this.retiredIds.add(id); + if (this.retiredIds.size <= 256) return; + const oldest = this.retiredIds.values().next().value as string | undefined; + if (oldest) this.retiredIds.delete(oldest); + } +} diff --git a/electron/coding-runtime/pi/rpc-framer.ts b/electron/coding-runtime/pi/rpc-framer.ts new file mode 100644 index 0000000..06c3756 --- /dev/null +++ b/electron/coding-runtime/pi/rpc-framer.ts @@ -0,0 +1,91 @@ +import { PiProcessError } from './process-errors'; + +export const DEFAULT_MAX_PI_RPC_LINE_BYTES = 1024 * 1024; + +type StrictLfJsonlFramerOptions = { + maxLineBytes?: number; + onRecord(record: unknown): void; +}; + +function protocolError(message: string): PiProcessError { + return new PiProcessError('PI_RPC_PROTOCOL_ERROR', message); +} + +export class StrictLfJsonlFramer { + private readonly decoder = new TextDecoder('utf-8', { fatal: true }); + private readonly maxLineBytes: number; + private readonly onRecord: (record: unknown) => void; + private buffered = Buffer.alloc(0); + private finished = false; + + constructor(options: StrictLfJsonlFramerOptions) { + const maxLineBytes = options.maxLineBytes ?? DEFAULT_MAX_PI_RPC_LINE_BYTES; + if (!Number.isSafeInteger(maxLineBytes) || maxLineBytes <= 0) { + throw new Error('maxLineBytes must be a positive safe integer'); + } + this.maxLineBytes = maxLineBytes; + this.onRecord = options.onRecord; + } + + push(chunk: Uint8Array | string): void { + if (this.finished) throw protocolError('Pi RPC stdout continued after stream end'); + const bytes = typeof chunk === 'string' + ? Buffer.from(chunk, 'utf8') + : Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength); + let offset = 0; + + while (offset < bytes.length) { + const newline = bytes.indexOf(0x0a, offset); + if (newline === -1) { + this.append(bytes.subarray(offset)); + return; + } + this.append(bytes.subarray(offset, newline)); + this.emitLine(); + offset = newline + 1; + } + } + + finish(): void { + if (this.finished) return; + this.finished = true; + if (this.buffered.length > 0) { + throw protocolError('Pi RPC stdout ended with a partial line'); + } + } + + private append(segment: Uint8Array): void { + if (this.buffered.length + segment.byteLength > this.maxLineBytes) { + throw protocolError(`Pi RPC stdout line exceeded ${this.maxLineBytes} bytes`); + } + if (segment.byteLength === 0) return; + this.buffered = this.buffered.length === 0 + ? Buffer.from(segment) + : Buffer.concat([this.buffered, segment], this.buffered.length + segment.byteLength); + } + + private emitLine(): void { + let line = this.buffered; + this.buffered = Buffer.alloc(0); + if (line.at(-1) === 0x0d) line = line.subarray(0, -1); + if (line.length === 0) throw protocolError('Pi RPC stdout emitted a blank line'); + + let source: string; + try { + source = this.decoder.decode(line); + } catch (error) { + throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi RPC stdout was not valid UTF-8', { + cause: error, + }); + } + + try { + this.onRecord(JSON.parse(source)); + } catch (error) { + if (error instanceof PiProcessError) throw error; + throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi RPC stdout contained malformed JSON', { + cause: error, + }); + } + } +} diff --git a/electron/coding-runtime/pi/worker-process.ts b/electron/coding-runtime/pi/worker-process.ts new file mode 100644 index 0000000..03ac609 --- /dev/null +++ b/electron/coding-runtime/pi/worker-process.ts @@ -0,0 +1,356 @@ +import { + execFile, + spawn, + type ChildProcessWithoutNullStreams, +} from 'node:child_process'; +import { platform } from 'node:os'; +import { PiProcessError } from './process-errors'; +import { + PiRpcClient, + type PiRpcCommand, + type PiRpcEvent, + type PiRpcRequestOptions, + type PiRpcResponse, +} from './rpc-client'; +import { StrictLfJsonlFramer } from './rpc-framer'; + +const DEFAULT_COMMAND_TIMEOUT_MS = 10_000; +const DEFAULT_SHUTDOWN_GRACE_MS = 3_000; +const DEFAULT_DIAGNOSTIC_BYTES = 16_000; +const ANSI_COLOR_PATTERN = new RegExp(`${String.fromCharCode(27)}\\[[0-9;]*m`, 'g'); + +export type PiWorkerStopResult = { + mode: 'not-started' | 'stdin-close' | 'forced-tree-kill'; + code: number | null; + signal: NodeJS.Signals | null; +}; + +export type PiWorkerProcessOptions = { + executablePath: string; + cliPath: string; + cwd: string; + configDir: string; + sessionDir: string; + additionalArgs?: readonly string[]; + env?: NodeJS.ProcessEnv; + sensitiveValues?: readonly string[]; + commandTimeoutMs?: number; + shutdownGraceMs?: number; + maxLineBytes?: number; + diagnosticBytes?: number; +}; + +export function buildPiRpcArgs( + sessionDir: string, + additionalArgs: readonly string[] = [], +): string[] { + return [ + '--mode', 'rpc', + '--offline', + '--session-dir', sessionDir, + '--no-extensions', + '--no-skills', + '--no-prompt-templates', + '--no-themes', + '--no-context-files', + '--no-approve', + '--no-tools', + ...additionalArgs, + ]; +} + +export function sanitizePiDiagnostic( + source: string, + sensitiveValues: readonly string[] = [], +): string { + let sanitized = source + .replace(ANSI_COLOR_PATTERN, '') + .replace(/(authorization\s*[:=]\s*(?:bearer\s+)?)[^\s,;]+/gi, '$1[REDACTED]') + .replace(/((?:x-api-key|api[_-]?key|token|secret)\s*[:=]\s*)[^\s,;]+/gi, '$1[REDACTED]'); + for (const value of sensitiveValues) { + if (value.length < 4) continue; + sanitized = sanitized.split(value).join('[REDACTED]'); + } + return sanitized; +} + +function boundedUtf8Tail(source: string, maxBytes: number): string { + const bytes = Buffer.from(source, 'utf8'); + if (bytes.length <= maxBytes) return source; + const decoder = new TextDecoder('utf-8', { fatal: true }); + for (let start = bytes.length - maxBytes; start < bytes.length; start += 1) { + try { + return decoder.decode(bytes.subarray(start)); + } catch { + // A UTF-8 code point occupies at most four bytes; advance to its boundary. + } + } + return ''; +} + +function positiveInteger(value: number | undefined, fallback: number, name: string): number { + const resolved = value ?? fallback; + if (!Number.isSafeInteger(resolved) || resolved <= 0) { + throw new Error(`${name} must be a positive safe integer`); + } + return resolved; +} + +function runExecutable(executable: string, args: readonly string[]): Promise { + return new Promise((resolve, reject) => { + execFile(executable, [...args], { windowsHide: true }, (error) => { + if (error) reject(error); + else resolve(); + }); + }); +} + +export class PiWorkerProcess { + private readonly options: PiWorkerProcessOptions; + private readonly commandTimeoutMs: number; + private readonly shutdownGraceMs: number; + private readonly diagnosticBytes: number; + private child: ChildProcessWithoutNullStreams | null = null; + private rpc: PiRpcClient | null = null; + private generationValue = 1; + private diagnostic = ''; + private stdoutTail = ''; + private invalidation: PiProcessError | null = null; + private stopping = false; + private stopPromise: Promise | null = null; + private exitResult: Promise<{ code: number | null; signal: NodeJS.Signals | null }> | null = null; + + constructor(options: PiWorkerProcessOptions) { + this.options = options; + this.commandTimeoutMs = positiveInteger( + options.commandTimeoutMs, + DEFAULT_COMMAND_TIMEOUT_MS, + 'commandTimeoutMs', + ); + this.shutdownGraceMs = positiveInteger( + options.shutdownGraceMs, + DEFAULT_SHUTDOWN_GRACE_MS, + 'shutdownGraceMs', + ); + this.diagnosticBytes = positiveInteger( + options.diagnosticBytes, + DEFAULT_DIAGNOSTIC_BYTES, + 'diagnosticBytes', + ); + } + + get generation(): number { + return this.generationValue; + } + + get pendingCommandCount(): number { + return this.rpc?.pendingCount ?? 0; + } + + get stderrDiagnostic(): string { + return this.diagnostic; + } + + get protocolError(): PiProcessError | null { + return this.invalidation?.code === 'PI_RPC_PROTOCOL_ERROR' ? this.invalidation : null; + } + + async start(): Promise { + if (this.child) throw new Error('Pi worker process already started'); + const generation = this.generationValue; + const child = spawn( + this.options.executablePath, + [this.options.cliPath, ...buildPiRpcArgs(this.options.sessionDir, this.options.additionalArgs)], + { + cwd: this.options.cwd, + env: { + ...process.env, + ...this.options.env, + ELECTRON_RUN_AS_NODE: '1', + PI_CODING_AGENT_DIR: this.options.configDir, + PI_OFFLINE: '1', + PI_TELEMETRY: '0', + }, + stdio: ['pipe', 'pipe', 'pipe'], + windowsHide: true, + detached: platform() !== 'win32', + }, + ); + this.child = child; + this.rpc = new PiRpcClient(child.stdin, { + generation, + defaultTimeoutMs: this.commandTimeoutMs, + idPrefix: 'makelore-pi', + onEventListenerError: (error) => { + this.appendDiagnostic(`[event-listener] ${error instanceof Error ? error.message : String(error)}\n`); + }, + }); + + this.exitResult = new Promise((resolve) => { + child.once('exit', (code, signal) => { + resolve({ code, signal }); + }); + }); + child.once('close', (code, signal) => { + if (!this.stopping && !this.invalidation) { + this.invalidate(new PiProcessError( + 'PI_RPC_EXITED', + `Pi worker exited unexpectedly (code=${code ?? 'null'}, signal=${signal ?? 'none'})`, + { generation }, + )); + } + }); + + const framer = new StrictLfJsonlFramer({ + maxLineBytes: this.options.maxLineBytes, + onRecord: (record) => this.acceptRecord(record), + }); + child.stdout.on('data', (chunk: Buffer) => { + this.stdoutTail = boundedUtf8Tail( + `${this.stdoutTail}${sanitizePiDiagnostic(chunk.toString('utf8'), this.options.sensitiveValues)}`, + Math.min(this.diagnosticBytes, 2_048), + ); + try { + framer.push(chunk); + } catch (error) { + this.handleProtocolFailure(error); + } + }); + child.stdout.on('end', () => { + try { + framer.finish(); + } catch (error) { + this.handleProtocolFailure(error); + } + }); + child.stderr.on('data', (chunk: Buffer) => this.appendDiagnostic(chunk.toString('utf8'))); + + try { + await new Promise((resolve, reject) => { + child.once('spawn', resolve); + child.once('error', reject); + }); + } catch (error) { + const failure = new PiProcessError('PI_WORKER_START_FAILED', 'Could not start Pi worker', { + cause: error, + generation, + diagnostic: this.diagnostic, + }); + this.invalidate(failure); + throw failure; + } + return this; + } + + request( + command: PiRpcCommand, + options?: PiRpcRequestOptions, + ): Promise> { + if (!this.rpc) { + return Promise.reject(new PiProcessError( + 'PI_WORKER_START_FAILED', + 'Pi worker has not started', + { generation: this.generationValue }, + )); + } + return this.rpc.request(command, options); + } + + subscribe(listener: (event: PiRpcEvent) => void): () => void { + if (!this.rpc) throw new Error('Pi worker has not started'); + return this.rpc.subscribe(listener); + } + + stop(): Promise { + if (!this.stopPromise) this.stopPromise = this.performStop(); + return this.stopPromise; + } + + private acceptRecord(record: unknown): void { + try { + this.rpc?.accept(record); + } catch (error) { + this.handleProtocolFailure(error); + } + } + + private handleProtocolFailure(error: unknown): void { + if (this.invalidation) return; + const message = error instanceof Error ? error.message : 'Pi RPC protocol failure'; + this.appendDiagnostic(`[stdout-protocol] ${this.stdoutTail}\n`); + const failure = new PiProcessError('PI_RPC_PROTOCOL_ERROR', message, { + cause: error, + generation: this.generationValue, + diagnostic: this.diagnostic, + }); + this.invalidate(failure); + void this.forceKillTree(); + } + + private invalidate(error: PiProcessError): void { + if (this.invalidation) return; + this.invalidation = error; + this.rpc?.invalidate(error); + this.generationValue += 1; + } + + private appendDiagnostic(source: string): void { + const sanitized = sanitizePiDiagnostic(source, this.options.sensitiveValues); + this.diagnostic = boundedUtf8Tail(`${this.diagnostic}${sanitized}`, this.diagnosticBytes); + } + + private async performStop(): Promise { + const child = this.child; + const exitResult = this.exitResult; + if (!child || !exitResult) { + return { mode: 'not-started', code: null, signal: null }; + } + + this.stopping = true; + this.invalidate(new PiProcessError('PI_WORKER_STOPPED', 'Pi worker stopped', { + generation: this.generationValue, + })); + child.stdin.end(); + const graceful = await this.waitForExit(this.shutdownGraceMs); + if (graceful) return { mode: 'stdin-close', ...graceful }; + + await this.forceKillTree(); + const forced = await this.waitForExit(this.shutdownGraceMs); + if (forced) return { mode: 'forced-tree-kill', ...forced }; + throw new PiProcessError('PI_WORKER_STOP_FAILED', 'Pi worker did not exit after forced tree kill', { + generation: this.generationValue, + diagnostic: this.diagnostic, + }); + } + + private async waitForExit( + timeoutMs: number, + ): Promise<{ code: number | null; signal: NodeJS.Signals | null } | null> { + if (!this.exitResult) return null; + let timer: ReturnType | undefined; + try { + return await Promise.race([ + this.exitResult, + new Promise((resolve) => { + timer = setTimeout(() => resolve(null), timeoutMs); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } + } + + private async forceKillTree(): Promise { + const child = this.child; + if (!child?.pid) return; + if (platform() === 'win32') { + await runExecutable('taskkill.exe', ['/pid', String(child.pid), '/t', '/f']).catch(() => undefined); + return; + } + try { + process.kill(-child.pid, 'SIGKILL'); + } catch { + child.kill('SIGKILL'); + } + } +} diff --git a/package.json b/package.json index 823f107..57cfa72 100644 --- a/package.json +++ b/package.json @@ -34,13 +34,14 @@ "init": "pnpm install", "dev": "node scripts/run-cloud-image-workspace-dev.mjs", "dev:image-workspace:local": "node scripts/run-local-image-workspace-dev.mjs", - "build": "pnpm run build:vite && node scripts/bundle-opencode.mjs && node scripts/run-electron-builder.mjs", + "build": "pnpm run build:vite && node scripts/bundle-opencode.mjs && node scripts/bundle-pi-runtime.mjs --release-targets && node scripts/run-electron-builder.mjs", "build:vite": "node --max-old-space-size=6144 ./node_modules/vite/bin/vite.js build", "perf:budget": "node scripts/check-performance-budget.mjs", "probe:pi-runtime": "node scripts/probe-pi-runtime.mjs --stage", "probe:pi-runtime:packaged": "node scripts/probe-pi-packaged-runtime.mjs", "probe:pi-provider-contracts": "node scripts/probe-pi-provider-contracts.mjs", "bundle:opencode": "node scripts/bundle-opencode.mjs", + "bundle:pi-runtime": "node scripts/bundle-pi-runtime.mjs", "lint": "eslint . --fix", "lint:check": "eslint .", "typecheck": "tsc --noEmit", @@ -66,8 +67,8 @@ "node:download:win": "zx scripts/download-bundled-node.mjs --platform=win", "prep:win-binaries": "pnpm run uv:download:win && pnpm run node:download:win", "icons": "zx scripts/generate-icons.mjs", - "package": "pnpm run build:vite && node scripts/bundle-opencode.mjs", - "package:stage:win-x64": "pnpm run build:vite && node scripts/bundle-opencode.mjs --platform=win32 --arch=x64", + "package": "pnpm run build:vite && node scripts/bundle-opencode.mjs && node scripts/bundle-pi-runtime.mjs --release-targets", + "package:stage:win-x64": "pnpm run build:vite && node scripts/bundle-opencode.mjs --platform=win32 --arch=x64 && node scripts/bundle-pi-runtime.mjs --target win32-x64", "package:mac": "pnpm run python:download:mac && pnpm run uv:download:mac && pnpm run package && node scripts/run-electron-builder.mjs --mac --publish never", "package:mac:local": "pnpm run python:download:mac && pnpm run uv:download:mac && SKIP_PREINSTALLED_SKILLS=1 pnpm run package && node scripts/run-electron-builder.mjs --mac --publish never", "package:win": "pnpm run python:download:win && pnpm run uv:download:win && pnpm run package:stage:win-x64 && node scripts/run-electron-builder.mjs --win --publish never", diff --git a/scripts/bundle-pi-runtime.mjs b/scripts/bundle-pi-runtime.mjs new file mode 100644 index 0000000..e322e3e --- /dev/null +++ b/scripts/bundle-pi-runtime.mjs @@ -0,0 +1,123 @@ +import { realpathSync } from 'node:fs'; +import { arch, platform } from 'node:os'; +import { join, relative, resolve } from 'node:path'; +import { pathToFileURL } from 'node:url'; +import { + PI_RUNTIME_PACKAGE, + defaultPiBundleTarget, + stagePiRuntimeBundle, +} from './lib/pi-runtime-bundle.mjs'; + +function parseTarget(value) { + const separator = value.lastIndexOf('-'); + if (separator <= 0 || separator === value.length - 1) { + throw new Error(`Invalid Pi runtime target: ${value}`); + } + const target = { + platform: value.slice(0, separator), + arch: value.slice(separator + 1), + }; + if (!['win32', 'darwin', 'linux'].includes(target.platform)) { + throw new Error(`Unsupported Pi runtime platform: ${target.platform}`); + } + if (!['x64', 'arm64'].includes(target.arch)) { + throw new Error(`Unsupported Pi runtime architecture: ${target.arch}`); + } + return target.platform === 'linux' ? { ...target, libc: 'glibc' } : target; +} + +export function parseBundleArgs(argv) { + const options = { outputRoot: resolve('build/pi-runtime'), targets: [] }; + for (let index = 0; index < argv.length; index += 1) { + const argument = argv[index]; + if (argument === '--output') { + const value = argv[index + 1]; + if (!value || value.startsWith('--')) throw new Error('--output requires a value'); + options.outputRoot = resolve(value); + index += 1; + } + else if (argument === '--target') { + const value = argv[index + 1]; + if (!value || value.startsWith('--')) throw new Error('--target requires a value'); + options.targets.push(parseTarget(value)); + index += 1; + } + else if (argument === '--release-targets') { + const currentPlatform = platform(); + const arches = currentPlatform === 'win32' ? ['x64'] : ['x64', 'arm64']; + options.targets.push(...arches.map((targetArch) => parseTarget(`${currentPlatform}-${targetArch}`))); + } + else if (argument === '--help') options.help = true; + else throw new Error(`Unknown argument: ${argument}`); + } + if (options.targets.length === 0) options.targets.push(defaultPiBundleTarget()); + + const unique = new Map(options.targets.map((target) => [ + `${target.platform}-${target.arch}`, + target, + ])); + options.targets = [...unique.values()]; + for (const target of options.targets) { + if (target.platform !== platform()) { + throw new Error( + `Pi runtime ${target.platform}-${target.arch} must be staged on ${target.platform}, current host is ${platform()}-${arch()}`, + ); + } + } + return options; +} + +function printHelp() { + process.stdout.write('Usage: node scripts/bundle-pi-runtime.mjs [options]\n\n'); + process.stdout.write(' --target Stage one runtime target (repeatable)\n'); + process.stdout.write(' --release-targets Stage every configured architecture for the host OS\n'); + process.stdout.write(' --output Output root (default: build/pi-runtime)\n'); +} + +export async function bundlePiRuntime(options, projectRoot = process.cwd()) { + const resolvedProjectRoot = resolve(projectRoot); + const sourcePackageRoot = realpathSync(join( + resolvedProjectRoot, + 'node_modules', + ...PI_RUNTIME_PACKAGE.split('/'), + )); + const results = []; + for (const target of options.targets) { + const destination = join(options.outputRoot, `${target.platform}-${target.arch}`); + const destinationRelative = relative(options.outputRoot, destination); + if (destinationRelative.startsWith('..') || destinationRelative === '') { + throw new Error(`Unsafe Pi runtime destination: ${destination}`); + } + const manifest = await stagePiRuntimeBundle({ + projectRoot: resolvedProjectRoot, + sourcePackageRoot, + destination, + target, + }); + results.push({ destination, manifest }); + } + return results; +} + +async function main() { + const options = parseBundleArgs(process.argv.slice(2)); + if (options.help) { + printHelp(); + return; + } + const results = await bundlePiRuntime(options); + process.stdout.write(`${JSON.stringify(results.map(({ destination, manifest }) => ({ + destination, + target: manifest.target, + packageCount: manifest.productionPackages.length, + assetCount: manifest.runtimeAssets.length, + })), null, 2)}\n`); +} + +const isMain = process.argv[1] && pathToFileURL(resolve(process.argv[1])).href === import.meta.url; +if (isMain) { + main().catch((error) => { + process.stderr.write(`${error.stack ?? error.message}\n`); + process.exitCode = 1; + }); +} diff --git a/scripts/lib/pi-runtime-bundle.mjs b/scripts/lib/pi-runtime-bundle.mjs new file mode 100644 index 0000000..c6caa25 --- /dev/null +++ b/scripts/lib/pi-runtime-bundle.mjs @@ -0,0 +1,301 @@ +import { spawn } from 'node:child_process'; +import { cp, mkdir, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises'; +import { realpathSync } from 'node:fs'; +import { createRequire } from 'node:module'; +import { arch as hostArch, platform as hostPlatform } from 'node:os'; +import { dirname, join, relative, resolve, sep } from 'node:path'; + +export const PI_RUNTIME_PACKAGE = '@earendil-works/pi-coding-agent'; +export const PI_RUNTIME_VERSION = '0.84.2'; +export const PI_RUNTIME_CLI_ENTRY = 'dist/cli.js'; +export const PI_RUNTIME_MANIFEST = 'makelore-pi-runtime.json'; +export const PI_RUNTIME_STAGING_NPM_VERSION = '11.6.2'; + +function readJson(path) { + return readFile(path, 'utf8').then((source) => JSON.parse(source)); +} + +async function pathExists(path) { + try { + await stat(path); + return true; + } catch { + return false; + } +} + +function portable(path) { + return path.split(sep).join('/'); +} + +function constraintMatches(constraints, current) { + if (!constraints || constraints.length === 0 || !current) return true; + const positive = constraints.filter((value) => !value.startsWith('!')); + const negative = constraints.filter((value) => value.startsWith('!')).map((value) => value.slice(1)); + return !negative.includes(current) && (positive.length === 0 || positive.includes(current)); +} + +export function lockEntrySupportsTarget(entry, target) { + return constraintMatches(entry.os, target.platform) + && constraintMatches(entry.cpu, target.arch) + && constraintMatches(entry.libc, target.libc); +} + +function packageNameFromLockPath(packagePath) { + const marker = packagePath.lastIndexOf('node_modules/'); + if (marker === -1) return null; + const remainder = packagePath.slice(marker + 'node_modules/'.length).split('/'); + return remainder[0]?.startsWith('@') ? remainder.slice(0, 2).join('/') : remainder[0]; +} + +function packageIdentity(name, version) { + return `${name}@${version}`; +} + +export function requiredPackageIdentities(shrinkwrap, target) { + const identities = new Set(); + let platformSkippedPackages = 0; + for (const [packagePath, entry] of Object.entries(shrinkwrap.packages ?? {})) { + if (!packagePath || entry.dev) continue; + if (!lockEntrySupportsTarget(entry, target)) { + platformSkippedPackages += 1; + continue; + } + const name = entry.name ?? packageNameFromLockPath(packagePath); + if (!name || !entry.version) { + throw new Error(`Pi shrinkwrap package identity is incomplete at ${packagePath}`); + } + identities.add(packageIdentity(name, entry.version)); + } + return { + identities: [...identities].sort(), + platformSkippedPackages, + }; +} + +async function visitInstalledPackages(nodeModulesPath, identities) { + if (!await pathExists(nodeModulesPath)) return; + for (const entry of await readdir(nodeModulesPath, { withFileTypes: true })) { + if (!entry.isDirectory() || entry.name === '.bin') continue; + if (entry.name.startsWith('@')) { + const scopePath = join(nodeModulesPath, entry.name); + for (const scopedEntry of await readdir(scopePath, { withFileTypes: true })) { + if (scopedEntry.isDirectory()) { + await visitInstalledPackage(join(scopePath, scopedEntry.name), identities); + } + } + continue; + } + await visitInstalledPackage(join(nodeModulesPath, entry.name), identities); + } +} + +async function visitInstalledPackage(packageRoot, identities) { + const packagePath = join(packageRoot, 'package.json'); + if (!await pathExists(packagePath)) return; + const packageJson = await readJson(packagePath); + if (packageJson.name && packageJson.version) { + identities.add(packageIdentity(packageJson.name, packageJson.version)); + } + await visitInstalledPackages(join(packageRoot, 'node_modules'), identities); +} + +export async function installedPackageIdentities(stageRoot) { + const identities = new Set(); + const rootPackage = await readJson(join(stageRoot, 'package.json')); + identities.add(packageIdentity(rootPackage.name, rootPackage.version)); + await visitInstalledPackages(join(stageRoot, 'node_modules'), identities); + return [...identities].sort(); +} + +export function assertPackageIdentities(required, installed) { + const installedSet = new Set(installed); + const missing = required.filter((identity) => !installedSet.has(identity)); + if (missing.length > 0) { + throw new Error(`Pi production closure is missing package identities: ${missing.join(', ')}`); + } +} + +async function collectRuntimeAssets(stageRoot) { + const assets = []; + const visit = async (directory) => { + if (!await pathExists(directory)) return; + for (const entry of await readdir(directory, { withFileTypes: true })) { + const path = join(directory, entry.name); + if (entry.isDirectory()) await visit(path); + else if (entry.name.endsWith('.node') || entry.name.endsWith('.wasm')) { + assets.push(portable(relative(stageRoot, path))); + } + } + }; + await visit(join(stageRoot, 'dist')); + await visit(join(stageRoot, 'node_modules')); + return assets.sort(); +} + +export async function preparePublishedPackageRoot(sourcePackageRoot, destination) { + await mkdir(dirname(destination), { recursive: true }); + await rm(destination, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }); + await cp(sourcePackageRoot, destination, { + recursive: true, + filter(source) { + return relative(sourcePackageRoot, source).split(sep)[0] !== 'node_modules'; + }, + }); + + const packagePath = join(destination, 'package.json'); + const packageJson = await readJson(packagePath); + const omittedDevDependencies = Object.keys(packageJson.devDependencies ?? {}).sort(); + delete packageJson.devDependencies; + await writeFile(packagePath, `${JSON.stringify(packageJson, null, 2)}\n`); + return { omittedDevDependencies }; +} + +function runCommand(executable, args, options) { + return new Promise((resolvePromise, reject) => { + const child = spawn(executable, args, { + cwd: options.cwd, + env: options.env, + stdio: ['ignore', 'pipe', 'pipe'], + windowsHide: true, + }); + let stdout = ''; + let stderr = ''; + child.stdout.on('data', (chunk) => { stdout += chunk.toString(); }); + child.stderr.on('data', (chunk) => { stderr += chunk.toString(); }); + child.once('error', reject); + child.once('exit', (code, signal) => { + if (code === 0) resolvePromise({ stdout, stderr }); + else { + reject(new Error( + `${executable} exited with code ${code ?? 'null'} signal ${signal ?? 'none'}: ${stderr || stdout}`, + )); + } + }); + }); +} + +export function resolvePinnedNpmCli(projectRoot) { + const requireFromProject = createRequire(join(resolve(projectRoot), 'package.json')); + return join(realpathSync(dirname(requireFromProject.resolve('npm/package.json'))), 'bin', 'npm-cli.js'); +} + +export async function validatePinnedNpm(projectRoot) { + const resolvedRoot = resolve(projectRoot); + const requireFromProject = createRequire(join(resolvedRoot, 'package.json')); + const projectPackage = await readJson(join(resolvedRoot, 'package.json')); + const npmPackagePath = requireFromProject.resolve('npm/package.json'); + const npmPackage = await readJson(npmPackagePath); + if (projectPackage.dependencies?.npm !== PI_RUNTIME_STAGING_NPM_VERSION) { + throw new Error(`Root npm dependency must be exactly ${PI_RUNTIME_STAGING_NPM_VERSION}`); + } + if (npmPackage.version !== PI_RUNTIME_STAGING_NPM_VERSION) { + throw new Error( + `Installed npm must be ${PI_RUNTIME_STAGING_NPM_VERSION}, got ${npmPackage.version ?? 'missing'}`, + ); + } + return npmPackagePath; +} + +export async function installProductionShrinkwrap({ + projectRoot, + stageRoot, + target, + npmCli = resolvePinnedNpmCli(projectRoot), +}) { + await validatePinnedNpm(projectRoot); + return await runCommand(process.execPath, [ + npmCli, + 'ci', + '--omit=dev', + '--ignore-scripts', + '--no-audit', + '--no-fund', + `--os=${target.platform}`, + `--cpu=${target.arch}`, + ], { + cwd: stageRoot, + env: { + ...process.env, + npm_config_update_notifier: 'false', + }, + }); +} + +export async function createPiRuntimeManifest(stageRoot, target, omittedDevDependencies) { + const rootPackage = await readJson(join(stageRoot, 'package.json')); + const shrinkwrap = await readJson(join(stageRoot, 'npm-shrinkwrap.json')); + if (rootPackage.name !== PI_RUNTIME_PACKAGE || rootPackage.version !== PI_RUNTIME_VERSION) { + throw new Error( + `Pi runtime must be ${PI_RUNTIME_PACKAGE}@${PI_RUNTIME_VERSION}, got ${rootPackage.name}@${rootPackage.version}`, + ); + } + if (rootPackage.bin?.pi !== PI_RUNTIME_CLI_ENTRY) { + throw new Error(`Pi CLI entry must be ${PI_RUNTIME_CLI_ENTRY}`); + } + if (!await pathExists(join(stageRoot, ...PI_RUNTIME_CLI_ENTRY.split('/')))) { + throw new Error(`Pi CLI is missing at ${PI_RUNTIME_CLI_ENTRY}`); + } + + const required = requiredPackageIdentities(shrinkwrap, target); + const installed = await installedPackageIdentities(stageRoot); + assertPackageIdentities(required.identities, installed); + const assets = await collectRuntimeAssets(stageRoot); + const requiredPhotonAsset = 'node_modules/@silvia-odwyer/photon-node/photon_rs_bg.wasm'; + if (!assets.includes(requiredPhotonAsset)) { + throw new Error(`Pi runtime asset is missing: ${requiredPhotonAsset}`); + } + + return { + schemaVersion: 1, + runtime: { + packageName: PI_RUNTIME_PACKAGE, + version: PI_RUNTIME_VERSION, + cliEntry: PI_RUNTIME_CLI_ENTRY, + nodeEngine: rootPackage.engines?.node ?? null, + }, + target: { + platform: target.platform, + arch: target.arch, + ...(target.libc ? { libc: target.libc } : {}), + }, + lockfileVersion: shrinkwrap.lockfileVersion, + productionPackages: required.identities, + platformSkippedPackages: required.platformSkippedPackages, + runtimeAssets: assets, + staging: { + npmVersion: PI_RUNTIME_STAGING_NPM_VERSION, + omitDev: true, + ignoreScripts: true, + omittedPublishedDevDependencies: [...omittedDevDependencies].sort(), + }, + }; +} + +export async function stagePiRuntimeBundle({ + projectRoot, + sourcePackageRoot, + destination, + target, +}) { + const prepared = await preparePublishedPackageRoot(sourcePackageRoot, destination); + await installProductionShrinkwrap({ projectRoot, stageRoot: destination, target }); + const manifest = await createPiRuntimeManifest( + destination, + target, + prepared.omittedDevDependencies, + ); + await writeFile( + join(destination, PI_RUNTIME_MANIFEST), + `${JSON.stringify(manifest, null, 2)}\n`, + ); + return manifest; +} + +export function defaultPiBundleTarget() { + return { + platform: hostPlatform(), + arch: hostArch(), + ...(hostPlatform() === 'linux' ? { libc: 'glibc' } : {}), + }; +} diff --git a/scripts/probe-pi-packaged-runtime.mjs b/scripts/probe-pi-packaged-runtime.mjs index 2697f2e..118081e 100644 --- a/scripts/probe-pi-packaged-runtime.mjs +++ b/scripts/probe-pi-packaged-runtime.mjs @@ -8,13 +8,16 @@ import { parseProbeArgs, resolvePiRuntime, runProbe, - stagePiRuntime, validatePiIdentity, - validateStagedClosure, } from './probe-pi-runtime.mjs'; +import { + PI_RUNTIME_MANIFEST, + defaultPiBundleTarget, + stagePiRuntimeBundle, +} from './lib/pi-runtime-bundle.mjs'; const PRODUCT_NAME = 'MakelorePiProbe'; -const ARTIFACT_LABEL = 'controlled-electron-builder-dir-app-asar'; +const ARTIFACT_LABEL = 'controlled-electron-builder-dir-extra-resources'; async function pathExists(path) { try { @@ -92,9 +95,12 @@ async function inspectPackagedClosure(executable, resourcesDirectory, assets) { const path = require('node:path'); const resources = process.env.PI_PROBE_RESOURCES_DIRECTORY; if (!resources) throw new Error('PI_PROBE_RESOURCES_DIRECTORY is required'); - const root = path.join(resources, 'app.asar'); - const unpackedRoot = path.join(resources, 'app.asar.unpacked'); + const root = path.join(resources, 'pi-runtime'); const packageJson = JSON.parse(fs.readFileSync(path.join(root, 'package.json'), 'utf8')); + const runtimeManifestPath = path.join(root, '${PI_RUNTIME_MANIFEST}'); + const runtimeManifest = fs.existsSync(runtimeManifestPath) + ? JSON.parse(fs.readFileSync(runtimeManifestPath, 'utf8')) + : null; const shrinkwrap = JSON.parse(fs.readFileSync(path.join(root, 'npm-shrinkwrap.json'), 'utf8')); const assets = JSON.parse(process.env.PI_PROBE_ASSETS_JSON); const libc = process.platform === 'linux' @@ -140,12 +146,11 @@ async function inspectPackagedClosure(executable, resourcesDirectory, assets) { missingPackages.push(packagePath); } const missingAssets = assets.filter((asset) => !fs.existsSync(path.join(root, ...asset.split('/')))); - const missingUnpackedNativeAssets = assets - .filter((asset) => asset.endsWith('.node')) - .filter((asset) => !fs.existsSync(path.join(unpackedRoot, ...asset.split('/')))); process.stdout.write(JSON.stringify({ packageName: packageJson.name, packageVersion: packageJson.version, + runtimeManifestExists: Boolean(runtimeManifest), + runtimeManifestTarget: runtimeManifest?.target ?? null, cliExists: fs.existsSync(path.join(root, 'dist', 'cli.js')), lockedPackages: Math.max(0, Object.keys(shrinkwrap.packages || {}).length - 1), expectedPackages, @@ -154,7 +159,7 @@ async function inspectPackagedClosure(executable, resourcesDirectory, assets) { relocatedPackages, assetCount: assets.length, missingAssets, - missingUnpackedNativeAssets, + nativeAssetsOutsideAsar: assets.filter((asset) => asset.endsWith('.node')).length, })); `; const { stdout } = await runCommand(executable, ['-e', script], { @@ -168,9 +173,9 @@ async function inspectPackagedClosure(executable, resourcesDirectory, assets) { const result = JSON.parse(stdout.trim()); if ( !result.cliExists + || !result.runtimeManifestExists || result.missingPackages.length > 0 || result.missingAssets.length > 0 - || result.missingUnpackedNativeAssets.length > 0 ) { throw new Error(`Packaged Pi closure is incomplete: ${JSON.stringify(result)}`); } @@ -187,14 +192,20 @@ function assertControlledOutput(projectRoot, outputDirectory) { } } -async function buildControlledArtifact(projectRoot, appDirectory, outputDirectory) { +async function buildControlledArtifact(projectRoot, appDirectory, runtimeDirectory, outputDirectory) { const electronPackage = JSON.parse(await readFile(join(projectRoot, 'node_modules', 'electron', 'package.json'), 'utf8')); const { platformTarget, architecture } = currentTarget(); const buildResources = join(appDirectory, 'build-resources'); - await mkdir(buildResources); + await mkdir(buildResources, { recursive: true }); + await writeFile(join(appDirectory, 'package.json'), `${JSON.stringify({ + name: 'makelore-pi-production-shape-probe', + version: '1.0.0', + main: 'main.js', + }, null, 2)}\n`); + await writeFile(join(appDirectory, 'main.js'), 'process.exit(0);\n'); const builderArtifacts = await build({ - // Use the staged Pi root as the project root so this controlled probe does - // not inherit Makelore's production hooks or unrelated extraResources. + // Keep the probe application minimal while copying the permanent stage to + // the same resources/pi-runtime location used by the product configuration. projectDir: appDirectory, targets: platformTarget.createTarget('dir', architecture), publish: 'never', @@ -208,13 +219,22 @@ async function buildControlledArtifact(projectRoot, appDirectory, outputDirector buildResources, }, files: [ - 'dist/**/*', + 'main.js', 'package.json', - 'npm-shrinkwrap.json', - 'node_modules/**/*', + ], + extraResources: [ + { + from: runtimeDirectory, + to: 'pi-runtime', + filter: ['**/*', '!node_modules{,/**/*}'], + }, + { + from: join(runtimeDirectory, 'node_modules'), + to: 'pi-runtime/node_modules', + filter: ['**/*'], + }, ], asar: true, - asarUnpack: ['**/*.node'], npmRebuild: false, win: { executableName: PRODUCT_NAME, @@ -239,22 +259,37 @@ export async function runPackagedProbe(options, projectRoot = process.cwd()) { assertControlledOutput(resolvedProjectRoot, outputDirectory); const scratchRoot = await mkdtemp(join(tmpdir(), 'makelore-pi-packaged-probe-')); const appDirectory = join(scratchRoot, 'app'); + const runtimeDirectory = join(scratchRoot, 'runtime'); try { const sourceRuntime = await resolvePiRuntime(resolvedProjectRoot); const identity = await validatePiIdentity(sourceRuntime); - await stagePiRuntime(sourceRuntime, appDirectory); - const stagedClosure = await validateStagedClosure(appDirectory); + const runtimeManifest = await stagePiRuntimeBundle({ + projectRoot: resolvedProjectRoot, + sourcePackageRoot: sourceRuntime.packageRoot, + destination: runtimeDirectory, + target: defaultPiBundleTarget(), + }); + const stagedClosure = { + lockfileVersion: runtimeManifest.lockfileVersion, + expectedPackages: runtimeManifest.productionPackages.length, + platformSkippedPackages: runtimeManifest.platformSkippedPackages, + assets: runtimeManifest.runtimeAssets, + omittedPublishedDevDependencies: runtimeManifest.staging.omittedPublishedDevDependencies, + manifest: runtimeManifest, + }; await rm(outputDirectory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }); await mkdir(dirname(outputDirectory), { recursive: true }); const builderArtifacts = await buildControlledArtifact( resolvedProjectRoot, appDirectory, + runtimeDirectory, outputDirectory, ); const executable = await findPackagedExecutable(outputDirectory); const resourcesDirectory = packagedResourcesDirectory(executable); const appAsar = join(resourcesDirectory, 'app.asar'); - const cliPath = join(appAsar, 'dist', 'cli.js'); + const packagedRuntime = join(resourcesDirectory, 'pi-runtime'); + const cliPath = join(packagedRuntime, 'dist', 'cli.js'); if (!await pathExists(appAsar)) throw new Error(`Packaged app.asar is missing: ${appAsar}`); const packagedClosure = await inspectPackagedClosure( executable, @@ -279,7 +314,7 @@ export async function runPackagedProbe(options, projectRoot = process.cwd()) { outputDirectory, executable, appAsar, - appAsarUnpacked: join(resourcesDirectory, 'app.asar.unpacked'), + runtimeRoot: packagedRuntime, builderArtifacts, }, stagedClosure, @@ -317,6 +352,6 @@ const isMain = process.argv[1] if (isMain) { main().catch((error) => { process.stderr.write(`${error.stack ?? error.message}\n`); - process.exitCode = 1; + process.exit(1); }); } diff --git a/tests/fixtures/fake-pi-rpc-child.mjs b/tests/fixtures/fake-pi-rpc-child.mjs new file mode 100644 index 0000000..e8ed268 --- /dev/null +++ b/tests/fixtures/fake-pi-rpc-child.mjs @@ -0,0 +1,128 @@ +import { spawn } from 'node:child_process'; + +let input = ''; +let held = null; +let keepAlive = null; +let descendant = null; +const counts = Object.create(null); + +function response(command, data = {}) { + return JSON.stringify({ + type: 'response', + id: command.id, + command: command.type, + success: true, + data, + }); +} + +function writeRecord(record) { + process.stdout.write(`${typeof record === 'string' ? record : JSON.stringify(record)}\n`); +} + +function handle(command) { + counts[command.type] = (counts[command.type] ?? 0) + 1; + if (command.type === 'echo') { + writeRecord(response(command, { value: command.value })); + return; + } + if (command.type === 'emit_event') { + writeRecord({ type: 'agent_start', marker: command.marker }); + writeRecord(response(command, { emitted: true })); + return; + } + if (command.type === 'hold') { + held = command; + return; + } + if (command.type === 'release') { + writeRecord(response(command, { order: 'first' })); + if (held) { + writeRecord(response(held, { order: 'second' })); + held = null; + } + return; + } + if (command.type === 'partial') { + const record = `${response(command, { partial: true })}\n`; + process.stdout.write(record.slice(0, 7)); + setTimeout(() => process.stdout.write(record.slice(7, 19)), 5); + setTimeout(() => process.stdout.write(record.slice(19)), 10); + return; + } + if (command.type === 'malformed') { + process.stdout.write('not-json\n'); + return; + } + if (command.type === 'blank') { + process.stdout.write('\n'); + return; + } + if (command.type === 'invalid_utf8') { + process.stdout.write(Buffer.from([0xff, 0x0a])); + return; + } + if (command.type === 'trailing_partial') { + process.stdout.write(response(command, { partial: true })); + setTimeout(() => process.exit(0), 5); + return; + } + if (command.type === 'large') { + writeRecord(response(command, { value: 'x'.repeat(command.bytes ?? 4096) })); + return; + } + if (command.type === 'get_state' && command.fakeRetry) { + if (counts.get_state > 1) writeRecord(response(command, { attempts: counts.get_state })); + return; + } + if (command.type === 'no_response' || command.type === 'prompt') return; + if (command.type === 'stats') { + writeRecord(response(command, { counts })); + return; + } + if (command.type === 'stderr_secret') { + const secret = process.env.FAKE_PI_SECRET ?? 'missing-secret'; + process.stderr.write(`${'diagnostic '.repeat(50)}\n`); + process.stderr.write(`Authorization: Bearer ${secret}\n`); + process.stderr.write(`token=${secret}\n`); + writeRecord(response(command, { wrote: true })); + return; + } + if (command.type === 'spawn_descendant') { + descendant = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { + stdio: 'ignore', + windowsHide: true, + }); + keepAlive = setInterval(() => {}, 1000); + writeRecord(response(command, { pid: descendant.pid })); + return; + } + if (command.type === 'crash') { + process.exit(7); + } + writeRecord({ + type: 'response', + id: command.id, + command: command.type, + success: false, + error: 'unsupported fake command', + }); +} + +process.stdin.setEncoding('utf8'); +process.stdin.on('data', (chunk) => { + input += chunk; + while (true) { + const newline = input.indexOf('\n'); + if (newline === -1) break; + const line = input.slice(0, newline).replace(/\r$/, ''); + input = input.slice(newline + 1); + if (line) handle(JSON.parse(line)); + } +}); +process.stdin.on('end', () => { + if (!keepAlive) process.exit(0); +}); +process.on('exit', () => { + if (keepAlive) clearInterval(keepAlive); +}); diff --git a/tests/unit/pi-rpc-foundation.test.ts b/tests/unit/pi-rpc-foundation.test.ts new file mode 100644 index 0000000..a82d98b --- /dev/null +++ b/tests/unit/pi-rpc-foundation.test.ts @@ -0,0 +1,271 @@ +// @vitest-environment node + +import { afterEach, describe, expect, it } from 'vitest'; +import { mkdir, mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { Writable } from 'node:stream'; +import { PiProcessError } from '../../electron/coding-runtime/pi/process-errors'; +import { PiRpcClient } from '../../electron/coding-runtime/pi/rpc-client'; +import { StrictLfJsonlFramer } from '../../electron/coding-runtime/pi/rpc-framer'; +import { + PiWorkerProcess, + buildPiRpcArgs, + sanitizePiDiagnostic, +} from '../../electron/coding-runtime/pi/worker-process'; + +const fakeChildPath = resolve('tests/fixtures/fake-pi-rpc-child.mjs'); +const scratchRoots: string[] = []; +const workers: PiWorkerProcess[] = []; + +async function makeWorker( + options: Partial[0]> = {}, +): Promise { + const root = await mkdtemp(join(tmpdir(), 'makelore-pi-rpc-test-')); + 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 worker = new PiWorkerProcess({ + executablePath: process.execPath, + cliPath: fakeChildPath, + cwd, + configDir, + sessionDir, + commandTimeoutMs: 1_000, + shutdownGraceMs: 500, + ...options, + }); + workers.push(worker); + return await worker.start(); +} + +async function processAlive(pid: number): Promise { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +afterEach(async () => { + await Promise.all(workers.splice(0).map((worker) => worker.stop().catch(() => undefined))); + await Promise.all(scratchRoots.splice(0).map((root) => rm(root, { + recursive: true, + force: true, + maxRetries: 3, + }))); +}); + +describe('strict Pi LF JSONL framing', () => { + it('handles chunk boundaries, multiple records, CRLF, and Unicode separators', () => { + const records: unknown[] = []; + const framer = new StrictLfJsonlFramer({ onRecord: (record) => records.push(record) }); + const source = Buffer.from( + `${JSON.stringify({ text: 'left\u2028middle\u2029right' })}\n${JSON.stringify({ ok: true })}\r\n`, + ); + + framer.push(source.subarray(0, 8)); + framer.push(source.subarray(8, 23)); + framer.push(source.subarray(23)); + framer.finish(); + + expect(records).toEqual([ + { text: 'left\u2028middle\u2029right' }, + { ok: true }, + ]); + }); + + it('fails closed for malformed, blank, invalid UTF-8, oversized, and partial records', () => { + const make = (maxLineBytes = 64) => new StrictLfJsonlFramer({ + maxLineBytes, + onRecord: () => undefined, + }); + + expect(() => make().push('not-json\n')).toThrow(/malformed JSON/); + expect(() => make().push('\n')).toThrow(/blank line/); + expect(() => make().push(Buffer.from([0xff, 0x0a]))).toThrow(/valid UTF-8/); + expect(() => make(4).push('12345')).toThrow(/exceeded 4 bytes/); + const partial = make(); + partial.push('{"ok":true}'); + expect(() => partial.finish()).toThrow(/partial line/); + }); +}); + +describe('Pi RPC client', () => { + it('waits for writable completion when the stream applies backpressure', async () => { + let written = ''; + let flush: (() => void) | undefined; + const writable = new Writable({ + highWaterMark: 1, + write(chunk, _encoding, callback) { + written += chunk.toString(); + flush = callback; + }, + }); + const client = new PiRpcClient(writable, { generation: 3, defaultTimeoutMs: 500 }); + let settled = false; + const requested = client.request({ type: 'get_state' }).then((response) => { + settled = true; + return response; + }); + await new Promise((resolvePromise) => setImmediate(resolvePromise)); + const command = JSON.parse(written) as { id: string }; + client.accept({ type: 'response', id: command.id, success: true, data: { ready: true } }); + await new Promise((resolvePromise) => setImmediate(resolvePromise)); + expect(settled).toBe(false); + flush?.(); + + await expect(requested).resolves.toMatchObject({ data: { ready: true } }); + }); + + it('rejects retry policy for prompt-like mutation commands', async () => { + const writable = new Writable({ write(_chunk, _encoding, callback) { callback(); } }); + const client = new PiRpcClient(writable, { generation: 1 }); + + await expect(client.request( + { type: 'prompt', message: 'do not replay' }, + { retry: 'read-only-once' }, + )).rejects.toThrow(/not a retryable read-only command/); + await expect(client.request( + { type: 'set_model', provider: 'provider-a', modelId: 'model-a' }, + { retry: 'read-only-once' }, + )).rejects.toThrow(/not a retryable read-only command/); + }); + + it('settles and removes the pending command when the writable fails', async () => { + const writable = new Writable({ + write(_chunk, _encoding, callback) { + callback(new Error('closed pipe')); + }, + }); + const client = new PiRpcClient(writable, { generation: 1 }); + + await expect(client.request({ type: 'get_state' })) + .rejects.toMatchObject({ code: 'PI_RPC_WRITE_FAILED' }); + expect(client.pendingCount).toBe(0); + }); +}); + +describe('Pi worker process', () => { + it('uses the locked offline and no-discovery RPC arguments', () => { + expect(buildPiRpcArgs('sessions', ['--model', 'model-a'])).toEqual([ + '--mode', 'rpc', + '--offline', + '--session-dir', 'sessions', + '--no-extensions', + '--no-skills', + '--no-prompt-templates', + '--no-themes', + '--no-context-files', + '--no-approve', + '--no-tools', + '--model', 'model-a', + ]); + }); + + it('correlates out-of-order responses, dispatches events, and reassembles partial lines', async () => { + const worker = await makeWorker(); + const events: unknown[] = []; + worker.subscribe(() => { throw new Error('consumer failed'); }); + worker.subscribe((event) => events.push(event)); + const held = worker.request<{ order: string }>({ type: 'hold' }); + const released = worker.request<{ order: string }>({ type: 'release' }); + + await expect(released).resolves.toMatchObject({ data: { order: 'first' } }); + await expect(held).resolves.toMatchObject({ data: { order: 'second' } }); + await expect(worker.request({ type: 'emit_event', marker: 'event-a' })) + .resolves.toMatchObject({ data: { emitted: true } }); + await expect(worker.request({ type: 'partial' })) + .resolves.toMatchObject({ data: { partial: true } }); + expect(events).toContainEqual({ type: 'agent_start', marker: 'event-a' }); + expect(worker.stderrDiagnostic).toContain('[event-listener] consumer failed'); + await expect(worker.request({ type: 'echo', value: 'listener-isolated' })) + .resolves.toMatchObject({ data: { value: 'listener-isolated' } }); + }); + + it('settles timeout and abort without replaying prompt, while read-only retry runs once', async () => { + const worker = await makeWorker({ commandTimeoutMs: 500 }); + await expect(worker.request( + { type: 'get_state', fakeRetry: true }, + { retry: 'read-only-once', timeoutMs: 500 }, + )).resolves.toMatchObject({ data: { attempts: 2 } }); + + await expect(worker.request( + { type: 'prompt', message: 'one attempt' }, + { timeoutMs: 100 }, + )).rejects.toMatchObject({ code: 'PI_RPC_TIMEOUT' }); + const stats = await worker.request<{ counts: Record }>({ type: 'stats' }); + expect(stats.data?.counts.prompt).toBe(1); + + const controller = new AbortController(); + const pending = worker.request({ type: 'no_response' }, { signal: controller.signal }); + const aborted = expect(pending).rejects.toMatchObject({ code: 'PI_RPC_ABORTED' }); + controller.abort(); + await aborted; + expect(worker.pendingCommandCount).toBe(0); + }); + + it.each(['malformed', 'blank', 'invalid_utf8', 'trailing_partial', 'large'])( + 'invalidates only the target worker for %s stdout', + async (failureType) => { + const left = await makeWorker({ maxLineBytes: 128 }); + const right = await makeWorker({ maxLineBytes: 128 }); + const command = failureType === 'large' + ? { type: failureType, bytes: 512 } + : { type: failureType }; + + await expect(left.request(command)).rejects.toMatchObject({ code: 'PI_RPC_PROTOCOL_ERROR' }); + expect(left.generation).toBe(2); + expect(left.protocolError).toBeInstanceOf(PiProcessError); + expect(left.protocolError?.diagnostic).toContain('[stdout-protocol]'); + await expect(right.request({ type: 'echo', value: 'still-alive' })) + .resolves.toMatchObject({ data: { value: 'still-alive' } }); + }, + ); + + it('settles every pending command after an unexpected exit', async () => { + const worker = await makeWorker(); + const pending = worker.request({ type: 'no_response' }); + const crash = worker.request({ type: 'crash' }); + + await expect(Promise.all([pending, crash])).rejects.toMatchObject({ code: 'PI_RPC_EXITED' }); + expect(worker.pendingCommandCount).toBe(0); + expect(worker.generation).toBe(2); + }); + + it('keeps only bounded redacted stderr diagnostics', async () => { + const secret = 'credential-that-must-not-leak'; + const worker = await makeWorker({ + env: { FAKE_PI_SECRET: secret }, + sensitiveValues: [secret], + diagnosticBytes: 160, + }); + + await worker.request({ type: 'stderr_secret' }); + await new Promise((resolvePromise) => setTimeout(resolvePromise, 10)); + expect(worker.stderrDiagnostic).not.toContain(secret); + expect(worker.stderrDiagnostic).toContain('[REDACTED]'); + expect(Buffer.byteLength(worker.stderrDiagnostic)).toBeLessThanOrEqual(160); + expect(sanitizePiDiagnostic(`token=${secret}`, [secret])).toBe('token=[REDACTED]'); + }); + + it('forces the complete child tree down after the graceful deadline', async () => { + const worker = await makeWorker({ shutdownGraceMs: 100 }); + const response = await worker.request<{ pid: number }>({ type: 'spawn_descendant' }); + const descendantPid = response.data?.pid; + expect(descendantPid).toBeTypeOf('number'); + + await expect(worker.stop()).resolves.toMatchObject({ mode: 'forced-tree-kill' }); + for (let attempt = 0; attempt < 20 && await processAlive(descendantPid!); attempt += 1) { + await new Promise((resolvePromise) => setTimeout(resolvePromise, 25)); + } + expect(await processAlive(descendantPid!)).toBe(false); + }); +}); diff --git a/tests/unit/pi-runtime-bundle.test.ts b/tests/unit/pi-runtime-bundle.test.ts new file mode 100644 index 0000000..16dc7b1 --- /dev/null +++ b/tests/unit/pi-runtime-bundle.test.ts @@ -0,0 +1,137 @@ +// @vitest-environment node + +import { afterEach, describe, expect, it } from 'vitest'; +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import YAML from 'yaml'; +import { + assertPackageIdentities, + installProductionShrinkwrap, + preparePublishedPackageRoot, + requiredPackageIdentities, +} from '../../scripts/lib/pi-runtime-bundle.mjs'; +import { parseBundleArgs } from '../../scripts/bundle-pi-runtime.mjs'; + +const scratchRoots: string[] = []; + +async function scratch(prefix: string): Promise { + const root = await mkdtemp(join(tmpdir(), prefix)); + scratchRoots.push(root); + return root; +} + +afterEach(async () => { + await Promise.all(scratchRoots.splice(0).map((root) => rm(root, { + recursive: true, + force: true, + maxRetries: 3, + }))); +}); + +describe('Pi runtime production bundler', () => { + it('turns the published root dev mismatch into a reproducible npm 11 production install', async () => { + const root = await scratch('makelore-pi-bundle-mismatch-'); + const source = join(root, 'published'); + const stage = join(root, 'stage'); + await mkdir(source); + await writeFile(join(source, 'package.json'), `${JSON.stringify({ + name: '@fixture/pi-runtime', + version: '1.0.0', + devDependencies: { 'missing-from-production-lock': '1.0.0' }, + }, null, 2)}\n`); + await writeFile(join(source, 'npm-shrinkwrap.json'), `${JSON.stringify({ + name: '@fixture/pi-runtime', + version: '1.0.0', + lockfileVersion: 3, + requires: true, + packages: { + '': { name: '@fixture/pi-runtime', version: '1.0.0' }, + }, + }, null, 2)}\n`); + + const prepared = await preparePublishedPackageRoot(source, stage); + expect(prepared.omittedDevDependencies).toEqual(['missing-from-production-lock']); + expect(JSON.parse(await readFile(join(stage, 'package.json'), 'utf8'))) + .not.toHaveProperty('devDependencies'); + await expect(installProductionShrinkwrap({ + projectRoot: process.cwd(), + stageRoot: stage, + target: { platform: process.platform, arch: process.arch }, + })).resolves.toMatchObject({}); + }, 15_000); + + it('accepts same-version nested-to-root flattening by identity instead of physical lock path', () => { + const shrinkwrap = { + packages: { + 'node_modules/p-retry/node_modules/@types/retry': { + version: '0.12.0', + }, + }, + }; + const required = requiredPackageIdentities(shrinkwrap, { + platform: process.platform, + arch: process.arch, + }); + + expect(required.identities).toEqual(['@types/retry@0.12.0']); + expect(() => assertPackageIdentities(required.identities, [ + '@types/retry@0.12.0', + ])).not.toThrow(); + }); + + it('keeps target selection explicit and rejects cross-OS staging', () => { + const current = `${process.platform}-${process.arch}`; + expect(parseBundleArgs(['--target', current]).targets).toEqual([ + expect.objectContaining({ platform: process.platform, arch: process.arch }), + ]); + const foreign = process.platform === 'win32' ? 'linux-x64' : 'win32-x64'; + expect(() => parseBundleArgs(['--target', foreign])).toThrow(/must be staged on/); + expect(() => parseBundleArgs(['--target', 'unknown-x64'])).toThrow(/Unsupported/); + }); + + it('wires every package target to its matching staged Pi runtime', async () => { + const packageJson = JSON.parse(await readFile('package.json', 'utf8')) as { + scripts: Record; + }; + const builder = YAML.parse(await readFile('electron-builder.yml', 'utf8')) as Record< + string, + { extraResources?: Array<{ from: string; to: string }> } + >; + + expect(packageJson.scripts.build).toContain('bundle-pi-runtime.mjs --release-targets'); + expect(packageJson.scripts.package).toContain('bundle-pi-runtime.mjs --release-targets'); + expect(packageJson.scripts['package:stage:win-x64']) + .toContain('bundle-pi-runtime.mjs --target win32-x64'); + expect(builder.mac.extraResources).toContainEqual({ + from: 'build/pi-runtime/darwin-${arch}', + to: 'pi-runtime', + filter: ['**/*', '!node_modules{,/**/*}'], + }); + expect(builder.mac.extraResources).toContainEqual({ + from: 'build/pi-runtime/darwin-${arch}/node_modules', + to: 'pi-runtime/node_modules', + filter: ['**/*'], + }); + expect(builder.win.extraResources).toContainEqual({ + from: 'build/pi-runtime/win32-${arch}', + to: 'pi-runtime', + filter: ['**/*', '!node_modules{,/**/*}'], + }); + expect(builder.win.extraResources).toContainEqual({ + from: 'build/pi-runtime/win32-${arch}/node_modules', + to: 'pi-runtime/node_modules', + filter: ['**/*'], + }); + expect(builder.linux.extraResources).toContainEqual({ + from: 'build/pi-runtime/linux-${arch}', + to: 'pi-runtime', + filter: ['**/*', '!node_modules{,/**/*}'], + }); + expect(builder.linux.extraResources).toContainEqual({ + from: 'build/pi-runtime/linux-${arch}/node_modules', + to: 'pi-runtime/node_modules', + filter: ['**/*'], + }); + }); +}); diff --git a/tests/unit/pi-worker-process-real.test.ts b/tests/unit/pi-worker-process-real.test.ts new file mode 100644 index 0000000..e16d188 --- /dev/null +++ b/tests/unit/pi-worker-process-real.test.ts @@ -0,0 +1,61 @@ +// @vitest-environment node + +import { afterEach, describe, expect, it } from 'vitest'; +import { realpathSync } from 'node:fs'; +import { mkdir, mkdtemp, readFile, rm } from 'node:fs/promises'; +import { createRequire } from 'node:module'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { PiWorkerProcess } from '../../electron/coding-runtime/pi/worker-process'; + +const scratchRoots: string[] = []; + +afterEach(async () => { + await Promise.all(scratchRoots.splice(0).map((root) => rm(root, { + recursive: true, + force: true, + maxRetries: 3, + }))); +}); + +describe('locked Pi worker process smoke', () => { + it('loads the real Pi 0.84.2 entry through Electron Node and exits by closing stdin', 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 packageJson = JSON.parse(await readFile(join(packageRoot, 'package.json'), 'utf8')) as { + version: string; + }; + expect(packageJson.version).toBe('0.84.2'); + + const root = await mkdtemp(join(tmpdir(), 'makelore-pi-real-worker-')); + 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 worker = await new PiWorkerProcess({ + executablePath: electronExecutable, + cliPath: join(packageRoot, 'dist', 'cli.js'), + cwd, + configDir, + sessionDir, + commandTimeoutMs: 5_000, + }).start(); + try { + await expect(worker.request({ type: 'get_state' })).resolves.toMatchObject({ + type: 'response', + command: 'get_state', + success: true, + }); + await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); + } finally { + await worker.stop().catch(() => undefined); + } + }, 15_000); +});