diff --git a/config/build-plugins/main-build-entry-points.ts b/config/build-plugins/main-build-entry-points.ts new file mode 100644 index 00000000000..a1c9427ef6a --- /dev/null +++ b/config/build-plugins/main-build-entry-points.ts @@ -0,0 +1,60 @@ +import { resolve } from 'node:path' +import { ORCAD_LOCAL_SERVE_SELECTION_ENTRY } from '../../src/shared/orcad-local-serve-selection' +import { CLI_MAIN_ENTRY_NAMES } from './plain-node-entry-guard' + +/** Main-build Rollup inputs: output file name (without .js) to its source entry. */ +export const MAIN_BUILD_ENTRY_POINTS: Record = { + index: resolve('src/main/index.ts'), + // Why: sandboxed webview preloads cannot load Rollup helper chunks. + 'browser-window-close-preload': resolve('src/preload/browser-window-close.ts'), + 'doc-preview-link-preload': resolve('src/preload/doc-preview-link.ts'), + 'daemon-entry': resolve('src/main/daemon/daemon-entry.ts'), + 'plugin-host-entry': resolve('src/main/plugins/plugin-host-entry.ts'), + 'computer-sidecar': resolve('src/main/computer/sidecar-entry.ts'), + 'stt-worker': resolve('src/main/speech/stt-worker.ts'), + 'warp-theme-parser-worker': resolve('src/main/warp-themes/warp-theme-parser-worker.ts'), + 'foreign-sqlite-reader-entry': resolve( + 'src/main/foreign-sqlite-readers/foreign-sqlite-reader-entry.ts' + ), + 'session-scanner-service-entry': resolve('src/main/ai-vault/session-scanner-service-entry.ts'), + 'wsl-transcript-fs-process-entry': resolve( + 'src/main/native-chat/wsl-transcript-fs-process-entry.ts' + ), + // Why: libuv spawns processes inline on the calling loop, so the port + // scan's probe commands run on a worker thread instead of the UI one. + 'port-scan-command-worker-entry': resolve('src/main/ports/port-scan-command-worker-entry.ts'), + // Why (#27362): the Codex usage probe spawns on window focus; a cold + // CreateProcessW must not block the UI loop. + 'worker-thread-child-process-entry': resolve( + 'src/main/worker-thread-child-process/worker-thread-child-process-entry.ts' + ), + // Why: the Claude/Codex/OpenCode usage scans walk whole history + // corpora and read SQLite synchronously; a worker thread keeps that + // off the main-process event loop. + 'usage-scan-worker-entry': resolve('src/main/usage/usage-scan-worker-entry.ts'), + // Why: a first account setup can merge a large history tree with sync fs calls. + 'claude-profile-setup-worker-entry': resolve( + 'src/main/claude-accounts/claude-profile-setup-worker-entry.ts' + ), + 'profile-state-backup-worker-entry': resolve( + 'src/main/persistence/profile-state/profile-state-backup-worker-entry.ts' + ), + 'profile-state-writer-worker-entry': resolve( + 'src/main/persistence/profile-state/profile-state-writer-worker-entry.ts' + ), + // Why: forked with ELECTRON_RUN_AS_NODE so @parcel/watcher faults + // can't take down the main process (issue #7547). + 'parcel-watcher-process-entry': resolve('src/main/ipc/parcel-watcher-process-entry.ts'), + // Why: `orca serve` runs it under ELECTRON_RUN_AS_NODE so the CLI never bundles orcad prep. + [ORCAD_LOCAL_SERVE_SELECTION_ENTRY]: resolve( + 'src/main/orcad/orcad-local-serve-selection-entry.ts' + ), + // Why: a worker thread survives the macOS 26 AppKit main-thread deadlock + // without paying for another Electron process. + 'main-thread-hang-watchdog-entry': resolve( + 'src/main/hang-watchdog/main-thread-hang-watchdog-entry.ts' + ), + ...Object.fromEntries( + CLI_MAIN_ENTRY_NAMES.map((module) => [module, resolve(`src/main/${module}.ts`)]) + ) +} diff --git a/config/build-plugins/plain-node-entry-guard.ts b/config/build-plugins/plain-node-entry-guard.ts index f31f16012ab..fdd9e7d14ca 100644 --- a/config/build-plugins/plain-node-entry-guard.ts +++ b/config/build-plugins/plain-node-entry-guard.ts @@ -62,6 +62,7 @@ const WORKER_THREAD_ENTRY_NAMES = [ 'foreign-sqlite-reader-entry', 'main-thread-hang-watchdog-entry', 'port-scan-command-worker-entry', + 'worker-thread-child-process-entry', 'usage-scan-worker-entry', 'claude-profile-setup-worker-entry', 'profile-state-backup-worker-entry', diff --git a/config/knip.json b/config/knip.json index 8bb86d8d261..fefbeabd39d 100644 --- a/config/knip.json +++ b/config/knip.json @@ -11,6 +11,7 @@ "src/main/warp-themes/warp-theme-parser-worker.ts", "src/main/foreign-sqlite-readers/foreign-sqlite-reader-entry.ts", "src/main/ports/port-scan-command-worker-entry.ts", + "src/main/worker-thread-child-process/worker-thread-child-process-entry.ts", "src/main/ipc/parcel-watcher-process-entry.ts", "src/main/hang-watchdog/main-thread-hang-watchdog-entry.ts", "src/main/agent-hooks/managed-agent-hook-controls.ts", diff --git a/config/scripts/electron-builder-config.test.mjs b/config/scripts/electron-builder-config.test.mjs index 1f21d2e7192..0e450ae8945 100644 --- a/config/scripts/electron-builder-config.test.mjs +++ b/config/scripts/electron-builder-config.test.mjs @@ -8,6 +8,12 @@ import { writeMobileWebBundleFixtureTree } from './mobile-web-bundle-fixture-tre const REPO_ROOT = join(import.meta.dirname, '..', '..') const SRC_MAIN_DIR = join(REPO_ROOT, 'src', 'main') +const MAIN_BUILD_ENTRY_POINTS_PATH = join( + REPO_ROOT, + 'config', + 'build-plugins', + 'main-build-entry-points.ts' +) const require = createRequire(import.meta.url) const electronBuilderConfig = require('../electron-builder.config.cjs') @@ -353,8 +359,10 @@ describe('electron-builder config', () => { const entryFilename = 'wsl-transcript-fs-process-entry.js' expect(electronBuilderConfig.asarUnpack).toContain(`out/main/${entryFilename}`) - const viteConfig = await readFile(join(REPO_ROOT, 'electron.vite.config.ts'), 'utf8') - expect(viteConfig).toMatch(new RegExp(`'${entryFilename.replace(/\.js$/, '')}':\\s*resolve\\(`)) + const entryPoints = await readFile(MAIN_BUILD_ENTRY_POINTS_PATH, 'utf8') + expect(entryPoints).toMatch( + new RegExp(`'${entryFilename.replace(/\.js$/, '')}':\\s*resolve\\(`) + ) }) // Why: the CLI runs this entry under ELECTRON_RUN_AS_NODE, which cannot read @@ -401,7 +409,10 @@ describe('electron-builder config', () => { // longer exists. const viteConfig = await readFile(join(REPO_ROOT, 'electron.vite.config.ts'), 'utf8') expect(viteConfig).toContain("entryFileNames: '[name].js'") - expect(viteConfig).toMatch(new RegExp(`'${entryFilename.replace(/\.js$/, '')}':\\s*resolve\\(`)) + const entryPoints = await readFile(MAIN_BUILD_ENTRY_POINTS_PATH, 'utf8') + expect(entryPoints).toMatch( + new RegExp(`'${entryFilename.replace(/\.js$/, '')}':\\s*resolve\\(`) + ) }) it('keeps the worker-thread hang watchdog inside app.asar', () => { diff --git a/config/scripts/vitest-node-runtime-files.mjs b/config/scripts/vitest-node-runtime-files.mjs index 96b08ec1d65..9c085bc44b5 100644 --- a/config/scripts/vitest-node-runtime-files.mjs +++ b/config/scripts/vitest-node-runtime-files.mjs @@ -44,6 +44,8 @@ export const NODE_RUNTIME_INCLUDE = [ 'src/main/agent-hooks/server-transport-interference.test.ts', 'src/main/plugins/plugin-worker-supervision.integration.test.ts', 'src/main/usage/usage-scan-worker-event-loop.test.ts', + // Asserts Node's spawn-failure close signal and that a dead worker closes its children's pipes. + 'src/main/worker-thread-child-process/worker-thread-child-process.test.ts', 'src/main/computer/macos-native-provider-client.test.ts', 'src/main/ai-vault-search/session-search-schema.test.ts', 'src/main/ai-vault-search/session-search-fts5-contract.test.ts', diff --git a/electron.vite.config.ts b/electron.vite.config.ts index 101e1f5333c..2455c093eb5 100644 --- a/electron.vite.config.ts +++ b/electron.vite.config.ts @@ -7,11 +7,8 @@ import tailwindcss from '@tailwindcss/vite' import { createBootstrapFatalExitBanner } from './config/build-plugins/bootstrap-fatal-exit-banner' import { createPdfjsViewerAssetsPlugin } from './config/build-plugins/pdfjs-viewer-assets' import { createProcessHostDevRebuildPlugin } from './config/build-plugins/process-host-dev-rebuild' -import { - CLI_MAIN_ENTRY_NAMES, - createPlainNodeEntryGuardPlugin -} from './config/build-plugins/plain-node-entry-guard' -import { ORCAD_LOCAL_SERVE_SELECTION_ENTRY } from './src/shared/orcad-local-serve-selection' +import { MAIN_BUILD_ENTRY_POINTS } from './config/build-plugins/main-build-entry-points' +import { createPlainNodeEntryGuardPlugin } from './config/build-plugins/plain-node-entry-guard' import packageJson from './package.json' with { type: 'json' } import { resolveBuildSourceCommit } from './config/scripts/build-source-commit.mjs' @@ -235,60 +232,7 @@ export const electronViteConfig: UserConfig = { // Why: native dependencies must resolve from packaged node_modules, // while the unpacked daemon needs its pure-JS xterm graph bundled. external: isExternalMainModule, - input: { - index: resolve('src/main/index.ts'), - // Why: sandboxed webview preloads cannot load Rollup helper chunks. - 'browser-window-close-preload': resolve('src/preload/browser-window-close.ts'), - 'doc-preview-link-preload': resolve('src/preload/doc-preview-link.ts'), - 'daemon-entry': resolve('src/main/daemon/daemon-entry.ts'), - 'plugin-host-entry': resolve('src/main/plugins/plugin-host-entry.ts'), - 'computer-sidecar': resolve('src/main/computer/sidecar-entry.ts'), - 'stt-worker': resolve('src/main/speech/stt-worker.ts'), - 'warp-theme-parser-worker': resolve('src/main/warp-themes/warp-theme-parser-worker.ts'), - 'foreign-sqlite-reader-entry': resolve( - 'src/main/foreign-sqlite-readers/foreign-sqlite-reader-entry.ts' - ), - 'session-scanner-service-entry': resolve( - 'src/main/ai-vault/session-scanner-service-entry.ts' - ), - 'wsl-transcript-fs-process-entry': resolve( - 'src/main/native-chat/wsl-transcript-fs-process-entry.ts' - ), - // Why: libuv spawns processes inline on the calling loop, so the port - // scan's probe commands run on a worker thread instead of the UI one. - 'port-scan-command-worker-entry': resolve( - 'src/main/ports/port-scan-command-worker-entry.ts' - ), - // Why: the Claude/Codex/OpenCode usage scans walk whole history - // corpora and read SQLite synchronously; a worker thread keeps that - // off the main-process event loop. - 'usage-scan-worker-entry': resolve('src/main/usage/usage-scan-worker-entry.ts'), - // Why: a first account setup can merge a large history tree with sync fs calls. - 'claude-profile-setup-worker-entry': resolve( - 'src/main/claude-accounts/claude-profile-setup-worker-entry.ts' - ), - 'profile-state-backup-worker-entry': resolve( - 'src/main/persistence/profile-state/profile-state-backup-worker-entry.ts' - ), - 'profile-state-writer-worker-entry': resolve( - 'src/main/persistence/profile-state/profile-state-writer-worker-entry.ts' - ), - // Why: forked with ELECTRON_RUN_AS_NODE so @parcel/watcher faults - // can't take down the main process (issue #7547). - 'parcel-watcher-process-entry': resolve('src/main/ipc/parcel-watcher-process-entry.ts'), - // Why: `orca serve` runs it under ELECTRON_RUN_AS_NODE so the CLI never bundles orcad prep. - [ORCAD_LOCAL_SERVE_SELECTION_ENTRY]: resolve( - 'src/main/orcad/orcad-local-serve-selection-entry.ts' - ), - // Why: a worker thread survives the macOS 26 AppKit main-thread deadlock - // without paying for another Electron process. - 'main-thread-hang-watchdog-entry': resolve( - 'src/main/hang-watchdog/main-thread-hang-watchdog-entry.ts' - ), - ...Object.fromEntries( - CLI_MAIN_ENTRY_NAMES.map((module) => [module, resolve(`src/main/${module}.ts`)]) - ) - }, + input: MAIN_BUILD_ENTRY_POINTS, // Why: Rolldown's SSR default is ESM, but Electron and sidecar launchers // consume these stable CommonJS paths. output: { diff --git a/src/main/rate-limits/codex-fetcher-auth-errors.test.ts b/src/main/rate-limits/codex-fetcher-auth-errors.test.ts index 20405233c9d..04b9d5aa9a0 100644 --- a/src/main/rate-limits/codex-fetcher-auth-errors.test.ts +++ b/src/main/rate-limits/codex-fetcher-auth-errors.test.ts @@ -7,6 +7,12 @@ const { childSpawnMock, resolveCodexCommandMock, ptySpawnMock } = vi.hoisted(() ptySpawnMock: vi.fn() })) +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('node:child_process', () => ({ spawn: childSpawnMock })) diff --git a/src/main/rate-limits/codex-fetcher-backend.test.ts b/src/main/rate-limits/codex-fetcher-backend.test.ts index 74ba03959f7..b06fa48abd9 100644 --- a/src/main/rate-limits/codex-fetcher-backend.test.ts +++ b/src/main/rate-limits/codex-fetcher-backend.test.ts @@ -9,6 +9,12 @@ const { childSpawnMock, readFileMock, ptySpawnMock } = vi.hoisted(() => ({ ptySpawnMock: vi.fn() })) +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('node:child_process', () => ({ spawn: childSpawnMock })) vi.mock('node:fs/promises', () => ({ readFile: readFileMock })) vi.mock('node-pty', () => ({ spawn: ptySpawnMock })) diff --git a/src/main/rate-limits/codex-fetcher-credit-balance.test.ts b/src/main/rate-limits/codex-fetcher-credit-balance.test.ts index 016357735e9..9fa8a10a48a 100644 --- a/src/main/rate-limits/codex-fetcher-credit-balance.test.ts +++ b/src/main/rate-limits/codex-fetcher-credit-balance.test.ts @@ -6,6 +6,12 @@ const { childSpawnMock, readFileMock } = vi.hoisted(() => ({ readFileMock: vi.fn() })) +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('node:child_process', () => ({ spawn: childSpawnMock })) vi.mock('node:fs/promises', () => ({ readFile: readFileMock })) vi.mock('../codex-cli/command', () => ({ resolveCodexCommand: () => 'codex' })) diff --git a/src/main/rate-limits/codex-fetcher-probe-shutdown.test.ts b/src/main/rate-limits/codex-fetcher-probe-shutdown.test.ts index 791bcf4c7bb..3532d6bea04 100644 --- a/src/main/rate-limits/codex-fetcher-probe-shutdown.test.ts +++ b/src/main/rate-limits/codex-fetcher-probe-shutdown.test.ts @@ -8,6 +8,12 @@ const { childSpawnMock, readFileMock, resolveCodexCommandMock, ptySpawnMock } = ptySpawnMock: vi.fn() })) +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('node:child_process', () => ({ spawn: childSpawnMock })) diff --git a/src/main/rate-limits/codex-fetcher-process-contract.test.ts b/src/main/rate-limits/codex-fetcher-process-contract.test.ts index 7310a36f56b..066052dcc9f 100644 --- a/src/main/rate-limits/codex-fetcher-process-contract.test.ts +++ b/src/main/rate-limits/codex-fetcher-process-contract.test.ts @@ -11,6 +11,12 @@ const { ptySpawnMock, resolveCodexCommandMock, stubScript } = vi.hoisted(() => ( })) // Runs the node stub in place of the resolved CLI, through the real chokepoint. +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('@orca/process-host', async (importOriginal) => { const actual = await importOriginal() return { diff --git a/src/main/rate-limits/codex-fetcher-rpc-exit-diagnostics.test.ts b/src/main/rate-limits/codex-fetcher-rpc-exit-diagnostics.test.ts index 12d74f5a5bb..a251d09d8b0 100644 --- a/src/main/rate-limits/codex-fetcher-rpc-exit-diagnostics.test.ts +++ b/src/main/rate-limits/codex-fetcher-rpc-exit-diagnostics.test.ts @@ -12,6 +12,12 @@ const { ptySpawnMock, resolveCodexCommandMock, stubScript } = vi.hoisted(() => ( })) // Runs the node stub in place of the resolved CLI, through the real chokepoint. +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('@orca/process-host', async (importOriginal) => { const actual = await importOriginal() return { diff --git a/src/main/rate-limits/codex-fetcher-runtime-pairing.test.ts b/src/main/rate-limits/codex-fetcher-runtime-pairing.test.ts index 3414d2b609d..98c699dc5a9 100644 --- a/src/main/rate-limits/codex-fetcher-runtime-pairing.test.ts +++ b/src/main/rate-limits/codex-fetcher-runtime-pairing.test.ts @@ -12,6 +12,12 @@ const { childSpawnMock, readFileMock, resolveCodexCommandMock, ptySpawnMock } = ptySpawnMock: vi.fn() })) +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('node:child_process', () => ({ spawn: childSpawnMock })) // The chokepoint is the seam: assertions see what the fetcher asked for, before shim resolution. vi.mock('@orca/process-host', async (importOriginal) => ({ diff --git a/src/main/rate-limits/codex-fetcher-session-supplement.test.ts b/src/main/rate-limits/codex-fetcher-session-supplement.test.ts index 44c08613721..8d7114ca7bc 100644 --- a/src/main/rate-limits/codex-fetcher-session-supplement.test.ts +++ b/src/main/rate-limits/codex-fetcher-session-supplement.test.ts @@ -6,6 +6,12 @@ const { childSpawnMock, readFileMock } = vi.hoisted(() => ({ readFileMock: vi.fn() })) +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('node:child_process', () => ({ spawn: childSpawnMock })) vi.mock('node:fs/promises', () => ({ readFile: readFileMock })) vi.mock('../codex-cli/command', () => ({ resolveCodexCommand: () => 'codex' })) diff --git a/src/main/rate-limits/codex-fetcher.test.ts b/src/main/rate-limits/codex-fetcher.test.ts index d700b8c36d4..feac7000585 100644 --- a/src/main/rate-limits/codex-fetcher.test.ts +++ b/src/main/rate-limits/codex-fetcher.test.ts @@ -19,6 +19,12 @@ const { startBackfillRecoveryMock: vi.fn(() => Promise.resolve(null)) })) +// The probe's child runs in-process here; the worker transport has its own suite. +vi.mock( + '../worker-thread-child-process/worker-thread-child-process', + () => + import('../worker-thread-child-process/worker-thread-child-process-in-process.test-fixtures') +) vi.mock('node:child_process', () => ({ spawn: childSpawnMock })) diff --git a/src/main/rate-limits/codex-fetcher.ts b/src/main/rate-limits/codex-fetcher.ts index d96525a3d4e..c3aa28c053b 100644 --- a/src/main/rate-limits/codex-fetcher.ts +++ b/src/main/rate-limits/codex-fetcher.ts @@ -13,7 +13,7 @@ import { } from '../codex-cli/codex-home-process-lock' import { isCodexStateDbBackfillPending } from '../codex/codex-state-db' import { startCodexStateDbBackfillRecoveryInBackground } from '../codex/codex-state-db-backfill-recovery' -import { spawnProcess } from '@orca/process-host' +import { spawnProcessOnWorkerThread } from '../worker-thread-child-process/worker-thread-child-process' import { buildConfiguredProxyEnv } from '../../shared/network-proxy' import { getMainHttpClient } from '../network/http-client' import { probeCodexAuthPresence } from './codex-auth-presence' @@ -28,10 +28,7 @@ import { consumeCodexRateLimitResetCreditFromBackend, supplementCodexRateLimitResetCredits } from './codex-reset-credit-client' -import { - readCodexRateLimitsViaRpc, - type CodexRpcRateLimitChild -} from './codex-rpc-rate-limit-probe' +import { readCodexRateLimitsViaRpc } from './codex-rpc-rate-limit-probe' import { getHiddenRateLimitWslCwdSetupCommands, resolveHiddenRateLimitPtyCwd @@ -95,10 +92,10 @@ async function fetchViaRpc(options?: CodexRateLimitFetchOptions): Promise +} + +export type WorkerThreadChildRequest = + | { type: 'spawn'; id: number; spec: WorkerThreadChildSpec } + | { type: 'stdin-write'; id: number; data: string } + | { type: 'stdin-end'; id: number } + | { type: 'kill'; id: number; signal?: NodeJS.Signals } + +export type WorkerThreadChildErrorPayload = { message: string; code?: string } + +export type WorkerThreadChildEvent = + | { type: 'spawn'; id: number; pid: number | undefined } + // Structured clone turns a Buffer into a plain Uint8Array. + | { type: 'stdout' | 'stderr'; id: number; chunk: Uint8Array } + // 'spawn-error': process creation threw, so no exit or close will follow. + | { type: 'error' | 'spawn-error'; id: number; error: WorkerThreadChildErrorPayload } + | { type: 'exit' | 'close'; id: number; code: number | null; signal: NodeJS.Signals | null } diff --git a/src/main/worker-thread-child-process/worker-thread-child-process-server.ts b/src/main/worker-thread-child-process/worker-thread-child-process-server.ts new file mode 100644 index 00000000000..81b10725fd7 --- /dev/null +++ b/src/main/worker-thread-child-process/worker-thread-child-process-server.ts @@ -0,0 +1,81 @@ +import type { PipedChildProcess } from '@orca/process-host/process-spec' +import { spawnProcess } from '@orca/process-host' +import type { + WorkerThreadChildErrorPayload, + WorkerThreadChildEvent, + WorkerThreadChildRequest +} from './worker-thread-child-process-protocol' + +// Worker-thread half; must stay electron-free (plain-Node entry guard). + +export type WorkerThreadChildPort = { + on(event: 'message', listener: (request: WorkerThreadChildRequest) => void): unknown + postMessage(event: WorkerThreadChildEvent): void +} + +function toErrorPayload(error: unknown): WorkerThreadChildErrorPayload { + if (error instanceof Error) { + return 'code' in error && typeof error.code === 'string' + ? { message: error.message, code: error.code } + : { message: error.message } + } + return { message: String(error) } +} + +export function serveWorkerThreadChildProcesses(port: WorkerThreadChildPort): void { + const children = new Map() + + const spawnChild = ({ id, spec }: Extract): void => { + let child: PipedChildProcess + try { + child = spawnProcess({ ...spec, stdio: ['pipe', 'pipe', 'pipe'] }) + } catch (error) { + port.postMessage({ type: 'spawn-error', id, error: toErrorPayload(error) }) + return + } + children.set(id, child) + const postError = (error: Error): void => + port.postMessage({ type: 'error', id, error: toErrorPayload(error) }) + child.on('spawn', () => port.postMessage({ type: 'spawn', id, pid: child.pid })) + // Copied so the clone carries only this chunk, not the whole backing allocation. + child.stdout?.on('data', (chunk: Buffer) => + port.postMessage({ type: 'stdout', id, chunk: new Uint8Array(chunk) }) + ) + child.stderr?.on('data', (chunk: Buffer) => + port.postMessage({ type: 'stderr', id, chunk: new Uint8Array(chunk) }) + ) + // An unhandled stream error would take the whole worker, and every other child, down. + child.stdout?.on('error', () => {}) + child.stderr?.on('error', () => {}) + // The stand-in surfaces stdin failures as child errors; its caller treats both alike. + child.stdin?.on('error', postError) + child.on('error', postError) + child.on('exit', (code, signal) => port.postMessage({ type: 'exit', id, code, signal })) + child.on('close', (code, signal) => { + children.delete(id) + port.postMessage({ type: 'close', id, code, signal }) + }) + } + + port.on('message', (request) => { + if (request.type === 'spawn') { + spawnChild(request) + return + } + const child = children.get(request.id) + if (!child) { + return + } + try { + if (request.type === 'stdin-write') { + child.stdin?.write(request.data) + } else if (request.type === 'stdin-end') { + child.stdin?.end() + } else { + child.kill(request.signal) + } + } catch { + // Races with exit; the exit/close events still reach the stand-in. + } + }) +} diff --git a/src/main/worker-thread-child-process/worker-thread-child-process.test.ts b/src/main/worker-thread-child-process/worker-thread-child-process.test.ts new file mode 100644 index 00000000000..44568d55fbe --- /dev/null +++ b/src/main/worker-thread-child-process/worker-thread-child-process.test.ts @@ -0,0 +1,295 @@ +import { EventEmitter } from 'node:events' +import { existsSync } from 'node:fs' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { MessageChannel, Worker, type MessagePort } from 'node:worker_threads' +import { runProcess } from '@orca/process-host' +import { build } from 'esbuild' +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest' +import { + IDLE_TEARDOWN_MS, + WorkerThreadChildProcessHost, + type WorkerThreadChildProcess +} from './worker-thread-child-process' +import { serveWorkerThreadChildProcesses } from './worker-thread-child-process-server' + +// Both halves joined by a real MessageChannel, so every event crosses structured +// clone exactly as it does between the main thread and the worker. +const openPorts: MessagePort[] = [] +const terminations: number[] = [] + +function makeConnectedHost(): WorkerThreadChildProcessHost { + return new WorkerThreadChildProcessHost(() => { + const { port1, port2 } = new MessageChannel() + openPorts.push(port1, port2) + serveWorkerThreadChildProcesses(port2) + const transport = new EventEmitter() + port1.on('message', (message) => transport.emit('message', message)) + return Object.assign(transport, { + postMessage: (message: unknown) => port1.postMessage(message), + unref: () => port1.unref(), + terminate: async () => { + terminations.push(Date.now()) + port1.close() + return 0 + } + }) + }) +} + +function waitForClose(child: WorkerThreadChildProcess): Promise<[number | null, string | null]> { + return new Promise((resolve) => child.once('close', (code, signal) => resolve([code, signal]))) +} + +function waitForSpawn(child: WorkerThreadChildProcess): Promise { + return new Promise((resolve, reject) => + child.once('spawn', () => + child.pid === undefined ? reject(new Error('spawn reported no pid')) : resolve(child.pid) + ) + ) +} + +// Why the zombie check: libuv reaps a child only through its own loop's handle, so +// once the worker's loop is gone an exited POSIX child stays a zombie until Orca exits. +async function isRunning(pid: number): Promise { + try { + process.kill(pid, 0) + } catch { + return false + } + if (process.platform === 'win32') { + return true + } + const { stdout } = await runProcess({ program: 'ps', args: ['-o', 'stat=', '-p', String(pid)] }) + const state = stdout.trim() + return state !== '' && !state.startsWith('Z') +} + +async function waitUntilExited(pid: number): Promise { + for (let attempt = 0; attempt < 200; attempt++) { + if (!(await isRunning(pid))) { + return + } + await new Promise((resolve) => setTimeout(resolve, 25)) + } + throw new Error(`process ${pid} is still running`) +} + +const ECHO_SCRIPT = ` +process.stdin.setEncoding('utf8') +process.stdin.on('data', (line) => process.stdout.write('echo:' + line)) +process.stdin.on('end', () => { process.stderr.write('bye'); process.exit(7) }) +` +const IDLE_SCRIPT = 'setInterval(() => {}, 1000)' + +afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() + terminations.length = 0 + for (const port of openPorts.splice(0)) { + port.close() + } +}) + +describe('WorkerThreadChildProcessHost over a message channel', () => { + it('runs a real piped child across the message boundary', async () => { + const child = makeConnectedHost().spawn({ + program: process.execPath, + args: ['-e', ECHO_SCRIPT] + }) + const stdout: string[] = [] + const stderr: string[] = [] + // Data must arrive as a Buffer: the RPC reader calls chunk.toString(). + child.stdout.on('data', (chunk: Buffer) => stdout.push(chunk.toString())) + child.stderr.on('data', (chunk: Buffer) => stderr.push(chunk.toString())) + const closed = waitForClose(child) + await waitForSpawn(child) + + child.stdin.write('{"id":1}\n') + await new Promise((resolve) => child.stdout.once('data', resolve)) + child.stdin.end() + + await expect(closed).resolves.toEqual([7, null]) + expect(stdout.join('')).toBe('echo:{"id":1}\n') + expect(stderr.join('')).toBe('bye') + expect(child.exitCode).toBe(7) + expect(child.kill()).toBe(false) + }, 20_000) + + it('passes env and cwd through to the child', async () => { + const child = makeConnectedHost().spawn({ + program: process.execPath, + args: ['-e', 'process.stdout.write(process.env.ORCA_PROBE_MARKER + "|" + process.cwd())'], + cwd: process.cwd(), + env: { ...process.env, ORCA_PROBE_MARKER: 'marker' } + }) + const stdout: string[] = [] + child.stdout.on('data', (chunk: Buffer) => stdout.push(chunk.toString())) + + await expect(waitForClose(child)).resolves.toEqual([0, null]) + expect(stdout.join('')).toBe(`marker|${process.cwd()}`) + }, 20_000) + + it('reports a missing program as ENOENT so callers can say the CLI is not installed', async () => { + const child = makeConnectedHost().spawn({ + program: 'orca-definitely-missing-program', + args: [] + }) + const error = new Promise((resolve) => child.once('error', resolve)) + + await expect(error).resolves.toMatchObject({ code: 'ENOENT' }) + }, 20_000) + + it('drops an error nobody listens for instead of throwing on the main loop', async () => { + const child = makeConnectedHost().spawn({ + program: 'orca-definitely-missing-program', + args: [] + }) + + await expect(waitForClose(child)).resolves.toEqual([expect.any(Number), null]) + }, 20_000) + + it('closes a child whose creation threw on the worker', async () => { + const child = makeConnectedHost().spawn({ program: '', args: [] }) + const errors: NodeJS.ErrnoException[] = [] + child.on('error', (error) => errors.push(error)) + + await expect(waitForClose(child)).resolves.toEqual([-1, null]) + expect(errors).toMatchObject([{ code: 'ERR_INVALID_ARG_VALUE' }]) + expect(child.exitCode).toBe(-1) + }, 20_000) + + it('kills a running child on request', async () => { + const child = makeConnectedHost().spawn({ + program: process.execPath, + args: ['-e', IDLE_SCRIPT] + }) + await waitForSpawn(child) + const closed = waitForClose(child) + + expect(child.kill()).toBe(true) + + const [code, signal] = await closed + expect(code !== 0 || signal !== null).toBe(true) + }, 20_000) + + it('fails closed, after listeners can attach, when no worker can be started', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + const host = new WorkerThreadChildProcessHost(() => { + throw new Error('entry missing') + }) + + const child = host.spawn({ program: process.execPath, args: [] }) + const events: unknown[][] = [] + child.on('error', (error: Error) => events.push(['error', error.message])) + child.on('exit', (code) => events.push(['exit', code])) + child.on('close', (code) => events.push(['close', code])) + expect(events).toEqual([]) + await new Promise((resolve) => process.nextTick(resolve)) + + expect(events).toEqual([ + ['error', 'Child process worker is unavailable'], + ['exit', -1], + ['close', -1] + ]) + expect(child.kill()).toBe(false) + }) + + it('tears the idle worker down only after the last child closes', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + const host = makeConnectedHost() + const first = host.spawn({ program: process.execPath, args: ['-e', ''] }) + const second = host.spawn({ program: process.execPath, args: ['-e', IDLE_SCRIPT] }) + await waitForSpawn(second) + await waitForClose(first) + await vi.advanceTimersByTimeAsync(IDLE_TEARDOWN_MS) + expect(terminations).toHaveLength(0) + + const secondClosed = waitForClose(second) + second.kill() + await secondClosed + await vi.advanceTimersByTimeAsync(IDLE_TEARDOWN_MS - 1) + expect(terminations).toHaveLength(0) + await vi.advanceTimersByTimeAsync(1) + expect(terminations).toHaveLength(1) + }, 20_000) + + it('cancels a pending idle teardown when a new child spawns', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + const host = makeConnectedHost() + await waitForClose(host.spawn({ program: process.execPath, args: ['-e', ''] })) + + await vi.advanceTimersByTimeAsync(IDLE_TEARDOWN_MS - 1) + const next = host.spawn({ program: process.execPath, args: ['-e', IDLE_SCRIPT] }) + await vi.advanceTimersByTimeAsync(IDLE_TEARDOWN_MS) + expect(terminations).toHaveLength(0) + + const closed = waitForClose(next) + next.kill() + await closed + }, 20_000) +}) + +// Why a real worker: only a real thread's death closes the child's pipes, which is +// the graceful stop an abandoned probe relies on instead of a pid-addressed kill. +describe('WorkerThreadChildProcessHost on a real worker thread', () => { + let directory = '' + let entryPath = '' + + beforeAll(async () => { + directory = await mkdtemp(join(tmpdir(), 'orca-worker-child-')) + entryPath = join(directory, 'worker-thread-child-process-entry.cjs') + await build({ + entryPoints: [join(__dirname, 'worker-thread-child-process-entry.ts')], + outfile: entryPath, + bundle: true, + platform: 'node', + format: 'cjs', + logLevel: 'silent' + }) + }, 60_000) + + afterAll(async () => { + await rm(directory, { recursive: true, force: true }) + }) + + it.each([ + ['terminated', (worker: Worker) => void worker.terminate()], + // A null request throws inside the worker's message handler. + ['crashed', (worker: Worker) => worker.postMessage(null)] + ])( + 'lets the child exit on stdin EOF when its worker is %s', + async (_, endWorker) => { + let worker: Worker | undefined + const host = new WorkerThreadChildProcessHost(() => (worker = new Worker(entryPath))) + const marker = join(directory, `eof-${Date.now()}-${Math.random()}`) + const child = host.spawn({ + program: process.execPath, + args: [ + '-e', + `process.stdin.resume();process.stdin.on('end',()=>{require('fs').writeFileSync(${JSON.stringify(marker)},'');process.exit(0)});${IDLE_SCRIPT}` + ] + }) + const pid = await waitForSpawn(child) + const abandoned = new Promise((resolve) => child.once('error', resolve)) + let closed = false + child.on('close', () => (closed = true)) + if (!worker) { + throw new Error('worker was not started') + } + + endWorker(worker) + + await expect(abandoned).resolves.toMatchObject({ + message: expect.stringMatching(/^Child process worker (exited|failed)/) + }) + await waitUntilExited(pid) + expect(existsSync(marker)).toBe(true) + expect(closed).toBe(false) + expect(() => child.stdin.end()).not.toThrow() + expect(child.kill()).toBe(false) + }, + 30_000 + ) +}) diff --git a/src/main/worker-thread-child-process/worker-thread-child-process.ts b/src/main/worker-thread-child-process/worker-thread-child-process.ts new file mode 100644 index 00000000000..3129d1d0892 --- /dev/null +++ b/src/main/worker-thread-child-process/worker-thread-child-process.ts @@ -0,0 +1,224 @@ +import { EventEmitter } from 'node:events' +import { existsSync } from 'node:fs' +import { Worker } from 'node:worker_threads' +import { LazyWorkerThreadHost, type WorkerThreadFactory } from '../lazy-worker-thread-host' +import { currentWorkerEntryLayout, resolveWorkerThreadEntryPath } from '../worker-thread-entry-path' +import type { + WorkerThreadChildErrorPayload, + WorkerThreadChildEvent, + WorkerThreadChildRequest, + WorkerThreadChildSpec +} from './worker-thread-child-process-protocol' + +// Why (#27362): libuv creates processes inline on the calling loop, and on +// Windows CreateProcessW can spend seconds in code integrity hashing a large, +// uncached binary. Spawning from a worker thread keeps that off the UI loop. +// Like the port-scan worker (#11161) this fails closed rather than spawning +// in-process when no worker can be started. + +// Starting a worker costs a few ms; a few minutes covers back-to-back probes. +export const IDLE_TEARDOWN_MS = 2 * 60_000 + +const WORKER_ENTRY_FILENAME = 'worker-thread-child-process-entry.js' +// Node reports a negative errno as the exit code of a child that never started. +const SPAWN_FAILED_EXIT_CODE = -1 + +/** False when the child is no longer reachable through a live worker. */ +type SendRequest = (request: WorkerThreadChildRequest) => boolean + +function toError(payload: WorkerThreadChildErrorPayload): Error { + const error: NodeJS.ErrnoException = new Error(payload.message) + if (payload.code) { + error.code = payload.code + } + return error +} + +// Never emits: the worker reports stdin failures as child 'error' events. +class WorkerThreadChildStdin extends EventEmitter { + constructor( + private readonly id: number, + private readonly send: SendRequest + ) { + super() + } + + write(data: string): boolean { + this.send({ type: 'stdin-write', id: this.id, data }) + return true + } + + end(): void { + this.send({ type: 'stdin-end', id: this.id }) + } +} + +/** Main-thread stand-in for a piped ChildProcess whose real process lives on the worker. */ +export class WorkerThreadChildProcess extends EventEmitter { + pid: number | undefined = undefined + exitCode: number | null = null + signalCode: NodeJS.Signals | null = null + readonly stdin: WorkerThreadChildStdin + readonly stdout = new EventEmitter() + readonly stderr = new EventEmitter() + + constructor( + readonly id: number, + private readonly send: SendRequest + ) { + super() + this.stdin = new WorkerThreadChildStdin(id, send) + } + + kill(signal?: NodeJS.Signals): boolean { + if (this.exitCode !== null || this.signalCode !== null) { + return false + } + return this.send({ type: 'kill', id: this.id, signal }) + } + + receive(event: WorkerThreadChildEvent): void { + switch (event.type) { + case 'spawn': + this.pid = event.pid + this.emit('spawn') + return + case 'stdout': + case 'stderr': + this[event.type].emit( + 'data', + Buffer.from(event.chunk.buffer, event.chunk.byteOffset, event.chunk.byteLength) + ) + return + case 'error': + this.emitErrorIfObserved(event.error) + return + case 'spawn-error': + this.emitErrorIfObserved(event.error) + this.receive({ type: 'exit', id: this.id, code: SPAWN_FAILED_EXIT_CODE, signal: null }) + this.receive({ type: 'close', id: this.id, code: SPAWN_FAILED_EXIT_CODE, signal: null }) + return + case 'exit': + this.exitCode = event.code + this.signalCode = event.signal + this.emit('exit', event.code, event.signal) + return + case 'close': + this.emit('close', event.code, event.signal) + } + } + + // Why: a late error after the caller detached would otherwise throw inside the + // worker message handler and surface as an uncaught main-process exception. + private emitErrorIfObserved(payload: WorkerThreadChildErrorPayload): void { + if (this.listenerCount('error') > 0) { + this.emit('error', toError(payload)) + } + } +} + +export class WorkerThreadChildProcessHost { + private readonly host: LazyWorkerThreadHost + private readonly children = new Map() + private nextId = 1 + + constructor(factory: WorkerThreadFactory) { + this.host = new LazyWorkerThreadHost({ + factory, + idleTeardownMs: IDLE_TEARDOWN_MS, + onMessage: (event) => this.route(event), + onError: (error) => this.abandonAll(`Child process worker failed: ${error.message}`), + onExit: (code) => this.abandonAll(`Child process worker exited with code ${code}`), + isIdle: () => this.children.size === 0, + onUnavailable: (error) => + console.warn( + `[worker-thread-child-process] worker unavailable. ${error instanceof Error ? error.message : String(error)}` + ) + }) + } + + spawn(spec: WorkerThreadChildSpec): WorkerThreadChildProcess { + const id = this.nextId++ + const child = new WorkerThreadChildProcess(id, (request) => this.post(request)) + if (this.host.ensure()) { + this.host.clearIdleTimer() + this.children.set(id, child) + if (this.post({ type: 'spawn', id, spec })) { + return child + } + this.forget(id) + } + // Deferred like Node's own spawn errors, so the caller can attach listeners first. + process.nextTick(() => + child.receive({ + type: 'spawn-error', + id, + error: { message: 'Child process worker is unavailable' } + }) + ) + return child + } + + private post(request: WorkerThreadChildRequest): boolean { + const worker = this.host.current + if (!worker || !this.children.has(request.id)) { + return false + } + try { + worker.postMessage(request) + return true + } catch { + return false + } + } + + private route(event: WorkerThreadChildEvent): void { + const child = this.children.get(event.id) + if (!child) { + return + } + if (event.type === 'close' || event.type === 'spawn-error') { + this.forget(event.id) + } + child.receive(event) + } + + private forget(id: number): void { + this.children.delete(id) + if (this.children.size === 0) { + this.host.scheduleIdleTeardown() + } + } + + // Why no 'close': the worker's death closes each child's stdin, which is its + // graceful stop, but nothing here can observe the exit, so callers' drain + // windows (and the Codex home lock they hold) still run their course. + private abandonAll(message: string): void { + const abandoned = [...this.children.values()] + this.children.clear() + this.host.destroy() + for (const child of abandoned) { + child.receive({ type: 'error', id: child.id, error: { message } }) + } + } +} + +function defaultWorkerFactory(): Worker { + const workerPath = resolveWorkerThreadEntryPath( + currentWorkerEntryLayout(__dirname), + WORKER_ENTRY_FILENAME + ) + // Why: throw synchronously so spawn() fails closed instead of waiting on a worker that never loads. + if (!existsSync(workerPath)) { + throw new Error(`Child process worker entry not found: ${workerPath}`) + } + return new Worker(workerPath) +} + +let sharedHost: WorkerThreadChildProcessHost | null = null + +/** Spawns a piped child from the shared worker thread so process creation never blocks the main loop. */ +export function spawnProcessOnWorkerThread(spec: WorkerThreadChildSpec): WorkerThreadChildProcess { + sharedHost ??= new WorkerThreadChildProcessHost(defaultWorkerFactory) + return sharedHost.spawn(spec) +}