diff --git a/js/.changeset/stream-compound-output.md b/js/.changeset/stream-compound-output.md new file mode 100644 index 0000000..4aee138 --- /dev/null +++ b/js/.changeset/stream-compound-output.md @@ -0,0 +1,5 @@ +--- +'command-stream': patch +--- + +Stream compound commands progressively while preserving virtual-command dispatch, nested-process cancellation, and safe real-shell fallback. diff --git a/js/BEST-PRACTICES.md b/js/BEST-PRACTICES.md index 01677de..9e01899 100644 --- a/js/BEST-PRACTICES.md +++ b/js/BEST-PRACTICES.md @@ -8,6 +8,7 @@ This document covers best practices, common patterns, and pitfalls to avoid when - [String Interpolation](#string-interpolation) - [Security Best Practices](#security-best-practices) - [Error Handling](#error-handling) +- [Real-time Streaming](#real-time-streaming) - [Performance Tips](#performance-tips) - [Common Pitfalls](#common-pitfalls) - [7. try/catch for Exit Code Detection (Silent Bug)](#7-trycatch-for-exit-code-detection-silent-bug) @@ -207,6 +208,129 @@ switch (result.code) { --- +## Real-time Streaming + +### Stream Compound Commands Progressively + +Use `stream()` when code must react to output before the full command finishes. +Sequences containing `;`, `&&`, `||`, subshells, or pipelines preserve +command-stream built-ins and registered virtual commands. Output from nested +system processes is forwarded as soon as it arrives: + +```javascript +import { $ } from 'command-stream'; + +const build = $({ + mirror: false, +})`echo "build started"; npm run build; echo "build finished"`; + +for await (const chunk of build.stream()) { + if (chunk.type === 'stdout') { + consumeBuildOutput(chunk.data); + } else if (chunk.type === 'stderr') { + consumeBuildError(chunk.data); + } else if (chunk.type === 'exit') { + console.log('build exited with', chunk.code); + } +} +``` + +Always check `chunk.type`: the final `{ type: 'exit', code }` chunk has no +`data` property. + +Breaking out of the loop stops the currently active command, including a real +process nested inside a compound command. Keep a runner reference when you also +need explicit cancellation: + +```javascript +const watcher = $`echo "watching"; npm run watch`; + +for await (const chunk of watcher.stream()) { + if (chunk.type === 'stdout' && isReady(chunk.data)) { + watcher.kill('SIGINT'); + } +} +``` + +### Choose the Right Execution Model + +JavaScript process libraries generally use one of three models: + +1. **Parse and interpret shell syntax in the library.** [Bun Shell](https://bun.sh/docs/runtime/shell) + uses a lexer, parser, and interpreter with cross-platform built-ins. This is + the model command-stream uses for supported operators so built-in and custom + virtual commands remain available. +2. **Run an explicit system shell.** [zx](https://google.github.io/zx/shell) + is shell-oriented, while [Node's `child_process`](https://nodejs.org/api/child_process.html) + exposes a `shell` option. This is appropriate for trusted scripts that need + platform shell features such as command substitution, glob expansion, file + descriptor redirection, or background jobs. command-stream automatically + falls back to the system shell when a compound command uses unsupported + syntax. +3. **Spawn executables directly and compose streams programmatically.** + [Execa](https://github.com/sindresorhus/execa) emphasizes direct process + execution, iterable output, and programmatic pipelines; Node also exposes + piped child streams directly. In command-stream, prefer `.pipe()` when the + pipeline structure is built dynamically or should not depend on shell + syntax. + +Use the internal command-stream path for portable built-ins, registered virtual +commands, and ordinary sequences. Use a real shell only when its extra syntax +is required, because behavior then depends on the installed shell and custom +virtual commands do not exist inside that external process. Use `.pipe()` for +explicit application-controlled composition. + +### Streaming Pitfalls + +```javascript +// WRONG: waits for completion before application code can consume stdout. +const result = await $({ mirror: false })`long-running-build`; +consumeBuildOutput(result.stdout); + +// CORRECT: consumes each chunk while the build is still running. +for await (const chunk of $({ mirror: false })`long-running-build`.stream()) { + if (chunk.type === 'stdout') consumeBuildOutput(chunk.data); +} +``` + +Do not wrap a sequence in `sh -c` merely to make it stream. That creates a +separate shell where registered virtual commands are unavailable: + +```javascript +import { $, register } from 'command-stream'; + +register('project-status', statusHandler); + +// WRONG: the external shell cannot resolve the JavaScript virtual command. +await $`sh -c 'echo checking; project-status'`; + +// CORRECT: command-stream keeps both commands in its own orchestration. +await $`echo checking; project-status`; +``` + +Never insert untrusted text as raw shell syntax: + +```javascript +import { $, raw } from 'command-stream'; + +// WRONG: raw user input can introduce operators or command substitution. +await $`${raw(request.body.command)}`; + +// CORRECT: interpolation treats external data as an argument. +await $`project-status ${request.body.project}`; +``` + +Finally, command-stream can forward only what a producer writes. Some programs +buffer output when stdout is a pipe; use that program's line-buffered, +unbuffered, or watch option when available. Increasing a test timeout does not +fix producer-side buffering. + +See +[`examples/streaming-compound-commands.mjs`](./examples/streaming-compound-commands.mjs) +for a runnable compound-stream example. + +--- + ## Performance Tips ### Use Streaming for Large Outputs diff --git a/js/README.md b/js/README.md index 85649f3..50cbd9a 100644 --- a/js/README.md +++ b/js/README.md @@ -504,6 +504,27 @@ arrives, followed by a final `{ type: 'exit', code }` chunk when the process exits. Always guard on `chunk.type` before reading `chunk.data`, since the `exit` chunk carries `code` instead of `data`. +Compound commands stream progressively too. Supported sequences keep built-in +and registered virtual commands in command-stream while forwarding nested +system-process output in real time: + +```javascript +const build = $({ + mirror: false, +})`echo "build started"; npm run build; echo "build finished"`; + +for await (const chunk of build.stream()) { + if (chunk.type === 'stdout') { + consumeBuildOutput(chunk.data); + } +} +``` + +Breaking the loop or calling `build.kill()` also stops the active nested +command. For guidance on system-shell fallback, programmatic pipelines, +producer-side buffering, and patterns to avoid, see +[Real-time Streaming in Best Practices](BEST-PRACTICES.md#real-time-streaming). + The iterator terminates as soon as the process exits, even if a grandchild keeps the stdout/stderr pipe open (e.g. `sh -c 'background-task & echo done'`). Any output still buffered is drained within a short grace period (the `exitPumpGrace` diff --git a/js/examples/streaming-compound-commands.mjs b/js/examples/streaming-compound-commands.mjs new file mode 100644 index 0000000..19e98cf --- /dev/null +++ b/js/examples/streaming-compound-commands.mjs @@ -0,0 +1,16 @@ +import { $ } from '../src/$.mjs'; + +const startedAt = Date.now(); +const command = $({ + mirror: false, +})`echo "build started"; sleep 0.25; echo "build finished"`; + +for await (const chunk of command.stream()) { + const elapsed = `${Date.now() - startedAt}ms`; + + if (chunk.type === 'stdout' || chunk.type === 'stderr') { + process.stdout.write(`[${elapsed}] ${chunk.type}: ${chunk.data}`); + } else if (chunk.type === 'exit') { + console.log(`[${elapsed}] exit: ${chunk.code}`); + } +} diff --git a/js/src/$.process-runner-base.mjs b/js/src/$.process-runner-base.mjs index e5702d7..c6f6911 100644 --- a/js/src/$.process-runner-base.mjs +++ b/js/src/$.process-runner-base.mjs @@ -241,6 +241,7 @@ class ProcessRunner extends StreamEmitter { this._cancelled = false; this._cancellationSignal = null; this._virtualGenerator = null; + this._activeNestedRunner = null; this._abortController = new AbortController(); // Set to true once user code starts consuming this runner (await / then / diff --git a/js/src/$.process-runner-execution.mjs b/js/src/$.process-runner-execution.mjs index dba273d..bdc5c4b 100644 --- a/js/src/$.process-runner-execution.mjs +++ b/js/src/$.process-runner-execution.mjs @@ -26,23 +26,11 @@ function hasShellOperators(command) { command.includes('||') || command.includes('(') || command.includes(';') || + command.includes('&') || (command.includes('cd ') && command.includes('&&')) ); } -/** - * Check if command is a streaming pattern - * @param {string} command - Command to check - * @returns {boolean} - */ -function isStreamingPattern(command) { - return ( - command.includes('sleep') && - command.includes(';') && - (command.includes('echo') || command.includes('printf')) - ); -} - /** * Determine if shell operators should be used * @param {object} runner - ProcessRunner instance @@ -50,14 +38,7 @@ function isStreamingPattern(command) { * @returns {boolean} */ function shouldUseShellOperators(runner, command) { - const hasOps = hasShellOperators(command); - const isStreaming = isStreamingPattern(command); - return ( - runner.options.shellOperators && - hasOps && - !isStreaming && - !runner._isStreaming - ); + return runner.options.shellOperators && hasShellOperators(command); } /** @@ -863,6 +844,7 @@ async function handleShellMode(runner, deps) { ); const useShellOps = shouldUseShellOperators(runner, command); + const requiresRealShell = useShellOps && needsRealShell(command); trace( 'ProcessRunner', @@ -870,18 +852,18 @@ async function handleShellMode(runner, deps) { `Shell operator detection | ${JSON.stringify({ hasShellOperators: hasShellOperators(command), shellOperatorsEnabled: runner.options.shellOperators, - isStreamingPattern: isStreamingPattern(command), isStreaming: runner._isStreaming, shouldUseShellOperators: useShellOps, + requiresRealShell, command: command.slice(0, 100), })}` ); - if ( - !runner.options._bypassVirtual && - useShellOps && - !needsRealShell(command) - ) { + if (requiresRealShell) { + return null; + } + + if (!runner.options._bypassVirtual && useShellOps) { const result = await tryEnhancedShellParser(runner, command); if (result) { return result; @@ -1127,7 +1109,7 @@ export function attachExecutionMethods(ProcessRunner, deps) { if (this.spec.mode === 'shell') { const shellResult = await handleShellMode(this, deps); if (shellResult) { - return shellResult; + return this.finish(shellResult); } } diff --git a/js/src/$.process-runner-orchestration.mjs b/js/src/$.process-runner-orchestration.mjs index 0cde381..6c66c2f 100644 --- a/js/src/$.process-runner-orchestration.mjs +++ b/js/src/$.process-runner-orchestration.mjs @@ -148,6 +148,10 @@ export function attachOrchestrationMethods(ProcessRunner, deps) { let combinedStderr = ''; for (let i = 0; i < sequence.commands.length; i++) { + if (this._cancelled) { + break; + } + const command = sequence.commands[i]; const operator = i > 0 ? sequence.operators[i - 1] : null; @@ -217,7 +221,7 @@ export function attachOrchestrationMethods(ProcessRunner, deps) { if (isVirtualCommandsEnabled() && virtualCommands.has(cmd)) { trace('ProcessRunner', () => `Using virtual command: ${cmd}`); const argValues = args.map((a) => a.value || a); - const result = await this._runVirtual(cmd, argValues); + const result = await this._runVirtual(cmd, argValues, null, false); await handleRedirects(result, redirects); return result; } @@ -234,7 +238,23 @@ export function attachOrchestrationMethods(ProcessRunner, deps) { { ...this.options, cwd: currentCwd, _bypassVirtual: true } ); - return await runner; + const forwardData = (chunk) => { + if (!this._cancelled && !this.finished) { + this._emitProcessedData(chunk.type, chunk.data); + } + }; + + runner.on('data', forwardData); + this._activeNestedRunner = runner; + + try { + return await runner; + } finally { + runner.off('data', forwardData); + if (this._activeNestedRunner === runner) { + this._activeNestedRunner = null; + } + } }; ProcessRunner.prototype.pipe = function (destination) { diff --git a/js/src/$.process-runner-stream-kill.mjs b/js/src/$.process-runner-stream-kill.mjs index cb9d921..4d3201c 100644 --- a/js/src/$.process-runner-stream-kill.mjs +++ b/js/src/$.process-runner-stream-kill.mjs @@ -216,6 +216,15 @@ function killRunner(runner, signal) { runner._cancellationSignal = signal; killPipelineComponents(runner.spec, signal); + if ( + runner._activeNestedRunner && + !runner._activeNestedRunner.finished && + typeof runner._activeNestedRunner.kill === 'function' + ) { + trace('ProcessRunner', () => 'Killing active nested command'); + runner._activeNestedRunner.kill(signal); + } + if (runner._cancelResolve) { trace('ProcessRunner', () => 'Resolving cancel promise'); runner._cancelResolve(); diff --git a/js/src/$.process-runner-virtual.mjs b/js/src/$.process-runner-virtual.mjs index efe07db..78103ee 100644 --- a/js/src/$.process-runner-virtual.mjs +++ b/js/src/$.process-runner-virtual.mjs @@ -77,7 +77,7 @@ function emitOutput(runner, type, data) { * @param {object} shellSettings - Global shell settings * @returns {object} Result object */ -function handleVirtualError(runner, error, shellSettings) { +function handleVirtualError(runner, error, shellSettings, shouldFinish) { let exitCode = error.code ?? 1; if (runner._cancelled && runner._cancellationSignal) { exitCode = getCancellationExitCode(runner._cancellationSignal); @@ -91,7 +91,9 @@ function handleVirtualError(runner, error, shellSettings) { }; emitOutput(runner, 'stderr', result.stderr); - runner.finish(result); + if (shouldFinish) { + runner.finish(result); + } if (shellSettings.errexit) { error.result = result; @@ -236,7 +238,8 @@ export function attachVirtualCommandMethods(ProcessRunner, deps) { ProcessRunner.prototype._runVirtual = async function ( cmd, args, - originalCommand = null + originalCommand = null, + shouldFinish = true ) { trace('ProcessRunner', () => `_runVirtual | cmd=${cmd}`); @@ -278,7 +281,9 @@ export function attachVirtualCommandMethods(ProcessRunner, deps) { ? await runGeneratorHandler(this, handler, argValues, stdinData) : await runRegularHandler(this, handler, argValues, stdinData); - this.finish(result); + if (shouldFinish) { + this.finish(result); + } if (globalShellSettings.errexit && result.code !== 0) { const error = new Error(`Command failed with exit code ${result.code}`); @@ -291,7 +296,7 @@ export function attachVirtualCommandMethods(ProcessRunner, deps) { return result; } catch (error) { - return handleVirtualError(this, error, globalShellSettings); + return handleVirtualError(this, error, globalShellSettings, shouldFinish); } }; } diff --git a/js/src/commands/$.sleep.mjs b/js/src/commands/$.sleep.mjs index c31e807..9b7ef3d 100644 --- a/js/src/commands/$.sleep.mjs +++ b/js/src/commands/$.sleep.mjs @@ -24,7 +24,45 @@ export default async function sleep({ args, abortSignal, isCancelled }) { // Use abort signal if available, otherwise use setTimeout try { await new Promise((resolve, reject) => { - const timeoutId = setTimeout(resolve, seconds * 1000); + let settled = false; + let checkInterval; + + const cleanup = () => { + clearTimeout(timeoutId); + if (checkInterval) { + clearInterval(checkInterval); + } + abortSignal?.removeEventListener('abort', handleAbort); + }; + + const settle = (callback, value) => { + if (settled) { + return; + } + settled = true; + cleanup(); + callback(value); + }; + + const complete = () => settle(resolve); + const cancel = () => settle(reject, new Error('Sleep cancelled')); + const handleAbort = () => { + trace( + 'VirtualCommand', + () => + `sleep: abort signal received | ${JSON.stringify( + { + seconds, + signalAborted: abortSignal.aborted, + }, + null, + 2 + )}` + ); + cancel(); + }; + + const timeoutId = setTimeout(complete, seconds * 1000); // Handle cancellation via abort signal if (abortSignal) { @@ -40,22 +78,7 @@ export default async function sleep({ args, abortSignal, isCancelled }) { )}` ); - abortSignal.addEventListener('abort', () => { - trace( - 'VirtualCommand', - () => - `sleep: abort signal received | ${JSON.stringify( - { - seconds, - signalAborted: abortSignal.aborted, - }, - null, - 2 - )}` - ); - clearTimeout(timeoutId); - reject(new Error('Sleep cancelled')); - }); + abortSignal.addEventListener('abort', handleAbort, { once: true }); // Check if already aborted if (abortSignal.aborted) { @@ -64,8 +87,7 @@ export default async function sleep({ args, abortSignal, isCancelled }) { () => `sleep: signal already aborted | ${JSON.stringify({ seconds }, null, 2)}` ); - clearTimeout(timeoutId); - reject(new Error('Sleep cancelled')); + cancel(); return; } } else { @@ -83,18 +105,17 @@ export default async function sleep({ args, abortSignal, isCancelled }) { () => `sleep: setting up isCancelled polling | ${JSON.stringify({ seconds }, null, 2)}` ); - const checkInterval = setInterval(() => { + checkInterval = setInterval(() => { if (isCancelled()) { trace( 'VirtualCommand', () => `sleep: isCancelled returned true | ${JSON.stringify({ seconds }, null, 2)}` ); - clearTimeout(timeoutId); - clearInterval(checkInterval); - reject(new Error('Sleep cancelled')); + cancel(); } }, 100); + checkInterval.unref?.(); } }); diff --git a/js/src/shell-parser.mjs b/js/src/shell-parser.mjs index d9a3506..f276937 100644 --- a/js/src/shell-parser.mjs +++ b/js/src/shell-parser.mjs @@ -152,6 +152,18 @@ function tokenize(command) { break; } + // A lone ampersand is a background-process operator, which this parser + // intentionally leaves to the real shell. Guard here as well as in + // needsRealShell() so malformed input can never leave the tokenizer at the + // same index forever. + if ( + command[i] === '&' && + command[i - 1] !== '&' && + command[i + 1] !== '&' + ) { + throw new Error('Background commands require a real shell'); + } + // Check for operators const operator = matchOperator(command, i); if (operator) { @@ -413,28 +425,83 @@ export function parseShellCommand(command) { } } +function scanQuotedFeature(command, index, quote) { + const char = command[index]; + + if (quote === "'") { + return { + quote: char === quote ? null : quote, + skip: 0, + unsupported: false, + }; + } + if (char === '\\') { + return { quote, skip: 1, unsupported: false }; + } + if (char === quote) { + return { quote: null, skip: 0, unsupported: false }; + } + + // Parameter and command substitution still occur in double quotes. + return { + quote, + skip: 0, + unsupported: char === '`' || char === '$', + }; +} + +function isSingleAmpersand(command, index) { + return ( + command[index] === '&' && + command[index - 1] !== '&' && + command[index + 1] !== '&' + ); +} + +function isUnsupportedUnquotedFeature(command, index) { + const char = command[index]; + if ('`$~*?['.includes(char) || isSingleAmpersand(command, index)) { + return true; + } + + const remainder = command.slice(index); + return ( + remainder.startsWith('2>') || + remainder.startsWith('&>') || + remainder.startsWith('>&') || + remainder.startsWith('<<') + ); +} + /** * Check if a command needs shell features we don't handle */ export function needsRealShell(command) { - // Check for features we don't handle yet - const unsupported = [ - '`', // Command substitution - '$(', // Command substitution - '${', // Variable expansion - '~', // Home expansion (at start of word) - '*', // Glob patterns - '?', // Glob patterns - '[', // Glob patterns - '2>', // stderr redirection - '&>', // Combined redirection - '>&', // File descriptor duplication - '<<', // Here documents - '<<<', // Here strings - ]; - - for (const feature of unsupported) { - if (command.includes(feature)) { + let quote = null; + for (let i = 0; i < command.length; i++) { + const char = command[i]; + + if (quote) { + const scan = scanQuotedFeature(command, i, quote); + if (scan.unsupported) { + return true; + } + quote = scan.quote; + i += scan.skip; + continue; + } + + if (char === '\\') { + i++; + continue; + } + + if (char === '"' || char === "'") { + quote = char; + continue; + } + + if (isUnsupportedUnquotedFeature(command, i)) { return true; } } diff --git a/js/tests/issue-43-stream-output.test.mjs b/js/tests/issue-43-stream-output.test.mjs new file mode 100644 index 0000000..3170886 --- /dev/null +++ b/js/tests/issue-43-stream-output.test.mjs @@ -0,0 +1,217 @@ +import { expect, test } from 'bun:test'; +import { rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import './test-helper.mjs'; +import { $, enableVirtualCommands, register, unregister } from '../src/$.mjs'; + +function nextWithin(iterator, timeoutMs = 1000) { + let timer; + const timeout = new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`stream did not yield within ${timeoutMs}ms`)), + timeoutMs + ); + }); + + return Promise.race([iterator.next(), timeout]).finally(() => + clearTimeout(timer) + ); +} + +test('streams a virtual command before the next command in a sequence completes', async () => { + enableVirtualCommands(); + + let releaseBlockedCommand; + const blockedCommand = new Promise((resolve) => { + releaseBlockedCommand = resolve; + }); + + let markBlockedCommandStarted; + const blockedCommandStarted = new Promise((resolve) => { + markBlockedCommandStarted = resolve; + }); + + register('issue-43-blocked', async () => { + markBlockedCommandStarted(); + await blockedCommand; + return { code: 0, stdout: 'after\n' }; + }); + + const runner = $({ + mirror: false, + })`echo before; issue-43-blocked`; + const iterator = runner.stream()[Symbol.asyncIterator](); + + try { + const first = await nextWithin(iterator); + expect(first.done).toBe(false); + expect(first.value.type).toBe('stdout'); + expect(first.value.data.toString()).toBe('before\n'); + + await blockedCommandStarted; + + const pendingSecond = iterator.next(); + const prematureResult = await Promise.race([ + pendingSecond.then(() => 'yielded'), + new Promise((resolve) => setTimeout(() => resolve('blocked'), 25)), + ]); + expect(prematureResult).toBe('blocked'); + + releaseBlockedCommand(); + + const second = await pendingSecond; + expect(second.done).toBe(false); + expect(second.value.type).toBe('stdout'); + expect(second.value.data.toString()).toBe('after\n'); + + const exit = await nextWithin(iterator); + expect(exit.done).toBe(false); + expect(exit.value).toEqual({ type: 'exit', code: 0 }); + + const end = await nextWithin(iterator); + expect(end.done).toBe(true); + } finally { + releaseBlockedCommand(); + await iterator.return(); + unregister('issue-43-blocked'); + } +}, 3000); + +test('forwards output from a real command nested in a streamed sequence', async () => { + enableVirtualCommands(); + + const releaseFile = join( + tmpdir(), + `command-stream-issue-43-${process.pid}-${Date.now()}` + ); + const childScript = [ + 'import { existsSync } from "node:fs";', + 'process.stdout.write("child-ready\\n");', + 'while (!existsSync(process.argv.at(1))) {', + ' await new Promise((resolve) => setTimeout(resolve, 10));', + '}', + 'process.stdout.write("child-done\\n");', + ].join(' '); + + const runner = $({ + mirror: false, + })`echo before; ${process.execPath} -e ${childScript} ${releaseFile}`; + const iterator = runner.stream()[Symbol.asyncIterator](); + + try { + const first = await nextWithin(iterator); + expect(first.value.type).toBe('stdout'); + expect(first.value.data.toString()).toBe('before\n'); + + const ready = await nextWithin(iterator); + expect(ready.value.type).toBe('stdout'); + expect(ready.value.data.toString()).toBe('child-ready\n'); + + writeFileSync(releaseFile, ''); + + const done = await nextWithin(iterator); + expect(done.value.type).toBe('stdout'); + expect(done.value.data.toString()).toBe('child-done\n'); + + const exit = await nextWithin(iterator); + expect(exit.value).toEqual({ type: 'exit', code: 0 }); + } finally { + writeFileSync(releaseFile, ''); + await iterator.return(); + rmSync(releaseFile, { force: true }); + } +}, 3000); + +test('keeps quoted shell metacharacters inside virtual command arguments', async () => { + enableVirtualCommands(); + register('issue-43-custom', async () => ({ + code: 0, + stdout: 'custom\n', + })); + + try { + const stdout = []; + for await (const chunk of $({ + mirror: false, + })`echo "literal; * [ ?"; issue-43-custom`.stream()) { + if (chunk.type === 'stdout') { + stdout.push(chunk.data.toString()); + } + } + + expect(stdout.join('')).toBe('literal; * [ ?\ncustom\n'); + } finally { + unregister('issue-43-custom'); + } +}); + +test('breaking a sequence stream kills its active nested command', async () => { + enableVirtualCommands(); + + const childScript = [ + 'process.stdout.write("child-ready\\n");', + 'setInterval(function () {}, 1000);', + ].join(' '); + const runner = $({ + mirror: false, + })`echo before; ${process.execPath} -e ${childScript}`; + + let sawNestedOutput = false; + for await (const chunk of runner.stream()) { + if (chunk.type === 'stdout' && chunk.data.toString() === 'child-ready\n') { + sawNestedOutput = true; + break; + } + } + + expect(sawNestedOutput).toBe(true); + expect(runner.finished).toBe(true); + expect(runner.result.code).toBe(143); +}, 3000); + +test.skipIf(process.platform === 'win32')( + 'falls back to the real shell for a background operator', + async () => { + enableVirtualCommands(); + + const stdout = []; + for await (const chunk of $({ + mirror: false, + })`sleep 0.01 & echo background`.stream()) { + if (chunk.type === 'stdout') { + stdout.push(chunk.data.toString()); + } + } + + expect(stdout.join('')).toBe('background\n'); + } +); + +test('a streamed sequence with virtual sleep releases its timers', async () => { + const examplePath = fileURLToPath( + new URL('../examples/streaming-compound-commands.mjs', import.meta.url) + ); + const child = Bun.spawn([process.execPath, examplePath], { + stdout: 'pipe', + stderr: 'pipe', + }); + + let timer; + try { + const exitCode = await Promise.race([ + child.exited, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('streaming example did not exit')), + 1000 + ); + }), + ]); + expect(exitCode).toBe(0); + } finally { + clearTimeout(timer); + child.kill(); + } +}, 3000);