Fix Codex wire frames and bound grace timers

packages/subprocess/subprocess-local/package.json

@@ -29,11 +29,13 @@
   "peerDependencies": {
     "@deepseek-ai/dsh-invariants": "^0.0.1",
     "@deepseek-ai/dsh-subprocess": "^0.0.1",
+    "@deepseek-ai/dsh-timeout": "^0.0.1",
     "cordis": "^4.0.0-rc.7"
   },
   "devDependencies": {
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-timeout": "workspace:^",
     "cordis": "^4.0.0-rc.7"
   }
 }

packages/subprocess/subprocess-local/src/spawn.ts

@@ -15,6 +15,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { setTimeout as sleepMs } from 'node:timers/promises'
 import { scrubbedParentEnv } from '@deepseek-ai/dsh-subprocess'
+import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
 import type {
   CollectedOutput,
   SubprocessCollect,
@@ -55,47 +56,6 @@ function sleepTick(): Promise<void> {
   return sleepMs(15)
 }
 
-/** Largest delay Node schedules without collapsing it to one millisecond. */
-const MAX_TIMER_DELAY_MS = 2_147_483_647n
-
-/**
- * Schedule a positive finite millisecond delay across as many Node-safe timer
- * segments as necessary. Fractional milliseconds round up so a grace never
- * expires earlier than configured.
- * @param delayMs - positive finite delay in milliseconds.
- * @param callback - work to run after the complete delay.
- * @returns a handle that cancels the active segment and all future segments.
- */
-export function scheduleFiniteTimeout(
-  delayMs: number,
-  callback: () => void,
-): { cancel(): void } {
-  let remaining = BigInt(Math.ceil(delayMs))
-  let timer: ReturnType<typeof setTimeout> | undefined
-  const arm = (): void => {
-    const chunk = remaining > MAX_TIMER_DELAY_MS
-      ? MAX_TIMER_DELAY_MS
-      : remaining
-    remaining -= chunk
-    timer = setTimeout(() => {
-      timer = undefined
-      if (remaining === 0n) {
-        callback()
-      } else {
-        arm()
-      }
-    }, Number(chunk))
-  }
-  arm()
-  return {
-    cancel(): void {
-      if (timer === undefined) return
-      clearTimeout(timer)
-      timer = undefined
-    },
-  }
-}
-
 let spillCounter = 0
 let defaultSpillDir: string | undefined
 
@@ -339,8 +299,12 @@ function signalTree(
  * @param spec - fully resolved argv, cwd, stdio, grace, cancellation, environment.
  * @param internals - test-only spill-directory, platform, and taskkill overrides.
  * @returns live subprocess handle.
+ * @throws when `graceMs` cannot be represented by one Node timer.
  */
 export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInternals = {}): SubprocessHandle {
+  if (!Number.isFinite(spec.graceMs) || spec.graceMs <= 0 || spec.graceMs > MAX_TIMER_DELAY_MS) {
+    throw new Error(`subprocess graceMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`)
+  }
   const spillDir = internals.spillDir ?? privateSpillDir()
   const platform = internals.platform ?? process.platform
   const taskkill = internals.taskkill ?? taskkillProcessTree
@@ -382,7 +346,7 @@ export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInter
   const stdoutCollector = collectStream(outMode, child.stdout, 'stdout')
   const stderrCollector = collectStream(errMode, child.stderr, 'stderr')
 
-  let graceTimer: ReturnType<typeof scheduleFiniteTimeout> | undefined
+  let graceTimer: ReturnType<typeof setTimeout> | undefined
   let treeExitObserved = false
   let treeExitObservation: Promise<void> | undefined
   let settled = false
@@ -426,7 +390,7 @@ export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInter
     treeExitObservation ??= (async () => {
       while (treeAlive()) await sleepTick()
       treeExitObserved = true
-      graceTimer?.cancel()
+      if (graceTimer !== undefined) clearTimeout(graceTimer)
       graceTimer = undefined
     })()
     return treeExitObservation
@@ -457,7 +421,7 @@ export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInter
     // kill() re-probes tree liveness before force-killing. It stays ref'd:
     // the pending SIGKILL is a commitment, and a parent exiting before it
     // fires would orphan a trapped survivor. Self-bounds at graceMs.
-    graceTimer = scheduleFiniteTimeout(spec.graceMs, () => { kill('SIGKILL') })
+    graceTimer = setTimeout(() => { kill('SIGKILL') }, spec.graceMs)
   }
 
   // The caller owns timeout classification; this layer only reacts to abort.
@@ -472,7 +436,7 @@ export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInter
   }
 
   const done = new Promise<SubprocessOutcome>((resolve, reject) => {
-    let pipeDrainTimer: ReturnType<typeof scheduleFiniteTimeout> | undefined
+    let pipeDrainTimer: ReturnType<typeof setTimeout> | undefined
     const settle = (exitCode: number | null, signal: NodeJS.Signals | null): void => {
       if (settled) return
       settled = true
@@ -495,15 +459,15 @@ export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInter
       // A surviving descendant that inherited a pipe must not hold the
       // outcome open indefinitely: after exit, the same bounded grace that
       // governs kills also bounds the close wait.
-      pipeDrainTimer = scheduleFiniteTimeout(spec.graceMs, () => {
+      pipeDrainTimer = setTimeout(() => {
         settle(exitCode, signal)
-      })
+      }, spec.graceMs)
     })
     child.on('close', settle)
     function cleanup(): void {
       // graceTimer deliberately NOT cleared: the SIGKILL escalation must be
       // able to reach tree survivors after the direct child settles.
-      pipeDrainTimer?.cancel()
+      if (pipeDrainTimer !== undefined) clearTimeout(pipeDrainTimer)
       spec.signal?.removeEventListener('abort', onAbort)
     }
   })

