diff --git a/src/main/codex/codex-structured-owner-identity.ts b/src/main/codex/codex-structured-owner-identity.ts
new file mode 100644
index 00000000000..c27979cf8c5
--- /dev/null
+++ b/src/main/codex/codex-structured-owner-identity.ts
@@ -0,0 +1,48 @@
+import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types'
+import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
+import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record'
+import { readProcessStartTimeMs } from '../runtime/agent-session-process-identity-probe'
+
+// What the lease records about the child Codex just handed back: the process it
+// will later re-prove, and the provider handle link the journal binds to. Both
+// must describe the thread Codex actually opened, never the one a client asked
+// for.
+
+/** The child echoes its spawn token here so the owner probe can tell a live
+ *  child of THIS reservation from a same-pid stranger. */
+export const CODEX_SPAWN_TOKEN_ENV = 'ORCA_AGENT_SESSION_SPAWN_TOKEN'
+
+export async function codexProcessIdentity(
+  input: {
+    identity: AgentSessionJournalIdentity
+    spawnToken: string
+    pid: number | undefined
+  },
+  readStartTime: (pid: number) => Promise<number | null> = readProcessStartTimeMs
+): Promise<AgentSessionProcessIdentity> {
+  if (input.pid === undefined) {
+    throw new Error('codex app-server started without a pid')
+  }
+  return {
+    hostId: input.identity.hostId,
+    pid: input.pid,
+    processStartTimeMs: await readStartTime(input.pid),
+    spawnToken: input.spawnToken
+  }
+}
+
+export function codexProviderHandleLink(input: {
+  threadId: string
+  resumed: boolean
+  fence: number
+  linkId?: string
+  observedAt: number
+}): AgentSessionProviderHandleLink {
+  return {
+    linkId: input.linkId ?? `codex-${input.fence}-${input.threadId}`.slice(0, 128),
+    handle: { provider: 'codex', threadId: input.threadId },
+    origin: input.resumed ? 'resumed' : 'created',
+    mintedAtFence: input.fence,
+    observedAt: input.observedAt
+  }
+}
diff --git a/src/main/codex/codex-structured-session-adapter.test.ts b/src/main/codex/codex-structured-session-adapter.test.ts
new file mode 100644
index 00000000000..e1dac82894f
--- /dev/null
+++ b/src/main/codex/codex-structured-session-adapter.test.ts
@@ -0,0 +1,600 @@
+import { describe, expect, it, vi } from 'vitest'
+import type {
+  AgentJournalMessageItem,
+  AgentSessionJournalIdentity
+} from '../../shared/agent-session-journal-types'
+import { CodexAppServerRequestError } from './codex-app-server-connection'
+import type {
+  CodexAppServerConnection,
+  CodexAppServerConnectionHandlers,
+  CodexAppServerLaunch,
+  openCodexAppServerConnection
+} from './codex-app-server-connection'
+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 CodexStructuredSessionEvent
+} from './codex-structured-session-adapter'
+
+const THREAD_ID = 'thread-abc'
+
+function identityFor(sessionId: string): AgentSessionJournalIdentity {
+  return {
+    sessionId,
+    workspaceId: 'ws-1',
+    hostId: 'host-1',
+    agent: 'codex',
+    providerHandle: { kind: 'codex', threadId: THREAD_ID }
+  }
+}
+
+const USER_MESSAGE: AgentJournalMessageItem = {
+  kind: 'message',
+  role: 'user',
+  blocks: [{ type: 'text', text: 'ship it' }]
+}
+
+type Route = (params: Record<string, unknown> | undefined) => unknown
+
+type FakeConnection = CodexAppServerConnection & {
+  launch: CodexAppServerLaunch
+  handlers: CodexAppServerConnectionHandlers
+  calls: { method: string; params?: Record<string, unknown> }[]
+  replies: { id: number | string; result?: unknown; code?: number; message?: string }[]
+  closeCount: number
+}
+
+/** Stands in for a live `codex app-server`: every RPC is answered from `routes`,
+ *  and the test drives Codex's own traffic through `handlers`. */
+function fakeCodex(routes: Record<string, Route> = {}): {
+  connections: FakeConnection[]
+  openConnection: typeof openCodexAppServerConnection
+  routes: Record<string, Route>
+} {
+  const connections: FakeConnection[] = []
+  const openConnection = (async (launch, handlers = {}) => {
+    const connection: FakeConnection = {
+      launch,
+      handlers,
+      calls: [],
+      replies: [],
+      closeCount: 0,
+      pid: 4321,
+      closed: false,
+      request: async (method, params) => {
+        connection.calls.push({ method, params })
+        const route = routes[method]
+        return route ? route(params) : {}
+      },
+      notify: () => {},
+      respond: (id, result) => connection.replies.push({ id, result }),
+      respondWithError: (id, code, message) => connection.replies.push({ id, code, message }),
+      close: async () => {
+        connection.closeCount += 1
+      }
+    }
+    connections.push(connection)
+    return connection
+  }) as typeof openCodexAppServerConnection
+  routes['thread/start'] ??= () => ({ thread: { id: THREAD_ID, path: '/rollouts/abc.jsonl' } })
+  routes['thread/resume'] ??= (params) => ({
+    thread: { id: (params as { threadId: string }).threadId }
+  })
+  return { connections, openConnection, routes }
+}
+
+function adapterFor(
+  codex: ReturnType<typeof fakeCodex>,
+  launch: Partial<CodexStructuredLaunch> = {},
+  events: CodexStructuredSessionEvent[] = []
+): CodexStructuredSessionAdapter {
+  return new CodexStructuredSessionAdapter({
+    resolveLaunch: async () => ({
+      command: 'codex',
+      args: ['app-server'],
+      cwd: '/work/repo',
+      codexHome: null,
+      resumeThreadId: null,
+      ...launch
+    }),
+    onEvent: (event) => events.push(event),
+    openConnection: codex.openConnection,
+    readProcessStartTime: async () => 1_700_000_000_000,
+    now: () => 1_700_000_000_500
+  })
+}
+
+async function acquired(
+  codex: ReturnType<typeof fakeCodex>,
+  launch: Partial<CodexStructuredLaunch> = {},
+  events: CodexStructuredSessionEvent[] = []
+): Promise<CodexStructuredSessionAdapter> {
+  const adapter = adapterFor(codex, launch, events)
+  await adapter.acquire({ identity: identityFor('session-1'), fence: 7, spawnToken: 'spawn-9' })
+  return adapter
+}
+
+describe('CodexStructuredSessionAdapter.acquire', () => {
+  it('starts a new thread and reports the process and link the lease will prove', async () => {
+    const codex = fakeCodex()
+    const adapter = adapterFor(codex, { codexHome: '/codex/home' })
+
+    const acquisition = await adapter.acquire({
+      identity: identityFor('session-1'),
+      fence: 7,
+      spawnToken: 'spawn-9'
+    })
+
+    expect(codex.connections[0].launch.env).toEqual({
+      [CODEX_SPAWN_TOKEN_ENV]: 'spawn-9',
+      CODEX_HOME: '/codex/home'
+    })
+    expect(codex.connections[0].calls[0]).toEqual({
+      method: 'thread/start',
+      params: { cwd: '/work/repo' }
+    })
+    expect(acquisition.process).toEqual({
+      hostId: 'host-1',
+      pid: 4321,
+      processStartTimeMs: 1_700_000_000_000,
+      spawnToken: 'spawn-9'
+    })
+    expect(acquisition.link).toEqual({
+      linkId: `codex-7-${THREAD_ID}`,
+      handle: { provider: 'codex', threadId: THREAD_ID },
+      origin: 'created',
+      mintedAtFence: 7,
+      observedAt: 1_700_000_000_500
+    })
+  })
+
+  it('resumes the thread the durable handle chain names, not the client one', async () => {
+    const codex = fakeCodex()
+    const adapter = adapterFor(codex, { resumeThreadId: 'thread-proven' })
+
+    const acquisition = await adapter.acquire({
+      identity: identityFor('session-1'),
+      fence: 9,
+      spawnToken: 'spawn-9'
+    })
+
+    expect(codex.connections[0].calls[0]).toEqual({
+      method: 'thread/resume',
+      params: { threadId: 'thread-proven', cwd: '/work/repo' }
+    })
+    expect(acquisition.link.origin).toBe('resumed')
+    expect(acquisition.link.handle).toEqual({ provider: 'codex', threadId: 'thread-proven' })
+  })
+
+  it('refuses a resume that lands on a different thread and reaps the child', async () => {
+    const codex = fakeCodex({ 'thread/resume': () => ({ thread: { id: 'thread-other' } }) })
+    const adapter = adapterFor(codex, { resumeThreadId: 'thread-proven' })
+
+    await expect(
+      adapter.acquire({ identity: identityFor('session-1'), fence: 9, spawnToken: 'spawn-9' })
+    ).rejects.toThrow('resumed thread-other instead of thread-proven')
+    expect(codex.connections[0].closeCount).toBe(1)
+  })
+
+  it('refuses a thread Codex never named', async () => {
+    const codex = fakeCodex({ 'thread/start': () => ({}) })
+    const adapter = adapterFor(codex)
+
+    await expect(
+      adapter.acquire({ identity: identityFor('session-1'), fence: 1, spawnToken: 'spawn-9' })
+    ).rejects.toThrow('did not name the thread')
+    expect(codex.connections[0].closeCount).toBe(1)
+  })
+
+  it('closes the previous child before re-acquiring at a new fence', async () => {
+    const codex = fakeCodex()
+    const adapter = await acquired(codex)
+
+    await adapter.acquire({ identity: identityFor('session-1'), fence: 8, spawnToken: 'spawn-10' })
+
+    expect(codex.connections).toHaveLength(2)
+    expect(codex.connections[0].closeCount).toBe(1)
+    expect(codex.connections[1].closeCount).toBe(0)
+  })
+
+  it('reports the rollout path Codex named, and null when it named none', async () => {
+    const withPath = fakeCodex()
+    const adapter = await acquired(withPath)
+    expect(await adapter.historyFilePath({ identity: identityFor('session-1') })).toBe(
+      '/rollouts/abc.jsonl'
+    )
+
+    const withoutPath = fakeCodex({ 'thread/start': () => ({ thread: { id: THREAD_ID } }) })
+    const bare = await acquired(withoutPath)
+    expect(await bare.historyFilePath({ identity: identityFor('session-1') })).toBeNull()
+  })
+})
+
+describe('CodexStructuredSessionAdapter.dispatch', () => {
+  it('accepts a turn Codex names in its response', async () => {
+    const codex = fakeCodex({ 'turn/start': () => ({ turn: { id: 'turn-1' } }) })
+    const adapter = await acquired(codex)
+
+    const outcome = await adapter.dispatch({
+      sessionId: 'session-1',
+      clientMessageId: 'client-1',
+      body: {
+        kind: 'message',
+        role: 'user',
+        blocks: [
+          { type: 'text', text: 'ship it' },
+          { type: 'image-ref', path: '/tmp/shot.png' },
+          { type: 'image-ref', url: 'https://example.test/a.png' }
+        ]
+      },
+      fence: 7
+    })
+
+    expect(outcome).toEqual({
+      state: 'accepted',
+      providerIdentity: { provider: 'codex', threadId: THREAD_ID, turnId: 'turn-1', ordinal: 0 }
+    })
+    expect(codex.connections[0].calls[1].params).toEqual({
+      threadId: THREAD_ID,
+      clientUserMessageId: 'client-1',
+      input: [
+        { type: 'text', text: 'ship it' },
+        { type: 'localImage', path: '/tmp/shot.png' },
+        { type: 'image', url: 'https://example.test/a.png' }
+      ]
+    })
+  })
+
+  it('accepts a turn named only by the notification that raced the ack', async () => {
+    const codex = fakeCodex()
+    const events: CodexStructuredSessionEvent[] = []
+    const adapter = await acquired(codex, {}, events)
+    codex.routes['turn/start'] = () => {
+      codex.connections[0].handlers.onNotification?.('turn/started', {
+        threadId: THREAD_ID,
+        turn: { id: 'turn-late' }
+      })
+      return {}
+    }
+
+    const outcome = await adapter.dispatch({
+      sessionId: 'session-1',
+      clientMessageId: 'client-1',
+      body: USER_MESSAGE,
+      fence: 7
+    })
+
+    expect(outcome).toMatchObject({ state: 'accepted' })
+    expect(outcome).toMatchObject({ providerIdentity: { turnId: 'turn-late' } })
+    expect(events.at(-1)).toMatchObject({ type: 'notification', method: 'turn/started' })
+  })
+
+  it('settles unknown rather than failed when Codex never names the turn', async () => {
+    vi.useFakeTimers()
+    try {
+      const codex = fakeCodex()
+      const adapter = await acquired(codex)
+
+      const dispatching = adapter.dispatch({
+        sessionId: 'session-1',
+        clientMessageId: 'client-1',
+        body: USER_MESSAGE,
+        fence: 7
+      })
+      await vi.advanceTimersByTimeAsync(10_000)
+
+      expect(await dispatching).toEqual({
+        state: 'unknown',
+        reason: 'codex app-server started a turn it did not name in time'
+      })
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('rejects only when Codex answered and declined', async () => {
+    const codex = fakeCodex({
+      'turn/start': () => {
+        throw new CodexAppServerRequestError('turn/start', -32602, 'turn already running')
+      }
+    })
+    const adapter = await acquired(codex)
+
+    expect(
+      await adapter.dispatch({
+        sessionId: 'session-1',
+        clientMessageId: 'client-1',
+        body: USER_MESSAGE,
+        fence: 7
+      })
+    ).toEqual({ state: 'rejected', reason: 'turn already running' })
+  })
+
+  it('rethrows a dead child so the wire settles the submission unknown', async () => {
+    const codex = fakeCodex({
+      'turn/start': () => {
+        throw new Error('codex app-server connection ended')
+      }
+    })
+    const adapter = await acquired(codex)
+
+    await expect(
+      adapter.dispatch({
+        sessionId: 'session-1',
+        clientMessageId: 'client-1',
+        body: USER_MESSAGE,
+        fence: 7
+      })
+    ).rejects.toThrow('connection ended')
+  })
+
+  it('applies an option change to the next turn only', async () => {
+    const codex = fakeCodex({ 'turn/start': () => ({ turn: { id: 'turn-1' } }) })
+    const adapter = await acquired(codex)
+
+    await adapter.setOption({ sessionId: 'session-1', key: 'model', value: 'gpt-5', fence: 7 })
+    await adapter.setOption({ sessionId: 'session-1', key: 'effort', value: 'high', fence: 7 })
+    await expect(
+      adapter.setOption({ sessionId: 'session-1', key: 'sandboxEscape', value: 'yes', fence: 7 })
+    ).rejects.toThrow('no thread option named sandboxEscape')
+    await adapter.dispatch({
+      sessionId: 'session-1',
+      clientMessageId: 'client-1',
+      body: USER_MESSAGE,
+      fence: 7
+    })
+
+    expect(codex.connections[0].calls[1].params).toMatchObject({ model: 'gpt-5', effort: 'high' })
+    expect(codex.connections[0].calls[1].params).not.toHaveProperty('sandboxEscape')
+  })
+})
+
+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?.({
+      id: 11,
+      method: 'item/commandExecution/requestApproval',
+      params: { itemId: 'codex-item-1', threadId: THREAD_ID, turnId: 'turn-1' }
+    })
+  }
+
+  it('surfaces an approval request and answers it exactly once', async () => {
+    const codex = fakeCodex()
+    const events: CodexStructuredSessionEvent[] = []
+    const adapter = await acquired(codex, {}, events)
+
+    askApproval(codex)
+    adapter.bindPromptItemId('session-1', 'codex:thread-abc:turn-1:3', 'codex-item-1')
+    await adapter.answerPrompt({
+      sessionId: 'session-1',
+      itemId: 'codex:thread-abc:turn-1:3',
+      kind: 'approval',
+      optionId: 'accept',
+      fence: 7
+    })
+
+    expect(events.at(-1)).toMatchObject({ type: 'prompt', codexItemId: 'codex-item-1' })
+    expect(codex.connections[0].replies).toEqual([{ id: 11, result: { decision: 'accept' } }])
+
+    await expect(
+      adapter.answerPrompt({
+        sessionId: 'session-1',
+        itemId: 'codex:thread-abc:turn-1:3',
+        kind: 'approval',
+        optionId: 'decline',
+        fence: 7
+      })
+    ).rejects.toThrow('no longer waiting on')
+    expect(codex.connections[0].replies).toHaveLength(1)
+  })
+
+  it('rejects an option id that is not a Codex decision', async () => {
+    const codex = fakeCodex()
+    const adapter = await acquired(codex)
+
+    askApproval(codex)
+
+    await expect(
+      adapter.answerPrompt({
+        sessionId: 'session-1',
+        itemId: 'codex-item-1',
+        kind: 'approval',
+        optionId: 'yolo',
+        fence: 7
+      })
+    ).rejects.toThrow('is not a Codex approval decision')
+    expect(codex.connections[0].replies).toEqual([])
+  })
+
+  it('holds a multi-question request until every question is answered', async () => {
+    const codex = fakeCodex()
+    const adapter = await acquired(codex)
+    codex.connections[0].handlers.onServerRequest?.({
+      id: 12,
+      method: 'item/tool/requestUserInput',
+      params: {
+        itemId: 'codex-item-2',
+        threadId: THREAD_ID,
+        turnId: 'turn-1',
+        questions: [{ id: 'q1' }, { id: 'q2' }]
+      }
+    })
+
+    await adapter.answerPrompt({
+      sessionId: 'session-1',
+      itemId: 'codex-item-2',
+      kind: 'question',
+      optionId: encodeCodexQuestionOptionId('q1', 'yes'),
+      fence: 7
+    })
+    expect(codex.connections[0].replies).toEqual([])
+
+    await adapter.answerPrompt({
+      sessionId: 'session-1',
+      itemId: 'codex-item-2',
+      kind: 'question',
+      optionId: encodeCodexQuestionOptionId('q2', 'no'),
+      fence: 7
+    })
+
+    expect(codex.connections[0].replies).toEqual([
+      { id: 12, result: { answers: { q1: { answers: ['yes'] }, q2: { answers: ['no'] } } } }
+    ])
+  })
+
+  it('refuses a request it does not model instead of leaving the turn blocked', async () => {
+    const codex = fakeCodex()
+    const events: CodexStructuredSessionEvent[] = []
+    await acquired(codex, {}, events)
+
+    codex.connections[0].handlers.onServerRequest?.({
+      id: 13,
+      method: 'mcpServer/elicitation/request',
+      params: { itemId: 'codex-item-3', threadId: THREAD_ID }
+    })
+
+    expect(codex.connections[0].replies).toEqual([
+      { id: 13, code: -32601, message: 'Orca does not handle mcpServer/elicitation/request' }
+    ])
+    expect(events.some((event) => event.type === 'prompt')).toBe(false)
+  })
+
+  it('surfaces an answer to a prompt Codex already forgot', async () => {
+    const codex = fakeCodex()
+    const adapter = await acquired(codex)
+
+    await expect(
+      adapter.answerPrompt({
+        sessionId: 'session-1',
+        itemId: 'codex-item-gone',
+        kind: 'approval',
+        optionId: 'accept',
+        fence: 7
+      })
+    ).rejects.toThrow('no longer waiting on codex-item-gone')
+  })
+})
+
+describe('CodexStructuredSessionAdapter lifecycle', () => {
+  it('keeps sessions isolated and closes each child once', async () => {
+    const codex = fakeCodex()
+    const adapter = adapterFor(codex)
+    await adapter.acquire({ identity: identityFor('session-1'), fence: 1, spawnToken: 'spawn-a' })
+    await adapter.acquire({ identity: identityFor('session-2'), fence: 1, spawnToken: 'spawn-b' })
+
+    codex.connections[0].handlers.onServerRequest?.({
+      id: 21,
+      method: 'item/fileChange/requestApproval',
+      params: { itemId: 'codex-item-1', threadId: THREAD_ID, turnId: 'turn-1' }
+    })
+    await expect(
+      adapter.answerPrompt({
+        sessionId: 'session-2',
+        itemId: 'codex-item-1',
+        kind: 'approval',
+        optionId: 'accept',
+        fence: 1
+      })
+    ).rejects.toThrow('no longer waiting on')
+
+    await adapter.closeAll()
+    expect(codex.connections.map((connection) => connection.closeCount)).toEqual([1, 1])
+    await expect(
+      adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 1 })
+    ).rejects.toThrow('no live codex app-server for session session-1')
+  })
+
+  it('drops a session whose child died and reports it once', async () => {
+    const codex = fakeCodex()
+    const events: CodexStructuredSessionEvent[] = []
+    const adapter = await acquired(codex, {}, events)
+
+    codex.connections[0].handlers.onExit?.(new Error('codex app-server connection ended'))
+
+    expect(events.at(-1)).toEqual({
+      type: 'ended',
+      sessionId: 'session-1',
+      reason: 'codex app-server connection ended'
+    })
+    await expect(
+      adapter.dispatch({
+        sessionId: 'session-1',
+        clientMessageId: 'client-1',
+        body: USER_MESSAGE,
+        fence: 7
+      })
+    ).rejects.toThrow('no live codex app-server')
+    expect(await adapter.historyFilePath({ identity: identityFor('session-1') })).toBeNull()
+  })
+
+  it('ignores Codex traffic that arrives after the session is gone', async () => {
+    const codex = fakeCodex()
+    const adapter = await acquired(codex)
+    const connection = codex.connections[0]
+
+    await adapter.closeSession('session-1')
+    connection.handlers.onNotification?.('item/agentMessage/delta', { delta: 'x' })
+    connection.handlers.onServerRequest?.({
+      id: 31,
+      method: 'item/fileChange/requestApproval',
+      params: { itemId: 'codex-item-9', threadId: THREAD_ID }
+    })
+
+    expect(connection.replies).toEqual([])
+  })
+})
diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts
new file mode 100644
index 00000000000..5c462a9074e
--- /dev/null
+++ b/src/main/codex/codex-structured-session-adapter.ts
@@ -0,0 +1,333 @@
+import type {
+  AgentJournalMessageItem,
+  AgentSessionJournalIdentity
+} from '../../shared/agent-session-journal-types'
+import type {
+  AgentSessionAcquisition,
+  AgentSessionDispatchOutcome,
+  StructuredAgentSessionAdapter
+} from '../native-chat/agent-session-wire/structured-agent-session-adapter'
+import {
+  isCodexAppServerRequestError,
+  openCodexAppServerConnection,
+  type CodexAppServerConnection,
+  type CodexAppServerServerRequest
+} from './codex-app-server-connection'
+import { isCodexAppServerUnsupportedError } from './codex-app-server-session'
+import {
+  CODEX_SPAWN_TOKEN_ENV,
+  codexProcessIdentity,
+  codexProviderHandleLink
+} from './codex-structured-owner-identity'
+import {
+  answerCodexPrompt,
+  CodexPromptRegistry,
+  receiveCodexPromptRequest
+} from './codex-structured-prompt-replies'
+import {
+  readCodexThreadId,
+  readCodexThreadPath,
+  readCodexTurnId
+} from './codex-structured-thread-facts'
+import {
+  CODEX_USER_MESSAGE_ORDINAL,
+  isCodexTurnOptionKey,
+  startCodexTurn
+} from './codex-structured-turn-start'
+
+// The Codex half of the structured agent-session wire: one long-lived
+// `codex app-server` child per session, started or resumed under the lease the
+// host already reserved. Everything durable — journal, fence, idempotency —
+// belongs to the wire; this adapter only starts the process, names the turn the
+// provider accepted, and answers Codex's blocking prompt requests.
+
+export type CodexStructuredLaunch = {
+  command: string
+  args: string[]
+  /** Thread working directory, passed to Codex rather than to the child. */
+  cwd: string
+  /** Pinned account home. Null inherits whatever CODEX_HOME the host has. */
+  codexHome: string | null
+  /** Thread this session already proved. Null starts a new one. Taken from the
+   *  durable handle chain, never from the client-declared identity. */
+  resumeThreadId: string | null
+}
+
+export type CodexStructuredSessionEvent =
+  | { type: 'notification'; sessionId: string; threadId: string; method: string; params: unknown }
+  | {
+      type: 'prompt'
+      sessionId: string
+      threadId: string
+      method: string
+      params: unknown
+      codexItemId: string
+    }
+  /** The child died without the host asking it to. The lease is now stale. */
+  | { type: 'ended'; sessionId: string; reason: string }
+
+export type CodexStructuredSessionAdapterDeps = {
+  /** Command, args, cwd, and pinned home for one session. The wire's adapter
+   *  contract carries none of these, so the runtime resolves them. */
+  resolveLaunch: (input: {
+    identity: AgentSessionJournalIdentity
+  }) => Promise<CodexStructuredLaunch>
+  /** Every Codex notification and prompt, in arrival order. The translation
+   *  module subscribes here; without it the session still runs, unjournaled. */
+  onEvent?: (event: CodexStructuredSessionEvent) => void
+  openConnection?: typeof openCodexAppServerConnection
+  readProcessStartTime?: (pid: number) => Promise<number | null>
+  mintLinkId?: () => string
+  now?: () => number
+  requestTimeoutMs?: number
+}
+
+type CodexSession = {
+  connection: CodexAppServerConnection
+  threadId: string
+  historyPath: string | null
+  prompts: CodexPromptRegistry
+  /** Applied to the next `turn/start`; Codex has no thread-settings write. */
+  options: Map<string, string>
+  turnIdWaiters: ((turnId: string) => void)[]
+}
+
+export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdapter {
+  private readonly sessions = new Map<string, CodexSession>()
+
+  constructor(private readonly deps: CodexStructuredSessionAdapterDeps) {}
+
+  private get requestTimeoutMs(): number | undefined {
+    return this.deps.requestTimeoutMs
+  }
+
+  async acquire(input: {
+    identity: AgentSessionJournalIdentity
+    fence: number
+    spawnToken: string
+  }): Promise<AgentSessionAcquisition> {
+    // A re-acquire at a new fence must not leave the old child writing.
+    await this.closeSession(input.identity.sessionId)
+    const sessionId = input.identity.sessionId
+    const launch = await this.deps.resolveLaunch({ identity: input.identity })
+    const prompts = new CodexPromptRegistry()
+    const open = this.deps.openConnection ?? openCodexAppServerConnection
+    const connection = await open(
+      {
+        command: launch.command,
+        args: launch.args,
+        env: {
+          [CODEX_SPAWN_TOKEN_ENV]: input.spawnToken,
+          ...(launch.codexHome ? { CODEX_HOME: launch.codexHome } : {})
+        }
+      },
+      {
+        onNotification: (method, params) => this.handleNotification(sessionId, method, params),
+        onServerRequest: (request) => this.handleServerRequest(sessionId, request),
+        onExit: (error) => {
+          this.sessions.delete(sessionId)
+          prompts.clear()
+          this.deps.onEvent?.({ type: 'ended', sessionId, reason: error.message })
+        }
+      }
+    )
+
+    try {
+      const started = await connection.request(
+        launch.resumeThreadId ? 'thread/resume' : 'thread/start',
+        launch.resumeThreadId
+          ? { threadId: launch.resumeThreadId, cwd: launch.cwd }
+          : { cwd: launch.cwd },
+        { timeoutMs: this.requestTimeoutMs }
+      )
+      const threadId = readCodexThreadId(started)
+      if (!threadId) {
+        throw new Error('codex app-server did not name the thread it opened')
+      }
+      // A resume that lands on a different thread is a fork wearing a resume's
+      // name; the handle chain would record a lie.
+      if (launch.resumeThreadId && threadId !== launch.resumeThreadId) {
+        throw new Error(`codex app-server resumed ${threadId} instead of ${launch.resumeThreadId}`)
+      }
+      this.sessions.set(sessionId, {
+        connection,
+        threadId,
+        historyPath: readCodexThreadPath(started),
+        prompts,
+        options: new Map(),
+        turnIdWaiters: []
+      })
+      return {
+        process: await codexProcessIdentity(
+          { ...input, pid: connection.pid },
+          this.deps.readProcessStartTime
+        ),
+        link: codexProviderHandleLink({
+          threadId,
+          resumed: launch.resumeThreadId !== null,
+          fence: input.fence,
+          linkId: this.deps.mintLinkId?.(),
+          observedAt: this.deps.now?.() ?? Date.now()
+        })
+      }
+    } catch (error) {
+      await connection.close()
+      throw error
+    }
+  }
+
+  private handleNotification(sessionId: string, method: string, params: unknown): void {
+    const session = this.sessions.get(sessionId)
+    if (!session) {
+      return
+    }
+    if (method === 'turn/started') {
+      const turnId = readCodexTurnId(params)
+      const waiter = turnId ? session.turnIdWaiters.shift() : undefined
+      waiter?.(turnId as string)
+    }
+    this.deps.onEvent?.({
+      type: 'notification',
+      sessionId,
+      threadId: session.threadId,
+      method,
+      params
+    })
+  }
+
+  private handleServerRequest(sessionId: string, request: CodexAppServerServerRequest): void {
+    const session = this.sessions.get(sessionId)
+    if (!session) {
+      return
+    }
+    const prompt = receiveCodexPromptRequest(session.prompts, session.connection, request)
+    if (!prompt) {
+      return
+    }
+    this.deps.onEvent?.({
+      type: 'prompt',
+      sessionId,
+      threadId: session.threadId,
+      method: request.method,
+      params: request.params,
+      codexItemId: prompt.codexItemId
+    })
+  }
+
+  /** Lets the translation module address a live prompt by the journal item it
+   *  became; until then the Codex item id addresses it directly. */
+  bindPromptItemId(sessionId: string, journalItemId: string, codexItemId: string): void {
+    this.sessions.get(sessionId)?.prompts.bindJournalItemId(journalItemId, codexItemId)
+  }
+
+  async dispatch(input: {
+    sessionId: string
+    clientMessageId: string
+    body: AgentJournalMessageItem
+    fence: number
+  }): Promise<AgentSessionDispatchOutcome> {
+    const session = this.session(input.sessionId)
+    let turnId: string | null
+    try {
+      turnId = await startCodexTurn(session, { ...input, timeoutMs: this.requestTimeoutMs })
+    } catch (error) {
+      // Codex answering and declining is a rejection; a timeout or a dead child
+      // is not, and the wire must not tell the user their message failed.
+      if (isCodexAppServerRequestError(error) || isCodexAppServerUnsupportedError(error)) {
+        return { state: 'rejected', reason: (error as Error).message }
+      }
+      throw error
+    }
+    return turnId === null
+      ? {
+          state: 'unknown',
+          reason: 'codex app-server started a turn it did not name in time'
+        }
+      : {
+          state: 'accepted',
+          providerIdentity: {
+            provider: 'codex',
+            threadId: session.threadId,
+            turnId,
+            ordinal: CODEX_USER_MESSAGE_ORDINAL
+          }
+        }
+  }
+
+  async cancelTurn(input: {
+    sessionId: string
+    turnId: string
+    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.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
+    }
+  }
+
+  async answerPrompt(input: {
+    sessionId: string
+    itemId: string
+    kind: 'approval' | 'question'
+    optionId: string
+    fence: number
+  }): Promise<void> {
+    const session = this.session(input.sessionId)
+    answerCodexPrompt(session.prompts, session.connection, input.itemId, input.optionId)
+  }
+
+  async setOption(input: {
+    sessionId: string
+    key: string
+    value: string
+    fence: number
+  }): Promise<void> {
+    const session = this.session(input.sessionId)
+    if (!isCodexTurnOptionKey(input.key)) {
+      throw new Error(`codex app-server has no thread option named ${input.key}`)
+    }
+    // Codex applies model, effort, and policy on `turn/start`, so an option set
+    // between turns takes effect on the next one rather than immediately.
+    session.options.set(input.key, input.value)
+  }
+
+  async historyFilePath(input: { identity: AgentSessionJournalIdentity }): Promise<string | null> {
+    return this.sessions.get(input.identity.sessionId)?.historyPath ?? null
+  }
+
+  /** Reaps one session's child. The proven handle chain is already durable, so
+   *  a graceful close loses nothing. */
+  async closeSession(sessionId: string): Promise<void> {
+    const session = this.sessions.get(sessionId)
+    if (!session) {
+      return
+    }
+    this.sessions.delete(sessionId)
+    session.prompts.clear()
+    await session.connection.close()
+  }
+
+  async closeAll(): Promise<void> {
+    await Promise.all([...this.sessions.keys()].map((sessionId) => this.closeSession(sessionId)))
+  }
+
+  private session(sessionId: string): CodexSession {
+    const session = this.sessions.get(sessionId)
+    if (!session) {
+      throw new Error(`no live codex app-server for session ${sessionId}`)
+    }
+    return session
+  }
+}
diff --git a/src/main/codex/codex-structured-thread-facts.ts b/src/main/codex/codex-structured-thread-facts.ts
new file mode 100644
index 00000000000..349246ecf29
--- /dev/null
+++ b/src/main/codex/codex-structured-thread-facts.ts
@@ -0,0 +1,39 @@
+// The handful of facts Orca reads out of Codex app-server payloads. Codex has
+// moved these fields between the envelope and a nested `thread` / `turn` object
+// across releases, so each reader accepts both shapes rather than pinning one.
+
+function record(value: unknown): Record<string, unknown> | null {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+    ? (value as Record<string, unknown>)
+    : null
+}
+
+function nonEmptyString(value: unknown): string | null {
+  return typeof value === 'string' && value.length > 0 ? value : null
+}
+
+/** `thread/start`, `thread/resume`, and `thread/started` all name the thread. */
+export function readCodexThreadId(payload: unknown): string | null {
+  const root = record(payload)
+  if (!root) {
+    return null
+  }
+  return nonEmptyString(record(root.thread)?.id) ?? nonEmptyString(root.threadId)
+}
+
+/** Rollout file for the thread, when Codex reports one. Journal recovery reads
+ *  it; a null just falls back to the existing session-file resolver. */
+export function readCodexThreadPath(payload: unknown): string | null {
+  const root = record(payload)
+  return root ? nonEmptyString(record(root.thread)?.path) : null
+}
+
+/** `turn/start` responses carry `turn.id`; `turn/started` notifications carry
+ *  the same under `turn`, and older builds put `turnId` on the envelope. */
+export function readCodexTurnId(payload: unknown): string | null {
+  const root = record(payload)
+  if (!root) {
+    return null
+  }
+  return nonEmptyString(record(root.turn)?.id) ?? nonEmptyString(root.turnId)
+}
diff --git a/src/main/codex/codex-structured-turn-start.ts b/src/main/codex/codex-structured-turn-start.ts
new file mode 100644
index 00000000000..d2a37c4e40c
--- /dev/null
+++ b/src/main/codex/codex-structured-turn-start.ts
@@ -0,0 +1,91 @@
+import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types'
+import type { NativeChatBlock } from '../../shared/native-chat-types'
+import type { CodexAppServerConnection } from './codex-app-server-connection'
+import { readCodexTurnId } from './codex-structured-thread-facts'
+
+// Starting a Codex turn and learning its id, which are not the same event:
+// `turn/start` returns the id on newer builds and acks before it exists on
+// older ones, where it arrives as a `turn/started` notification instead.
+
+/** Codex records the user message first in a turn, so the submission Orca just
+ *  accepted is ordinal 0 of `(threadId, turnId)`. */
+export const CODEX_USER_MESSAGE_ORDINAL = 0
+
+/** Past this the turn is real but unnameable, which the journal renders as
+ *  delivery unconfirmed rather than failure. */
+const TURN_ID_WAIT_MS = 10_000
+
+/** Keys Codex accepts as per-turn overrides. An unlisted key would otherwise
+ *  become an arbitrary client-controlled `turn/start` parameter. */
+const CODEX_TURN_OPTION_KEYS = new Set([
+  'model',
+  'effort',
+  'approvalPolicy',
+  'approvalsReviewer',
+  'personality',
+  'serviceTier'
+])
+
+export function isCodexTurnOptionKey(key: string): boolean {
+  return CODEX_TURN_OPTION_KEYS.has(key)
+}
+
+/** The session state one turn needs. `turnIdWaiters` is shared with the
+ *  notification handler, which resolves the head of the queue — correct because
+ *  Codex runs one turn per thread, so starts and `turn/started` share an order. */
+export type CodexTurnHost = {
+  connection: Pick<CodexAppServerConnection, 'request'>
+  threadId: string
+  options: Map<string, string>
+  turnIdWaiters: ((turnId: string) => void)[]
+}
+
+function turnInputFor(body: AgentJournalMessageItem): Record<string, unknown>[] {
+  const input: Record<string, unknown>[] = []
+  for (const block of body.blocks as NativeChatBlock[]) {
+    if (block.type === 'text' && block.text.length > 0) {
+      input.push({ type: 'text', text: block.text })
+    } else if (block.type === 'image-ref' && block.path) {
+      input.push({ type: 'localImage', path: block.path })
+    } else if (block.type === 'image-ref' && block.url) {
+      input.push({ type: 'image', url: block.url })
+    }
+  }
+  return input
+}
+
+/**
+ * Resolves the turn id, or null when Codex owns a turn it never named. Throws
+ * only for outcomes the wire must not read as acceptance.
+ */
+export async function startCodexTurn(
+  host: CodexTurnHost,
+  input: { clientMessageId: string; body: AgentJournalMessageItem; timeoutMs?: number }
+): Promise<string | null> {
+  // Registered BEFORE the call: on builds that ack first, `turn/started` can
+  // land while the response is still in flight.
+  let notified: ((turnId: string) => void) | null = null
+  const fromNotification = new Promise<string | null>((resolve) => {
+    notified = resolve
+    host.turnIdWaiters.push(resolve)
+    setTimeout(() => resolve(null), TURN_ID_WAIT_MS).unref?.()
+  })
+  try {
+    const started = await host.connection.request(
+      'turn/start',
+      {
+        threadId: host.threadId,
+        clientUserMessageId: input.clientMessageId,
+        input: turnInputFor(input.body),
+        ...Object.fromEntries(host.options)
+      },
+      { timeoutMs: input.timeoutMs }
+    )
+    return readCodexTurnId(started) ?? (await fromNotification)
+  } finally {
+    const index = notified ? host.turnIdWaiters.indexOf(notified) : -1
+    if (index !== -1) {
+      host.turnIdWaiters.splice(index, 1)
+    }
+  }
+}