zx ProcessPromise 完全指南：Promise 语义下的进程控制、管道流与输出格式化

在 Node.js 脚本中处理子进程，长期要面对回调地狱、错误语义混乱和流管道拼接的麻烦。zx 的解法是让全局的 $ 模板标签直接返回一个 ProcessPromise 实例——它继承自原生 Promise，resolve 后得到 ProcessOutput 对象。本文基于 docs/process-promise.md 逐节展开：从 stage 状态机、stdin/stdout/stderr 流访问，到 pipe() 管道的缓冲回放与分叉合并、kill()/abort() 终止控制、stdio() 配置，以及 nothrow()、quiet()、verbose()、timeout() 等修饰方法，并结合 src/core.ts 中的 ProcessPromise 类源码，说明每个行为的底层实现依据。读完本文，你可以掌握在 zx 脚本中可靠地管理进程生命周期、构建任意拓扑的管道流、以及格式化进程输出的完整方案。

ProcessPromise 是什么：从 $ 到 ProcessOutput

最基本的用法：

const p = $`command` // ProcessPromise
const o = await p    // ProcessOutput

$ 默认会立即派生一个新进程；如果想延迟启动、手动触发执行，使用 halt: true 选项再调用 run()：

const p = $({halt: true})`command`
const o = await p.run()

从源码看，ProcessPromise 定义于 src/core.ts#L257：export class ProcessPromise extends Promise<ProcessOutput>，确实继承自原生 Promise，且泛型明确 resolve 值为 ProcessOutput。$ 模板标签在 src/core.ts#L194 中创建实例后，只有当 !pp.isHalted() 时才调用 pp.run()——这正是"默认立即启动、halt 后手动 run"行为的实现来源。run() 方法本身（src/core.ts#L309）会先通过 ProcessPromise.bus.runBack(this) 唤醒管道中等待它的上游，再检查 isRunning() || isSettled() 防止重复启动，然后把 _stage 置为 'running'，最终调用 vendor 层的 exec 完成真正的进程派生（src/core.ts#L334）。

值得注意的实现细节：$ 是一个基于 AsyncLocalStorage 的快照机制（src/core.ts#L155）。每次调用模板标签时都会生成一份 Snapshot（包含 ac: AbortController、ee: EventEmitter、cmd 等），作为该进程实例的私有配置快照。这也是后文 halt、nothrow、quiet 等方法能"只影响当前这条命令而不污染全局 $"的原因——它们修改的都是当前实例的 _snapshot。

stage：进程状态机

stage 属性展示当前进程所处的阶段，取值为：initial | halted | running | fulfilled | rejected。

const p = $`echo foo`
p.stage // 'running'
await p
p.stage // 'fulfilled'

源码中状态类型定义在 src/core.ts#L234：

type ProcessStage = 'initial' | 'halted' | 'running' | 'fulfilled' | 'rejected'

状态迁移由 finalize() 驱动（src/core.ts#L418）：当子进程结束并构造出 ProcessOutput 后，若 output.ok || this.isNothrow() 则 _stage = 'fulfilled' 并 resolve；否则置为 'rejected' 并 reject（同步模式下直接 throw）。halted 状态则在构造函数中由 snapshot.halt 决定（src/core.ts#L282）。因此 stage 是一个可靠的只读观察点，适合在长任务中做轮询判断。

stdin：可写的标准输入流

p.stdin 返回子进程的 stdin 可写流。访问该 getter 会触发以 stdio('pipe') 派生子进程。使用要点是不要忘记结束流：

const p = $`while read; do echo $REPLY; done`
p.stdin.write('Hello, World!\n')
p.stdin.end()

默认情况下，每个进程的 stdin 处于 inherit 模式。从源码看（src/core.ts#L529），getter 实际返回 this.child?.stdin；而实例初始持有一个 _stdin = new VoidStream()，进程尚未派生前对 stdin 的写入会先进入这个 VoidStream 缓冲，待 run() 时由 exec 的 stdin 参数接走（src/core.ts#L338）。

