Merge PR1 Codex provider fixes into PR2

packages/subagent/subagent-acp/src/run.ts

@@ -90,15 +90,41 @@ export const DEFAULT_DISPOSE_EOF_GRACE_MS = 6_000
 /** Default POSIX grace between SIGTERM and SIGKILL on dispose (the `disposeGraceMs` config). */
 export const DEFAULT_DISPOSE_GRACE_MS = 3_000
 
-/** Bounded whole-tree exit wait: polls the handle's tree liveness until it exits or `ms` elapses. */
-async function treeExitsWithin(child: SubprocessHandle, ms: number): Promise<boolean> {
-  const controller = new AbortController()
-  const timer = setTimeout(() => { controller.abort() }, ms)
-  try {
-    return await child.waitForExit(controller.signal)
-  } finally {
-    clearTimeout(timer)
+/** Largest delay Node schedules without collapsing it to one millisecond. */
+const MAX_TIMER_DELAY_MS = 2_147_483_647n
+
+function scaledFiniteMilliseconds(ms: number, scale: number): bigint {
+  const whole = Math.floor(ms)
+  return BigInt(whole) * BigInt(scale)
+    + BigInt(Math.ceil((ms - whole) * scale))
+}
+
+/**
+ * Bounded whole-tree exit wait across Node-safe timer segments.
+ * @param child - process tree whose liveness is authoritative.
+ * @param ms - positive finite base window in milliseconds.
+ * @param scale - integer multiplier applied without Number overflow.
+ */
+async function treeExitsWithin(
+  child: SubprocessHandle,
+  ms: number,
+  scale = 1,
+): Promise<boolean> {
+  let remaining = scaledFiniteMilliseconds(ms, scale)
+  while (remaining > 0n) {
+    const chunk = remaining > MAX_TIMER_DELAY_MS
+      ? MAX_TIMER_DELAY_MS
+      : remaining
+    remaining -= chunk
+    const controller = new AbortController()
+    const timer = setTimeout(() => { controller.abort() }, Number(chunk))
+    try {
+      if (await child.waitForExit(controller.signal)) return true
+    } finally {
+      clearTimeout(timer)
+    }
   }
+  return false
 }
 
 /**
@@ -125,7 +151,7 @@ export async function disposeAcpChild(child: SubprocessHandle, eofGraceMs: numbe
   // (this plugin passes disposeGraceMs there), so the bound covers both the
   // escalation window and an equal confirmation window after the SIGKILL.
   child.terminate()
-  if (!(await treeExitsWithin(child, graceMs * 2))) {
+  if (!(await treeExitsWithin(child, graceMs, 2))) {
     throw new Error('ACP child process tree did not exit within its dispose windows')
   }
 }

packages/subagent/subagent-acp/tests/subagent-acp.spec.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import Loader from '@cordisjs/plugin-loader'
 import { chmodSync, existsSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from 'node:fs'
@@ -190,6 +190,72 @@ describe('disposeAcpChild (the backend-owned teardown ladder over seam verbs)',
     await expect(disposeAcpChild(never, 20, 20)).rejects.toThrow(/did not exit within its dispose windows/)
   })
 
+  it('keeps an oversized finite escalation window instead of collapsing it to one millisecond', async () => {
+    vi.useFakeTimers()
+    try {
+      let waitCount = 0
+      let reportExited!: (exited: boolean) => void
+      const terminate = vi.fn()
+      const waitForExit = vi.fn((signal?: AbortSignal) => {
+        waitCount += 1
+        return new Promise<boolean>((resolve) => {
+          signal?.addEventListener('abort', () => { resolve(false) }, { once: true })
+          if (waitCount === 2) reportExited = resolve
+        })
+      })
+      const child: Parameters<typeof disposeAcpChild>[0] = {
+        pid: 1,
+        stdin: undefined,
+        stdout: undefined,
+        stderr: undefined,
+        collected: {},
+        done: new Promise(() => {}),
+        terminate,
+        waitForExit,
+      }
+      const disposal = disposeAcpChild(child, 0.25, Number.MAX_VALUE)
+      await vi.advanceTimersByTimeAsync(1)
+      expect(terminate).toHaveBeenCalledOnce()
+      expect(waitForExit).toHaveBeenCalledTimes(2)
+      const escalationSignal = waitForExit.mock.calls[1]?.[0]
+      await vi.advanceTimersByTimeAsync(1)
+      expect(escalationSignal?.aborted).toBe(false)
+      reportExited(true)
+      await expect(disposal).resolves.toBeUndefined()
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('chains a doubled grace beyond one Node timer segment', async () => {
+    vi.useFakeTimers()
+    try {
+      const waitForExit = vi.fn((signal?: AbortSignal) => new Promise<boolean>((resolve) => {
+        signal?.addEventListener('abort', () => { resolve(false) }, { once: true })
+      }))
+      const child: Parameters<typeof disposeAcpChild>[0] = {
+        pid: 1,
+        stdin: undefined,
+        stdout: undefined,
+        stderr: undefined,
+        collected: {},
+        done: new Promise(() => {}),
+        terminate: vi.fn(),
+        waitForExit,
+      }
+      const disposal = disposeAcpChild(child, 0.25, 1_073_741_823.75)
+      const rejected = expect(disposal).rejects.toThrow(/did not exit within its dispose windows/)
+      await vi.advanceTimersByTimeAsync(1)
+      await vi.advanceTimersByTimeAsync(2_147_483_647)
+      expect(waitForExit).toHaveBeenCalledTimes(3)
+      await vi.advanceTimersByTimeAsync(1)
+      await rejected
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('observes a spawn-level rejection and returns without a process to reap', async () => {
     const child = spawnSubprocess({
       argv: ['bash', '-c', 'true'],

packages/subagent/subagent-codex/src/run.ts

@@ -24,6 +24,47 @@ import { CodexAppServerWire } from './wire.ts'
 /** Default POSIX grace between subprocess termination tiers. */
 export const DEFAULT_DISPOSE_GRACE_MS = 3_000
 
+/** Largest delay Node schedules without collapsing it to one millisecond. */
+const MAX_TIMER_DELAY_MS = 2_147_483_647n
+
+/**
+ * Bound final exit observation at twice a positive finite grace without
+ * narrowing the public config to Node's single-timer integer range.
+ */
+function doubledGraceWindow(graceMs: number): {
+  readonly signal: AbortSignal
+  readonly cancel: () => void
+} {
+  const whole = Math.floor(graceMs)
+  let remaining = BigInt(whole) * 2n
+    + BigInt(Math.ceil((graceMs - whole) * 2))
+  const controller = new AbortController()
+  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) {
+        controller.abort()
+      } else {
+        arm()
+      }
+    }, Number(chunk))
+  }
+  arm()
+  return {
+    signal: controller.signal,
+    cancel: () => {
+      if (timer === undefined) return
+      clearTimeout(timer)
+      timer = undefined
+    },
+  }
+}
+
 /** Fully resolved inputs for one Codex app-server run. */
 export interface CodexRunSpec {
   /** Parent Session workspace, also supplied to `thread/start`. */
@@ -88,8 +129,13 @@ export async function disposeCodexChild(
     // A concurrently closed stdin does not change tree ownership below.
   }
   child.terminate()
-  if (!(await child.waitForExit(AbortSignal.timeout(graceMs * 2)))) {
-    throw new Error('subagent-codex: app-server process tree did not exit within its dispose window')
+  const exitWindow = doubledGraceWindow(graceMs)
+  try {
+    if (!(await child.waitForExit(exitWindow.signal))) {
+      throw new Error('subagent-codex: app-server process tree did not exit within its dispose window')
+    }
+  } finally {
+    exitWindow.cancel()
   }
   await child.done
 }

packages/subagent/subagent-codex/src/wire.ts

@@ -17,12 +17,17 @@ type JsonObject = Record<string, unknown>
 interface Deferred<T> {
   readonly promise: Promise<T>
   readonly resolve: (value: T) => void
+  readonly reject: (reason?: unknown) => void
 }
 
 function deferred<T>(): Deferred<T> {
   let resolve!: (value: T) => void
-  const promise = new Promise<T>((settle) => { resolve = settle })
-  return { promise, resolve }
+  let reject!: (reason?: unknown) => void
+  const promise = new Promise<T>((settle, fail) => {
+    resolve = settle
+    reject = fail
+  })
+  return { promise, resolve, reject }
 }
 
 function object(value: unknown, label: string): JsonObject {
@@ -93,7 +98,7 @@ async function raceAbort<T>(pending: Promise<T>, signal: AbortSignal): Promise<T
  */
 export class CodexAppServerWire {
   private readonly transport: JsonRpcLineTransport
-  private readonly fatal = deferred<Error>()
+  private readonly fatal = deferred<never>()
   private threadId: string | undefined
   private turnId: string | undefined
   private pendingTurnId: string | undefined
@@ -111,6 +116,10 @@ export class CodexAppServerWire {
     output: Writable,
   ) {
     this.transport = new JsonRpcLineTransport(input, output)
+    // Fatal protocol state can arrive after the current guarded operation has
+    // already settled. Keep the shared rejection observed without inserting
+    // another promise-adoption hop into active races.
+    void this.fatal.promise.catch(() => {})
     this.transport.onRequest((method, params) => this.handleServerRequest(method, params))
     this.transport.onNotification((method, params) => {
       try {
@@ -157,9 +166,8 @@ export class CodexAppServerWire {
    * Create the run's private ephemeral thread and retain its identity.
    * @param cwd - parent Session workspace.
    * @param signal - unpublished-start cancellation.
-   * @returns the app-server thread id.
    */
-  async startThread(cwd: string, signal: AbortSignal): Promise<string> {
+  async startThread(cwd: string, signal: AbortSignal): Promise<void> {
     const response = object(await this.guarded(this.transport.request('thread/start', {
       cwd,
       ephemeral: true,
@@ -170,7 +178,6 @@ export class CodexAppServerWire {
       throw new Error('subagent-codex: app-server did not create an ephemeral thread')
     }
     this.threadId = id
-    return id
   }
 
   /**
@@ -249,15 +256,12 @@ export class CodexAppServerWire {
   }
 
   private async guarded<T>(pending: Promise<T>, signal: AbortSignal): Promise<T> {
-    const withFatal = Promise.race([
-      pending,
-      this.fatal.promise.then((error): Promise<never> => Promise.reject(error)),
-    ])
+    const withFatal = Promise.race([this.fatal.promise, pending])
     return raceAbort(withFatal, signal)
   }
 
   private fail(error: Error): void {
-    this.fatal.resolve(error)
+    this.fatal.reject(error)
   }
 
   private readonly onInputError = (error: Error): void => {

packages/subagent/subagent-codex/tests/subagent-codex.spec.ts

@@ -210,7 +210,7 @@ async function initializeWire(): Promise<{
   const starting = wire.startThread(process.cwd(), new AbortController().signal)
   const threadStart = await child.peer.nextMethod('thread/start')
   child.peer.respond(threadStart, { thread: { id: 'thread-1', ephemeral: true } })
-  await expect(starting).resolves.toBe('thread-1')
+  await starting
   return { child, wire }
 }
 
@@ -369,8 +369,9 @@ describe('CodexAppServerWire', () => {
         { type: 'text', text: 'second', text_elements: [] },
       ],
     })
+    child.peer.respond(turnStart, { turn: { id: 'turn-1' } })
+    await nextTask()
     child.peer.send(
-      { id: turnStart.id, result: { turn: { id: 'turn-1' } } },
       {
         method: 'turn/started',
         params: { threadId: 'thread-1', turn: { id: 'turn-1' } },
@@ -516,6 +517,20 @@ describe('CodexAppServerWire', () => {
     }
   })
 
+  it('keeps an unsupported request authoritative over an early terminal in the same chunk', async () => {
+    const { child, wire } = await initializeWire()
+    const result = wire.runTurn(['task'], new AbortController().signal, () => false)
+    const turnStart = await child.peer.nextMethod('turn/start')
+    child.peer.send(
+      { id: turnStart.id, result: { turn: { id: 'turn-1' } } },
+      { id: 'future-request', method: 'future/request', params: {} },
+      agentMessage('early answer', 'final_answer'),
+      turnCompleted('completed'),
+    )
+    await expect(result).rejects.toThrow('unsupported app-server request')
+    wire.close()
+  })
+
   it('gives local cancellation precedence over a remote completed turn', async () => {
     const { child, wire } = await initializeWire()
     let cancelled = false
@@ -1038,6 +1053,37 @@ describe('disposeCodexChild', () => {
     expect(child.waitForExit).toHaveBeenCalledTimes(1)
   })
 
+  it('accepts fractional and larger-than-Node grace windows', async () => {
+    for (const graceMs of [0.25, Number.MAX_VALUE]) {
+      const child = fakeChild()
+      const wire = new CodexAppServerWire(child.handle.stdout!, child.handle.stdin!)
+      await expect(disposeCodexChild(wire, child.handle, graceMs))
+        .resolves.toBeUndefined()
+      const signal = vi.mocked(child.waitForExit).mock.calls[0]?.[0]
+      expect(signal?.aborted).toBe(false)
+    }
+  })
+
+  it('chains a doubled grace window beyond one Node timer segment', async () => {
+    vi.useFakeTimers()
+    try {
+      const child = fakeChild({ exitOnTerminate: false })
+      const wire = new CodexAppServerWire(child.handle.stdout!, child.handle.stdin!)
+      const disposal = disposeCodexChild(
+        wire,
+        child.handle,
+        1_073_741_823.75,
+      )
+      const rejected = expect(disposal)
+        .rejects.toThrow('did not exit within its dispose window')
+      await vi.advanceTimersByTimeAsync(2_147_483_647)
+      await vi.advanceTimersByTimeAsync(1)
+      await rejected
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('contains a concurrently closed stdin error', async () => {
     const child = fakeChild()
     const wire = new CodexAppServerWire(child.handle.stdout!, child.handle.stdin!)