diff --git a/apps/web/app/api/agent-sessions/[id]/route.test.ts b/apps/web/app/api/agent-sessions/[id]/route.test.ts index 91ed68a..1ace842 100644 --- a/apps/web/app/api/agent-sessions/[id]/route.test.ts +++ b/apps/web/app/api/agent-sessions/[id]/route.test.ts @@ -236,8 +236,8 @@ describe("agent session route", () => { expect(mocks.updateAgentSessionMetadata).not.toHaveBeenCalled(); }); - it("forwards steering as an active-turn command", async () => { - const response = await POST( + it("rejects delivery modes that bypass the connector-owned queue", async () => { + const steer = await POST( request("POST", { type: "steer", message: "Focus on the failing test", @@ -245,19 +245,18 @@ describe("agent session route", () => { }), context, ); - - expect(response.status).toBe(200); - expect(mocks.daemonRequest).toHaveBeenCalledWith( - "connector", - expect.objectContaining({ - commandId: "steer-message", - command: expect.objectContaining({ type: "steer" }), + const interrupt = await POST( + request("POST", { + type: "interrupt", + message: "Replace the current approach", + clientMessageId: "interrupt-message", }), + context, ); - expect(mocks.updateAgentSessionMetadata).toHaveBeenCalledWith( - "session", - { providerModifiedAt: expect.any(Date) }, - ); + + expect(steer.status).toBe(400); + expect(interrupt.status).toBe(400); + expect(mocks.daemonRequest).not.toHaveBeenCalled(); }); it("creates a new connector-owned workspace session", async () => { diff --git a/apps/web/app/api/agent-sessions/[id]/route.ts b/apps/web/app/api/agent-sessions/[id]/route.ts index 0d73ab7..3c0ce45 100644 --- a/apps/web/app/api/agent-sessions/[id]/route.ts +++ b/apps/web/app/api/agent-sessions/[id]/route.ts @@ -141,7 +141,6 @@ export async function POST( if ( command.type === "prompt" || command.type === "implement_plan" || - command.type === "steer" || command.type === "steer_queued_message" ) { await updateAgentSessionMetadata(id, { diff --git a/apps/web/components/agents/AgentComposer.tsx b/apps/web/components/agents/AgentComposer.tsx index 744331b..eb75609 100644 --- a/apps/web/components/agents/AgentComposer.tsx +++ b/apps/web/components/agents/AgentComposer.tsx @@ -16,6 +16,7 @@ import { Loader2, Pencil, Square, + Trash2, X, } from "lucide-react"; import { Button } from "@/components/ui/button"; @@ -75,6 +76,7 @@ export function AgentComposer({ onSubmit, onStop, onEditQueued, + onDeleteQueued, onSteerQueued, restoreDraftKey, }: { @@ -92,15 +94,16 @@ export function AgentComposer({ onSubmit: ( message: string, images: AgentPromptImage[], - delivery: "prompt" | "queue" | "steer", ) => Promise; onStop: () => void; onEditQueued: (id: string) => Promise; + onDeleteQueued: (id: string) => Promise; onSteerQueued: (id: string) => Promise; restoreDraftKey?: string; }) { const [input, setInput] = useState(""); const [submitting, setSubmitting] = useState(false); + const submittingRef = useRef(false); const [dismissedDraft, setDismissedDraft] = useState(null); const [activeIndex, setActiveIndex] = useState(0); const textareaRef = useRef(null); @@ -178,7 +181,7 @@ export function AgentComposer({ !pending && !disabled ) { - void submitMessage(`/${command.name}`, [], "prompt"); + void submitMessage(`/${command.name}`, []); return; } @@ -195,30 +198,30 @@ export function AgentComposer({ async function submitMessage( message: string, images: AgentPromptImage[], - delivery: "prompt" | "queue" | "steer", ) { - if ((!message && images.length === 0) || pending || submitting || disabled) + if ( + (!message && images.length === 0) || + pending || + submittingRef.current || + disabled + ) return; + submittingRef.current = true; setSubmitting(true); try { - const accepted = await onSubmit(message, images, delivery); + const accepted = await onSubmit(message, images); if (!accepted) return; setInput(""); clearAttachments(); setDismissedDraft(null); setActiveIndex(0); } finally { + submittingRef.current = false; setSubmitting(false); } } - function submit( - delivery: "prompt" | "queue" | "steer" = running - ? supportsSteer - ? "steer" - : "queue" - : "prompt", - ) { + function submit() { const message = input.trim(); const prefix = "/api/uploads/"; const images = readyParts.flatMap((part) => @@ -235,7 +238,7 @@ export function AgentComposer({ ] : [], ); - void submitMessage(message, images, delivery); + void submitMessage(message, images); } function addImageFiles(files: readonly File[]) { @@ -291,12 +294,13 @@ export function AgentComposer({ async function editQueuedMessage(message: AgentQueuedMessage) { if ( pending || - submitting || + submittingRef.current || disabled || input.trim() || attachments.length > 0 ) return; + submittingRef.current = true; setSubmitting(true); try { const accepted = await onEditQueued(message.id); @@ -312,16 +316,31 @@ export function AgentComposer({ ); requestAnimationFrame(() => textareaRef.current?.focus()); } finally { + submittingRef.current = false; setSubmitting(false); } } async function steerQueuedMessage(id: string) { - if (pending || submitting || disabled) return; + if (pending || submittingRef.current || disabled) return; + submittingRef.current = true; setSubmitting(true); try { await onSteerQueued(id); } finally { + submittingRef.current = false; + setSubmitting(false); + } + } + + async function deleteQueuedMessage(id: string) { + if (pending || submittingRef.current || disabled) return; + submittingRef.current = true; + setSubmitting(true); + try { + await onDeleteQueued(id); + } finally { + submittingRef.current = false; setSubmitting(false); } } @@ -421,74 +440,101 @@ export function AgentComposer({ {queuedMessages.length > 0 && (
-
- {queuedMessages.map((queuedMessage) => { - const sending = queuedMessage.status === "sending"; - return ( -
+ {queuedMessages.map((queuedMessage) => { + const sending = queuedMessage.status === "sending"; + const imageCount = queuedMessage.images?.length ?? 0; + return ( +
+ {sending ? ( + + ) : ( - +

- {queuedMessage.message.replace(/\s+/g, " ") || - `${queuedMessage.images?.length ?? 0} attached ${(queuedMessage.images?.length ?? 0) === 1 ? "image" : "images"}`} - - {!sending && ( - <> + {queuedMessage.message || + `${imageCount} attached ${imageCount === 1 ? "image" : "images"}`} +

+

+ {sending ? "Sending" : "Queued"} + {imageCount > 0 + ? ` ยท ${imageCount} ${imageCount === 1 ? "image" : "images"}` + : ""} +

+
+ {!sending && ( +
+ + {supportsSteer && running && ( - {supportsSteer && running && ( - - )} - - )} - - {sending ? "Sending" : "Queued"} - -
- ); - })} -
+ )} + + + )} + + ); + })}
)} @@ -573,7 +619,7 @@ export function AgentComposer({ /> )} - {running && supportsSteer && ( + {running && ( - )} diff --git a/apps/web/components/agents/AgentSessionView.tsx b/apps/web/components/agents/AgentSessionView.tsx index 2546a98..8272db8 100644 --- a/apps/web/components/agents/AgentSessionView.tsx +++ b/apps/web/components/agents/AgentSessionView.tsx @@ -213,7 +213,6 @@ export function AgentSessionView({ async function submit( message: string, images: AgentPromptImage[], - delivery: "prompt" | "queue" | "steer", ): Promise { let input: AgentSessionCommand; try { @@ -222,10 +221,10 @@ export function AgentSessionView({ snapshot?.state ?? {}, ); input = - normalized.type !== "prompt" || delivery === "prompt" + normalized.type !== "prompt" || snapshot?.status !== "running" ? normalized : { - type: delivery, + type: "queue", message, ...(images.length > 0 ? { images } : {}), }; @@ -472,6 +471,9 @@ export function AgentSessionView({ onEditQueued={(id) => run({ type: "remove_queued_message", id }) } + onDeleteQueued={(id) => + run({ type: "remove_queued_message", id }) + } onSteerQueued={(id) => run({ type: "steer_queued_message", id }) } diff --git a/apps/web/e2e/agent-runtime.spec.ts b/apps/web/e2e/agent-runtime.spec.ts index 29a26b9..fc3aea7 100644 --- a/apps/web/e2e/agent-runtime.spec.ts +++ b/apps/web/e2e/agent-runtime.spec.ts @@ -406,35 +406,35 @@ test("shows durable turn activity without changing completed tool status", async } if ( command.type === "abort" || - command.type === "steer" || command.type === "queue" || command.type === "remove_queued_message" || command.type === "steer_queued_message" ) { await new Promise((resolve) => setTimeout(resolve, 750)); } + const queueResult = + command.type === "queue" + ? { + queuedMessages: [ + { + id: "queued-message", + message: command.message, + ...(Array.isArray(command.images) + ? { images: command.images } + : {}), + status: "pending", + }, + ], + } + : command.type === "remove_queued_message" || + command.type === "steer_queued_message" + ? { queuedMessages: [] } + : {}; await route.fulfill({ contentType: "application/json", body: JSON.stringify({ accepted: true, - ...(command.type === "queue" - ? { - queuedMessages: [ - { - id: "queued-message", - message: command.message, - ...(Array.isArray(command.images) - ? { images: command.images } - : {}), - status: "pending", - }, - ], - } - : command.type === "remove_queued_message" - ? { queuedMessages: [] } - : command.type === "steer_queued_message" - ? { queuedMessages: [] } - : {}), + ...queueResult, }), }); return; @@ -694,9 +694,14 @@ test("shows durable turn activity without changing completed tool status", async page.getByText("Turn changes", { exact: true }), ).toHaveCount(0); - const composer = page.getByPlaceholder( + const composer = agentComposer.getByRole("combobox"); + await expect(composer).toHaveAttribute( + "placeholder", "Message Codex or type / for commands", ); + await expect( + page.getByLabel("Active turn message actions"), + ).toHaveCount(0); await expect(page.getByRole("button", { name: "Attach images" })).toBeVisible(); await composer.evaluate(async (element) => { const canvas = document.createElement("canvas"); @@ -749,7 +754,10 @@ test("shows durable turn activity without changing completed tool status", async }, ], }); - + await page.screenshot({ + path: testInfo.outputPath("runtime-queued-message-desktop.png"), + fullPage: true, + }); await page.getByRole("button", { name: "Edit queued message" }).click(); await expect(composer).toHaveValue("Run the tests"); await expect(composer).toBeEnabled(); @@ -763,22 +771,75 @@ test("shows durable turn activity without changing completed tool status", async await composer.fill(""); await composer.fill("Focus on the failing test"); - await page.getByRole("button", { name: "Steer Codex" }).click(); + await composer.press("Enter"); await expect(composer).toHaveValue("Focus on the failing test"); await expect(composer).toBeDisabled(); await expect(composer).toHaveValue(""); - - await composer.fill("Then summarize"); - await page.getByRole("button", { name: "Queue message for Codex" }).click(); await expect( - page.getByText("Then summarize", { exact: true }), + page.getByText("Focus on the failing test", { exact: true }), ).toBeVisible(); await page .getByRole("button", { name: "Steer with queued message" }) .click(); + await expect.poll(() => submittedCommands.at(-1)).toMatchObject({ + type: "steer_queued_message", + }); + await expect( + page.getByText("Focus on the failing test", { exact: true }), + ).toHaveCount(0); + + await expect(composer).toHaveAttribute( + "placeholder", + "Message Codex or type / for commands", + ); + await composer.fill("Delete this follow-up"); + await composer.press("Enter"); + await expect( + page.getByText("Delete this follow-up", { exact: true }), + ).toBeVisible(); + await page.getByRole("button", { name: "Delete queued message" }).click(); + await expect( + page.getByText("Delete this follow-up", { exact: true }), + ).toHaveCount(0); + await expect( - page.getByText("Then summarize", { exact: true }), + page.getByRole("button", { name: "Interrupt Codex and send message" }), ).toHaveCount(0); + await expect( + page.getByRole("button", { name: "Steer Codex" }), + ).toHaveCount(0); + + await composer.fill("Queue with the alternate shortcut"); + await composer.press("Control+Enter"); + await expect.poll(() => submittedCommands.at(-1)).toMatchObject({ + type: "queue", + message: "Queue with the alternate shortcut", + }); + await page.getByRole("button", { name: "Delete queued message" }).click(); + await expect(composer).toBeEnabled(); + + await composer.fill("Submit this only once"); + const commandCountBeforeDoubleSubmit = submittedCommands.length; + await composer.evaluate((element) => { + for (let index = 0; index < 2; index += 1) { + element.dispatchEvent( + new KeyboardEvent("keydown", { + key: "Enter", + bubbles: true, + cancelable: true, + }), + ); + } + }); + await expect.poll(() => submittedCommands.length).toBe( + commandCountBeforeDoubleSubmit + 1, + ); + await expect.poll(() => submittedCommands.at(-1)).toMatchObject({ + type: "queue", + message: "Submit this only once", + }); + await expect(composer).toBeEnabled(); + expect(submittedCommands).toHaveLength(commandCountBeforeDoubleSubmit + 1); await page.getByRole("button", { name: "Stop Codex" }).click(); await expect(genericActivity).toContainText("Stopping"); diff --git a/apps/web/e2e/connections.spec.ts b/apps/web/e2e/connections.spec.ts index 3ae0d58..94eacbb 100644 --- a/apps/web/e2e/connections.spec.ts +++ b/apps/web/e2e/connections.spec.ts @@ -521,16 +521,14 @@ test("connect local Pi, attach a workspace, and open a native session", async ({ timeout: 150_000, }); await expect( - page.getByPlaceholder("Message Pi or type / for commands"), + page.getByTestId("agent-composer").getByRole("combobox"), ).toBeVisible(); await expect(page.getByLabel("Session usage")).toBeVisible(); await expect(page.getByLabel("Session actions")).toBeVisible(); }); await test.step("select and execute a built-in slash command", async () => { - const composer = page.getByPlaceholder( - "Message Pi or type / for commands", - ); + const composer = page.getByTestId("agent-composer").getByRole("combobox"); await composer.fill("/na"); const nameCommand = page.getByRole("option", { name: /\/name.*Set the session name.*Built-in/, @@ -601,7 +599,7 @@ test("connect local Pi, attach a workspace, and open a native session", async ({ await page.setViewportSize({ width: 390, height: 844 }); await expect( - page.getByPlaceholder("Message Pi or type / for commands"), + page.getByTestId("agent-composer").getByRole("combobox"), ).toBeVisible(); await page.getByLabel("Open sidebar").click(); await expect( @@ -708,9 +706,7 @@ test("connect local Oh My Pi and use its native commands", async ({ timeout: 150_000, }); - const composer = page.getByPlaceholder( - "Message Oh My Pi or type / for commands", - ); + const composer = page.getByTestId("agent-composer").getByRole("combobox"); await composer.fill("/model"); await composer.press("Enter"); await expect(page.getByText(/Current model:/)).toBeVisible({ @@ -793,9 +789,7 @@ test("connect local Codex and resume a native thread", async ({ timeout: 150_000, }); - const composer = page.getByPlaceholder( - "Message Codex or type / for commands", - ); + const composer = page.getByTestId("agent-composer").getByRole("combobox"); await composer.fill("/usage"); await composer.press("Enter"); const usageDialog = page.getByRole("dialog", { name: "Codex usage" }); @@ -994,9 +988,7 @@ test("connect to Oh My Pi through an existing SSH alias", async ({ timeout: 150_000, }); - const composer = page.getByPlaceholder( - "Message Oh My Pi or type / for commands", - ); + const composer = page.getByTestId("agent-composer").getByRole("combobox"); await composer.fill("/model"); await composer.press("Enter"); await expect(page.getByText(/Current model:/)).toBeVisible({ diff --git a/apps/web/lib/queries/agentSessions.ts b/apps/web/lib/queries/agentSessions.ts index 4a4fb6a..11adb92 100644 --- a/apps/web/lib/queries/agentSessions.ts +++ b/apps/web/lib/queries/agentSessions.ts @@ -109,7 +109,6 @@ export function useAgentSessionCommand(id: string) { }> => { const wireCommand = (command.type === "prompt" || - command.type === "steer" || command.type === "queue" || command.type === "implement_plan") && !command.clientMessageId @@ -157,7 +156,6 @@ export function useAgentSessionCommand(id: string) { if ( command.type === "prompt" || command.type === "implement_plan" || - command.type === "steer" || command.type === "steer_queued_message" || command.type === "set_session_name" || command.type === "new_session" || diff --git a/packages/agent-bridge/src/agents.ts b/packages/agent-bridge/src/agents.ts index 07d29f3..5d66562 100644 --- a/packages/agent-bridge/src/agents.ts +++ b/packages/agent-bridge/src/agents.ts @@ -320,12 +320,6 @@ export const agentSessionCommandSchema = z.discriminatedUnion("type", [ clientMessageId: clientMessageIdSchema, }), z.object({ type: z.literal("abort") }), - z.object({ - type: z.literal("steer"), - message: z.string().trim().max(200_000), - images: z.array(agentPromptImageSchema).max(MAX_AGENT_IMAGES).optional(), - clientMessageId: clientMessageIdSchema, - }), z.object({ type: z.literal("queue"), message: z.string().trim().max(200_000), diff --git a/packages/agent-bridge/src/state.test.ts b/packages/agent-bridge/src/state.test.ts index 851cfa8..899af2e 100644 --- a/packages/agent-bridge/src/state.test.ts +++ b/packages/agent-bridge/src/state.test.ts @@ -73,6 +73,87 @@ describe("agent runtime event reducer", () => { ]); }); + it("atomically restores canonical turn order after optimistic steer races", () => { + const initial = { ...snapshot(), messages: [] }; + const prompt = applyAgentRuntimeEnvelope( + initial, + event({ + type: "overtchat_submission", + message: { + role: "user", + content: "write a paragraph on overtchat", + overtchatSubmissionId: "prompt", + }, + }), + )!; + const steered = applyAgentRuntimeEnvelope( + prompt, + event({ + type: "overtchat_submission", + message: { + role: "user", + content: "2 more now", + overtchatSubmissionId: "steer", + }, + }), + )!; + const reconciled = applyAgentRuntimeEnvelope( + steered, + event({ + type: "overtchat_turn_update", + turnId: "turn-1", + messages: [ + { + id: "turn-1:user:0", + role: "user", + content: "write a paragraph on overtchat", + overtchatSubmissionId: "prompt", + overtchatTurnId: "turn-1", + }, + { + id: "assistant-1", + role: "assistant", + content: [{ type: "text", text: "First paragraph" }], + overtchatTurnId: "turn-1", + }, + { + id: "turn-1:user:1", + role: "user", + content: "2 more now", + overtchatSubmissionId: "steer", + overtchatTurnId: "turn-1", + }, + { + id: "assistant-2", + role: "assistant", + content: [{ type: "text", text: "Two more paragraphs" }], + overtchatTurnId: "turn-1", + }, + { + id: "turn-1:footer", + role: "turnFooter", + content: "First paragraph\n\nTwo more paragraphs", + overtchatTurnId: "turn-1", + }, + ], + }), + )!; + + expect( + reconciled.messages.map((message) => + message && typeof message === "object" + ? [Reflect.get(message, "role"), Reflect.get(message, "id")] + : null, + ), + ).toEqual([ + ["user", "turn-1:user:0"], + ["assistant", "assistant-1"], + ["user", "turn-1:user:1"], + ["assistant", "assistant-2"], + ["turnFooter", "turn-1:footer"], + ]); + }); + it("replaces repeated user lifecycle events with the same timestamp", () => { const started = applyAgentRuntimeEnvelope( snapshot(), @@ -412,6 +493,61 @@ describe("agent runtime event reducer", () => { ]); }); + it("reconciles identical submissions by client identity", () => { + const first = applyAgentRuntimeEnvelope( + snapshot(), + event({ + type: "overtchat_submission", + message: { + role: "user", + content: "Same follow-up", + timestamp: 100, + overtchatSubmissionId: "submission:1", + }, + }), + )!; + const second = applyAgentRuntimeEnvelope( + first, + event({ + type: "overtchat_submission", + message: { + role: "user", + content: "Same follow-up", + timestamp: 100, + overtchatSubmissionId: "submission:2", + }, + }), + )!; + const acknowledged = applyAgentRuntimeEnvelope( + second, + event({ + type: "message_start", + message: { + id: "provider:2", + role: "user", + content: "Same follow-up", + overtchatSubmissionId: "submission:2", + }, + }), + )!; + + expect(acknowledged.messages).toEqual([ + { role: "user", content: "Hello" }, + { + role: "user", + content: "Same follow-up", + timestamp: 100, + overtchatSubmissionId: "submission:1", + }, + { + id: "provider:2", + role: "user", + content: "Same follow-up", + overtchatSubmissionId: "submission:2", + }, + ]); + }); + it("removes a submission rejected before the provider starts it", () => { const submitted = applyAgentRuntimeEnvelope( snapshot(), diff --git a/packages/agent-bridge/src/state.ts b/packages/agent-bridge/src/state.ts index d467e93..9618738 100644 --- a/packages/agent-bridge/src/state.ts +++ b/packages/agent-bridge/src/state.ts @@ -59,22 +59,75 @@ function submissionIdOf(message: unknown): string | null { return typeof id === "string" ? id : null; } +function turnIdOf(message: unknown): string | null { + if (!message || typeof message !== "object") return null; + const id = Reflect.get(message, "overtchatTurnId"); + return typeof id === "string" && id ? id : null; +} + +function replaceTurnMessages( + messages: unknown[], + turnId: string, + incoming: unknown[], +): unknown[] { + // The provider projection owns order within a turn. Replace that block as one + // unit so neither connector nor browser delivery timing can reorder its rows. + const incomingSubmissionIds = new Set( + incoming.flatMap((message) => { + const id = submissionIdOf(message); + return id ? [id] : []; + }), + ); + const matchedIndexes: number[] = []; + const remaining = messages.filter((message, index) => { + const submissionId = submissionIdOf(message); + const matched = + turnIdOf(message) === turnId || + (submissionId !== null && incomingSubmissionIds.has(submissionId)); + if (matched) matchedIndexes.push(index); + return !matched; + }); + const insertionIndex = Math.min( + matchedIndexes.length > 0 ? Math.min(...matchedIndexes) : remaining.length, + remaining.length, + ); + return [ + ...remaining.slice(0, insertionIndex), + ...incoming, + ...remaining.slice(insertionIndex), + ]; +} + function upsertMessage(messages: unknown[], message: unknown): unknown[] { const role = roleOf(message); if (!role) return messages; const next = [...messages]; if (role === "user") { - const text = textOf(message); - const pendingIndex = next.findIndex( - (candidate) => - roleOf(candidate) === "user" && - submissionIdOf(candidate) !== null && - textOf(candidate) === text, - ); - if (pendingIndex >= 0) { - next[pendingIndex] = message; + const submissionId = submissionIdOf(message); + if (submissionId) { + const submissionIndex = next.findIndex( + (candidate) => submissionIdOf(candidate) === submissionId, + ); + if (submissionIndex >= 0) { + next[submissionIndex] = message; + return next; + } + next.push(message); return next; } + if (!submissionId) { + const text = textOf(message); + const pendingIndex = next.findIndex( + (candidate) => + roleOf(candidate) === "user" && + submissionIdOf(candidate) !== null && + textOf(candidate) === text, + ); + if (pendingIndex >= 0) { + next[pendingIndex] = message; + return next; + } + } } const id = idOf(message); if (id) { @@ -153,6 +206,14 @@ export function applyAgentRuntimeMessageEvent( messages: unknown[], event: AgentRuntimeEvent, ): unknown[] { + if ( + event.type === "overtchat_turn_update" && + typeof event.turnId === "string" && + event.turnId && + Array.isArray(event.messages) + ) { + return replaceTurnMessages(messages, event.turnId, event.messages); + } if ( event.type === "overtchat_submission" && event.message !== undefined @@ -288,8 +349,11 @@ export function applyAgentRuntimeEnvelope( }; } if ( - ["message_start", "message_update", "message_end"].includes(event.type) && - event.message !== undefined + event.type === "overtchat_turn_update" || + (["message_start", "message_update", "message_end"].includes( + event.type, + ) && + event.message !== undefined) ) { return { ...current, diff --git a/packages/agent-runtime/src/codex/client.test.ts b/packages/agent-runtime/src/codex/client.test.ts index e20a7db..61515c0 100644 --- a/packages/agent-runtime/src/codex/client.test.ts +++ b/packages/agent-runtime/src/codex/client.test.ts @@ -933,18 +933,184 @@ describe("CodexRuntimeClient", () => { expect.arrayContaining([ expect.objectContaining({ type: "turn_start", turnId: "turn-1" }), expect.objectContaining({ - type: "message_update", - message: expect.objectContaining({ - role: "user", - content: "Inspect the tests", - }), + type: "overtchat_turn_update", + turnId: "turn-1", + messages: expect.arrayContaining([ + expect.objectContaining({ + role: "user", + content: "Inspect the tests", + overtchatTurnId: "turn-1", + }), + ]), }), - expect.objectContaining({ type: "message_update" }), + expect.objectContaining({ type: "overtchat_turn_update" }), expect.objectContaining({ type: "turn_end", status: "completed" }), ]), ); }); + it("reconciles the native Codex user item with its submitted identity", async () => { + const client = new CodexRuntimeClient( + { transport: "local" }, + { executable: "codex", cwd: "/workspace" }, + ); + await client.getState(); + await client.prompt("Inspect the tests", [], { + clientMessageId: "client-message", + }); + server.emit("turn/started", { + threadId: "thread-1", + turn: { + id: "turn-1", + status: "inProgress", + startedAt: 10, + items: [], + }, + }); + server.emit("item/started", { + threadId: "thread-1", + turnId: "turn-1", + item: { + id: "provider-message", + type: "userMessage", + content: [ + { + type: "text", + text: "Inspect the tests", + text_elements: [], + }, + ], + }, + }); + + expect( + (await client.getMessages()).messages.filter( + (message) => + message && + typeof message === "object" && + Reflect.get(message, "role") === "user", + ), + ).toEqual([ + expect.objectContaining({ + id: "turn-1:user:0", + content: "Inspect the tests", + overtchatSubmissionId: "client-message", + }), + ]); + + server.emit("turn/completed", { + threadId: "thread-1", + turn: { + id: "turn-1", + status: "completed", + startedAt: 10, + completedAt: 11, + items: [ + { + id: "completed-provider-message", + type: "userMessage", + content: [ + { + type: "text", + text: "Inspect the tests", + text_elements: [], + }, + ], + }, + ], + }, + }); + + const completedUsers = (await client.getMessages()).messages.filter( + (message) => + message && + typeof message === "object" && + Reflect.get(message, "role") === "user", + ); + expect(completedUsers).toHaveLength(1); + }); + + it("emits one authoritative ordered turn after a mid-turn steer", async () => { + const client = new CodexRuntimeClient( + { transport: "local" }, + { executable: "codex", cwd: "/workspace" }, + ); + const events: Array> = []; + client.onEvent((event) => events.push(event)); + await client.getState(); + await client.prompt("write a paragraph on overtchat", [], { + clientMessageId: "client-prompt", + }); + server.emit("turn/started", { + threadId: "thread-1", + turn: { + id: "turn-1", + status: "inProgress", + startedAt: 10, + items: [], + }, + }); + server.emit("item/started", { + threadId: "thread-1", + turnId: "turn-1", + item: { + id: "assistant-1", + type: "agentMessage", + text: "First paragraph", + phase: "final_answer", + }, + }); + await client.steer("2 more now", [], { + clientMessageId: "client-steer", + }); + server.emit("item/started", { + threadId: "thread-1", + turnId: "turn-1", + item: { + id: "provider-steer", + type: "userMessage", + content: [ + { + type: "text", + text: "2 more now", + text_elements: [], + }, + ], + }, + }); + server.emit("item/started", { + threadId: "thread-1", + turnId: "turn-1", + item: { + id: "assistant-2", + type: "agentMessage", + text: "Two more paragraphs", + phase: "final_answer", + }, + }); + + const update = events.findLast( + (event) => event.type === "overtchat_turn_update", + ); + const messages = Array.isArray(update?.messages) ? update.messages : []; + expect( + messages.map((message) => + message && typeof message === "object" + ? [ + Reflect.get(message, "role"), + Reflect.get(message, "id"), + Reflect.get(message, "overtchatTurnId"), + ] + : null, + ), + ).toEqual([ + ["user", "turn-1:user:0", "turn-1"], + ["assistant", "assistant-1", "turn-1"], + ["user", "turn-1:user:1", "turn-1"], + ["assistant", "assistant-2", "turn-1"], + ]); + }); + it("preserves streamed work when the completed turn only includes the final answer", async () => { const client = new CodexRuntimeClient( { transport: "local" }, @@ -1131,7 +1297,7 @@ describe("CodexRuntimeClient", () => { await expect(client.getMessages()).resolves.toEqual({ messages: expect.arrayContaining([ expect.objectContaining({ - id: "user-history", + id: "turn-history:user:0", role: "user", content: "Resume this thread", }), @@ -1177,6 +1343,75 @@ describe("CodexRuntimeClient", () => { ); }); + it("restores an active turn and appends steering input after its existing user message", async () => { + server.threadReads.set("thread-1", { + id: "thread-1", + cwd: "/workspace", + preview: "Original prompt", + path: "/tmp/thread-1.jsonl", + name: null, + createdAt: 1, + updatedAt: 2, + turns: [ + { + id: "turn-active", + status: "inProgress", + startedAt: 10, + completedAt: null, + items: [ + { + id: "hydrated-provider-user", + type: "userMessage", + content: [ + { + type: "text", + text: "Original prompt", + text_elements: [], + }, + ], + }, + ], + }, + ], + }); + const client = new CodexRuntimeClient( + { transport: "local" }, + { + executable: "codex", + cwd: "/workspace", + resume: { + providerSessionId: "thread-1", + providerSessionPath: "/tmp/thread-1.jsonl", + }, + }, + ); + + await expect(client.getState()).resolves.toMatchObject({ + isStreaming: true, + }); + await client.steer("Steering follow-up", [], { + clientMessageId: "client-steer", + }); + + const users = (await client.getMessages()).messages.filter( + (message) => + message && + typeof message === "object" && + Reflect.get(message, "role") === "user", + ); + expect(users).toEqual([ + expect.objectContaining({ + id: "turn-active:user:0", + content: "Original prompt", + }), + expect.objectContaining({ + id: "turn-active:user:1", + content: "Steering follow-up", + overtchatSubmissionId: "client-steer", + }), + ]); + }); + it("rehydrates persisted subagent activity after resuming", async () => { server.threadReads.set("thread-1", { id: "thread-1", @@ -1339,7 +1574,7 @@ describe("CodexRuntimeClient", () => { await expect(client.getMessages()).resolves.toEqual({ messages: expect.arrayContaining([ expect.objectContaining({ - id: "user-history", + id: "turn-history:user:0", content: "Resume this thread", }), expect.objectContaining({ @@ -1836,7 +2071,7 @@ describe("CodexRuntimeClient", () => { await client.getState(); await expect( - client.forkSession("user-history", "edit"), + client.forkSession("turn-history:user:0", "edit"), ).resolves.toEqual({ session: { providerSessionId: "thread-fork", @@ -1865,7 +2100,7 @@ describe("CodexRuntimeClient", () => { }); await expect(client.getMessages()).resolves.toMatchObject({ messages: expect.arrayContaining([ - expect.objectContaining({ id: "user-history", role: "user" }), + expect.objectContaining({ id: "turn-history:user:0", role: "user" }), expect.objectContaining({ id: "commentary-history", role: "assistant", @@ -1952,7 +2187,7 @@ describe("CodexRuntimeClient", () => { }); await expect(client.getMessages()).resolves.toMatchObject({ messages: expect.arrayContaining([ - expect.objectContaining({ id: "user-history", role: "user" }), + expect.objectContaining({ id: "turn-history:user:0", role: "user" }), expect.objectContaining({ id: "commentary-history", role: "assistant", @@ -1984,7 +2219,9 @@ describe("CodexRuntimeClient", () => { ); await client.getState(); - await expect(client.forkSession("user-history", "edit")).rejects.toThrow( + await expect( + client.forkSession("turn-history:user:0", "edit"), + ).rejects.toThrow( "Editing the first message requires a newer Codex installation.", ); }); diff --git a/packages/agent-runtime/src/codex/client.ts b/packages/agent-runtime/src/codex/client.ts index cfabe29..4bf8b56 100644 --- a/packages/agent-runtime/src/codex/client.ts +++ b/packages/agent-runtime/src/codex/client.ts @@ -116,6 +116,7 @@ type PendingInteraction = type KnownUserInput = { id: string; text: string; + afterItemId: string | null; images: Array<{ uploadId: string; filename: string; @@ -130,10 +131,26 @@ function textInput(text: string) { function isSyntheticUserItem(item: CodexItem): boolean { return ( item.type === "userMessage" && - item.id.startsWith("overtchat:codex-user:") + item.overtchatSyntheticUserInput === true ); } +function codexUserMessageId(turnId: string, userIndex: number): string { + return `${turnId}:user:${userIndex}`; +} + +function userItemForMessageId( + turn: CodexTurn, + messageId: string, +): CodexItem | undefined { + let userIndex = 0; + for (const item of turn.items) { + if (item.type !== "userMessage") continue; + if (codexUserMessageId(turn.id, userIndex++) === messageId) return item; + } + return undefined; +} + function itemText(item: CodexItem): string { const content = item.content; if (typeof content === "string") return content; @@ -426,8 +443,10 @@ function canonicalTurnMessages(turn: CodexTurn): unknown[] { assistantOrders.push(order); }; + let userIndex = 0; for (const [itemIndex, item] of turn.items.entries()) { if (item.type === "userMessage") { + const messageId = codexUserMessageId(turn.id, userIndex++); const text = itemText(item); const images = Array.isArray(item.overtchatImages) ? item.overtchatImages.flatMap((value) => { @@ -448,9 +467,11 @@ function canonicalTurnMessages(turn: CodexTurn): unknown[] { }) : []; if (text || images.length > 0) { + const submissionId = stringOf(item, "overtchatSubmissionId"); messages.push({ - id: item.id, + id: messageId, role: "user", + overtchatTurnId: turn.id, content: images.length > 0 ? [ @@ -459,6 +480,9 @@ function canonicalTurnMessages(turn: CodexTurn): unknown[] { ] : text, timestamp: startedAt + itemIndex, + ...(submissionId + ? { overtchatSubmissionId: submissionId } + : {}), }); } continue; @@ -610,6 +634,7 @@ function canonicalTurnMessages(turn: CodexTurn): unknown[] { messages.push({ id: item.id, role: "custom", + overtchatTurnId: turn.id, display: true, content: "Conversation context compacted.", timestamp: startedAt + itemIndex, @@ -628,6 +653,7 @@ function canonicalTurnMessages(turn: CodexTurn): unknown[] { results.push({ id: `${item.id}:result`, role: "toolResult", + overtchatTurnId: turn.id, toolCallId: item.id, toolName: tool.name, content: [{ type: "text", text: tool.output }], @@ -660,6 +686,7 @@ function canonicalTurnMessages(turn: CodexTurn): unknown[] { messages.push({ id: `${turn.id}:footer`, role: "turnFooter", + overtchatTurnId: turn.id, messageId: assistantContent.length > 0 ? `${turn.id}:assistant` : null, content: assistantContent @@ -977,6 +1004,7 @@ export class CodexRuntimeClient implements AgentRuntimeClient { this.assertInteractive(); if (!this.activeTurnId) throw new Error("Codex has no active turn to steer."); const turnId = this.activeTurnId; + const afterItemId = this.lastNativeItemId(turnId); const response = await this.server.request("turn/steer", { threadId: this.thread!.id, expectedTurnId: turnId, @@ -987,7 +1015,12 @@ export class CodexRuntimeClient implements AgentRuntimeClient { }); this.rememberUserInput( turnId, - this.createKnownUserInput(message, images, options.clientMessageId), + this.createKnownUserInput( + message, + images, + options.clientMessageId, + afterItemId, + ), ); return response; } @@ -1298,9 +1331,7 @@ export class CodexRuntimeClient implements AgentRuntimeClient { .map((turn, turnIndex) => ({ turn, turnIndex, - item: turn.items.find( - (item) => item.id === messageId && item.type === "userMessage", - ), + item: userItemForMessageId(turn, messageId), })) .find((candidate) => candidate.item); if (!target?.item) { @@ -1424,8 +1455,10 @@ export class CodexRuntimeClient implements AgentRuntimeClient { this.turns.clear(); this.subagentOwners.clear(); this.pendingSubagentNotifications.clear(); + this.activeTurnId = null; for (const turn of this.thread.turns) { this.turns.set(turn.id, turn); + if (turn.status === "inProgress") this.activeTurnId = turn.id; for (const item of turn.items) { this.registerSubagentOwners(turn.id, item); } @@ -1804,7 +1837,9 @@ export class CodexRuntimeClient implements AgentRuntimeClient { const completedTurn = this.withKnownUserInputs( parseCodexTurn(data?.turn), ); - const turn = this.reconcileCompletedTurn(completedTurn); + const turn = this.withKnownUserInputs( + this.reconcileCompletedTurn(completedTurn), + ); this.turns.set(turn.id, turn); this.emitTurn(turn); if (this.activeTurnId === turn.id) this.activeTurnId = null; @@ -2110,12 +2145,14 @@ export class CodexRuntimeClient implements AgentRuntimeClient { text: string, images: readonly ResolvedAgentImage[] = [], clientMessageId?: string, + afterItemId: string | null = null, ): KnownUserInput { return { id: clientMessageId ?? `overtchat:codex-user:${++this.nextUserInputId}`, text, + afterItemId, images: images.map(({ uploadId, filename, mediaType }) => ({ uploadId, filename, @@ -2197,15 +2234,22 @@ export class CodexRuntimeClient implements AgentRuntimeClient { this.emitTurn(next); } + private lastNativeItemId(turnId: string): string | null { + const items = this.turns + .get(turnId) + ?.items.filter((item) => !isSyntheticUserItem(item)); + return items?.at(-1)?.id ?? null; + } + private withKnownUserInputs(turn: CodexTurn): CodexTurn { const known = this.knownUserInputs.get(turn.id); if (!known?.length) return turn; - const items = turn.items.filter((item) => !isSyntheticUserItem(item)); - const nativeUsers = items.flatMap((item, index) => + const nativeItems = turn.items.filter((item) => !isSyntheticUserItem(item)); + const nativeUsers = nativeItems.flatMap((item, index) => item.type === "userMessage" ? [{ index, text: itemText(item) }] : [], ); const claimedNativeUsers = new Set(); - const synthetic: CodexItem[] = []; + const placements: Array<{ input: KnownUserInput; item: CodexItem }> = []; for (const input of known) { const native = nativeUsers.find( (candidate) => @@ -2214,26 +2258,54 @@ export class CodexRuntimeClient implements AgentRuntimeClient { ); if (native) { claimedNativeUsers.add(native.index); - if (input.images.length > 0) { - items[native.index] = { - ...items[native.index], - overtchatImages: input.images, - }; - } + placements.push({ + input, + item: { + ...nativeItems[native.index], + overtchatSubmissionId: input.id, + ...(input.images.length > 0 + ? { overtchatImages: input.images } + : {}), + }, + }); continue; } - synthetic.push({ - id: input.id, - type: "userMessage", - content: textInput(input.text), - ...(input.images.length > 0 - ? { overtchatImages: input.images } - : {}), + placements.push({ + input, + item: { + id: input.id, + type: "userMessage", + content: textInput(input.text), + overtchatSubmissionId: input.id, + overtchatSyntheticUserInput: true, + ...(input.images.length > 0 + ? { overtchatImages: input.images } + : {}), + }, }); } - return synthetic.length === 0 && items.length === turn.items.length - ? turn - : { ...turn, items: [...synthetic, ...items] }; + const items = nativeItems.filter( + (_item, index) => !claimedNativeUsers.has(index), + ); + const lastPlacementByAnchor = new Map(); + for (const { input, item } of placements) { + const anchorKey = input.afterItemId ?? "overtchat:turn-start"; + const previousAtAnchor = lastPlacementByAnchor.get(anchorKey); + const anchorIndex = previousAtAnchor + ? items.indexOf(previousAtAnchor) + : input.afterItemId + ? items.findIndex((candidate) => candidate.id === input.afterItemId) + : -1; + const insertionIndex = + anchorIndex >= 0 + ? anchorIndex + 1 + : input.afterItemId === null + ? 0 + : items.length; + items.splice(insertionIndex, 0, item); + lastPlacementByAnchor.set(anchorKey, item); + } + return { ...turn, items }; } private reconcileCompletedTurn(completed: CodexTurn): CodexTurn { @@ -2243,8 +2315,16 @@ export class CodexRuntimeClient implements AgentRuntimeClient { const indexById = new Map( items.map((item, index) => [item.id, index] as const), ); + const streamedUserIndexes = items.flatMap((item, index) => + item.type === "userMessage" ? [index] : [], + ); + let completedUserIndex = 0; for (const item of completed.items) { - const index = indexById.get(item.id); + const userIndex = + item.type === "userMessage" ? completedUserIndex++ : null; + const index = + indexById.get(item.id) ?? + (userIndex === null ? undefined : streamedUserIndexes[userIndex]); if (index === undefined) { indexById.set(item.id, items.length); items.push(item); @@ -2292,9 +2372,10 @@ export class CodexRuntimeClient implements AgentRuntimeClient { const index = turn.items.findIndex((candidate) => candidate.id === id); if (index >= 0) turn.items[index] = item; else turn.items.push(item); - this.turns.set(turnId, turn); + const reconciled = this.withKnownUserInputs(turn); + this.turns.set(turnId, reconciled); this.registerSubagentOwners(turnId, item); - this.emitTurn(turn); + this.emitTurn(reconciled); } private registerSubagentOwners(turnId: string, item: CodexItem): void { @@ -2608,9 +2689,11 @@ export class CodexRuntimeClient implements AgentRuntimeClient { } private emitTurn(turn: CodexTurn): void { - for (const message of canonicalTurnMessages(turn)) { - this.emit({ type: "message_update", message }); - } + this.emit({ + type: "overtchat_turn_update", + turnId: turn.id, + messages: canonicalTurnMessages(turn), + }); } private updateTokenUsage(data: UnknownRecord | null): void { diff --git a/packages/agent-runtime/src/runtime/registry.test.ts b/packages/agent-runtime/src/runtime/registry.test.ts index 294b382..f222855 100644 --- a/packages/agent-runtime/src/runtime/registry.test.ts +++ b/packages/agent-runtime/src/runtime/registry.test.ts @@ -1,9 +1,14 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { AgentProviderId } from "@overtchat/agent-bridge"; +import type { AgentRuntimeEvent } from "@overtchat/agent-runtime/providers/types"; const mocks = vi.hoisted(() => ({ prompt: vi.fn(), + steer: vi.fn(), + abort: vi.fn(), stop: vi.fn(), saveQueue: vi.fn(), + eventSubscriber: null as ((event: AgentRuntimeEvent) => void) | null, })); const stats = { @@ -25,14 +30,17 @@ const stats = { }; vi.mock("@overtchat/agent-runtime/providers/registry", () => ({ - agentProviderAdapter: () => ({ - provider: "codex", + agentProviderAdapter: (provider: AgentProviderId) => ({ + provider, capabilities: { steer: true }, probeConnection: vi.fn(), probeTarget: vi.fn(), listWorkspaceSessions: vi.fn(), startSession: () => ({ - onEvent: vi.fn(), + onEvent: vi.fn((subscriber: (event: AgentRuntimeEvent) => void) => { + mocks.eventSubscriber = subscriber; + return vi.fn(); + }), getState: vi.fn().mockResolvedValue({ isStreaming: false, sessionId: "provider-session", @@ -46,6 +54,8 @@ vi.mock("@overtchat/agent-runtime/providers/registry", () => ({ getAvailableThinkingLevels: vi.fn().mockResolvedValue([]), getCommands: vi.fn().mockResolvedValue([]), prompt: mocks.prompt, + steer: mocks.steer, + abort: mocks.abort, stop: mocks.stop, }), sessionIdentity: () => ({ @@ -54,7 +64,10 @@ vi.mock("@overtchat/agent-runtime/providers/registry", () => ({ sessionName: null, }), createEventClassifier: () => ({ - classify: () => ({ started: false, ended: false }), + classify: (event: AgentRuntimeEvent) => ({ + started: event.type === "agent_start", + terminal: event.type === "agent_end", + }), reset: vi.fn(), }), commandsFromEvent: () => null, @@ -65,12 +78,15 @@ vi.mock("@overtchat/agent-runtime/providers/registry", () => ({ import { AgentRuntimeRegistry } from "./registry.js"; -describe("agent runtime queue recovery", () => { +describe("agent runtime", () => { beforeEach(() => { vi.clearAllMocks(); mocks.prompt.mockResolvedValue({ accepted: true }); + mocks.steer.mockResolvedValue({ accepted: true }); + mocks.abort.mockResolvedValue({ interrupted: true }); mocks.stop.mockResolvedValue(undefined); mocks.saveQueue.mockResolvedValue(undefined); + mocks.eventSubscriber = null; }); it("resubmits a journaled queue item with its original message identity", async () => { @@ -110,4 +126,508 @@ describe("agent runtime queue recovery", () => { }); await registry.stopAll(); }); + + it("drains queued messages once in FIFO order as turns become idle", async () => { + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "Start the task" }, + "initial-message", + ); + await runtime.command( + { type: "queue", message: "First follow-up" }, + "queued-first", + ); + await runtime.command( + { type: "queue", message: "Second follow-up" }, + "queued-second", + ); + + mocks.eventSubscriber?.({ type: "agent_end", messages: [] }); + await vi.waitFor(() => expect(mocks.prompt).toHaveBeenCalledTimes(2)); + expect(mocks.prompt.mock.calls[1]).toEqual([ + "First follow-up", + undefined, + { clientMessageId: "queued-first" }, + ]); + expect(runtime.snapshot().queuedMessages).toEqual([ + expect.objectContaining({ + id: "queued-second", + status: "pending", + }), + ]); + + mocks.eventSubscriber?.({ type: "agent_end", messages: [] }); + await vi.waitFor(() => expect(mocks.prompt).toHaveBeenCalledTimes(3)); + expect(mocks.prompt.mock.calls[2]).toEqual([ + "Second follow-up", + undefined, + { clientMessageId: "queued-second" }, + ]); + expect(runtime.snapshot().queuedMessages).toEqual([]); + await registry.stopAll(); + }); + + it("restores a queued message when steering rejects it", async () => { + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "Start the task" }, + "initial-message", + ); + await runtime.command( + { type: "queue", message: "Try this approach" }, + "queued-message", + ); + mocks.steer.mockRejectedValueOnce(new Error("Provider rejected steer")); + + await expect( + runtime.command({ + type: "steer_queued_message", + id: "queued-message", + }), + ).rejects.toThrow("Provider rejected steer"); + expect(runtime.snapshot().queuedMessages).toEqual([ + expect.objectContaining({ + id: "queued-message", + status: "pending", + }), + ]); + expect( + runtime + .snapshot() + .messages.some( + (message) => + message && + typeof message === "object" && + Reflect.get(message, "overtchatSubmissionId") === + "queued-message", + ), + ).toBe(false); + await registry.stopAll(); + }); + + it("stops the active turn without starting a replacement prompt", async () => { + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "Start the task" }, + "initial-message", + ); + await runtime.command({ type: "abort" }); + + expect(mocks.abort).toHaveBeenCalledOnce(); + expect(mocks.prompt).toHaveBeenCalledOnce(); + expect(runtime.snapshot().status).toBe("idle"); + await registry.stopAll(); + }); + + it.each(["pi", "omp"] as const)( + "uses the shared durable queue and steering path for %s", + async (provider) => { + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider, + target: { transport: "local" }, + executable: provider, + cwd: "/workspace", + sessionId: `session-${provider}`, + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "Start the task" }, + `initial-${provider}`, + ); + await runtime.command( + { type: "queue", message: "Replace the approach" }, + `queued-${provider}`, + ); + await runtime.command({ + type: "steer_queued_message", + id: `queued-${provider}`, + }); + + expect(mocks.steer).toHaveBeenCalledWith( + "Replace the approach", + undefined, + { clientMessageId: `queued-${provider}` }, + ); + expect(runtime.snapshot().provider).toBe(provider); + expect(runtime.snapshot().queuedMessages).toEqual([]); + await registry.stopAll(); + }, + ); + + it("publishes one canonical user message when a provider echoes before accepting", async () => { + mocks.prompt.mockImplementation(async () => { + mocks.eventSubscriber?.({ + type: "message_start", + message: { + id: "provider-message", + role: "user", + content: "Continue the task", + timestamp: 123, + }, + }); + return { accepted: true }; + }); + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "Continue the task" }, + "client-message", + ); + + expect(runtime.snapshot().messages).toEqual([ + { + id: "provider-message", + role: "user", + content: "Continue the task", + timestamp: 123, + }, + ]); + await registry.stopAll(); + }); + + it("keeps an acknowledged prompt when the transport rejects afterward", async () => { + mocks.prompt.mockImplementation(async (_message, _images, options) => { + mocks.eventSubscriber?.({ + type: "message_start", + message: { + id: "provider-message", + role: "user", + content: "Continue the task", + timestamp: 123, + overtchatSubmissionId: options?.clientMessageId, + }, + }); + throw new Error("Transport closed after provider acceptance"); + }); + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await expect( + runtime.command( + { type: "prompt", message: "Continue the task" }, + "client-message", + ), + ).resolves.toEqual({ accepted: true, providerAcknowledged: true }); + expect(runtime.snapshot().messages).toEqual([ + expect.objectContaining({ + id: "provider-message", + content: "Continue the task", + overtchatSubmissionId: "client-message", + }), + ]); + expect(runtime.snapshot().error).toBeUndefined(); + await registry.stopAll(); + }); + + it("removes the canonical user message when the provider rejects it", async () => { + mocks.prompt.mockRejectedValue(new Error("Provider rejected the prompt")); + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await expect( + runtime.command( + { type: "prompt", message: "Continue the task" }, + "client-message", + ), + ).rejects.toThrow("Provider rejected the prompt"); + expect(runtime.snapshot().messages).toEqual([]); + await registry.stopAll(); + }); + + it("reconciles a native steering echo with the canonical steer message", async () => { + mocks.prompt.mockImplementation(async () => { + mocks.eventSubscriber?.({ + type: "message_start", + message: { + id: "provider-prompt", + role: "user", + content: "Start the task", + timestamp: 123, + }, + }); + return { accepted: true }; + }); + mocks.steer.mockImplementation(async () => { + mocks.eventSubscriber?.({ + type: "message_start", + message: { + id: "provider-steer", + role: "user", + content: "Use the other approach", + timestamp: 124, + }, + }); + return { accepted: true }; + }); + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "Start the task" }, + "client-prompt", + ); + await runtime.command( + { type: "queue", message: "Use the other approach" }, + "client-steer", + ); + await runtime.command({ + type: "steer_queued_message", + id: "client-steer", + }); + + expect(runtime.snapshot().messages).toEqual([ + expect.objectContaining({ + id: "provider-prompt", + content: "Start the task", + }), + expect.objectContaining({ + id: "provider-steer", + content: "Use the other approach", + }), + ]); + await registry.stopAll(); + }); + + it("keeps the connector snapshot in canonical turn order after a steer race", async () => { + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "write a paragraph on overtchat" }, + "client-prompt", + ); + await runtime.command( + { type: "queue", message: "2 more now" }, + "client-steer", + ); + await runtime.command({ + type: "steer_queued_message", + id: "client-steer", + }); + mocks.eventSubscriber?.({ + type: "overtchat_turn_update", + turnId: "turn-1", + messages: [ + { + id: "turn-1:user:0", + role: "user", + content: "write a paragraph on overtchat", + overtchatSubmissionId: "client-prompt", + overtchatTurnId: "turn-1", + }, + { + id: "assistant-1", + role: "assistant", + content: [{ type: "text", text: "First paragraph" }], + overtchatTurnId: "turn-1", + }, + { + id: "turn-1:user:1", + role: "user", + content: "2 more now", + overtchatSubmissionId: "client-steer", + overtchatTurnId: "turn-1", + }, + { + id: "assistant-2", + role: "assistant", + content: [{ type: "text", text: "Two more paragraphs" }], + overtchatTurnId: "turn-1", + }, + { + id: "turn-1:footer", + role: "turnFooter", + content: "First paragraph\n\nTwo more paragraphs", + overtchatTurnId: "turn-1", + }, + ], + }); + + expect( + runtime.snapshot().messages.map((message) => + message && typeof message === "object" + ? [Reflect.get(message, "role"), Reflect.get(message, "id")] + : null, + ), + ).toEqual([ + ["user", "turn-1:user:0"], + ["assistant", "assistant-1"], + ["user", "turn-1:user:1"], + ["assistant", "assistant-2"], + ["turnFooter", "turn-1:footer"], + ]); + await registry.stopAll(); + }); + + it("keeps an acknowledged steer when the transport rejects afterward", async () => { + mocks.steer.mockImplementation(async (_message, _images, options) => { + mocks.eventSubscriber?.({ + type: "overtchat_turn_update", + turnId: "turn-1", + messages: [ + { + id: "provider-steer", + role: "user", + content: "Use the other approach", + timestamp: 124, + overtchatSubmissionId: options?.clientMessageId, + overtchatTurnId: "turn-1", + }, + ], + }); + throw new Error("Transport closed after provider acceptance"); + }); + const registry = new AgentRuntimeRegistry({ + resolveImages: async () => [], + }); + const runtime = await registry.getOrStart({ + connectionId: "connection", + workspaceId: "workspace", + provider: "codex", + target: { transport: "local" }, + executable: "codex", + cwd: "/workspace", + sessionId: "session", + providerSessionId: "provider-session", + providerSessionPath: "/sessions/provider-session.jsonl", + }); + + await runtime.command( + { type: "prompt", message: "Start the task" }, + "client-prompt", + ); + await expect( + runtime + .command( + { type: "queue", message: "Use the other approach" }, + "client-steer", + ) + .then(() => + runtime.command({ + type: "steer_queued_message", + id: "client-steer", + }), + ), + ).resolves.toEqual({ accepted: true, providerAcknowledged: true }); + expect(runtime.snapshot().messages).toEqual([ + expect.objectContaining({ content: "Start the task" }), + expect.objectContaining({ + id: "provider-steer", + content: "Use the other approach", + overtchatSubmissionId: "client-steer", + }), + ]); + expect(runtime.snapshot().error).toBeUndefined(); + await registry.stopAll(); + }); }); diff --git a/packages/agent-runtime/src/runtime/registry.ts b/packages/agent-runtime/src/runtime/registry.ts index d240843..2b1f58c 100644 --- a/packages/agent-runtime/src/runtime/registry.ts +++ b/packages/agent-runtime/src/runtime/registry.ts @@ -51,6 +51,7 @@ type PendingSubmission = { id: string; message: string; published: boolean; + providerAcknowledged: boolean; }; export type AgentWorkspaceDescriptor = RuntimeOwner & { @@ -184,6 +185,26 @@ function messageRole(message: unknown): string | null { : null; } +function messageSubmissionId(message: unknown): string | null { + if (!message || typeof message !== "object") return null; + const id = Reflect.get(message, "overtchatSubmissionId"); + return typeof id === "string" && id ? id : null; +} + +function eventUserMessages(event: AgentRuntimeEvent): unknown[] { + if ( + event.type === "overtchat_turn_update" && + Array.isArray(event.messages) + ) { + return event.messages.filter((message) => messageRole(message) === "user"); + } + return ["message_start", "message_update", "message_end"].includes( + event.type, + ) && messageRole(event.message) === "user" + ? [event.message] + : []; +} + export class AgentSessionRuntime { private readonly subscribers = new Set(); private readonly replay: AgentRuntimeEnvelope[] = []; @@ -264,17 +285,15 @@ export class AgentSessionRuntime { const classification = this.eventClassifier.classify(event); this.messages = applyAgentRuntimeMessageEvent(this.messages, event); this.state = applyAgentRuntimeStateEvent(this.state, event); - if ( - ["message_start", "message_update", "message_end"].includes( - event.type, - ) && - messageRole(event.message) === "user" - ) { - const text = messageText(event.message); - const submission = [...this.pendingSubmissions.values()].find( - (candidate) => candidate.message.trim() === text, - ); - if (submission) this.pendingSubmissions.delete(submission.id); + for (const userMessage of eventUserMessages(event)) { + const text = messageText(userMessage); + const submissionId = messageSubmissionId(userMessage); + const submission = submissionId + ? this.pendingSubmissions.get(submissionId) + : [...this.pendingSubmissions.values()].find( + (candidate) => candidate.message.trim() === text, + ); + if (submission) submission.providerAcknowledged = true; } if (event.type === "process_exit") { this.stopped = true; @@ -340,20 +359,26 @@ export class AgentSessionRuntime { event.type === "rpc_error" && typeof event.error === "string" ) { - this.error = event.error; if (event.command === "prompt" && this.promptAwaitingStart) { this.promptAwaitingStart = false; const submission = this.promptSubmissionId ? this.pendingSubmissions.get(this.promptSubmissionId) : undefined; - if (submission?.published) { - this.rejectSubmission(submission.id); + if (submission?.providerAcknowledged) { + this.pendingSubmissions.delete(submission.id); + } else { + this.error = event.error; + if (submission?.published) { + this.rejectSubmission(submission.id); + } + settleRejectedPrompt = true; } if (this.promptSubmissionId) { this.pendingSubmissions.delete(this.promptSubmissionId); } this.promptSubmissionId = undefined; - settleRejectedPrompt = true; + } else { + this.error = event.error; } } this.publish({ type: "runtime_event", data: event }); @@ -411,6 +436,9 @@ export class AgentSessionRuntime { } if (classification.terminal) { this.promptAwaitingStart = false; + for (const submission of this.pendingSubmissions.values()) { + submission.providerAcknowledged = true; + } this.pendingSubmissions.clear(); this.promptSubmissionId = undefined; this.clearPendingInteraction(); @@ -507,12 +535,6 @@ export class AgentSessionRuntime { ); case "abort": return this.abortActiveRun(); - case "steer": - return this.submitSteer( - command.message, - command.images ?? [], - clientMessageId, - ); case "queue": return this.enqueueMessage( command.message, @@ -839,11 +861,13 @@ export class AgentSessionRuntime { const submission = { id: submissionId, message, - published: false, + published: true, + providerAcknowledged: false, }; this.pendingSubmissions.set(submission.id, submission); this.promptSubmissionId = submission.id; this.publishStatus(); + this.publishSubmission(submission.id, submission.message, images); try { const result = images.length > 0 @@ -853,13 +877,20 @@ export class AgentSessionRuntime { : await this.client.prompt(message, undefined, { clientMessageId: submissionId, }); - if (this.pendingSubmissions.get(submission.id) === submission) { - submission.published = true; - this.publishSubmission(submission.id, submission.message, images); + this.pendingSubmissions.delete(submission.id); + if (this.promptSubmissionId === submission.id) { + this.promptSubmissionId = undefined; } return result; } catch (error) { this.promptAwaitingStart = false; + if (submission.providerAcknowledged) { + this.pendingSubmissions.delete(submission.id); + if (this.promptSubmissionId === submission.id) { + this.promptSubmissionId = undefined; + } + return { accepted: true, providerAcknowledged: true }; + } if (this.pendingSubmissions.get(submission.id) === submission) { if (submission.published) this.rejectSubmission(submission.id); this.pendingSubmissions.delete(submission.id); @@ -921,9 +952,11 @@ export class AgentSessionRuntime { const submission: PendingSubmission = { id: submissionId, message, - published: false, + published: true, + providerAcknowledged: false, }; this.pendingSubmissions.set(submission.id, submission); + this.publishSubmission(submission.id, submission.message, images); const request = images.length > 0 ? this.client.steer(message, images, { @@ -933,10 +966,7 @@ export class AgentSessionRuntime { clientMessageId: submissionId, }); return request.then((result) => { - if (this.pendingSubmissions.get(submission.id) === submission) { - submission.published = true; - this.publishSubmission(submission.id, submission.message, images); - } + this.pendingSubmissions.delete(submission.id); return result; }); }; @@ -947,6 +977,10 @@ export class AgentSessionRuntime { ) .catch((error) => { const submission = this.pendingSubmissions.get(submissionId); + if (submission?.providerAcknowledged) { + this.pendingSubmissions.delete(submission.id); + return { accepted: true, providerAcknowledged: true }; + } if (submission) { if (submission.published) this.rejectSubmission(submission.id); this.pendingSubmissions.delete(submission.id); @@ -994,7 +1028,9 @@ export class AgentSessionRuntime { .finally(() => { if (this.abortPromise !== settled) return; this.abortPromise = null; - if (this.status === "idle") void this.drainQueuedMessage(); + if (this.status === "idle") { + void this.drainQueuedMessage(); + } }); this.abortPromise = settled; return settled;