stdout/stderr：可读的标准输出流

p.stdout 和 p.stderr 返回对应的可读流，可以逐块消费：

const p = $`npm init`
for await (const chunk of p.stdout) {
  echo(chunk)
}

这里体现了一个设计取舍：await p 拿到的是拼接好的完整 ProcessOutput，而如果需要在进程运行期间实时处理输出，就使用流 API。stdout 与 stderr 是分离的，stdall（合并流）则体现在 ProcessOutput 层面。

exitCode：只取退出码

exitCode 是一个 promise，resolve 为进程退出码，且无论成功失败都能取到值：

if (await $`[[ -d path ]]`.exitCode == 0) {
  // ...
}

其实现非常简洁（src/core.ts#L541）：

get exitCode(): Promise<number | null> {
  return this.then(
    (o) => o.exitCode,
    (o) => o.exitCode
  )
}

即对 then 的成功和失败分支都取 o.exitCode——这就是为什么非零退出码不会在这里抛错。相比 nothrow()，它是"轻量版"：不改变异常行为，只提取退出码。

输出格式化器：json()、text()、lines()、buffer()、blob()

ProcessPromise 上的一组输出格式化方法，全部委托给 ProcessOutput 的同名方法：

const p = $`echo 'foo\nbar'`

await p.text()        // foo\nbar\n
await p.text('hex')   // 666f6f0a6261720a
await p.buffer()      // Buffer.from('foo\nbar\n')
await p.lines()       // ['foo', 'bar']

// 可以自定义分隔符：
await $`touch foo bar baz; find ./ -type f -print0`
  .lines('\0')        // ['./bar', './baz', './foo']

// 输出是合法 JSON 时原地解析：
await $`echo '{"foo": "bar"}'`
  .json()             // {foo: 'bar'}

对应源码在 src/core.ts#L577：每个方法都是 this.then((o) => o.xxx()) 的薄封装。ProcessOutput 端的实现（src/core.ts#L936）中，json() 对 stdall 做 JSON.parse，text(encoding) 支持非 UTF-8 编码（先转 buffer() 再解码），lines(delimiter) 走行迭代器并支持自定义分隔符。

进程元数据：pid、cwd、cmd、fullCmd

四个只读元数据 getter：

const p = $`sleep 1`
p.pid       // 进程 id
p.cwd       // 进程工作目录
p.cmd       // 命令: "sleep 1"
p.fullCmd   // 含前后缀的完整命令: "set -euo pipefail;sleep 1"

pid 来自 this.child?.pid（src/core.ts#L508），进程尚未派生时为 undefined；cmd 是模板标签拼出的原始命令，fullCmd 则是 prefix + cmd + postfix 的拼接（src/core.ts#L520）。从源码结构看，默认 shell 前缀为 set -euo pipefail;（见 src/core.ts#L1009 的 useBash()），这解释了 fullCmd 为何带这一前缀——它保证管道中任一环节失败都会让整条命令失败。

[Symbol.asyncIterator]：把进程当成逐行异步迭代器

ProcessPromise 实现了异步迭代器协议，直接迭代它的 stdout，按行产出：

const p = $`echo "Line1\nLine2\nLine3"`
for await (const line of p) {
  console.log(line)
}

// 可以指定自定义分隔符：
for await (const line of $({
  delimiter: '\0'
})`touch foo bar baz; find ./ -type f -print0`) {
  console.log(line)
}

实现位于 src/core.ts#L796，逻辑分三步：

1. 先遍历 this._zurk.store.stdout 中已缓冲的块（进程可能已经输出一部分）；
2. 再 for await 订阅实时 this.stdout 流，处理后续块；
3. 最后 await this，把进程退出码异常也传播给迭代调用方。

行切分使用 getLines(chunk, memo, dlmtr)，默认分隔符为 /\r?\n/（兼容 CRLF，常量定义于 src/core.ts#L75）。这种"回放已缓冲块 + 订阅未来块"的模式，正是下文 pipe() 任意时刻接管能力的同一底层机制。

