diff --git a/src/tui/runner/exit.test.ts b/src/tui/runner/exit.test.ts index f2749f152..d3358ea2c 100644 --- a/src/tui/runner/exit.test.ts +++ b/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; +} { + 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(); + } + }); +}); diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index b7877c615..94731eb60 100644 --- a/src/tui/runner/exit.ts +++ b/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); }); }, diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index 8d4ee8564..0a40f6d36 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -313,6 +313,7 @@ export function createSubmitPath( text: string, pending: readonly PendingImageAttachment[], ): Promise => { + 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), );