fix(cli): stop parseOpArgs hanging on a non-TTY stdin with no input (#3513)

parseOpArgs read stdin for stdin-capable ops via readFileSync(0), assuming non-TTY implies piped content. In a non-TTY with no piped input — a CI step, a cron job, an agent harness that inherits a non-TTY stdin without ever writing to it — that call never returns. The stdin fill moves out of parseOpArgs into an async applyStdinParam with a bounded read (readStdinBounded), called by the op dispatch right after arg parsing: - TTY: skipped, as before. - Regular file / /dev/null (fstat says not a pipe/socket): readFileSync returns without blocking — `gbrain put x < file` and `< /dev/null` behave exactly as before (empty-but-readable still yields ''). - FIFO/socket: stream-read with a deadline on the FIRST byte only (default 5000ms, GBRAIN_STDIN_TIMEOUT_MS overrides). Real pipes (`echo foo | gbrain put x`, heredocs) deliver their first byte in milliseconds; once any data arrives the deadline lifts and the read drains to EOF, so slow producers keep working. An empty pipe that closes yields ''. A pipe that never delivers a byte times out, the param stays unset, and the existing required-param usage error fails fast with exit 1. Regression tests spawn the real CLI with a held-open, never-written pipe (hangs 20s+ on pre-fix code; exits ~1s fixed) plus parity cases for data pipes, /dev/null, and empty closed pipes, and a subprocess driver pinning content preservation through applyStdinParam. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-31 04:07:52 +00:00 · 2026-07-28 13:32:03 -07:00
parent 6920744dd8
commit 33c3b79a7f
2 changed files with 267 additions and 12 deletions
@@ -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 <path>`): 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<string, unkno
    }
  }

-  // Read stdin for content params
-  if (op.cliHints?.stdin && !params[op.cliHints.stdin] && !process.stdin.isTTY) {
-    const stdinContent = readFileSync(0, 'utf-8');
-    const MAX_STDIN = 5_000_000; // 5MB
-    if (Buffer.byteLength(stdinContent, 'utf-8') > 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<string, unknown>,
+): Promise<void> {
+  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<string | null> {
+  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<string | null>((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
@@ -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<CliRun> {
+  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<string, unknown> | 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<string, unknown> | 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);
+});