Merge master into fix/workspace-recency-order

packages/experimental/code-runtime-python/tests/boot-write-failure.spec.ts

@@ -2,16 +2,13 @@ import { EventEmitter } from 'node:events'
 import { existsSync } from 'node:fs'
 import { dirname } from 'node:path'
 import { PassThrough } from 'node:stream'
-import { afterEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it, onTestFinished, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 
 /**
- * A synchronous `proto.write` throw on the fd-3 pipe is the one boot path a real
- * subprocess cannot be coerced into from a test: the pipe accepts queued bytes
- * until the kernel buffer fills, and a same-tick EPIPE needs fd 3 already closed
- * before the first write. `spawn` is mocked so fd 3 throws on the boot frame,
- * which is exactly the branch that regressed. The mock is confined to this file
- * so the real-subprocess suite in runtime.spec.ts is untouched.
+ * Mocked subprocess pipes control synchronous write failures and backpressure
+ * transitions independently of kernel buffering. The real-subprocess suite
+ * remains in runtime.spec.ts.
  */
 const { execFileSyncMock, spawnMock } = vi.hoisted(() => ({ execFileSyncMock: vi.fn(), spawnMock: vi.fn() }))
 vi.mock('node:child_process', async (importOriginal) => {
@@ -128,7 +125,67 @@ function fakeChildBackpressuredThenDestroyed(): { child: EventEmitter; proto: Pa
   return { child, proto }
 }
 
-describe('PythonCodeRuntime — boot-write failure', () => {
+describe('PythonCodeRuntime — controlled subprocess pipes', () => {
+  it('preserves pending replies across compaction while the pipe stays backpressured', async () => {
+    const spawned = Promise.withResolvers<undefined>()
+    let written = Promise.withResolvers<undefined>()
+    const replies: unknown[] = []
+    const proto = new PassThrough()
+    const stdout = new PassThrough()
+    const stderr = new PassThrough()
+    const stdin = new PassThrough()
+    const child = Object.assign(new EventEmitter(), { stdout, stderr, stdio: [stdin, stdout, stderr, proto] })
+    proto.write = (chunk: unknown) => {
+      const frame = JSON.parse(String(chunk)) as { type: string }
+      if (frame.type !== 'reply') return true
+      replies.push(frame)
+      written.resolve(undefined)
+      return false
+    }
+    spawnMock.mockImplementation(() => { spawned.resolve(undefined); return child })
+    const ctx = new Context()
+    const fiber = await ctx.plugin(PythonCodeRuntime)
+    const runtime = ctx.codeRuntime as InstanceType<typeof PythonCodeRuntime>
+    const run = runtime.run({
+      program: 'return 1',
+      bindings: [{ global: 'tools', functions: { echo: async (value: unknown) => value as number } }],
+    })
+    onTestFinished(async () => {
+      await fiber.dispose()
+      await run
+      for (const stream of [stdin, stdout, stderr, proto]) stream.destroy()
+    })
+    const calls = (start: number): void => {
+      proto.emit('data', Buffer.from(Array.from({ length: 512 }, (_, offset) => JSON.stringify({
+        type: 'call', id: start + offset, global: 'tools', name: 'echo', args: start + offset,
+      })).join('\n') + '\n'))
+    }
+    const drainThrough = async (count: number): Promise<void> => {
+      while (replies.length < count) {
+        written = Promise.withResolvers<undefined>()
+        expect(proto.listenerCount('drain')).toBe(1)
+        proto.emit('drain')
+        await written.promise
+      }
+    }
+    await spawned.promise
+    proto.emit('data', Buffer.from('{"type":"boot-ack"}\n'))
+    calls(0)
+    await written.promise
+    await drainThrough(256)
+    calls(512)
+    await drainThrough(768)
+    calls(1024)
+    // Each write remains blocked until this fixture emits drain, keeping
+    // pending replies behind the consumed-prefix compaction at frame 1024.
+    await drainThrough(1536)
+    expect(replies).toEqual(Array.from({ length: 1536 }, (_, id) => ({ type: 'reply', id, ok: true, value: id })))
+    proto.emit('drain')
+    proto.emit('data', Buffer.from('{"type":"done","value":"done"}\n'))
+    expect(await run).toMatchObject({ value: 'done' })
+    expect(proto.listenerCount('drain')).toBe(0)
+  })
+
   it('force-kills a version probe that exceeds its load-time deadline', async () => {
     const ctx = new Context()
     const fiber = await ctx.plugin(PythonCodeRuntime)

packages/experimental/code-runtime-python/tests/runtime.spec.ts

@@ -5411,54 +5411,6 @@ describe('PythonCodeRuntime — hostile peer', () => {
     expect(result.value).toContain('duplicate dict key')
   }, 30_000)
 
-  it('compacts the reply queue mid-drain without dropping pending frames', async () => {
-    // A reply larger than the writable high-water mark makes the FIRST write
-    // return false, suspending the drain loop; the frames queued behind it
-    // push the drain's consumed head past MAX_PENDING_REPLIES, so the resumed
-    // drain compacts the queue mid-run. The child reads fd 3 itself (blocking
-    // the asyncio pump, so its reads cannot race the host's pushes) and sends
-    // a second wave of calls AFTER reading part of the first wave's replies —
-    // those replies are still pending when the drain's head crosses the
-    // compaction bound, so a compaction that dropped pending frames would
-    // leave the child's reply count short and the read loop spinning to the
-    // wall clock. No fixed sleep: the child's reads pace at the drain's
-    // delivery rate (each write blocks until the child reads), and the host
-    // finishes pushing all of a wave within milliseconds — orders of magnitude
-    // before the head crosses the bound — so the queue is always full at the
-    // splice. Newlines are counted per chunk (each reply carries exactly one),
-    // never by re-scanning the accumulated total, which would be O(n²).
-    const { runtime } = await setup({ maxWallMs: 60_000 })
-    const result = await runtime.run({
-      program: [
-        'import os',
-        'frame = b\'{"type":"call","id":%d,"global":"tools","name":"big","args":{}}\\n\'',
-        'for i in range(1024):',
-        '    view = memoryview(frame % i)',
-        '    while view:',
-        '        view = view[os.write(3, view):]',
-        'seen = 0',
-        'while seen < 500:',
-        '    chunk = os.read(3, 65536)',
-        '    if not chunk:',
-        '        break',
-        '    seen += chunk.count(b"\\n")',
-        'for i in range(500):',
-        '    view = memoryview(frame % (1024 + i))',
-        '    while view:',
-        '        view = view[os.write(3, view):]',
-        'while seen < 1524:',
-        '    chunk = os.read(3, 65536)',
-        '    if not chunk:',
-        '        break',
-        '    seen += chunk.count(b"\\n")',
-        'return "done"',
-      ].join('\n'),
-      bindings: [{ global: 'tools', functions: { big: async () => 'x'.repeat(65 * 1024) } }],
-    })
-    expect(result.error).toBeUndefined()
-    expect(result.value).toBe('done')
-  }, 60_000)
-
   it('bounds a flood of zero-byte log lines through the per-entry separator charge', async () => {
     // Blank print() lines carry zero content bytes; without the +1 separator
     // charge they would bypass maxLogBytes entirely and grow the retained

packages/subprocess/subprocess-local/README.i18n.yaml

@@ -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: 3222d7bf9b9b498baf939212ebbf7da9bf599d76
-README.zh.md: e8d2ddc9398ffe186260a1b4ed3a0cd116ff8afb
+README.md: 10f0b01d147dfd0afefd2214264084e55d2b213d
+README.zh.md: 14479e688f122712041741270b995ca47a2e1166

packages/subprocess/subprocess-local/README.md

@@ -52,7 +52,7 @@ Collect mode keeps the last `maxBytes` of a stream in memory — errors and fina
 
 Normal disposal terminates every running managed range and terminal session and awaits quiescence. During a JavaScript-observable host exit — direct `process.exit()`, default uncaught exceptions, default unhandled rejections — synchronous finalization asks a Linux scope to kill its members, kills each Windows runner so its sole Job handle closes, and uses the existing PGID, `taskkill`, or captured-identity operation for fallbacks. It creates no promises or timers and does not claim quiescence. The same exit removes the private per-process spill directory when it holds no completed spill file; completed spill files remain as full-output recovery artifacts until an external cleanup. Unhandled `SIGTERM`/`SIGINT`/`SIGHUP`, `SIGKILL`, fatal OOM, native crashes, and power loss need an external supervisor.
 
-Linux ordinary and terminal cancellation preserves the observed termination signal even before the bootstrap consumes its launch request. An unconsumed request still reports startup failure when no matching termination was requested; a recorded pre-exec error always takes precedence. `waitForExit()` independently proves the scope empty, including a scope the manager leaves active with no processes after a payload dies before it enters that scope's cgroup.
+Linux ordinary and terminal cancellation preserves the observed termination signal even before the bootstrap consumes its launch request. An unconsumed request still reports startup failure when no matching termination was requested; a recorded pre-exec error always takes precedence. `waitForExit()` independently proves the scope empty, including a scope the manager leaves active with no processes after a payload dies before it enters that scope's cgroup. State queries interrupted by a termination signal are repeated before deciding whether cleanup succeeded. A failed final signal does not reject a range subsequently proven empty; active ranges without that proof retain the signal failure.
 
 ### What can go wrong
 

packages/subprocess/subprocess-local/README.zh.md

@@ -52,7 +52,7 @@ kind: "package-reference"
 
 正常 dispose 会终止每个仍在运行的受管范围与终端会话并等待其完全停稳。在 JavaScript 可观察的宿主退出期间——直接 `process.exit()`、默认未捕获异常、默认未处理 rejection——同步最终清理会请求 Linux scope 终止其成员，同步终止每个 Windows runner 以关闭其唯一 Job handle，并为 fallback 使用既有 PGID、`taskkill` 或已捕获身份操作。它不创建 Promise 或定时器，也不声称已经完全停稳。同一退出阶段会删除未持有任何已完成 spill 文件的每进程私有 spill 目录；已完成的 spill 文件作为完整输出恢复产物保留，直到外部机制清理。未处理的 `SIGTERM`/`SIGINT`/`SIGHUP`、`SIGKILL`、fatal OOM、native crash 与断电需要外部 supervisor。
 
-Linux 普通进程和终端进程即使在 bootstrap 消费启动请求前被取消，也会保留实际观察到的终止信号。如果没有请求对应的终止信号，未消费的请求仍会报启动失败；已记录的 pre-exec 错误始终优先。`waitForExit()` 独立证明 scope 已为空，其中也包括 payload 在进入该 scope 的 cgroup 前就被杀死、manager 因此让它保持 active 却没有任何进程的 scope。
+Linux 普通进程和终端进程即使在 bootstrap 消费启动请求前被取消，也会保留实际观察到的终止信号。如果没有请求对应的终止信号，未消费的请求仍会报启动失败；已记录的 pre-exec 错误始终优先。`waitForExit()` 独立证明 scope 已为空，其中也包括 payload 在进入该 scope 的 cgroup 前就被杀死、manager 因此让它保持 active 却没有任何进程的 scope。状态查询期间若发出终止信号，会重新查询后再判定清理是否成功。即使最终信号发送失败，之后证明范围已为空仍可成功结束；未获得这一证明的 active 范围仍报告信号失败。
 
 ### 可能出错的地方
 

packages/subprocess/subprocess-local/src/linux-scope.ts

@@ -318,6 +318,7 @@ class SystemdScopeOwner implements BoundProcessOwner {
 
   private async rangeActive(): Promise<boolean> {
     this.observeRequestConsumption()
+    const generation = this.wakeGeneration
     const result = await this.query(this.systemctl, [
       '--user',
       'show',
@@ -326,6 +327,8 @@ class SystemdScopeOwner implements BoundProcessOwner {
       '--property=ActiveState',
       '--property=TasksCurrent',
     ])
+    // A signal invalidates state queried before its delivery and direct fallback.
+    if (generation !== this.wakeGeneration) return true
     const output = `${result.stdout}\n${result.stderr}`
     if (result.status === 0) {
       const { loadState, activeState, tasksCurrent } = this.parseUnitState(result.stdout)
@@ -340,11 +343,11 @@ class SystemdScopeOwner implements BoundProcessOwner {
       if (!['active', 'activating', 'reloading', 'deactivating'].includes(activeState)) {
         throw new Error(`systemctl returned unknown ActiveState for ${this.unit}: ${JSON.stringify(activeState)}`)
       }
-      if (this.killFailure !== undefined) throw this.killFailure
       if (this.emptyRange(tasksCurrent)) {
         this.releaseEmptyRange()
         return false
       }
+      if (this.killFailure !== undefined) throw this.killFailure
       return true
     }
     if (!MISSING_UNIT.test(output)) {

packages/subprocess/subprocess-local/tests/linux-scope.spec.ts

@@ -430,6 +430,82 @@ describe('Linux scope establishment and quiescence', () => {
     killFailed.result.owner.cleanup?.()
   })
 
+  it('rechecks a pre-signal observation before reporting a failed final kill', async () => {
+    denyProcessGroups()
+    const beforeKill = Promise.withResolvers<ReturnType<typeof activeUnit>>()
+    const query = vi.fn()
+      .mockImplementationOnce(() => beforeKill.promise)
+      .mockResolvedValueOnce(activeUnit('inactive'))
+    const sleep = vi.fn(async () => {})
+    const launched = launch(query, {
+      sleep,
+      spawnSync: vi.fn(() => ({ status: 1, stdout: '', stderr: 'Invalid argument' })) as never,
+    })
+    consumeLinuxLaunchRequest(launched.requestPath)
+    const waiting = launched.result.owner.waitForExit()
+    launched.result.owner.signal('SIGKILL')
+    launched.child.exit(null, 'SIGKILL')
+    beforeKill.resolve(activeUnit())
+    await expect(waiting).resolves.toBeUndefined()
+    await expect(launched.result.direct).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
+    expect(query).toHaveBeenCalledTimes(2)
+    expect(sleep).not.toHaveBeenCalled()
+    launched.result.owner.cleanup?.()
+  })
+
+  it('does not accept a pre-signal empty observation when the fresh range remains populated', async () => {
+    denyProcessGroups()
+    const beforeKill = Promise.withResolvers<ReturnType<typeof activeUnit>>()
+    const query = vi.fn()
+      .mockImplementationOnce(() => beforeKill.promise)
+      .mockResolvedValueOnce(activeUnitWithTasks('1'))
+    const launched = launch(query, {
+      spawnSync: vi.fn(() => ({ status: 1, stdout: '', stderr: 'Invalid argument' })) as never,
+    })
+    consumeLinuxLaunchRequest(launched.requestPath)
+    const waiting = launched.result.owner.waitForExit()
+    launched.result.owner.signal('SIGKILL')
+    launched.child.exit(null, 'SIGKILL')
+    beforeKill.resolve(activeUnit('inactive'))
+    await expect(waiting).rejects.toThrow('Invalid argument')
+    await launched.result.direct
+    expect(query).toHaveBeenCalledTimes(2)
+    launched.result.owner.cleanup?.()
+  })
+
+  it('accepts a confirmed empty range after a failed final kill', async () => {
+    denyProcessGroups()
+    const spawnSync = recordingSystemctl()
+      .mockReturnValueOnce({ status: 1, stdout: '', stderr: 'Invalid argument' })
+    const launched = launch(async () => activeUnitWithTasks('0'), { spawnSync: spawnSync as never })
+    consumeLinuxLaunchRequest(launched.requestPath)
+    launched.result.owner.signal('SIGKILL')
+    launched.child.exit(null, 'SIGKILL')
+    await expect(launched.result.owner.waitForExit()).resolves.toBeUndefined()
+    await expect(launched.result.direct).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
+    expect(spawnSync.mock.calls.map(call => call[1]?.[1])).toEqual(['kill', 'stop'])
+    launched.result.owner.cleanup?.()
+  })
+
+  it.each([
+    { tasks: '1', clientRunning: false },
+    { tasks: '[not set]', clientRunning: false },
+    { tasks: '0', clientRunning: true },
+  ])('retains a failed kill with tasks=$tasks and clientRunning=$clientRunning', async ({ tasks, clientRunning }) => {
+    denyProcessGroups()
+    const spawnSync = recordingSystemctl()
+      .mockReturnValueOnce({ status: 1, stdout: '', stderr: 'Invalid argument' })
+    const launched = launch(async () => activeUnitWithTasks(tasks), { spawnSync: spawnSync as never })
+    consumeLinuxLaunchRequest(launched.requestPath)
+    launched.result.owner.signal('SIGKILL')
+    if (!clientRunning) launched.child.exit(null, 'SIGKILL')
+    await expect(launched.result.owner.waitForExit()).rejects.toThrow('Invalid argument')
+    expect(spawnSync).toHaveBeenCalledOnce()
+    if (clientRunning) launched.child.exit(null, 'SIGKILL')
+    await launched.result.direct
+    launched.result.owner.cleanup?.()
+  })
+
   it('reports command-query failures from the default systemctl adapter', async () => {
     childProcessMocks.execFile.mockImplementationOnce((...args: unknown[]) => {
       const callback = args.at(-1) as (error: Error | null, stdout: string, stderr: string) => void

scripts/run-gates.spec.ts

@@ -2,6 +2,7 @@ import { readFileSync } from 'node:fs'
 import { describe, expect, it, vi, type MockInstance } from 'vitest'
 import {
   cliGateOptions,
+  collectDescendants,
   defaultConcurrency,
   formatGateResultReason,
   gatesForMode,
@@ -917,6 +918,30 @@ describe('fail-fast scheduling', () => {
 })
 
 describe('process-table parsing', () => {
+  it('excludes the root when a parent link returns to it', () => {
+    expect(collectDescendants(100, [[200, 100], [100, 200], [300, 200]]))
+      .toEqual([200, 300])
+  })
+
+  it('visits duplicate and cyclic descendant links only once', () => {
+    expect(collectDescendants(100, [
+      [200, 100], [200, 100], [300, 100], [200, 200], [400, 200], [200, 400], [500, 300], [900, 800],
+    ])).toEqual([200, 300, 400, 500])
+  })
+
+  it('returns no descendants for an isolated or self-parented root', () => {
+    expect(collectDescendants(100, [])).toEqual([])
+    expect(collectDescendants(100, [[100, 100]])).toEqual([])
+  })
+
+  it('walks a wide child set without spreading it into call arguments', () => {
+    const children = Array.from({ length: 150_000 }, (_, i): [number, number] => [i + 3, 2])
+    const descendants = collectDescendants(1, [[2, 1], ...children])
+    expect(descendants).toHaveLength(children.length + 1)
+    expect(descendants[0]).toBe(2)
+    expect(descendants.at(-1)).toBe(150_002)
+  })
+
   it('parses `pid ppid` rows from a POSIX ps dump', () => {
     expect(parsePidPpidLines('  123   1\n456 123\n  789 456\n')).toEqual([[123, 1], [456, 123], [789, 456]])
   })

scripts/run-gates.ts

@@ -1505,23 +1505,29 @@ export function taskkillArgs(rootPid: number, descendants: number[]): string[][]
   return [rootPid, ...descendants].map(pid => ['/PID', String(pid), '/T', '/F'])
 }
 
-/** Breadth-first walk of the pid/ppid rows starting at `root`. */
-function collectDescendants(root: number, rows: Array<[number, number]>): number[] {
+/**
+ * Walk a process-table snapshot without revisiting duplicate or cyclic PID links.
+ * @param root - process whose descendants are collected; excluded from the result.
+ * @param rows - observed PID and parent PID pairs.
+ * @returns distinct reachable descendants in breadth-first order.
+ */
+export function collectDescendants(root: number, rows: Array<[number, number]>): number[] {
   const byParent = new Map<number, number[]>()
   for (const [pid, ppid] of rows) {
     const children = byParent.get(ppid) ?? []
     children.push(pid)
     byParent.set(ppid, children)
   }
-  const result: number[] = []
-  const queue = byParent.get(root) ?? []
-  for (let index = 0; index < queue.length; index += 1) {
-    const pid = queue[index]
-    if (pid === undefined) continue
-    result.push(pid)
-    queue.push(...(byParent.get(pid) ?? []))
+  const seen = new Set([root])
+  const queue = [root]
+  for (const parent of queue) {
+    for (const pid of byParent.get(parent) ?? []) {
+      if (seen.has(pid)) continue
+      seen.add(pid)
+      queue.push(pid)
+    }
   }
-  return result
+  return queue.slice(1)
 }
 
 /**