Stream compound command output progressively (#43)

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.

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

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`

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}`);
+  }
+}

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 /

js/src/$.process-runner-execution.mjs

@@ -26,38 +26,19 @@ 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
  * @param {string} command - Command to check
  * @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,25 +844,26 @@ async function handleShellMode(runner, deps) {
   );
 
   const useShellOps = shouldUseShellOperators(runner, command);
+  const requiresRealShell = useShellOps && needsRealShell(command);
 
   trace(
     'ProcessRunner',
     () =>
       `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);
         }
       }
 

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) {

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();

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);
     }
   };
 }