fix(codex): commit active input ownership atomically

Persist accepted input before publishing successor turn ownership across the queue, event bus, operation tracker, and Codex runtime. Transfer abortability at the same synchronous boundary and cover failure, restart, and settlement races deterministically.

Author: yongkangc

server-agents/codex/src/agents/codex/__tests__/execution.test.js

@@ -3,6 +3,8 @@ import { AssistantMessage, UserMessage } from '@garcon/common/chat-types';
 import { renderTranscriptSeed } from '@garcon/common/transcript-seed';
 import { AgentEventEmitterRuntime } from '@garcon/server-agent-common/shared/event-emitter-runtime';
 import { createPathNativeSessionCodec } from '@garcon/server-agent-common/native-session/path-native-session';
+import { AgentEventBus } from '../../../../../../server/agents/event-bus.ts';
+import { QueueExecutionAttempt } from '../../../../../../server/chat-execution/execution-attempt.ts';
 import { CodexExecution } from '../execution.ts';
 
 function createRuntime() {
@@ -67,7 +69,12 @@ function startRequest(overrides = {}) {
   };
 }
 
-function activeInputRequest(operation, beforeDelivery = async () => undefined) {
+async function commitHandoff(handoff) {
+  handoff.validate();
+  handoff.commit();
+}
+
+function activeInputRequest(operation, beforeDelivery = commitHandoff) {
   return startRequest({
     agentSessionId: 'thread-1',
     nativeSession: {
@@ -236,7 +243,10 @@ describe('CodexExecution', () => {
     execution.subscribe((event) => events.push(event));
     await execution.start(startRequest());
     runtime.submitActiveInput.mockImplementation(async (_request, beforeDelivery) => {
-      await beforeDelivery();
+      await beforeDelivery({
+        validate: () => undefined,
+        commit: () => undefined,
+      });
       throw new Error('delivery outcome unknown');
     });
 
@@ -252,4 +262,101 @@ describe('CodexExecution', () => {
       operation: successor,
     }));
   });
+
+  it('commits retained, routed, tracked, and runtime ownership at one persistence boundary', async () => {
+    const runtime = createRuntime();
+    const execution = new CodexExecution(
+      createHost(),
+      runtime,
+      createPathNativeSessionCodec('codex'),
+      createConfig(),
+    );
+    const bus = new AgentEventBus({ list: () => [{ execution }] });
+    const predecessor = startRequest().operation;
+    const rejected = { ...predecessor, clientRequestId: 'request-rejected', turnId: 'turn-rejected' };
+    const successor = { ...predecessor, clientRequestId: 'request-b', turnId: 'turn-b' };
+    const attempt = new QueueExecutionAttempt(predecessor);
+    const successorAbortable = mock(() => undefined);
+    let runtimeMetadata = {
+      clientRequestId: predecessor.clientRequestId,
+      turnId: predecessor.turnId,
+    };
+    const trackedMessages = [];
+    const routedMessages = [];
+    execution.subscribe((event) => {
+      if (event.type === 'messages') {
+        trackedMessages.push({ content: event.messages[0].content, turnId: event.operation.turnId });
+      }
+    });
+    bus.onMessages((_chatId, messages, metadata) => {
+      routedMessages.push({ content: messages[0].content, turnId: metadata.turnId });
+    });
+    const emitOutput = (content) => runtime.emitMessages(
+      'chat-1',
+      [new AssistantMessage('2026-07-24T00:00:00.000Z', content)],
+      runtimeMetadata,
+    );
+    const assertOwner = (turnId, content) => {
+      expect(runtimeMetadata.turnId).toBe(turnId);
+      expect(attempt.identity().turnId).toBe(turnId);
+      expect(bus.getActiveTurn('chat-1').turnId).toBe(turnId);
+      expect(trackedMessages.at(-1)).toEqual({ content, turnId });
+      expect(routedMessages.at(-1)).toEqual({ content, turnId });
+    };
+    const composeHandoff = (operationHandoff, next) => attempt.handoffTurn(
+      predecessor,
+      next,
+      bus.handoffTurn('chat-1', predecessor, next, operationHandoff),
+    );
+
+    bus.trackTurn('chat-1', predecessor);
+    bus.markTurnAbortable('chat-1', predecessor);
+    attempt.markAbortable();
+    await execution.start(startRequest());
+    runtime.submitActiveInput.mockImplementation(async (request, beforeDelivery) => {
+      await beforeDelivery({
+        validate: () => undefined,
+        commit: () => {
+          runtimeMetadata = {
+            clientRequestId: request.clientRequestId,
+            turnId: request.turnId,
+          };
+          request.onAbortable?.();
+        },
+      });
+      return true;
+    });
+
+    await expect(execution.submitActiveInput(activeInputRequest(rejected, async (operationHandoff) => {
+      const handoff = composeHandoff(operationHandoff, rejected);
+      handoff.validate();
+      await Promise.resolve();
+      emitOutput('A while persistence fails');
+      assertOwner('turn-1', 'A while persistence fails');
+      throw new Error('persistence failed');
+    }))).rejects.toThrow('persistence failed');
+    emitOutput('A after persistence failed');
+    assertOwner('turn-1', 'A after persistence failed');
+
+    await expect(execution.submitActiveInput({
+      ...activeInputRequest(successor, async (operationHandoff) => {
+        const handoff = composeHandoff(operationHandoff, successor);
+        handoff.validate();
+        await Promise.resolve();
+        emitOutput('A before persistence commits');
+        assertOwner('turn-1', 'A before persistence commits');
+        handoff.validate();
+        handoff.commit();
+        emitOutput('B after the atomic commit');
+        assertOwner('turn-b', 'B after the atomic commit');
+      }),
+      admission: {
+        signal: new AbortController().signal,
+        markStarted: mock(() => undefined),
+        markAbortable: successorAbortable,
+      },
+    })).resolves.toBe(true);
+    expect(successorAbortable).toHaveBeenCalledTimes(1);
+    await expect(bus.waitUntilTurnAbortable('chat-1', successor)).resolves.toBe(true);
+  });
 });

server-agents/codex/src/agents/codex/app-server/__tests__/app-server.test.js

@@ -3780,6 +3780,87 @@ describe('CodexAppServerRuntime', () => {
     expect(fake.shutdown).not.toHaveBeenCalled();
   });
 
+  it('transfers an already-abortable active session to the successor callback at commit', async () => {
+    let fake;
+    fake = new FakeClient({
+      getThreadGoal: async () => ({ goal: null }),
+      setThreadGoal: async (threadId, params) => {
+        queueMicrotask(() => fake.emit('notification', {
+          method: 'turn/started',
+          params: { threadId, turn: makeTurn({ id: 'goal-turn', status: 'inProgress' }) },
+        }));
+        return { goal: makeGoal(threadId, params.objective) };
+      },
+    });
+    const provider = new CodexAppServerRuntime({ createClient: () => fake });
+    const predecessorAbortable = mock(() => undefined);
+    const successorAbortable = mock(() => undefined);
+    await provider.runTurn(makeRequest({
+      agentSessionId: 'thread-1',
+      codexGoalCommand: { kind: 'set', objective: 'Long-running work' },
+      nativePath: null,
+      onAbortable: predecessorAbortable,
+    }));
+    expect(predecessorAbortable).toHaveBeenCalledTimes(1);
+
+    await provider.submitActiveInput(makeRequest({
+      agentSessionId: 'thread-1',
+      command: 'Take ownership now',
+      nativePath: null,
+      onAbortable: successorAbortable,
+    }), async (handoff) => {
+      expect(successorAbortable).not.toHaveBeenCalled();
+      handoff.validate();
+      handoff.commit();
+      expect(successorAbortable).toHaveBeenCalledTimes(1);
+    });
+  });
+
+  it('targets a successor when the provider becomes abortable after handoff commit', async () => {
+    const turnStarted = createDeferred();
+    const fake = new FakeClient({
+      getThreadGoal: async () => ({
+        goal: makeGoal('thread-1', 'Restored work'),
+      }),
+      steerTurn: async () => ({ turnId: 'continuation-turn' }),
+    });
+    const provider = new CodexAppServerRuntime({ createClient: () => fake });
+    const predecessorAbortable = mock(() => undefined);
+    const successorAbortable = mock(() => undefined);
+    await provider.runTurn(makeRequest({
+      agentSessionId: 'thread-1',
+      command: '/goal status',
+      codexGoalCommand: { kind: 'status' },
+      nativePath: null,
+      onAbortable: predecessorAbortable,
+    }));
+    expect(predecessorAbortable).not.toHaveBeenCalled();
+
+    const delivery = provider.submitActiveInput(makeRequest({
+      agentSessionId: 'thread-1',
+      command: 'Continue after restoration',
+      nativePath: null,
+      onAbortable: successorAbortable,
+    }), async (handoff) => {
+      handoff.validate();
+      handoff.commit();
+      turnStarted.resolve();
+    });
+    await turnStarted.promise;
+    expect(successorAbortable).not.toHaveBeenCalled();
+    fake.emit('notification', {
+      method: 'turn/started',
+      params: {
+        threadId: 'thread-1',
+        turn: makeTurn({ id: 'continuation-turn', status: 'inProgress' }),
+      },
+    });
+
+    await expect(delivery).resolves.toBe(true);
+    expect(predecessorAbortable).not.toHaveBeenCalled();
+    expect(successorAbortable).toHaveBeenCalledTimes(1);
+  });
+
   it('reconciles active input through the delivered payload and native history loader', async () => {
     const content = 'Preserve active input & literal markup <exactly>';
     const nativePath = path.join(tmpDir, 'active-input.jsonl');
@@ -3819,12 +3900,14 @@ describe('CodexAppServerRuntime', () => {
       command: content,
       clientMessageId: 'message-steer',
       nativePath,
-    }), async () => {
+    }), async (handoff) => {
       await pendingInputs.register('chat-1', content, {
         clientRequestId: 'request-steer',
         clientMessageId: 'message-steer',
         createdAt: '2026-06-01T00:00:00.000Z',
       });
+      handoff.validate();
+      handoff.commit();
     })).resolves.toBe(true);
 
     expect(pendingInputs.listForChat('chat-1')).toHaveLength(1);
@@ -4364,7 +4447,11 @@ describe('CodexAppServerRuntime', () => {
         agentSessionId: 'thread-1',
         command: `Deliver after ${turnKind}`,
         nativePath: null,
-      }), async () => { accepted = true; });
+      }), async (handoff) => {
+        accepted = true;
+        handoff.validate();
+        handoff.commit();
+      });
       await nonSteerableRejected;
       expect(accepted).toBe(true);
       fake.emit('notification', {
@@ -4465,7 +4552,7 @@ describe('CodexAppServerRuntime', () => {
     expect(fake.startTurn).not.toHaveBeenCalled();
   });
 
-  it('adopts active-input metadata before flushing a terminal finish pending at delivery', async () => {
+  it('keeps predecessor metadata when persistence fails with a terminal finish pending', async () => {
     let fake;
     fake = new FakeClient({
       getThreadGoal: async () => ({ goal: null }),
@@ -4505,14 +4592,15 @@ describe('CodexAppServerRuntime', () => {
           error: { message: 'terminal failure', codexErrorInfo: null, additionalDetails: null },
         },
       });
+      throw new Error('registration failed');
     });
-    await expect(first).rejects.toThrow('terminal failure');
+    await expect(first).rejects.toThrow('registration failed');
     await expect(failed).resolves.toEqual({
       chatId: 'chat-1',
       message: 'terminal failure',
       metadata: {
-        clientRequestId: 'request-b',
-        turnId: 'turn-b',
+        clientRequestId: 'request-a',
+        turnId: 'turn-a',
       },
     });
     let accepted = false;

server-agents/codex/src/agents/codex/app-server/runtime.ts

@@ -2,7 +2,7 @@ import { AssistantMessage, ErrorMessage, PermissionCancelledMessage, PermissionR
 import { promises as fs } from 'fs';
 import { AgentEventEmitterRuntime } from '@garcon/server-agent-common/shared/event-emitter-runtime';
 import type { RuntimeEventMetadata } from '@garcon/server-agent-common/shared/event-emitter-runtime';
-import type { AgentLogger } from '@garcon/server-agent-interface';
+import type { AgentActiveInputHandoff, AgentLogger } from '@garcon/server-agent-interface';
 import { CodexHistoryService } from '../history-source.js';
 import {
   assertCodexExecutionOpen,
@@ -374,7 +374,10 @@ export class CodexAppServerRuntime extends AgentEventEmitterRuntime {
 
   async submitActiveInput(
     request: CodexResumeRequest,
-    beforeDelivery: () => Promise<void> = async () => {},
+    beforeDelivery: (handoff: AgentActiveInputHandoff) => Promise<void> = async (handoff) => {
+      handoff.validate();
+      handoff.commit();
+    },
   ): Promise<boolean> {
     const session = this.#sessions.get(request.agentSessionId);
     if (!session || session.status === 'completed' || session.status === 'failed' || session.status === 'aborted') {
@@ -392,8 +395,27 @@ export class CodexAppServerRuntime extends AgentEventEmitterRuntime {
       ) return false;
       session.activeDeliveryReservations += 1;
       try {
-        await beforeDelivery();
-        session.eventMetadata = codexEventMetadata(request);
+        const validate = () => {
+          if (
+            this.#sessions.get(request.agentSessionId) !== session
+            || hasTerminalPendingFinish(session)
+            || isTerminalSessionStatus(session.status)
+          ) {
+            throw new Error(session.pendingFinish?.failedMessage ?? 'Codex session ended before active input delivery');
+          }
+        };
+        let committed = false;
+        validate();
+        await beforeDelivery({
+          validate,
+          commit: () => {
+            session.eventMetadata = codexEventMetadata(request);
+            session.onAbortable = request.onAbortable;
+            committed = true;
+            if (session.abortableNotified) session.onAbortable?.();
+          },
+        });
+        if (!committed) throw new Error('Codex active input handoff was not committed');
         if (hasTerminalPendingFinish(session) || isTerminalSessionStatus(session.status)) {
           throw new Error(session.pendingFinish?.failedMessage ?? 'Codex session ended before active input delivery');
         }
@@ -457,6 +479,7 @@ export class CodexAppServerRuntime extends AgentEventEmitterRuntime {
           const turn = await session.client.startTurn(startParams);
           if (!this.#canApplyTurnAttempt(session, turnAttemptGeneration)) return;
           session.activeTurnId = turn.turn.id;
+          this.#notifyAbortable(session);
           return;
         } catch (error) {
           const isTurnTransition = isActiveTurnConflictError(error) || isActiveTurnNotSteerableError(error);

server-agents/codex/src/agents/codex/execution.ts

@@ -106,12 +106,12 @@ export class CodexExecution implements AgentExecution {
     );
     return this.runtime.submitActiveInput(
       runtimeRequest,
-      () => this.#operations.handoff(
+      (handoff) => request.beforeDelivery(this.#operations.handoff(
         request.chatId,
         predecessor,
         request.operation,
-        request.beforeDelivery,
-      ),
+        handoff,
+      )),
     );
   }