fix(subprocess): clean runtime state after teardown failure

2026-08-08 21:27:57 +08:00
parent 0fec058a48
commit be3533fa19
5 changed files with 60 additions and 15 deletions
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/subprocess/subprocess-local/README.md
-README.md: 38b26c33f3854acdb4a2bb48612da73361d727ba
-README.zh.md: fc4a5d4961fb7618b9db84b21063f907361a26d8
+README.md: 61ee8ff9e09d5177ff6d5b3805dd84fb74c59961
+README.zh.md: 06a8e4bdd225e212aa67817ebbf66581314ab2bf
@@ -10,7 +10,7 @@ Local implementation of the [`@deepseek-ai/dsh-subprocess`](../subprocess/README
 - **Per-stream dispositions** — `'pipe'` hands the raw stream to the caller untouched (protocol framing stays consumer-owned); `'inherit'` passes the parent descriptor through; collect mode keeps the in-memory TAIL beyond its cap (errors and results cluster at the end — pi/OpenCode rationale) while the FULL stream is appended to a private temp file when a spill cap is configured — omitting `spill` keeps only the tail, the diagnostic shape. A stream larger than the spill cap discards its now-incomplete spill and returns only the marked truncated tail; spill fds are sealed at settlement, and a failed final close withholds the path rather than advertising an incomplete file. Spill files are `0600` with random names under a lazily-created `0700` per-process directory.
 - **Credential scrub + explicit merge** — `process.env` minus credential-shaped vars (`*KEY*`/`*SECRET*`/`*TOKEN*`) and all ambient `DSH_*` names; the spec's explicit `env` merges after that scrub with no namespace validation, so a deliberately supplied credential or current `DSH_*` fact wins while stale nested-harness identity cannot leak in ambiently. Supplied stdin is written and closed; otherwise fd 0 is `/dev/null`. See the [stdin/env Agent Note](../../../.agents/notes/implemented/architecture/2026-06-30-bash-stdin-env-trusted-plugin-surface.md) and [managed environment Agent Note](../../../.agents/notes/implemented/feature/2026-07-10-agent-session-identity-and-log-location.md).
 - **Offset-based reads** — collect-mode readers return deltas in whole-stream byte coordinates; the service never holds a cursor, so consumer-owned cursors (the bash background read path) and full-stream re-reads coexist, before and after settlement.
- **Execution-world coordinates** — `cwd` is the host process cwd, `runtimeRoot` is an owner-private temporary directory removed on disposal, and `resolveExecutable` checks absolute files or searches the scrubbed effective PATH with platform-aware executable extensions.
+- **Execution-world coordinates** — `cwd` is the host process cwd, `runtimeRoot` is an owner-private temporary directory removed on disposal before any process-cleanup failure is reported, and `resolveExecutable` checks absolute files or searches the scrubbed effective PATH with platform-aware executable extensions.
 - **Terminal-process ownership** — `spawnTerminal` allocates `node-pty`, bridges UTF-8 terminal bytes, inspects and signals the current foreground process group, and cleans descendants before the top-level shell. Each foreground inspection retains exact identities from the rooted tree; Linux also enumerates the POSIX session after its leader exits. A previously observed macOS descendant and any same-session Linux member therefore remain fenced after reparenting, while pid/start identity prevents cleanup from following PID reuse. The higher PTY backend owns prompt readiness, buffers, and model-facing operations.
 - **Terminate-and-join disposal** — the service retains live handles only so its own disposal can escalate every running tree and await its exit; settled and spawn-failed handles leave the live set on settlement.

@@ -10,7 +10,7 @@
 - **按流划分的处置方式**：`'pipe'` 把原始流原样交给调用方（协议分帧仍归消费方所有）；`'inherit'` 直通父进程的描述符；收集模式（collect）在输出超过上限后于内存中保留尾部（错误与结果通常聚集在末尾，沿用 pi/OpenCode 的理由），并在配置了 spill 上限时把完整流追加到一个私有临时文件；省略 `spill` 则只保留尾部，即诊断尾部的形状。某条流大于 spill 上限时，会丢弃已不完整的 spill，仅返回带截断标记的尾部；spill 文件描述符在结算时封存，最终关闭失败时则不公布路径，以免声称存在不完整的文件。spill 文件权限为 `0600`、名称随机，位于按需延迟创建的 `0700` 每进程目录之下。
 - **凭据清除 + 显式合并**：以 `process.env` 为基础，移除形似凭据的变量（`*KEY*`／`*SECRET*`／`*TOKEN*`）和所有环境中已有的 `DSH_*` 名称；spec 的显式 `env` 在该清除之后合并且不做命名空间校验，因此有意提供的凭据或当前 `DSH_*` 事实会胜出，而陈旧的嵌套 harness 身份无法从环境中隐式漏入。提供的 stdin 会被写入后关闭；否则 fd 0 指向 `/dev/null`。参见 [stdin/env Agent Note（agent 决策记录）](../../../.agents/notes/implemented/architecture/2026-06-30-bash-stdin-env-trusted-plugin-surface.md)与[受管环境 Agent Note](../../../.agents/notes/implemented/feature/2026-07-10-agent-session-identity-and-log-location.md)。
 - **基于偏移量的读取**：收集模式的读取器以全流字节坐标返回增量；服务自身从不持有游标，因此消费方自有的游标（bash 的后台读取路径）与完整流重读可以共存，结算前后皆然。
- **执行世界坐标**：`cwd` 是宿主进程 cwd，`runtimeRoot` 是所有者私有的临时目录，在资源释放时删除；`resolveExecutable` 检查绝对文件，或使用平台感知的可执行扩展名在清理后的有效 PATH 中查找。
+- **执行世界坐标**：`cwd` 是宿主进程 cwd，`runtimeRoot` 是所有者私有的临时目录，会在资源释放时删除，并且删除发生在报告任何进程清理失败之前；`resolveExecutable` 检查绝对文件，或使用平台感知的可执行扩展名在清理后的有效 PATH 中查找。
 - **终端进程所有权**：`spawnTerminal` 分配 `node-pty`，桥接 UTF-8 终端字节，检查当前前台进程组并向其发送信号，并先于顶层 shell 清理后代。每次前台检查都会保留有根进程树中的精确身份；Linux 还会在会话 leader 退出后枚举该 POSIX 会话。因此，先前观察到的 macOS 后代以及任何同会话 Linux 成员在重新设定父进程后仍受身份围栏保护，而 pid／启动身份可防止清理因 PID 复用而跟随到其他进程。上层 PTY 后端负责提示符就绪检测、缓冲和面向模型的操作。
 - **先终止再等待退出的 dispose**：服务保留存活句柄，只为让自身的 dispose 能对每个仍在运行的进程树执行升级并等待其退出；已结算与 spawn 失败的句柄在结算时即离开存活集合。

@@ -67,11 +67,15 @@ export class LocalSubprocessService extends SubprocessService {
        pending.push(terminal.waitForExit().then(() => { this.terminals.delete(terminal) }))
      }
      this.live.clear()
-      const outcomes = await Promise.allSettled(pending)
-      for (const outcome of outcomes) {
-        if (outcome.status === 'rejected') throw outcome.reason
-      }
-      await rm(this.runtimeRoot, { recursive: true, force: true })
+      const outcomes = [
+        ...await Promise.allSettled(pending),
+        ...await Promise.allSettled([rm(this.runtimeRoot, { recursive: true, force: true })]),
+      ]
+      const failures = outcomes.flatMap<unknown>(outcome => outcome.status === 'rejected'
+        ? [outcome.reason as unknown]
+        : [])
+      if (failures.length === 1) throw failures[0]
+      if (failures.length > 1) throw new AggregateError(failures, 'local subprocess teardown failed')
    }, 'local subprocess teardown')
  }