packages/subprocess/subprocess-local/tests/spawn.spec.ts

@@ -5,11 +5,11 @@ import { describe, expect, it, vi } from 'vitest'
 import {
   killGroup,
   OutputCollector,
-  scheduleFiniteTimeout,
   spawnSubprocess,
   taskkillProcessTree,
 } from '../src/spawn.ts'
 import type { SubprocessHandle, SubprocessOutputReader } from '@deepseek-ai/dsh-subprocess'
+import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
 
 const { failNextClose, failNextUnlink } = vi.hoisted(() => ({
   failNextClose: { value: false },
@@ -107,31 +107,15 @@ async function waitForPidFile(path: string, timeoutMs = 5_000): Promise<number>
   throw new Error(`pid file ${path} was not written after ${timeoutMs}ms`)
 }
 
-describe('scheduleFiniteTimeout', () => {
-  it('rounds fractions up, chains Node-safe segments, and cancels idempotently', async () => {
-    vi.useFakeTimers()
-    try {
-      const fired = vi.fn()
-      const chained = scheduleFiniteTimeout(2_147_483_647.25, fired)
-      await vi.advanceTimersByTimeAsync(2_147_483_647)
-      expect(fired).not.toHaveBeenCalled()
-      await vi.advanceTimersByTimeAsync(1)
-      expect(fired).toHaveBeenCalledOnce()
-      chained.cancel()
-
-      const cancelled = vi.fn()
-      const timer = scheduleFiniteTimeout(0.25, cancelled)
-      timer.cancel()
-      timer.cancel()
-      await vi.advanceTimersByTimeAsync(1)
-      expect(cancelled).not.toHaveBeenCalled()
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-})
-
 describe('spawnSubprocess', () => {
+  it.each([0, -1, Number.NaN, Number.POSITIVE_INFINITY, MAX_TIMER_DELAY_MS + 1])(
+    'rejects an invalid grace before spawning: %s',
+    (graceMs) => {
+      expect(() => spawnSubprocess(spec('true', { graceMs })))
+        .toThrow(`subprocess graceMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`)
+    },
+  )
+
   it('captures stdout on success', async () => {
     const result = await finish(spawnSubprocess(spec('echo hello')))
     expect(result.exitCode).toBe(0)
@@ -194,17 +178,6 @@ describe('spawnSubprocess', () => {
     expect(result.signal).toBe('SIGKILL')
   })
 
-  it('cancels a larger-than-Node escalation timer once SIGTERM removes the tree', async () => {
-    const running = spawnSubprocess(spec('echo ready; sleep 60', {
-      graceMs: Number.MAX_VALUE,
-    }))
-    await waitForStdout(running, 'ready\n')
-    running.terminate()
-    const result = await running.done
-    expect(result.signal).toBe('SIGTERM')
-    await expect(running.waitForExit()).resolves.toBe(true)
-  })
-
   it('cancels escalation when the terminated group vanishes before collected pipes drain', async () => {
     const pidFile = join(spillDir, `escaped-pipe-holder-${Date.now()}.pid`)
     const graceMs = 160

packages/subprocess/subprocess-local/tsconfig.json

@@ -17,6 +17,9 @@
     {
       "path": "../subprocess"
     },
+    {
+      "path": "../../util/timeout"
+    },
     {
       "path": "../../support/invariants"
     }