refactor(worker): move Codex turn logic into CodexAdapter behind AgentRuntime

2026-07-11 02:53:26 -04:00
parent c1e816741d
commit e5bba720b4
5 changed files with 382 additions and 268 deletions
@@ -0,0 +1,184 @@
+import { mkdir, readFile, writeFile } from 'node:fs/promises';
+import path from 'node:path';
+import { randomUUID } from 'node:crypto';
+
+import type { NormalizedAgentEvent } from '../agent-events';
+import { normalizeCodexJsonLine } from '../agent-events';
+import { prepareCodexWorkspaceFiles } from '../codex-runtime';
+import { env } from '../env';
+import type { AdapterFactory, TurnResult } from './agent-runtime';
+import {
+  buildMarkedCommand,
+  killBoxProcessesByMarker,
+  streamExecInContainer,
+} from './docker';
+import type { AdapterWorkspace } from './provider';
+import { codexModelArgs, providerEnvironment } from './provider';
+
+// Reused verbatim from the worker: normalize + pretty-print an auth JSON blob
+// before writing it with owner-only permissions.
+const writeJsonFile = async (filePath: string, content: string) => {
+  let normalized = content.trim();
+  try {
+    normalized = `${JSON.stringify(JSON.parse(normalized), null, 2)}\n`;
+  } catch {
+    throw new Error('Codex auth JSON is not valid JSON.');
+  }
+  await mkdir(path.dirname(filePath), { recursive: true });
+  await writeFile(filePath, normalized, { mode: 0o600 });
+};
+
+const isNoise = (line: string) => {
+  const trimmed = line.trim();
+  return (
+    !trimmed ||
+    trimmed === 'Reading additional input from stdin...' ||
+    trimmed.includes('`[features].codex_hooks` is deprecated')
+  );
+};
+
+export const createCodexAdapter: AdapterFactory = (deps) => {
+  const execStream = deps?.execStream ?? streamExecInContainer;
+
+  const prepareAuth = async (workspace: AdapterWorkspace) => {
+    const secret = workspace.claim.aiProviderProfile?.secret;
+    if (!secret) {
+      throw new Error('Codex auth profile is missing auth.json contents.');
+    }
+    await prepareCodexWorkspaceFiles({
+      workdir: workspace.workdir,
+      repoDir: workspace.repoDir,
+    });
+    const codexAuthPath = path.join(workspace.workdir, '.codex', 'auth.json');
+    await writeJsonFile(codexAuthPath, secret);
+
+    // Also seed OpenCode's auth location with the saved JSON for forward
+    // compatibility if this profile later runs through OpenCode directly.
+    const openCodeAuthPath = path.join(
+      workspace.workdir,
+      '.local',
+      'share',
+      'opencode',
+      'auth.json',
+    );
+    await writeJsonFile(openCodeAuthPath, secret);
+  };
+
+  const runTurn = async (
+    workspace: AdapterWorkspace,
+    prompt: string,
+    onEvent: (event: NormalizedAgentEvent) => Promise<void>,
+  ): Promise<TurnResult> => {
+    workspace.runtime = 'codex';
+    workspace.turnMarker = `spoon-turn-${randomUUID()}`;
+    const marker = workspace.turnMarker;
+
+    const outputFileName = `last-message-${workspace.claim.job._id}.txt`;
+    const outputFileHostPath = path.join(
+      workspace.workdir,
+      '.codex',
+      outputFileName,
+    );
+    const outputFileContainerPath = path.posix.join(
+      workspace.containerHome,
+      '.codex',
+      outputFileName,
+    );
+    const codexArgv = workspace.codexSessionId
+      ? [
+          'codex',
+          'exec',
+          'resume',
+          '--json',
+          ...codexModelArgs(workspace.claim),
+          '--dangerously-bypass-approvals-and-sandbox',
+          '--output-last-message',
+          outputFileContainerPath,
+          workspace.codexSessionId,
+          prompt,
+        ]
+      : [
+          'codex',
+          'exec',
+          '--json',
+          ...codexModelArgs(workspace.claim),
+          '--dangerously-bypass-approvals-and-sandbox',
+          '--output-last-message',
+          outputFileContainerPath,
+          '--cd',
+          workspace.containerRepo,
+          prompt,
+        ];
+    const command = buildMarkedCommand(marker, codexArgv);
+
+    const aiEnv = providerEnvironment(workspace.claim, workspace.containerHome);
+    const secretEnv = Object.fromEntries(
+      workspace.claim.secrets.map((secret) => [secret.name, secret.value]),
+    );
+
+    let capturedSessionId: string | undefined;
+    let capturedError: string | undefined;
+
+    // execa's own `timeout` only kills the local docker-exec client, not the
+    // in-box process group. Schedule an explicit marker kill on timeout and
+    // clear it once the exec resolves.
+    const killTimer = setTimeout(() => {
+      void killBoxProcessesByMarker({
+        containerName: workspace.boxName,
+        marker,
+      });
+    }, env.jobTimeoutMs);
+
+    let result;
+    try {
+      result = await execStream({
+        containerName: workspace.boxName,
+        containerCwd: workspace.containerRepo,
+        command,
+        environment: { ...aiEnv, ...secretEnv },
+        redact: workspace.redact,
+        timeoutMs: env.jobTimeoutMs,
+        onStdoutLine: async (line) => {
+          for (const event of normalizeCodexJsonLine(line)) {
+            if (event.kind === 'session') capturedSessionId = event.sessionId;
+            if (event.kind === 'error') capturedError = event.message;
+            await onEvent(event);
+          }
+        },
+        onStderrLine: async (line) => {
+          if (isNoise(line)) return;
+          await onEvent({ kind: 'status', status: line });
+        },
+      });
+    } finally {
+      clearTimeout(killTimer);
+    }
+
+    if (result.exitCode !== 0) {
+      capturedError = result.output;
+    }
+
+    let finalMessage: string | undefined;
+    try {
+      const lastMessage = await readFile(outputFileHostPath, 'utf8');
+      if (lastMessage.trim()) finalMessage = lastMessage.trim();
+    } catch (error) {
+      const code = error && typeof error === 'object' ? 'code' in error : false;
+      if (!code || (error as { code?: string }).code !== 'ENOENT') {
+        throw error;
+      }
+    }
+
+    return { finalMessage, sessionId: capturedSessionId, error: capturedError };
+  };
+
+  const abort = async (workspace: AdapterWorkspace) => {
+    if (!workspace.turnMarker) return;
+    await killBoxProcessesByMarker({
+      containerName: workspace.boxName,
+      marker: workspace.turnMarker,
+    });
+  };
+
+  return { name: 'codex', prepareAuth, runTurn, abort };
+};
@@ -0,0 +1,4 @@
+import { registerAdapter } from './agent-runtime';
+import { createCodexAdapter } from './codex-adapter';
+
+registerAdapter('codex', createCodexAdapter);
@@ -18,9 +18,8 @@ import type { NormalizedAgentEvent } from './agent-events';
 import type { OpenCodeSession } from './opencode-session';
 import type { AdapterWorkspace, Claim } from './runtime/provider';
 import type { BoxHandle } from './user-container';
