From 89efb007f94e7c3809d29dce70ba2cf098fcafcd Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 22 Apr 2026 18:54:20 +0700 Subject: [PATCH] Fix Codex replacement turn idle flash --- .../contexts/session-stream-lifecycle.test.ts | 7 + .../src/contexts/session-stream-lifecycle.ts | 5 +- .../contexts/session-stream-reducers.test.ts | 22 ++ ...nd-during-tool-call-codex.real.e2e.test.ts | 304 ++++++++++++++++++ 4 files changed, 337 insertions(+), 1 deletion(-) create mode 100644 packages/server/src/server/daemon-e2e/send-during-tool-call-codex.real.e2e.test.ts diff --git a/packages/app/src/contexts/session-stream-lifecycle.test.ts b/packages/app/src/contexts/session-stream-lifecycle.test.ts index 99abd41ba..3a004cc07 100644 --- a/packages/app/src/contexts/session-stream-lifecycle.test.ts +++ b/packages/app/src/contexts/session-stream-lifecycle.test.ts @@ -13,10 +13,17 @@ const turnFailedEvent: AgentStreamEventPayload = { error: "failed", }; +const turnCanceledEvent: AgentStreamEventPayload = { + type: "turn_canceled", + provider: "codex", + reason: "interrupted", +}; + describe("session stream lifecycle helpers", () => { it("derives optimistic terminal lifecycle only when current status is running", () => { expect(deriveOptimisticLifecycleStatus("running", turnCompletedEvent)).toBe("idle"); expect(deriveOptimisticLifecycleStatus("running", turnFailedEvent)).toBe("error"); + expect(deriveOptimisticLifecycleStatus("running", turnCanceledEvent)).toBe(null); expect(deriveOptimisticLifecycleStatus("initializing", turnCompletedEvent)).toBe(null); expect(deriveOptimisticLifecycleStatus("idle", turnFailedEvent)).toBe(null); }); diff --git a/packages/app/src/contexts/session-stream-lifecycle.ts b/packages/app/src/contexts/session-stream-lifecycle.ts index 28145a636..f31e65e32 100644 --- a/packages/app/src/contexts/session-stream-lifecycle.ts +++ b/packages/app/src/contexts/session-stream-lifecycle.ts @@ -10,10 +10,13 @@ export function deriveOptimisticLifecycleStatus( } switch (event.type) { case "turn_completed": - case "turn_canceled": return "idle"; case "turn_failed": return "error"; + case "turn_canceled": + // A canceled turn can be either a final user cancel or an interrupt before + // a replacement turn starts. The daemon snapshot is authoritative here. + return null; default: return null; } diff --git a/packages/app/src/contexts/session-stream-reducers.test.ts b/packages/app/src/contexts/session-stream-reducers.test.ts index ba78f1301..ccd2c9757 100644 --- a/packages/app/src/contexts/session-stream-reducers.test.ts +++ b/packages/app/src/contexts/session-stream-reducers.test.ts @@ -602,6 +602,28 @@ describe("processAgentStreamEvent", () => { expect(result.agent!.status).toBe("error"); }); + it("does not derive optimistic idle status on turn_canceled for running agent", () => { + const turnCanceledEvent: AgentStreamEventPayload = { + type: "turn_canceled", + provider: "codex", + reason: "interrupted", + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: turnCanceledEvent, + currentAgent: { + status: "running", + updatedAt: new Date(1000), + lastActivityAt: new Date(1000), + }, + timestamp: new Date(2000), + }); + + expect(result.agentChanged).toBe(false); + expect(result.agent).toBe(null); + }); + it("does not change agent when status is not running", () => { const turnCompletedEvent: AgentStreamEventPayload = { type: "turn_completed", diff --git a/packages/server/src/server/daemon-e2e/send-during-tool-call-codex.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/send-during-tool-call-codex.real.e2e.test.ts new file mode 100644 index 000000000..9d3cdc747 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/send-during-tool-call-codex.real.e2e.test.ts @@ -0,0 +1,304 @@ +import { beforeAll, beforeEach, describe, expect, test } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; + +import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js"; +import type { AgentTimelineItem } from "../agent/agent-sdk-types.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { createMessageCollector } from "../test-utils/message-collector.js"; +import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import type { SessionOutboundMessage } from "../messages.js"; +import { getFullAccessConfig, isProviderAvailable } from "./agent-configs.js"; + +function tmpCwd(): string { + return mkdtempSync(path.join(tmpdir(), "daemon-real-codex-tool-interrupt-")); +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function generateClientMessageId(): string { + return `msg_${Date.now()}_${Math.random().toString(36).slice(2, 11)}`; +} + +function getAgentStatuses(messages: SessionOutboundMessage[], agentId: string): string[] { + return messages + .filter( + (message) => + message.type === "agent_update" && + message.payload.kind === "upsert" && + message.payload.agent.id === agentId, + ) + .map((message) => message.payload.agent.status); +} + +function getAgentStatusesBeforeFirstAssistant( + messages: SessionOutboundMessage[], + agentId: string, +): string[] { + const firstAssistantIndex = messages.findIndex( + (message) => + message.type === "agent_stream" && + message.payload.agentId === agentId && + message.payload.event.type === "timeline" && + message.payload.event.item.type === "assistant_message", + ); + const observedPrefix = + firstAssistantIndex < 0 ? messages : messages.slice(0, firstAssistantIndex); + return getAgentStatuses(observedPrefix, agentId); +} + +function getAssistantTexts(messages: SessionOutboundMessage[], agentId: string): string[] { + return messages + .filter( + (message) => + message.type === "agent_stream" && + message.payload.agentId === agentId && + message.payload.event.type === "timeline" && + message.payload.event.item.type === "assistant_message", + ) + .map((message) => message.payload.event.item.text); +} + +function hasProviderLimitText(text: string): boolean { + return /hit your limit|rate limit|quota|credits/i.test(text); +} + +function assertNoProviderLimit(timeline: Awaited>) { + const assistantTexts = timeline.entries + .filter((entry) => entry.item.type === "assistant_message") + .slice(-5) + .map((entry) => entry.item.text); + const limitText = assistantTexts.find((text) => hasProviderLimitText(text)); + if (limitText) { + throw new Error(`Codex provider rejected the run: ${limitText}`); + } +} + +function summarizeTimelineItems(timeline: Awaited>) { + return timeline.entries.slice(-15).map((entry) => { + const item = entry.item; + if (item.type === "assistant_message") { + return { type: item.type, text: item.text }; + } + if (item.type === "tool_call") { + return { + type: item.type, + name: item.name, + status: item.status, + callId: item.callId, + }; + } + if (item.type === "user_message") { + return { type: item.type, text: item.text }; + } + return { type: item.type }; + }); +} + +async function waitForAssistantWaitingOnSleep( + client: DaemonClient, + agentId: string, + timeoutMs = 75_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }); + assertNoProviderLimit(timeline); + const assistantText = timeline.entries + .filter((entry) => entry.item.type === "assistant_message") + .map((entry) => entry.item.text) + .join("\n"); + if (/still running|waiting/i.test(assistantText) && !/`done`|\bdone\b/i.test(assistantText)) { + return; + } + + await sleep(500); + } + + const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }).catch(() => null); + const recentToolCalls = + timeline?.entries + .filter((entry) => entry.item.type === "tool_call") + .slice(-10) + .map((entry) => ({ + name: entry.item.name, + status: entry.item.status, + callId: entry.item.callId, + })) ?? []; + const recentAssistantTexts = + timeline?.entries + .filter((entry) => entry.item.type === "assistant_message") + .slice(-5) + .map((entry) => entry.item.text) ?? []; + throw new Error( + `Timed out waiting for Codex to report it was waiting on sleep. Recent tool_calls=${JSON.stringify(recentToolCalls)} recent assistant text=${JSON.stringify(recentAssistantTexts)}`, + ); +} + +describe("daemon E2E (real codex) - send message during tool call", () => { + let canRun = false; + + beforeAll(async () => { + canRun = await isProviderAvailable("codex"); + }); + + beforeEach((context) => { + if (!canRun) { + context.skip(); + } + }); + + test("does not emit an idle agent_update between UI send and the replacement Codex turn", async () => { + const logger = pino({ level: "silent" }); + const cwd = tmpCwd(); + const daemon = await createTestPaseoDaemon({ + agentClients: { codex: new CodexAppServerAgentClient(logger) }, + logger, + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + let collector: ReturnType | null = null; + + try { + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "primary" } }); + + const agent = await client.createAgent({ + cwd, + title: "codex-tool-interrupt-repro", + ...getFullAccessConfig("codex"), + }); + + collector = createMessageCollector(client); + + await client.sendMessage( + agent.id, + "Run `sleep 60 && echo done` and tell me what it outputs. Be brief.", + ); + await client.waitForAgentUpsert( + agent.id, + (snapshot) => snapshot.status === "running", + 60_000, + ); + await waitForAssistantWaitingOnSleep(client, agent.id); + + collector.clear(); + + await client.sendAgentMessage(agent.id, "Reply with exactly: INTERRUPT_RECEIVED", { + messageId: generateClientMessageId(), + }); + + const finish = await client.waitForFinish(agent.id, 120_000); + const postSendMessages = [...collector.messages]; + const postSendStatuses = getAgentStatuses(postSendMessages, agent.id); + const statusesBeforeFirstAssistant = getAgentStatusesBeforeFirstAssistant( + postSendMessages, + agent.id, + ); + const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 }); + + if (finish.status !== "idle") { + const snapshot = await client.fetchAgent(agent.id); + throw new Error( + `Expected idle after replacement, got ${finish.status}. postSendStatuses=${JSON.stringify(postSendStatuses)} statusesBeforeFirstAssistant=${JSON.stringify(statusesBeforeFirstAssistant)} postSendAssistantTexts=${JSON.stringify(getAssistantTexts(postSendMessages, agent.id))} agentStatus=${snapshot?.agent.status ?? null} recentTimeline=${JSON.stringify(summarizeTimelineItems(timeline))}`, + ); + } + + expect(statusesBeforeFirstAssistant).not.toContain("idle"); + expect(statusesBeforeFirstAssistant).not.toContain("error"); + + const assistantTexts = timeline.entries + .filter((entry) => entry.item.type === "assistant_message") + .map((entry) => { + const item = entry.item as Extract; + return item.text; + }); + expect(assistantTexts.some((text) => text.includes("[System Error]"))).toBe(false); + expect(assistantTexts.some((text) => text.toUpperCase().includes("INTERRUPT_RECEIVED"))).toBe( + true, + ); + } finally { + collector?.unsubscribe(); + await client.close(); + await daemon.close(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 300_000); + + test("does not emit an idle agent_update when a second prompt is sent 200ms after the first", async () => { + const logger = pino({ level: "silent" }); + const cwd = tmpCwd(); + const daemon = await createTestPaseoDaemon({ + agentClients: { codex: new CodexAppServerAgentClient(logger) }, + logger, + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + let collector: ReturnType | null = null; + + try { + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "primary" } }); + + const agent = await client.createAgent({ + cwd, + title: "codex-quick-follow-up-repro", + ...getFullAccessConfig("codex"), + }); + + collector = createMessageCollector(client); + + await client.sendAgentMessage( + agent.id, + "Run `sleep 60 && echo done` and tell me what it outputs. Be brief.", + { + messageId: generateClientMessageId(), + }, + ); + await sleep(200); + + collector.clear(); + + await client.sendAgentMessage(agent.id, "Reply with exactly: QUICK_FOLLOW_UP_RECEIVED", { + messageId: generateClientMessageId(), + }); + + const finish = await client.waitForFinish(agent.id, 120_000); + const postSendMessages = [...collector.messages]; + const postSendStatuses = getAgentStatuses(postSendMessages, agent.id); + const statusesBeforeFirstAssistant = getAgentStatusesBeforeFirstAssistant( + postSendMessages, + agent.id, + ); + const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 }); + + if (finish.status !== "idle") { + const snapshot = await client.fetchAgent(agent.id); + throw new Error( + `Expected idle after quick follow-up, got ${finish.status}. postSendStatuses=${JSON.stringify(postSendStatuses)} statusesBeforeFirstAssistant=${JSON.stringify(statusesBeforeFirstAssistant)} postSendAssistantTexts=${JSON.stringify(getAssistantTexts(postSendMessages, agent.id))} agentStatus=${snapshot?.agent.status ?? null} recentTimeline=${JSON.stringify(summarizeTimelineItems(timeline))}`, + ); + } + + expect(statusesBeforeFirstAssistant).not.toContain("idle"); + expect(statusesBeforeFirstAssistant).not.toContain("error"); + + const assistantTexts = timeline.entries + .filter((entry) => entry.item.type === "assistant_message") + .map((entry) => { + const item = entry.item as Extract; + return item.text; + }); + expect(assistantTexts.some((text) => text.includes("[System Error]"))).toBe(false); + expect( + assistantTexts.some((text) => text.toUpperCase().includes("QUICK_FOLLOW_UP_RECEIVED")), + ).toBe(true); + } finally { + collector?.unsubscribe(); + await client.close(); + await daemon.close(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 300_000); +});