diff --git a/packages/app/src/components/prompt-input/submit.test.ts b/packages/app/src/components/prompt-input/submit.test.ts index 5397816afcba..b09880d2b0b8 100644 --- a/packages/app/src/components/prompt-input/submit.test.ts +++ b/packages/app/src/components/prompt-input/submit.test.ts @@ -235,6 +235,9 @@ beforeAll(async () => { session: { remember: () => undefined, setStatus: () => undefined, + // Delegates straight to the API client; optimistic admission and + // rollback are covered by the data-layer tests in packages/tui. + prompt: (input: unknown) => rootClient.api.session.prompt(input as never), }, location: { info: () => ({ project: { id: "project", directory: "/repo/main" } }), diff --git a/packages/app/src/components/prompt-input/submit.ts b/packages/app/src/components/prompt-input/submit.ts index ecb9c2bfc82f..e5a1541ea020 100644 --- a/packages/app/src/components/prompt-input/submit.ts +++ b/packages/app/src/components/prompt-input/submit.ts @@ -132,7 +132,9 @@ export async function sendFollowupDraft(input: FollowupSendInput) { }) } - await input.api.prompt({ + // The data layer admits optimistically: the prompt renders immediately + // and rolls back if the server rejects it. + await input.data.session.prompt({ sessionID: input.draft.sessionID, id: messageID, text: request.text, diff --git a/packages/client/src/solid/data.ts b/packages/client/src/solid/data.ts index 32eb7546eb1d..3760f5137fd8 100644 --- a/packages/client/src/solid/data.ts +++ b/packages/client/src/solid/data.ts @@ -34,10 +34,11 @@ import type { WebSearchProvider, } from "../promise" import { Worktree } from "@opencode-ai/schema/worktree" -import { isPermissionNotFoundError } from "../promise" +import { SessionMessage } from "@opencode-ai/schema/session-message" +import { isPermissionNotFoundError, type SessionPromptInput } from "../promise" import { createStore, produce, reconcile } from "solid-js/store" import type { SessionInbox } from "@opencode-ai/schema/session-inbox" -import { createEffect, createSignal, onCleanup } from "solid-js" +import { batch, createEffect, createSignal, onCleanup } from "solid-js" export type DataSessionStatus = "idle" | "running" @@ -178,11 +179,6 @@ export function createData(config: CreateDataInput) { setStore("session", "active", sessionID, status) } - function addPending(item: SessionInboxInfo) { - if (store.session.pending[item.sessionID]?.some((pending) => pending.id === item.id)) return - setStore("session", "pending", item.sessionID, [...(store.session.pending[item.sessionID] ?? []), item]) - } - function removePending(sessionID: string, inboxID?: string) { if (!inboxID) return if (store.session.pending[sessionID]?.some((item) => item.id === inboxID)) @@ -219,6 +215,60 @@ export function createData(config: CreateDataInput) { setStore("session", "pending", sessionID, index, { ...item, delivery }) } + // Inbox IDs of optimistic prompt admissions still awaiting their durable + // echo. This is the one deliberate piece of in-flight bookkeeping in this + // layer: it exists so a rejection only rolls back rows the server never + // acknowledged, and so a concurrent pending re-fetch cannot wipe a row the + // server does not know about yet. Entries clear on the enqueued echo or on + // rollback — not on POST success, which typically precedes the echo. + const outbox = new Set() + + // Upsert an admitted inbox item into pending, input, and (for user and + // synthetic items) the visible transcript. Used by the inbox.enqueued + // handler and by optimistic prompt admission; the upsert is what reconciles + // the durable echo with an optimistic placeholder — the durable payload and + // times replace the client's guess. + function admitLocal(item: SessionInboxInfo) { + batch(() => { + const pending = store.session.pending[item.sessionID] ?? [] + const at = pending.findIndex((entry) => entry.id === item.id) + setStore( + "session", + "pending", + item.sessionID, + at < 0 ? [...pending, item] : pending.map((entry, index) => (index === at ? item : entry)), + ) + const input = store.session.input[item.sessionID] ?? [] + if (!input.includes(item.id)) setStore("session", "input", item.sessionID, [...input, item.id]) + if (item.type !== "user" && item.type !== "synthetic") return + message.update(item.sessionID, (draft, index) => { + const row = + item.type === "user" + ? { id: item.id, type: "user" as const, ...item.payload, time: { created: item.timeCreated } } + : { id: item.id, type: "synthetic" as const, ...item.payload, time: { created: item.timeCreated } } + const position = index.get(item.id) + if (position === undefined) return message.append(draft, index, row) + draft[position] = row + }) + }) + } + + // Remove an inbox item from pending, input, and the visible transcript. + // Used by the inbox.cancelled handler and by optimistic rollback. + function retractLocal(sessionID: string, inboxID: string) { + batch(() => { + removePending(sessionID, inboxID) + if (!messageIndex.get(sessionID)?.has(inboxID)) return + message.update(sessionID, (draft, index) => { + const position = index.get(inboxID) + if (position === undefined) return + draft.splice(position, 1) + index.delete(inboxID) + message.reindex(draft, index, position) + }) + }) + } + const message = { update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { setStore( @@ -325,6 +375,7 @@ export function createData(config: CreateDataInput) { } function removeSession(sessionID: string) { + store.session.pending[sessionID]?.forEach((item) => outbox.delete(item.id)) messageIndex.delete(sessionID) sync.invalidate(`session:${sessionID}`) sync.invalidate(`session.pending:${sessionID}`) @@ -493,49 +544,16 @@ export function createData(config: CreateDataInput) { updatePending(event.data.sessionID, event.data.inboxID, event.data.delivery) return case "session.inbox.cancelled": { - removePending(event.data.sessionID, event.data.inboxID) - if (messageIndex.get(event.data.sessionID)?.has(event.data.inboxID)) - message.update(event.data.sessionID, (draft, index) => { - const position = index.get(event.data.inboxID) - if (position === undefined) return - draft.splice(position, 1) - index.delete(event.data.inboxID) - message.reindex(draft, index, position) - }) + retractLocal(event.data.sessionID, event.data.inboxID) return } case "session.inbox.enqueued": { - const item = event.data.item - addPending({ + outbox.delete(event.data.inboxID) + admitLocal({ id: event.data.inboxID, sessionID: event.data.sessionID, timeCreated: event.created, - ...item, - }) - if (!store.session.input[event.data.sessionID]?.includes(event.data.inboxID)) - setStore("session", "input", event.data.sessionID, [ - ...(store.session.input[event.data.sessionID] ?? []), - event.data.inboxID, - ]) - if (item.type !== "user" && item.type !== "synthetic") return - message.update(event.data.sessionID, (draft, index) => { - message.append( - draft, - index, - item.type === "user" - ? { - id: event.data.inboxID, - type: "user", - ...item.payload, - time: { created: event.created }, - } - : { - id: event.data.inboxID, - type: "synthetic", - ...item.payload, - time: { created: event.created }, - }, - ) + ...event.data.item, }) return } @@ -1062,12 +1080,19 @@ export function createData(config: CreateDataInput) { sync(sessionID: string) { return sync.run(`session.pending:${sessionID}`, async () => { const pending = await api().session.inbox.list({ sessionID }) - setStore("session", "pending", sessionID, reconcile(pending)) + // Keep optimistic rows still awaiting their echo: this fetch may + // have raced ahead of an in-flight admission the server does not + // know about yet. + const inflight = (store.session.pending[sessionID] ?? []).filter( + (item) => outbox.has(item.id) && !pending.some((row) => row.id === item.id), + ) + const merged = inflight.length === 0 ? pending : [...pending, ...inflight] + setStore("session", "pending", sessionID, reconcile(merged)) setStore( "session", "input", sessionID, - reconcile(pending.filter((item) => item.type !== "compaction").map((item) => item.id)), + reconcile(merged.filter((item) => item.type !== "compaction").map((item) => item.id)), ) }) }, @@ -1075,6 +1100,47 @@ export function createData(config: CreateDataInput) { sync.invalidate(`session.pending:${sessionID}`) }, }, + // Optimistic prompt admission: render the prompt immediately under a + // client-minted ID, send it, and let the durable inbox.enqueued echo + // upsert that same ID with the server's payload. Server admission is + // idempotent per ID, so retrying with the identical payload cannot + // double-admit. + prompt(input: SessionPromptInput) { + const id = input.id ?? SessionMessage.ID.create() + // A retry may reuse an ID that is already rendered — and possibly + // already durable. Admit optimistically only for new IDs so a failed + // retry cannot roll back acknowledged state. + const fresh = + !messageIndex.get(input.sessionID)?.has(id) && + !store.session.pending[input.sessionID]?.some((item) => item.id === id) + if (fresh) { + outbox.add(id) + admitLocal({ + id, + sessionID: input.sessionID, + timeCreated: Date.now(), + type: "user", + delivery: input.delivery ?? "steer", + // Files and skills stay off the optimistic row: their durable + // forms are server-loaded (content, mime, resolution), so they + // fill in when the echo upserts the row. + payload: { + text: input.text, + agents: input.agents?.map((agent) => ({ ...agent })), + metadata: input.metadata, + }, + }) + } + // Wrapped so even a synchronous client failure reaches the rollback. + return Promise.resolve() + .then(() => api().session.prompt({ ...input, id })) + .catch((error) => { + // Roll back only rows this call admitted and the echo has not + // acknowledged: anything else is server state. + if (fresh && outbox.delete(id)) retractLocal(input.sessionID, id) + throw error + }) + }, sync(sessionID: string, options?: { children?: boolean }) { return sync.run(options?.children ? `session.family:${sessionID}` : `session:${sessionID}`, async () => { const [info, children] = await Promise.all([ @@ -1114,7 +1180,14 @@ export function createData(config: CreateDataInput) { sync(sessionID: string) { return sync.run(`session.message:${sessionID}`, async () => { const response = await api().message.list({ sessionID, limit: 200, order: "desc" }) - const messages = response.data.toReversed() + const fetched = response.data.toReversed() + // Same protection as the pending sync: a re-fetch racing an + // optimistic admission must not wipe the in-flight transcript row. + const ids = new Set(fetched.map((item) => item.id)) + const inflight = (store.session.message[sessionID] ?? []).filter( + (item) => outbox.has(item.id) && !ids.has(item.id), + ) + const messages = inflight.length === 0 ? fetched : [...fetched, ...inflight] messageIndex.set(sessionID, new Map(messages.map((message, index) => [message.id, index]))) setStore("session", "message", sessionID, reconcile(messages)) setStore("session", "messageCursor", sessionID, response.cursor.next ?? undefined) diff --git a/packages/tui/src/component/prompt/index.tsx b/packages/tui/src/component/prompt/index.tsx index e3e37358a8d6..461cb39fc8c4 100644 --- a/packages/tui/src/component/prompt/index.tsx +++ b/packages/tui/src/component/prompt/index.tsx @@ -1304,7 +1304,12 @@ export function Prompt(props: PromptProps) { return false } } - const error = await client.api.session + // The data layer admits optimistically: the prompt renders immediately + // and rolls back if the server rejects it, so submission does not wait + // on the network. On rejection the row is already rolled back; restore + // the composer unless the user has started typing something new. + const entry = { ...store.prompt, mode: currentMode } + data.session .prompt({ sessionID, text: inputText, @@ -1313,14 +1318,15 @@ export function Prompt(props: PromptProps) { skills: store.prompt.skills?.length ? store.prompt.skills : undefined, delivery, }) - .then( - () => undefined, - (error) => error, - ) - if (error) { - toast.show({ title: "Failed to send prompt", message: errorMessage(error), variant: "error" }) - return false - } + .catch((error) => { + toast.show({ title: "Failed to send prompt", message: errorMessage(error), variant: "error" }) + if (disposed || input.isDestroyed || input.plainText !== "") return + input.setText(entry.text) + setStore("prompt", entry) + setStore("mode", entry.mode ?? "normal") + restoreExtmarksFromPrompt(entry) + input.cursorOffset = entry.text.length + }) if (pendingEditorSelection) editor.markSelectionSent() } history.append({ @@ -1332,15 +1338,14 @@ export function Prompt(props: PromptProps) { setStore("extmarkToPart", new Map()) props.onSubmit?.() - // temporary hack to make sure the message is sent + // Optimistic admission puts the message in the store synchronously, so + // the session view renders it on arrival. if (!props.sessionID) { if (pendingEditorSelection) editor.preserveSelectionFromNewSession() - setTimeout(() => { - route.navigate({ - type: "session", - sessionID, - }) - }, 50) + route.navigate({ + type: "session", + sessionID, + }) } input.clear() if (finishMoveProgress) move.finishSubmit() diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 4f7a3b852119..e1eb763ea9d6 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -3094,3 +3094,291 @@ test("stops at the last non-repeating ancestor on a parent cycle", async () => { app.renderer.destroy() } }) + +test("admits prompts optimistically and reconciles with the durable echo", async () => { + const events = createEventStream() + const sessionID = "session-1" + let release!: (response: Response) => void + const deferred = new Promise((resolve) => { + release = resolve + }) + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/prompt`) return deferred + // The server does not know about the in-flight admission yet. + if (url.pathname === `/api/session/${sessionID}/inbox`) return json({ data: [] }) + }, events) + let sync!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + sync = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + const promise = sync.session.prompt({ sessionID, text: "hello" }) + const settled = promise.then( + () => undefined, + (error) => error, + ) + + // Optimistic: the row renders before the server responds. + const optimistic = sync.session.pending.list(sessionID)[0] + expect(optimistic).toMatchObject({ sessionID, type: "user", payload: { text: "hello" }, delivery: "steer" }) + const messageID = optimistic!.id + expect(messageID.startsWith("msg_")).toBe(true) + expect(sync.session.input.list(sessionID)).toEqual([messageID]) + expect(sync.session.message.list(sessionID).map((message) => message.id)).toEqual([messageID]) + + // A pending re-fetch racing the in-flight admission cannot wipe the row. + await sync.session.pending.sync(sessionID) + expect(sync.session.pending.list(sessionID).map((item) => item.id)).toEqual([messageID]) + expect(sync.session.input.list(sessionID)).toEqual([messageID]) + + // The durable echo upserts by ID instead of duplicating: server-loaded + // payload (files) and durable times replace the optimistic placeholder. + const received: string[] = [] + const unsubscribe = sync.listen((event) => received.push(event.name)) + const echoFile = { data: "aGVsbG8=", mime: "text/plain", source: { type: "uri" as const, uri: "file:///a.txt" } } + emitEvent(events, { + id: "evt_echo_1", + created: 5, + type: "session.inbox.enqueued", + durable: durable(sessionID), + data: { + sessionID, + inboxID: messageID, + item: { type: "user", payload: { text: "hello", files: [echoFile] }, delivery: "steer" }, + }, + }) + await wait(() => received.includes("session.inbox.enqueued")) + unsubscribe() + expect(sync.session.pending.list(sessionID)).toEqual([ + { + id: messageID, + sessionID, + timeCreated: 5, + type: "user", + payload: { text: "hello", files: [echoFile] }, + delivery: "steer", + }, + ]) + const echoed = sync.session.message.list(sessionID)[0] + expect(echoed?.type).toBe("user") + if (echoed?.type !== "user") return + expect(echoed.time.created).toBe(5) + expect(echoed.files).toEqual([echoFile]) + + // A late transport failure after the echo must not delete acknowledged state. + release(json({ _tag: "UnknownError", message: "response lost" }, { status: 500 })) + expect(await settled).toBeDefined() + expect(sync.session.pending.list(sessionID).map((item) => item.id)).toEqual([messageID]) + expect(sync.session.message.list(sessionID).map((message) => message.id)).toEqual([messageID]) + } finally { + release(json({ _tag: "UnknownError", message: "cleanup" }, { status: 500 })) + app.renderer.destroy() + } +}) + +test("keeps the row when the response lands before the echo", async () => { + const events = createEventStream() + const sessionID = "session-1" + const messageID = "msg_early_1" + const admission = { + id: messageID, + sessionID, + timeCreated: 1, + type: "user", + payload: { text: "hello" }, + delivery: "steer", + } + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/prompt`) return json({ data: admission }) + // The server's listings still miss the admission (projection lag). + if (url.pathname === `/api/session/${sessionID}/inbox`) return json({ data: [] }) + if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) + }, events) + let sync!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + sync = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + await sync.session.prompt({ sessionID, id: messageID, text: "hello" }) + + // POST resolved but the echo has not arrived: racing pending and message + // re-fetches still cannot wipe the row. + await sync.session.pending.sync(sessionID) + sync.session.pending.invalidate(sessionID) + await sync.session.pending.sync(sessionID) + await sync.session.message.sync(sessionID) + sync.session.message.invalidate(sessionID) + await sync.session.message.sync(sessionID) + expect(sync.session.pending.list(sessionID).map((item) => item.id)).toEqual([messageID]) + expect(sync.session.input.list(sessionID)).toEqual([messageID]) + expect(sync.session.message.list(sessionID).map((message) => message.id)).toEqual([messageID]) + } finally { + app.renderer.destroy() + } +}) + +test("rolls back an optimistic prompt the server rejected", async () => { + const events = createEventStream() + const sessionID = "session-1" + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/prompt`) + return json({ _tag: "InvalidRequestError", message: "invalid" }, { status: 400 }) + }, events) + let sync!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + sync = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + const promise = sync.session.prompt({ sessionID, text: "rejected" }) + expect(sync.session.message.list(sessionID)).toHaveLength(1) + + await expect(promise).rejects.toThrow() + expect(sync.session.pending.list(sessionID)).toEqual([]) + expect(sync.session.input.list(sessionID)).toEqual([]) + expect(sync.session.message.list(sessionID)).toEqual([]) + } finally { + app.renderer.destroy() + } +}) + +test("a retry under the same client-minted ID cannot duplicate rows", async () => { + const events = createEventStream() + const sessionID = "session-1" + const messageID = "msg_retry_1" + const admission = { + id: messageID, + sessionID, + timeCreated: 1, + type: "user", + payload: { text: "hello" }, + delivery: "steer", + } + const posts: string[] = [] + let fail = false + const calls = createFetch(async (url, request) => { + if (url.pathname === `/api/session/${sessionID}/prompt`) { + posts.push(((await request.json()) as { id: string }).id) + if (fail) return json({ _tag: "UnknownError", message: "transient" }, { status: 500 }) + return json({ data: admission }) + } + }, events) + let sync!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + sync = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + await sync.session.prompt({ sessionID, id: messageID, text: "hello" }) + // Retry with the identical payload: server admission is idempotent per ID, + // and the local dedupe keeps a single row. + await sync.session.prompt({ sessionID, id: messageID, text: "hello" }) + + expect(posts).toEqual([messageID, messageID]) + expect(sync.session.pending.list(sessionID).map((item) => item.id)).toEqual([messageID]) + expect(sync.session.input.list(sessionID)).toEqual([messageID]) + expect(sync.session.message.list(sessionID).map((message) => message.id)).toEqual([messageID]) + + // The row is acknowledged (echo applied): a FAILED retry under the same + // ID must not roll back acknowledged state. + const received: string[] = [] + const unsubscribe = sync.listen((event) => received.push(event.name)) + emitEvent(events, { + id: "evt_ack_1", + created: 2, + type: "session.inbox.enqueued", + durable: durable(sessionID), + data: { sessionID, inboxID: messageID, item: { type: "user", payload: { text: "hello" }, delivery: "steer" } }, + }) + await wait(() => received.includes("session.inbox.enqueued")) + unsubscribe() + fail = true + await expect(sync.session.prompt({ sessionID, id: messageID, text: "hello" })).rejects.toThrow() + expect(sync.session.pending.list(sessionID).map((item) => item.id)).toEqual([messageID]) + expect(sync.session.message.list(sessionID).map((message) => message.id)).toEqual([messageID]) + } finally { + app.renderer.destroy() + } +})