-import { normalizeCodexJsonLine } from './agent-events';
-import { abortManagedCodexProcess, managedCodexCommand } from './codex-process';
-import { prepareCodexWorkspaceFiles } from './codex-runtime';
+import { getAdapter } from './runtime/agent-runtime';
+import './runtime/register';
 import { env } from './env';
 import {
  cloneRepository,
@@ -44,10 +43,8 @@ import {
  runExecInContainer,
  startWorkspaceContainer,
  stopWorkspaceContainer,
-  streamExecInContainer,
 } from './runtime/docker';
 import {
-  codexModelArgs,
  collectJsonStringValues,
  isCodexLoginProfile,
  opencodeModel,
@@ -66,15 +63,10 @@ type ActiveWorkspace = AdapterWorkspace & {
  containerId?: string;
  opencodePassword?: string;
  opencodeSession?: OpenCodeSession;
-  codexPidFile?: string;
  agentTurnActive?: boolean;
  cancelRequested?: boolean;
  resolveTurn?: () => void;
  lastRecordedDiffSignature?: string;
-  // Captures the most recent Codex `error`/`turn.failed` event for the active
-  // turn so the failure surfaces the real reason instead of a generic
-  // "no assistant response" message.
-  codexTurnError?: string;
 };

 type FileTreeNode = {
@@ -280,6 +272,22 @@ const setCodexSessionId = async (
    codexSessionId,
  });

+// Persist the runtime session id for continuity, keyed on the workspace runtime
+// (replaces the per-runtime setter call sites so dispatch stays generic).
+const persistRuntimeSession = async (
+  workspace: ActiveWorkspace,
+  sessionId: string,
+) => {
+  if (workspace.runtime === 'codex') {
+    workspace.codexSessionId = sessionId;
+    await setCodexSessionId(workspace.claim.job._id, sessionId);
+  } else if (workspace.runtime === 'opencode') {
+    workspace.opencodeSessionId = sessionId;
+  } else {
+    workspace.claudeSessionId = sessionId;
+  }
+};
+
 const createInteractionRequest = async (args: {
  jobId: Id<'agentJobs'>;
  runtime: 'opencode' | 'codex';
@@ -326,48 +334,6 @@ const commandToShell = (command: string) => ['bash', '-lc', command];
 const workspaceContainerName = (jobId: string) =>
  `spoon-agent-job-${jobId.replace(/[^a-zA-Z0-9_.-]/g, '-')}`;

-const writeJsonFile = async (filePath: string, content: string) => {
-  let normalized = content.trim();
-  try {
-    normalized = `${JSON.stringify(JSON.parse(normalized), null, 2)}\n`;
-  } catch {
-    throw new Error('Codex auth JSON is not valid JSON.');
-  }
-  await mkdir(path.dirname(filePath), { recursive: true });
-  await writeFile(filePath, normalized, { mode: 0o600 });
-};
-
-const prepareCodexAuth = async (workspace: ActiveWorkspace) => {
-  const secret = workspace.claim.aiProviderProfile?.secret;
-  if (!secret) {
-    throw new Error('Codex auth profile is missing auth.json contents.');
-  }
-  await prepareCodexWorkspaceFiles({
-    workdir: workspace.workdir,
-    repoDir: workspace.repoDir,
-  });
-  const codexAuthPath = path.join(workspace.workdir, '.codex', 'auth.json');
-  await writeJsonFile(codexAuthPath, secret);
-
-  // Also seed OpenCode's auth location with the saved JSON for forward
-  // compatibility if this profile later runs through OpenCode directly.
-  const openCodeAuthPath = path.join(
-    workspace.workdir,
-    '.local',
-    'share',
-    'opencode',
-    'auth.json',
-  );
-  await writeJsonFile(openCodeAuthPath, secret);
-
-  await appendEvent(
-    workspace.claim.job._id,
-    'info',
-    'clone',
-    'Prepared Codex auth JSON for the isolated workspace.',
-  );
-};
-
 const agentFailurePrefix = (claim: Claim) =>
  isCodexLoginProfile(claim) ? 'codex failed' : 'opencode failed';

@@ -412,9 +378,11 @@ const handleAgentEvent = async (args: {
    return;
  }
  if (event.kind === 'session') {
+    // Session persistence is centralized in persistRuntimeSession (called from
+    // the dispatch after the turn); here we only track it in memory so a later
+    // turn can resume.
    if (workspace.runtimeMode === 'codex_exec') {
      workspace.codexSessionId = event.sessionId;
-      await setCodexSessionId(jobId, event.sessionId);
    }
    return;
  }
@@ -491,10 +459,9 @@ const handleAgentEvent = async (args: {
    return;
  }
  // event.kind === 'error'
-  // Record the real Codex failure reason on the workspace so the turn can
-  // surface it (Codex can emit `error`/`turn.failed` events and still exit 0
-  // in some versions, which otherwise looks like an empty response).
-  workspace.codexTurnError = event.message;
+  // The adapter captures the real Codex failure reason (Codex can emit
+  // `error`/`turn.failed` events and still exit 0) and surfaces it as
+  // TurnResult.error; here we just log it.
  await appendEvent(jobId, 'error', 'plan', truncate(event.message, 20_000));
 };

@@ -584,177 +551,6 @@ const workspaceCurrentContent = new Map<
  }
 >();

-// Reading through a function boundary prevents TypeScript from narrowing the
-// field to `undefined` after the synchronous reset in `runCodexTurn`; it is set
-// asynchronously by the stream event handler.
-const readCodexTurnError = (workspace: ActiveWorkspace) =>
-  workspace.codexTurnError;
-
-const runCodexTurn = async (args: {
-  workspace: ActiveWorkspace;
-  prompt: string;
-  assistantMessageId: Id<'agentJobMessages'>;
-  assistantContent: { value: string };
-}) => {
-  const { workspace, prompt, assistantMessageId, assistantContent } = args;
-  workspace.runtimeMode = 'codex_exec';
-  workspace.codexTurnError = undefined;
-  await setRuntimeSession({
-    jobId: workspace.claim.job._id,
-    agentRuntimeMode: 'codex_exec',
-    codexSessionId: workspace.codexSessionId,
-  });
-  const outputFileName = `last-message-${workspace.claim.job._id}.txt`;
-  const outputFileHostPath = path.join(
-    workspace.workdir,
-    '.codex',
-    outputFileName,
-  );
-  const outputFileContainerPath = path.posix.join(
-    workspace.containerHome,
-    '.codex',
-    outputFileName,
-  );
-  if (workspace.cancelRequested) {
-    throw new Error('Workspace cancellation requested.');
-  }
-  const codexCommand = workspace.codexSessionId
-    ? [
-        'codex',
-        'exec',
-        'resume',
-        '--json',
-        ...codexModelArgs(workspace.claim),
-        '--dangerously-bypass-approvals-and-sandbox',
-        '--output-last-message',
-        outputFileContainerPath,
-        workspace.codexSessionId,
-        prompt,
-      ]
-    : [
-        'codex',
-        'exec',
-        '--json',
-        ...codexModelArgs(workspace.claim),
-        '--dangerously-bypass-approvals-and-sandbox',
-        '--output-last-message',
-        outputFileContainerPath,
-        '--cd',
-        workspace.containerRepo,
-        prompt,
-      ];
-  const pidFileName = `codex-turn-${workspace.claim.job._id}.pid`;
-  const pidFileHostPath = path.join(workspace.workdir, '.codex', pidFileName);
-  const pidFileContainerPath = path.posix.join(
-    workspace.containerHome,
-    '.codex',
-    pidFileName,
-  );
-  workspace.codexPidFile = pidFileContainerPath;
-  const command = managedCodexCommand({
-    jobId: workspace.claim.job._id,
-    pidFile: pidFileContainerPath,
-    command: codexCommand,
-  });
-  const aiEnv = providerEnvironment(workspace.claim, workspace.containerHome);
-  const secretEnv = Object.fromEntries(
-    workspace.claim.secrets.map((secret) => [secret.name, secret.value]),
-  );
-  let result;
-  try {
-    result = await streamExecInContainer({
-      containerName: workspace.boxName,
-      containerCwd: workspace.containerRepo,
-      command,
-      environment: {
-        ...aiEnv,
-        ...secretEnv,
-      },
-      redact: workspace.redact,
-      timeoutMs: env.jobTimeoutMs,
-      onStdoutLine: async (line) => {
-        for (const event of normalizeCodexJsonLine(line)) {
-          await handleAgentEvent({
-            workspace,
-            event,
-            assistantMessageId,
-            assistantContent,
-          });
-        }
-      },
-      onStderrLine: async (line) => {
-        const trimmed = line.trim();
-        if (
-          trimmed &&
-          trimmed !== 'Reading additional input from stdin...' &&
-          !trimmed.includes('`[features].codex_hooks` is deprecated')
-        ) {
-          await appendEvent(
-            workspace.claim.job._id,
-            'info',
-            'plan',
-            truncate(trimmed, 10_000),
-          );
-        }
-      },
-    });
-  } finally {
-    workspace.codexPidFile = undefined;
-    await rm(pidFileHostPath, { force: true });
-  }
-  await appendEvent(
-    workspace.claim.job._id,
-    'info',
-    'plan',
-    `Codex CLI exited with code ${result.exitCode}; captured output length ${result.output.length}; assistant length ${assistantContent.value.length}.`,
-  );
-  if (result.exitCode !== 0) {
-    throw new Error(`codex failed:\n${result.output}`);
-  }
-  if (!assistantContent.value.trim()) {
-    try {
-      const lastMessage = await readFile(outputFileHostPath, 'utf8');
-      if (lastMessage.trim()) {
-        assistantContent.value = truncate(
-          workspace.redact(lastMessage.trim()),
-          40_000,
-        );
-        await updateMessage({
-          messageId: assistantMessageId,
-          content: assistantContent.value,
-          status: 'streaming',
-        });
-        await appendEvent(
-          workspace.claim.job._id,
-          'info',
-          'plan',
-          `Recovered assistant response from Codex output file ${outputFileName}.`,
-        );
-      }
-    } catch (error) {
-      const code = error && typeof error === 'object' ? 'code' in error : false;
-      if (!code || (error as { code?: string }).code !== 'ENOENT') {
-        throw error;
-      }
-      await appendEvent(
-        workspace.claim.job._id,
-        'warn',
-        'plan',
-        `Codex output file ${outputFileName} was not created.`,
-      );
-    }
-  }
-  // Codex can report a failure via a JSON `error`/`turn.failed` event while
-  // still exiting 0. If the turn produced no assistant text but did report an
-  // error, surface that real reason rather than a generic empty response.
-  // Read through a helper so it is not narrowed away by the reset above (the
-  // field is mutated asynchronously inside the stream handler).
-  const codexTurnError = readCodexTurnError(workspace);
-  if (!assistantContent.value.trim() && codexTurnError) {
-    throw new Error(`codex failed:\n${codexTurnError}`);
-  }
-};
-
 const runOpenCodeTurn = async (args: {
  workspace: ActiveWorkspace;
  prompt: string;
@@ -1279,8 +1075,9 @@ const runClaim = async (claim: Claim) => {
      boxHandle,
      githubToken,
      redact,
-      // Placeholder until Task 3 resolves the runtime from the claim/profile.
-      runtime: 'codex',
+      // job.runtime becomes authoritative in Task 8; until then resolve from
+      // the profile so dispatch stays byte-identical to today.
+      runtime: isCodexLoginProfile(claim) ? 'codex' : 'opencode',
    };
    if (userEnv) {
      await appendEvent(
@@ -1298,7 +1095,13 @@ const runClaim = async (claim: Claim) => {
      });
    }
    if (isCodexLoginProfile(claim)) {
-      await prepareCodexAuth(workspace);
+      await getAdapter(workspace.runtime).prepareAuth(workspace);
+      await appendEvent(
+        jobId,
+        'info',
+        'clone',
+        'Prepared Codex auth JSON for the isolated workspace.',
+      );
    }
    await materializeEnvFile(workspace);
    const detected = await detectPackageCommands(repoDir);
@@ -1505,20 +1308,8 @@ export const getWorkspaceAgentStatus = (jobId: string) => {
 const abortWorkspaceAgentForWorkspace = async (workspace: ActiveWorkspace) => {
  if (workspace.opencodeSession) {
    await abortOpenCodeSession(workspace.opencodeSession);
-  } else if (workspace.runtimeMode === 'codex_exec' && workspace.codexPidFile) {
-    await abortManagedCodexProcess({
-      jobId: workspace.claim.job._id,
-      pidFile: workspace.codexPidFile,
-      execute: async (command) =>
-        await runExecInContainer({
-          containerName: workspace.boxName,
-          containerCwd: workspace.containerRepo,
-          command,
-          environment: {},
-          redact: workspace.redact,
-          timeoutMs: env.jobTimeoutMs,
-        }),
-    });
+  } else if (workspace.runtime === 'codex' && workspace.turnMarker) {
+    await getAdapter(workspace.runtime).abort(workspace);
  }
  workspace.agentTurnActive = false;
  workspace.resolveTurn?.();
@@ -1607,18 +1398,58 @@ export const sendWorkspaceMessage = async (
    });
    const assistantContent = { value: '' };
    if (isCodexLoginProfile(claim)) {
+      const messageId = assistantMessageId;
      await appendEvent(
        claim.job._id,
        'info',
        'plan',
        'Starting Codex CLI turn with the configured login profile.',
      );
-      await runCodexTurn({
+      workspace.runtimeMode = 'codex_exec';
+      await setRuntimeSession({
+        jobId: claim.job._id,
+        agentRuntimeMode: 'codex_exec',
+        codexSessionId: workspace.codexSessionId,
+      });
+      if (workspace.cancelRequested) {
+        throw new Error('Workspace cancellation requested.');
+      }
+      const onEvent = (event: NormalizedAgentEvent) =>
+        handleAgentEvent({
          workspace,
-        prompt,
-        assistantMessageId,
+          event,
+          assistantMessageId: messageId,
          assistantContent,
        });
+      const adapter = getAdapter(workspace.runtime);
+      const result = await adapter.runTurn(workspace, prompt, onEvent);
+      if (result.sessionId) {
+        await persistRuntimeSession(workspace, result.sessionId);
+      }
+      if (!assistantContent.value.trim() && result.finalMessage) {
+        assistantContent.value = truncate(redact(result.finalMessage), 40_000);
+        await updateMessage({
+          messageId,
+          content: assistantContent.value,
+          status: 'streaming',
+        });
+      }
+      if (!assistantContent.value.trim()) {
+        console.error(
+          `Codex completed without producing an assistant response for job ${claim.job._id}.`,
+        );
+        throw new Error(
+          result.error
+            ? `${workspace.runtime} failed:\n${result.error}`
+            : 'Codex completed without producing an assistant response.',
+        );
+      }
+      await updateMessage({
+        messageId,
+        status: 'completed',
+        content: assistantContent.value,
+      });
+      workspace.agentTurnActive = false;
      console.log(
        `Codex turn completed for job ${claim.job._id}; response length=${assistantContent.value.length}`,
      );
@@ -1665,24 +1496,6 @@ export const sendWorkspaceMessage = async (
        throw new Error(`${agentFailurePrefix(claim)}:\n${result.output}`);
      }
    }
-    if (isCodexLoginProfile(claim)) {
-      if (!assistantContent.value.trim()) {
-        console.error(
-          `Codex completed without producing an assistant response for job ${claim.job._id}.`,
-        );
-        throw new Error(
-          workspace.codexTurnError
-            ? `Codex failed: ${workspace.codexTurnError}`
-            : 'Codex completed without producing an assistant response.',
-        );
-      }
-      await updateMessage({
-        messageId: assistantMessageId,
-        status: 'completed',
-        content: assistantContent.value,
-      });
-      workspace.agentTurnActive = false;
-    }
    workspace.agentTurnActive = false;
    if (claim.job.jobType === 'maintenance_review') {
      const decision = parseMaintenanceDecision(assistantContent.value);
@@ -0,0 +1,111 @@
+import { describe, expect, test, vi } from 'vitest';
+
+import type { NormalizedAgentEvent } from '../../src/agent-events';
+import type { ExecStreamFn } from '../../src/runtime/agent-runtime';
+import type { AdapterWorkspace, Claim } from '../../src/runtime/provider';
+
+const loadAdapter = async () => {
+  process.env.SPOON_WORKER_TOKEN = 'test-worker-token';
+  process.env.GITHUB_APP_ID = '123';
+  process.env.GITHUB_APP_PRIVATE_KEY =
+    '-----BEGIN PRIVATE KEY-----\\ntest\\n-----END PRIVATE KEY-----';
+  return await import('../../src/runtime/codex-adapter');
+};
+
+const claim = {
+  job: {
+    _id: 'job-1',
+    prompt: 'hi',
+    baseBranch: 'main',
+    workBranch: 'spoon/x',
+    forkOwner: 'o',
+    forkRepo: 'r',
+    upstreamOwner: 'u',
+    upstreamRepo: 'r',
+  },
+  spoon: { name: 'Spoon' },
+  openai: { model: 'gpt-5', reasoningEffort: 'medium' },
+  aiProviderProfile: {
+    id: 'p',
+    name: 'Codex',
+    provider: 'opencode_openai_login',
+    authType: 'opencode_auth_json',
+    model: 'gpt-5',
+    reasoningEffort: 'medium',
+  },
+  github: {},
+  secrets: [],
+} as unknown as Claim;
+
+const makeWorkspace = (): AdapterWorkspace => ({
+  claim,
+  // A non-existent workdir so the --output-last-message read is a clean ENOENT.
+  workdir: '/tmp/spoon-codex-adapter-test-missing',
+  homeDir: '/tmp/spoon-codex-adapter-test-missing',
+  username: 'tester',
+  containerHome: '/home/tester',
+  containerRepo: '/home/tester/Code/spoon',
+  repoDir: '/tmp/spoon-codex-adapter-test-missing/Code/spoon',
+  boxName: 'spoon-box-tester',
+  redact: (value: string) => value,
+  runtime: 'codex',
+});
+
+describe('CodexAdapter', () => {
+  test('streams normalized events and marks the turn via buildMarkedCommand', async () => {
+    const { createCodexAdapter } = await loadAdapter();
+    const captured: unknown[] = [];
+    const execStream = vi.fn(async (args) => {
+      captured.push(args.command);
+      await args.onStdoutLine?.(
+        JSON.stringify({
+          type: 'item.completed',
+          item: { id: 'item-1', type: 'agent_message', text: 'done' },
+        }),
+      );
+      await args.onStdoutLine?.(JSON.stringify({ type: 'turn.completed' }));
+      return { exitCode: 0, output: '' };
+    }) as unknown as ExecStreamFn;
+
+    const adapter = createCodexAdapter({ execStream });
+    const events: NormalizedAgentEvent[] = [];
+    const onEvent = (event: NormalizedAgentEvent): Promise<void> => {
+      events.push(event);
+      return Promise.resolve();
+    };
+
+    const result = await adapter.runTurn(makeWorkspace(), 'do it', onEvent);
+
+    expect(events).toContainEqual({
+      kind: 'assistant_delta',
+      content: 'done\n\n',
+      externalMessageId: 'item-1',
+    });
+    expect(events).toContainEqual({ kind: 'assistant_completed' });
+    expect(result).toBeTypeOf('object');
+    // The turn ran through buildMarkedCommand (setsid bash session).
+    const command = captured[0] as string[];
+    expect(command[0]).toBe('setsid');
+    expect(command[1]).toBe('bash');
+  });
+
+  test('surfaces a turn.failed error while exiting zero', async () => {
+    const { createCodexAdapter } = await loadAdapter();
+    const execStream = vi.fn(async (args) => {
+      await args.onStdoutLine?.(
+        JSON.stringify({
+          type: 'turn.failed',
+          error: { message: 'boom' },
+        }),
+      );
+      return { exitCode: 0, output: '' };
+    }) as unknown as ExecStreamFn;
+
+    const adapter = createCodexAdapter({ execStream });
+    const result = await adapter.runTurn(makeWorkspace(), 'do it', () =>
+      Promise.resolve(),
+    );
+
+    expect(result.error).toContain('boom');
+  });
+});
@@ -49,6 +49,8 @@ vi.mock('../../src/opencode-session', () => ({
  replyOpenCodePermission: vi.fn(),
 }));
 vi.mock('../../src/runtime/docker', () => ({
+  buildMarkedCommand: vi.fn(() => ['setsid', 'bash', '-lc', '# marker']),
+  killBoxProcessesByMarker: vi.fn(() => Promise.resolve()),
  listWorkspaceContainerNames: vi.fn(() => Promise.resolve([])),
  runExecInContainer: vi.fn(),
  startWorkspaceContainer: vi.fn(),