fix(agents): hand off active input at delivery boundary

Prevents stale predecessor restoration during active-input admission.

Author: yongkangc

integration-tests/support/integration-fixture.ts

@@ -1,4 +1,4 @@
-import { mkdir, mkdtemp, readFile, rm, utimes, writeFile } from 'node:fs/promises';
+import { appendFile, mkdir, mkdtemp, readFile, rm, utimes, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { dirname, join, resolve } from 'node:path';
 import { fileURLToPath } from 'node:url';
@@ -258,14 +258,35 @@ export class IntegrationFixture {
     await this.#startReplacementGarcon();
   }
 
-  async #removeFinalNativeUserRow(input: { chatId: string; clientRequestId: string }): Promise<void> {
+  async appendDirectOpenAiNativeMessage(input: {
+    chatId: string;
+    role: 'user' | 'assistant';
+    content: string;
+    clientRequestId?: string;
+    turnId?: string;
+  }): Promise<void> {
+    const nativePath = await this.#directOpenAiNativePath(input.chatId);
+    const raw = await readFile(nativePath, 'utf8');
+    if (raw.length > 0 && !raw.endsWith('\n')) {
+      throw new Error('Direct native transcript has an incomplete tail.');
+    }
+    await appendFile(nativePath, `${JSON.stringify({
+      role: input.role,
+      content: input.content,
+      timestamp: new Date().toISOString(),
+      ...(input.clientRequestId ? { clientRequestId: input.clientRequestId } : {}),
+      ...(input.turnId ? { turnId: input.turnId } : {}),
+    })}\n`, 'utf8');
+  }
+
+  async #directOpenAiNativePath(chatId: string): Promise<string> {
     const registry = JSON.parse(
       await readFile(join(this.dirs.workspace, 'chats.json'), 'utf8'),
     ) as { sessions?: Record<string, Record<string, unknown>> };
-    const chat = registry.sessions?.[input.chatId];
-    if (!chat) throw new Error(`Chat ${input.chatId} was not persisted before crash.`);
+    const chat = registry.sessions?.[chatId];
+    if (!chat) throw new Error(`Chat ${chatId} was not persisted before restart.`);
     if (chat.agentId !== DIRECT_OPENAI_CHAT_COMPLETIONS_COMPATIBLE_AGENT_ID) {
-      throw new Error(`Chat ${input.chatId} is not a direct OpenAI-compatible chat.`);
+      throw new Error(`Chat ${chatId} is not a direct OpenAI-compatible chat.`);
     }
     const nativeSession = chat.nativeSession && typeof chat.nativeSession === 'object'
       ? chat.nativeSession as Record<string, unknown>
@@ -290,9 +311,13 @@ export class IntegrationFixture {
       || !nativePath
       || resolve(nativePath) !== expectedPath
     ) {
-      throw new Error(`Chat ${input.chatId} has an unexpected native transcript path.`);
+      throw new Error(`Chat ${chatId} has an unexpected native transcript path.`);
     }
+    return expectedPath;
+  }
 
+  async #removeFinalNativeUserRow(input: { chatId: string; clientRequestId: string }): Promise<void> {
+    const expectedPath = await this.#directOpenAiNativePath(input.chatId);
     const raw = await readFile(expectedPath, 'utf8');
     if (!raw.endsWith('\n')) throw new Error('Direct native transcript has an incomplete tail.');
     const lines = raw.split('\n').filter((line) => line.length > 0);

integration-tests/tests/server/persistence-lifecycle.test.ts

