diff --git a/client/src/api.ts b/client/src/api.ts index 9b584c08..cc5d20cf 100644 --- a/client/src/api.ts +++ b/client/src/api.ts @@ -506,20 +506,25 @@ export async function steerSession( projectId: string, sessionId: string, message: string, - worktreeId?: string -): Promise { + worktreeId?: string, + queuedMessageId?: string +): Promise<{ disposition: "steered" | "queued"; message?: QueuedMessage }> { const res = await fetch( `${BASE}/projects/${projectId}/sessions/${sessionId}/steer${withWorktree(worktreeId)}`, { method: "POST", headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ message }), + body: JSON.stringify({ message, queuedMessageId }), } ); if (!res.ok) { const body = await res.json().catch(() => ({})) as { error?: string }; throw new Error(body.error ?? "Failed to steer session"); } + return await res.json() as { + disposition: "steered" | "queued"; + message?: QueuedMessage; + }; } /** A message enqueued to run after the active turn completes (see issue #113). */ diff --git a/client/src/pages/SessionView.tsx b/client/src/pages/SessionView.tsx index dc4082e0..b29e9412 100644 --- a/client/src/pages/SessionView.tsx +++ b/client/src/pages/SessionView.tsx @@ -2929,6 +2929,9 @@ export function SessionView({ // draining (when there's no own SSE). The event poller keys off this so it // engages once our SSE closes but the run continues server-side (#113). const [ownStreamActive, setOwnStreamActive] = useState(false); + // Codex ends native steering at its terminal event, before the enclosing + // SSE emits `done`. Keep that narrower lifecycle separate from `streaming`. + const nativeSteerAvailableRef = useRef(false); // Messages enqueued while a run is streaming (replayed one-at-a-time on // clean completion). The server is the source of truth; this mirrors it // for rendering. See issue #113. @@ -4633,6 +4636,7 @@ export function SessionView({ } else if (data.type === "anita_event") { const adaEvent = data.event; if (adaEvent.type === "run.started") { + nativeSteerAvailableRef.current = true; detectedSessionId = adaEvent.sessionId; attachToSession(adaEvent.sessionId); } else if (adaEvent.type === "assistant.text") { @@ -4780,6 +4784,7 @@ export function SessionView({ ]); } } else if (adaEvent.type === "run.failed") { + nativeSteerAvailableRef.current = false; runFailed = true; if (isVisible()) { setStreamItems((prev) => [ @@ -4788,6 +4793,7 @@ export function SessionView({ ]); } } else if (adaEvent.type === "run.completed") { + nativeSteerAvailableRef.current = false; if (adaEvent.sessionId) detectedSessionId = adaEvent.sessionId; if ((adaEvent.status === "max_iterations" || adaEvent.stopReason === "max_turns") && isVisible()) { setStreamItems((prev) => [ @@ -5157,7 +5163,7 @@ export function SessionView({ promotedId = first.id; } - if (promotedId) { + if (promotedId && !providerUsesNativeSteering) { try { await removeSessionQueuedMessage(projectId, targetSessionId, promotedId); setQueue((prev) => prev.filter((m) => m.id !== promotedId)); @@ -5166,15 +5172,40 @@ export function SessionView({ } } - setMessage(""); - // The draft text was just consumed by the steer. Clear it explicitly: the - // write-through effect is skipped while steerInProgress is set. - clearComposerDraft(composerDraftKey); - setStreamItems((prev) => [...prev, { type: "user_message", text: steerText, at: Date.now() }]); - if (providerUsesNativeSteering) { try { - await steerSession(projectId, targetSessionId, steerText, worktreeId); + // Once the terminal event is visible, use the ordinary durable queue + // directly. The route independently performs the same fallback for + // requests already racing the event. + if (!nativeSteerAvailableRef.current) { + await handleEnqueue(); + return; + } + const result = await steerSession( + projectId, + targetSessionId, + steerText, + worktreeId, + promotedId ?? undefined + ); + if (promotedId && result.disposition === "steered") { + setQueue((prev) => prev.filter((item) => item.id !== promotedId)); + } else if (result.disposition === "queued") { + if (!promotedId && result.message) { + setQueue((prev) => [...prev, result.message!]); + } + if (!promotedId) { + setMessage(""); + clearComposerDraft(composerDraftKey); + } + return; + } + setMessage(""); + clearComposerDraft(composerDraftKey); + setStreamItems((prev) => [ + ...prev, + { type: "user_message", text: steerText, at: Date.now() }, + ]); } catch (err) { setStreamItems((prev) => [ ...prev, @@ -5184,6 +5215,13 @@ export function SessionView({ return; } + setMessage(""); + clearComposerDraft(composerDraftKey); + setStreamItems((prev) => [ + ...prev, + { type: "user_message", text: steerText, at: Date.now() }, + ]); + // Emulated steer (Claude/Anita): stop the current run; the stream's // `done` handler resumes with the steer text once the process exits. // The composer stays disabled until the resumed run starts so a second diff --git a/server/lib/__tests__/codex-steer.test.ts b/server/lib/__tests__/codex-steer.test.ts new file mode 100644 index 00000000..6555a07e --- /dev/null +++ b/server/lib/__tests__/codex-steer.test.ts @@ -0,0 +1,60 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { acceptCodexSteer } from "../codex-steer.js"; +import type { QueuedMessage, QueuedMessageInput } from "../session-queue.js"; + +function queued(id = "queued-1"): QueuedMessage { + return { + id, + text: "follow up", + visibleText: "follow up", + provider: "codex", + model: "gpt-5", + mode: "default", + attachmentIds: [], + createdAt: "2026-08-06T00:00:00.000Z", + }; +} + +function options(outcome: "steered" | "turn-ended", queuedMessageId?: string) { + const calls = { enqueued: 0, removed: 0 }; + const item = queued(queuedMessageId); + return { + calls, + input: { + queuedMessageId, + steer: async () => outcome, + getQueuedMessage: async () => item, + removeQueuedMessage: async () => { calls.removed += 1; }, + buildFollowUp: async (): Promise => item, + enqueueFollowUp: async () => { calls.enqueued += 1; return item; }, + }, + }; +} + +test("accepted native steer does not enqueue and removes a promoted item once", async () => { + const fixture = options("steered", "queued-1"); + assert.deepEqual(await acceptCodexSteer(fixture.input), { disposition: "steered" }); + assert.deepEqual(fixture.calls, { enqueued: 0, removed: 1 }); +}); + +test("turn-finalization race preserves typed text as a queued follow-up", async () => { + const fixture = options("turn-ended"); + const result = await acceptCodexSteer(fixture.input); + assert.equal(result.disposition, "queued"); + assert.deepEqual(fixture.calls, { enqueued: 1, removed: 0 }); +}); + +test("turn-finalization race leaves a promoted queue item exactly once", async () => { + const fixture = options("turn-ended", "queued-1"); + const result = await acceptCodexSteer(fixture.input); + assert.equal(result.disposition, "queued"); + assert.deepEqual(fixture.calls, { enqueued: 0, removed: 0 }); +}); + +test("genuine native steer failure does not enqueue or remove queue state", async () => { + const fixture = options("steered"); + fixture.input.steer = async () => { throw new Error("protocol failure"); }; + await assert.rejects(() => acceptCodexSteer(fixture.input), /protocol failure/); + assert.deepEqual(fixture.calls, { enqueued: 0, removed: 0 }); +}); diff --git a/server/lib/codex-app-server.ts b/server/lib/codex-app-server.ts index fa1b7c3f..5a3cd9df 100644 --- a/server/lib/codex-app-server.ts +++ b/server/lib/codex-app-server.ts @@ -253,19 +253,29 @@ export class CodexAppServerManager { }); } - async steerSession(sessionId: string, message: string): Promise { + async steerSession(sessionId: string, message: string): Promise<"steered" | "turn-ended"> { const runtime = this.sessions.get(sessionId); if (!runtime?.turnInProgress) { - throw new Error("No active turn for this session"); + return "turn-ended"; } if (!runtime.currentTurnId) { throw new Error("Turn ID not yet available"); } - await this.call("turn/steer", { - threadId: runtime.threadId, - input: [{ type: "text", text: message, text_elements: [] }], - expectedTurnId: runtime.currentTurnId, - }); + try { + await this.call("turn/steer", { + threadId: runtime.threadId, + input: [{ type: "text", text: message, text_elements: [] }], + expectedTurnId: runtime.currentTurnId, + }); + return "steered"; + } catch (error) { + // A terminal notification can arrive while turn/steer is in flight. + // Once that happens the app-server no longer owns the message, so let + // the route preserve it as a follow-up. Other failures are genuine and + // must remain visible to the composer without claiming delivery. + if (!runtime.turnInProgress) return "turn-ended"; + throw error; + } } async submitUserInput( diff --git a/server/lib/codex-steer.ts b/server/lib/codex-steer.ts new file mode 100644 index 00000000..649b247b --- /dev/null +++ b/server/lib/codex-steer.ts @@ -0,0 +1,41 @@ +import type { QueuedMessage, QueuedMessageInput } from "./session-queue.js"; + +export type CodexSteerResult = + | { disposition: "steered" } + | { disposition: "queued"; message?: QueuedMessage }; + +interface AcceptCodexSteerOptions { + queuedMessageId?: string; + steer: () => Promise<"steered" | "turn-ended">; + getQueuedMessage: (id: string) => Promise; + removeQueuedMessage: (id: string) => Promise; + buildFollowUp: () => Promise; + enqueueFollowUp: (input: QueuedMessageInput) => Promise; +} + +/* + * Transfer ownership of a Codex composer submission exactly once. A terminal + * event can win while turn/steer is in flight; in that case a typed message is + * made durable, while a promoted queue item simply remains owned by the queue. + */ +export async function acceptCodexSteer( + options: AcceptCodexSteerOptions +): Promise { + const outcome = await options.steer(); + if (outcome === "steered") { + if (options.queuedMessageId) { + await options.removeQueuedMessage(options.queuedMessageId); + } + return { disposition: "steered" }; + } + + if (options.queuedMessageId) { + return { + disposition: "queued", + message: await options.getQueuedMessage(options.queuedMessageId), + }; + } + + const queued = await options.enqueueFollowUp(await options.buildFollowUp()); + return { disposition: "queued", message: queued }; +} diff --git a/server/routes/sessions.ts b/server/routes/sessions.ts index 81bf6d42..a44ac277 100644 --- a/server/routes/sessions.ts +++ b/server/routes/sessions.ts @@ -80,6 +80,7 @@ import { type QueuedMessage, type QueuedMessageInput, } from "../lib/session-queue.js"; +import { acceptCodexSteer } from "../lib/codex-steer.js"; // Strip ANSI escape codes (color, cursor, etc.) const ANSI_RE = /\x1b\[[0-9;]*[a-zA-Z]/g; @@ -1540,7 +1541,11 @@ export async function handleSessionStream( // This happens server-side so the queue drains regardless of // whether any client is connected (see issue #113). if (streamSessionId && !pausedForClaudeUserInput && code === 0) { - void advanceSessionQueue(req.params.projectId, worktreeId, streamSessionId); + void scheduleSessionQueueAdvance( + req.params.projectId, + worktreeId, + streamSessionId + ); } }) .catch(() => {}); @@ -1619,6 +1624,39 @@ async function advanceSessionQueue( await advanceSessionQueue(projectId, worktreeId, sessionId); } +/* + * Finalization and a late /steer fallback can both discover that a queue is + * ready. Serialize those triggers and re-check runtime state so they cannot + * start two follow-up turns concurrently. + */ +const queueAdvanceChains = new Map>(); + +function scheduleSessionQueueAdvance( + projectId: string, + worktreeId: string, + sessionId: string +): Promise { + const previous = queueAdvanceChains.get(sessionId) ?? Promise.resolve(); + const next = previous + .then(async () => { + if (getSessionRuntime(sessionId).active) return; + await advanceSessionQueue(projectId, worktreeId, sessionId); + }) + .catch((error) => { + console.error( + `[session] queue advancement failed (session=${sessionId}):`, + error instanceof Error ? error.message : error + ); + }); + queueAdvanceChains.set(sessionId, next); + void next.finally(() => { + if (queueAdvanceChains.get(sessionId) === next) { + queueAdvanceChains.delete(sessionId); + } + }); + return next; +} + /** Persist a visible error for a queued message that could not be started. */ async function recordQueueAdvanceFailure( projectId: string, @@ -1779,7 +1817,7 @@ async function streamCodexPlanSession( // Drain the next enqueued message on a clean completion (server-side, // independent of any client; see issue #113). if (streamSessionId && exitCode === 0) { - void advanceSessionQueue(projectId, worktreeId, streamSessionId); + void scheduleSessionQueueAdvance(projectId, worktreeId, streamSessionId); } } @@ -2052,13 +2090,50 @@ sessionsRouter.post( } const message = req.body.message as string | undefined; + const queuedMessageId = req.body.queuedMessageId as string | undefined; if (!message || typeof message !== "string") { res.status(400).json({ error: "message is required" }); return; } try { - await codexAppServerManager.steerSession(req.params.sessionId, message); + const result = await acceptCodexSteer({ + queuedMessageId, + steer: () => codexAppServerManager.steerSession(req.params.sessionId, message), + getQueuedMessage: async (id) => + (await listQueue(req.params.sessionId)).find((item) => item.id === id), + removeQueuedMessage: async (id) => { + await removeFromQueue(req.params.sessionId, id); + }, + buildFollowUp: async () => { + const session = await getSession(worktree.path, req.params.sessionId); + if (!session?.provider || !session.model) { + throw new Error("Session metadata is unavailable for the follow-up"); + } + return { + text: message, + visibleText: message, + provider: session.provider, + model: session.model, + reasoningEffort: session.reasoningEffort, + serviceTier: session.serviceTier === "fast" ? "fast" : undefined, + mode: session.mode === "plan" ? "plan" : "default", + attachmentIds: [], + }; + }, + enqueueFollowUp: (input) => enqueueMessage(req.params.sessionId, input), + }); + if (result.disposition === "queued") { + res.json({ ok: true, ...result }); + if (!getSessionRuntime(req.params.sessionId).active) { + void scheduleSessionQueueAdvance( + req.params.projectId, + worktree.id, + req.params.sessionId + ); + } + return; + } await appendEvent(worktree.path, req.params.sessionId, { id: randomUUID(), sessionId: req.params.sessionId, @@ -2066,7 +2141,7 @@ sessionsRouter.post( type: "user_message", data: { text: message }, }); - res.json({ ok: true }); + res.json({ ok: true, ...result }); } catch (error) { res.status(400).json({ error: error instanceof Error ? error.message : String(error),