fix: cancel superseded relay exec children (#4283)

src/relay/agent-exec-handler.test.ts

@@ -197,6 +197,100 @@ describe('AgentExecHandler', () => {
     })
   })
 
+  it('kills the active command when the request aborts', async () => {
+    const child = createFakeChild()
+    spawnMock.mockReturnValue(child as never)
+    const handlers = createHandlers()
+    const controller = new AbortController()
+
+    const pending = handlers.get('agent.execNonInteractive')!(
+      {
+        binary: 'agent',
+        args: [],
+        cwd: '/repo',
+        stdin: null,
+        timeoutMs: 5_000
+      },
+      { clientId: 1, isStale: () => controller.signal.aborted, signal: controller.signal }
+    )
+
+    controller.abort()
+
+    if (process.platform === 'win32') {
+      expect(execMock).toHaveBeenCalledWith('taskkill /pid 12345 /T /F', expect.any(Function))
+    } else {
+      expect(child.kill).toHaveBeenCalledWith('SIGKILL')
+    }
+
+    child.emit('close', null)
+    await expect(pending).resolves.toMatchObject({
+      exitCode: null,
+      timedOut: false,
+      canceled: true
+    })
+    expect(child.stdout.listenerCount('data')).toBe(0)
+    expect(child.stderr.listenerCount('data')).toBe(0)
+    expect(child.listenerCount('error')).toBe(0)
+    expect(child.listenerCount('close')).toBe(0)
+  })
+
+  it('cancels a superseded command in the same operation lane', async () => {
+    const firstChild = createFakeChild()
+    const secondChild = createFakeChild()
+    secondChild.pid = 12346
+    spawnMock.mockReturnValueOnce(firstChild as never).mockReturnValueOnce(secondChild as never)
+    const handlers = createHandlers()
+
+    const first = handlers.get('agent.execNonInteractive')!(
+      {
+        binary: 'agent',
+        args: ['first'],
+        cwd: '/repo',
+        stdin: null,
+        timeoutMs: 5_000,
+        operation: 'commit-message'
+      },
+      requestContext()
+    )
+    const second = handlers.get('agent.execNonInteractive')!(
+      {
+        binary: 'agent',
+        args: ['second'],
+        cwd: '/repo',
+        stdin: null,
+        timeoutMs: 5_000,
+        operation: 'commit-message'
+      },
+      requestContext()
+    )
+
+    if (process.platform === 'win32') {
+      expect(execMock).toHaveBeenCalledWith('taskkill /pid 12345 /T /F', expect.any(Function))
+      expect(execMock).not.toHaveBeenCalledWith('taskkill /pid 12346 /T /F', expect.any(Function))
+    } else {
+      expect(firstChild.kill).toHaveBeenCalledWith('SIGKILL')
+      expect(secondChild.kill).not.toHaveBeenCalled()
+    }
+
+    await expect(
+      handlers.get('agent.cancelExec')!(
+        { cwd: '/repo', operation: 'commit-message' },
+        requestContext()
+      )
+    ).resolves.toEqual({ canceled: true })
+
+    if (process.platform === 'win32') {
+      expect(execMock).toHaveBeenCalledWith('taskkill /pid 12346 /T /F', expect.any(Function))
+    } else {
+      expect(secondChild.kill).toHaveBeenCalledWith('SIGKILL')
+    }
+
+    firstChild.emit('close', null)
+    secondChild.emit('close', null)
+    await expect(first).resolves.toMatchObject({ canceled: true })
+    await expect(second).resolves.toMatchObject({ canceled: true })
+  })
+
   it('reports when cancellation has no matching in-flight command', async () => {
     const handlers = createHandlers()
 

src/relay/agent-exec-handler.ts

@@ -1,7 +1,7 @@
 import { exec, spawn, type ChildProcess } from 'child_process'
 import { existsSync } from 'fs'
 import { delimiter, join } from 'path'
-import type { RelayDispatcher } from './dispatcher'
+import type { RelayDispatcher, RequestContext } from './dispatcher'
 
 const DEFAULT_TIMEOUT_MS = 60_000
 const MAX_TIMEOUT_MS = 5 * 60 * 1000
@@ -107,7 +107,7 @@ function laneKeyFor(cwd: string, operation: unknown): string {
   return JSON.stringify([op, cwd])
 }
 
-type InFlightExec = { child: ChildProcess; markCanceled: () => void }
+type InFlightExec = { child: ChildProcess; cancel: () => void }
 
 type ExecResult = {
   stdout: string
@@ -137,7 +137,9 @@ export class AgentExecHandler {
   }
 
   constructor(dispatcher: RelayDispatcher) {
-    dispatcher.onRequest('agent.execNonInteractive', (p) => this.exec(p as ExecParams))
+    dispatcher.onRequest('agent.execNonInteractive', (p, context) =>
+      this.exec(p as ExecParams, context)
+    )
     dispatcher.onRequest('agent.cancelExec', (p) => this.cancel(p as CancelParams))
   }
 
@@ -147,12 +149,11 @@ export class AgentExecHandler {
     if (!entry) {
       return { canceled: false }
     }
-    entry.markCanceled()
-    killProcessTree(entry.child)
+    entry.cancel()
     return { canceled: true }
   }
 
-  private async exec(params: ExecParams): Promise<ExecResult> {
+  private async exec(params: ExecParams, context?: RequestContext): Promise<ExecResult> {
     const binary = typeof params.binary === 'string' ? params.binary : ''
     if (!binary) {
       throw new Error('agent.execNonInteractive: binary is required')
@@ -201,6 +202,7 @@ export class AgentExecHandler {
       let entry: InFlightExec | null = null
       let timer: ReturnType<typeof setTimeout> | null = null
       let detachChildListeners = (): void => {}
+      let detachRequestAbortListener = (): void => {}
       const finish = (result: ExecResult): void => {
         if (settled) {
           return
@@ -210,18 +212,26 @@ export class AgentExecHandler {
           clearTimeout(timer)
           timer = null
         }
+        detachRequestAbortListener()
         detachChildListeners()
         if (laneKey && entry && this.inFlightByLane.get(laneKey) === entry) {
           this.inFlightByLane.delete(laneKey)
         }
         resolve(result)
       }
+      const cancelCurrent = (): void => {
+        canceled = true
+        killProcessTree(child)
+      }
       if (laneKey) {
+        // Why: the relay owns one visible non-interactive job per cwd+operation.
+        // Replacing the lane without canceling the prior child would orphan
+        // that process until timeout because future cancelExec calls reach only
+        // the newest map entry.
+        this.inFlightByLane.get(laneKey)?.cancel()
         entry = {
           child,
-          markCanceled: () => {
-            canceled = true
-          }
+          cancel: cancelCurrent
         }
         this.inFlightByLane.set(laneKey, entry)
       }
@@ -274,6 +284,17 @@ export class AgentExecHandler {
         child.off('close', onClose)
       }
 
+      if (context?.signal) {
+        if (context.signal.aborted) {
+          cancelCurrent()
+        } else {
+          context.signal.addEventListener('abort', cancelCurrent, { once: true })
+          detachRequestAbortListener = () => {
+            context.signal?.removeEventListener('abort', cancelCurrent)
+          }
+        }
+      }
+
       if (stdinPayload !== null) {
         child.stdin?.end(stdinPayload)
       } else {