Compare commits

...
26 changed files with 2962 additions and 243 deletions
@@ -141,7 +141,9 @@ for (const direction of ["ltr", "rtl"]) {
await page.locator("html").evaluate((element, direction) => element.setAttribute("dir", direction), direction)
const pending = await submitPending(page, mock)
await draftFollowUp(page)
await page.locator('[data-component="composer-editor"]').press("ControlOrMeta+Home")
await page
.locator('[data-component="composer-editor"]')
.press(process.platform === "darwin" ? "Meta+ArrowUp" : "Control+Home")
const title = page.locator("[data-session-title]").getByRole("heading", { level: 1 })
const before = await title.boundingBox()
const messageBefore = await pending.message.boundingBox()
@@ -224,26 +226,94 @@ for (const failure of ["worktree", "session"]) {
})
}
test("preserves both inputs when the initial prompt cannot be sent", async ({ page }) => {
const mock = await openDraft(page)
const pending = await submitPending(page, mock)
await draftFollowUp(page)
await page.route(`**/api/session/${pending.sessionID}/prompt`, (route) =>
route.fulfill({
status: 500,
json: { message: "Prompt admission failed" },
headers,
}),
)
for (const action of ["Retry", "Cancel"]) {
test(`preserves both inputs until ${action} when the initial prompt cannot be sent`, async ({ page }) => {
const mock = await openDraft(page)
const pending = await submitPending(page, mock)
await draftFollowUp(page)
const promptURL = `${server}/api/session/${pending.sessionID}/prompt`
const attempts: Record<string, unknown>[] = []
page.on("request", (request) => {
if (request.method() === "POST" && request.url() === promptURL) attempts.push(request.postDataJSON())
})
await page.route(promptURL, (route) =>
route.fulfill({ status: 500, json: { message: "Prompt admission failed" }, headers }),
)
mock.worktree.resolve({ status: 200, json: { directory: workspace } })
mock.worktree.resolve({ status: 200, json: { directory: workspace } })
await expect(pending.shimmer).toHaveCount(0)
await expect(page.locator('[data-component="composer-editor"]')).toHaveText(`${text}\n\n${followUp}`)
await expect(page.locator('[data-action="composer-submit"]')).toBeEnabled()
await expect.poll(() => mock.calls).toEqual(["worktree", "session", "prompt", "prompt"])
expect(mock.prompts).toEqual([])
})
const editor = page.locator('[data-component="composer-editor"]')
const recovery = page.locator(`[data-component="prompt-submission"][data-submission-id="${pending.messageID}"]`)
const message = page.locator(`[data-component="user-message"][data-timeline-part-id="${pending.messageID}:text:0"]`)
await expect(recovery.getByRole("status")).toHaveText("Could not confirm prompt delivery")
await expect(page).toHaveURL(pending.url)
await expect(pending.shimmer).toHaveCount(0)
await expect(pending.message).toHaveCount(1)
await expect(message.locator('[data-slot="user-message-text"]')).toHaveText(text)
await expect(editor).toHaveText(followUp)
await expect(page.locator('[data-action="composer-submit"]')).toBeEnabled()
expect(mock.creates).toEqual([
expect.objectContaining({ id: pending.sessionID, location: { directory: workspace } }),
])
expect(attempts).toHaveLength(4)
expect(attempts[0]).toMatchObject({ id: pending.messageID, text, delivery: "steer" })
expect(attempts).toEqual(Array(4).fill(attempts[0]))
expect(mock.prompts).toEqual([])
await page.locator(`[data-titlebar-tab-link][href="${sessionPath}${otherID}"]`).click()
await expect(page).toHaveURL(`${sessionPath}${otherID}`)
await expect(editor).toBeEditable()
await expect(editor).toHaveText("")
await expect(recovery).toHaveCount(0)
await expect(message).toHaveCount(0)
await editor.fill("Keep this other session's draft")
await page.locator(`[data-titlebar-tab-link][href="${sessionPath}${pending.sessionID}"]`).click()
await expect(page).toHaveURL(pending.url)
await expect(recovery.getByRole("status")).toHaveText("Could not confirm prompt delivery")
await expect(message.locator('[data-slot="user-message-text"]')).toHaveText(text)
await expect(editor).toHaveText(followUp)
await page.unroute(promptURL)
if (action === "Retry") {
await recovery.getByRole("button", { name: action, exact: true }).click()
await expect.poll(() => mock.prompts).toEqual([{ sessionID: pending.sessionID, body: attempts[0] }])
expect(attempts).toHaveLength(5)
expect(attempts[4]).toEqual(attempts[0])
await expect(message.locator('[data-slot="user-message-text"]')).toHaveText(text)
}
if (action === "Cancel") {
const cancelled = page.waitForResponse(
(response) =>
response.request().method() === "DELETE" &&
response.url() === `${server}/api/session/${pending.sessionID}/inbox/${pending.messageID}`,
)
await recovery.getByRole("button", { name: action, exact: true }).click()
expect((await cancelled).status()).toBe(204)
await expect(message).toHaveCount(0)
expect(attempts).toHaveLength(4)
expect(mock.prompts).toEqual([])
}
await expect(recovery).toHaveCount(0)
await expect(editor).toHaveText(followUp)
await expect(page.locator('[data-action="composer-submit"]')).toBeEnabled()
await page.locator('[data-action="composer-submit"]').click()
await expect.poll(() => mock.prompts.length).toBe(action === "Retry" ? 2 : 1)
const sent = mock.prompts.find((prompt) => prompt.body.text === followUp)!
expect(sent).toMatchObject({ sessionID: pending.sessionID, body: { id: expect.any(String), text: followUp } })
expect(sent.body.id).not.toBe(pending.messageID)
await expect(
page.locator(`[data-timeline-part-id="${sent.body.id}:text:0"] [data-slot="user-message-text"]`),
).toHaveText(followUp)
await expect(pending.message).toHaveCount(action === "Retry" ? 2 : 1)
await expect(editor).toHaveText("")
await expect(page).toHaveURL(pending.url)
expect(mock.worktreeRequests).toHaveLength(1)
expect(mock.creates).toHaveLength(1)
await page.locator(`[data-titlebar-tab-link][href="${sessionPath}${otherID}"]`).click()
await expect(page).toHaveURL(`${sessionPath}${otherID}`)
await expect(editor).toHaveText("Keep this other session's draft")
})
}
test("restores the original draft when worktree creation fails", async ({ page }) => {
const mock = await openDraft(page)
@@ -145,6 +145,10 @@ async function openSession(page: Page, mock: ReturnType<typeof createQueueMock>,
await page.goto(`/server/${base64Encode(server)}/session/${sessionID}`)
const composer = page.locator('[data-component="composer"]')
await expectAppVisible(composer)
// The editor is interactive before provider/model selection finishes loading.
await expect(composer.getByRole("button", { name: "Queue Model", exact: true })).toBeVisible()
await expect(composer.getByRole("button", { name: "Queue Model", exact: true })).toBeEnabled()
await expect(composer.getByRole("textbox", { name: "Prompt", exact: true })).toBeEditable()
return {
composer,
input: composer.locator('[data-component="composer-editor"]'),
@@ -234,6 +238,145 @@ test("editing restores the existing draft and replaces only the original queue p
expect(mock.log[0]).toBe("prompt:queue")
})
for (const action of ["edit", "reorder"] as const) {
test(`retires an exhausted queue ${action} replacement without offering a standalone retry`, async ({ page }) => {
const mock = createQueueMock(["first queued prompt", "second queued prompt"])
const view = await openSession(page, mock)
const attempts: string[] = []
await page.route(`**/api/session/${sessionID}/prompt`, (route) => {
const body = route.request().postDataJSON() as { id: string }
attempts.push(body.id)
return route.abort("failed")
})
await expect(view.rows).toHaveCount(2)
const first = view.rows.filter({ hasText: "first queued prompt" })
if (action === "edit") {
await first.getByRole("button", { name: "first queued prompt", exact: true }).click()
await view.input.fill("replacement that cannot be admitted")
await view.input.press("Enter")
}
if (action === "reorder") {
await first.getByRole("button", { name: "Reorder queued prompt" }).hover()
await page.mouse.down()
const target = await view.rows.filter({ hasText: "second queued prompt" }).boundingBox()
if (!target) throw new Error("The target queue row is not visible")
await page.mouse.move(target.x + target.width / 2, target.y + target.height / 2, { steps: 10 })
await page.mouse.up()
}
await expect(page.getByText("Request failed", { exact: true })).toBeVisible()
expect(attempts).toHaveLength(4)
expect(new Set(attempts).size).toBe(1)
expect(mock.changes).toEqual([{ inboxID: attempts[0], action: "cancel" }])
expect(mock.rows.map((row) => row.id)).toEqual(["inb_seed_1", "inb_seed_2"])
await expect(view.rows.locator('[data-action="session-queue-edit"]')).toHaveText([
"first queued prompt",
"second queued prompt",
])
await expect(page.locator('[data-component="prompt-submission"]')).toHaveCount(0)
})
}
for (const delivery of ["steer", "queue"] as const) {
for (const width of [390, 1440]) {
test(`recovers independent failed ${delivery} submissions at ${width}px`, async ({ page }, testInfo) => {
await page.setViewportSize({ width, height: 900 })
const mock = createQueueMock([])
const view = await openSession(page, mock, delivery)
const attempts: { id: string; text: string }[] = []
const accepted = new Set<string>()
await page.route(`**/api/session/${sessionID}/prompt`, (route) => {
const body = route.request().postDataJSON() as { id: string; text: string }
attempts.push(body)
return accepted.has(body.id) ? route.fallback() : route.abort("failed")
})
const recovery = page.locator('[data-component="prompt-submission"]')
await view.input.fill("first failed submission")
await view.input.press("Enter")
await expect(recovery.getByRole("status")).toHaveText("Could not confirm prompt delivery")
const firstID = attempts[0].id
expect(attempts.map((attempt) => attempt.id)).toEqual([firstID, firstID, firstID, firstID])
await expect(view.input).toHaveText("")
await view.input.fill("second failed submission")
await view.input.press("Enter")
await expect(recovery.getByRole("status")).toHaveText([
"Could not confirm prompt delivery",
"Could not confirm prompt delivery",
])
const secondID = attempts[4].id
expect(secondID).not.toBe(firstID)
expect(attempts.slice(4).map((attempt) => attempt.id)).toEqual([secondID, secondID, secondID, secondID])
await expect(view.input).toHaveText("")
if (delivery === "queue") {
await expect(view.rows).toHaveCount(2)
await expect(view.rows.getByRole("button", { name: "first failed submission", exact: true })).toBeDisabled()
await expect(view.rows.getByRole("button", { name: "second failed submission", exact: true })).toBeDisabled()
await expect(
view.rows
.filter({ hasText: "first failed submission" })
.getByRole("button", { name: "Reorder queued prompt" }),
).toBeDisabled()
await expect(view.rows.getByRole("button", { name: "Steer", exact: true })).toHaveCount(0)
}
await testInfo.attach("failed-submissions", { body: await page.screenshot(), contentType: "image/png" })
const firstRecovery = page.locator(`[data-component="prompt-submission"][data-submission-id="${firstID}"]`)
const secondRecovery = page.locator(`[data-component="prompt-submission"][data-submission-id="${secondID}"]`)
accepted.add(firstID)
await firstRecovery.getByRole("button", { name: "Retry", exact: true }).click()
await expect(firstRecovery).toHaveCount(0)
await expect.poll(() => mock.prompts.map((prompt) => prompt.id)).toEqual([firstID])
expect(attempts).toHaveLength(9)
expect(attempts[8]).toEqual(attempts[0])
await expect(secondRecovery.getByRole("status")).toHaveText("Could not confirm prompt delivery")
await secondRecovery.getByRole("button", { name: "Cancel", exact: true }).click()
await expect(recovery).toHaveCount(0)
await expect.poll(() => mock.changes).toEqual([{ inboxID: secondID, action: "cancel" }])
await expect(view.input).toHaveText("")
if (delivery === "queue") {
await expect(view.rows.locator('[data-action="session-queue-edit"]')).toHaveText(["first failed submission"])
return
}
const messages = page.locator('[data-timeline-row="UserMessage"]')
await expect(messages).toHaveCount(1)
await expect(messages).toContainText("first failed submission")
await expect(messages).toHaveAttribute("data-message-id", firstID)
})
}
test(`keeps recovery visible for an image-only failed ${delivery} submission`, async ({ page }) => {
const mock = createQueueMock([])
const view = await openSession(page, mock, delivery)
await page.route(`**/api/session/${sessionID}/prompt`, (route) => route.abort("failed"))
const chooser = page.waitForEvent("filechooser")
await view.composer.getByRole("button", { name: "Add images and files", exact: true }).click()
await page.getByRole("menuitem", { name: /^Images and files/ }).click()
await (
await chooser
).setFiles({
name: "pixel.png",
mimeType: "image/png",
buffer: Buffer.from(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aX1cAAAAASUVORK5CYII=",
"base64",
),
})
await expect(view.composer.locator('[data-component="composer-attachments"]')).toBeVisible()
await view.input.press("Enter")
const recovery = page.locator('[data-component="prompt-submission"]')
await expect(recovery.getByRole("status")).toHaveText("Could not confirm prompt delivery")
await expect(recovery.getByRole("button", { name: "Retry", exact: true })).toBeVisible()
await recovery.getByRole("button", { name: "Cancel", exact: true }).click()
await expect(recovery).toHaveCount(0)
await expect(view.rows).toHaveCount(0)
await expect(page.locator('[data-timeline-row="UserMessage"]')).toHaveCount(0)
await expect(view.composer.locator('[data-component="composer-attachments"]')).toHaveCount(0)
await expect(view.input).toHaveText("")
})
}
for (const delivery of ["steer", "queue"] as const) {
test(`keeps finished tools above a pending ${delivery === "queue" ? "queue-to-steer" : "steer"} follow-up`, async ({
page,
+2
View File
@@ -1,6 +1,7 @@
import { ImagePreview } from "@opencode-ai/ui/image-preview"
import { useDialog } from "@opencode-ai/ui/context/dialog"
import type { ReferenceInfo } from "@opencode-ai/client/promise"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { createComponent, createEffect, createMemo, on } from "solid-js"
import type { ComposerSuggestion } from "./types"
import { createComposerEditor, createComposerEditorState, type ComposerEditorModel } from "./editor/interaction"
@@ -418,6 +419,7 @@ export function createComposerModel(adapter: ComposerAdapter, options?: { queue?
}
function composerErrorMessage(language: ReturnType<typeof useLanguage>, error: unknown) {
if (error instanceof PromptSubmissionError) return language.t("session.submission.failed")
if (error && typeof error === "object" && "message" in error && typeof error.message === "string") {
return error.message
}
+89 -7
View File
@@ -1,6 +1,7 @@
import { describe, expect, test } from "bun:test"
import type { ModelSelection } from "@/providers/models/selection"
import type { SessionMessageUser } from "@opencode-ai/client/promise"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { Skill } from "@opencode-ai/schema/skill"
import type { ActiveComposerAdapter, ComposerControls, ComposerSession, NewSessionComposerAdapter } from "./adapter"
import { createMemoryComposerState } from "./state"
@@ -52,6 +53,7 @@ function submitInput(
adapter: ActiveComposerAdapter | NewSessionComposerAdapter,
notify = { missingSelection() {}, failed(_kind: "shell" | "command" | "prompt", _error: unknown) {} },
mode: "normal" | "shell" = "normal",
delivery: "steer" | "queue" = "steer",
) {
return createComposerSubmit({
adapter,
@@ -63,6 +65,7 @@ function submitInput(
setMode() {},
closePopover() {},
notify,
delivery: () => delivery,
comments: { capture: () => [], clear() {}, restore() {} },
})
}
@@ -99,6 +102,7 @@ function session(input: {
setStatus: (_sessionID, status) => input.statuses?.push(status),
prompt: async (value) => {
input.calls.push("prompt")
await value.prepare?.()
await input.prompt(value)
},
},
@@ -107,7 +111,7 @@ function session(input: {
}
describe("Composer submission", () => {
test("sends one captured value with explicit delivery after selection switches", async () => {
test("captures model selection in one shared prompt command and prepares the agent", async () => {
const state = createMemoryComposerState({ prompt: "ship it" }).capture()
const calls: string[] = []
const admitted = Promise.withResolvers<Parameters<ComposerSession["data"]["session"]["prompt"]>[0]>()
@@ -131,7 +135,8 @@ describe("Composer submission", () => {
await submitInput(adapter).submit(new Event("submit"))
const request = await admitted.promise
expect(calls).toEqual(["switch-agent", "switch-model", "prompt"])
expect(calls).toEqual(["prompt", "switch-agent"])
expect(request.model).toEqual({ id: "model-1", providerID: "provider-1", variant: "balanced" })
expect(request.delivery).toBe("steer")
expect(request.text).toBe("ship it")
expect(request.id).toMatch(/^msg_/)
@@ -170,13 +175,13 @@ describe("Composer submission", () => {
const submitted = submitInput(adapter).submit(new Event("submit"))
const request = await admitted.promise
expect(calls).toEqual(["start", "switch-agent", "switch-model", "prompt"])
expect(calls).toEqual(["start", "prompt", "switch-agent"])
expect(statuses).toEqual(["running"])
expect(promoted.current()).toMatchObject([{ type: "text", content: "restored draft" }])
cleanupReady.resolve()
await submitted
expect(calls).toEqual(["start", "switch-agent", "switch-model", "prompt", "submitted"])
expect(calls).toEqual(["start", "prompt", "switch-agent", "submitted"])
expect(request.delivery).toBe("steer")
expect(request.text).toBe("first prompt")
expect(draft.current()).toEqual([{ type: "text", content: "", start: 0, end: 0 }])
@@ -309,7 +314,7 @@ describe("Composer submission", () => {
await checked.promise
expect(state.current()).toEqual([{ type: "text", content: "", start: 0, end: 0 }])
expect(attempts).toHaveLength(2)
expect(attempts).toHaveLength(1)
expect(new Set(attempts).size).toBe(1)
})
@@ -382,7 +387,7 @@ describe("Composer submission", () => {
}
const notify = {
missingSelection() {},
failed: () => (attempts.length === 2 ? first.resolve() : second.resolve()),
failed: () => (attempts.length === 1 ? first.resolve() : second.resolve()),
}
const submission = submitInput(adapter, notify)
@@ -391,12 +396,89 @@ describe("Composer submission", () => {
await submission.submit(new Event("submit"))
await second.promise
expect(attempts).toHaveLength(4)
expect(attempts).toHaveLength(2)
expect(new Set(attempts).size).toBe(1)
expect(statuses).toEqual(["running", "idle", "running", "idle"])
expect(state.current()).toMatchObject([{ type: "text", content: "retry me" }])
})
test("queued follow-ups keep model metadata without changing the active selection", async () => {
const state = createMemoryComposerState({ prompt: "later" }).capture()
const calls: string[] = []
const admitted = Promise.withResolvers<Parameters<ComposerSession["data"]["session"]["prompt"]>[0]>()
const target = session({ calls, prompt: async (value) => admitted.resolve(value) })
const adapter: ActiveComposerAdapter = {
kind: "active-session",
state,
ready: () => true,
controls,
working: () => true,
session: () => target,
interrupt: async () => undefined,
submitted() {},
setEditor() {},
}
await submitInput(adapter, undefined, "normal", "queue").submit(new Event("submit"))
const request = await admitted.promise
expect(calls).toEqual(["prompt"])
expect(request.delivery).toBe("queue")
expect(request.model).toBeUndefined()
expect(request.prepare).toBeUndefined()
expect(request.metadata?.model).toEqual({ providerID: "provider-1", modelID: "model-1", variant: "balanced" })
})
test.each(["failed", "cancelled"] as const)(
"%s shared submission clears first-prompt handoff and busy state without restoring a duplicate draft",
async (reason) => {
const state = createMemoryComposerState({ prompt: "retained by the shared client" }).capture()
const statuses: ("idle" | "running")[] = []
const calls: string[] = []
const settled = Promise.withResolvers<void>()
const notified: unknown[] = []
const cleared: string[] = []
const target = session({
calls,
statuses,
admitted: () => {
throw new Error("An optimistic row is not an acknowledgement")
},
handoff: { set() {}, clear: (id) => cleared.push(id) },
prompt: async (value) => {
throw new PromptSubmissionError(reason, "session-1", value.id!)
},
})
target.data.session.setStatus = (_id, status) => {
statuses.push(status)
if (status === "idle") settled.resolve()
}
const adapter: NewSessionComposerAdapter = {
kind: "new-session",
state,
ready: () => true,
controls,
working: () => false,
submitted() {},
async start() {
return { session: target, cleanupReady: Promise.resolve() }
},
}
await submitInput(adapter, { missingSelection() {}, failed: (_kind, error) => notified.push(error) }).submit(
new Event("submit"),
)
await settled.promise
expect(calls.filter((call) => call === "prompt")).toHaveLength(1)
expect(statuses).toEqual(["running", "idle"])
expect(cleared).toHaveLength(1)
expect(cleared[0]).toMatch(/^msg_/)
expect(notified).toHaveLength(reason === "failed" ? 1 : 0)
expect(state.current()).toEqual([{ type: "text", content: "", start: 0, end: 0 }])
expect(state.retry.current()).toBeUndefined()
},
)
test("forwards structured mentions to custom commands", async () => {
const state = createMemoryComposerState().capture()
state.set([
+23 -24
View File
@@ -1,5 +1,6 @@
import { SessionMessage } from "@opencode-ai/schema/session-message"
import type { SessionMessageUser } from "@opencode-ai/client/promise"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { Event } from "@opencode-ai/schema/event"
import type { Accessor } from "solid-js"
import type { PromptHistoryComment } from "./history/entry"
@@ -310,31 +311,23 @@ async function sendPrompt(session: ComposerSession, value: ComposerSubmission) {
// selection applies now; a queued follow-up must not reconfigure the turn it
// waits behind, so it runs with the session selection at delivery time (the
// intended selection stays recorded in its metadata).
if (value.delivery === "steer") {
const current = session.current()
if (current?.agent !== value.selection.agent) {
await session.api.switchAgent({ sessionID: session.id, agent: value.selection.agent })
}
if (
current?.model?.providerID !== value.selection.model.providerID ||
current.model.id !== value.selection.model.modelID ||
(current.model.variant ?? "default") !== (value.selection.variant ?? "default")
) {
await session.api.switchModel({
sessionID: session.id,
model: {
id: value.selection.model.modelID,
providerID: value.selection.model.providerID,
variant: value.selection.variant,
},
})
}
}
const admission = {
await session.data.session.prompt({
id: value.id,
sessionID: session.id,
delivery: value.delivery,
...(value.delivery === "steer"
? {
model: {
id: value.selection.model.modelID,
providerID: value.selection.model.providerID,
variant: value.selection.variant,
},
prepare: async () => {
if (session.current()?.agent !== value.selection.agent)
await session.api.switchAgent({ sessionID: session.id, agent: value.selection.agent })
},
}
: {}),
text: request.text,
files: request.files.map((file) => ({ uri: file.uri, name: file.name, mention: file.mention })),
agents: request.agents,
@@ -348,8 +341,7 @@ async function sendPrompt(session: ComposerSession, value: ComposerSubmission) {
...(value.selection.variant ? { variant: value.selection.variant } : {}),
},
},
}
await session.data.session.prompt(admission).catch(() => session.data.session.prompt(admission))
})
}
async function buildSubmissionRequest(session: ComposerSession, value: ComposerSubmission) {
@@ -378,6 +370,13 @@ function failSubmission(
messageID?: string,
rollback?: () => void,
) {
if (error instanceof PromptSubmissionError) {
session.handoff?.clear(error.id)
rollback?.()
// The shared client retains failed rows for retry; cancellation discards them.
if (error.reason === "failed") input.notify.failed(kind, error)
return
}
if (messageID && session.admitted(messageID)) return
if (messageID) session.handoff?.clear(messageID)
rollback?.()
+4
View File
@@ -694,6 +694,10 @@ export const dict = {
"session.queue.remove": "Remove",
"session.queue.reorder": "Reorder queued prompt",
"session.queue.attachments": "+ attachments",
"session.submission.sending": "Sending...",
"session.submission.retrying": "Retrying...",
"session.submission.failed": "Could not confirm prompt delivery",
"session.submission.retry": "Retry",
"session.timeline.working": "Working",
"session.timeline.notice.finished": "{{actor}} finished",
"session.timeline.notice.failed": "{{actor}} failed",
@@ -11,6 +11,7 @@ import { IconButton } from "@opencode-ai/ui/icon-button"
import { Tooltip } from "@opencode-ai/ui/tooltip"
import { useLanguage } from "@/runtime/i18n/language"
import type { SessionQueueView } from "./queue"
import { PromptSubmission } from "../prompt-submission"
// Pullout above the composer listing the prompts queued behind the current
// turn. The panel slides under the composer card (negative margin, opaque
@@ -58,11 +59,7 @@ export function SessionQueuePanel(props: { queue: SessionQueueView }) {
>
{/* Keyed on row IDs so store updates move row elements instead of
remounting them, which would kill an in-flight drag. */}
<div
ref={listRef}
class="flex flex-col gap-px"
classList={{ "max-h-[131px] overflow-y-auto": count() > 3 }}
>
<div ref={listRef} class="flex flex-col gap-px" classList={{ "max-h-[131px] overflow-y-auto": count() > 3 }}>
<For each={props.queue.rows().map((row) => row.id)}>
{(id, index) => <SessionQueueRow queue={props.queue} id={id} index={index()} />}
</For>
@@ -89,7 +86,7 @@ function SessionQueueRow(props: { queue: SessionQueueView; id: string; index: nu
return props.index
},
get disabled() {
return props.queue.busy()
return props.queue.busy() || !props.queue.reorderable()
},
})
return (
@@ -110,6 +107,7 @@ function SessionQueueRow(props: { queue: SessionQueueView; id: string; index: nu
type="button"
class="grid shrink-0 cursor-grab touch-none grid-cols-2 gap-x-[2px] gap-y-[2.25px] p-1"
aria-label={language.t("session.queue.reorder")}
disabled={props.queue.busy() || !props.queue.reorderable()}
>
<For each={Array.from({ length: 6 })}>
{() => <span class="size-[2px] bg-v2-background-bg-layer-04" />}
@@ -120,7 +118,7 @@ function SessionQueueRow(props: { queue: SessionQueueView; id: string; index: nu
type="button"
data-action="session-queue-edit"
dir="auto"
disabled={props.queue.busy()}
disabled={props.queue.busy() || !!props.queue.submission(props.id)}
class="max-w-full min-w-0 self-start truncate rounded-sm text-start text-[13px] font-[440] leading-[var(--line-height-compact)]"
classList={{
"text-v2-text-text-faint": editing(),
@@ -135,6 +133,7 @@ function SessionQueueRow(props: { queue: SessionQueueView; id: string; index: nu
{language.t("session.queue.attachments")}
</span>
</Show>
<PromptSubmission sessionID={props.queue.sessionID} id={props.id} />
</div>
</div>
<div
@@ -146,7 +145,7 @@ function SessionQueueRow(props: { queue: SessionQueueView; id: string; index: nu
"pointer-events-none": props.queue.busy(),
}}
>
<Show when={!editing()}>
<Show when={!editing() && !props.queue.submission(props.id)}>
<Tooltip
placement="top"
inactive={!props.queue.working()}
@@ -166,18 +165,20 @@ function SessionQueueRow(props: { queue: SessionQueueView; id: string; index: nu
</Button>
</Tooltip>
</Show>
<Tooltip placement="top" value={language.t("session.queue.remove")}>
<IconButton
data-action="session-queue-remove"
type="button"
size="small"
variant="ghost-muted"
icon={<Icon name="outline-xmark" />}
disabled={props.queue.busy()}
aria-label={language.t("session.queue.remove")}
onClick={() => void props.queue.remove(props.id)}
/>
</Tooltip>
<Show when={!props.queue.submission(props.id)}>
<Tooltip placement="top" value={language.t("session.queue.remove")}>
<IconButton
data-action="session-queue-remove"
type="button"
size="small"
variant="ghost-muted"
icon={<Icon name="outline-xmark" />}
disabled={props.queue.busy()}
aria-label={language.t("session.queue.remove")}
onClick={() => void props.queue.remove(props.id)}
/>
</Tooltip>
</Show>
</div>
</div>
)}
+34 -8
View File
@@ -2,6 +2,7 @@ import { createEffect, createMemo, onCleanup, type Accessor } from "solid-js"
import { createStore } from "solid-js/store"
import { useMutation } from "@tanstack/solid-query"
import type { SessionInboxInfo } from "@opencode-ai/client/promise"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { SessionMessage } from "@opencode-ai/schema/session-message"
import type { ComposerDelivery } from "@/composer/adapter"
import type { ComposerModel } from "@/composer/model"
@@ -38,6 +39,13 @@ export function createSessionQueue(input: {
const language = useLanguage()
const [state, setState] = createStore<{ editing?: { id: string; stash: EditStash } }>({})
const notify = () => showToast({ title: language.t("common.requestFailed") })
// A standalone row retry cannot finish the surrounding cancel/reorder workflow.
const admitReplacement = (request: Parameters<typeof data.session.prompt>[0]) =>
data.session.prompt(request).catch(async (error: unknown) => {
if (error instanceof PromptSubmissionError && error.reason === "failed")
await data.session.pending.cancel(error.sessionID, error.id)
throw error
})
const mutation = useMutation(() => ({
mutationFn: async (
change:
@@ -62,13 +70,13 @@ export function createSessionQueue(input: {
change.text,
)
// Admit before cancelling so a failed replacement never discards the original.
const admitted = await data.session.prompt({
const admitted = await admitReplacement({
...replacement,
id: change.replacement,
delivery: change.delivery,
...(change.delivery === "queue" ? { resume: false } : {}),
})
await server.api.session.inbox.cancel({ sessionID: input.sessionID, inboxID: change.original })
await data.session.pending.cancel(input.sessionID, change.original)
cancelEdit()
if (change.delivery === "queue")
await rewrite(change.inboxIDs.map((id) => (id === change.original ? admitted.id : id)))
@@ -82,6 +90,9 @@ export function createSessionQueue(input: {
.list(input.sessionID)
.filter((item): item is QueuedPrompt => item.type === "user" && item.delivery === "queue"),
)
const submission = (id: string) => data.session.submission.get(input.sessionID, id)
// Reordering replaces IDs; unconfirmed admissions must keep theirs for retry.
const reorderable = () => !queued().some((item) => submission(item.id))
const rows = createMemo(() => {
const replacement = mutation.isPending ? mutation.variables : undefined
return queuedPromptRows(
@@ -108,8 +119,9 @@ export function createSessionQueue(input: {
if (changed < 0) return
// Existing inbox APIs cannot reorder rows, so replace only the changed suffix.
// This is not atomic: earlier replacements remain if a later admission fails.
for (const item of ordered.slice(changed)) {
await data.session.prompt({
await admitReplacement({
sessionID: input.sessionID,
text: item.payload.text,
files: item.payload.files?.map((file) => ({
@@ -126,24 +138,25 @@ export function createSessionQueue(input: {
})
}
for (const item of current.slice(changed)) {
await server.api.session.inbox.cancel({ sessionID: input.sessionID, inboxID: item.id })
await data.session.pending.cancel(input.sessionID, item.id)
}
}
const steer = (id: string) => {
if (submission(id)) return Promise.resolve()
if (state.editing?.id === id) cancelEdit()
return server.api.session.inbox.steer({ sessionID: input.sessionID, inboxID: id }).catch(() => notify())
}
const remove = (id: string) => {
if (state.editing?.id === id) cancelEdit()
return server.api.session.inbox.cancel({ sessionID: input.sessionID, inboxID: id }).catch(() => notify())
return data.session.pending.cancel(input.sessionID, id).catch(() => notify())
}
const reorder = (inboxIDs: string[]) => {
if (mutation.isPending) return Promise.resolve()
if (mutation.isPending || !reorderable()) return Promise.resolve()
return mutation.mutateAsync({ type: "reorder", inboxIDs }).catch(() => undefined)
}
const edit = (id: string) => {
if (mutation.isPending) return false
if (mutation.isPending || submission(id)) return false
if (state.editing?.id === id) return true
const item = queued().find((entry) => entry.id === id)
if (!item) return false
@@ -202,6 +215,9 @@ export function createSessionQueue(input: {
}
return {
sessionID: input.sessionID,
submission,
reorderable,
count: () => queued().length,
delivery: () => (input.working() ? input.behavior() : "steer"),
alternate: () => {
@@ -228,7 +244,17 @@ export type SessionQueue = ReturnType<typeof createSessionQueue>
// The slice of the queue the panel renders and drives.
export type SessionQueueView = Pick<
SessionQueue,
"rows" | "editing" | "working" | "busy" | "steer" | "remove" | "edit" | "reorder"
| "sessionID"
| "submission"
| "reorderable"
| "rows"
| "editing"
| "working"
| "busy"
| "steer"
| "remove"
| "edit"
| "reorder"
>
export function queuedPromptRows(items: QueuedPrompt[], replacement?: { original: string; replacement: string }) {
@@ -0,0 +1,52 @@
import { Show } from "solid-js"
import { createStore } from "solid-js/store"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { Button } from "@opencode-ai/ui/button"
import { useLanguage } from "@/runtime/i18n/language"
import { useData } from "@/runtime/server/current"
import { showToast } from "@/shell/notifications/toast"
export function PromptSubmission(props: { sessionID: string; id: string; class?: string }) {
const data = useData()
const language = useLanguage()
const [state, setState] = createStore({ cancelling: false })
const submission = () => data.session.submission.get(props.sessionID, props.id)
const failed = (error: unknown) => {
if (error instanceof PromptSubmissionError) return
showToast({ title: language.t("common.requestFailed") })
}
const cancel = () => {
setState("cancelling", true)
void data.session.pending
.cancel(props.sessionID, props.id)
.catch(failed)
.finally(() => setState("cancelling", false))
}
return (
<Show when={submission()}>
{(submission) => (
<div
data-component="prompt-submission"
data-submission-id={props.id}
class={`flex min-h-7 flex-wrap items-center gap-x-2 text-[13px] leading-[var(--line-height-compact)] text-v2-text-text-muted ${props.class ?? ""}`}
>
<span role="status">{language.t(`session.submission.${submission().status}`)}</span>
<Show when={submission().status === "failed"}>
<Button
type="button"
size="small"
variant="ghost-muted"
disabled={state.cancelling}
onClick={() => void data.session.submission.retry(props.sessionID, props.id).catch(failed)}
>
{language.t("session.submission.retry")}
</Button>
</Show>
<Button type="button" size="small" variant="ghost-muted" disabled={state.cancelling} onClick={cancel}>
{language.t("common.cancel")}
</Button>
</div>
)}
</Show>
)
}
@@ -32,6 +32,7 @@ import { parseCommentNote, readPromptPresentation } from "@/composer/comment-not
import { useCommand } from "@/shell/commands/command"
import { useSettings } from "@/settings/model"
import { SessionTitleHeader } from "../session-identity-header"
import { PromptSubmission } from "../prompt-submission"
type BackgroundTask = {
id: string
@@ -630,7 +631,20 @@ function MessageTimelineView(
const content = Timeline.resolveContent(messageByID().get(row.group.ref.messageID), row.group.ref.partID)
return content?.type === "tool" && ["edit", "write"].includes(content.name)
}}
renderRow={(row, onSizeChange) => <rowRenderer.Row row={row} onSizeChange={onSizeChange} />}
renderRow={(row, onSizeChange) => (
<>
<rowRenderer.Row row={row} onSizeChange={onSizeChange} />
<Show when={row()._tag === "UserMessage" && sessionID()}>
{(id) => (
<PromptSubmission
sessionID={id()}
id={row().userMessageID}
class={`${turnPadding()} ${props.centered ? "md:max-w-[1000px] md:mx-auto" : ""}`}
/>
)}
</Show>
</>
)}
header={
<Show when={!props.hideHeader}>
<SessionTitleHeader>
@@ -0,0 +1,119 @@
# Optimistic Submissions
The client uses two internal, dependency-free modules for local submissions.
Neither imports Solid, Effect, QUARK, or the HTTP client. These are not a new
query cache or a separately published library.
- `Command` owns captured input, ordered execution, retry attempts, and external confirmation.
- `Optimistic` owns local contributions and composes them with authoritative collections.
- `solid/data.ts` translates server events, chooses retry policy, and bridges changes into Solid.
## Commands
```ts
const queue = Command.queue()
const send = Command.make({
key: (input: PromptInput) => input.id,
group: (input) => input.sessionID,
queue,
execute:
(input) =>
async ({ signal }) =>
api.session.prompt(input, { signal }),
retry: {
delays: [250, 750, 1500],
when: isRetryableAdmissionFailure,
},
})
const operation = send.submit(input, { wait: sessionCreation })
await operation.request
```
Input is cloned once at submission. Reusing a key in the same group returns the
existing operation and ignores the new payload; cross-group reuse fails. A
manual `operation.retry()` joins active work or restarts a failed operation with
the original input. It does not create a new remote identity.
Commands sharing a queue serialize within each group. Different groups execute
independently. In OpenCode, prompts and compactions share the session's admission
queue. Cancelling an item waiting behind another item cannot let its successors
bypass the earlier item.
`execute(input)` is a per-cycle factory: it runs once initially and once for
each explicit manual retry. Its returned function runs for each automatic
attempt. This lets model selection succeed once during an automatic retry
cycle, while a manual retry reapplies the captured model after intervening user
changes. The separate `prepare` submission option runs at most once and is never
automatically retried.
## Confirmation Is Not HTTP Settlement
| Observation | Command State | Local Contribution |
| ------------------------------------------ | ------------------------------ | ----------------------------------- |
| Request starts | `sending` | Visible |
| Transient failure with attempts remaining | `retrying` | Visible |
| Attempts exhausted | `failed` | Retained for explicit retry/cancel |
| HTTP returns canonical payload | `accepted` | Retained until projected |
| Canonical event or positive read | Accepted and retired | Removed after canonical publication |
| Definitive rejection or local cancellation | Rejected/cancelled and retired | Removed |
```ts
// Publish the canonical item before retiring its local contribution.
batch(() => {
publishCanonical(item)
send.confirm(item.id, item)
})
```
Confirmation resolves active waiters and interrupts outstanding local work.
A subsequent HTTP rejection cannot overturn it. A caller already holding a
settled failed Promise still has that failure; the operation's current request
can expose later canonical confirmation.
Delivery events may prove acceptance without carrying the canonical payload.
`confirmFrom(key, load)` records this positive proof and obtains the actual
payload. A failed lookup cannot turn a proven admission into a definitive
rollback, and a later observation can retry the lookup. OpenCode additionally
reloads ordered history rather than appending a late message lookup behind its
assistant response.
## Views, Not Rollbacks
```ts
const local = Optimistic.make<InboxItem>({
key: (item) => item.id,
group: (item) => item.sessionID,
})
local.set(preview)
const visible = Optimistic.merge(serverItems, local.list(sessionID), (item) => item.id)
```
Canonical items win by ID. The merge preserves their order and object references,
then appends unmatched local contributions in submission order. When there are
no unmatched local contributions, it returns the canonical array itself.
Observation must retire local contributions explicitly. Merely hiding a local
row behind a matching canonical ID would allow it to reappear after cache
eviction. Conversely, absence from an inbox or history snapshot is not evidence
that a local submission failed.
The Solid adapter maintains per-session local preview arrays independently of
submission status. Retrying does not rebuild the transcript, and canonical
assistant/tool/text proxies keep their existing fine-grained subscriptions.
Local prompt and compaction contributions share the same overlay so their
relative submission order is preserved.
## Ownership
Client ownership, not component observation, controls operation lifetime.
Unsubscribing does not cancel work. Ordinary session cache eviction preserves
unconfirmed contributions. Session deletion and client disposal cancel them.
`cancel` only stops local work. OpenCode separately calls the server's inbox
cancellation endpoint when admission may have been attempted. Neither aborting a
fetch nor disposing a client proves that a remote write was undone.
This module does not establish the cause of the original lost-submission report.
Same-ID retries and retained local state are mitigations, not incident resolution.
+324
View File
@@ -0,0 +1,324 @@
export namespace Command {
export type State = Readonly<
{ attempt: number; stage: string } & (
| { status: "sending" | "accepted" }
| { status: "retrying"; delay: number; error: unknown }
| { status: "failed" | "rejected" | "cancelled"; error: unknown }
)
>
export type Context = {
signal: AbortSignal
attempt: number
stage(name: string): void
}
export type Operation<Input, Output> = {
readonly key: string
readonly group: string
readonly input: Input
readonly createdAt: number
readonly signal: AbortSignal
readonly state: State
readonly accepted: Output | undefined
readonly started: boolean
readonly request: Promise<Output>
retry(): Promise<Output>
cancel(): void
confirm(output: Output): void
confirmFrom(load: (signal: AbortSignal) => Promise<Output>): void
}
export type Event<Input, Output> = { operation: Operation<Input, Output>; removed: boolean }
export type Queue = ReturnType<typeof queue>
export class Error extends globalThis.Error {
constructor(
readonly reason: "cancelled" | "failed",
readonly key: string,
options?: ErrorOptions,
) {
super(`Command ${key} ${reason}`, options)
this.name = "Command.Error"
}
}
export function queue() {
const tails = new Map<string, Promise<unknown>>()
return {
pending: (group: string) => tails.has(group),
run<A>(group: string, fn: () => Promise<A>, options?: { wait?: Promise<unknown>; signal?: AbortSignal }) {
const previous = tails.get(group)
const ready = Promise.all([previous, options?.wait])
const request = interrupt(
ready.then(() => {
options?.signal?.throwIfAborted()
return fn()
}),
options?.signal,
)
// Cancellation releases this caller, but cannot release an earlier queue owner.
const tail = Promise.allSettled([previous, request]).then(() => undefined)
tails.set(group, tail)
void tail.then(() => {
if (tails.get(group) === tail) tails.delete(group)
})
return request
},
}
}
export function make<Input, Output>(config: {
key: (input: Input) => string
group: (input: Input) => string
queue: Queue
execute: (input: Input) => (context: Context) => Promise<Output>
retry?: { delays: readonly number[]; when: (error: unknown) => boolean }
}) {
const operations = new Map<string, Operation<Input, Output>>()
const listeners = new Set<(event: Event<Input, Output>) => void>()
let disposed = false
return {
get: (key: string) => operations.get(key),
values: (): readonly Operation<Input, Output>[] => Array.from(operations.values()),
confirm: (key: string, output: Output) => operations.get(key)?.confirm(output),
confirmFrom: (key: string, load: (signal: AbortSignal) => Promise<Output>) =>
operations.get(key)?.confirmFrom(load),
cancel: (key: string) => operations.get(key)?.cancel(),
subscribe(listener: (event: Event<Input, Output>) => void) {
listeners.add(listener)
return () => {
listeners.delete(listener)
}
},
dispose() {
disposed = true
operations.forEach((operation) => operation.cancel())
listeners.clear()
},
submit(value: Input, options?: { wait?: Promise<unknown>; prepare?: () => Promise<unknown> }) {
if (disposed) throw new globalThis.Error("Command manager is disposed")
const input = structuredClone(value)
const key = config.key(input)
const group = config.group(input)
const existing = operations.get(key)
if (existing) {
if (existing.group !== group) throw new globalThis.Error(`Command ${key} belongs to another group`)
return existing
}
const abort = new AbortController()
let state: State = { status: "sending", attempt: 0, stage: "waiting" }
let accepted: Output | undefined
let started = false
let proof = false
let preparation: Promise<unknown> | undefined
let confirmation: Promise<void> | undefined
let cycle = deferred<Output>()
function notify(removed = false) {
listeners.forEach((listener) => listener({ operation, removed }))
}
function update(next: State) {
state = next
notify()
}
function isRunning() {
return state.status === "sending" || state.status === "retrying"
}
function retire(next: State) {
state = next
operations.delete(key)
abort.abort()
notify()
notify(true)
}
const operation: Operation<Input, Output> = {
key,
group,
input,
createdAt: Date.now(),
signal: abort.signal,
get state() {
return state
},
get accepted() {
return accepted
},
get started() {
return started
},
get request() {
return cycle.promise
},
retry() {
if (state.status !== "failed") return cycle.promise
cycle = deferred<Output>()
run()
return cycle.promise
},
cancel() {
if (abort.signal.aborted) return
const error = new Error("cancelled", key)
if (!isRunning()) cycle = deferred<Output>()
cycle.reject(error)
retire({ status: "cancelled", attempt: state.attempt, stage: state.stage, error })
},
confirm(output) {
if (abort.signal.aborted) return
accepted = output
if (!isRunning()) cycle = deferred<Output>()
cycle.resolve(output)
retire({ status: "accepted", attempt: state.attempt, stage: state.stage })
},
confirmFrom(load) {
if (abort.signal.aborted || confirmation) return
proof = true
confirmation = interrupt(
Promise.resolve().then(() => {
abort.signal.throwIfAborted()
return load(abort.signal)
}),
abort.signal,
).then(
(output) => {
confirmation = undefined
operation.confirm(output)
},
() => {
// Keep proof, not a permanently failed lookup: a later echo can retry it.
confirmation = undefined
},
)
},
}
function run() {
state = { status: "sending", attempt: 0, stage: "waiting" }
const current = cycle
let exhausted = false
const request = config.queue.run(
group,
async () => {
update({ status: "sending", attempt: 0, stage: "prepare" })
preparation ??= Promise.resolve().then(() => {
abort.signal.throwIfAborted()
return options?.prepare?.()
})
await interrupt(preparation, abort.signal)
abort.signal.throwIfAborted()
const execute = config.execute(input)
for (let attempt = 1; ; attempt++) {
abort.signal.throwIfAborted()
update({ status: "sending", attempt, stage: "execute" })
try {
return await interrupt(
Promise.resolve().then(() => {
abort.signal.throwIfAborted()
started = true
const result = execute({
signal: abort.signal,
attempt,
stage(name) {
if (!abort.signal.aborted && isRunning() && cycle === current)
update({ ...state, stage: name })
},
})
if (!abort.signal.aborted) notify()
return result
}),
abort.signal,
)
} catch (error) {
abort.signal.throwIfAborted()
if (confirmation) await interrupt(confirmation, abort.signal)
abort.signal.throwIfAborted()
if (!proof && !config.retry?.when(error)) throw error
const delay = config.retry?.delays[attempt - 1]
if (delay === undefined) {
exhausted = true
throw error
}
update({ status: "retrying", attempt, stage: state.stage, delay, error })
await pause(delay, abort.signal)
}
}
},
{ wait: options?.wait, signal: abort.signal },
)
void request.then(
(output) => {
if (abort.signal.aborted) return
accepted = output
current.resolve(output)
update({ status: "accepted", attempt: state.attempt, stage: state.stage })
},
(cause: unknown) => {
if (abort.signal.aborted) return
const error = exhausted || proof ? new Error("failed", key, { cause }) : cause
current.reject(error)
const next: State = {
status: exhausted || proof ? "failed" : "rejected",
attempt: state.attempt,
stage: state.stage,
error,
}
if (next.status === "rejected") return retire(next)
update(next)
},
)
notify()
}
operations.set(key, operation)
run()
return operation
},
}
}
function deferred<A>() {
const result = Promise.withResolvers<A>()
void result.promise.catch(() => undefined)
return result
}
function interrupt<A>(promise: Promise<A>, signal?: AbortSignal): Promise<A> {
if (!signal) return promise
return new Promise<A>((resolve, reject) => {
const abort = () => reject(signal.reason)
if (signal.aborted) abort()
else signal.addEventListener("abort", abort, { once: true })
void promise.then(
(value) => {
signal.removeEventListener("abort", abort)
resolve(value)
},
(error) => {
signal.removeEventListener("abort", abort)
reject(error)
},
)
})
}
function pause(delay: number, signal: AbortSignal) {
return new Promise<void>((resolve, reject) => {
const abort = () => {
clearTimeout(timer)
reject(signal.reason)
}
const timer = setTimeout(() => {
signal.removeEventListener("abort", abort)
resolve()
}, delay)
if (signal.aborted) abort()
else signal.addEventListener("abort", abort, { once: true })
})
}
}
+61
View File
@@ -0,0 +1,61 @@
export namespace Optimistic {
/** Local contributions never write into the authoritative collection. */
export function make<A extends object>(options: { key: (value: A) => string; group: (value: A) => string }) {
const values = new Map<string, A>()
const groups = new Map<string, readonly A[]>()
const listeners = new Set<(group: string, values: readonly A[]) => void>()
const empty: readonly A[] = []
const publish = (group: string, next: readonly A[]) => {
if (next.length) groups.set(group, next)
else groups.delete(group)
listeners.forEach((listener) => listener(group, next))
}
return {
get: (key: string) => values.get(key),
has: (key: string) => values.has(key),
list: (group: string) => groups.get(group) ?? empty,
set(value: A) {
const key = options.key(value)
const group = options.group(value)
const previous = values.get(key)
if (previous === value) return
if (previous && options.group(previous) !== group) throw new Error("Optimistic key belongs to another group")
values.set(key, value)
const current = groups.get(group) ?? empty
publish(
group,
previous ? current.map((item) => (options.key(item) === key ? value : item)) : [...current, value],
)
},
remove(key: string) {
const value = values.get(key)
if (!value) return false
values.delete(key)
const group = options.group(value)
publish(
group,
(groups.get(group) ?? empty).filter((item) => options.key(item) !== key),
)
return true
},
clear(group: string) {
const current = groups.get(group)
if (!current) return
current.forEach((item) => values.delete(options.key(item)))
publish(group, empty)
},
subscribe(listener: (group: string, values: readonly A[]) => void) {
listeners.add(listener)
return () => listeners.delete(listener)
},
}
}
/** Preserve authoritative order and object identity; append only unmatched local values. */
export function merge<A>(canonical: A[], local: readonly A[], key: (value: A) => string): A[] {
if (!local.length) return canonical
const known = new Set(canonical.map(key))
const missing = local.filter((value) => !known.has(key(value)))
return missing.length ? [...canonical, ...missing] : canonical
}
}
+290 -126
View File
@@ -29,6 +29,7 @@ import type {
SessionMessageAssistantTool,
SessionInfo,
SessionInboxInfo,
SessionInboxUser,
SessionInboxCompaction,
ShellInfo,
SkillInfo,
@@ -44,13 +45,34 @@ import {
isFormAlreadySettledError,
isFormNotFoundError,
isPermissionNotFoundError,
isConflictError,
isSessionNotFoundError,
type SessionPromptInput,
} from "../promise"
import { createStore, produce, reconcile } from "solid-js/store"
import { createStore, produce, reconcile, unwrap } from "solid-js/store"
import { Command } from "../command"
import { Optimistic } from "../optimistic"
import { promptFailure } from "./prompt-retry"
import type { SessionInbox } from "@opencode-ai/schema/session-inbox"
import { batch, createEffect, createMemo, createSignal, onCleanup } from "solid-js"
export type DataSessionStatus = "idle" | "running"
export type PromptSubmission = {
readonly id: string
readonly sessionID: string
readonly status: "sending" | "retrying" | "failed"
readonly attempt: number
}
export class PromptSubmissionError extends Error {
constructor(
readonly reason: "cancelled" | "failed",
readonly sessionID: string,
readonly id: string,
) {
super(reason === "cancelled" ? "Prompt submission cancelled" : "Could not confirm prompt delivery")
}
}
type OpenCodeEventMap = { [Type in OpenCodeEvent["type"]]: Extract<OpenCodeEvent, { type: Type }> }
export type CreateDataInput = {
@@ -66,6 +88,7 @@ export type CreateDataInput = {
readonly connection?: {
readonly status: () => "connected" | "connecting" | "reconnecting"
}
readonly log?: { readonly info?: (message: string, data?: Readonly<Record<string, unknown>>) => void }
}
const messageIDFromEvent = (eventID: string) => eventID.replace(/^evt_/, "msg_")
@@ -212,6 +235,8 @@ export function createData(config: CreateDataInput) {
})
const [defaultLocation, setDefaultLocation] = createSignal<LocationRef>({ directory: config.directory })
const [submissions, setSubmissions] = createStore<Record<string, PromptSubmission | undefined>>({})
const [local, setLocal] = createStore<Record<string, readonly LocalAdmission[]>>({})
const sessions = createMemo(() =>
Object.values(store.session.info).toSorted((a, b) => b.time.updated - a.time.updated),
)
@@ -286,12 +311,18 @@ export function createData(config: CreateDataInput) {
setStore("session", "pending", sessionID, index, { ...item, delivery })
}
// Inbox IDs of optimistic admissions awaiting acknowledgement, so rejection
// only rolls back unacknowledged rows and a pending re-fetch cannot wipe a
// row the server does not know about yet. Prompts clear on their durable
// echo, positive pending read, or rollback; compactions also reconcile the
// POST's canonical ID.
const outbox = new Set<string>()
type LocalAdmission = {
id: string
sessionID: string
item: SessionInboxInfo
message?: SessionMessageInfo
}
const optimistic = Optimistic.make<LocalAdmission>({ key: (item) => item.id, group: (item) => item.sessionID })
const stopOptimistic = optimistic.subscribe((sessionID, items) => setLocal(sessionID, reconcile(items)))
function stageLocal(item: SessionInboxInfo) {
optimistic.set({ id: item.id, sessionID: item.sessionID, item, message: inboxMessage(item) })
}
// Session IDs of optimistic create admissions still awaiting acknowledgement
// (the session.created echo or the create response itself). A failed create
@@ -304,10 +335,95 @@ export function createData(config: CreateDataInput) {
// to exist server-side instead of failing with "not found".
const creating = new Map<string, Promise<unknown>>()
// Per-session send chain: prompts and compactions must be admitted in
// submission order. Each waits for the previous POST to settle, so one
// failure does not block the next.
const sending = new Map<string, Promise<unknown>>()
const admissions = Command.queue()
const prompts = Command.make({
key: (input: { request: SessionPromptInput & { id: string }; model?: ModelRef }) => input.request.id,
group: (input) => input.request.sessionID,
queue: admissions,
execute: (input) => {
// Automatic retries retain successful setup; manual retry reapplies the
// captured selection after any intervening user changes.
let selected = !input.model
return async (context) => {
if (!selected && input.model) {
context.stage("model")
await api().session.switchModel(
{ sessionID: input.request.sessionID, model: input.model },
{ signal: context.signal },
)
selected = true
}
context.signal.throwIfAborted()
context.stage("prompt")
return api().session.prompt(input.request, { signal: context.signal })
}
},
retry: { delays: [250, 750, 1500], when: (error) => promptFailure(error).retryable },
})
const stopPrompts = prompts.subscribe(({ operation, removed }) => {
batch(() => {
const state = operation.state
const status = state.status
setSubmissions(
operation.key,
!removed && (status === "sending" || status === "retrying" || status === "failed")
? { id: operation.key, sessionID: operation.group, status, attempt: state.attempt }
: undefined,
)
if (removed) optimistic.remove(operation.key)
if (removed && status !== "accepted") return
if (
status === "accepted" &&
operation.accepted &&
prompts.get(operation.key) === operation &&
optimistic.has(operation.key)
)
stageLocal(operation.accepted)
config.log?.info?.("prompt submission", {
sessionID: operation.group,
messageID: operation.key,
stage: state.stage,
attempt: state.attempt,
outcome: removed ? "confirmed" : status === "sending" ? (state.attempt === 0 ? "started" : "attempt") : status,
...("error" in state ? promptFailure(state.error) : {}),
...("delay" in state ? { delay: state.delay } : {}),
})
})
})
onCleanup(() => {
prompts.dispose()
stopPrompts()
stopOptimistic()
})
function promptResult(sessionID: string, id: string, request: Promise<SessionInboxUser>) {
return request.catch((error: unknown) => {
if (error instanceof Command.Error) throw new PromptSubmissionError(error.reason, sessionID, id)
throw error
})
}
function confirmAdmission(item: SessionInboxInfo) {
const operation = prompts.get(item.id)
if (item.type === "user" && operation?.group === item.sessionID) operation.confirm(item)
if (optimistic.get(item.id)?.sessionID === item.sessionID) optimistic.remove(item.id)
}
function observeMessages(sessionID: string, items: readonly SessionMessageInfo[]) {
items.forEach((item) => {
if (item.type !== "user") return
removePending(sessionID, item.id)
const { id, type, time, ...payload } = item
confirmAdmission({
id,
type,
sessionID,
payload,
timeCreated: time.created,
delivery: prompts.get(id)?.input.request.delivery ?? "steer",
})
})
}
const messageLoads = new Map<string, Promise<unknown>>()
const compacting = new Map<string, { id: string; observed: Set<string>; request: Promise<SessionInboxCompaction> }>()
onCleanup(() => compacting.clear())
@@ -322,27 +438,8 @@ export function createData(config: CreateDataInput) {
void promise.then(settle, settle)
}
// Capture creation before settlement clears its entry, so dependent RPCs still see a failed create.
function sendAdmission<Value>(sessionID: string, send: () => Promise<Value>, gate?: Promise<unknown>) {
const created = creating.get(sessionID)
const previous = sending.get(sessionID)
const request = Promise.resolve()
.then(() => Promise.all([gate, created, previous]))
.then(send)
track(
sending,
sessionID,
request.catch(() => undefined),
)
return request
}
// 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 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) {
// Only server-confirmed inbox items enter the canonical projection.
function admitCanonical(item: SessionInboxInfo) {
batch(() => {
const pending = store.session.pending[item.sessionID] ?? []
const at = pending.findIndex((entry) => entry.id === item.id)
@@ -360,21 +457,24 @@ export function createData(config: CreateDataInput) {
}
function materializeInboxMessage(item: SessionInboxInfo) {
if (item.type !== "user" && item.type !== "synthetic") return
const row = inboxMessage(item)
if (!row) 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
})
}
function inboxMessage(item: SessionInboxInfo): SessionMessageInfo | undefined {
if (item.type === "user") return { id: item.id, type: "user", ...item.payload, time: { created: item.timeCreated } }
if (item.type === "synthetic")
return { id: item.id, type: "synthetic", ...item.payload, time: { created: item.timeCreated } }
}
// 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) {
// Only server cancellation removes authoritative transcript state.
function retractCanonical(sessionID: string, inboxID: string) {
batch(() => {
removePending(sessionID, inboxID)
if (!messageIndex.get(sessionID)?.has(inboxID)) return
@@ -498,11 +598,8 @@ export function createData(config: CreateDataInput) {
sync.invalidate(`session.pending:${sessionID}`)
sync.invalidate(`session.message:${sessionID}`)
messageLoads.delete(sessionID)
// Keep unacknowledged submissions until their echo or rollback settles them.
const pending = store.session.pending[sessionID]?.filter((item) => outbox.has(item.id)) ?? []
const messages = store.session.message[sessionID]?.filter((item) => outbox.has(item.id)) ?? []
// Local submissions belong to the client, not to the evictable read cache.
messageIndex.delete(sessionID)
if (messages.length) messageIndex.set(sessionID, new Map(messages.map((item, index) => [item.id, index])))
setStore(
"session",
produce((draft) => {
@@ -511,18 +608,16 @@ export function createData(config: CreateDataInput) {
delete draft.messageLoading[sessionID]
delete draft.pending[sessionID]
delete draft.input[sessionID]
if (messages.length) draft.message[sessionID] = messages
if (pending.length) {
draft.pending[sessionID] = pending
draft.input[sessionID] = pending.filter((item) => item.type !== "compaction").map((item) => item.id)
}
}),
)
}
function removeSession(sessionID: string) {
prompts.values().forEach((operation) => {
if (operation.group === sessionID) operation.cancel()
})
optimistic.clear(sessionID)
activeUpdates?.set(sessionID, undefined)
store.session.pending[sessionID]?.forEach((item) => outbox.delete(item.id))
messageIndex.delete(sessionID)
sync.invalidate(`session:${sessionID}`)
sync.invalidate(`session.family:${sessionID}`)
@@ -706,6 +801,33 @@ export function createData(config: CreateDataInput) {
return
}
case "session.inbox.delivered": {
const active = prompts.get(event.data.inboxID)
const operation = active?.group === event.data.sessionID ? active : undefined
const pending = store.session.pending[event.data.sessionID]?.find((item) => item.id === event.data.inboxID)
if (pending?.type === "user") confirmAdmission(pending)
if (operation && !pending) {
prompts.confirmFrom(event.data.inboxID, async (signal) => {
// Delivery proves acceptance, not the payload of a local preview.
const row = await api().session.message(
{ sessionID: event.data.sessionID, messageID: event.data.inboxID },
{ signal },
)
if (row.type !== "user") throw new Error("Delivered prompt is not a user message")
// Hydrate the server's order rather than appending a late lookup
// behind an assistant that may already have streamed its response.
result.session.message.invalidate(event.data.sessionID)
await result.session.message.sync(event.data.sessionID)
const { id, type, time, ...payload } = row
return {
id,
type,
sessionID: event.data.sessionID,
timeCreated: time.created,
payload,
delivery: operation.input.request.delivery ?? "steer",
}
})
}
const admitted = store.session.input[event.data.sessionID]?.includes(event.data.inboxID) ?? false
removePending(event.data.sessionID, event.data.inboxID)
message.update(event.data.sessionID, (draft, index) => {
@@ -725,23 +847,29 @@ export function createData(config: CreateDataInput) {
updatePending(event.data.sessionID, event.data.inboxID, event.data.delivery)
return
case "session.inbox.cancelled": {
retractLocal(event.data.sessionID, event.data.inboxID)
const operation = prompts.get(event.data.inboxID)
if (operation?.group === event.data.sessionID) operation.cancel()
if (optimistic.get(event.data.inboxID)?.sessionID === event.data.sessionID)
optimistic.remove(event.data.inboxID)
retractCanonical(event.data.sessionID, event.data.inboxID)
compacting.get(event.data.sessionID)?.observed.add(event.data.inboxID)
return
}
case "session.inbox.enqueued": {
outbox.delete(event.data.inboxID)
admitLocal({
const item = {
id: event.data.inboxID,
sessionID: event.data.sessionID,
timeCreated: event.created,
...event.data.item,
}
batch(() => {
admitCanonical(item)
confirmAdmission(item)
})
if (event.data.item.type === "compaction") {
const active = compacting.get(event.data.sessionID)
active?.observed.add(event.data.inboxID)
if (active && active.id !== event.data.inboxID && outbox.delete(active.id))
removePending(event.data.sessionID, active.id)
if (active && active.id !== event.data.inboxID) optimistic.remove(active.id)
}
return
}
@@ -1323,40 +1451,61 @@ export function createData(config: CreateDataInput) {
},
input: {
list(sessionID: string) {
return store.session.input[sessionID] ?? []
return Optimistic.merge(
store.session.input[sessionID] ?? [],
(local[sessionID] ?? []).filter((entry) => entry.item.type !== "compaction").map((entry) => entry.id),
(id) => id,
)
},
has(sessionID: string, inboxID: string) {
return store.session.input[sessionID]?.includes(inboxID) ?? false
return (
store.session.input[sessionID]?.includes(inboxID) ||
(local[sessionID] ?? []).some((entry) => entry.id === inboxID && entry.item.type !== "compaction")
)
},
},
pending: {
async cancel(sessionID: string, inboxID: string) {
const operation = prompts.get(inboxID)
const active = operation?.group === sessionID ? operation : undefined
const attempted = active?.started || !optimistic.has(inboxID)
active?.cancel()
if (active && !attempted) return
await api()
.session.inbox.cancel({ sessionID, inboxID })
.catch((error: unknown) => {
if (active && (isConflictError(error) || isSessionNotFoundError(error))) return
throw error
})
},
list(sessionID: string) {
return store.session.pending[sessionID] ?? []
return Optimistic.merge(
store.session.pending[sessionID] ?? [],
(local[sessionID] ?? []).map((entry) => entry.item),
(item) => item.id,
)
},
sync(sessionID: string) {
return sync.run(`session.pending:${sessionID}`, async () => {
const pending = await api().session.inbox.list({ sessionID })
// A positive read acknowledges admission even when its SSE echo is delayed.
pending.forEach((item) => outbox.delete(item.id))
// Compactions also coalesce by Session, not just by the proposed ID.
if (pending.some((item) => item.type === "compaction"))
store.session.pending[sessionID]
?.filter((item) => item.type === "compaction")
.forEach((item) => outbox.delete(item.id))
// 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))
const merged = inflight.length === 0 ? pending : [...pending, ...inflight]
batch(() => {
setStore("session", "pending", sessionID, reconcile(merged))
setStore("session", "pending", sessionID, reconcile(pending))
setStore(
"session",
"input",
sessionID,
reconcile(merged.filter((item) => item.type !== "compaction").map((item) => item.id)),
reconcile(pending.filter((item) => item.type !== "compaction").map((item) => item.id)),
)
merged.forEach(materializeInboxMessage)
pending.forEach((item) => {
materializeInboxMessage(item)
confirmAdmission(item)
})
// A compaction may coalesce onto another ID at the server.
if (pending.some((item) => item.type === "compaction"))
optimistic
.list(sessionID)
.filter((entry) => entry.item.type === "compaction")
.forEach((entry) => optimistic.remove(entry.id))
})
})
},
@@ -1426,9 +1575,8 @@ export function createData(config: CreateDataInput) {
// A known pending control ID may be consumed while setup waits. Propose
// a fresh ID and let the server coalesce, without duplicating its row.
const id = SessionMessage.ID.create()
if (!store.session.pending[input.sessionID]?.some((item) => item.type === "compaction")) {
outbox.add(id)
admitLocal({
if (!result.session.pending.list(input.sessionID).some((item) => item.type === "compaction")) {
stageLocal({
id,
sessionID: input.sessionID,
timeCreated: Date.now(),
@@ -1441,20 +1589,24 @@ export function createData(config: CreateDataInput) {
// speculative row on an echo, and remember consumed IDs until the POST
// settles so its older response cannot resurrect a queued row.
const observed = new Set<string>()
const request = sendAdmission(input.sessionID, async () => {
if (input.model) await api().session.switchModel({ sessionID: input.sessionID, model: input.model })
return api().session.compact({ sessionID: input.sessionID, id })
})
const request = admissions
.run(
input.sessionID,
async () => {
if (input.model) await api().session.switchModel({ sessionID: input.sessionID, model: input.model })
return api().session.compact({ sessionID: input.sessionID, id })
},
{ wait: creating.get(input.sessionID) },
)
.then((item) => {
batch(() => {
outbox.delete(id)
if (item.id !== id) removePending(input.sessionID, id)
if (!observed.has(item.id) && !messageIndex.get(input.sessionID)?.has(item.id)) admitLocal(item)
optimistic.remove(id)
if (!observed.has(item.id) && !messageIndex.get(input.sessionID)?.has(item.id)) admitCanonical(item)
})
return item
})
.catch((error) => {
if (outbox.delete(id)) removePending(input.sessionID, id)
optimistic.remove(id)
throw error
})
.finally(() => {
@@ -1463,51 +1615,53 @@ export function createData(config: CreateDataInput) {
compacting.set(input.sessionID, { id, observed, request })
return request
},
// 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 & { gate?: Promise<unknown>; prepare?: () => Promise<unknown> }) {
const { gate, prepare, ...request } = input
submission: {
get(sessionID: string, id: string) {
const value = submissions[id]
return value?.sessionID === sessionID ? value : undefined
},
retry(sessionID: string, id: string) {
const operation = prompts.get(id)
if (!operation || operation.group !== sessionID)
return Promise.reject(new Error("Prompt submission not found"))
return promptResult(sessionID, id, operation.retry())
},
},
// Capture once. The command owns attempts; the overlay owns presentation.
prompt(
input: SessionPromptInput & { gate?: Promise<unknown>; prepare?: () => Promise<unknown>; model?: ModelRef },
) {
const { gate, prepare, model, ...request } = input
const id = request.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 existing = prompts.get(id)
if (existing) {
if (existing.group !== request.sessionID)
return Promise.reject(new Error("Prompt submission belongs to another session"))
return promptResult(request.sessionID, id, existing.retry())
}
const fresh =
!messageIndex.get(request.sessionID)?.has(id) &&
!store.session.pending[request.sessionID]?.some((item) => item.id === id)
const operation = prompts.submit(
{ request: unwrap({ ...request, id }), model: model && { ...model } },
{ wait: Promise.all([gate, creating.get(request.sessionID)]), prepare },
)
if (fresh) {
outbox.add(id)
admitLocal({
stageLocal({
id,
sessionID: request.sessionID,
timeCreated: Date.now(),
timeCreated: operation.createdAt,
type: "user",
delivery: request.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.
delivery: operation.input.request.delivery ?? "steer",
// Files and skills are resolved by the server, not guessed locally.
payload: {
text: request.text,
agents: request.agents?.map((agent) => ({ ...agent })),
metadata: request.metadata,
text: operation.input.request.text,
agents: operation.input.request.agents?.map((agent) => ({ ...agent })),
metadata: operation.input.request.metadata,
},
})
}
return sendAdmission(
request.sessionID,
async () => {
await prepare?.()
return api().session.prompt({ ...request, id })
},
gate,
).catch((error) => {
// Roll back only rows this call admitted and the server has not
// acknowledged: anything else is server state.
if (fresh && outbox.delete(id)) retractLocal(request.sessionID, id)
throw error
})
return promptResult(request.sessionID, id, operation.request)
},
sync(sessionID: string, options?: { children?: boolean }) {
return sync.run(options?.children ? `session.family:${sessionID}` : `session:${sessionID}`, async () => {
@@ -1540,32 +1694,41 @@ export function createData(config: CreateDataInput) {
},
message: {
list(sessionID: string) {
return store.session.message[sessionID] ?? []
return Optimistic.merge(
store.session.message[sessionID] ?? [],
(local[sessionID] ?? []).flatMap((entry) => (entry.message ? [entry.message] : [])),
(item) => item.id,
)
},
get(sessionID: string, messageID: string) {
const messages = store.session.message[sessionID]
const position = messageIndex.get(sessionID)?.get(messageID)
return position === undefined ? undefined : messages?.[position]
return position === undefined
? local[sessionID]?.find((entry) => entry.id === messageID)?.message
: messages?.[position]
},
sync(sessionID: string) {
return sync.run(`session.message:${sessionID}`, async () => {
const response = await api().message.list({ sessionID, limit: messagePageLimit, order: "desc" })
const fetched = response.data.toReversed()
// Same protection as the pending sync: a re-fetch racing an
// admission must not wipe its local transcript row.
const ids = new Set(fetched.map((item) => item.id))
const admitted = new Set(
(store.session.pending[sessionID] ?? []).flatMap((item) =>
item.type === "user" || item.type === "synthetic" ? [item.id] : [],
),
)
const local = (store.session.message[sessionID] ?? []).filter(
(item) => !ids.has(item.id) && (outbox.has(item.id) || admitted.has(item.id)),
// An admitted inbox item belongs in the transcript before delivery.
// Local guesses are composed by selectors and never enter this cache.
const pending = (store.session.message[sessionID] ?? []).filter(
(item) => !ids.has(item.id) && admitted.has(item.id),
)
const messages = local.length === 0 ? fetched : [...fetched, ...local]
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)
const messages = pending.length === 0 ? fetched : [...fetched, ...pending]
batch(() => {
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)
observeMessages(sessionID, fetched)
})
})
},
more(sessionID: string) {
@@ -1629,6 +1792,7 @@ export function createData(config: CreateDataInput) {
messageIndex.set(sessionID, new Map(messages.map((item, position) => [item.id, position])))
setStore("session", "message", sessionID, reconcile(messages))
setStore("session", "messageCursor", sessionID, next)
observeMessages(sessionID, fetched)
})
return true
})()
+26
View File
@@ -0,0 +1,26 @@
import { ClientError, isServiceUnavailableError } from "../promise"
// Only classifications reach submission diagnostics, never raw causes or payloads.
export function promptFailure(error: unknown): { retryable: boolean; reason: string; status?: number } {
if (error instanceof Error && !(error instanceof ClientError) && error.cause) return promptFailure(error.cause)
if (isServiceUnavailableError(error)) return { retryable: true, reason: "ServiceUnavailable", status: 503 }
if (!(error instanceof ClientError)) return { retryable: false, reason: "Rejected" }
if (error.reason === "Transport") {
const aborted = error.cause instanceof Error && error.cause.name === "AbortError"
return { retryable: !aborted, reason: aborted ? "Aborted" : "Transport" }
}
if (error.reason === "MalformedResponse") return { retryable: true, reason: error.reason }
const status =
error.reason === "UnexpectedStatus" &&
typeof error.cause === "object" &&
error.cause !== null &&
"status" in error.cause &&
typeof error.cause.status === "number"
? error.cause.status
: undefined
return {
retryable: status !== undefined && [408, 429, 500, 502, 503, 504].includes(status),
reason: error.reason,
status,
}
}
+523
View File
@@ -0,0 +1,523 @@
import { describe, expect, test } from "bun:test"
import { Command } from "../src/command"
describe("Command.queue", () => {
test("captures wait at enqueue, serializes each group, and releases successors after failure", async () => {
const queue = Command.queue()
const gate = Promise.withResolvers<void>()
const calls: string[] = []
const failure = new Error("creation failed")
const first = queue.run("one", async () => calls.push("first"), { wait: gate.promise })
const second = queue.run("one", async () => calls.push("second"))
const other = queue.run("two", async () => calls.push("other"))
expect(queue.pending("one")).toBe(true)
await other
expect(calls).toEqual(["other"])
gate.reject(failure)
await expect(first).rejects.toBe(failure)
await second
expect(calls).toEqual(["other", "second"])
await wait(() => !queue.pending("one"))
})
test("an aborted gated item settles promptly without bypassing its predecessor", async () => {
const queue = Command.queue()
const predecessor = Promise.withResolvers<void>()
const gate = Promise.withResolvers<void>()
const abort = new AbortController()
const calls: string[] = []
const first = queue.run("one", async () => {
calls.push("first")
await predecessor.promise
})
const middle = queue.run("one", async () => calls.push("middle"), { wait: gate.promise, signal: abort.signal })
const last = queue.run("one", async () => calls.push("last"))
await wait(() => calls.length === 1)
abort.abort(new Error("cancelled gate"))
await expect(middle).rejects.toBe(abort.signal.reason)
await Bun.sleep(1)
expect(calls).toEqual(["first"])
predecessor.resolve()
await Promise.all([first, last])
expect(calls).toEqual(["first", "last"])
gate.reject(new Error("late gate failure"))
})
test("does not invoke an already aborted item", async () => {
const queue = Command.queue()
const abort = new AbortController()
abort.abort()
let called = false
await expect(queue.run("one", async () => (called = true), { signal: abort.signal })).rejects.toBe(
abort.signal.reason,
)
expect(called).toBe(false)
})
})
type Input = { id: string; session: string; payload: { text: string }; model?: { id: string } }
function input(id = "message", session = "session"): Input {
return { id, session, payload: { text: "original" }, model: { id: "demo" } }
}
function setup(
execute: (input: Input) => (context: Command.Context) => Promise<string>,
options?: { queue?: Command.Queue; retry?: { delays: readonly number[]; when: (error: unknown) => boolean } },
) {
return Command.make({
key: (input: Input) => input.id,
group: (input: Input) => input.session,
queue: options?.queue ?? Command.queue(),
execute,
retry: options?.retry,
})
}
describe("Command.make", () => {
test("captures one payload and key; duplicate submissions preserve the first operation", async () => {
const gate = Promise.withResolvers<void>()
const seen: Input[] = []
const manager = setup((value) => async () => {
seen.push(value)
return value.payload.text
})
const value = input()
const operation = manager.submit(value, { wait: gate.promise })
value.payload.text = "changed"
value.model!.id = "changed"
expect(operation.input).toEqual(input())
expect(operation.input).not.toBe(value)
expect(operation.started).toBe(false)
expect(operation.createdAt).toBeGreaterThan(0)
expect(manager.submit(value)).toBe(operation)
expect(() => manager.submit(input(value.id, "other"))).toThrow("belongs to another group")
expect(manager.get(value.id)).toBe(operation)
expect(manager.values()).toEqual([operation])
gate.resolve()
await expect(operation.request).resolves.toBe("original")
expect(seen).toEqual([input()])
expect(seen[0]).toBe(operation.input)
expect(operation.started).toBe(true)
manager.dispose()
})
test("shares FIFO between managers while other sessions execute independently", async () => {
const queue = Command.queue()
const first = Promise.withResolvers<string>()
const calls: string[] = []
const prompts = setup(
(value) => async () => {
calls.push(value.id)
return first.promise
},
{ queue },
)
const controls = setup(
(value) => async () => {
calls.push(value.id)
return value.id
},
{ queue },
)
const prompt = prompts.submit(input("prompt"))
const compact = controls.submit(input("compact"))
const other = controls.submit(input("other", "other"))
await other.request
expect(calls).toEqual(["prompt", "other"])
first.resolve("prompt")
await Promise.all([prompt.request, compact.request])
expect(calls).toEqual(["prompt", "other", "compact"])
prompts.dispose()
controls.dispose()
})
test("retries captured data, prepares once, and rebuilds per-cycle factory state only on manual retry", async () => {
const seen: Input[] = []
const attempts: number[] = []
const contexts: Command.Context[] = []
let prepared = 0
let factories = 0
let selections = 0
let succeed = false
const failure = new Error("offline")
const manager = setup(
(value) => {
factories++
let selected = false
return async (context) => {
contexts.push(context)
seen.push(value)
attempts.push(context.attempt)
if (!selected) {
selections++
selected = true
context.stage("model")
}
context.stage("prompt")
if (!succeed) throw failure
return "accepted"
}
},
{ retry: { delays: [1, 1], when: (error) => error === failure } },
)
const operation = manager.submit(input(), {
prepare: async () => {
prepared++
},
})
expect(operation.retry()).toBe(operation.request)
const original = operation.request
await expect(original).rejects.toMatchObject({ reason: "failed", key: "message", cause: failure })
expect(operation.state).toMatchObject({ status: "failed", attempt: 3, stage: "prompt" })
expect(manager.get(operation.key)).toBe(operation)
expect(prepared).toBe(1)
expect(factories).toBe(1)
expect(selections).toBe(1)
succeed = true
const retried = operation.retry()
contexts[0].stage("stale cycle")
expect(operation.state).toMatchObject({ status: "sending", attempt: 0, stage: "waiting" })
expect(retried).not.toBe(original)
expect(operation.request).toBe(retried)
expect(operation.retry()).toBe(retried)
await expect(retried).resolves.toBe("accepted")
expect(attempts).toEqual([1, 2, 3, 1])
expect(seen.every((value) => value === operation.input)).toBe(true)
expect(prepared).toBe(1)
expect(factories).toBe(2)
expect(selections).toBe(2)
manager.dispose()
})
test("preparation and gate failures are definitive and never retried", async () => {
for (const stage of ["prepare", "waiting"]) {
const failure = new Error(stage)
let factories = 0
let classified = 0
const manager = setup(
() => {
factories++
return async () => "unexpected"
},
{
retry: {
delays: [1],
when: () => {
classified++
return true
},
},
},
)
const operation = manager.submit(input(), {
wait: stage === "waiting" ? Promise.reject(failure) : undefined,
prepare: async () => {
throw failure
},
})
await expect(operation.request).rejects.toBe(failure)
expect(operation.state).toMatchObject({ status: "rejected", stage, attempt: 0, error: failure })
expect(operation.started).toBe(false)
expect(factories).toBe(0)
expect(classified).toBe(0)
expect(manager.values()).toEqual([])
}
})
test("definitive execution rejection preserves the original error and removes the operation", async () => {
const failure = new Error("invalid")
const manager = setup(() => async () => {
throw failure
})
const operation = manager.submit(input())
await expect(operation.request).rejects.toBe(failure)
expect(operation.state).toMatchObject({ status: "rejected", attempt: 1, error: failure })
expect(manager.values()).toEqual([])
operation.confirm("too late")
expect(operation.accepted).toBeUndefined()
await expect(operation.retry()).rejects.toBe(failure)
})
test("HTTP acceptance is retained until canonical confirmation, with deterministic removal events", async () => {
const manager = setup(() => async () => "http")
const events: { status: string; removed: boolean; retained: boolean }[] = []
manager.subscribe(({ operation, removed }) => {
events.push({ status: operation.state.status, removed, retained: !!manager.get(operation.key) })
})
const operation = manager.submit(input())
await expect(operation.request).resolves.toBe("http")
expect(operation.accepted).toBe("http")
expect(operation.signal.aborted).toBe(false)
expect(manager.values()).toEqual([operation])
expect(events.at(-1)).toEqual({ status: "accepted", removed: false, retained: true })
manager.confirm(operation.key, "canonical")
expect(operation.accepted).toBe("canonical")
expect(operation.signal.aborted).toBe(true)
expect(manager.values()).toEqual([])
expect(events.slice(-2)).toEqual([
{ status: "accepted", removed: false, retained: false },
{ status: "accepted", removed: true, retained: false },
])
operation.cancel()
await expect(operation.retry()).resolves.toBe("canonical")
})
test("confirmation settles behind a gate without executing or releasing an earlier owner", async () => {
const queue = Command.queue()
const predecessor = Promise.withResolvers<void>()
const owner = queue.run("session", () => predecessor.promise)
let executed = 0
const manager = setup(
() => async () => {
executed++
return "http"
},
{ queue },
)
const operation = manager.submit(input(), { wait: new Promise(() => {}) })
let followed = false
const follower = queue.run("session", async () => {
followed = true
})
manager.confirm(operation.key, "canonical")
await expect(operation.request).resolves.toBe("canonical")
await Bun.sleep(1)
expect(followed).toBe(false)
expect(executed).toBe(0)
expect(operation.started).toBe(false)
predecessor.resolve()
await Promise.all([owner, follower])
expect(followed).toBe(true)
})
test("cancelling preparation settles immediately and ignores its eventual result", async () => {
const preparation = Promise.withResolvers<void>()
const preparing = Promise.withResolvers<void>()
let factories = 0
const manager = setup(() => {
factories++
return async () => "http"
})
const operation = manager.submit(input(), {
prepare: async () => {
preparing.resolve()
return preparation.promise
},
})
await preparing.promise
manager.cancel(operation.key)
await expect(operation.request).rejects.toMatchObject({ reason: "cancelled" })
preparation.reject(new Error("late prepare"))
await Bun.sleep(1)
expect(factories).toBe(0)
expect(operation.started).toBe(false)
expect(manager.values()).toEqual([])
})
test("confirmation at the prepare boundary prevents preparation and execution from starting", async () => {
let prepared = false
const manager = setup(() => async () => "unexpected")
manager.subscribe(({ operation }) => {
if (operation.state.status === "sending" && operation.state.stage === "prepare") operation.confirm("canonical")
})
const operation = manager.submit(input(), {
prepare: async () => {
prepared = true
},
})
await expect(operation.request).resolves.toBe("canonical")
await Bun.sleep(1)
expect(prepared).toBe(false)
expect(operation.started).toBe(false)
expect(operation.state.status).toBe("accepted")
})
test("supports undefined canonical output without confusing it with an unconfirmed operation", async () => {
const manager = Command.make<string, void>({
key: (input) => input,
group: () => "session",
queue: Command.queue(),
execute: () => async () => {
throw new Error("must not execute")
},
})
const operation = manager.submit("void", { wait: new Promise(() => {}) })
operation.confirm(undefined)
await expect(operation.request).resolves.toBeUndefined()
await expect(operation.retry()).resolves.toBeUndefined()
expect(operation.state.status).toBe("accepted")
expect(manager.values()).toEqual([])
})
test("confirmation cancels backoff promptly and can recover an exhausted operation", async () => {
for (const phase of ["retrying", "failed"]) {
const failure = new Error("offline")
let calls = 0
const manager = setup(
() => async () => {
calls++
throw failure
},
{
retry: { delays: phase === "retrying" ? [60_000] : [], when: () => true },
},
)
const reached = Promise.withResolvers<void>()
manager.subscribe(({ operation }) => {
if (operation.state.status === phase) reached.resolve()
})
const operation = manager.submit(input())
const first = operation.request
await reached.promise
manager.confirm(operation.key, "canonical")
await expect(operation.request).resolves.toBe("canonical")
if (phase === "failed") await expect(first).rejects.toBeInstanceOf(Command.Error)
else await expect(first).resolves.toBe("canonical")
expect(operation.state.status).toBe("accepted")
expect(calls).toBe(1)
expect(manager.values()).toEqual([])
}
})
test.each(["resolve", "reject"])(
"late HTTP %s and stage reports cannot resurrect confirmed or cancelled handles",
async (outcome) => {
for (const action of ["confirm", "cancel"]) {
const response = Promise.withResolvers<string>()
const invoked = Promise.withResolvers<Command.Context>()
const manager = setup(() => async (context) => {
invoked.resolve(context)
return response.promise
})
const events: string[] = []
manager.subscribe(({ operation, removed }) => events.push(`${operation.state.status}:${removed}`))
const operation = manager.submit(input())
const context = await invoked.promise
if (action === "confirm") operation.confirm("canonical")
else operation.cancel()
if (action === "confirm") await expect(operation.request).resolves.toBe("canonical")
else await expect(operation.request).rejects.toMatchObject({ reason: "cancelled", key: "message" })
const count = events.length
context.stage("late")
if (outcome === "resolve") response.resolve("late")
else response.reject(new Error("late"))
await Bun.sleep(1)
expect(events).toHaveLength(count)
expect(manager.values()).toEqual([])
expect(operation.state.status).toBe(action === "confirm" ? "accepted" : "cancelled")
expect(operation.accepted).toBe(action === "confirm" ? "canonical" : undefined)
expect(context.signal.aborted).toBe(true)
}
},
)
test("confirmFrom uses canonical output, deduplicates in-flight lookups, and wins over HTTP failure", async () => {
const response = Promise.withResolvers<string>()
const canonical = Promise.withResolvers<string>()
const invoked = Promise.withResolvers<void>()
let loads = 0
const manager = setup(() => async () => {
invoked.resolve()
return response.promise
})
const operation = manager.submit(input())
await invoked.promise
const load = async () => {
loads++
return canonical.promise
}
manager.confirmFrom(operation.key, load)
operation.confirmFrom(load)
response.reject(new Error("lost HTTP response"))
await wait(() => loads === 1)
expect(operation.accepted).toBeUndefined()
canonical.resolve("canonical, not optimistic")
await expect(operation.request).resolves.toBe("canonical, not optimistic")
expect(loads).toBe(1)
expect(manager.values()).toEqual([])
})
test("failed confirmation lookup retains proof and permits lookup retry after bounded transport exhaustion", async () => {
const response = Promise.withResolvers<string>()
const invoked = Promise.withResolvers<void>()
const lookup = Promise.withResolvers<string>()
let calls = 0
const manager = setup(
() => async () => {
calls++
invoked.resolve()
return response.promise
},
{
retry: { delays: [1], when: () => false },
},
)
const operation = manager.submit(input())
await invoked.promise
operation.confirmFrom(() => lookup.promise)
lookup.reject(new Error("lookup unavailable"))
response.reject(new Error("not normally retryable"))
await expect(operation.request).rejects.toMatchObject({ reason: "failed" })
expect(calls).toBe(2)
expect(operation.state.status).toBe("failed")
expect(manager.get(operation.key)).toBe(operation)
operation.confirmFrom(async () => "recovered canonical")
await wait(() => operation.state.status === "accepted")
expect(operation.accepted).toBe("recovered canonical")
expect(manager.values()).toEqual([])
})
test("unsubscribe leaves work alive; dispose cancels requests, gates, lookups, and retained operations", async () => {
const invoked = Promise.withResolvers<void>()
const lookupStarted = Promise.withResolvers<AbortSignal>()
const late = Promise.withResolvers<string>()
const manager = setup((value) => async () => {
if (value.id === "accepted") return "http"
invoked.resolve()
return new Promise(() => {})
})
let notifications = 0
const unsubscribe = manager.subscribe(() => notifications++)
const accepted = manager.submit(input("accepted", "accepted"))
const running = manager.submit(input("running", "running"))
const gated = manager.submit(input("gated"), { wait: new Promise(() => {}) })
await Promise.all([accepted.request, invoked.promise])
unsubscribe()
const count = notifications
running.confirmFrom(async (signal) => {
lookupStarted.resolve(signal)
return late.promise
})
const signal = await lookupStarted.promise
expect(running.signal.aborted).toBe(false)
manager.dispose()
await expect(running.request).rejects.toMatchObject({ reason: "cancelled" })
await expect(gated.request).rejects.toMatchObject({ reason: "cancelled" })
await expect(accepted.request).rejects.toMatchObject({ reason: "cancelled" })
expect(signal.aborted).toBe(true)
expect(notifications).toBe(count)
expect(manager.values()).toEqual([])
expect(() => manager.submit(input("new"))).toThrow("disposed")
late.resolve("too late")
await Bun.sleep(1)
expect(running.state.status).toBe("cancelled")
})
test("unused request rejections are internally observed without changing the returned rejection", async () => {
const manager = setup(() => async () => {
throw new Error("unused")
})
const operation = manager.submit(input())
await wait(() => operation.state.status === "rejected")
await expect(operation.request).rejects.toThrow("unused")
})
})
async function wait(predicate: () => boolean) {
for (let attempt = 0; attempt < 1000; attempt++) {
if (predicate()) return
await Bun.sleep(1)
}
throw new Error("Timed out waiting for command state")
}
+56
View File
@@ -0,0 +1,56 @@
import { expect, test } from "bun:test"
import { Optimistic } from "../src/optimistic"
test("local contributions preserve order, publish by group, and retire permanently", () => {
const overlay = Optimistic.make<{ id: string; group: string; text: string }>({
key: (value) => value.id,
group: (value) => value.group,
})
const changed: string[] = []
const unsubscribe = overlay.subscribe((group) => changed.push(group))
overlay.set({ id: "a", group: "one", text: "First" })
overlay.set({ id: "b", group: "one", text: "Second" })
overlay.set({ id: "c", group: "two", text: "Independent" })
overlay.set({ id: "a", group: "one", text: "Canonical payload" })
expect(overlay.list("one").map((item) => item.id)).toEqual(["a", "b"])
expect(overlay.get("a")?.text).toBe("Canonical payload")
expect(changed).toEqual(["one", "one", "two", "one"])
expect(overlay.remove("a")).toBe(true)
expect(overlay.remove("a")).toBe(false)
expect(overlay.has("a")).toBe(false)
overlay.clear("one")
expect(overlay.list("one")).toEqual([])
expect(overlay.list("two")).toHaveLength(1)
unsubscribe()
overlay.clear("two")
expect(changed.at(-1)).toBe("one")
})
test("authoritative rows win without cloning or changing their order", () => {
const canonical = [
{ id: "b", text: "Server" },
{ id: "a", text: "Earlier" },
]
const local = [
{ id: "b", text: "Guess" },
{ id: "c", text: "Pending" },
]
const view = Optimistic.merge(canonical, local, (row) => row.id)
expect(view.map((row) => row.id)).toEqual(["b", "a", "c"])
expect(view[0]).toBe(canonical[0])
expect(view[1]).toBe(canonical[1])
expect(view[2]).toBe(local[1])
expect(Optimistic.merge(canonical, [], (row) => row.id)).toBe(canonical)
expect(Optimistic.merge(canonical, [local[0]], (row) => row.id)).toBe(canonical)
})
test("overlays cannot move an identity between owners", () => {
const overlay = Optimistic.make<{ id: string; group: string }>({
key: (item) => item.id,
group: (item) => item.group,
})
overlay.set({ id: "a", group: "one" })
expect(() => overlay.set({ id: "a", group: "two" })).toThrow("another group")
expect(overlay.list("two")).toEqual([])
expect(overlay.get("a")?.group).toBe("one")
})
@@ -190,7 +190,8 @@ test.each(["compaction", "canonical compaction", "user"])(
await fixture.data.session.pending.sync(sessionID)
expect(fixture.data.session.pending.list(sessionID)).toEqual([durable])
fixture.response.resolve(new Response("Lost response", { status: 500 }))
expect(await result).toBeInstanceOf(Error)
if (type === "user") expect(await result).toEqual(durable)
if (type !== "user") expect(await result).toBeInstanceOf(Error)
expect(fixture.data.session.pending.list(sessionID)).toEqual([durable])
if (type === "user")
expect(fixture.data.session.message.list(sessionID)).toMatchObject([{ id, type: "user", text: "Follow up" }])
@@ -200,14 +201,17 @@ test.each(["compaction", "canonical compaction", "user"])(
test("keeps one event listener and removes it when the data owner is disposed during a gate", async () => {
using fixture = setup()
const gate = Promise.withResolvers<void>()
const first = fixture.data.session.prompt({ sessionID, text: "First", gate: gate.promise })
const first = fixture.data.session
.prompt({ sessionID, text: "First", gate: gate.promise })
.catch((error: unknown) => error)
const compact = fixture.data.session.compact({ sessionID })
expect(fixture.listeners.size).toBe(1)
fixture.dispose()
expect(fixture.listeners.size).toBe(0)
gate.resolve()
fixture.response.resolve(Response.json({ data: item("msg_canonical") }))
await Promise.all([first, compact])
expect(await first).toMatchObject({ reason: "cancelled" })
await compact
expect(fixture.listeners.size).toBe(0)
})
@@ -0,0 +1,797 @@
import { expect, test } from "bun:test"
import { createComputed, createRoot } from "solid-js"
import { isServer } from "solid-js/web"
import { createData, type CreateDataInput } from "../src/solid"
import { OpenCode, type OpenCodeEvent, type SessionInboxUser, type SessionMessageInfo } from "../src/promise"
test("retries transient admission with the same ID and captured body", async () => {
const bodies: string[] = []
using fixture = setup(async (request) => {
bodies.push(await request.text())
if (bodies.length === 1) return new Response(null, { status: 503 })
return Response.json({ data: item(JSON.parse(bodies[0]).id) })
})
const input = { sessionID, text: "Original", files: [{ uri: "file:///original.txt" }] }
const sent = fixture.data.session.prompt(input).catch((error: unknown) => error)
input.text = "Changed"
input.files[0].uri = "file:///changed.txt"
await wait(() => bodies.length === 2)
expect(await sent).toMatchObject({ type: "user" })
expect(bodies[1]).toBe(bodies[0])
expect(JSON.parse(bodies[0])).toMatchObject({ text: "Original", files: [{ uri: "file:///original.txt" }] })
expect(fixture.data.session.message.list(sessionID)).toHaveLength(1)
})
test("retries model selection but runs arbitrary preparation once and preserves admission ordering", async () => {
const calls: string[] = []
let modelCalls = 0
let promptCalls = 0
using fixture = setup(async (request) => {
const rpc = request.url.split("/").at(-1)
const body = await request.json()
calls.push(rpc === "prompt" ? body.text : rpc)
if (rpc === "model") {
modelCalls += 1
return new Response(null, { status: modelCalls === 1 ? 503 : 204 })
}
if (body.text === "First" && ++promptCalls === 1) throw new Error("Connection reset")
return Response.json({ data: item(body.id) })
})
let preparations = 0
const first = fixture.data.session.prompt({
sessionID,
text: "First",
model: { providerID: "demo", id: "model" },
prepare: async () => {
preparations += 1
},
})
const second = fixture.data.session.prompt({ sessionID, text: "Second" })
await Promise.all([first, second])
expect(preparations).toBe(1)
expect(calls).toEqual(["model", "model", "First", "First", "Second"])
})
test.each([400, 401, 404, 409, 422])("does not retry definitive HTTP %i rejection", async (status) => {
let calls = 0
using fixture = setup(async () => {
calls += 1
return Response.json({ message: "Invalid request" }, { status })
})
await expect(fixture.data.session.prompt({ sessionID, text: "Invalid" })).rejects.toBeDefined()
expect(calls).toBe(1)
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
expect(fixture.data.session.message.list(sessionID)).toEqual([])
})
test("an internal server error is ambiguous and retries the original admission", async () => {
let calls = 0
using fixture = setup(async (request) => {
calls += 1
if (calls === 1) return new Response(null, { status: 500 })
return Response.json({ data: item((await request.json()).id) })
})
expect(await fixture.data.session.prompt({ sessionID, id: "msg_500", text: "Retry" })).toMatchObject({
id: "msg_500",
})
expect(calls).toBe(2)
})
test("cancelling an existing admission reaches the server even while its local retry is gated", async () => {
const methods: string[] = []
using fixture = setup(async (request) => {
methods.push(request.method)
return new Response(null, { status: 204 })
})
fixture.enqueue("msg_existing")
const gate = Promise.withResolvers<void>()
const sent = fixture.data.session
.prompt({ sessionID, id: "msg_existing", text: "Retry", gate: gate.promise })
.catch((error: unknown) => error)
await fixture.data.session.pending.cancel(sessionID, "msg_existing")
expect(await sent).toMatchObject({ reason: "cancelled" })
expect(methods).toEqual(["DELETE"])
gate.resolve()
})
test("keeps an exhausted submission visible and manual retry reuses its immutable capture", async () => {
const bodies: string[] = []
const models: string[] = []
let preparations = 0
using fixture = setup(async (request) => {
if (request.url.endsWith("/model")) {
models.push((await request.json()).model.id)
return new Response(null, { status: 204 })
}
bodies.push(await request.text())
if (bodies.length <= 4) return new Response(null, { status: 502 })
return Response.json({ data: item(JSON.parse(bodies[0]).id) })
})
const input = {
sessionID,
text: "Keep me",
files: [{ uri: "file:///original.txt" }],
metadata: { source: "original" },
model: { providerID: "demo", id: "original" },
prepare: async () => {
preparations += 1
},
}
const first = fixture.data.session.prompt(input)
input.text = "Changed"
input.files[0].uri = "file:///changed.txt"
input.metadata.source = "changed"
input.model.id = "changed"
await expect(first).rejects.toMatchObject({ reason: "failed" })
const id = JSON.parse(bodies[0]).id
expect(bodies).toHaveLength(4)
expect(fixture.data.session.submission.get(sessionID, id)).toMatchObject({ status: "failed", attempt: 4 })
expect(fixture.data.session.message.list(sessionID)).toMatchObject([{ id, text: "Keep me" }])
await fixture.api.session.switchModel({ sessionID, model: { providerID: "demo", id: "different" } })
await fixture.data.session.submission.retry(sessionID, id)
expect(bodies).toHaveLength(5)
expect(new Set(bodies).size).toBe(1)
expect(JSON.parse(bodies[0])).toMatchObject({
text: "Keep me",
files: [{ uri: "file:///original.txt" }],
metadata: { source: "original" },
})
expect(models).toEqual(["original", "different", "original"])
expect(preparations).toBe(1)
expect(fixture.data.session.submission.get(sessionID, id)).toBeUndefined()
})
test("an enqueue acknowledgement settles a lost response without another POST", async () => {
const requested = Promise.withResolvers<string>()
let calls = 0
using fixture = setup(async (request) => {
calls += 1
requested.resolve((await request.json()).id)
return new Promise((_, reject) =>
request.signal.addEventListener("abort", () => reject(new Error("Closed")), { once: true }),
)
})
const sent = fixture.data.session.prompt({ sessionID, text: "Accepted" })
const id = await requested.promise
fixture.enqueue(id)
expect(await sent).toEqual(item(id))
expect(calls).toBe(1)
expect(fixture.data.session.submission.get(sessionID, id)).toBeUndefined()
})
test("accepted HTTP responses retain a preview and same-ID calls join until canonical observation", async () => {
let calls = 0
using fixture = setup(async (request) => {
calls += 1
return Response.json({ data: item((await request.json()).id) })
})
const input = { sessionID, id: "msg_accepted", text: "Original" }
expect(await fixture.data.session.prompt(input)).toEqual(item(input.id))
const preview = fixture.data.session.message.list(sessionID)[0]
expect(preview).toMatchObject({ id: input.id, text: "Accepted" })
expect(fixture.data.session.message.get(sessionID, input.id)).toBe(preview)
expect(fixture.data.session.submission.get(sessionID, input.id)).toBeUndefined()
expect(await fixture.data.session.prompt({ ...input, text: "Must not replace", delivery: "queue" })).toEqual(
item(input.id),
)
expect(await fixture.data.session.submission.retry(sessionID, input.id)).toEqual(item(input.id))
expect(calls).toBe(1)
expect(fixture.data.session.message.list(sessionID)[0]).toBe(preview)
fixture.enqueue(input.id)
expect(fixture.data.session.message.list(sessionID)).toHaveLength(1)
fixture.data.session.evict(sessionID)
expect(fixture.data.session.message.list(sessionID)).toEqual([])
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
})
test("another session can send while one session is retrying", async () => {
const calls: string[] = []
using fixture = setup(async (request) => {
const body = await request.json()
calls.push(body.text)
if (body.text === "Blocked") return new Response(null, { status: 503 })
return Response.json({ data: { ...item(body.id), sessionID: "ses_other" } })
})
const first = fixture.data.session
.prompt({ sessionID, id: "msg_blocked", text: "Blocked" })
.catch((error: unknown) => error)
await wait(() => fixture.data.session.submission.get(sessionID, "msg_blocked")?.status === "retrying")
await fixture.data.session.prompt({ sessionID: "ses_other", text: "Independent" })
expect(calls).toEqual(["Blocked", "Independent"])
fixture.dispose()
expect(await first).toMatchObject({ reason: "cancelled" })
})
test.each(["pending-read", "enqueue", "delivered", "cancelled"])(
"a foreign-session %s observation cannot settle or remove an explicit-ID submission",
async (observation) => {
const calls: string[] = []
using fixture = setup(async (request) => {
if (request.method === "GET") {
if (request.url.endsWith("/inbox")) return Response.json({ data: [item("msg_existing")] })
return Response.json({
data: [{ id: "msg_history", type: "user", text: "Owner history", time: { created: 1 } }],
cursor: {},
})
}
calls.push(new URL(request.url).pathname)
expect((await request.json()).id).toBe("msg_existing")
return Response.json({ _tag: "ConflictError", message: "Message belongs to another session" }, { status: 409 })
})
await fixture.data.session.message.sync(sessionID)
fixture.enqueue("msg_existing")
const history = fixture.data.session.message.get(sessionID, "msg_history")
const gate = Promise.withResolvers<void>()
let settled = false
const sent = fixture.data.session
.prompt({ sessionID: "ses_other", id: "msg_existing", text: "Foreign local", gate: gate.promise })
.catch((error: unknown) => error)
.finally(() => {
settled = true
})
const preview = fixture.data.session.message.get("ses_other", "msg_existing")
if (observation === "pending-read") await fixture.data.session.pending.sync(sessionID)
if (observation === "enqueue") fixture.enqueue("msg_existing")
if (observation === "delivered") {
const event = {
id: "evt_delivered",
created: 20,
type: "session.inbox.delivered",
durable: { aggregateID: sessionID, seq: 2, version: 1 },
data: { sessionID, inboxID: "msg_existing", messageID: "msg_existing" },
} satisfies OpenCodeEvent
fixture.emit(event)
// Also exercise delivery without a remaining pending row.
fixture.emit(event)
}
if (observation === "cancelled")
fixture.emit({
id: "evt_cancelled",
created: 20,
type: "session.inbox.cancelled",
durable: { aggregateID: sessionID, seq: 2, version: 1 },
data: { sessionID, inboxID: "msg_existing" },
})
await Bun.sleep(0)
expect(settled).toBe(false)
expect(calls).toEqual([])
expect(fixture.data.session.submission.get("ses_other", "msg_existing")).toMatchObject({
status: "sending",
attempt: 0,
})
expect(fixture.data.session.message.get("ses_other", "msg_existing")).toBe(preview)
expect(preview).toMatchObject({ text: "Foreign local" })
expect(fixture.data.session.pending.list("ses_other")).toMatchObject([{ id: "msg_existing" }])
const canonical = JSON.stringify({
messages: fixture.data.session.message.list(sessionID),
pending: fixture.data.session.pending.list(sessionID),
})
gate.resolve()
expect(await sent).toEqual({ _tag: "ConflictError", message: "Message belongs to another session" })
expect(calls).toEqual(["/api/session/ses_other/prompt"])
expect(fixture.data.session.message.list("ses_other")).toEqual([])
expect(fixture.data.session.pending.list("ses_other")).toEqual([])
expect(fixture.data.session.submission.get("ses_other", "msg_existing")).toBeUndefined()
expect(fixture.data.session.message.get(sessionID, "msg_history")).toBe(history)
expect(
JSON.stringify({
messages: fixture.data.session.message.list(sessionID),
pending: fixture.data.session.pending.list(sessionID),
}),
).toBe(canonical)
},
)
test("a projected message acknowledges delivery and a late delivery event preserves canonical order", async () => {
const requested = Promise.withResolvers<string>()
using fixture = setup(async (request) => {
if (request.method === "GET")
return Response.json({
data: [
{ id: "msg_answer", type: "assistant", time: { created: 30 }, content: [] },
{ id: await requested.promise, type: "user", text: "Canonical text", time: { created: 25 } },
],
cursor: {},
})
requested.resolve((await request.json()).id)
return new Promise((_, reject) =>
request.signal.addEventListener("abort", () => reject(new Error("Closed")), { once: true }),
)
})
const sent = fixture.data.session.prompt({ sessionID, text: "Original text" })
const id = await requested.promise
await fixture.data.session.message.sync(sessionID)
expect(await sent).toMatchObject({ payload: { text: "Canonical text" }, timeCreated: 25 })
expect(fixture.data.session.submission.get(sessionID, id)).toBeUndefined()
const canonical = fixture.data.session.message.list(sessionID)
expect(canonical.map((row) => row.id)).toEqual([id, "msg_answer"])
fixture.emit({
id: "evt_delivered",
created: 25,
type: "session.inbox.delivered",
durable: { aggregateID: sessionID, seq: 2, version: 1 },
data: { sessionID, inboxID: id, messageID: id },
})
expect(fixture.data.session.input.list(sessionID)).toEqual([])
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
expect(fixture.data.session.message.list(sessionID).map((row) => row.id)).toEqual([id, "msg_answer"])
expect(fixture.data.session.message.get(sessionID, id)).toBe(canonical[0])
fixture.data.session.evict(sessionID)
expect(fixture.data.session.message.list(sessionID)).toEqual([])
})
test("delivery without an enqueue echo reads canonical content and refreshes authoritative order", async () => {
const requested = Promise.withResolvers<string>()
const reads: string[] = []
using fixture = setup(async (request) => {
if (request.method === "GET") {
const row = { id: await requested.promise, type: "user", text: "Canonical text", time: { created: 25 } }
if (new URL(request.url).pathname.endsWith(`/message/${row.id}`)) {
reads.push("get")
return Response.json({ data: row })
}
reads.push("list")
return Response.json({
data: [{ id: "msg_answer", type: "assistant", time: { created: 30 }, content: [] }, row],
cursor: {},
})
}
requested.resolve((await request.json()).id)
return new Promise((_, reject) =>
request.signal.addEventListener("abort", () => reject(new Error("Closed")), { once: true }),
)
})
const sent = fixture.data.session.prompt({ sessionID, text: "Original text" })
const id = await requested.promise
fixture.emit({
id: "evt_delivered",
created: 25,
type: "session.inbox.delivered",
durable: { aggregateID: sessionID, seq: 2, version: 1 },
data: { sessionID, inboxID: id, messageID: id },
})
expect(await sent).toMatchObject({ payload: { text: "Canonical text" }, timeCreated: 25 })
expect(reads).toEqual(["get", "list"])
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
expect(fixture.data.session.message.list(sessionID).map((row) => row.id)).toEqual([id, "msg_answer"])
expect(fixture.data.session.message.get(sessionID, id)).toBe(fixture.data.session.message.list(sessionID)[0])
})
test("a cancellation echo stops retries without resurrecting the prompt", async () => {
let calls = 0
using fixture = setup(async () => {
calls += 1
return new Response(null, { status: 503 })
})
const sent = fixture.data.session
.prompt({ sessionID, id: "msg_cancelled", text: "Cancel" })
.catch((error: unknown) => error)
await wait(() => fixture.data.session.submission.get(sessionID, "msg_cancelled")?.status === "retrying")
fixture.emit({
id: "evt_cancel",
created: 20,
type: "session.inbox.cancelled",
durable: { aggregateID: sessionID, seq: 2, version: 1 },
data: { sessionID, inboxID: "msg_cancelled" },
})
expect(await sent).toMatchObject({ reason: "cancelled" })
expect(fixture.data.session.message.list(sessionID)).toEqual([])
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
await Bun.sleep(300)
expect(calls).toBe(1)
})
test("a late HTTP success cannot resurrect a cancelled in-flight prompt", async () => {
const requested = Promise.withResolvers<void>()
const response = Promise.withResolvers<Response>()
using fixture = setup(async () => {
requested.resolve()
return response.promise
})
const sent = fixture.data.session
.prompt({ sessionID, id: "msg_cancelled", text: "Cancel" })
.catch((error: unknown) => error)
await requested.promise
fixture.emit({
id: "evt_cancel",
created: 20,
type: "session.inbox.cancelled",
durable: { aggregateID: sessionID, seq: 2, version: 1 },
data: { sessionID, inboxID: "msg_cancelled" },
})
expect(await sent).toMatchObject({ reason: "cancelled" })
response.resolve(Response.json({ data: item("msg_cancelled") }))
await Bun.sleep(10)
expect(fixture.data.session.message.list(sessionID)).toEqual([])
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
expect(fixture.data.session.input.list(sessionID)).toEqual([])
expect(fixture.data.session.submission.get(sessionID, "msg_cancelled")).toBeUndefined()
})
test("does not retry arbitrary preparation callbacks", async () => {
let calls = 0
using fixture = setup(async () => {
calls += 1
return new Response(null, { status: 503 })
})
let prepared = 0
await expect(
fixture.data.session.prompt({
sessionID,
text: "Prepare",
prepare: async () => {
prepared += 1
throw new Error("Preparation failed")
},
}),
).rejects.toThrow("Preparation failed")
expect(prepared).toBe(1)
expect(calls).toBe(0)
})
test("cancellation interrupts backoff and releases a following submission", async () => {
const calls: string[] = []
using fixture = setup(async (request) => {
if (request.method === "DELETE") return new Response(null, { status: 204 })
const body = await request.json()
calls.push(body.text)
if (body.text === "Cancel me") return new Response(null, { status: 503 })
return Response.json({ data: item(body.id) })
})
const first = fixture.data.session
.prompt({ sessionID, id: "msg_cancel", text: "Cancel me" })
.catch((error: unknown) => error)
const second = fixture.data.session.prompt({ sessionID, text: "Next" })
await wait(() => fixture.data.session.submission.get(sessionID, "msg_cancel")?.status === "retrying")
await fixture.data.session.pending.cancel(sessionID, "msg_cancel")
expect(await first).toMatchObject({ reason: "cancelled" })
await second
expect(calls).toEqual(["Cancel me", "Next"])
})
test("disposal stops a prompt waiting behind a gate without sending it", async () => {
let calls = 0
using fixture = setup(async () => {
calls += 1
return Response.json({ data: item("msg_disposed") })
})
const gate = Promise.withResolvers<void>()
const sent = fixture.data.session
.prompt({ sessionID, text: "Never send", gate: gate.promise })
.catch((error: unknown) => error)
fixture.dispose()
expect(await sent).toMatchObject({ reason: "cancelled" })
gate.resolve()
await Bun.sleep(10)
expect(calls).toBe(0)
})
test("diagnostics exclude prompt data and raw error messages", async () => {
let calls = 0
using fixture = setup(async (request) => {
calls += 1
if (calls === 1) throw new Error("private transport details")
return Response.json({ data: item((await request.json()).id) })
})
await fixture.data.session.prompt({ sessionID, text: "private prompt", metadata: { private: "private metadata" } })
expect(fixture.logs).toContainEqual(expect.objectContaining({ stage: "prompt", outcome: "retrying", attempt: 1 }))
expect(fixture.logs).toContainEqual(expect.objectContaining({ stage: "prompt", outcome: "accepted", attempt: 2 }))
expect(JSON.stringify(fixture.logs)).not.toContain("private")
})
test("local add, retry status, rejection, and HTTP acknowledgement leave canonical rows untouched", async () => {
const rejection = Promise.withResolvers<Response>()
let calls = 0
using fixture = setup(async (request) => {
if (request.method === "GET") {
if (request.url.endsWith("/inbox")) return Response.json({ data: [item("msg_pending")] })
return Response.json({
data: [{ id: "msg_history", type: "user", text: "History", time: { created: 1 } }],
cursor: {},
})
}
const body = await request.json()
if (body.text === "Accepted") return Response.json({ data: item(body.id) })
calls += 1
if (calls === 1) return new Response(null, { status: 503 })
return rejection.promise
})
await fixture.data.session.pending.sync(sessionID)
await fixture.data.session.message.sync(sessionID)
const messages = fixture.data.session.message.list(sessionID)
const pending = fixture.data.session.pending.list(sessionID)
const content = JSON.stringify({ messages, pending })
const assertCanonical = () => {
expect(JSON.stringify({ messages, pending })).toBe(content)
expect(fixture.data.session.message.list(sessionID).slice(0, 2)).toEqual(messages)
messages.forEach((row, index) => {
expect(fixture.data.session.message.list(sessionID)[index]).toBe(row)
expect(fixture.data.session.message.get(sessionID, row.id)).toBe(row)
})
expect(fixture.data.session.pending.list(sessionID)[0]).toBe(pending[0])
}
const sent = fixture.data.session
.prompt({ sessionID, id: "msg_rejected", text: "Reject me" })
.catch((error: unknown) => error)
assertCanonical()
expect(messages.map((row) => row.id)).toEqual(["msg_history", "msg_pending"])
expect(pending.map((row) => row.id)).toEqual(["msg_pending"])
expect(fixture.data.session.message.list(sessionID)).toHaveLength(3)
expect(fixture.data.session.pending.list(sessionID)).toHaveLength(2)
const preview = fixture.data.session.message.list(sessionID)[2]
expect(fixture.data.session.message.get(sessionID, "msg_rejected")).toBe(preview)
await wait(() => fixture.data.session.submission.get(sessionID, "msg_rejected")?.status === "retrying")
assertCanonical()
expect(fixture.data.session.message.list(sessionID)[2]).toBe(preview)
rejection.resolve(new Response(null, { status: 422 }))
expect(await sent).toBeInstanceOf(Error)
assertCanonical()
expect(fixture.data.session.message.list(sessionID)).toBe(messages)
expect(fixture.data.session.pending.list(sessionID)).toBe(pending)
await fixture.data.session.prompt({ sessionID, id: "msg_accepted", text: "Accepted" })
assertCanonical()
expect(fixture.data.session.message.list(sessionID)).toHaveLength(3)
expect(fixture.data.session.pending.list(sessionID)).toHaveLength(2)
fixture.enqueue("msg_accepted")
expect(fixture.data.session.message.list(sessionID)).toHaveLength(3)
expect(fixture.data.session.pending.list(sessionID)).toHaveLength(2)
expect(fixture.data.session.message.get(sessionID, "msg_history")).toBe(messages[0])
expect(fixture.data.session.pending.list(sessionID)[0]).toBe(pending[0])
})
test("absent snapshots and cache eviction preserve local work until a positive message observation", async () => {
const gate = Promise.withResolvers<void>()
const rows: SessionMessageInfo[] = []
using fixture = setup(async (request) => {
if (request.method === "GET")
return Response.json(request.url.endsWith("/inbox") ? { data: [] } : { data: rows, cursor: {} })
return Response.json({ data: item((await request.json()).id) })
})
const sent = fixture.data.session.prompt({ sessionID, id: "msg_local", text: "Local", gate: gate.promise })
await fixture.data.session.pending.sync(sessionID)
await fixture.data.session.message.sync(sessionID)
expect(fixture.data.session.message.list(sessionID)).toMatchObject([{ id: "msg_local", text: "Local" }])
expect(fixture.data.session.pending.list(sessionID)).toMatchObject([{ id: "msg_local" }])
gate.resolve()
await sent
fixture.data.session.pending.invalidate(sessionID)
fixture.data.session.message.invalidate(sessionID)
await fixture.data.session.pending.sync(sessionID)
await fixture.data.session.message.sync(sessionID)
expect(fixture.data.session.message.list(sessionID)).toMatchObject([{ id: "msg_local", text: "Accepted" }])
expect(fixture.data.session.pending.list(sessionID)).toMatchObject([{ id: "msg_local" }])
fixture.data.session.evict(sessionID)
expect(fixture.data.session.message.list(sessionID)).toHaveLength(1)
rows.push({ id: "msg_local", type: "user", text: "Canonical", time: { created: 25 } })
await fixture.data.session.message.sync(sessionID)
expect(fixture.data.session.message.list(sessionID)).toEqual(rows)
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
fixture.data.session.evict(sessionID)
expect(fixture.data.session.message.list(sessionID)).toEqual([])
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
expect(fixture.data.session.input.list(sessionID)).toEqual([])
})
test.each(["accepted", "failed"])(
"an older message page retires %s local work without resurrection on eviction",
async (status) => {
const canonical = { id: "msg_older", type: "user", text: "Canonical", time: { created: 1 } }
let calls = 0
using fixture = setup(async (request) => {
if (request.method !== "GET") {
calls += 1
if (status === "failed") return new Response(null, { status: 503 })
return Response.json({ data: item((await request.json()).id) })
}
if (request.url.endsWith("/inbox")) return Response.json({ data: [] })
if (new URL(request.url).searchParams.has("cursor")) return Response.json({ data: [canonical], cursor: {} })
return Response.json({
data: [{ id: "msg_newer", type: "assistant", time: { created: 2 }, content: [] }],
cursor: { next: "older" },
})
})
const sent = fixture.data.session.prompt({ sessionID, id: canonical.id, text: "Local" })
if (status === "failed") await expect(sent).rejects.toMatchObject({ reason: "failed" })
if (status === "accepted") await sent
await fixture.data.session.pending.sync(sessionID)
await fixture.data.session.message.sync(sessionID)
expect(fixture.data.session.message.list(sessionID).map((row) => row.id)).toEqual(["msg_newer", canonical.id])
expect(fixture.data.session.pending.list(sessionID)).toMatchObject([{ id: canonical.id }])
expect(fixture.data.session.message.more(sessionID)).toBe(true)
if (status === "failed")
expect(fixture.data.session.submission.get(sessionID, canonical.id)).toMatchObject({ status: "failed" })
await fixture.data.session.message.loadMore(sessionID)
expect(fixture.data.session.message.list(sessionID).map((row) => row.id)).toEqual([canonical.id, "msg_newer"])
expect(fixture.data.session.message.get(sessionID, canonical.id)).toEqual(canonical)
expect(fixture.data.session.message.get(sessionID, canonical.id)).toBe(
fixture.data.session.message.list(sessionID)[0],
)
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
expect(fixture.data.session.input.list(sessionID)).toEqual([])
expect(fixture.data.session.submission.get(sessionID, canonical.id)).toBeUndefined()
expect(fixture.data.session.message.more(sessionID)).toBe(false)
fixture.data.session.evict(sessionID)
expect(fixture.data.session.message.list(sessionID)).toEqual([])
expect(fixture.data.session.pending.list(sessionID)).toEqual([])
expect(calls).toBe(status === "accepted" ? 1 : 4)
},
)
test.each(["prompt", "compaction"])(
"local %s first preserves both overlays and serializes admissions",
async (first) => {
const promptResponse = Promise.withResolvers<void>()
const compactResponse = Promise.withResolvers<void>()
const calls: string[] = []
using fixture = setup(async (request) => {
if (request.method === "GET")
return Response.json(request.url.endsWith("/inbox") ? { data: [] } : { data: [], cursor: {} })
const body = await request.json()
if (request.url.endsWith("/prompt")) {
calls.push("prompt")
await promptResponse.promise
return Response.json({ data: item(body.id) })
}
calls.push("compaction")
await compactResponse.promise
return Response.json({ data: { ...item(body.id), type: "compaction", payload: {} } })
})
await fixture.data.session.pending.sync(sessionID)
await fixture.data.session.message.sync(sessionID)
const canonicalPending = fixture.data.session.pending.list(sessionID)
const canonicalMessages = fixture.data.session.message.list(sessionID)
const submit = (kind: string) =>
kind === "prompt"
? fixture.data.session.prompt({ sessionID, text: "Follow up" })
: fixture.data.session.compact({ sessionID })
const second = first === "prompt" ? "compaction" : "prompt"
const firstRequest = submit(first)
const secondRequest = submit(second)
expect(fixture.data.session.pending.list(sessionID).map((row) => row.type)).toEqual([
first === "prompt" ? "user" : "compaction",
second === "prompt" ? "user" : "compaction",
])
expect(fixture.data.session.message.list(sessionID)).toMatchObject([{ type: "user", text: "Follow up" }])
expect(fixture.data.session.input.list(sessionID)).toHaveLength(1)
expect(canonicalPending).toEqual([])
expect(canonicalMessages).toEqual([])
await wait(() => calls.length === 1)
expect(calls).toEqual([first])
if (first === "prompt") promptResponse.resolve()
if (first === "compaction") compactResponse.resolve()
await firstRequest
await wait(() => calls.length === 2)
expect(calls).toEqual([first, second])
promptResponse.resolve()
compactResponse.resolve()
await secondRequest
expect(fixture.data.session.pending.list(sessionID)).toHaveLength(2)
expect(fixture.data.session.message.list(sessionID)).toHaveLength(1)
},
)
test("local changes do not recompute another session's structural selectors", async () => {
// Bun normally resolves Solid's inert SSR build; exercise real subscriptions.
if (isServer) {
const child = Bun.spawn(
[
process.execPath,
"--conditions=browser",
"test",
import.meta.path,
"--test-name-pattern",
"local changes do not recompute",
],
{ stdout: "pipe", stderr: "pipe" },
)
const [code, stdout, stderr] = await Promise.all([
child.exited,
new Response(child.stdout).text(),
new Response(child.stderr).text(),
])
expect({ code, output: code === 0 ? "" : stdout + stderr }).toEqual({ code: 0, output: "" })
return
}
let calls = 0
using fixture = setup(async (request) => {
if (request.method === "GET")
return Response.json(request.url.endsWith("/inbox") ? { data: [] } : { data: [], cursor: {} })
calls += 1
return new Response(null, { status: 503 })
})
await fixture.data.session.pending.sync("ses_other")
await fixture.data.session.message.sync("ses_other")
let computations = 0
const observer = createRoot((dispose) => {
createComputed(() => {
fixture.data.session.message.list("ses_other").map((row) => row.id)
fixture.data.session.pending.list("ses_other").map((row) => row.type)
fixture.data.session.input.list("ses_other").slice()
computations += 1
})
return { [Symbol.dispose]: dispose }
})
using subscription = observer
expect(computations).toBe(1)
const sent = fixture.data.session
.prompt({ sessionID, id: "msg_local", text: "Local" })
.catch((error: unknown) => error)
await wait(() => fixture.data.session.submission.get(sessionID, "msg_local")?.status === "retrying")
expect(calls).toBe(1)
expect(computations).toBe(1)
fixture.enqueue("msg_local")
await sent
expect(computations).toBe(1)
const gate = Promise.withResolvers<void>()
const other = fixture.data.session
.prompt({ sessionID: "ses_other", text: "Positive control", gate: gate.promise })
.catch((error: unknown) => error)
expect(computations).toBeGreaterThan(1)
fixture.dispose()
expect(await other).toMatchObject({ reason: "cancelled" })
gate.resolve()
})
const sessionID = "ses_retry"
const item = (id: string): SessionInboxUser => ({
id,
sessionID,
type: "user",
payload: { text: "Accepted" },
delivery: "steer",
timeCreated: 10,
})
function setup(handle: (request: Request) => Promise<Response>) {
const listeners = new Set<Parameters<CreateDataInput["event"]["listen"]>[0]>()
const logs: Readonly<Record<string, unknown>>[] = []
const api = OpenCode.make({
baseUrl: "http://opencode.local",
fetch: async (input, init) => handle(input instanceof Request ? input : new Request(input, init)),
})
const root = createRoot((dispose) => ({
data: createData({
api: () => api,
directory: "/project",
event: {
on: () => () => {},
listen(handler) {
listeners.add(handler)
return () => listeners.delete(handler)
},
},
log: {
info: (_message, data) => {
if (data) logs.push(data)
},
},
}),
dispose,
}))
const emit = (details: OpenCodeEvent) => listeners.forEach((listener) => listener({ name: details.type, details }))
return {
data: root.data,
api,
dispose: root.dispose,
[Symbol.dispose]: root.dispose,
logs,
emit,
enqueue(id: string) {
emit({
id: "evt_enqueue",
created: 10,
type: "session.inbox.enqueued",
durable: { aggregateID: sessionID, seq: 1, version: 1 },
data: { sessionID, inboxID: id, item: { type: "user", payload: { text: "Accepted" }, delivery: "steer" } },
})
},
}
}
async function wait(predicate: () => boolean) {
for (let attempt = 0; attempt < 200; attempt++) {
if (predicate()) return
await Bun.sleep(5)
}
throw new Error("Timed out waiting for submission")
}
+12 -13
View File
@@ -53,6 +53,7 @@ import { useConfig } from "../../config"
import { usePromptMove } from "./move"
import { resolvePastedAttachments } from "./local-attachment"
import { locationKey, useData } from "../../context/data"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { useLocation } from "../../context/location"
import { Keymap, type KeymapCommand } from "../../context/keymap"
import { abbreviateHome } from "../../runtime"
@@ -1349,10 +1350,9 @@ export function Prompt(props: PromptProps) {
}
}
}
// 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.
// Admission owns retries and their identity. Definitive failures restore
// the draft; uncertain outcomes remain on the original retryable row.
let cancelCommit: (() => void) | undefined
data.session
.prompt({
sessionID: target,
@@ -1361,19 +1361,18 @@ export function Prompt(props: PromptProps) {
agents: entry.agents,
skills: entry.skills?.length ? entry.skills : undefined,
delivery,
model,
gate: newSession?.gate,
prepare: () => {
// Commit the captured selection after earlier admissions, including
// compaction setup. Cached state may still precede their SSE echoes;
// the server makes an unchanged selection a no-op.
const cancelCommit = local.model.trackSessionCommit(target, model)
return client.api.session.switchModel({ sessionID: target, model }).catch((error) => {
cancelCommit()
throw new Error(`Failed to switch model: ${errorMessage(error)}`, { cause: error })
})
prepare: async () => {
// Track selection after earlier admissions. The data layer owns
// the retryable model commit and skips it once it succeeds.
cancelCommit = local.model.trackSessionCommit(target, model)
},
})
.catch((error) => {
cancelCommit?.()
// Unknown outcomes retain their original admission ID for explicit retry.
if (error instanceof PromptSubmissionError) return
if (newSession) return newSession.recover(error)
toast.show({ title: "Failed to send prompt", message: errorMessage(error), variant: "error" })
restoreEntry()
+3
View File
@@ -3,6 +3,7 @@ import type { Plugin } from "@opencode-ai/plugin/tui"
import { createStore } from "solid-js/store"
import { createSimpleContext } from "./helper"
import { useClient } from "./client"
import { useLog } from "./log"
export { locationKey } from "@opencode-ai/client/solid"
export type { FormWithLocation } from "@opencode-ai/client/solid"
@@ -11,11 +12,13 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
name: "Data",
init: (props: { directory: string }) => {
const client = useClient()
const log = useLog({ component: "prompt" })
const data = createData({
api: () => client.api,
event: client.event,
connection: client.connection,
directory: props.directory,
log,
})
data satisfies Plugin.Context["data"]
const [generatingTitles, setGeneratingTitles] = createStore<Record<string, boolean | undefined>>({})
+56 -12
View File
@@ -21,6 +21,8 @@ import { mkdir, writeFile } from "node:fs/promises"
import { useRoute, useRouteData } from "../../context/route"
import { createStore } from "solid-js/store"
import { useData } from "../../context/data"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { PromptSubmissionStatus } from "./prompt-submission"
import { SplitBorder } from "../../ui/border"
import { useTuiPaths, useTuiTerminalEnvironment } from "../../context/runtime"
import { Spinner, SPINNER_FRAMES } from "../../component/spinner"
@@ -124,7 +126,7 @@ const BACKGROUND_TOOL_HINT_DELAY = 3_000
// The tail comfortably overfills a tall viewport; older rows mount as the reader approaches them.
const TRANSCRIPT_TAIL_ROWS = 40
const TRANSCRIPT_BACKFILL_CHUNK = 60
type PendingAction = "steer" | "queue" | "cancel"
type PendingAction = "steer" | "queue" | "cancel" | "retry"
const context = createContext<{
/** Content width: terminal width minus vertical tabs, sidebar, and padding. */
@@ -570,16 +572,19 @@ export function Session(props: {
const mutatePending = async (action: PendingAction, inboxID: string) => {
const result = await runPendingAction(inboxID, async () => {
const request =
action === "steer"
? client.api.session.inbox.steer({ sessionID: route.sessionID, inboxID })
: action === "queue"
? client.api.session.inbox.queue({ sessionID: route.sessionID, inboxID })
: client.api.session.inbox.cancel({ sessionID: route.sessionID, inboxID })
action === "retry"
? data.session.submission.retry(route.sessionID, inboxID)
: action === "steer"
? client.api.session.inbox.steer({ sessionID: route.sessionID, inboxID })
: action === "queue"
? client.api.session.inbox.queue({ sessionID: route.sessionID, inboxID })
: data.session.pending.cancel(route.sessionID, inboxID)
const error = await request.then(
() => undefined,
(error) => error,
)
if (!error) return true
if (error instanceof PromptSubmissionError) return false
const label = action === "cancel" ? "delete" : action
toast.show({ title: `Failed to ${label} pending prompt`, message: errorMessage(error), variant: "error" })
return false
@@ -591,12 +596,14 @@ export function Session(props: {
<DialogSelect
title="Queued prompts"
options={queuedPrompts().map((prompt, index) => ({
title: prompt.text,
title: `${data.session.submission.get(route.sessionID, prompt.id)?.status === "failed" ? "Send not confirmed: " : ""}${prompt.text}`,
value: prompt.id,
footer: `${index + 1} of ${queuedPrompts().length}`,
}))}
onSelect={(option) => {
void mutatePending("steer", option.value).then((steered) => {
const state = data.session.submission.get(route.sessionID, option.value)
if (state && state.status !== "failed") return
void mutatePending(state ? "retry" : "steer", option.value).then((steered) => {
if (steered) dialog.clear()
})
}}
@@ -612,7 +619,7 @@ export function Session(props: {
},
},
]}
footerHints={[{ title: "steer", label: "enter" }]}
footerHints={[{ title: "steer / retry", label: "enter" }]}
/>
))
const unavailable = (feature: string) => {
@@ -1266,6 +1273,21 @@ export function Session(props: {
dialog.clear()
},
},
{
title: "Retry failed prompt",
id: "session.retry_prompt",
group: "Prompt",
enabled: pendingUsers().some(
(item) => data.session.submission.get(route.sessionID, item.id)?.status === "failed",
),
run: () => {
const item = pendingUsers().find(
(item) => data.session.submission.get(route.sessionID, item.id)?.status === "failed",
)
if (item) void mutatePending("retry", item.id)
dialog.clear()
},
},
{
title: "View queued prompts",
id: "session.queued_prompts",
@@ -1441,7 +1463,7 @@ export function Session(props: {
</box>
<box flexShrink={0}>
<Show when={!composer.open && !disabled() && queuedPrompts().length > 0}>
<QueuedPromptDock prompts={queuedPrompts()} onOpen={openQueuedPrompts} />
<QueuedPromptDock sessionID={route.sessionID} prompts={queuedPrompts()} onOpen={openQueuedPrompts} />
</Show>
<Slot path="session.composer.top" input={{ sessionID: route.sessionID }} />
<Composer
@@ -2372,6 +2394,7 @@ function UserMessage(props: { message: SessionMessageUser }) {
const [hover, setHover] = createSignal(false)
const color = createMemo(() => local.agent.color(data.session.get(ctx.sessionID)?.agent ?? "build"))
const delivery = createMemo(() => ctx.pendingDelivery(props.message.id))
const submission = createMemo(() => data.session.submission.get(ctx.sessionID, props.message.id))
const dialog = useDialog()
const renderer = useRenderer()
const promptRef = usePromptRef()
@@ -2398,6 +2421,23 @@ function UserMessage(props: { message: SessionMessageUser }) {
}}
onMouseUp={() => {
if (renderer.getSelection()?.getSelectedText()) return
if (submission()) {
dialog.replace(() => (
<DialogSelect
title="Prompt submission"
options={[
...(submission()?.status === "failed" ? [{ title: "Retry send", value: "retry" as const }] : []),
{ title: "Cancel send", value: "cancel" as const },
]}
onSelect={(option) => {
void ctx.mutatePending(option.value, props.message.id).then((ok) => {
if (ok) dialog.clear()
})
}}
/>
))
return
}
if (delivery() === "steer") {
dialog.replace(() => (
<DialogSelect
@@ -2428,6 +2468,7 @@ function UserMessage(props: { message: SessionMessageUser }) {
flexShrink={0}
>
<text fg={theme.text.default}>{props.message.text}</text>
<PromptSubmissionStatus sessionID={ctx.sessionID} messageID={props.message.id} />
<Show when={skills().length}>
<box flexDirection="row" paddingTop={1} gap={1} flexWrap="wrap">
<For each={skills()}>
@@ -2482,7 +2523,7 @@ function UserMessage(props: { message: SessionMessageUser }) {
)
}
function QueuedPromptDock(props: { prompts: { id: string; text: string }[]; onOpen: () => void }) {
function QueuedPromptDock(props: { sessionID: string; prompts: { id: string; text: string }[]; onOpen: () => void }) {
const theme = useTheme("elevated")
const [hover, setHover] = createSignal(false)
const next = createMemo(() => props.prompts[0]?.text.replaceAll("\n", " "))
@@ -2503,12 +2544,15 @@ function QueuedPromptDock(props: { prompts: { id: string; text: string }[]; onOp
paddingLeft={2}
paddingRight={1}
backgroundColor={hover() ? theme.raise(theme.background.default) : theme.background.default}
flexDirection="row"
flexDirection="column"
>
<text fg={theme.text.subdued} wrapMode="none" truncate flexGrow={1} flexShrink={1} minWidth={0}>
<span style={{ fg: theme.text.default }}>{props.prompts.length} queued</span>
<Show when={next()}>{(text) => <> · {text()}</>}</Show>
</text>
<Show when={props.prompts[0]}>
{(prompt) => <PromptSubmissionStatus sessionID={props.sessionID} messageID={prompt().id} />}
</Show>
</box>
</box>
)
@@ -0,0 +1,60 @@
import { createMemo, Show } from "solid-js"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { useData } from "../../context/data"
import { useTheme } from "../../context/theme"
import { useToast } from "../../ui/toast"
import { errorMessage } from "../../util/error"
export function PromptSubmissionStatus(props: { sessionID: string; messageID: string }) {
const data = useData()
const theme = useTheme()
const toast = useToast()
const submission = createMemo(() => data.session.submission.get(props.sessionID, props.messageID))
const act = (action: "retry" | "cancel") => {
const request =
action === "retry"
? data.session.submission.retry(props.sessionID, props.messageID)
: data.session.pending.cancel(props.sessionID, props.messageID)
void request.catch((error: unknown) => {
if (error instanceof PromptSubmissionError) return
toast.show({ title: "Prompt submission failed", message: errorMessage(error), variant: "error" })
})
}
return (
<Show when={submission()}>
{(state) => (
<box flexDirection="row" gap={2} flexWrap="wrap" paddingTop={1}>
<text fg={state().status === "failed" ? theme.text.feedback.error.default : theme.text.subdued}>
{state().status === "failed"
? "Send not confirmed"
: state().status === "retrying"
? `Retrying send (attempt ${state().attempt + 1}/4)`
: "Sending..."}
</text>
<Show when={state().status === "failed"}>
<text
fg={theme.text.action.primary.default}
onMouseUp={(event) => {
if (event.button !== 0) return
event.stopPropagation()
act("retry")
}}
>
retry
</text>
</Show>
<text
fg={theme.text.action.primary.default}
onMouseUp={(event) => {
if (event.button !== 0) return
event.stopPropagation()
act("cancel")
}}
>
cancel
</text>
</box>
)}
</Show>
)
}
+24 -4
View File
@@ -761,9 +761,19 @@ test("session startup prompt is submitted exactly once", async () => {
data: [{ id: "model", providerID: "provider", name: "Model", variants: [] }],
})
if (url.pathname === "/api/session/dummy/prompt") {
bodies.push(await request.json())
const input = await request.json()
bodies.push(input)
promptSubmitted.resolve()
return json({ data: {} })
return json({
data: {
id: input.id,
sessionID: "dummy",
type: "user",
payload: { text: input.text },
delivery: input.delivery ?? "steer",
timeCreated: 1,
},
})
}
}, events)
const server = Bun.serve({ port: 0, fetch: (request) => calls.fetch(request) })
@@ -836,8 +846,18 @@ test.each([false, true])("uses the resolved launch directory for new prompts (fa
return json({ data: session })
}
if (/^\/api\/session\/[^/]+\/prompt$/.test(url.pathname)) {
submitted.resolve(await request.json())
return json({ data: {} })
const input = await request.json()
submitted.resolve(input)
return json({
data: {
id: input.id,
sessionID: url.pathname.split("/")[3],
type: "user",
payload: { text: input.text },
delivery: input.delivery ?? "steer",
timeCreated: 1,
},
})
}
if (/^\/api\/session\/[^/]+\/(message|inbox|permission)$/.test(url.pathname))
return json({ data: [], cursor: {} })
+10 -6
View File
@@ -2,6 +2,7 @@
import { expect, test } from "bun:test"
import { testRender } from "@opentui/solid"
import type { OpenCodeEvent } from "@opencode-ai/client"
import { PromptSubmissionError } from "@opencode-ai/client/solid"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { Bus } from "@opencode-ai/core/bus"
import { Event } from "@opencode-ai/schema/event"
@@ -3198,9 +3199,9 @@ test("admits prompts optimistically and reconciles with the durable echo", async
expect(echoed.time.created).toBe(5)
expect(echoed.files).toEqual([echoFile])
// A late transport failure after the echo must not delete acknowledged state.
// The durable echo settles admission even when its HTTP response is lost.
release(json({ _tag: "UnknownError", message: "response lost" }, { status: 500 }))
expect(await settled).toBeDefined()
expect(await settled).toBeUndefined()
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 {
@@ -3416,11 +3417,10 @@ test("a retry under the same client-minted ID cannot duplicate rows", async () =
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.
// Repeated callers join the HTTP-accepted operation until its durable echo.
await sync.session.prompt({ sessionID, id: messageID, text: "hello" })
expect(posts).toEqual([messageID, messageID])
expect(posts).toEqual([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])
@@ -3439,7 +3439,11 @@ test("a retry under the same client-minted ID cannot duplicate rows", async () =
await wait(() => received.includes("session.inbox.enqueued"))
unsubscribe()
fail = true
await expect(sync.session.prompt({ sessionID, id: messageID, text: "hello" })).rejects.toThrow()
await expect(sync.session.prompt({ sessionID, id: messageID, text: "hello" })).rejects.toBeInstanceOf(
PromptSubmissionError,
)
expect(posts).toEqual(Array(5).fill(messageID))
expect(sync.session.submission.get(sessionID, messageID)).toMatchObject({ status: "failed", attempt: 4 })
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 {
+122
View File
@@ -0,0 +1,122 @@
import { expect, test } from "bun:test"
import { createTestRenderer } from "@opentui/core/testing"
import { TextareaRenderable } from "@opentui/core"
import { Effect, FileSystem } from "effect"
import { Global } from "@opencode-ai/util/global"
import { createEventStream, createFetch, directory, json } from "./fixture/tui-client"
import { tmpdir } from "./fixture/fixture"
test.each(["steer", "queue"] as const)(
"retains and retries an unconfirmed %s with the original ID",
async (delivery) => {
await using state = await tmpdir()
const setup = await createTestRenderer({ width: 120, height: 30, useThread: false, kittyKeyboard: true })
setup.renderer.start()
const ready = Promise.withResolvers<void>()
const retrying = Promise.withResolvers<void>()
const exhausted = Promise.withResolvers<void>()
const accepted = Promise.withResolvers<void>()
const events = createEventStream()
const sessionID = `ses_retry_${delivery}`
const location = { directory, project: { id: "project", directory, canonical: directory } }
const bodies: string[] = []
const calls = createFetch(async (url, request) => {
if (url.pathname === `/api/session/${sessionID}`)
return json({
data: {
id: sessionID,
projectID: "project",
title: "Prompt retry",
agent: "build",
model: { providerID: "demo", id: "model" },
location: { directory },
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: 0, updated: 0 },
},
})
if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} })
if (url.pathname === `/api/session/${sessionID}/inbox` || url.pathname === `/api/session/${sessionID}/permission`)
return json({ data: [] })
if (url.pathname === "/api/agent")
return json({ location, data: [{ id: "build", mode: "primary", hidden: false, permissions: [] }] })
if (url.pathname === "/api/provider") return json({ location, data: [{ id: "demo", name: "Demo" }] })
if (url.pathname === "/api/model")
return json({ location, data: [{ id: "model", providerID: "demo", name: "Demo Model", variants: [] }] })
if (url.pathname === `/api/session/${sessionID}/prompt`) {
const text = await request.text()
bodies.push(text)
if (bodies.length <= 4) return new Response(null, { status: 503 })
const body = JSON.parse(text)
const item = { type: "user" as const, payload: { text: body.text }, delivery }
events.emit({
id: "evt_accepted",
created: 10,
type: "session.inbox.enqueued",
durable: { aggregateID: sessionID, seq: 1, version: 1 },
data: { sessionID, inboxID: body.id, item },
})
return json({ data: { id: body.id, sessionID, timeCreated: 10, ...item } })
}
return undefined
}, events)
const server = Bun.serve({ port: 0, idleTimeout: 0, fetch: (request) => calls.fetch(request) })
const { run } = await import("../src/app")
const task = Effect.runPromise(
run({
app: { name: "test", version: "test", channel: "test" },
server: { endpoint: { url: server.url.toString() } },
config: {
get: async () => ({ animations: false, keybinds: { "prompt.queue": "f6" } }),
update: async () => ({}),
},
packages: { resolve: async () => undefined },
terminalHandoff: async () => ({ renderer: setup.renderer, mode: "dark", complete: ready.resolve }),
args: { sessionID },
log: (_level, message, tags) => {
if (message !== "prompt submission") return
if (tags.outcome === "retrying") retrying.resolve()
if (tags.outcome === "failed") exhausted.resolve()
if (tags.outcome === "accepted") accepted.resolve()
},
}).pipe(Effect.provide(Global.layerWith({ state: state.path })), Effect.provide(FileSystem.layerNoop({}))),
)
try {
await ready.promise
await setup.waitForFrame((frame) => frame.includes("Demo Model"))
await setup.mockInput.typeText("Please keep this prompt")
if (delivery === "queue") setup.mockInput.pressKey("F6")
else setup.mockInput.pressEnter()
await retrying.promise
await setup.waitForFrame((frame) => frame.includes("Retrying send"))
await exhausted.promise
const failed = await setup.waitForFrame((frame) => frame.includes("Send not confirmed"))
expect(failed).toContain("Please keep this prompt")
const input = setup.renderer.currentFocusedRenderable
expect(input).toBeInstanceOf(TextareaRenderable)
if (!(input instanceof TextareaRenderable)) throw new Error("composer is not focused")
expect(input.plainText).toBe("")
expect(bodies).toHaveLength(4)
setup.resize(60, 30)
await setup.waitForVisualIdle()
const narrow = await setup.waitForFrame((frame) => frame.includes("Send not confirmed"))
expect(narrow).toContain("Please keep this prompt")
expect(narrow).toContain("cancel")
const lines = narrow.split("\n")
const row = lines.findLastIndex((line) => line.includes("retry"))
expect(row).toBeGreaterThanOrEqual(0)
await setup.mockMouse.click(lines[row].indexOf("retry"), row)
await accepted.promise
await setup.waitForFrame((frame) => !frame.includes("Send not confirmed") && !frame.includes("Retrying send"))
expect(bodies).toHaveLength(5)
expect(new Set(bodies).size).toBe(1)
expect(JSON.parse(bodies[0]).delivery).toBe(delivery)
expect(input.plainText).toBe("")
} finally {
setup.renderer.destroy()
await task
await server.stop()
}
},
)