Drop a submitted prompt after session rotation

A clear after Enter could finish token refresh against the rebuilt agent, so the outgoing prompt started the new session.

Author: TheGreatAxios

src/tui/runner/exit.test.ts

@@ -1,9 +1,19 @@
 import { describe, expect, spyOn, test } from "bun:test";
+import { EventEmitter } from "node:events";
+import { type Agent } from "@intx/agent";
 import { getLogger } from "@intx/log";
+import type { InferenceSource } from "@intx/types/runtime";
 
+import * as codexSession from "../../auth/codex/session.js";
 import { LOG_NAMESPACE_ROOT } from "../../branding.js";
 import { defined } from "../../../tests/helpers/defined.js";
-import { finalizeTUIRun } from "./exit.js";
+import { createDeliveryGeneration } from "../queued-delivery.js";
+import { createSessionOperationQueue } from "../session-operation-queue.js";
+import {
+  createRunLifecycle,
+  finalizeTUIRun,
+  resetSessionForRotation,
+} from "./exit.js";
 import type { RunnerServices, RunnerState } from "./state.js";
 
 function stubQuit(args: {
@@ -114,3 +124,136 @@ describe("finalizeTUIRun quit order", () => {
     await expect(pending).rejects.toThrow("stop");
   });
 });
+
+const liveSource: InferenceSource = {
+  id: "codex/work",
+  provider: "openai",
+  baseURL: "https://example.test",
+  apiKey: "old-token",
+  model: "m",
+};
+
+function recordingAgent(sends: string[]): Agent {
+  return {
+    send: async (content) => {
+      sends.push(typeof content === "string" ? content : String(content));
+      return { type: "reply", reply: "", turn: {} as never };
+    },
+    stream: async function* stream() {
+      yield* [];
+    },
+    deliver: () => undefined,
+    close: async () => undefined,
+    setSource: () => undefined,
+    setSources: () => undefined,
+    history: async () => [],
+    checkpoints: async () => [],
+    readAt: async () => [],
+    get blobReader() {
+      return {} as Agent["blobReader"];
+    },
+  };
+}
+
+function stubSendLifecycle(agent: Agent): {
+  state: RunnerState;
+  services: RunnerServices;
+} {
+  const state = {
+    runTaskTitle: "keep-title",
+    liveSource,
+    connectedMcpServers: [],
+    config: { cwd: "/tmp", task: "keep-title" },
+    sessionId: "s",
+    startedAt: 1,
+    inFlight: 0,
+    fatalBuildError: null,
+    sendAborted: false,
+    initialCodexProfile: "work",
+    initialXaiProfile: undefined,
+    stampProvider: { fn: undefined },
+  } as unknown as RunnerState;
+  const services = {
+    crashGuard: {
+      isFinalized: () => true,
+      setPartialFlush: () => undefined,
+    },
+    cycleRecorder: {
+      dispose: async () => "",
+      handleEvent: () => undefined,
+    },
+    providerFailureAttempts: {},
+    correlationAcceptance: { observe: () => undefined },
+    runSink: { sink: () => undefined },
+    sessionCost: { addTurn: () => undefined },
+    buildAgent: async () => agent,
+    emitter: new EventEmitter(),
+    sessionOps: createSessionOperationQueue(),
+    deliveryGeneration: createDeliveryGeneration(),
+    toolset: { setToolPromoter: () => undefined },
+    subAgentSessions: { cancelAll: async () => [] },
+    activeRunHandle: { task: "", startedAt: 0, model: "" },
+  } as unknown as RunnerServices;
+  return { state, services };
+}
+
+function hangCodexRefresh(): {
+  settle: (value: { access: string }) => void;
+  spy: ReturnType<typeof spyOn>;
+} {
+  let settle: ((value: { access: string }) => void) | undefined;
+  const spy = spyOn(codexSession, "getValidCodexToken").mockImplementation(
+    () =>
+      new Promise<{ access: string }>((resolve) => {
+        settle = resolve;
+      }),
+  );
+  return {
+    spy,
+    settle: (value) => defined(settle, "settleRefresh")(value),
+  };
+}
+
+describe("agentProxy.send vs /clear", () => {
+  test("a /clear during hung OAuth after awaitTail does not send into the rebuilt agent", async () => {
+    const oldSends: string[] = [];
+    const newSends: string[] = [];
+    const oldAgent = recordingAgent(oldSends);
+    const newAgent = recordingAgent(newSends);
+    const { state, services } = stubSendLifecycle(oldAgent);
+    const hung = hangCodexRefresh();
+    try {
+      const { agentProxy } = await createRunLifecycle(state, services);
+      const pending = agentProxy.send("keep me out of the new session");
+      await Promise.resolve();
+      await Promise.resolve();
+
+      resetSessionForRotation(state, services);
+      state.currentAgent = newAgent;
+      hung.settle({ access: "fresh-token" });
+      await Promise.allSettled([pending]);
+
+      expect(newSends).toEqual([]);
+      expect(oldSends).toEqual([]);
+    } finally {
+      hung.spy.mockRestore();
+    }
+  });
+
+  test("hung OAuth after awaitTail still sends when the session is not rotated", async () => {
+    const sends: string[] = [];
+    const { state, services } = stubSendLifecycle(recordingAgent(sends));
+    const hung = hangCodexRefresh();
+    try {
+      const { agentProxy } = await createRunLifecycle(state, services);
+      const pending = agentProxy.send("deliver this");
+      await Promise.resolve();
+      await Promise.resolve();
+      hung.settle({ access: "fresh-token" });
+      await pending;
+      expect(sends).toEqual(["deliver this"]);
+    } finally {
+      hung.spy.mockRestore();
+    }
+  });
+});