@@ -69,7 +69,7 @@ describe('persistence lifecycle', () => {
     await withIntegrationFixture('ephemeral-queue-restart', async (fixture) => {
       const chatId = fixture.newChatId();
       const held = fixture.fakeProviders.openAi.holdNext({ lastUserText: 'ephemeral-active' });
-      await fixture.client.startDirectChat({
+      const active = await fixture.client.startDirectChat({
         chatId,
         content: 'ephemeral-active',
         projectPath: fixture.dirs.project,
@@ -80,14 +80,45 @@ describe('persistence lifecycle', () => {
       const paused = await fixture.client.pauseQueue(chatId);
       expect(paused.control.queue.entries.map((entry) => entry.content)).toEqual(['discard-on-restart']);
       expect(paused.control.queue.pause?.kind).toBe('manual');
+      expect((await fixture.client.reconnectState([chatId])).processing).toEqual({
+        outcome: 'snapshot',
+        runningChatIds: [chatId],
+      });
 
       const activeAborted = held.expectAbort();
-      await fixture.restartGarcon();
+      await fixture.restartGarcon({
+        beforeStart: () => fixture.appendDirectOpenAiNativeMessage({
+          chatId,
+          role: 'assistant',
+          content: 'terminal persisted after disconnect',
+          clientRequestId: active.clientRequestId,
+          turnId: active.turnId,
+        }),
+      });
       await activeAborted;
       held.releaseTruncatedStream();
 
-      const control = await fixture.client.getExecutionControl(chatId);
-      expect(control).toEqual({
+      const restarted = await fixture.client.reconnectState([chatId]);
+      expect(restarted.processing).toEqual({
+        outcome: 'snapshot',
+        runningChatIds: [],
+      });
+      expect(restarted.controlResults).toEqual([{
+        chatId,
+        outcome: 'snapshot',
+        control: {
+          queue: {
+            entries: [],
+            dispatchingEntryId: null,
+            recentlyDispatched: [],
+            pause: null,
+            reorderRevision: 0,
+          },
+          version: 0,
+          updatedAt: null,
+        },
+      }]);
+      expect(await fixture.client.getExecutionControl(chatId)).toEqual({
         queue: {
           entries: [],
           dispatchingEntryId: null,
@@ -98,9 +129,12 @@ describe('persistence lifecycle', () => {
         version: 0,
         updatedAt: null,
       });
+      const reloaded = await fixture.client.reloadChat(chatId);
+      expect(assistantContents(reloaded.messages)).toContain('terminal persisted after disconnect');
       const restored = await fixture.client.getMessages(chatId);
       expect(restored.pendingUserInputs).toEqual([]);
       expect(countUserContent(restored.messages, 'discard-on-restart')).toBe(0);
+      expect(assistantContents(restored.messages)).toContain('terminal persisted after disconnect');
 
       const next = await fixture.client.runDirectChat({
         chatId,

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

@@ -67,6 +67,20 @@ function startRequest(overrides = {}) {
   };
 }
 
+function activeInputRequest(operation, beforeDelivery = async () => undefined) {
+  return startRequest({
+    agentSessionId: 'thread-1',
+    nativeSession: {
+      ownerId: 'codex',
+      schemaVersion: 1,
+      value: { path: '/tmp/thread-1.jsonl', agentSessionId: 'thread-1' },
+    },
+    operation,
+    beforeDelivery,
+    carryOver: undefined,
+  });
+}
+
 describe('CodexExecution', () => {
   it('preserves admission, endpoint configuration, session identity, and event identity', async () => {
     const runtime = createRuntime();
@@ -170,4 +184,72 @@ describe('CodexExecution', () => {
     await expect(execution.start(startRequest({ prompt: '/goal clear' })))
       .rejects.toMatchObject({ code: 'INVALID_SETTINGS' });
   });
+
+  for (const outcome of ['decline', 'failure']) {
+    it(`keeps the predecessor operation visible when active input has a pre-boundary ${outcome}`, async () => {
+      const runtime = createRuntime();
+      const execution = new CodexExecution(
+        createHost(),
+        runtime,
+        createPathNativeSessionCodec('codex'),
+        createConfig(),
+      );
+      const predecessor = startRequest().operation;
+      const successor = { ...predecessor, clientRequestId: 'request-2', turnId: 'turn-2' };
+      const next = { ...predecessor, clientRequestId: 'request-3', turnId: 'turn-3' };
+      const events = [];
+      execution.subscribe((event) => events.push(event));
+      await execution.start(startRequest());
+      runtime.submitActiveInput.mockImplementation(async () => {
+        runtime.emitFinished('chat-1', 0, {
+          clientRequestId: predecessor.clientRequestId,
+          turnId: predecessor.turnId,
+        });
+        if (outcome === 'failure') throw new Error('failed before delivery boundary');
+        return false;
+      });
+
+      const activeInput = execution.submitActiveInput(activeInputRequest(successor));
+      if (outcome === 'failure') await expect(activeInput).rejects.toThrow('failed before delivery boundary');
+      else await expect(activeInput).resolves.toBe(false);
+      expect(events).toContainEqual(expect.objectContaining({
+        type: 'finished',
+        operation: predecessor,
+      }));
+
+      await execution.resume(activeInputRequest(next));
+      expect(runtime.runTurn).toHaveBeenCalledOnce();
+    });
+  }
+
+  it('retains successor ownership after a post-boundary delivery failure', async () => {
+    const runtime = createRuntime();
+    const execution = new CodexExecution(
+      createHost(),
+      runtime,
+      createPathNativeSessionCodec('codex'),
+      createConfig(),
+    );
+    const predecessor = startRequest().operation;
+    const successor = { ...predecessor, clientRequestId: 'request-2', turnId: 'turn-2' };
+    const events = [];
+    execution.subscribe((event) => events.push(event));
+    await execution.start(startRequest());
+    runtime.submitActiveInput.mockImplementation(async (_request, beforeDelivery) => {
+      await beforeDelivery();
+      throw new Error('delivery outcome unknown');
+    });
+
+    await expect(execution.submitActiveInput(activeInputRequest(successor)))
+      .rejects.toThrow('delivery outcome unknown');
+    runtime.emitFailed('chat-1', 'delivery failed', {
+      clientRequestId: successor.clientRequestId,
+      turnId: successor.turnId,
+    });
+
+    expect(events).toContainEqual(expect.objectContaining({
+      type: 'failed',
+      operation: successor,
+    }));
+  });
 });

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

@@ -98,23 +98,21 @@ export class CodexExecution implements AgentExecution {
   async submitActiveInput(
     request: Parameters<NonNullable<AgentExecution['submitActiveInput']>>[0],
   ): Promise<boolean> {
-    this.#operations.register(request.chatId, request.operation);
-    try {
-      const runtimeRequest = prepareResumeRequest(
-        request,
-        await this.#runtimeConfiguration(request),
-        this.nativeSessions,
-      );
-      const accepted = await this.runtime.submitActiveInput(
-        runtimeRequest,
+    const predecessor = this.#operations.current(request.chatId);
+    const runtimeRequest = prepareResumeRequest(
+      request,
+      await this.#runtimeConfiguration(request),
+      this.nativeSessions,
+    );
+    return this.runtime.submitActiveInput(
+      runtimeRequest,
+      () => this.#operations.handoff(
+        request.chatId,
+        predecessor,
+        request.operation,
         request.beforeDelivery,
-      );
-      if (!accepted) this.#operations.finish(request.chatId, request.operation);
-      return accepted;
-    } catch (error) {
-      this.#operations.finish(request.chatId, request.operation);
-      throw error;
-    }
+      ),
+    );
   }
 
   async compact(

server-agents/common/src/execution/operation-tracker.ts

@@ -8,6 +8,27 @@ export class AgentOperationTracker {
     this.#operations.set(chatId, operation);
   }
 
+  async handoff(
+    chatId: string,
+    predecessor: AgentOperationIdentity | null,
+    successor: AgentOperationIdentity,
+    commit: () => Promise<void>,
+  ): Promise<void> {
+    if ((this.#operations.get(chatId) ?? null) !== predecessor) {
+      throw new Error(`Cannot hand off operation for chat ${chatId} after its active operation changed`);
+    }
+    this.#operations.set(chatId, successor);
+    try {
+      await commit();
+    } catch (error) {
+      if (this.#operations.get(chatId) === successor) {
+        if (predecessor) this.#operations.set(chatId, predecessor);
+        else this.#operations.delete(chatId);
+      }
+      throw error;
+    }
+  }
+
   current(
     chatId: string,
     metadata?: RuntimeEventMetadata,

server/agents/__tests__/event-bus.test.js

@@ -60,7 +60,7 @@ describe('AgentEventBus', () => {
     const { bus } = makeBus();
     bus.trackTurn('chat-1', operation('turn-1'));
     bus.markTurnAbortable('chat-1', operation('turn-1'));
-    bus.replaceTurn('chat-1', operation('turn-2'));
+    await bus.handoffTurn('chat-1', operation('turn-1'), operation('turn-2'), async () => undefined);
     const controller = new AbortController();
     const waiting = bus.waitUntilTurnAbortable('chat-1', operation('turn-2'), controller.signal);
     controller.abort();
@@ -79,15 +79,48 @@ describe('AgentEventBus', () => {
     expect(bus.getActiveTurn('chat-1')?.turnId).toBe('turn-1');
   });
 
-  it('allows an explicit active-input identity replacement', () => {
+  it('commits an explicit active-input identity handoff at its delivery boundary', async () => {
     const { bus } = makeBus();
     bus.trackTurn('chat-1', operation('turn-1'));
 
-    bus.replaceTurn('chat-1', operation('turn-2'));
+    await bus.handoffTurn('chat-1', operation('turn-1'), operation('turn-2'), async () => {
+      expect(bus.getActiveTurn('chat-1')?.turnId).toBe('turn-2');
+    });
 
     expect(bus.getActiveTurn('chat-1')?.turnId).toBe('turn-2');
   });
 
+  it('restores the predecessor only when the delivery boundary fails while the successor owns it', async () => {
+    const { bus } = makeBus();
+    bus.trackTurn('chat-1', operation('turn-1'));
+    bus.markTurnAbortable('chat-1', operation('turn-1'));
+
+    await expect(bus.handoffTurn(
+      'chat-1',
+      operation('turn-1'),
+      operation('turn-2'),
+      async () => { throw new Error('registration failed'); },
+    )).rejects.toThrow('registration failed');
+
+    expect(bus.getActiveTurn('chat-1')?.turnId).toBe('turn-1');
+    await expect(bus.waitUntilTurnAbortable('chat-1', operation('turn-1'))).resolves.toBe(true);
+  });
+
+  it('rejects a handoff after the predecessor identity has changed', async () => {
+    const { bus } = makeBus();
+    bus.trackTurn('chat-1', operation('turn-1'));
+    bus.settleTurn('chat-1', operation('turn-1'));
+    bus.trackTurn('chat-1', operation('turn-3'));
+
+    await expect(bus.handoffTurn(
+      'chat-1',
+      operation('turn-1'),
+      operation('turn-2'),
+      async () => undefined,
+    )).rejects.toThrow('active turn changed');
+    expect(bus.getActiveTurn('chat-1')?.turnId).toBe('turn-3');
+  });
+
   it('retains exact identity through duplicate terminal events until settlement', () => {
     const { bus, emit } = makeBus();
     const terminals = [];