fix: terminate Codex tools on stop

src/main/codex/codex-structured-session-adapter.test.ts

@@ -11,12 +11,12 @@ import type {
   openCodexAppServerConnection
 } from './codex-app-server-connection'
 import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
-import { CodexAppServerUnsupportedError } from './codex-app-server-session'
 import { CODEX_SPAWN_TOKEN_ENV } from './codex-structured-owner-identity'
 import { encodeCodexQuestionOptionId } from './codex-structured-prompt-replies'
 import {
   CodexStructuredSessionAdapter,
   type CodexStructuredLaunch,
+  type CodexStructuredSessionAdapterDeps,
   type CodexStructuredSessionEvent
 } from './codex-structured-session-adapter'
 
@@ -100,7 +100,10 @@ function fakeCodex(routes: Record<string, Route> = {}): {
 function adapterFor(
   codex: ReturnType<typeof fakeCodex>,
   launch: Partial<CodexStructuredLaunch> = {},
-  events: CodexStructuredSessionEvent[] = []
+  events: CodexStructuredSessionEvent[] = [],
+  processControl: Partial<
+    Pick<CodexStructuredSessionAdapterDeps, 'captureTurnProcesses' | 'terminateTurnProcesses'>
+  > = {}
 ): CodexStructuredSessionAdapter {
   return new CodexStructuredSessionAdapter({
     resolveLaunch: async () => ({
@@ -114,7 +117,10 @@ function adapterFor(
     onEvent: (event) => events.push(event),
     openConnection: codex.openConnection,
     readProcessStartTime: async () => 1_700_000_000_000,
-    now: () => 1_700_000_000_500
+    captureTurnProcesses: async () => ({ platform: 'win32', identities: new Map() }),
+    terminateTurnProcesses: async () => true,
+    now: () => 1_700_000_000_500,
+    ...processControl
   })
 }
 
@@ -564,58 +570,6 @@ describe('CodexStructuredSessionAdapter.dispatch', () => {
   })
 })
 
-describe('CodexStructuredSessionAdapter.cancelTurn', () => {
-  it('confirms an interrupt Codex acknowledged', async () => {
-    const codex = fakeCodex()
-    const adapter = await acquired(codex)
-
-    expect(
-      await adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
-    ).toEqual({ cancelled: true })
-    expect(codex.connections[0].calls[1]).toEqual({
-      method: 'turn/interrupt',
-      params: { threadId: THREAD_ID, turnId: 'turn-1' }
-    })
-  })
-
-  it('reports not-cancelled when Codex declines or lacks the method', async () => {
-    const declined = fakeCodex({
-      'turn/interrupt': () => {
-        throw new CodexAppServerRequestError('turn/interrupt', -32602, 'no such turn')
-      }
-    })
-    const absent = fakeCodex({
-      'turn/interrupt': () => {
-        throw new CodexAppServerUnsupportedError('no turn/interrupt')
-      }
-    })
-
-    expect(
-      await (
-        await acquired(declined)
-      ).cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
-    ).toEqual({ cancelled: false })
-    expect(
-      await (
-        await acquired(absent)
-      ).cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
-    ).toEqual({ cancelled: false })
-  })
-
-  it('rethrows an unsettled interrupt so the turn is not shown as cancelled', async () => {
-    const codex = fakeCodex({
-      'turn/interrupt': () => {
-        throw new Error('codex app-server turn/interrupt exceeded 30000ms')
-      }
-    })
-    const adapter = await acquired(codex)
-
-    await expect(
-      adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
-    ).rejects.toThrow('exceeded 30000ms')
-  })
-})
-
 describe('CodexStructuredSessionAdapter prompts', () => {
   function askApproval(codex: ReturnType<typeof fakeCodex>): void {
     codex.connections[0].handlers.onServerRequest?.({

src/main/codex/codex-structured-session-adapter.ts

@@ -11,11 +11,7 @@ import {
   type StructuredAgentSessionSetOptionInput
 } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
 import { createCodexJournalTranslator } from './codex-structured-journal-translation'
-import {
-  isCodexAppServerRequestError,
-  openCodexAppServerConnection
-} from './codex-app-server-connection'
-import { isCodexAppServerUnsupportedError } from './codex-app-server-session'
+import { openCodexAppServerConnection } from './codex-app-server-connection'
 import { codexProcessIdentity, codexProviderHandleLink } from './codex-structured-owner-identity'
 import { buildCodexStructuredChildEnvironment } from './codex-structured-child-environment'
 import { answerCodexPrompt } from './codex-structured-prompt-replies'
@@ -42,6 +38,7 @@ import {
   deliverCodexServerRequest,
   deliverCodexUnhandledFrame
 } from './codex-structured-provider-events'
+import { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation'
 
 export type {
   CodexStructuredLaunch,
@@ -52,8 +49,16 @@ export type {
 export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdapter {
   private readonly sessions = new Map<string, CodexSession>()
   private readonly acquisitions = new CodexAcquisitionRegistry()
+  private readonly turnCancellation: CodexStructuredTurnCancellation
 
-  constructor(private readonly deps: CodexStructuredSessionAdapterDeps) {}
+  constructor(private readonly deps: CodexStructuredSessionAdapterDeps) {
+    this.turnCancellation = new CodexStructuredTurnCancellation({
+      captureTurnProcesses: deps.captureTurnProcesses,
+      terminateTurnProcesses: deps.terminateTurnProcesses,
+      requestTimeoutMs: deps.requestTimeoutMs,
+      emit: (session, event) => this.emit(session, event)
+    })
+  }
 
   supportsLocation = supportsCodexStructuredLocation
 
@@ -132,7 +137,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
       }
       this.acquisitions.assertCurrent(sessionId, attempt)
       this.acquisitions.deleteIfCurrent(sessionId, attempt)
-      this.sessions.set(sessionId, {
+      const session: CodexSession = {
         connection,
         threadId: opened.threadId,
         historyPath: opened.historyPath,
@@ -141,7 +146,9 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
         reportedOptions: reportedCodexThreadOptions(opened),
         turnIdWaiters: [],
         translator
-      })
+      }
+      this.turnCancellation.register(session)
+      this.sessions.set(sessionId, session)
       for (const event of acquisition.drain()) {
         event()
       }
@@ -191,12 +198,12 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
   }
 
   private handleNotification(sessionId: string, method: string, params: unknown): void {
-    deliverCodexNotification(
-      sessionId,
-      this.sessions.get(sessionId),
-      method,
-      params,
-      (session, event) => this.emit(session, event)
+    const session = this.sessions.get(sessionId)
+    if (session && this.turnCancellation.handleNotification(sessionId, session, method, params)) {
+      return
+    }
+    deliverCodexNotification(sessionId, session, method, params, (session, event) =>
+      this.emit(session, event)
     )
   }
 
@@ -236,7 +243,9 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
     body: AgentJournalMessageItem
     fence: number
   }): Promise<AgentSessionDispatchOutcome> {
-    return dispatchCodexTurn(this.session(input.sessionId), input, this.deps.requestTimeoutMs)
+    const session = this.session(input.sessionId)
+    await this.turnCancellation.captureBaseline(session)
+    return dispatchCodexTurn(session, input, this.deps.requestTimeoutMs)
   }
 
   async cancelTurn(input: {
@@ -245,21 +254,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
     fence: number
   }): Promise<{ cancelled: boolean }> {
     const session = this.session(input.sessionId)
-    try {
-      await session.connection.request(
-        'turn/interrupt',
-        { threadId: session.threadId, turnId: input.turnId },
-        { timeoutMs: this.deps.requestTimeoutMs }
-      )
-      return { cancelled: true }
-    } catch (error) {
-      // Codex declining names a turn it no longer owns; anything else leaves the
-      // cancel unconfirmed and must surface as such.
-      if (isCodexAppServerRequestError(error) || isCodexAppServerUnsupportedError(error)) {
-        return { cancelled: false }
-      }
-      throw error
-    }
+    return this.turnCancellation.cancel(session, input.turnId)
   }
 
   async answerPrompt(input: {

src/main/codex/codex-structured-session-cancel.test.ts

@@ -0,0 +1,278 @@
+import { describe, expect, it, vi } from 'vitest'
+import type {
+  AgentJournalMessageItem,
+  AgentSessionJournalIdentity
+} from '../../shared/agent-session-journal-types'
+import {
+  CodexAppServerRequestError,
+  type CodexAppServerConnection,
+  type CodexAppServerConnectionHandlers,
+  type CodexAppServerLaunch,
+  type openCodexAppServerConnection
+} from './codex-app-server-connection'
+import { CodexAppServerUnsupportedError } from './codex-app-server-session'
+import {
+  CodexStructuredSessionAdapter,
+  type CodexStructuredSessionAdapterDeps,
+  type CodexStructuredSessionEvent
+} from './codex-structured-session-adapter'
+
+const THREAD_ID = 'thread-abc'
+const USER_MESSAGE: AgentJournalMessageItem = {
+  kind: 'message',
+  role: 'user',
+  blocks: [{ type: 'text', text: 'ship it' }]
+}
+
+type Route = (params: Record<string, unknown> | undefined) => unknown
+type FakeConnection = Omit<CodexAppServerConnection, 'closed'> & {
+  closed: boolean
+  launch: CodexAppServerLaunch
+  handlers: CodexAppServerConnectionHandlers
+  calls: { method: string; params?: Record<string, unknown> }[]
+}
+
+function identity(): AgentSessionJournalIdentity {
+  return {
+    sessionId: 'session-1',
+    workspaceId: 'ws-1',
+    hostId: 'host-1',
+    agent: 'codex',
+    providerHandle: { kind: 'codex', threadId: THREAD_ID }
+  }
+}
+
+function fakeCodex(): {
+  connections: FakeConnection[]
+  openConnection: typeof openCodexAppServerConnection
+  routes: Record<string, Route>
+} {
+  const connections: FakeConnection[] = []
+  const routes: Record<string, Route> = {
+    'thread/resume': () => ({ thread: { id: THREAD_ID } })
+  }
+  const openConnection = (async (launch, handlers = {}) => {
+    const connection: FakeConnection = {
+      launch,
+      handlers,
+      calls: [],
+      pid: 4321,
+      closed: false,
+      request: async (method, params) => {
+        connection.calls.push({ method, params })
+        return routes[method]?.(params) ?? {}
+      },
+      notify: () => {},
+      respond: () => {},
+      respondWithError: () => {},
+      close: async () => {
+        connection.closed = true
+      }
+    }
+    connections.push(connection)
+    return connection
+  }) as typeof openCodexAppServerConnection
+  return { connections, openConnection, routes }
+}
+
+async function acquired(
+  codex: ReturnType<typeof fakeCodex>,
+  events: CodexStructuredSessionEvent[] = [],
+  processControl: Partial<
+    Pick<CodexStructuredSessionAdapterDeps, 'captureTurnProcesses' | 'terminateTurnProcesses'>
+  > = {}
+): Promise<CodexStructuredSessionAdapter> {
+  const adapter = new CodexStructuredSessionAdapter({
+    resolveLaunch: async () => ({
+      command: 'codex',
+      args: ['app-server'],
+      cwd: '/work/repo',
+      codexHome: null,
+      resumeThreadId: THREAD_ID
+    }),
+    onEvent: (event) => events.push(event),
+    openConnection: codex.openConnection,
+    readProcessStartTime: async () => 1_700_000_000_000,
+    captureTurnProcesses: async () => ({ platform: 'win32', identities: new Map() }),
+    terminateTurnProcesses: async () => true,
+    ...processControl
+  })
+  await adapter.acquire({ identity: identity(), fence: 7, spawnToken: 'spawn-9' })
+  return adapter
+}
+
+function completeTurn(codex: ReturnType<typeof fakeCodex>, turnId = 'turn-1'): void {
+  codex.connections[0].handlers.onNotification?.('turn/completed', {
+    threadId: THREAD_ID,
+    turn: { id: turnId, status: 'interrupted' }
+  })
+}
+
+describe('CodexStructuredSessionAdapter.cancelTurn', () => {
+  it('confirms an interrupt Codex acknowledged', async () => {
+    const codex = fakeCodex()
+    const adapter = await acquired(codex)
+
+    await expect(
+      adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
+    ).resolves.toEqual({ cancelled: true })
+    expect(codex.connections[0].calls.at(-1)).toEqual({
+      method: 'turn/interrupt',
+      params: { threadId: THREAD_ID, turnId: 'turn-1' }
+    })
+  })
+
+  it('reports not-cancelled when Codex declines or lacks the method', async () => {
+    const declined = fakeCodex()
+    declined.routes['turn/interrupt'] = () => {
+      throw new CodexAppServerRequestError('turn/interrupt', -32602, 'no such turn')
+    }
+    const absent = fakeCodex()
+    absent.routes['turn/interrupt'] = () => {
+      throw new CodexAppServerUnsupportedError('no turn/interrupt')
+    }
+
+    await expect(
+      (await acquired(declined)).cancelTurn({
+        sessionId: 'session-1',
+        turnId: 'turn-1',
+        fence: 7
+      })
+    ).resolves.toEqual({ cancelled: false })
+    await expect(
+      (await acquired(absent)).cancelTurn({
+        sessionId: 'session-1',
+        turnId: 'turn-1',
+        fence: 7
+      })
+    ).resolves.toEqual({ cancelled: false })
+  })
+
+  it('rethrows an unsettled interrupt so the turn is not shown as cancelled', async () => {
+    const codex = fakeCodex()
+    codex.routes['turn/interrupt'] = () => {
+      throw new Error('codex app-server turn/interrupt exceeded 30000ms')
+    }
+
+    await expect(
+      (await acquired(codex)).cancelTurn({
+        sessionId: 'session-1',
+        turnId: 'turn-1',
+        fence: 7
+      })
+    ).rejects.toThrow('exceeded 30000ms')
+  })
+
+  it('publishes terminal state only after streaming interruption is physically settled', async () => {
+    const events: CodexStructuredSessionEvent[] = []
+    let finishTermination!: (terminated: boolean) => void
+    const termination = new Promise<boolean>((resolve) => {
+      finishTermination = resolve
+    })
+    const codex = fakeCodex()
+    codex.routes['turn/interrupt'] = () => {
+      completeTurn(codex)
+      return {}
+    }
+    const adapter = await acquired(codex, events, {
+      terminateTurnProcesses: async () => termination
+    })
+    codex.connections[0].handlers.onNotification?.('item/agentMessage/delta', {
+      threadId: THREAD_ID,
+      turnId: 'turn-1',
+      itemId: 'item-1',
+      delta: 'still streaming'
+    })
+
+    const pending = adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
+    await vi.waitFor(() => expect(codex.connections[0].calls.at(-1)?.method).toBe('turn/interrupt'))
+    expect(events).toContainEqual(expect.objectContaining({ method: 'item/agentMessage/delta' }))
+    expect(events).not.toContainEqual(expect.objectContaining({ method: 'turn/completed' }))
+
+    finishTermination(true)
+    await expect(pending).resolves.toEqual({ cancelled: true })
+    expect(events.at(-1)).toMatchObject({ method: 'turn/completed' })
+  })
+
+  it('starts physical termination without waiting for the interrupt receipt', async () => {
+    let finishInterrupt!: () => void
+    const interruptReceipt = new Promise<void>((resolve) => {
+      finishInterrupt = resolve
+    })
+    const terminateTurnProcesses = vi.fn(async () => true)
+    const codex = fakeCodex()
+    codex.routes['turn/interrupt'] = () => interruptReceipt
+    const adapter = await acquired(codex, [], { terminateTurnProcesses })
+
+    const pending = adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
+    await vi.waitFor(() => expect(terminateTurnProcesses).toHaveBeenCalledOnce())
+    finishInterrupt()
+
+    await expect(pending).resolves.toEqual({ cancelled: true })
+  })
+
+  it('keeps the turn live when process termination cannot be verified', async () => {
+    const events: CodexStructuredSessionEvent[] = []
+    const codex = fakeCodex()
+    codex.routes['turn/interrupt'] = () => {
+      completeTurn(codex)
+      return {}
+    }
+    const adapter = await acquired(codex, events, {
+      terminateTurnProcesses: async () => false
+    })
+
+    await expect(
+      adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
+    ).resolves.toEqual({ cancelled: false })
+    expect(events).not.toContainEqual(expect.objectContaining({ method: 'turn/completed' }))
+  })
+
+  it('accepts an immediate resend after verified interruption', async () => {
+    let nextTurn = 0
+    const codex = fakeCodex()
+    codex.routes['turn/start'] = () => ({ turn: { id: `turn-${++nextTurn}` } })
+    codex.routes['turn/interrupt'] = () => {
+      completeTurn(codex)
+      return {}
+    }
+    const adapter = await acquired(codex)
+
+    await adapter.dispatch({
+      sessionId: 'session-1',
+      clientMessageId: 'client-1',
+      body: USER_MESSAGE,
+      fence: 7
+    })
+    await adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
+
+    await expect(
+      adapter.dispatch({
+        sessionId: 'session-1',
+        clientMessageId: 'client-2',
+        body: USER_MESSAGE,
+        fence: 7
+      })
+    ).resolves.toMatchObject({
+      state: 'accepted',
+      providerIdentity: { turnId: 'turn-2' }
+    })
+  })
+
+  it('does not publish deferred completion when the interrupt receipt fails', async () => {
+    const events: CodexStructuredSessionEvent[] = []
+    const codex = fakeCodex()
+    codex.routes['turn/interrupt'] = () => {
+      completeTurn(codex)
+      throw new Error('interrupt receipt lost')
+    }
+    const adapter = await acquired(codex, events, {
+      terminateTurnProcesses: async () => true
+    })
+
+    await expect(
+      adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
+    ).rejects.toThrow('interrupt receipt lost')
+    expect(events).not.toContainEqual(expect.objectContaining({ method: 'turn/completed' }))
+  })
+})

src/main/codex/codex-structured-session-state.ts

@@ -5,6 +5,7 @@ import type {
 } from './codex-app-server-connection'
 import { CodexAcquisitionWindow } from './codex-structured-acquisition-window'
 import type { CodexJournalTranslator } from './codex-structured-journal-translation'
+import type { CodexTurnProcessSnapshot } from './codex-structured-turn-processes'
 
 export type CodexStructuredLaunch = {
   command: string
@@ -40,6 +41,11 @@ export type CodexStructuredSessionAdapterDeps = {
   mintLinkId?: () => string
   now?: () => number
   requestTimeoutMs?: number
+  captureTurnProcesses?: (rootPid: number) => Promise<CodexTurnProcessSnapshot | null>
+  terminateTurnProcesses?: (
+    rootPid: number,
+    baseline: CodexTurnProcessSnapshot | null
+  ) => Promise<boolean>
 }
 
 export type CodexSession = {

src/main/codex/codex-structured-turn-cancellation.ts

@@ -0,0 +1,150 @@
+import {
+  isCodexAppServerRequestError,
+  type CodexAppServerConnection
+} from './codex-app-server-connection'
+import { isCodexAppServerUnsupportedError } from './codex-app-server-session'
+import type {
+  CodexSession,
+  CodexStructuredSessionAdapterDeps,
+  CodexStructuredSessionEvent
+} from './codex-structured-session-state'
+import { readCodexThreadId, readCodexTurnId } from './codex-structured-thread-facts'
+import {
+  captureCodexTurnProcesses,
+  terminateCodexTurnProcesses,
+  type CodexTurnProcessSnapshot
+} from './codex-structured-turn-processes'
+
+type TurnProcessState = {
+  baseline: Promise<CodexTurnProcessSnapshot | null>
+  blockedCompletions: Set<string>
+  deferredCompletions: Map<string, CodexStructuredSessionEvent>
+}
+
+type TurnCancellationDeps = Pick<
+  CodexStructuredSessionAdapterDeps,
+  'captureTurnProcesses' | 'requestTimeoutMs' | 'terminateTurnProcesses'
+> & {
+  emit: (session: CodexSession, event: CodexStructuredSessionEvent) => void
+}
+
+export class CodexStructuredTurnCancellation {
+  private readonly states = new WeakMap<CodexSession, TurnProcessState>()
+
+  constructor(private readonly deps: TurnCancellationDeps) {}
+
+  register(session: CodexSession): void {
+    this.states.set(session, {
+      baseline: Promise.resolve(null),
+      blockedCompletions: new Set(),
+      deferredCompletions: new Map()
+    })
+  }
+
+  captureBaseline(session: CodexSession): Promise<CodexTurnProcessSnapshot | null> {
+    this.refreshBaseline(session)
+    return this.state(session).baseline
+  }
+
+  handleNotification(
+    sessionId: string,
+    session: CodexSession,
+    method: string,
+    params: unknown
+  ): boolean {
+    const threadId = readCodexThreadId(params) ?? session.threadId
+    if (method !== 'turn/completed' || threadId !== session.threadId) {
+      return false
+    }
+    const turnId = readCodexTurnId(params)
+    const state = this.state(session)
+    if (!turnId || !state.blockedCompletions.has(turnId)) {
+      return false
+    }
+    const event = {
+      type: 'notification' as const,
+      sessionId,
+      threadId,
+      method,
+      params
+    }
+    state.deferredCompletions.set(turnId, event)
+    return true
+  }
+
+  async cancel(session: CodexSession, turnId: string): Promise<{ cancelled: boolean }> {
+    const state = this.state(session)
+    state.blockedCompletions.add(turnId)
+    const baseline = await state.baseline
+    let requestError: unknown
+    const interruptReceipt = session.connection
+      .request(
+        'turn/interrupt',
+        { threadId: session.threadId, turnId },
+        { timeoutMs: this.deps.requestTimeoutMs }
+      )
+      .then(
+        () => true,
+        (error: unknown) => {
+          requestError = error
+          return false
+        }
+      )
+    const [acknowledged, terminated] = await Promise.all([
+      interruptReceipt,
+      this.terminate(session.connection, baseline)
+    ])
+    if (terminated && acknowledged) {
+      this.releaseCompletion(session, turnId)
+      return { cancelled: true }
+    }
+    if (
+      requestError &&
+      !isCodexAppServerRequestError(requestError) &&
+      !isCodexAppServerUnsupportedError(requestError)
+    ) {
+      throw requestError
+    }
+    return { cancelled: false }
+  }
+
+  private capture(pid: number | undefined): Promise<CodexTurnProcessSnapshot | null> {
+    return pid
+      ? (this.deps.captureTurnProcesses ?? captureCodexTurnProcesses)(pid)
+      : Promise.resolve(null)
+  }
+
+  private terminate(
+    connection: Pick<CodexAppServerConnection, 'pid'>,
+    baseline: CodexTurnProcessSnapshot | null
+  ): Promise<boolean> {
+    return connection.pid
+      ? (this.deps.terminateTurnProcesses ?? terminateCodexTurnProcesses)(connection.pid, baseline)
+      : Promise.resolve(false)
+  }
+
+  private refreshBaseline(session: CodexSession): void {
+    this.state(session).baseline = this.capture(session.connection.pid)
+  }
+
+  private releaseCompletion(
+    session: CodexSession,
+    turnId: string,
+    completion = this.state(session).deferredCompletions.get(turnId)
+  ): void {
+    const state = this.state(session)
+    state.blockedCompletions.delete(turnId)
+    state.deferredCompletions.delete(turnId)
+    if (completion) {
+      this.deps.emit(session, completion)
+    }
+  }
+
+  private state(session: CodexSession): TurnProcessState {
+    const state = this.states.get(session)
+    if (!state) {
+      throw new Error('codex turn process state is unavailable')
+    }
+    return state
+  }
+}

src/main/codex/codex-structured-turn-processes.integration.test.ts

@@ -0,0 +1,103 @@
+import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
+import { describe, expect, it } from 'vitest'
+import {
+  captureCodexTurnProcesses,
+  terminateCodexTurnProcesses
+} from './codex-structured-turn-processes'
+
+function nextLine(child: ChildProcessWithoutNullStreams): Promise<string> {
+  return new Promise((resolve, reject) => {
+    let buffer = ''
+    const onData = (chunk: Buffer): void => {
+      buffer += chunk.toString('utf8')
+      const newline = buffer.indexOf('\n')
+      if (newline === -1) {
+        return
+      }
+      child.stdout.off('data', onData)
+      resolve(buffer.slice(0, newline))
+    }
+    child.once('error', reject)
+    child.stdout.on('data', onData)
+  })
+}
+
+function processExists(pid: number): boolean {
+  try {
+    process.kill(pid, 0)
+    return true
+  } catch {
+    return false
+  }
+}
+
+describe.runIf(process.platform !== 'win32')('Codex structured turn process termination', () => {
+  it('removes the exact PID of a stopped 60-second command', async () => {
+    const root = spawn(
+      process.execPath,
+      [
+        '-e',
+        `const { spawn } = require('node:child_process');
+         process.stdin.once('data', () => {
+           const child = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 60000)'], { stdio: 'ignore' });
+           process.stdout.write(String(child.pid) + '\\n');
+         });
+         setTimeout(() => {}, 60000);`
+      ],
+      { stdio: ['pipe', 'pipe', 'pipe'] }
+    )
+    let commandPid = 0
+    try {
+      const baseline = await captureCodexTurnProcesses(root.pid!)
+      root.stdin.write('start\n')
+      commandPid = Number(await nextLine(root))
+      expect(processExists(commandPid)).toBe(true)
+
+      await expect(terminateCodexTurnProcesses(root.pid!, baseline)).resolves.toBe(true)
+      expect(processExists(commandPid)).toBe(false)
+    } finally {
+      if (commandPid > 0 && processExists(commandPid)) {
+        process.kill(commandPid, 'SIGKILL')
+      }
+      root.kill('SIGKILL')
+    }
+  }, 15_000)
+
+  it('preserves descendants that predate the turn', async () => {
+    const root = spawn(
+      process.execPath,
+      [
+        '-e',
+        `const { spawn } = require('node:child_process');
+         const persistent = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 60000)'], { stdio: 'ignore' });
+         process.stdout.write(String(persistent.pid) + '\\n');
+         process.stdin.once('data', () => {
+           const child = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 60000)'], { stdio: 'ignore' });
+           process.stdout.write(String(child.pid) + '\\n');
+         });
+         setTimeout(() => {}, 60000);`
+      ],
+      { stdio: ['pipe', 'pipe', 'pipe'] }
+    )
+    let persistentPid = 0
+    let commandPid = 0
+    try {
+      persistentPid = Number(await nextLine(root))
+      const baseline = await captureCodexTurnProcesses(root.pid!)
+      root.stdin.write('start\n')
+      commandPid = Number(await nextLine(root))
+
+      await expect(terminateCodexTurnProcesses(root.pid!, baseline)).resolves.toBe(true)
+      expect(processExists(persistentPid)).toBe(true)
+      expect(processExists(commandPid)).toBe(false)
+    } finally {
+      if (persistentPid > 0 && processExists(persistentPid)) {
+        process.kill(persistentPid, 'SIGKILL')
+      }
+      if (commandPid > 0 && processExists(commandPid)) {
+        process.kill(commandPid, 'SIGKILL')
+      }
+      root.kill('SIGKILL')
+    }
+  }, 15_000)
+})

src/main/codex/codex-structured-turn-processes.ts

@@ -0,0 +1,85 @@
+import { captureDescendantSnapshot, type DescendantSnapshot } from '../pty-descendant-termination'
+import { terminateDescendantSnapshotAndWait } from '../pty-descendant-exit-verification'
+import { queryWindowsProcessDescendants } from '../providers/windows-foreground-process-rows'
+import { terminateWindowsProcessTree } from '../windows-process-tree-kill'
+
+export type CodexTurnProcessSnapshot =
+  | { platform: 'posix'; snapshot: DescendantSnapshot }
+  | { platform: 'win32'; identities: ReadonlyMap<number, string> }
+
+function windowsIdentity(row: {
+  ppid: number
+  name: string
+  command: string
+  executablePath: string
+}): string {
+  return [row.ppid, row.name, row.command, row.executablePath].join('\0')
+}
+
+export async function captureCodexTurnProcesses(
+  rootPid: number
+): Promise<CodexTurnProcessSnapshot | null> {
+  if (process.platform === 'win32') {
+    const descendants = await queryWindowsProcessDescendants(rootPid, { fresh: true })
+    return descendants
+      ? {
+          platform: 'win32',
+          identities: new Map(descendants.map((row) => [row.pid, windowsIdentity(row)]))
+        }
+      : null
+  }
+  const snapshot = await captureDescendantSnapshot(rootPid)
+  return snapshot ? { platform: 'posix', snapshot } : null
+}
+
+function addedPosixDescendants(
+  baseline: DescendantSnapshot,
+  current: DescendantSnapshot
+): DescendantSnapshot {
+  const baselineRows = new Map(baseline.descendants.map((row) => [row.pid, row]))
+  return {
+    ...current,
+    descendants: current.descendants.filter((row) => {
+      const prior = baselineRows.get(row.pid)
+      return prior?.startedAt !== row.startedAt || prior.pgid !== row.pgid
+    })
+  }
+}
+
+async function terminateWindowsAddedProcesses(
+  rootPid: number,
+  baseline: ReadonlyMap<number, string>
+): Promise<boolean> {
+  const current = await queryWindowsProcessDescendants(rootPid, { fresh: true })
+  if (!current) {
+    return false
+  }
+  const added = current.filter((row) => baseline.get(row.pid) !== windowsIdentity(row))
+  const addedPids = new Set(added.map((row) => row.pid))
+  const roots = added.filter((row) => !addedPids.has(row.ppid))
+  await Promise.all(roots.map((row) => terminateWindowsProcessTree(row.pid)))
+  const targetIdentities = new Map(added.map((row) => [row.pid, windowsIdentity(row)]))
+  const remaining = await queryWindowsProcessDescendants(rootPid, { fresh: true })
+  return (
+    remaining !== null &&
+    remaining.every((row) => targetIdentities.get(row.pid) !== windowsIdentity(row))
+  )
+}
+
+export async function terminateCodexTurnProcesses(
+  rootPid: number,
+  baseline: CodexTurnProcessSnapshot | null
+): Promise<boolean> {
+  if (!baseline) {
+    return false
+  }
+  if (baseline.platform === 'win32') {
+    return terminateWindowsAddedProcesses(rootPid, baseline.identities)
+  }
+  const current = await captureDescendantSnapshot(rootPid)
+  if (!current) {
+    return false
+  }
+  const added = addedPosixDescendants(baseline.snapshot, current)
+  return terminateDescendantSnapshotAndWait(added)
+}

src/main/pty-descendant-exit-verification.ts

@@ -0,0 +1,78 @@
+import {
+  DESCENDANT_KILL_GRACE_MS,
+  DESCENDANT_SNAPSHOT_TIMEOUT_MS,
+  hasUnambiguousStartIdentity,
+  readProcessTable,
+  readProcessTableBeforeDeadline,
+  sendDescendantSignal,
+  type DescendantSnapshot,
+  type ProcessTableRow,
+  type TerminateDeps
+} from './pty-descendant-termination'
+
+export const DESCENDANT_KILL_VERIFY_MS = 3_500
+
+function waitForDelay(ms: number): Promise<void> {
+  return new Promise((resolve) => {
+    const timer = setTimeout(resolve, ms)
+    timer.unref?.()
+  })
+}
+
+function matchingSnapshotRows(
+  snapshot: DescendantSnapshot,
+  table: readonly ProcessTableRow[]
+): ProcessTableRow[] {
+  const expected = new Map(snapshot.descendants.map((row) => [row.pid, row]))
+  return table.filter((live) => {
+    const row = expected.get(live.pid)
+    return row?.startedAt === live.startedAt && row.pgid === live.pgid
+  })
+}
+
+type VerificationDeps = TerminateDeps & {
+  verifyMs?: number
+}
+
+/** An unreadable process table is never proof that a stopped descendant exited. */
+export async function terminateDescendantSnapshotAndWait(
+  snapshot: DescendantSnapshot,
+  deps: VerificationDeps = {}
+): Promise<boolean> {
+  const sendSignal = deps.sendSignal ?? sendDescendantSignal
+  const readTable = deps.readTable ?? readProcessTable
+  const graceMs = deps.graceMs ?? DESCENDANT_KILL_GRACE_MS
+  const verifyMs = deps.verifyMs ?? DESCENDANT_KILL_VERIFY_MS
+  const deadline = Date.now() + verifyMs
+  for (const row of snapshot.descendants) {
+    sendSignal(row.pid, 'SIGTERM')
+  }
+  let forced = false
+  while (Date.now() < deadline) {
+    const capture = await readProcessTableBeforeDeadline(
+      readTable,
+      deps.timeoutMs ?? DESCENDANT_SNAPSHOT_TIMEOUT_MS
+    )
+    if (!capture) {
+      return false
+    }
+    const live = matchingSnapshotRows(snapshot, capture.rows)
+    if (live.length === 0) {
+      return true
+    }
+    if (!forced && Date.now() >= deadline - verifyMs + graceMs) {
+      forced = true
+      for (const row of live) {
+        if (hasUnambiguousStartIdentity(row, snapshot.capturedAtMs)) {
+          sendSignal(row.pid, 'SIGKILL')
+        }
+      }
+    }
+    await waitForDelay(50)
+  }
+  const finalCapture = await readProcessTableBeforeDeadline(
+    readTable,
+    deps.timeoutMs ?? DESCENDANT_SNAPSHOT_TIMEOUT_MS
+  )
+  return finalCapture !== null && matchingSnapshotRows(snapshot, finalCapture.rows).length === 0
+}

src/main/pty-descendant-termination.test.ts

@@ -15,6 +15,7 @@ import {
   type ProcessTableCapture,
   type ProcessTableRow
 } from './pty-descendant-termination'
+import { terminateDescendantSnapshotAndWait } from './pty-descendant-exit-verification'
 
 const CAPTURED_AT_MS = Date.parse('Tue Jul 14 12:00:00 2026')
 
@@ -299,6 +300,49 @@ describe('terminateDescendantSnapshot', () => {
   })
 })
 
+describe('terminateDescendantSnapshotAndWait', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+  })
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('escalates an identity-matched survivor and verifies its exit', async () => {
+    const survivor = row(20, 10, 20)
+    const sendSignal = vi.fn()
+    const readTable = vi
+      .fn()
+      .mockResolvedValueOnce(tableCapture([survivor]))
+      .mockResolvedValueOnce(tableCapture([]))
+
+    const pending = terminateDescendantSnapshotAndWait(snapshot([survivor]), {
+      sendSignal,
+      readTable,
+      graceMs: 0,
+      verifyMs: 200
+    })
+    await vi.advanceTimersByTimeAsync(50)
+
+    await expect(pending).resolves.toBe(true)
+    expect(sendSignal.mock.calls).toEqual([
+      [20, 'SIGTERM'],
+      [20, 'SIGKILL']
+    ])
+  })
+
+  it('does not claim exit when the verification table is unavailable', async () => {
+    const sendSignal = vi.fn()
+    const result = await terminateDescendantSnapshotAndWait(snapshot([row(20, 10, 20)]), {
+      sendSignal,
+      readTable: vi.fn().mockRejectedValue(new Error('ps exploded'))
+    })
+
+    expect(result).toBe(false)
+    expect(sendSignal).toHaveBeenCalledWith(20, 'SIGTERM')
+  })
+})
+
 describe('createProcessTableSnapshotReader', () => {
   it('coalesces same-turn teardown requests onto one fresh scan', async () => {
     const capture = tableCapture([row(10, 1, 10)])

src/main/pty-descendant-termination.ts

@@ -119,9 +119,9 @@ export function createProcessTableSnapshotReader(
   }
 }
 
-const readProcessTable = createProcessTableSnapshotReader(readFreshProcessTable)
+export const readProcessTable = createProcessTableSnapshotReader(readFreshProcessTable)
 
-function readProcessTableBeforeDeadline(
+export function readProcessTableBeforeDeadline(
   readTable: ProcessTableReader,
   timeoutMs: number
 ): Promise<ProcessTableCapture | null> {
@@ -284,7 +284,7 @@ export async function killWithDescendantSweep(
   }
 }
 
-function defaultSendSignal(pid: number, signal: NodeJS.Signals): void {
+export function sendDescendantSignal(pid: number, signal: NodeJS.Signals): void {
   try {
     process.kill(pid, signal)
   } catch {
@@ -292,14 +292,14 @@ function defaultSendSignal(pid: number, signal: NodeJS.Signals): void {
   }
 }
 
-type TerminateDeps = {
+export type TerminateDeps = {
   readTable?: ProcessTableReader
   sendSignal?: SignalSender
   graceMs?: number
   timeoutMs?: number
 }
 
-function hasUnambiguousStartIdentity(row: ProcessTableRow, capturedAtMs: number): boolean {
+export function hasUnambiguousStartIdentity(row: ProcessTableRow, capturedAtMs: number): boolean {
   const startedAtMs = Date.parse(row.startedAt)
   if (!Number.isFinite(startedAtMs)) {
     return false
@@ -319,7 +319,7 @@ export function terminateDescendantSnapshot(
   snapshot: DescendantSnapshot,
   deps: TerminateDeps = {}
 ): void {
-  const sendSignal = deps.sendSignal ?? defaultSendSignal
+  const sendSignal = deps.sendSignal ?? sendDescendantSignal
   const readTable = deps.readTable ?? readProcessTable
   for (const row of snapshot.descendants) {
     sendSignal(row.pid, 'SIGTERM')