pipe()：比 bash 管道更强的流管道

pipe() 把进程输出重定向到目标，几乎等同于 bash 的 |，但有三类增强：目标是 zx 进程时可组合、支持在管道中途接管、支持字符串字面量目标。

const greeting = await $`printf "hello"`
  .pipe($`awk '{printf $1", world!"}'`)
  .pipe($`tr '[a-z]' '[A-Z]'`)

pipe() 接受任意 Writable、ProcessPromise 或文件路径三种目标：

await $`echo "Hello, stdout!"`
  .pipe(fs.createWriteStream('/tmp/output.txt'))

传字符串则隐式创建写文件流，等价于：

await $`echo "Hello, stdout!"`
  .pipe('/tmp/output.txt')

链式管道是 thenable

const p = $`echo "hello"`
  .pipe(getUpperCaseTransform())
  .pipe(fs.createWriteStream(tempfile()))  // <- stream
const o = await p

源码中 promisifyStream()（src/core.ts#L747）用 proxyOverride 给目标流挂上一个 then()，在流的 finish/EPF（end-piped-from）事件触发时以 from.output（即 ProcessOutput）resolve——这就是"管道链可以 await"的实现。同时它挂上了 run() 与 pipe() 透传，使得 ProcessPromise 也兼容标准 Stream.pipe API：

const { stdout } = await fs
  .createReadStream(await fs.writeFile(file, 'test'))
  .pipe(getUpperCaseTransform())
  .pipe($`cat`)

实时输出

await $`echo 1; sleep 1; echo 2; sleep 1; echo 3;`
  .pipe(process.stdout)

管道到 process.stdout 后，每个块一到就被转发，实现逐行实时打印。

时间机器：任意时刻接管管道

这是 pipe() 最独特的能力：在进程开始、中途甚至结束之后调用 pipe()，所有已产生的块都会被缓冲并按正确顺序处理。

const result = $`echo 1; sleep 1; echo 2; sleep 1; echo 3`
const piped1 = result.pipe`cat`
let piped2

setTimeout(() => { piped2 = result.pipe`cat` }, 1500)

(await piped1).toString()  // '1\n2\n3\n'
(await piped2).toString()  // '1\n2\n3\n'

从 src/core.ts#L640 的 _pipe() 实现可以看清机制：

• 若源进程已 settle（this.output 存在），走 fillSettled() 分支：把 _zurk.store[source] 中已缓冲的所有 chunk 一次性写入中转流 from（一个 VoidStream），然后 end()；
• 若源进程仍在运行，则用 ee.once(source, ...) 先回放缓冲、再 ee.on(source, onData) 订阅增量，ee 是快照中的 EventEmitter（src/core.ts#L167）。

由于 _pipe 内部还设置了 this._piped = true（src/core.ts#L657），run() 中的 stdout 日志回调会跳过打印（src/core.ts#L367 注释："If the process is piped, don't print its output"），避免管道内容在终端重复出现；而 stderr 无论是否管道都会打印。

分流：一个进程喂多个消费者

const p = $`some-command`
const [o1, o2] = await Process.all([
  p.pipe`log`,
  p.pipe`extract`
])

多次调用 pipe() 即产生扇出，每个消费者都拿到完整的一份输出——因为每次 _pipe 都基于同一份 store 缓冲独立回放。

管道中的 nothrow() 组合

await $`find ./examples -type f -print0`
  .pipe($`xargs -0 grep ${'missing' + 'part'}`.nothrow())
  .pipe($`wc -l`)

grep 无匹配时退出码非零，nothrow() 让该环节不抛错、管道继续走到 wc -l。

字符串字面量目标

await $`printf "hello"`
  .pipe`awk '{printf $1", world!"}'`
  .pipe`tr '[a-z]' '[A-Z]'`

_pipe() 里通过 isStringLiteral(dest, ...args) 判断模板字面量，自动构造一个 $({ halt: true, signal: this.signal })(...) 的下游进程（src/core.ts#L644）——halt: true 表示下游进程会等上游有数据/被触发时才运行，且 signal 自动从上游继承。