src/tui/runner/exit.ts

@@ -5,7 +5,11 @@
  * proxy), the stream sink, and the quit-time finalization tail.
  */
 
-import { AgentContextLockError, type Agent } from "@intx/agent";
+import {
+  AgentClosedError,
+  AgentContextLockError,
+  type Agent,
+} from "@intx/agent";
 import { getLogger } from "@intx/log";
 import type { InferenceSource } from "@intx/types/runtime";
 import { consumeStream } from "../../session/stream-consumer.js";
@@ -300,7 +304,8 @@ export async function createRunLifecycle(
 
   // Serial operation queue. Rotation (reload, interrupt, newSession), compaction
   // continuation, and proxy deliver enqueue async tasks; they run one at a time.
-  // `send` awaits the tail before dispatching so it never races a concurrent rebuild.
+  // `send` awaits the tail, then drops if /clear|/new bumped delivery generation
+  // during the wait or token refresh so the prompt cannot land on the rebuilt agent.
   const enqueueOp = services.sessionOps.enqueue;
 
   const reloadIfIdle = (): void => {
@@ -407,7 +412,14 @@ export async function createRunLifecycle(
   // Host mounts later; stampProvider.fn is wired once the bridge exists.
   const agentProxy: Agent = {
     send: async (content, opts) => {
+      const stillCurrent = services.deliveryGeneration.capture();
+      const dropIfRotated = (): void => {
+        if (stillCurrent()) return;
+        state.sendAborted = true;
+        throw new AgentClosedError();
+      };
       await services.sessionOps.awaitTail();
+      dropIfRotated();
       if (state.fatalBuildError !== null) throw state.fatalBuildError;
       const trimmed = typeof content === "string" ? content.trim() : "";
       if (trimmed.length > 0 && state.runTaskTitle.trim().length === 0) {
@@ -422,6 +434,7 @@ export async function createRunLifecycle(
       return await runWhileAgentBusy(state, async () => {
         await refreshCodexBeforeSend();
         await refreshXaiBeforeSend();
+        dropIfRotated();
         return await liveAgent(state).send(content, opts);
       });
     },

src/tui/runner/submit.ts

@@ -313,6 +313,7 @@ export function createSubmitPath(
     text: string,
     pending: readonly PendingImageAttachment[],
   ): Promise<void> => {
+    const stillCurrent = services.deliveryGeneration.capture();
     state.sendAborted = false;
     if (text.trim().length > 0) {
       void appendSentMessage(state.config.cwd, state.sessionId, text).catch(
@@ -329,6 +330,7 @@ export function createSubmitPath(
       imageAttachmentFromPath,
       pending,
     );
+    if (!stillCurrent()) return;
     await sendWithAttemptIdentity(
       userInboundMessage(ingested.text, ingested.attachments),
     );