vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-08-21 16:53:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

feat(client): optimistic prompt admission with client-minted IDs (#43520)

Browse source
This commit is contained in:
Kit Langton 2026-08-19 20:34:12 -04:00 • committed by GitHub
parent 3876f7aad6
commit 6b09b9e6a2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 435 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

3
packages/app/src/components/prompt-input/submit.test.ts
View file

@ -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" } }),

4
packages/app/src/components/prompt-input/submit.ts
View file

@ -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,

167
packages/client/src/solid/data.ts
View file

@ -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<string>()
// 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<string, number>) => 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)

37
packages/tui/src/component/prompt/index.tsx
View file

@ -1263,7 +1263,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,
@ -1272,14 +1277,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({
@ -1291,15 +1297,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()

288
packages/tui/test/cli/tui/data.test.tsx
View file

@ -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<Response>((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<typeof useData>
let ready!: () => void
const mounted = new Promise<void>((resolve) => {
ready = resolve
})
function Probe() {
sync = useData()
onMount(ready)
return <box />
}
const app = await testRender(() => (
<TestTuiContexts>
<ClientProvider api={createApi(calls.fetch)}>
<ProjectProvider>
<DataProvider>
<Probe />
</DataProvider>
</ProjectProvider>
</ClientProvider>
</TestTuiContexts>
))
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<typeof useData>
let ready!: () => void
const mounted = new Promise<void>((resolve) => {
ready = resolve
})
function Probe() {
sync = useData()
onMount(ready)
return <box />
}
const app = await testRender(() => (
<TestTuiContexts>
<ClientProvider api={createApi(calls.fetch)}>
<ProjectProvider>
<DataProvider>
<Probe />
</DataProvider>
</ProjectProvider>
</ClientProvider>
</TestTuiContexts>
))
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<typeof useData>
let ready!: () => void
const mounted = new Promise<void>((resolve) => {
ready = resolve
})
function Probe() {
sync = useData()
onMount(ready)
return <box />
}
const app = await testRender(() => (
<TestTuiContexts>
<ClientProvider api={createApi(calls.fetch)}>
<ProjectProvider>
<DataProvider>
<Probe />
</DataProvider>
</ProjectProvider>
</ClientProvider>
</TestTuiContexts>
))
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<typeof useData>
let ready!: () => void
const mounted = new Promise<void>((resolve) => {
ready = resolve
})
function Probe() {
sync = useData()
onMount(ready)
return <box />
}
const app = await testRender(() => (
<TestTuiContexts>
<ClientProvider api={createApi(calls.fetch)}>
<ProjectProvider>
<DataProvider>
<Probe />
</DataProvider>
</ProjectProvider>
</ClientProvider>
</TestTuiContexts>
))
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()
}
})