合并：多路汇入一个进程

pipe() 不仅可链式和分流，也能合并。经典模式是多个 halt 进程把 stdout 汇入一个 cat 消费者：

const $h = $({ halt: true })
const p1 = $`echo foo`
const p2 = $h`echo a && sleep 0.1 && echo c && sleep 0.2 && echo e`
const p3 = $h`sleep 0.05 && echo b && sleep 0.1 && echo d`
const p4 = $`sleep 0.4 && echo bar`
const p5 = $h`cat`

await p1
p1.pipe(p5)
p2.pipe(p5)
p3.pipe(p5)
p4.pipe(p5)

const { stdout } = await p5.run() // 'foo\na\nb\nc\nd\ne\nbar\n'

注意 p5 是 halt 的，需要显式 p5.run() 启动消费者（或依赖 runBack 机制：下游 run() 会唤醒所有上游，见 src/core.ts#L732）。

指定 stderr 管道

pipe() 默认操作 stdout 流，也可以用属性形式指定 stderr（getter pipe 上挂有 stdout/stderr/stdall 三个方法，见 src/core.ts#L624）：

const p = $`echo foo >&2; echo bar`
const o1 = (await p.pipe.stderr`cat`).toString()  // 'foo\n'
const o2 = (await p.pipe.stdout`cat`).toString()  // 'bar\n'

signal 沿管道传播

指定了 signal 选项时，它会沿管道向下传递，abort 时整条管道一起终止：

const ac = new AbortController()
const { signal } = ac
const p = $({ signal, nothrow: true })`echo test`.pipe`sleep 999`
setTimeout(() => ac.abort(), 50)

try {
  await p
} catch ({ message }) {
  message // The operation was aborted
}

