From 7c2f51872255f73e19b143f5bc3ddfcda5f86b01 Mon Sep 17 00:00:00 2001 From: gaebal-gajae Date: Tue, 1 Sep 2026 02:54:51 +0000 Subject: [PATCH] fix(hooks): guard all run.cjs protocol writes against closed consumers Generic pipe forwarding was not enough: trusted Worker flushes and timeout diagnostics still wrote directly to process.stdout/stderr, and the generic dest listener was removed on source end before queued writes could fail. Install a per-invocation protocol sink that owns destination error handlers until the runner finishes, routes Worker buffers and timeout diagnostics through guarded writes, and fail-opens on EPIPE. --- inventory/inventory-graph.json | 10 +- scripts/run.cjs | 117 ++++++++++++------ .../windows-prompt-hook-runner.test.ts | 49 +++++++- 3 files changed, 134 insertions(+), 42 deletions(-) diff --git a/inventory/inventory-graph.json b/inventory/inventory-graph.json index b25fadf47..824f42878 100644 --- a/inventory/inventory-graph.json +++ b/inventory/inventory-graph.json @@ -5,15 +5,15 @@ "provenance": { "base": "05c800f40d1ad53b42a78609d2667ef4f726808b", "planningHead": "0a91273e61dbbd47eb0af4c02844409251e08398", - "head": "b8d53c022cb3e21c0ef5e1d15cdc8bb3fbf3a619", - "sourceSha256": "7ea030666dea9a5eccbb137a296ab0132a3ecfd85ca0b0fefdf8c0d6e7efa7b1", + "head": "c9ce5f4ca192a0d3f95d276a462331bb104fe623", + "sourceSha256": "e4dddb487dae147e2f2a91cc5c2f0d99744be0034559efbb7e7f5531cda7130b", "generatedAt": null, "generator": "scripts/generate-inventory-graph.mjs" }, "base": "05c800f40d1ad53b42a78609d2667ef4f726808b", "planningHead": "0a91273e61dbbd47eb0af4c02844409251e08398", - "head": "b8d53c022cb3e21c0ef5e1d15cdc8bb3fbf3a619", - "sourceSha256": "7ea030666dea9a5eccbb137a296ab0132a3ecfd85ca0b0fefdf8c0d6e7efa7b1", + "head": "c9ce5f4ca192a0d3f95d276a462331bb104fe623", + "sourceSha256": "e4dddb487dae147e2f2a91cc5c2f0d99744be0034559efbb7e7f5531cda7130b", "counts": { "public": { "skills": 35, @@ -76665,5 +76665,5 @@ } }, "inventorySha256": "b3d68b6eb375492031545ea45f5699047ae9e1115224086a7b99201a2bf55374", - "manifestSha256": "39eae76d9b54bb0bf4e8524cec65cd484205df3efe82b6ad8615f0b9fb8b460a" + "manifestSha256": "7955057c93f73c197f95e0e45f7052c1e0ab374956f7aa2746a9e7df095f68a5" } diff --git a/scripts/run.cjs b/scripts/run.cjs index fde4ef0e2..526ac2fdd 100644 --- a/scripts/run.cjs +++ b/scripts/run.cjs @@ -250,11 +250,13 @@ function resolveTrustedSessionEndTarget(resolution, extraArgs) { } -function writeTimeoutDiagnostic(targetPath, manifestHook, timeoutMs) { +function writeTimeoutDiagnostic(targetPath, manifestHook, timeoutMs, sink) { const message = `[run.cjs] Hook ${basename(targetPath)} timed out after ${timeoutMs}ms; exiting fail-open.\n`; if (manifestHook?.event !== 'UserPromptSubmit' || isDebugHooksEnabled()) { - process.stderr.write(message); + if (sink) return sink.write(process.stderr, Buffer.from(message)); + try { process.stderr.write(message); } catch { /* protocol dest may already be closed */ } } + return undefined; } function captureProcessStartIdentity(pid) { @@ -363,30 +365,65 @@ function abandonProtocolSource(source, dest) { } catch { /* already flowing or destroyed */ } } -function attachProtocolForwarders(child) { - attachProtocolForwarder(child.stdout, process.stdout); - attachProtocolForwarder(child.stderr, process.stderr); -} +function createProtocolSink() { + const discarded = { stdout: false, stderr: false }; + const sources = { stdout: new Set(), stderr: new Set() }; + let installed = false; + const onStdoutError = (error) => handleDestError('stdout', process.stdout, error); + const onStderrError = (error) => handleDestError('stderr', process.stderr, error); -function attachProtocolForwarder(source, dest) { - if (!source || !dest) return; - source.pipe(dest, { end: false }); - const onDestError = (error) => { - if (!isClosedDestinationError(error)) { - const other = dest === process.stdout ? process.stderr : process.stdout; - try { - if (other && other.writable && !other.destroyed) { - const detail = error && (error.code || error.message || String(error)); - other.write(`[run.cjs] protocol stream error: ${detail}\n`); - } - } catch { /* both destinations may already be closed */ } + function handleDestError(name, dest, error) { + discarded[name] = true; + for (const source of sources[name]) abandonProtocolSource(source, dest); + sources[name].clear(); + if (!isClosedDestinationError(error) && name === 'stdout') { + void write(process.stderr, Buffer.from(`[run.cjs] protocol stream error: ${error.code || error.message}\n`)); } - abandonProtocolSource(source, dest); - }; - dest.on('error', onDestError); - const cleanup = () => dest.removeListener('error', onDestError); - source.once('end', cleanup); - source.once('close', cleanup); + } + + function install() { + if (installed) return; + installed = true; + process.stdout.on('error', onStdoutError); + process.stderr.on('error', onStderrError); + } + + function uninstall() { + if (!installed) return; + installed = false; + process.stdout.removeListener('error', onStdoutError); + process.stderr.removeListener('error', onStderrError); + } + + function write(dest, data) { + install(); + const name = dest === process.stderr ? 'stderr' : 'stdout'; + if (discarded[name] || !dest || dest.destroyed || !dest.writable) return Promise.resolve(); + return new Promise((resolve) => { + try { + dest.write(data, () => resolve()); + } catch (error) { + handleDestError(name, dest, error); + resolve(); + } + }); + } + + function attachChild(child) { + install(); + const bind = (source, dest, name) => { + if (!source) return; + sources[name].add(source); + source.pipe(dest, { end: false }); + const drop = () => sources[name].delete(source); + source.once('end', drop); + source.once('close', drop); + }; + bind(child.stdout, process.stdout, 'stdout'); + bind(child.stderr, process.stderr, 'stderr'); + } + + return { install, uninstall, write, attachChild }; } function detachProtocolStdio(child) { @@ -479,9 +516,15 @@ function superviseGenericChild(targetPath, extraArgs) { } function runGenericChild(targetPath, extraArgs, timeoutMs, manifestHook) { + const sink = createProtocolSink(); + sink.install(); return new Promise(resolve => { let terminal = false; let timer; + const finish = (status) => { + sink.uninstall(); + resolve(status); + }; const child = spawn(process.execPath, resolveGenericChildCommand(targetPath, extraArgs), { stdio: resolveGenericChildStdio(), env: { @@ -491,7 +534,7 @@ function runGenericChild(targetPath, extraArgs, timeoutMs, manifestHook) { windowsHide: true, detached: true, }); - attachProtocolForwarders(child); + sink.attachChild(child); // Capture the durable start identity immediately so reapTree can reject // a PID that was reused after the child exited. const childIdentity = child.pid ? captureProcessStartIdentity(child.pid) : null; @@ -510,6 +553,7 @@ function runGenericChild(targetPath, extraArgs, timeoutMs, manifestHook) { terminal = true; detachHandlers(); detachProtocolStdio(child); + sink.uninstall(); reapTree(child, childIdentity); process.exit(0); } @@ -528,8 +572,8 @@ function runGenericChild(targetPath, extraArgs, timeoutMs, manifestHook) { // taskkill must not keep Claude Code blocked on EOF past the host fuse. detachProtocolStdio(child); releaseGenericChild(child); - writeTimeoutDiagnostic(targetPath, manifestHook, timeoutMs); - resolve(0); + writeTimeoutDiagnostic(targetPath, manifestHook, timeoutMs, sink); + finish(0); reapTree(child, childIdentity); }, timeoutMs); @@ -543,7 +587,7 @@ function runGenericChild(targetPath, extraArgs, timeoutMs, manifestHook) { // (#3920 success-path hang, outer harness timeout 124). void settleProtocolStdio(child).then(() => { releaseGenericChild(child); - resolve(typeof code === 'number' ? code : 0); + finish(typeof code === 'number' ? code : 0); }); }); child.once('error', () => { @@ -551,7 +595,7 @@ function runGenericChild(targetPath, extraArgs, timeoutMs, manifestHook) { terminal = true; detachHandlers(); detachProtocolStdio(child); - resolve(0); + finish(0); }); for (const signal of RUNNER_TERMINATION_SIGNALS) process.on(signal, onRunnerSignal); @@ -566,6 +610,8 @@ async function runWorker(targetPath, manifestHook, timeoutMs) { let discardOutput = false; const stdout = []; const stderr = []; + const sink = createProtocolSink(); + sink.install(); const cleanupInput = () => { if (!worker) return; @@ -575,15 +621,12 @@ async function runWorker(targetPath, manifestHook, timeoutMs) { const waitForOutputEnd = stream => stream.readableEnded ? Promise.resolve() : new Promise(resolve => stream.once('end', resolve)); - const writeBuffer = (stream, buffer) => new Promise(resolve => { - stream.write(buffer, () => resolve()); - }); const forwardBuffers = async (workerError) => { - if (stdout.length) await writeBuffer(process.stdout, Buffer.concat(stdout)); - if (stderr.length) await writeBuffer(process.stderr, Buffer.concat(stderr)); + if (stdout.length) await sink.write(process.stdout, Buffer.concat(stdout)); + if (stderr.length) await sink.write(process.stderr, Buffer.concat(stderr)); if (workerError) { const diagnostic = workerError.stack || workerError.message || String(workerError); - await writeBuffer(process.stderr, Buffer.from(`${diagnostic}\n`)); + await sink.write(process.stderr, Buffer.from(`${diagnostic}\n`)); } }; const waitForWorkerOutput = () => Promise.all([ @@ -600,6 +643,7 @@ async function runWorker(targetPath, manifestHook, timeoutMs) { cleanupInput(); if (worker) await waitForWorkerOutput(); await forwardBuffers(workerError); + sink.uninstall(); resolve(status); }; @@ -613,7 +657,8 @@ async function runWorker(targetPath, manifestHook, timeoutMs) { } catch { // Termination is best-effort; the hook must still fail open. } - writeTimeoutDiagnostic(targetPath, manifestHook, timeoutMs); + await writeTimeoutDiagnostic(targetPath, manifestHook, timeoutMs, sink); + sink.uninstall(); resolve(0); }, timeoutMs); diff --git a/src/__tests__/windows-prompt-hook-runner.test.ts b/src/__tests__/windows-prompt-hook-runner.test.ts index 4635c1915..07248e2c7 100644 --- a/src/__tests__/windows-prompt-hook-runner.test.ts +++ b/src/__tests__/windows-prompt-hook-runner.test.ts @@ -1,4 +1,4 @@ -import { spawnSync } from 'node:child_process'; +import { spawn, spawnSync } from 'node:child_process'; import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -108,6 +108,53 @@ describe('Windows-safe prompt hook runner paths', () => { const innerMs = runCjs.resolveGenericTimeoutMs({ timeoutMs: 1000, event: 'UserPromptSubmit' }); expect(result.stderr).toContain(`Hook keyword-detector.mjs timed out after ${innerMs}ms; exiting fail-open.`); }); + it('fail-opens a trusted Worker when protocol stdout is closed', async () => { + const cacheBase = mkdtempSync(join(tmpdir(), 'omc worker closed stdout-')); + tempDirs.push(cacheBase); + const root = join(cacheBase, '4.8.0'); + makePlugin(root, workerProbe); + const target = join(root, 'scripts', 'keyword-detector.mjs'); + const runner = spawn(NODE, [RUN_CJS_PATH, target], { + stdio: ['pipe', 'pipe', 'pipe'], + env: { ...process.env, CLAUDE_PLUGIN_ROOT: root }, + windowsHide: true, + }); + runner.stdin.write('{}'); + runner.stdin.end(); + runner.stdout.destroy(); + const code = await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('trusted Worker hung after stdout close')), 5000); + runner.once('exit', status => { + clearTimeout(timer); + resolve(status); + }); + }); + expect(code).toBe(0); + }); + + it('fail-opens a trusted Worker timeout diagnostic when protocol stderr is closed', async () => { + const cacheBase = mkdtempSync(join(tmpdir(), 'omc worker closed stderr-')); + tempDirs.push(cacheBase); + const root = join(cacheBase, '4.9.0'); + const target = join(root, 'scripts', 'keyword-detector.mjs'); + makePlugin(root, "setInterval(() => {}, 1000);", 1); + const runner = spawn(NODE, [RUN_CJS_PATH, target], { + stdio: ['pipe', 'pipe', 'pipe'], + env: { ...process.env, CLAUDE_PLUGIN_ROOT: root, OMC_DEBUG_HOOKS: '1' }, + windowsHide: true, + }); + runner.stdin.write('{}'); + runner.stdin.end(); + runner.stderr.destroy(); + const code = await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('trusted Worker hung after stderr close')), 5000); + runner.once('exit', status => { + clearTimeout(timer); + resolve(status); + }); + }); + expect(code).toBe(0); + }); it('models an argv-delayed launch crossing 10s and failing open before the 30s host fuse', () => { const cacheBase = mkdtempSync(join(tmpdir(), 'omc delayed prompt launch-'));