@@ -1,6 +1,6 @@
 import { PassThrough } from 'node:stream'
 import { describe, expect, it, vi } from 'vitest'
-import { rm, stat } from 'node:fs/promises'
+import { stat } from 'node:fs/promises'
 import { basename, dirname, relative, resolve } from 'node:path'
 import { Context } from 'cordis'
 import LocalSubprocessService from '@deepseek-ai/dsh-subprocess-local'
@@ -119,11 +119,15 @@ describe('LocalSubprocessService', () => {
    expect(terminals.size).toBe(0)
  })

-  it('waits for every terminal cleanup and retains rejections', async () => {
+  it('waits for every terminal cleanup, removes runtime state, and retains rejections', async () => {
    const ctx = new Context()
    const fiber = await ctx.plugin(LocalSubprocessService)
    const service = ctx.subprocess
    const runtimeRoot = service.runtimeRoot
+    const firstFailure = new Error('first retryable cleanup failure')
+    const secondFailure = new Error('second retryable cleanup failure')
+    const disposalErrors: unknown[] = []
+    ctx.logger.error = ((error: unknown) => { disposalErrors.push(error) }) as typeof ctx.logger.error
    const failedTerminal: SubprocessTerminalHandle = {
      pid: 1,
      output: new PassThrough(),
@@ -132,7 +136,12 @@ describe('LocalSubprocessService', () => {
      inspectForeground: async () => undefined,
      signalForeground: async () => 1,
      terminate: vi.fn(),
-      waitForExit: vi.fn(async () => { throw new Error('retryable cleanup failure') }),
+      waitForExit: vi.fn(async () => { throw firstFailure }),
+    }
+    const secondFailedTerminal: SubprocessTerminalHandle = {
+      ...failedTerminal,
+      terminate: vi.fn(),
+      waitForExit: vi.fn(async () => { throw secondFailure }),
    }
    let finishCleanup!: () => void
    const cleanup = new Promise<boolean>((resolve) => {
@@ -145,6 +154,7 @@ describe('LocalSubprocessService', () => {
    }
    const terminals = (service as unknown as { terminals: Set<SubprocessTerminalHandle> }).terminals
    terminals.add(failedTerminal)
+    terminals.add(secondFailedTerminal)
    terminals.add(drainingTerminal)

    let disposed = false
@@ -153,9 +163,40 @@ describe('LocalSubprocessService', () => {
    expect(disposed).toBe(false)
    finishCleanup()
    await disposing
-    expect(terminals).toEqual(new Set([failedTerminal]))
-    expect((await stat(runtimeRoot)).isDirectory()).toBe(true)
-    await rm(runtimeRoot, { recursive: true, force: true })
+    expect(terminals).toEqual(new Set([failedTerminal, secondFailedTerminal]))
+    await expect(stat(runtimeRoot)).rejects.toMatchObject({ code: 'ENOENT' })
+    expect(disposalErrors).toHaveLength(1)
+    expect(disposalErrors[0]).toMatchObject({
+      errors: [firstFailure, secondFailure],
+      message: 'local subprocess teardown failed',
+    })
+  })
+
+  it('reports one cleanup failure without wrapping it after removing runtime state', async () => {
+    const ctx = new Context()
+    const failure = new Error('single cleanup failure')
+    const disposalErrors: unknown[] = []
+    ctx.logger.error = ((error: unknown) => { disposalErrors.push(error) }) as typeof ctx.logger.error
+    const fiber = await ctx.plugin(LocalSubprocessService)
+    const service = ctx.subprocess
+    const runtimeRoot = service.runtimeRoot
+    const terminal: SubprocessTerminalHandle = {
+      pid: 1,
+      output: new PassThrough(),
+      done: Promise.resolve({ exitCode: 0, signal: null }),
+      write: async () => {},
+      inspectForeground: async () => undefined,
+      signalForeground: async () => 1,
+      terminate: vi.fn(),
+      waitForExit: vi.fn(async () => { throw failure }),
+    }
+    const terminals = (service as unknown as { terminals: Set<SubprocessTerminalHandle> }).terminals
+    terminals.add(terminal)
+
+    await fiber.dispose()
+
+    await expect(stat(runtimeRoot)).rejects.toMatchObject({ code: 'ENOENT' })
+    expect(disposalErrors).toEqual([failure])
  })

  it('releases a terminal after top-level exit reaches quiescence', async () => {