这依赖 _pipe 中 pipe`` 字面量目标自动携带 signal: this.signal`（src/core.ts#L649）的传递链。

自由组合 zx 进程与原生流

const getUpperCaseTransform = () => new Transform({
  transform(chunk, encoding, callback) {
    callback(null, String(chunk).toUpperCase())
  },
})

// $ > stream (promisified) > $
const o1 = await $`echo "hello"`
  .pipe(getUpperCaseTransform())
  .pipe($`cat`)

o1.stdout //  'HELLO\n'

// stream > $
const file = tempfile()
await fs.writeFile(file, 'test')
const o2 = await fs
  .createReadStream(file)
  .pipe(getUpperCaseTransform())
  .pipe($`cat`)

o2.stdout //  'TEST'

$ 进程、原生 Transform、文件流可以任意顺序拼接，管道链始终 thenable。

unpipe()：从管道中移除一个源

unpipe() 是 pipe() 的反操作，把某个进程从管道中摘除，摘除前已发出的数据仍然保留在下游：

const p1 = $`echo foo && sleep 0.05 && echo bar && sleep 0.05 && echo baz && sleep 0.05 && echo qux`
const p2 = $`echo 1 && sleep 0.05 && echo 2 && sleep 0.05 && echo 3`
const p3 = $`cat`

p1.pipe(p3)
p2.pipe(p3)

setTimeout(() => p1.unpipe(p3), 105)

assert.equal((await p1).stdout, 'foo\nbar\nbaz\nqux')
assert.equal((await p2).stdout, '1\n2\n3')
assert.equal((await p3).stdout, 'foo\n1\nbar\n2\n3')

实现上 unpipe() 委托给 ProcessPromise.bus.unpipe(this, to)（src/core.ts#L634），bus 内部维护 Map<ProcessPromise, Set<PipeAcceptor>> 引用关系（src/core.ts#L711），删除对应引用后 _pipe 里 check() 检测到引用消失，就不再向该目标转发后续 chunk。这个例子直观展示了动态摘除的效果：p3 收到了 unpipe 时刻之前 p1 已发出的 foo 和 bar，但之后 p1 的输出不再进入 p3。

kill()：杀死进程及其全部子进程

const p = $`sleep 999`
setTimeout(() => p.kill('SIGINT'), 100)
await p

默认发送 SIGTERM，可通过参数指定其他信号。对已结束的进程调用 kill() 会抛错：

const p = await $`sleep 999`
p.kill() // Error: Too late to kill the process.

kill() 的源码（src/core.ts#L446）先检查 isSettled()（抛 Fail('Too late to kill the process.')）、child 与 pid 是否存在，然后调用全局的 kill(pid, signal)。真正的实现在 src/core.ts#L1057：信号默认取 $.killSignal || SIGTERM；Windows 上走 taskkill /pid <pid> /t /f；POSIX 上先用 ps.tree({ pid, recursive: true }) 递归收集整棵进程树逐个 process.kill，再尝试对进程组 -pid 和单独 +pid 兜底——这就是"kill the process and all children"的完整语义，避免 sh -c 派生出的孙进程逃逸。

abort()：通过 AbortController 终止

const ac = new AbortController()
const {signal} = ac
const p = $({signal})`sleep 999`

setTimeout(() => ac.abort('reason'), 100)
await p

如果不显式提供 ac 或 signal，zx 会自动创建一个（快照中的 ac: opts.ac || new AbortController()，src/core.ts#L166），并通过 p 上的 signal getter 暴露出来（src/core.ts#L548），可用于控制外部资源：

const p = $`sleep 999`
const {signal} = p

const res = fetch('https://example.com', {signal})
p.abort('reason')

abort() 的三个守卫条件（src/core.ts#L436）：已 settle 抛 Too late to abort the process.；signal 被别的进程控制时抛错；没有 child 进程时抛错。因此"可运行中 abort、结束后 abort 报错"的行为：

const p = $({nothrow: true})`sleep 999`
p.abort() // ok

await p
p.abort() // Error: Too late to abort the process.

kill() 与 abort() 的区别在于语义和粒度：kill() 直接按 pid 杀进程树（默认 SIGTERM，可换信号）；abort() 走标准 AbortController/AbortSignal 事件体系，便于与 fetch、setTimeout 等原生支持 signal 的 API 复用同一控制句柄。

stdio()：配置标准输入输出

const h$ = $({halt: true})
const p1 = h$`read`.stdio('inherit', 'pipe', null).run()
const p2 = h$`read`.stdio('pipe').run() // 等价于 ['pipe', 'pipe', 'pipe']

stdio() 接受单值（展开为 [stdin, 'pipe', 'pipe']）或完整数组（src/core.ts#L456）：

stdio(
  stdin: IOType | StdioOptions,
  stdout: IOType = 'pipe',
  stderr: IOType = 'pipe'
): this

注意 stdio 必须在进程启动前设置，所以用 halt 预设语法更合适：

await $({stdio: ['pipe', 'pipe', 'pipe']})`read`

从 run() 的实现看，$.stdio 会原样传给底层 exec（src/core.ts#L347），启动后再改 snapshot 已无法影响已派生的进程——这就是文档强调"preset 语法更优先"的原因。null 表示关闭该通道，'inherit' 表示继承父进程对应通道。

nothrow()：不抛错地拿到结果

改变 $ 的行为，使非零退出码不再抛异常，等价于 $({nothrow: true}) 选项：

await $`grep something from-file`.nothrow()

// 管道内部使用：
await $`find ./examples -type f -print0`
  .pipe($`xargs -0 grep something`.nothrow())
  .pipe($`wc -l`)

// 也可以传入参数为单条命令切换 nothrow 模式
$.nothrow = true
await $`echo foo`.nothrow(false)

实现是一行（src/core.ts#L467）：this._snapshot.nothrow = v。由于 finalize 时判断条件是 output.ok || this.isNothrow()（src/core.ts#L422），开启后非零退出码也走 fulfilled 分支，await 得到带真实 exitCode 的 ProcessOutput 而非 reject。nothrow(false) 参数形式则用于在全局 $.nothrow = true 的前提下，对单条命令恢复默认行为。

如果只需要退出码，直接用 exitCode 更简洁：

if (await $`[[ -d path ]]`.exitCode == 0) {
  // ...
}

// 等价于：
if ((await $`[[ -d path ]]`.nothrow()).exitCode == 0) {
  // ...
}

区别在于：exitCode 是只读提取器（成功失败都能取）；nothrow() 改变的是整条 promise 的异常语义，管道中的中间环节依赖后者。

quiet() 与 verbose()：日志级别控制

quiet() 开启静默模式，命令输出不在终端回显：

// 命令输出不会被显示
await $`grep something from-file`.quiet()

$.quiet = true
await $`echo foo`.quiet(false) // 对单条命令关闭

verbose() 开启详细输出，传 false 关闭：

await $`grep something from-file`.verbose()

$.verbose = true
await $`echo foo`.verbose(false) // 单条命令关闭 verbose

两者同样只是改 _snapshot.quiet / _snapshot.verbose（src/core.ts#L472），真正生效点在 run() 的日志回调里：stdout 事件仅在 !self._piped && self.isVerbose() 时打印（src/core.ts#L366），而 isVerbose() 的定义是 this._snapshot.verbose && !this.isQuiet()（src/core.ts#L602）——即 quiet 优先级高于 verbose。注意 stderr 始终打印（除非 quiet），这一点与上面 stdout 的静默逻辑形成对比。

timeout()：超时自动杀掉进程

await $`sleep 999`.timeout('5s')

// 也可以指定超时后使用的信号
await $`sleep 999`.timeout('5s', 'SIGKILL')

若进程已 settle，该方法不做任何事；传入 nullish 值会取消超时。源码（src/core.ts#L482）：

timeout(d: Duration = 0, signal = $.timeoutSignal): this {
  if (this.isSettled()) return this
  const $ = this._snapshot
  $.timeout = parseDuration(d)
  $.timeoutSignal = signal

  if (this._timeoutId) clearTimeout(this._timeoutId)
  if ($.timeout && this.isRunning()) {
    this._timeoutId = setTimeout(() => this.kill($.timeoutSignal), $.timeout)
    this.finally(() => clearTimeout(this._timeoutId)).catch(noop)
  }
  return this
}

三个要点都对应上文档描述：settle 后直接 return；parseDuration 支持 '5s' 这类时长字符串，传 0/nullish 时不注册定时器（即禁用）；超时动作就是 this.kill($.timeoutSignal)，默认信号为 SIGTERM（defaults.timeoutSignal，src/core.ts#L152），且 finally 里会清掉定时器，避免已自然结束的进程触发僵尸 kill。Duration 类型也从 src/util.ts 导入，支持秒级字符串解析。

小结：ProcessPromise 的能力版图

把整篇文档串起来，ProcessPromise 在 src/core.ts 中呈现为一个"进程 + Promise + 流代理"三合一的类型：

维度	能力	源码位置
生命周期	stage 五态状态机、halt/run() 延迟启动	src/core.ts#L234、src/core.ts#L309
流	stdin/stdout/stderr、[Symbol.asyncIterator] 逐行迭代	src/core.ts#L529、src/core.ts#L796
输出	text()/json()/lines()/buffer()/blob()、exitCode	src/core.ts#L541
管道	pipe() 任意时刻接管、分流/合并/字面量目标、unpipe()	src/core.ts#L640
终止	kill() 杀进程树、abort() 走 AbortController	src/core.ts#L436、src/core.ts#L1057
配置	stdio()、nothrow()、quiet()、verbose()、timeout()	src/core.ts#L456

理解它的核心在于两点：一是快照隔离——每条 $ 命令持有独立的 Snapshot，实例方法只改自己；二是块缓冲 + 事件回放——store 里始终保存已产出的 stdout/stderr 块，pipe() 与异步迭代器据此实现对"过去"数据的确定性重放，这正是 zx 管道能"事后接管、多路分叉"的根本原因。配套的输出对象语义可继续参考 docs/process-output.md，全局选项（nothrow、timeout 等）的完整说明见 docs/api.md。