diff --git a/src/cli.ts b/src/cli.ts old mode 100755 new mode 100644 index 779f0c2c8..65091bfa6 --- a/src/cli.ts +++ b/src/cli.ts @@ -9,7 +9,7 @@ installSigchldHandler(); import { installSignalHandlers as installCleanupSignalHandlers } from './core/process-cleanup.ts'; installCleanupSignalHandlers(); -import { readFileSync, existsSync, unlinkSync } from 'fs'; +import { readFileSync, existsSync, unlinkSync, fstatSync } from 'fs'; import { spawn } from 'child_process'; import { readUpdateCache, @@ -344,6 +344,11 @@ async function main() { // them out of the engine try/catch is safe and unlocks routing. const params = parseOpArgs(op, subArgs); + // #3513: stdin fill moved out of parseOpArgs so a non-TTY stdin with no + // piped input can't block the parse forever — the bounded read leaves the + // param unset on timeout and the required-param check below fails fast. + await applyStdinParam(op, params); + // v0.27.1 (`gbrain query --image `): swap the `image` param from // a filesystem path into base64 bytes + mime. The op accepts base64; the // CLI accepts a path. Helper is exported so tests can exercise the @@ -804,20 +809,97 @@ export function parseOpArgs(op: Operation, args: string[]): Record MAX_STDIN) { - console.error(`Error: stdin content exceeds ${MAX_STDIN} bytes. Split into smaller inputs.`); - process.exit(1); - } - params[op.cliHints.stdin] = stdinContent; - } - return params; } +/** + * #3513: read stdin into an op's stdin-capable param without ever blocking + * forever. The old inline `readFileSync(0)` in parseOpArgs assumed non-TTY + * implies piped content; a non-TTY stdin with NO input (CI step, cron job, + * agent harness holding an unwritten pipe open) blocked the read until kill. + * + * Strategy by fd kind (fstat): + * - TTY: skip, as before (interactive input is not an op-param source). + * - regular file / /dev/null / anything not a pipe or socket: readFileSync + * returns without blocking (`gbrain put x < file`, `< /dev/null` → ''). + * - FIFO/socket: stream-read with a deadline on the FIRST byte only. A real + * pipe (`echo foo | gbrain put x`, heredocs) delivers its first byte + * within milliseconds; once any data arrives the deadline is lifted and + * we read to EOF like readFileSync did (slow producers stay supported). + * An empty-but-closed pipe (`: | gbrain put x`) EOFs immediately → ''. + * A pipe that never delivers a byte times out → param stays unset, so + * the existing required-param usage error fires (fail fast, exit 1). + * + * GBRAIN_STDIN_TIMEOUT_MS overrides the first-byte deadline (default 5000). + * Exported for tests; called by the op dispatch right after parseOpArgs. + */ +export async function applyStdinParam( + op: Operation, + params: Record, +): Promise { + if (!op.cliHints?.stdin || params[op.cliHints.stdin] || process.stdin.isTTY) return; + const content = await readStdinBounded(); + if (content === null) return; // no input arrived — let the required-param check fail fast + const MAX_STDIN = 5_000_000; // 5MB + if (Buffer.byteLength(content, 'utf-8') > MAX_STDIN) { + console.error(`Error: stdin content exceeds ${MAX_STDIN} bytes. Split into smaller inputs.`); + process.exit(1); + } + params[op.cliHints.stdin] = content; +} + +/** First-byte deadline for pipe/socket stdin (#3513). Env-overridable escape hatch. */ +function stdinFirstByteTimeoutMs(): number { + const n = Number(process.env.GBRAIN_STDIN_TIMEOUT_MS); + return Number.isFinite(n) && n > 0 ? n : 5000; +} + +/** + * Returns the full stdin content, '' for a readable-but-empty stdin, or + * null when stdin is a pipe/socket that never delivered a byte within the + * first-byte deadline (or the fd is closed/unreadable). + */ +export async function readStdinBounded(): Promise { + let isPipeOrSocket: boolean; + try { + const st = fstatSync(0); + isPipeOrSocket = st.isFIFO() || st.isSocket(); + } catch { + return null; // closed/invalid fd — treat as no input + } + if (!isPipeOrSocket) { + // Regular file redirect, /dev/null, etc. — read returns without blocking. + try { + return readFileSync(0, 'utf-8'); + } catch { + return null; + } + } + return await new Promise((resolve) => { + const chunks: Buffer[] = []; + let gotData = false; + const timer = setTimeout(() => { + if (!gotData) { + process.stdin.destroy(); + resolve(null); + } + }, stdinFirstByteTimeoutMs()); + const finish = () => { + clearTimeout(timer); + resolve(Buffer.concat(chunks).toString('utf-8')); + }; + process.stdin.on('data', (c: Buffer) => { + if (!gotData) { + gotData = true; + clearTimeout(timer); // deadline applies to the FIRST byte only + } + chunks.push(c); + }); + process.stdin.once('end', finish); + process.stdin.once('error', finish); + }); +} + /** * #2098: thin-client source scoping. Locally, --source / GBRAIN_SOURCE / * .gbrain-source resolve to ctx.sourceId in makeContext; the thin-client diff --git a/test/cli-stdin-hang.test.ts b/test/cli-stdin-hang.test.ts new file mode 100644 index 000000000..38459b4c8 --- /dev/null +++ b/test/cli-stdin-hang.test.ts @@ -0,0 +1,173 @@ +/** + * #3513: parseOpArgs' stdin read must never block forever. + * + * In a non-TTY with no piped input — a CI step, a cron job, an agent + * harness that inherits a non-TTY stdin without writing to it — the old + * inline `readFileSync(0)` never returned. The fix bounds the read with a + * first-byte deadline (pipes/sockets only) and falls through to the + * existing required-param usage error on timeout. + * + * The load-bearing regression test spawns the REAL CLI with a held-open, + * never-written pipe: on pre-fix code it hangs until our observation window + * kills it; on fixed code it exits 1 with the usage error well inside the + * window. The stdin read + required-param check both run BEFORE engine + * connect, so no brain/DB is touched. + */ +import { describe, expect, test } from 'bun:test'; +import { dirname, join } from 'path'; + +const REPO = dirname(import.meta.dir); +const CLI = join(REPO, 'src', 'cli.ts'); + +interface CliRun { + exited: boolean; + exitCode: number | null; + stderr: string; +} + +/** Narrow Bun's `number | FileSink` stdin union to the pipe sink. */ +function pipeSink(proc: { stdin: unknown }): { write(d: string): unknown; end(): unknown } { + const s = proc.stdin; + if (!s || typeof s === 'number') throw new Error('expected a piped stdin sink'); + return s as { write(d: string): unknown; end(): unknown }; +} + +/** + * Spawn the CLI with the given stdin wiring. `holdPipeOpen` keeps the write + * end of the stdin pipe alive without ever writing — the #3513 repro. The + * observation window kills the child if it hasn't exited (pre-fix hang). + */ +async function runCliWithStdin( + args: string[], + stdin: 'hold-open' | 'closed-empty' | { data: string } | { file: string }, + windowMs: number, +): Promise { + const proc = Bun.spawn(['bun', 'run', CLI, ...args], { + cwd: REPO, + env: { ...process.env, GBRAIN_STDIN_TIMEOUT_MS: '500' }, + stdin: typeof stdin === 'object' && 'file' in stdin ? Bun.file(stdin.file) : 'pipe', + stdout: 'pipe', + stderr: 'pipe', + }); + if (typeof stdin === 'object' && 'data' in stdin) { + pipeSink(proc).write(stdin.data); + await pipeSink(proc).end(); + } else if (stdin === 'closed-empty') { + await pipeSink(proc).end(); + } + // 'hold-open': never write, never close — the CI/cron/agent-harness shape. + + let exited = true; + const killer = setTimeout(() => { + exited = false; + try { proc.kill('SIGKILL'); } catch { /* already dead */ } + }, windowMs); + const [exitCode, stderr] = await Promise.all([ + proc.exited, + new Response(proc.stderr).text(), + ]); + clearTimeout(killer); + try { pipeSink(proc).end(); } catch { /* hold-open cleanup */ } + return { exited, exitCode: exited ? exitCode : null, stderr }; +} + +describe('#3513 — stdin-capable op with a non-TTY, never-written stdin', () => { + test('exits fast with the usage error instead of blocking forever', async () => { + // `put` declares stdin:'content' (required). No inline content, no piped + // input → the bounded read times out at 500ms, content stays unset, and + // the required-param check prints usage and exits 1. Pre-fix: readFileSync(0) + // blocks until the 20s window kills the child. + const run = await runCliWithStdin(['put', 'stdin-hang-test-slug'], 'hold-open', 20_000); + expect(run.exited).toBe(true); // pre-#3513 this is false: the read never returns + expect(run.exitCode).toBe(1); + expect(run.stderr).toContain('Usage: gbrain put'); + }, 30_000); + + test('a genuine pipe with data is still consumed (no hang, no crash)', async () => { + // Piped content fills `content`; the missing positional slug then fails + // the required check — proving the stream path read stdin and moved on. + const run = await runCliWithStdin(['put'], { data: '# hello\n' }, 20_000); + expect(run.exited).toBe(true); + expect(run.exitCode).toBe(1); + expect(run.stderr).toContain('Usage: gbrain put'); + }, 30_000); + + test('empty-but-real input (`< /dev/null`) does not hang', async () => { + const run = await runCliWithStdin(['put'], { file: '/dev/null' }, 20_000); + expect(run.exited).toBe(true); + expect(run.exitCode).toBe(1); + }, 30_000); + + test('an empty pipe that closes immediately does not hang', async () => { + const run = await runCliWithStdin(['put'], 'closed-empty', 20_000); + expect(run.exited).toBe(true); + expect(run.exitCode).toBe(1); + }, 30_000); +}); + +describe('#3513 — applyStdinParam content preservation (subprocess driver)', () => { + // Drive the exported helper in a child process so we control the child's + // real fd 0 — bun test's own stdin is not a reliable fixture. + const DRIVER = ` + const { applyStdinParam } = await import(${JSON.stringify(CLI)}); + const op = { name: 'put', params: { content: { type: 'string', required: true } }, cliHints: { stdin: 'content' } }; + const params = {}; + await applyStdinParam(op, params); + console.log(JSON.stringify(params)); + process.exit(0); + `; + + async function runDriver( + stdin: 'hold-open' | 'closed-empty' | { data: string } | { file: string }, + ): Promise<{ exited: boolean; params: Record | null }> { + const proc = Bun.spawn(['bun', '-e', DRIVER], { + cwd: REPO, + env: { ...process.env, GBRAIN_STDIN_TIMEOUT_MS: '500' }, + stdin: typeof stdin === 'object' && 'file' in stdin ? Bun.file(stdin.file) : 'pipe', + stdout: 'pipe', + stderr: 'pipe', + }); + if (typeof stdin === 'object' && 'data' in stdin) { + pipeSink(proc).write(stdin.data); + await pipeSink(proc).end(); + } else if (stdin === 'closed-empty') { + await pipeSink(proc).end(); + } + let exited = true; + const killer = setTimeout(() => { + exited = false; + try { proc.kill('SIGKILL'); } catch { /* already dead */ } + }, 20_000); + const [stdout] = await Promise.all([new Response(proc.stdout).text(), proc.exited]); + clearTimeout(killer); + try { pipeSink(proc).end(); } catch { /* hold-open cleanup */ } + const line = stdout.trim().split('\n').pop() ?? ''; + let params: Record | null = null; + try { params = JSON.parse(line); } catch { /* child killed before printing */ } + return { exited, params }; + } + + test('piped data lands in the stdin param verbatim', async () => { + const { exited, params } = await runDriver({ data: '---\ntitle: x\n---\nbody' }); + expect(exited).toBe(true); + expect(params?.content).toBe('---\ntitle: x\n---\nbody'); + }, 30_000); + + test('/dev/null yields empty-string content (readable, empty — pre-fix parity)', async () => { + const { exited, params } = await runDriver({ file: '/dev/null' }); + expect(exited).toBe(true); + expect(params?.content).toBe(''); + }, 30_000); + + test('empty closed pipe yields empty-string content', async () => { + const { exited, params } = await runDriver('closed-empty'); + expect(exited).toBe(true); + expect(params?.content).toBe(''); + }, 30_000); + + test('held-open pipe times out and leaves the param unset', async () => { + const { exited, params } = await runDriver('hold-open'); + expect(exited).toBe(true); // completes inside the window instead of hanging + expect(params).toEqual({}); + }, 30_000); +});