Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 91 Packages Projects Releases Wiki Activity

mini: settle on wait and durable pending

Browse Source 
Turn completion no longer races stream lifecycle events.
Use session.wait as the authoritative idle fence, then
reconcile projected output to recover any suffix missed
by the live stream.
This commit is contained in:
Simon Klee
2026-07-20 15:10:06 +02:00
parent 61b7caa0b5
commit ccab563c39
16 changed files with 909 additions and 658 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli/src/run/noninteractive.ts
+184 -19
View File
@@ -5,6 +5,7 @@ import type {
LocationRef,
OpenCodeClient,
SessionMessageAssistantTool,
SessionMessageInfo,
} from "@opencode-ai/client/promise"
import { SessionMessage } from "@opencode-ai/schema/session-message"
import { EOL } from "node:os"
@@ -75,6 +76,9 @@ export async function runNonInteractivePrompt(input: Input) {
const messageID = SessionMessage.ID.create()
const starts = new Map<string, StartedPart>()
const tools = new Map<string, ToolState>()
const renderedText = new Map<string, string>()
const renderedReasoning = new Map<string, string>()
const renderedTools = new Set<string>()
let submitted = false
let promoted = false
let emittedError = false
@@ -82,6 +86,8 @@ export async function runNonInteractivePrompt(input: Input) {
let formCancelled = false
let interrupted = false
let v1InvalidOutput = false
let prePromotionError: { message: string; [key: string]: unknown } | undefined
let finalizing = false
let admission: AbortController | undefined
let pendingStep: { timestamp: number; part: Record<string, unknown>; label: string } | undefined
@@ -104,6 +110,17 @@ export async function runNonInteractivePrompt(input: Input) {
UI.empty()
}
const writeReasoning = (part: { text: string; [key: string]: unknown }, timestamp: number) => {
if (emit("reasoning", timestamp, { part })) return
const text = part.text.trim()
if (!text) return
const line = `Thinking: ${text}`
if (!process.stdout.isTTY) return void process.stdout.write(line + EOL)
UI.empty()
UI.println(`${UI.Style.TEXT_DIM}\u001b[3m${line}\u001b[0m${UI.Style.TEXT_NORMAL}`)
UI.empty()
}
const flushStep = () => {
if (!pendingStep) return
const value = pendingStep
@@ -181,6 +198,7 @@ export async function runNonInteractivePrompt(input: Input) {
if (event.type === "session.input.promoted") {
if (event.data.inputID === messageID) {
promoted = true
prePromotionError = undefined
continue
}
}
@@ -191,7 +209,12 @@ export async function runNonInteractivePrompt(input: Input) {
) {
return
}
if (!promoted && event.type === "session.execution.failed") {
prePromotionError = event.data.error
continue
}
if (!promoted) continue
if (finalizing) continue
if (event.type === "session.step.started") {
const part = {
@@ -219,12 +242,16 @@ export async function runNonInteractivePrompt(input: Input) {
if (event.type === "session.text.started") {
flushStep()
starts.set("text", { id: partID(event.id), timestamp: time })
starts.set(`text\u0000${contentKey(event.data.assistantMessageID, event.data.ordinal)}`, {
id: partID(event.id),
timestamp: time,
})
continue
}
if (event.type === "session.text.ended") {
const started = starts.get("text")
starts.delete("text")
const key = contentKey(event.data.assistantMessageID, event.data.ordinal)
const started = starts.get(`text\u0000${key}`)
starts.delete(`text\u0000${key}`)
const part = {
id: started?.id ?? partID(event.id),
sessionID: input.sessionID,
@@ -233,18 +260,23 @@ export async function runNonInteractivePrompt(input: Input) {
text: event.data.text,
time: { start: started?.timestamp ?? time, end: time },
}
renderedText.set(key, event.data.text)
writeText(part, time)
continue
}
if (event.type === "session.reasoning.started") {
flushStep()
starts.set("reasoning", { id: partID(event.id), timestamp: time })
starts.set(`reasoning\u0000${contentKey(event.data.assistantMessageID, event.data.ordinal)}`, {
id: partID(event.id),
timestamp: time,
})
continue
}
if (event.type === "session.reasoning.ended" && input.thinking) {
const started = starts.get("reasoning")
starts.delete("reasoning")
const key = contentKey(event.data.assistantMessageID, event.data.ordinal)
const started = starts.get(`reasoning\u0000${key}`)
starts.delete(`reasoning\u0000${key}`)
const part = {
id: started?.id ?? partID(event.id),
sessionID: input.sessionID,
@@ -254,17 +286,8 @@ export async function runNonInteractivePrompt(input: Input) {
metadata: event.data.state,
time: { start: started?.timestamp ?? time, end: time },
}
if (emit("reasoning", time, { part })) continue
const text = part.text.trim()
if (!text) continue
const line = `Thinking: ${text}`
if (!process.stdout.isTTY) {
process.stdout.write(line + EOL)
continue
}
UI.empty()
UI.println(`${UI.Style.TEXT_DIM}\u001b[3m${line}\u001b[0m${UI.Style.TEXT_NORMAL}`)
UI.empty()
renderedReasoning.set(key, event.data.text)
writeReasoning(part, time)
continue
}
@@ -360,6 +383,7 @@ export async function runNonInteractivePrompt(input: Input) {
},
}
tools.delete(key)
renderedTools.add(key)
if (!emit("tool_use", time, { part })) await input.renderTool(tool)
continue
}
@@ -405,6 +429,7 @@ export async function runNonInteractivePrompt(input: Input) {
},
}
tools.delete(key)
renderedTools.add(key)
if (input.compatibility === "v1" && (permissionRejected || formCancelled)) continue
if (!emit("tool_use", time, { part })) {
if (toolOutputText(current.tool, current.content).trim())
@@ -480,6 +505,122 @@ export async function runNonInteractivePrompt(input: Input) {
}
}
const projectedMessages = async () => {
const messages: SessionMessageInfo[] = []
let cursor: string | undefined
while (true) {
const page = await input.client.message.list(
cursor
? { sessionID: input.sessionID, limit: 200, cursor }
: { sessionID: input.sessionID, limit: 200, order: "desc" },
)
for (const message of page.data) {
if (message.id === messageID) return { found: true, messages: messages.toReversed() }
messages.push(message)
}
cursor = page.cursor.next ?? undefined
if (!cursor) return { found: false, messages: [] }
}
}
const reconcile = async () => {
const projected = await projectedMessages()
for (const message of projected.messages) {
if (message.type !== "assistant") continue
const timestamp = message.time.completed ?? message.time.created
let textOrdinal = 0
let reasoningOrdinal = 0
for (const item of message.content) {
if (item.type === "text") {
const ordinal = textOrdinal++
const key = contentKey(message.id, ordinal)
const rendered = renderedText.get(key) ?? ""
if (rendered === item.text || !item.text.startsWith(rendered)) continue
const text = item.text.slice(rendered.length)
writeText(
{
id: projectedPartID(message.id, `text-${ordinal}`),
sessionID: input.sessionID,
messageID: message.id,
type: "text",
text,
time: { start: message.time.created, end: timestamp },
},
timestamp,
)
renderedText.set(key, item.text)
continue
}
if (item.type === "reasoning") {
const ordinal = reasoningOrdinal++
if (!input.thinking) continue
const key = contentKey(message.id, ordinal)
const rendered = renderedReasoning.get(key) ?? ""
if (rendered === item.text || !item.text.startsWith(rendered)) continue
const text = item.text.slice(rendered.length)
const part = {
id: projectedPartID(message.id, `reasoning-${ordinal}`),
sessionID: input.sessionID,
messageID: message.id,
type: "reasoning",
text,
metadata: item.state,
time: { start: message.time.created, end: timestamp },
}
renderedReasoning.set(key, item.text)
writeReasoning(part, timestamp)
continue
}
const key = toolKey(message.id, item.id)
if (renderedTools.has(key) || item.state.status === "streaming" || item.state.status === "running") continue
const part: MiniToolPart = {
id: projectedPartID(message.id, `tool-${item.id}`),
sessionID: input.sessionID,
messageID: message.id,
type: "tool",
callID: item.id,
tool: item.name,
state:
item.state.status === "completed"
? {
status: "completed",
input: item.state.input,
output: toolOutputText(item.name, item.state.content),
title: item.name,
metadata: { structured: item.state.structured, content: item.state.content, result: item.state.result },
time: { start: item.time.ran ?? item.time.created, end: item.time.completed ?? timestamp },
}
: {
status: "error",
input: item.state.input,
error: item.state.error.message,
metadata: { structured: item.state.structured, content: item.state.content, result: item.state.result },
time: { start: item.time.ran ?? item.time.created, end: item.time.completed ?? timestamp },
},
}
renderedTools.add(key)
if (emit("tool_use", timestamp, { part })) continue
if (item.state.status === "completed") {
await input.renderTool(item)
continue
}
if (toolOutputText(item.name, item.state.content).trim()) {
await input.renderTool({ ...item, state: { ...item.state, status: "completed" } })
}
await input.renderToolError(item)
UI.error(item.state.error.message)
}
if (message.error && !emittedError) {
emittedError = true
process.exitCode = 1
if (!emit("error", timestamp, { error: message.error })) UI.error(message.error.message)
}
}
return projected.found
}
const interrupt = () => {
if (interrupted) process.exit(130)
interrupted = true
@@ -559,11 +700,27 @@ export async function runNonInteractivePrompt(input: Input) {
? globals.data.filter((form) => form.sessionID === GLOBAL_FORM_SESSION_ID).map(cancelForm)
: []),
])
await completed
if (input.compatibility === "v1") {
await completed
return
}
const waiting = input.client.session.wait({ sessionID: input.sessionID })
await Promise.race([waiting, completed.then(() => waiting)])
finalizing = true
controller.abort()
const found = await reconcile()
if (!found && !interrupted && !permissionRejected && !formCancelled && !emittedError) {
const error = prePromotionError ?? { type: "unknown", message: "Prompt was not promoted" }
emittedError = true
process.exitCode = 1
if (!emit("error", Date.now(), { error })) UI.error(error.message)
}
} finally {
process.off("SIGINT", interrupt)
controller.abort()
await stream.return?.(undefined).catch(() => {})
if (input.compatibility === "v1") await stream.return?.(undefined).catch(() => {})
else void stream.return?.(undefined).catch(() => {})
}
}
@@ -595,6 +752,14 @@ function toolKey(messageID: string, callID: string) {
return `${messageID}\u0000${callID}`
}
function contentKey(messageID: string, ordinal: number) {
return `${messageID}\u0000${ordinal}`
}
function projectedPartID(messageID: string, part: string) {
return `prt_${messageID.replace(/^msg_/, "")}_${part}`
}
function fallbackTool(event: {
id: string
created: number
packages/cli/test/run/noninteractive.test.ts
+81 -1
View File
@@ -1,5 +1,10 @@
import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"
import { OpenCode, type EventSubscribeOutput, type SessionMessageAssistantTool } from "@opencode-ai/client/promise"
import {
OpenCode,
type EventSubscribeOutput,
type SessionMessageAssistantTool,
type SessionMessageInfo,
} from "@opencode-ai/client/promise"
import { runNonInteractivePrompt } from "../../src/run/noninteractive"
type V2Event = EventSubscribeOutput
@@ -162,10 +167,13 @@ async function run(input: {
cancel?: (input: { sessionID: string; formID: string }) => Promise<void>
renderTool?: (part: SessionMessageAssistantTool) => Promise<void>
renderToolError?: (part: SessionMessageAssistantTool) => Promise<void>
messages?: (inputID: string) => SessionMessageInfo[]
wait?: () => Promise<void>
}) {
const sdk = OpenCode.make({ baseUrl: "https://opencode.test" })
const values: V2Event[] = [{ id: "evt_connected", type: "server.connected", data: {} }]
let wake: (() => void) | undefined
const wait = Promise.withResolvers<void>()
const stream = (async function* (): AsyncGenerator<V2Event, void, unknown> {
while (true) {
const value = values.shift()
@@ -175,6 +183,7 @@ async function run(input: {
})
continue
}
if (value.type.startsWith("session.execution.")) setTimeout(wait.resolve, 0)
yield value
}
})()
@@ -193,8 +202,19 @@ async function run(input: {
}) as never,
)
spyOn(sdk.form, "cancel").mockImplementation((request) => (input.cancel?.(request) ?? ok(undefined)) as never)
let promptID = "msg_prompt"
spyOn(sdk.session, "wait").mockImplementation(() => input.wait?.() ?? wait.promise)
spyOn(sdk.message, "list").mockImplementation(() =>
ok({
data: input.messages?.(promptID) ?? [
{ id: promptID, type: "user", text: "hello", time: { created: 1 } },
],
cursor: {},
}),
)
spyOn(sdk.session, "prompt").mockImplementation((request) => {
const messageID = request.id ?? "msg_prompt"
promptID = messageID
values.push(...input.turn(messageID))
wake?.()
wake = undefined
@@ -244,6 +264,63 @@ afterEach(() => {
})
describe("runNonInteractivePrompt", () => {
test("uses session.wait then reconciles projected output without a terminal event", async () => {
const idle = Promise.withResolvers<void>()
let done = false
const task = capture({
format: "json",
turn: (messageID) => [prompted(messageID)],
wait: () => idle.promise,
messages: (messageID) => [
{
id: "msg_assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", id: "test-model" },
content: [{ type: "text", text: "projected answer" }],
finish: "stop",
time: { created: 2, completed: 3 },
},
{ id: messageID, type: "user", text: "hello", time: { created: 1 } },
],
}).then((output) => {
done = true
return output
})
await Bun.sleep(0)
await Bun.sleep(0)
expect(done).toBe(false)
idle.resolve()
const output = await task
expect(
output.stdout
.split("\n")
.filter(Boolean)
.map((line) => JSON.parse(line)),
).toEqual([expect.objectContaining({ type: "text", part: expect.objectContaining({ text: "projected answer" }) })])
})
test("reports an observed execution failure before prompt promotion", async () => {
const output = await capture({
format: "json",
turn: () => [executionFailed("instructions unavailable")],
messages: () => [],
})
expect(
output.stdout
.split("\n")
.filter(Boolean)
.map((line) => JSON.parse(line)),
).toEqual([
expect.objectContaining({
type: "error",
error: { type: "provider.transport", message: "instructions unavailable" },
}),
])
})
test("cancels session and global form blockers and exits on pre-promotion interrupt", async () => {
const sdk = await run({
pendingForms: [form("frm_pending", "ses_1"), form("frm_pending_global", "global")],
@@ -307,6 +384,9 @@ describe("runNonInteractivePrompt", () => {
}),
])
expect(output.stderr).toBe("")
const sdk = await run({ compatibility: "v1", turn: (messageID) => [prompted(messageID), settled()] })
expect(sdk.session.wait).not.toHaveBeenCalled()
expect(sdk.message.list).not.toHaveBeenCalled()
})
test("V1 default output flushes step_start before an unrelated execution failure", async () => {
packages/tui/src/config/v1/keybind.ts
+1 -1
View File
@@ -96,7 +96,7 @@ export const Definitions = {
session_interrupt: keybind("escape", "Interrupt current session"),
session_background: keybind("ctrl+b", "Background blocking session tools"),
session_compact: keybind("<leader>c", "Compact the session"),
session_queued_prompts: keybind("<leader>q", "Manage queued prompts"),
session_queued_prompts: keybind("<leader>q", "View pending work"),
session_child_first: keybind("down,<leader>down", "Toggle subagent picker"),
session_child_cycle: keybind("right", "Go to next child session"),
session_child_cycle_reverse: keybind("left", "Go to previous child session"),
packages/tui/src/mini/footer.command.tsx
+6 -24
View File
@@ -465,8 +465,8 @@ export function RunCommandMenuBody(props: {
{
action: "queued" as const,
category: "Agent",
display: "Manage queued prompts",
footer: `${props.queued().length} queued`,
display: "View pending work",
footer: `${props.queued().length} pending`,
keywords: props
.queued()
.map((item) => item.prompt.text)
@@ -673,15 +673,13 @@ export function RunQueuedPromptSelectBody(props: {
theme: Accessor<RunFooterTheme>
prompts: Accessor<FooterQueuedPrompt[]>
onClose: () => void
onEdit: (prompt: FooterQueuedPrompt) => void | Promise<void>
onDelete: (prompt: FooterQueuedPrompt) => void | Promise<void>
onRows?: (rows: number) => void
}) {
const entries = createMemo<QueuedEntry[]>(() =>
props.prompts().map((prompt) => ({
category: "",
display: prompt.prompt.text.replaceAll("\n", " "),
footer: "queued · ctrl+e edit · ctrl+d remove",
footer: prompt.delivery,
keywords: prompt.prompt.text,
prompt,
})),
@@ -690,29 +688,13 @@ export function RunQueuedPromptSelectBody(props: {
entries,
limit: SUBAGENT_LIST_ROWS,
onClose: props.onClose,
onSelect: (item) => props.onEdit(item.prompt),
onSelect: props.onClose,
onRows: props.onRows,
onKey: (event, item) => {
const ctrl = event.ctrl && !event.meta && !event.shift && !event.super
if (item && (event.name === "delete" || (ctrl && event.name === "d"))) {
event.preventDefault()
props.onDelete(item.prompt)
return true
}
if (item && ctrl && event.name === "e") {
event.preventDefault()
props.onEdit(item.prompt)
return true
}
return false
},
})
return (
<PanelShell
title="Queued prompts"
title="Pending work"
query={controller.query()}
count={controller.items().length}
total={entries().length}
@@ -730,7 +712,7 @@ export function RunQueuedPromptSelectBody(props: {
offset={controller.menu.offset}
rows={controller.menu.rows}
limit={SUBAGENT_LIST_ROWS}
empty="No queued prompts"
empty="No pending work"
border={false}
paddingLeft={PANEL_PAD}
paddingRight={PANEL_PAD}
packages/tui/src/mini/footer.ts
-23
View File
@@ -115,10 +115,6 @@ function createEmptySubagentState(): FooterSubagentState {
}
function eventPatch(next: FooterEvent): FooterPatch | undefined {
if (next.type === "queue") {
return { queue: next.queue }
}
if (next.type === "first") {
return { first: next.first }
}
@@ -131,7 +127,6 @@ function eventPatch(next: FooterEvent): FooterPatch | undefined {
return {
phase: "running",
status: "sending prompt",
queue: next.queue,
interrupt: 0,
exit: 0,
}
@@ -141,7 +136,6 @@ function eventPatch(next: FooterEvent): FooterPatch | undefined {
return {
phase: "idle",
status: "",
queue: next.queue,
}
}
@@ -156,7 +150,6 @@ export class RunFooter implements FooterApi {
private closed = false
private destroyed = false
private prompts = new Set<(input: RunPrompt) => void>()
private queuedRemoves = new Set<(messageID: string) => boolean | Promise<boolean>>()
private closes = new Set<() => void>()
// Microtask-coalesced commit queue. Flushed on next microtask or on close/destroy.
private queue: StreamCommit[] = []
@@ -226,7 +219,6 @@ export class RunFooter implements FooterApi {
const [state, setState] = createSignal<FooterState>({
phase: "idle",
status: "",
queue: 0,
model: options.modelLabel,
usage: "",
first: options.first,
@@ -328,7 +320,6 @@ export class RunFooter implements FooterApi {
onStatus: footer.setStatus,
onSubagentSelect: options.onSubagentSelect,
onSubagentInterrupt: options.onSubagentInterrupt,
onQueuedRemove: footer.handleQueuedRemove,
})
},
}),
@@ -355,13 +346,6 @@ export class RunFooter implements FooterApi {
}
}
public onQueuedRemove(fn: (messageID: string) => boolean | Promise<boolean>): () => void {
this.queuedRemoves.add(fn)
return () => {
this.queuedRemoves.delete(fn)
}
}
public onClose(fn: () => void): () => void {
if (this.isClosed) {
fn()
@@ -487,7 +471,6 @@ export class RunFooter implements FooterApi {
const state = {
phase: next.phase ?? prev.phase,
status: typeof next.status === "string" ? next.status : prev.status,
queue: typeof next.queue === "number" ? Math.max(0, next.queue) : prev.queue,
model: typeof next.model === "string" ? next.model : prev.model,
usage: typeof next.usage === "string" ? next.usage : prev.usage,
first: typeof next.first === "boolean" ? next.first : prev.first,
@@ -665,11 +648,6 @@ export class RunFooter implements FooterApi {
this.requestExitHandler = fn
}
private handleQueuedRemove = async (messageID: string): Promise<boolean> => {
const fn = [...this.queuedRemoves][0]
return fn ? await fn(messageID) : false
}
private handleInputClear = (): void => {
this.clearInterruptTimer()
this.clearExitTimer()
@@ -1080,7 +1058,6 @@ export class RunFooter implements FooterApi {
for (const timeout of this.themeRefreshTimeouts) clearTimeout(timeout)
this.themeRefreshTimeouts.length = 0
this.prompts.clear()
this.queuedRemoves.clear()
this.closes.clear()
this.scrollback.destroy()
for (const theme of [...this.themes]) this.destroyTheme(theme)
packages/tui/src/mini/footer.view.tsx
+2 -10
View File
@@ -102,7 +102,6 @@ type RunFooterViewProps = {
onStatus: (text: string) => void
onSubagentSelect?: (sessionID: string | undefined) => void
onSubagentInterrupt?: (sessionID: string) => void
onQueuedRemove: (messageID: string) => Promise<boolean>
}
export function RunFooterView(props: RunFooterViewProps) {
@@ -179,7 +178,6 @@ export function RunFooterView(props: RunFooterViewProps) {
const busy = createMemo(() => props.state().phase === "running")
const armed = createMemo(() => props.state().interrupt > 0)
const exiting = createMemo(() => props.state().exit > 0)
const queue = createMemo(() => props.state().queue)
const usage = createMemo(() => props.state().usage)
const interruptLabel = createMemo(() => {
if (!interrupt()) {
@@ -414,7 +412,7 @@ export function RunFooterView(props: RunFooterViewProps) {
items.push({ kind: "background", key: backgroundShortcut(), label: "background" })
}
if (queuedPrompts().length > 0 && queuedShortcut()) {
items.push({ kind: "queued", key: queuedShortcut(), label: `${queue()} queued` })
items.push({ kind: "queued", key: queuedShortcut(), label: `${queuedPrompts().length} pending` })
}
if (activeTabs().length > 0 && subagentShortcut()) {
items.push({ kind: "subagents", key: subagentShortcut(), label: "subagents" })
@@ -495,7 +493,7 @@ export function RunFooterView(props: RunFooterViewProps) {
commands: [
{
id: "session.queued_prompts",
title: "Manage queued prompts",
title: "View pending work",
group: "Session",
run: openQueuedMenu,
},
@@ -656,12 +654,6 @@ export function RunFooterView(props: RunFooterViewProps) {
theme={theme}
prompts={queuedPrompts}
onClose={closePanel}
onDelete={(item) => void props.onQueuedRemove(item.messageID)}
onEdit={async (item) => {
if (!(await props.onQueuedRemove(item.messageID))) return
closePanel()
queueMicrotask(() => composer.replacePrompt(item.prompt))
}}
onRows={setSubagentMenuRows}
/>
</Match>
packages/tui/src/mini/runtime.queue.ts
+47 -86
View File
@@ -1,8 +1,8 @@
// Serial prompt queue for direct interactive mode.
//
// Prompts arrive from the footer (user types and hits enter) and queue up
// here. The queue drains one turn at a time; ordinary prompts waiting behind
// an active ordinary turn are exposed for edit/removal until they begin.
// Prompts arrive from the footer (user types and hits enter) and local
// operations drain one at a time. Ordinary prompts submitted during an active
// ordinary turn are admitted immediately to the server's durable queue.
//
// The queue also handles /exit, /quit, and /new commands, empty-prompt rejection,
// and tracks per-turn wall-clock duration for the footer status line.
@@ -11,84 +11,54 @@
import { SessionMessage } from "@opencode-ai/schema/session-message"
import { Locale } from "../util/locale"
import { isExitCommand, isNewCommand } from "./prompt.shared"
import type { FooterApi, FooterEvent, FooterQueuedPrompt, RunPrompt } from "./types"
import type { FooterApi, FooterEvent, RunPrompt } from "./types"
type Trace = {
write(type: string, data?: unknown): void
}
type Deferred<T = void> = {
promise: Promise<T>
resolve: (value: T | PromiseLike<T>) => void
reject: (error?: unknown) => void
}
export type QueueInput = {
footer: FooterApi
initialInput?: string
trace?: Trace
onSend?: (prompt: RunPrompt) => void
onSend?: (prompt: RunPrompt, delivery: "steer" | "queue") => void
onAdmissionError?: (prompt: RunPrompt, error: unknown) => void | Promise<void>
onNewSession?: () => void | Promise<void>
run: (prompt: RunPrompt, signal: AbortSignal) => Promise<void>
admit: (prompt: RunPrompt, signal: AbortSignal) => Promise<void>
settle: () => Promise<void>
run: (prompt: RunPrompt, signal: AbortSignal, admitted: () => void) => Promise<void>
}
type State = {
queue: RunPrompt[]
queued: FooterQueuedPrompt[]
active?: RunPrompt
admission?: Promise<void>
ctrl?: AbortController
closed: boolean
}
function defer<T = void>(): Deferred<T> {
let resolve!: (value: T | PromiseLike<T>) => void
let reject!: (error?: unknown) => void
const promise = new Promise<T>((next, fail) => {
resolve = next
reject = fail
})
return { promise, resolve, reject }
}
// Runs the prompt queue until the footer closes.
//
// Subscribes to footer prompt events and drains operations through input.run().
// Ordinary prompts submitted during an ordinary active turn remain local and
// are exposed by the footer for edit/removal until their turn begins.
// Ordinary prompts submitted during an ordinary active turn are admitted as
// durable queued work instead of remaining editable process-local state.
export async function runPromptQueue(input: QueueInput): Promise<void> {
const stop = defer<{ type: "closed" }>()
const done = defer()
const stop = Promise.withResolvers<{ type: "closed" }>()
const done = Promise.withResolvers<void>()
const state: State = {
queue: [],
queued: [],
closed: input.footer.isClosed,
}
let draining: Promise<void> | undefined
let admissions = Promise.resolve()
let admissionVersion = 0
const admissionController = new AbortController()
const emit = (next: FooterEvent, row: Record<string, unknown>) => {
input.trace?.write("ui.patch", row)
input.footer.event(next)
}
const syncQueue = () => {
const queue = state.queue.length
emit({ type: "queue", queue }, { queue })
emit(
{
type: "queued.prompts",
prompts: [...state.queued],
},
{ queued: state.queued.length },
)
}
const removeLocalQueued = (queued: FooterQueuedPrompt) => {
if (!state.queued.includes(queued)) return
state.queued = state.queued.filter((item) => item !== queued)
syncQueue()
}
const finish = () => {
if (!state.closed || draining) {
return
@@ -104,8 +74,8 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
state.closed = true
state.queue.length = 0
state.queued.length = 0
state.ctrl?.abort()
admissionController.abort()
stop.resolve({ type: "closed" })
finish()
}
@@ -123,11 +93,7 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
continue
}
const queued = state.queued.find((item) => item.prompt === prompt)
if (queued) removeLocalQueued(queued)
if (prompt.mode !== "shell" && isNewCommand(prompt.text)) {
syncQueue()
if (!input.onNewSession) {
emit(
{
@@ -149,13 +115,11 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
patch: {
phase: "running",
status: "starting new session",
queue: state.queue.length,
},
},
{
phase: "running",
status: "starting new session",
queue: state.queue.length,
},
)
await input.onNewSession()
@@ -167,24 +131,23 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
? prompt
: {
...prompt,
messageID: prompt.messageID ?? queued?.messageID ?? SessionMessage.ID.create(),
messageID: prompt.messageID ?? SessionMessage.ID.create(),
}
state.active = sent
emit(
{
type: "turn.send",
queue: state.queue.length,
},
{ type: "turn.send" },
{
phase: "running",
status: "sending prompt",
queue: state.queue.length,
},
)
const start = Date.now()
const ctrl = new AbortController()
const admission = Promise.withResolvers<void>()
const version = admissionVersion
state.ctrl = ctrl
state.admission = admission.promise
try {
await input.footer.idle()
@@ -203,13 +166,13 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
input.trace?.write("ui.commit", commit)
input.footer.append(commit)
}
input.onSend?.(sent)
input.onSend?.(sent, "steer")
if (state.closed) {
break
}
const task = input.run(sent, ctrl.signal).then(
const task = input.run(sent, ctrl.signal, admission.resolve).then(
() => ({ type: "done" as const }),
(error) => ({ type: "error" as const, error }),
)
@@ -223,10 +186,21 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
if (next.type === "error") {
throw next.error
}
if (sent.mode !== "shell" && admissionVersion !== version) {
do {
const current = admissionVersion
await admissions
if (state.closed) break
await input.settle()
if (current === admissionVersion) break
} while (!state.closed)
}
} finally {
admission.resolve()
if (state.ctrl === ctrl) {
state.ctrl = undefined
}
if (state.admission === admission.promise) state.admission = undefined
if (sent.mode !== "shell") {
const duration = Locale.duration(Math.max(0, Date.now() - start))
@@ -249,14 +223,10 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
} finally {
draining = undefined
emit(
{
type: "turn.idle",
queue: state.queue.length,
},
{ type: "turn.idle" },
{
phase: "idle",
status: "",
queue: state.queue.length,
},
)
}
@@ -279,23 +249,22 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
if (
active &&
active.mode !== "shell" &&
!active.command &&
prompt.mode !== "shell" &&
!prompt.command &&
prompt.command?.source !== "skill" &&
!isNewCommand(prompt.text)
) {
const queued: FooterQueuedPrompt = {
messageID: SessionMessage.ID.create(),
prompt,
}
state.queued = [...state.queued, queued]
state.queue.push(prompt)
syncQueue()
const sent = { ...prompt, messageID: SessionMessage.ID.create() }
const admission = state.admission
admissionVersion += 1
input.onSend?.(sent, "queue")
admissions = admissions
.then(() => admission)
.then(() => input.admit(sent, admissionController.signal))
.catch((error) => (state.closed ? undefined : input.onAdmissionError?.(sent, error)))
return
}
state.queue.push(prompt)
syncQueue()
if (prompt.mode !== "shell" && isNewCommand(prompt.text)) {
drain()
return
@@ -319,14 +288,6 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
const offClose = input.footer.onClose(() => {
close()
})
const offRemoveQueued = input.footer.onQueuedRemove((messageID) => {
const queued = state.queued.find((item) => item.messageID === messageID)
if (!queued) return false
state.queue = state.queue.filter((prompt) => prompt !== queued.prompt)
removeLocalQueued(queued)
return true
})
try {
if (state.closed) {
return
@@ -341,8 +302,8 @@ export async function runPromptQueue(input: QueueInput): Promise<void> {
} finally {
offPrompt()
offClose()
offRemoveQueued()
close()
await draining?.catch(() => {})
await admissions
}
}
packages/tui/src/mini/runtime.ts
+51 -28
View File
@@ -780,6 +780,22 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
}, RESIZE_DELAY)
})
const renderPromptError = async (prompt: RunPrompt, error: unknown, signal?: AbortSignal) => {
if (signal?.aborted || footer.isClosed) return
const text =
(await state.stream?.then((item) => item.mod).catch(() => undefined))?.formatUnknownError(error) ??
(error instanceof Error ? error.message : String(error))
const commit = {
kind: "error",
text,
phase: "start",
source: "system",
messageID: prompt.messageID,
} as const
rememberLocal(commit)
footer.append(commit)
}
const runQueue = async () => {
await firstPaint
if (footer.isClosed) return
@@ -798,10 +814,10 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
footer,
initialInput: input.initialInput,
trace: log,
onSend: (prompt) => {
onSend: (prompt, delivery) => {
state.shown = true
state.history.push(prompt)
if (prompt.mode !== "shell") {
if (prompt.mode !== "shell" && delivery === "steer") {
rememberLocal({
kind: "user",
text: prompt.text,
@@ -811,6 +827,24 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
})
}
},
admit: async (prompt, signal) => {
await state.switching?.catch(() => {})
const next = await ensureStream()
await next.handle.queuePromptTurn({
agent: state.agent,
model: state.model,
variant: state.activeVariant,
prompt,
files: input.files,
includeFiles: false,
signal,
})
},
onAdmissionError: renderPromptError,
settle: async () => {
const next = await ensureStream()
await next.handle.waitForIdle()
},
onNewSession: createSession
? async () => {
try {
@@ -856,6 +890,7 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
},
})
footer.event({ type: "stream.view", view: { type: "prompt" } })
footer.event({ type: "queued.prompts", prompts: [] })
footer.event({
type: "stream.patch",
patch: {
@@ -891,7 +926,7 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
}
}
: undefined,
run: async (prompt, signal) => {
run: async (prompt, signal, admitted) => {
if (state.demo && (await state.demo.prompt(prompt, signal))) {
return
}
@@ -900,15 +935,18 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
try {
const next = await ensureStream()
await next.handle.runPromptTurn({
agent: state.agent,
model: state.model,
variant: state.activeVariant,
prompt,
files: input.files,
includeFiles,
signal,
})
await next.handle.runPromptTurn(
{
agent: state.agent,
model: state.model,
variant: state.activeVariant,
prompt,
files: input.files,
includeFiles,
signal,
},
admitted,
)
if (prompt.messageID) {
state.localRows = state.localRows.filter(
(row) => row.commit.kind !== "user" || row.commit.messageID !== prompt.messageID,
@@ -918,22 +956,7 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
// pending for the next prompt-shaped turn.
if (prompt.mode !== "shell" && prompt.command?.source !== "skill") includeFiles = false
} catch (error) {
if (signal.aborted || footer.isClosed) {
return
}
const text =
(await state.stream?.then((item) => item.mod).catch(() => undefined))?.formatUnknownError(error) ??
(error instanceof Error ? error.message : String(error))
const commit = {
kind: "error",
text,
phase: "start",
source: "system",
messageID: prompt.messageID,
} as const
rememberLocal(commit)
footer.append(commit)
await renderPromptError(prompt, error, signal)
}
},
})
packages/tui/src/mini/stream-v2.transport.ts
+201 -90
View File
@@ -6,6 +6,7 @@ import type {
PermissionV2Request,
SessionMessageAssistantTool,
SessionMessageInfo,
SessionPendingInfo,
} from "@opencode-ai/client/promise"
import { Event } from "@opencode-ai/schema/event"
import { SessionMessage } from "@opencode-ai/schema/session-message"
@@ -20,6 +21,7 @@ import type {
LocalReplayRow,
MiniPermissionRequest,
MiniFormRequest,
FooterQueuedPrompt,
RunFilePart,
RunInput,
RunPrompt,
@@ -64,7 +66,9 @@ export type SessionResizeReplayInput = {
}
export type SessionTransport = {
runPromptTurn(input: SessionTurnInput): Promise<void>
runPromptTurn(input: SessionTurnInput, admitted?: () => void): Promise<void>
queuePromptTurn(input: SessionTurnInput): Promise<void>
waitForIdle(): Promise<void>
interruptActiveTurn(): Promise<void>
selectSubagent(sessionID: string | undefined): void
replayOnResize(input: SessionResizeReplayInput): Promise<boolean>
@@ -74,11 +78,12 @@ export type SessionTransport = {
type Wait = {
messageID: string
failureMessageID: string
promoted: boolean
promotionObserved: boolean
interrupted: boolean
failureRendered: boolean
resolve: () => void
reject: (error: unknown) => void
terminalError?: Error
}
// One active session.shell call. The HTTP response is the completion signal;
@@ -134,6 +139,8 @@ type State = {
rootActive: boolean
buffered?: ReplayBuffer
errors: Set<string>
pending: Map<string, FooterQueuedPrompt>
admitted: Set<string>
}
const money = new Intl.NumberFormat("en-US", { style: "currency", currency: "USD" })
@@ -167,6 +174,16 @@ function errorMessage(error: { message?: string; _tag?: string }) {
return error.message || error._tag || "Session execution failed"
}
function pendingPrompt(item: SessionPendingInfo): FooterQueuedPrompt | undefined {
if (item.type !== "user") return undefined
return {
messageID: item.id,
prompt: { messageID: item.id, text: item.data.text, parts: [] },
delivery: item.delivery,
admittedSeq: item.admittedSeq,
}
}
function wait(delay: number, signal: AbortSignal) {
return new Promise<void>((resolve) => {
const timer = setTimeout(done, delay)
@@ -369,6 +386,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
let sdk = input.sdk
let generation = 0
let activeAttempt: Attempt | undefined
let settlementClient: OpenCodeClient | undefined
input.signal?.addEventListener("abort", () => controller.abort(), { once: true })
const state: State = {
permissions: [],
@@ -389,6 +407,8 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
initial: true,
rootActive: false,
errors: new Set(),
pending: new Map(),
admitted: new Set(),
}
let readyResolve!: () => void
let readyReject!: (error: unknown) => void
@@ -443,6 +463,30 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
)
}
const syncPending = () => {
const prompts = [...state.pending.values()].toSorted((left, right) => left.admittedSeq - right.admittedSeq)
input.trace?.write("ui.patch", { pending: prompts.length })
input.footer.event({ type: "queued.prompts", prompts })
}
const mergePending = (item: SessionPendingInfo) => {
const prompt = pendingPrompt(item)
if (!prompt || state.messageIDs.has(prompt.messageID)) return
state.admitted.add(prompt.messageID)
state.pending.set(prompt.messageID, prompt)
syncPending()
}
const promoteWait = (wait: Wait, observed: boolean, messageID = wait.messageID) => {
const transition = messageID !== wait.failureMessageID || (observed ? !wait.promotionObserved : !wait.promoted)
wait.promoted = true
if (observed) wait.promotionObserved = true
if (!transition) return
wait.failureMessageID = messageID
wait.failureRendered = false
wait.terminalError = undefined
}
const syncBlockers = () => {
if (state.closed || controller.signal.aborted || input.footer.isClosed) return
const descendant = subagents.snapshot()
@@ -528,15 +572,19 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
const renderMessage = (message: SessionMessageInfo, render: boolean, reuseVisibleWait: boolean) => {
if (message.type === "user") {
const waiting = state.wait?.messageID === message.id
if (waiting && state.wait) state.wait.promoted = true
if (!render || state.messageIDs.has(message.id)) return
const admitted = state.admitted.delete(message.id)
if (state.wait && (admitted || (waiting && state.wait.failureMessageID === message.id)))
promoteWait(state.wait, false, message.id)
if (state.pending.delete(message.id)) syncPending()
if (state.messageIDs.has(message.id)) return
state.messageIDs.add(message.id)
if (!render) return
if (reuseVisibleWait && waiting) return
write([{ kind: "user", source: "system", text: message.text, phase: "start", messageID: message.id }])
return
}
if (message.type === "skill") {
if (state.wait?.messageID === message.id) state.wait.promoted = true
if (state.wait?.messageID === message.id) promoteWait(state.wait, false)
if (!render || state.skillMessages.has(message.id)) {
state.skillMessages.add(message.id)
return
@@ -616,8 +664,10 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
renderTool(message.id, item, render)
}
if (render && message.error && !state.errors.has(message.id)) {
if (message.error && !state.errors.has(message.id)) {
state.errors.add(message.id)
if (!render) return
if (state.wait) state.wait.failureRendered = true
write([
{
kind: "error",
@@ -630,6 +680,22 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
}
const projectedMessages = async (client: OpenCodeClient, signal: AbortSignal) =>
(
await client.message.list(
{ sessionID: input.sessionID, limit: input.replayLimit ?? 200, order: "desc" },
{ signal },
)
).data.toReversed()
const settleSession = async (client: OpenCodeClient) => {
await client.session.wait({ sessionID: input.sessionID }, { signal: controller.signal })
for (const message of await projectedMessages(client, controller.signal)) renderMessage(message, true, true)
state.rootActive = false
write([], { phase: "idle", status: blockerStatus(state.view) })
await input.footer.idle()
}
const resolvePermissionSources = async (
client: OpenCodeClient,
permissions: PermissionV2Request[],
@@ -672,8 +738,9 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
) => {
const client = attempt.client
const options = { signal: attempt.signal }
const [messages, permissions, forms, globals, active] = await Promise.all([
client.message.list({ sessionID: input.sessionID, limit: input.replayLimit ?? 200, order: "desc" }, options),
const [projected, pending, permissions, forms, globals, active] = await Promise.all([
projectedMessages(client, attempt.signal),
client.session.pending.list({ sessionID: input.sessionID }, options),
client.permission.list({ sessionID: input.sessionID }, options),
client.form.list({ sessionID: input.sessionID }, options),
input.location
@@ -687,7 +754,11 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
client.session.active(options),
])
if (!current(attempt)) return
const projected = structuredClone(messages.data).toReversed() as SessionMessageInfo[]
state.pending = new Map(pending.flatMap((item) => {
const prompt = pendingPrompt(item)
return prompt ? [[prompt.messageID, prompt] as const] : []
}))
syncPending()
state.permissions = permissions
pruneToolSources()
for (const message of projected) renderMessage(message, next.render, next.reuseVisibleWait)
@@ -714,11 +785,6 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
})
if (!state.rootActive) await input.footer.idle()
if (!current(attempt)) return
if (!state.rootActive && state.wait && (state.wait.promoted || state.wait.interrupted)) {
const current = state.wait
state.wait = undefined
current.resolve()
}
}
const apply = (attempt: Attempt, event: RunV2Event) => {
@@ -756,9 +822,37 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
input.trace?.write("recv.event", event)
subagents.main(client, event, attempt.signal)
if (event.type === "session.input.admitted") {
if (event.data.input.type !== "user") return
mergePending({
admittedSeq: event.durable.seq,
id: event.data.inputID,
sessionID: event.data.sessionID,
timeCreated: event.created,
...event.data.input,
})
return
}
if (event.type === "session.input.promoted") {
if (state.wait?.messageID === event.data.inputID) state.wait.promoted = true
state.messageIDs.add(event.data.inputID)
const waiting = state.wait?.messageID === event.data.inputID
if (state.wait) promoteWait(state.wait, true, event.data.inputID)
state.admitted.delete(event.data.inputID)
const pending = state.pending.get(event.data.inputID)
state.pending.delete(event.data.inputID)
syncPending()
const visible = state.messageIDs.has(event.data.inputID)
if (waiting || pending) state.messageIDs.add(event.data.inputID)
if (!waiting && pending && !visible) {
write([
{
kind: "user",
source: "system",
text: pending.prompt.text,
phase: "start",
messageID: event.data.inputID,
},
])
}
write([], { phase: "running", status: "waiting for assistant" })
return
}
@@ -768,7 +862,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
if (event.type === "session.skill.activated") {
const messageID = messageIDFromEvent(event.id)
if (state.wait?.messageID === messageID) state.wait.promoted = true
if (state.wait?.messageID === messageID) promoteWait(state.wait, true)
if (state.skillMessages.has(messageID)) return
state.skillMessages.add(messageID)
write([skillCommit(messageID, event.data.name)])
@@ -1047,25 +1141,17 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
state.rootActive = false
write([], { phase: "idle", status: "" })
const current = state.wait
if (!current || (!current.promoted && !current.interrupted)) return
state.wait = undefined
if (!current) return
if (current.interrupted && event.type === "session.execution.interrupted" && event.data.reason === "user") {
current.resolve()
return
}
if (event.type === "session.execution.failed") {
if (current.failureRendered) {
current.resolve()
return
}
current.reject(new Error(errorMessage(event.data.error)))
if (!current.failureRendered) current.terminalError = new Error(errorMessage(event.data.error))
return
}
if (event.type === "session.execution.interrupted") {
current.reject(new Error(`Session interrupted: ${event.data.reason}`))
return
current.terminalError = new Error(`Session interrupted: ${event.data.reason}`)
}
current.resolve()
}
}
@@ -1248,27 +1334,22 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
}
// Shared settlement scaffolding for prompt-shaped turns: registers the wait,
// wires interruption, sends, then blocks until the live settled event (or a
// hydration pass over an idle session) resolves it.
// Prompt-shaped turns complete through the process-local idle fence. Live
// lifecycle events remain presentation and best-effort outcome metadata.
const runTurnWait = async (
next: SessionTurnInput,
messageID: string,
turn: { promoted?: boolean; send: () => Promise<unknown> },
client: OpenCodeClient,
send: () => Promise<SessionPendingInfo | void>,
onAdmitted?: () => void,
) => {
let resolve!: () => void
let reject!: (error: unknown) => void
const done = new Promise<void>((ok, fail) => {
resolve = ok
reject = fail
})
const active: Wait = {
messageID,
promoted: turn.promoted === true,
failureMessageID: messageID,
promoted: false,
promotionObserved: false,
interrupted: false,
failureRendered: false,
resolve,
reject,
}
state.wait = active
const interrupt = () => {
@@ -1277,14 +1358,26 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
next.signal?.addEventListener("abort", interrupt, { once: true })
try {
await turn.send()
await done
const admitted = await send()
if (admitted) mergePending(admitted)
onAdmitted?.()
await settleSession(client)
if (active.terminalError && !active.failureRendered)
write([
{
kind: "error",
source: "system",
text: active.terminalError.message,
phase: "start",
messageID: active.failureMessageID,
},
])
} catch (error) {
if (state.wait === active) state.wait = undefined
if (next.signal?.aborted) return
throw error
} finally {
next.signal?.removeEventListener("abort", interrupt)
if (state.wait === active) state.wait = undefined
}
}
@@ -1362,6 +1455,46 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
let queuedResizeReplay: SessionResizeReplayInput | undefined
let closing: Promise<void> | undefined
const admitPrompt = async (next: SessionTurnInput, client: OpenCodeClient, delivery: "steer" | "queue") => {
const messageID = next.prompt.messageID
if (!messageID) throw new Error("Prompt message ID is required")
const command = next.prompt.command
const attachments = await prepareAttachments(next, command ? "command" : "prompt", input.readTextFile)
const agents = promptAgents(next)
if (!command) {
input.trace?.write("send.prompt", { sessionID: input.sessionID, messageID, delivery })
return client.session.prompt(
{
sessionID: input.sessionID,
id: messageID,
text: [next.prompt.text, ...attachments.text].join("\n\n"),
files: attachments.files.length ? attachments.files : undefined,
agents: agents.length ? agents : undefined,
delivery,
},
{ signal: next.signal },
)
}
const selected = await resolveSelectedModel(input, client, next)
if (next.variant && !selected) throw new Error("Cannot select a variant before selecting a model")
input.trace?.write("send.command", { sessionID: input.sessionID, messageID, command: command.name, delivery })
return client.session.command(
{
sessionID: input.sessionID,
id: messageID,
command: command.name,
arguments: command.arguments,
agent: next.agent,
model: selected,
files: attachments.files.length ? attachments.files : undefined,
agents: agents.length ? agents : undefined,
delivery,
},
{ signal: next.signal },
)
}
const replayOnResize = (next: SessionResizeReplayInput) => {
queuedResizeReplay = next
if (resizeReplay) return resizeReplay
@@ -1388,7 +1521,20 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
return {
async runPromptTurn(next) {
async queuePromptTurn(next) {
if (next.prompt.mode === "shell" || next.prompt.command?.source === "skill")
throw new Error("This prompt cannot be queued")
if (!state.connected) throw new Error("Event stream is reconnecting")
const client = sdk
mergePending(await admitPrompt(next, client, "queue"))
settlementClient = client
},
async waitForIdle() {
const client = settlementClient ?? sdk
await settleSession(client)
if (settlementClient === client) settlementClient = undefined
},
async runPromptTurn(next, admitted) {
if (next.prompt.mode === "shell") {
await runShellTurn(next)
return
@@ -1402,40 +1548,21 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
const command = next.prompt.command
if (command?.source === "skill") {
input.trace?.write("send.skill", { sessionID: input.sessionID, messageID, skill: command.name })
await runTurnWait(next, messageID, {
send: () =>
await runTurnWait(
next,
messageID,
client,
() =>
client.session.skill(
{ sessionID: input.sessionID, id: messageID, skill: command.name },
{ signal: next.signal },
),
})
admitted,
)
return
}
if (command) {
const selected = await resolveSelectedModel(input, client, next)
if (next.variant && !selected) throw new Error("Cannot select a variant before selecting a model")
// Agent and model ride the command payload; the server switches only
// when the command itself does not pin them.
const attachments = await prepareAttachments(next, "command")
const agents = promptAgents(next)
input.trace?.write("send.command", { sessionID: input.sessionID, messageID, command: command.name })
await runTurnWait(next, messageID, {
send: () =>
client.session.command(
{
sessionID: input.sessionID,
id: messageID,
command: command.name,
arguments: command.arguments,
agent: next.agent,
model: selected,
files: attachments.files.length ? attachments.files : undefined,
agents: agents.length ? agents : undefined,
delivery: "steer",
},
{ signal: next.signal },
),
})
await runTurnWait(next, messageID, client, () => admitPrompt(next, client, "steer"), admitted)
return
}
@@ -1447,23 +1574,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
if (selected)
await client.session.switchModel({ sessionID: input.sessionID, model: selected }, { signal: next.signal })
const attachments = await prepareAttachments(next, "prompt", input.readTextFile)
const agents = promptAgents(next)
input.trace?.write("send.prompt", { sessionID: input.sessionID, messageID })
await runTurnWait(next, messageID, {
send: () =>
client.session.prompt(
{
sessionID: input.sessionID,
id: messageID,
text: [next.prompt.text, ...attachments.text].join("\n\n"),
files: attachments.files.length ? attachments.files : undefined,
agents: agents.length ? agents : undefined,
delivery: "steer",
},
{ signal: next.signal },
),
})
await runTurnWait(next, messageID, client, () => admitPrompt(next, client, "steer"), admitted)
},
async interruptActiveTurn() {
// A running shell holds no drain, so session.interrupt cannot reach it;
packages/tui/src/mini/types.ts
+4 -14
View File
@@ -84,6 +84,8 @@ export type RunPrompt = {
export type FooterQueuedPrompt = {
messageID: string
prompt: RunPrompt
delivery: "steer" | "queue"
admittedSeq: number
}
export type RunAgent = {
@@ -161,7 +163,6 @@ export type FooterPhase = "idle" | "running"
export type FooterState = {
phase: FooterPhase
status: string
queue: number
model: string
usage: string
first: boolean
@@ -338,10 +339,6 @@ export type FooterEvent =
variants: string[]
current: string | undefined
}
| {
type: "queue"
queue: number
}
| {
type: "queued.prompts"
prompts: FooterQueuedPrompt[]
@@ -355,14 +352,8 @@ export type FooterEvent =
model: string
selection: NonNullable<RunInput["model"]>
}
| {
type: "turn.send"
queue: number
}
| {
type: "turn.idle"
queue: number
}
| { type: "turn.send" }
| { type: "turn.idle" }
| {
type: "turn.duration"
duration: string
@@ -438,7 +429,6 @@ export type LocalReplayRow = {
export type FooterApi = {
readonly isClosed: boolean
onPrompt(fn: (input: RunPrompt) => void): () => void
onQueuedRemove(fn: (messageID: string) => boolean | Promise<boolean>): () => void
onClose(fn: () => void): () => void
event(next: FooterEvent): void
append(commit: StreamCommit): void
packages/tui/test/mini/fixture/footer-api.ts
-9
View File
@@ -2,7 +2,6 @@ import type { FooterApi, FooterEvent, RunPrompt, StreamCommit } from "../../../s
export function createFooterApiFixture(input: { events?: FooterEvent[]; commits?: StreamCommit[] } = {}) {
const prompts = new Set<(input: RunPrompt) => void>()
const queuedRemoves = new Set<(messageID: string) => boolean | Promise<boolean>>()
const closes = new Set<() => void>()
const events = input.events ?? []
const commits = input.commits ?? []
@@ -17,10 +16,6 @@ export function createFooterApiFixture(input: { events?: FooterEvent[]; commits?
prompts.add(fn)
return () => prompts.delete(fn)
},
onQueuedRemove(fn) {
queuedRemoves.add(fn)
return () => queuedRemoves.delete(fn)
},
onClose(fn) {
if (closed) {
fn()
@@ -46,7 +41,6 @@ export function createFooterApiFixture(input: { events?: FooterEvent[]; commits?
destroy() {
api.close()
prompts.clear()
queuedRemoves.clear()
closes.clear()
},
}
@@ -60,8 +54,5 @@ export function createFooterApiFixture(input: { events?: FooterEvent[]; commits?
const prompt: RunPrompt = mode ? { text, parts: [], mode } : { text, parts: [] }
for (const fn of [...prompts]) fn(prompt)
},
removeQueued(messageID: string) {
for (const fn of [...queuedRemoves]) void fn(messageID)
},
}
}
packages/tui/test/mini/footer-keymap.test.tsx
-2
View File
@@ -12,7 +12,6 @@ test("down opens subagents from an empty prompt", async () => {
const [state] = createSignal<FooterState>({
phase: "idle",
status: "",
queue: 0,
model: "gpt-5",
usage: "",
first: false,
@@ -70,7 +69,6 @@ test("down opens subagents from an empty prompt", async () => {
onRows={() => {}}
onLayout={() => {}}
onStatus={() => {}}
onQueuedRemove={async () => true}
/>
</Keymap.Provider>
)
packages/tui/test/mini/footer.view.test.tsx
+24 -24
View File
@@ -92,7 +92,6 @@ function footerState(input: Partial<FooterState> = {}) {
return createSignal<FooterState>({
phase: "idle",
status: "",
queue: 0,
model: "gpt-5",
usage: "",
first: false,
@@ -158,7 +157,6 @@ async function renderFooter(
onRows={() => {}}
onLayout={() => {}}
onStatus={() => {}}
onQueuedRemove={async () => true}
/>
</Keymap.Provider>
)
@@ -680,10 +678,15 @@ test("direct subagent panel closes when moving up from the first item", async ()
}
})
test("direct queued prompt panel renders pending prompt actions", async () => {
const [prompts] = createSignal([{ messageID: "m-1", prompt: { text: "fix the auth test", parts: [] } }])
const edited: string[] = []
const deleted: string[] = []
test("direct pending panel shows durable delivery without edit actions", async () => {
const [prompts] = createSignal([
{
messageID: "m-1",
prompt: { text: "fix the auth test", parts: [] },
delivery: "queue" as const,
admittedSeq: 1,
},
])
const app = await testRender(
() => (
@@ -692,12 +695,6 @@ test("direct queued prompt panel renders pending prompt actions", async () => {
theme={() => RUN_THEME_FALLBACK.footer}
prompts={prompts}
onClose={() => {}}
onEdit={(prompt) => {
edited.push(prompt.messageID)
}}
onDelete={(prompt) => {
deleted.push(prompt.messageID)
}}
/>
</box>
),
@@ -709,16 +706,14 @@ test("direct queued prompt panel renders pending prompt actions", async () => {
const frame = app.captureCharFrame()
const list = panelMenu(app.renderer.root)
expect(frame).toContain("Queued prompts")
expect(frame).toContain("Pending work")
expect(frame).toContain("fix the auth test")
expect(frame).toContain("queued")
expect(frame).toContain("queue")
expect(frame).not.toContain("┌")
expect(frame).not.toContain("┃")
expectPaletteList(list, 0)
app.mockInput.pressKey("e", { ctrl: true })
app.mockInput.pressKey("DELETE")
expect(edited).toEqual(["m-1"])
expect(deleted).toEqual(["m-1"])
expect(frame).not.toContain("edit")
expect(frame).not.toContain("remove")
} finally {
app.renderer.destroy()
}
@@ -999,11 +994,10 @@ test.skip("direct footer clears the synthetic skills draft when the panel closes
}
})
test("direct footer shows editable prompts and additional queued work while running", async () => {
test("direct footer shows authoritative pending work while running", async () => {
const [state] = createSignal<FooterState>({
phase: "running",
status: "",
queue: 3,
model: "gpt-5",
usage: "",
first: false,
@@ -1036,7 +1030,14 @@ test("direct footer shows editable prompts and additional queued work while runn
state={state}
view={view}
subagent={subagents}
queuedPrompts={() => [{ messageID: "m-queued", prompt: { text: "follow up", parts: [] } }]}
queuedPrompts={() => [
{
messageID: "m-queued",
prompt: { text: "follow up", parts: [] },
delivery: "queue",
admittedSeq: 1,
},
]}
theme={() => RUN_THEME_FALLBACK}
tuiConfig={tuiConfig}
onSubmit={() => true}
@@ -1053,7 +1054,6 @@ test("direct footer shows editable prompts and additional queued work while runn
onRows={() => {}}
onLayout={() => {}}
onStatus={() => {}}
onQueuedRemove={async () => true}
/>
</Keymap.Provider>
)
@@ -1088,9 +1088,9 @@ test("direct footer shows editable prompts and additional queued work while runn
expect(spinner).toBeDefined()
expect(frame).toContain("a-model-name-long-enough-to-force-responsive-truncation")
expect(frame).toContain("3 queued")
expect(frame).toContain("1 pending")
expect(frame).toContain("ctrl+b background")
expect(frame).toContain("ctrl+x q 3 queued")
expect(frame).toContain("ctrl+x q 1 pending")
expect(frame).toContain("↓ subagents")
expect(frame).toContain("ctrl+p cmd")
expect(frame).toContain("a-model-name-long-enough-to-force-responsive-truncation")
packages/tui/test/mini/runtime.queue.test.ts
+56 -164
View File
@@ -1,8 +1,16 @@
import { describe, expect, test } from "bun:test"
import { runPromptQueue } from "../../src/mini/runtime.queue"
import { runPromptQueue as runPromptQueueBase, type QueueInput } from "../../src/mini/runtime.queue"
import type { RunPrompt } from "../../src/mini/types"
import { createFooterApiFixture } from "./fixture/footer-api"
function runPromptQueue(input: Omit<QueueInput, "admit" | "settle"> & Partial<Pick<QueueInput, "admit" | "settle">>) {
return runPromptQueueBase({
admit: async () => {},
settle: async () => {},
...input,
})
}
describe("run runtime queue", () => {
test("ignores empty prompts", async () => {
const ui = createFooterApiFixture()
@@ -171,206 +179,90 @@ describe("run runtime queue", () => {
])
})
test("passes prompts to onSend", async () => {
test("durably admits in-flight follow-ups in submission order", async () => {
const ui = createFooterApiFixture()
const seen: string[] = []
await runPromptQueue({
footer: ui.api,
initialInput: " hello ",
onSend: (input) => {
seen.push(input.text)
},
run: async () => {
ui.api.close()
},
})
expect(seen).toEqual([" hello "])
})
test("appends the user row before the turn starts", async () => {
const ui = createFooterApiFixture()
await runPromptQueue({
footer: ui.api,
initialInput: "/fmt bash",
run: async () => {
expect(ui.commits).toEqual([
{
kind: "user",
text: "/fmt bash",
phase: "start",
source: "system",
messageID: expect.any(String),
},
])
ui.api.close()
},
})
})
test("runs queued prompts in order", async () => {
const ui = createFooterApiFixture()
const seen: string[] = []
let wake: (() => void) | undefined
const gate = new Promise<void>((resolve) => {
wake = resolve
})
const admitted: string[] = []
const gate = Promise.withResolvers<void>()
const task = runPromptQueue({
footer: ui.api,
run: async (input) => {
seen.push(input.text)
if (seen.length === 1) {
await gate
return
}
ui.api.close()
run: async (input, _signal, onAdmitted) => {
admitted.push(`${input.text}:steer`)
onAdmitted()
await gate.promise
},
admit: async (input) => {
admitted.push(`${input.text}:queue`)
},
settle: async () => ui.api.close(),
})
ui.submit("one")
ui.submit("two")
await Promise.resolve()
expect(seen).toEqual(["one"])
wake?.()
await task
expect(seen).toEqual(["one", "two"])
})
test("exposes ordinary in-flight prompts for removal before sending", async () => {
const ui = createFooterApiFixture()
const turns: RunPrompt[] = []
let wake: (() => void) | undefined
const gate = new Promise<void>((resolve) => {
wake = resolve
})
const task = runPromptQueue({
footer: ui.api,
run: async (input) => {
turns.push(input)
await gate
},
})
ui.submit("one")
ui.submit("two")
await Promise.resolve()
await Promise.resolve()
expect(turns.map((item) => item.text)).toEqual(["one"])
expect(turns[0]?.messageID).toEqual(expect.any(String))
ui.submit("three")
while (admitted.length < 3) await Bun.sleep(0)
expect(admitted).toEqual(["one:steer", "two:queue", "three:queue"])
expect(ui.commits.map((item) => item.text)).toEqual(["one"])
const first = ui.events.find((item) => item.type === "queued.prompts")
const event = ui.events.findLast((item) => item.type === "queued.prompts")
expect(first?.type === "queued.prompts" ? first.prompts : []).toEqual([])
expect(
first?.type === "queued.prompts" && event?.type === "queued.prompts" ? first.prompts === event.prompts : true,
).toBe(false)
expect(ui.events.findLast((item) => item.type === "queue")).toEqual({ type: "queue", queue: 1 })
expect(event?.type === "queued.prompts" ? event.prompts.map((item) => item.prompt.text) : []).toEqual(["two"])
if (event?.type === "queued.prompts") ui.removeQueued(event.prompts[0]!.messageID)
await Promise.resolve()
wake?.()
ui.api.close()
gate.resolve()
await task
expect(turns.map((item) => item.text)).toEqual(["one"])
})
test("removing one managed queued prompt preserves the others", async () => {
test("continues durable admission after one fails", async () => {
const ui = createFooterApiFixture()
const turns: string[] = []
let wake: (() => void) | undefined
const gate = new Promise<void>((resolve) => {
wake = resolve
})
const admitted: string[] = []
const errors: string[] = []
const gate = Promise.withResolvers<void>()
const task = runPromptQueue({
footer: ui.api,
run: async (input) => {
turns.push(input.text)
if (input.text === "active") await gate
if (input.text === "queued three") ui.api.close()
run: async (_input, _signal, admitted) => {
admitted()
await gate.promise
},
})
ui.submit("active")
ui.submit("queued one")
ui.submit("queued two")
ui.submit("queued three")
await Promise.resolve()
await Promise.resolve()
const event = ui.events.findLast((item) => item.type === "queued.prompts")
if (event?.type === "queued.prompts") {
const second = event.prompts.find((item) => item.prompt.text === "queued two")
if (second) ui.removeQueued(second.messageID)
}
wake?.()
await task
expect(turns).toEqual(["active", "queued one", "queued three"])
})
test("drains a prompt queued during an in-flight turn", async () => {
const ui = createFooterApiFixture()
const seen: string[] = []
let wake: (() => void) | undefined
const gate = new Promise<void>((resolve) => {
wake = resolve
})
const task = runPromptQueue({
footer: ui.api,
run: async (input) => {
seen.push(input.text)
if (seen.length === 1) {
await gate
return
}
ui.api.close()
admit: async (input) => {
if (input.text === "two") throw new Error("admission failed")
admitted.push(input.text)
},
onAdmissionError: (_prompt, error) => {
errors.push(error instanceof Error ? error.message : String(error))
},
settle: async () => ui.api.close(),
})
ui.submit("one")
await Promise.resolve()
expect(seen).toEqual(["one"])
wake?.()
await Promise.resolve()
ui.submit("two")
ui.submit("three")
while (admitted.length === 0) await Bun.sleep(0)
gate.resolve()
await task
expect(seen).toEqual(["one", "two"])
expect(errors).toEqual(["admission failed"])
expect(admitted).toEqual(["three"])
})
test("close aborts the active run and drops pending queued work", async () => {
test("close aborts an in-flight durable admission", async () => {
const ui = createFooterApiFixture()
const seen: string[] = []
let hit = false
let admissionHit = false
const admissionStarted = Promise.withResolvers<void>()
const task = runPromptQueue({
footer: ui.api,
run: async (input, signal) => {
seen.push(input.text)
run: async (_input, signal, admitted) => {
admitted()
await new Promise<void>((resolve) => signal.addEventListener("abort", () => resolve(), { once: true }))
},
admit: async (_prompt, signal) => {
admissionStarted.resolve()
await new Promise<void>((resolve) => {
if (signal.aborted) {
hit = true
admissionHit = true
resolve()
return
}
signal.addEventListener(
"abort",
() => {
hit = true
admissionHit = true
resolve()
},
{ once: true },
@@ -382,11 +274,11 @@ describe("run runtime queue", () => {
ui.submit("one")
await Promise.resolve()
ui.submit("two")
await admissionStarted.promise
ui.api.close()
await task
expect(hit).toBe(true)
expect(seen).toEqual(["one"])
expect(admissionHit).toBe(true)
})
test("propagates run errors", async () => {
packages/tui/test/mini/runtime.test.ts
+4
View File
@@ -93,6 +93,8 @@ describe("run interactive runtime", () => {
streamStarted.resolve()
return {
runPromptTurn: async () => {},
queuePromptTurn: async () => {},
waitForIdle: async () => {},
interruptActiveTurn: async () => {},
selectSubagent: () => {},
settleForm: (sessionID: string, formID: string) => settled.push({ sessionID, formID }),
@@ -432,6 +434,8 @@ describe("run interactive runtime", () => {
setTimeout(() => input.footer.close(), 0)
return {
runPromptTurn: async () => {},
queuePromptTurn: async () => {},
waitForIdle: async () => {},
interruptActiveTurn: async () => {},
selectSubagent: () => {},
replayOnResize: async () => false,
packages/tui/test/mini/stream-v2.transport.test.ts
+248 -163
View File
@@ -127,6 +127,8 @@ function sdk(input: {
globals?: FormInfo[]
globalLocation?: { directory: string; workspaceID?: string }
permissions?: Record<string, PermissionV2Request[]>
pending?: Record<string, Awaited<ReturnType<OpenCodeClient["session"]["pending"]["list"]>>>
wait?: () => Promise<void>
}) {
const client = OpenCode.make({ baseUrl: "https://opencode.test" })
let subscription = 0
@@ -159,6 +161,8 @@ function sdk(input: {
}),
)
spyOn(client.session, "active").mockImplementation(() => ok(input.active?.() ?? {}))
spyOn(client.session.pending, "list").mockImplementation((request) => ok(input.pending?.[request.sessionID] ?? []))
spyOn(client.session, "wait").mockImplementation(() => input.wait?.() ?? ok(undefined))
spyOn(client.session, "message").mockImplementation((request) => {
const message = input.messages?.[request.sessionID]?.find((item) => item.id === request.messageID)
return message ? (ok(message) as never) : Promise.reject(new Error(`message not found: ${request.messageID}`))
@@ -417,35 +421,23 @@ describe("V2 mini transport", () => {
await transport.close()
})
test("finalizes an idle projection before reducing live output", async () => {
test("waits authoritatively and reconciles the projected terminal suffix", async () => {
const events = feed()
events.push(connected())
const settled = defer()
const messages: SessionMessages = []
const client = sdk({
streams: [events],
messages: {
ses_1: [
{
id: "msg_old",
type: "assistant",
agent: "build",
model: { providerID: "test", id: "model" },
content: [{ type: "text", text: "[Link](https://example.com)" }],
time: { created: 1 },
},
],
},
messages: { ses_1: messages },
wait: () => settled.promise,
})
const ui = footer()
const idle = spyOn(ui.api, "idle")
const transport = await createSessionTransport({
sdk: client,
sessionID: "ses_1",
thinking: true,
replay: true,
thinking: false,
footer: ui.api,
})
expect(ui.commits.map((item) => item.text)).toEqual(["[Link](https://example.com)"])
expect(idle).toHaveBeenCalledTimes(1)
let admitted = false
spyOn(client.session, "prompt").mockImplementation((request) => {
@@ -480,7 +472,7 @@ describe("V2 mini transport", () => {
sessionID: "ses_1",
assistantMessageID: "msg_assistant",
ordinal: 0,
delta: "answer",
delta: "ans",
},
})
events.push({
@@ -490,10 +482,222 @@ describe("V2 mini transport", () => {
durable: durable("ses_1"),
data: { sessionID: "ses_1" },
})
let done = false
void turn.then(() => {
done = true
})
await Bun.sleep(0)
expect(done).toBe(false)
messages.push(
{ id: "msg_prompt", type: "user", text: "hello", time: { created: 2 } },
{
id: "msg_assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", id: "model" },
content: [{ type: "text", text: "answer" }],
time: { created: 3, completed: 4 },
},
)
settled.resolve()
await turn
expect(ui.commits.map((item) => item.text)).toEqual(["[Link](https://example.com)", "answer"])
expect(ui.events).toContainEqual({ type: "stream.patch", patch: { phase: "idle", status: "" } })
expect(ui.commits.map((item) => item.text)).toEqual(["ans", "wer"])
await transport.close()
})
test("shows durable pending delivery and appends queued input on promotion", async () => {
const events = feed()
events.push(connected())
const client = sdk({
streams: [events],
pending: {
ses_1: [
{
admittedSeq: 1,
id: "msg_queued",
sessionID: "ses_1",
timeCreated: 1,
type: "user",
data: { text: "follow up" },
delivery: "queue",
},
],
},
})
const ui = footer()
const transport = await createSessionTransport({
sdk: client,
sessionID: "ses_1",
thinking: false,
footer: ui.api,
})
const pending = () =>
ui.events
.findLast((item) => item.type === "queued.prompts")
?.prompts.map((item) => [item.messageID, item.delivery, item.admittedSeq])
expect(pending()).toEqual([["msg_queued", "queue", 1]])
events.push({
id: "evt_promoted",
created: 2,
type: "session.input.promoted",
durable: durable("ses_1", 2),
data: { sessionID: "ses_1", inputID: "msg_queued" },
})
while (!ui.commits.some((item) => item.messageID === "msg_queued")) await Bun.sleep(0)
expect(ui.commits).toContainEqual(
expect.objectContaining({ kind: "user", messageID: "msg_queued", text: "follow up" }),
)
expect(pending()).toEqual([])
const prompt = spyOn(client.session, "prompt").mockImplementation((request) =>
ok({ ...promptAdmission(request), admittedSeq: 2 }) as never,
)
await transport.queuePromptTurn({
agent: undefined,
model: undefined,
variant: undefined,
prompt: { messageID: "msg_next", text: "another", parts: [] },
files: [],
includeFiles: false,
})
expect(prompt).toHaveBeenCalledWith(expect.objectContaining({ delivery: "queue" }), expect.anything())
events.push({
id: "evt_earlier_admission",
created: 3,
type: "session.input.admitted",
durable: durable("ses_1", 1),
data: {
sessionID: "ses_1",
inputID: "msg_earlier",
input: { type: "user", data: { text: "earlier" }, delivery: "steer" },
},
})
while (true) {
const pending = ui.events.findLast((item) => item.type === "queued.prompts")
if (pending?.type === "queued.prompts" && pending.prompts.length >= 2) break
await Bun.sleep(0)
}
expect(pending()).toEqual([
["msg_earlier", "steer", 1],
["msg_next", "queue", 2],
])
await transport.close()
})
test("reports an observed execution failure before prompt promotion", async () => {
const events = feed()
events.push(connected())
const idle = defer()
const client = sdk({ streams: [events], messages: { ses_1: [] }, wait: () => idle.promise })
const ui = footer()
const transport = await createSessionTransport({
sdk: client,
sessionID: "ses_1",
thinking: false,
footer: ui.api,
})
let admitted = false
spyOn(client.session, "prompt").mockImplementation((request) => {
admitted = true
return ok(promptAdmission(request)) as never
})
const turn = transport.runPromptTurn({
agent: undefined,
model: undefined,
variant: undefined,
prompt: { messageID: "msg_prompt", text: "hello", parts: [] },
files: [],
includeFiles: false,
})
while (!admitted) await Bun.sleep(0)
events.push({
id: "evt_failed",
created: 2,
type: "session.execution.failed",
durable: durable("ses_1", 2),
data: { sessionID: "ses_1", error: { type: "unknown", message: "instructions unavailable" } },
})
await Bun.sleep(0)
idle.resolve()
await turn
expect(ui.commits).toContainEqual(
expect.objectContaining({ kind: "error", messageID: "msg_prompt", text: "instructions unavailable" }),
)
await transport.close()
})
test("attributes an execution-only failure to the latest promoted prompt", async () => {
const events = feed()
events.push(connected())
const idle = defer()
const messages: SessionMessages = []
const client = sdk({ streams: [events], messages: { ses_1: messages }, wait: () => idle.promise })
const ui = footer()
const transport = await createSessionTransport({
sdk: client,
sessionID: "ses_1",
thinking: false,
footer: ui.api,
})
let admitted = false
spyOn(client.session, "prompt").mockImplementation((request) => {
admitted = true
return ok(promptAdmission(request)) as never
})
const turn = transport.runPromptTurn({
agent: undefined,
model: undefined,
variant: undefined,
prompt: { messageID: "msg_prompt", text: "hello", parts: [] },
files: [],
includeFiles: false,
})
while (!admitted) await Bun.sleep(0)
events.push({
id: "evt_prompt_promoted",
created: 2,
type: "session.input.promoted",
durable: durable("ses_1", 2),
data: { sessionID: "ses_1", inputID: "msg_prompt" },
})
await transport.queuePromptTurn({
agent: undefined,
model: undefined,
variant: undefined,
prompt: { messageID: "msg_queued", text: "follow up", parts: [] },
files: [],
includeFiles: false,
})
events.push({
id: "evt_queued_promoted",
created: 3,
type: "session.input.promoted",
durable: durable("ses_1", 3),
data: { sessionID: "ses_1", inputID: "msg_queued" },
})
events.push({
id: "evt_failed",
created: 4,
type: "session.execution.failed",
durable: durable("ses_1", 4),
data: { sessionID: "ses_1", error: { type: "unknown", message: "model unavailable" } },
})
await Bun.sleep(0)
messages.push(
{ id: "msg_prompt", type: "user", text: "hello", time: { created: 2 } },
{ id: "msg_queued", type: "user", text: "follow up", time: { created: 3 } },
)
idle.resolve()
await turn
expect(ui.commits).toContainEqual(
expect.objectContaining({ kind: "error", messageID: "msg_queued", text: "model unavailable" }),
)
await transport.close()
})
@@ -786,11 +990,12 @@ describe("V2 mini transport", () => {
await transport.close()
})
test("rebootstraps after disconnect and completes a promoted turn from idle active state", async () => {
test("reconnects and hydrates without completing before session.wait", async () => {
const first = feed()
const second = feed()
first.push(connected("evt_connected_1"))
second.push(connected("evt_connected_2"))
const idle = defer()
let running = true
const client = sdk({
streams: [first, second],
@@ -799,6 +1004,7 @@ describe("V2 mini transport", () => {
if (running) active.ses_1 = { type: "running" }
return active
},
wait: () => idle.promise,
})
let projected = false
spyOn(client.message, "list").mockImplementation(() =>
@@ -844,11 +1050,25 @@ describe("V2 mini transport", () => {
while (!admitted) await Bun.sleep(0)
projected = true
running = false
second.push({
id: "evt_prior_failed",
created: 1,
type: "session.execution.failed",
durable: durable("ses_1", 1),
data: { sessionID: "ses_1", error: { type: "unknown", message: "prior execution failed" } },
})
second.push({
id: "evt_prompted",
created: 2,
type: "session.input.promoted",
durable: durable("ses_1", 2),
data: { sessionID: "ses_1", inputID: "msg_prompt" },
})
first.close()
while (!ui.events.some((event) => event.type === "stream.patch" && event.patch.status === "reconnecting"))
await Bun.sleep(0)
idle.resolve()
await turn
expect(ui.events).toContainEqual({ type: "stream.patch", patch: { phase: "running", status: "reconnecting" } })
expect(ui.events).toContainEqual({ type: "stream.patch", patch: { phase: "idle", status: "" } })
await transport.close()
})
@@ -1789,52 +2009,6 @@ describe("V2 mini transport", () => {
await transport.close()
})
test("resolves an interrupted turn even when promotion never arrived", async () => {
const events = feed()
events.push(connected())
const client = sdk({
streams: [events],
active: () => ({ ses_1: { type: "running" } }),
})
const ui = footer()
const transport = await createSessionTransport({
sdk: client,
sessionID: "ses_1",
thinking: false,
footer: ui.api,
})
let admitted = false
// The generated method has conditional return types for throwOnError; this mock represents the successful branch.
// @ts-expect-error successful SDK response is valid for both modes at runtime
spyOn(client.session, "prompt").mockImplementation((request) => {
admitted = true
return ok({ data: promptAdmission(request) })
})
const interrupted = spyOn(client.session, "interrupt").mockImplementation(() => ok(undefined))
const turn = transport.runPromptTurn({
agent: undefined,
model: undefined,
variant: undefined,
prompt: { messageID: "msg_prompt", text: "hello", parts: [] },
files: [],
includeFiles: true,
})
while (!admitted) await Bun.sleep(0)
await transport.interruptActiveTurn()
events.push({
id: "evt_settled",
created: 0,
type: "session.execution.interrupted",
durable: durable("ses_1"),
data: { sessionID: "ses_1", reason: "user" },
})
await turn
expect(interrupted).toHaveBeenCalledWith({ sessionID: "ses_1" })
await transport.close()
})
test("falls back to the default model when selecting a variant on a fresh session", async () => {
const events = feed()
events.push(connected())
@@ -1901,7 +2075,8 @@ describe("V2 mini transport", () => {
test("interrupts the current Session when an active turn is aborted", async () => {
const events = feed()
events.push(connected())
const client = sdk({ streams: [events] })
const idle = defer()
const client = sdk({ streams: [events], wait: () => idle.promise })
const ui = footer()
const transport = await createSessionTransport({
sdk: client,
@@ -1940,13 +2115,7 @@ describe("V2 mini transport", () => {
})
await Bun.sleep(0)
controller.abort()
events.push({
id: "evt_settled",
created: 0,
type: "session.execution.interrupted",
durable: durable("ses_1"),
data: { sessionID: "ses_1", reason: "user" },
})
idle.resolve()
await turn
expect(interrupted).toHaveBeenCalledWith({ sessionID: "ses_1" })
@@ -2493,90 +2662,6 @@ describe("V2 mini transport", () => {
await transport.close()
})
test("does not resolve a skill turn before the matching activation is observed", async () => {
const events = feed()
events.push(connected())
const client = sdk({ streams: [events] })
const ui = footer()
const transport = await createSessionTransport({
sdk: client,
sessionID: "ses_1",
thinking: false,
footer: ui.api,
})
let sent = false
spyOn(client.session, "skill").mockImplementation(() => {
sent = true
return ok(undefined) as never
})
let done = false
const turn = transport
.runPromptTurn({
agent: undefined,
model: undefined,
variant: undefined,
prompt: {
messageID: "msg_skill",
text: "/tigerstyle",
parts: [],
command: { name: "tigerstyle", arguments: "", source: "skill" },
},
files: [],
includeFiles: true,
})
.then(() => {
done = true
})
while (!sent) await Bun.sleep(0)
events.push({
id: "evt_other",
created: 0,
type: "session.skill.activated",
durable: durable("ses_1"),
data: {
sessionID: "ses_1",
id: "other",
name: "other",
text: "other instructions",
},
})
events.push({
id: "evt_unrelated_settled",
created: 0,
type: "session.execution.succeeded",
durable: durable("ses_1"),
data: { sessionID: "ses_1" },
})
await Bun.sleep(0)
await Bun.sleep(0)
expect(done).toBe(false)
events.push({
id: "evt_skill",
created: 0,
type: "session.skill.activated",
durable: durable("ses_1"),
data: {
sessionID: "ses_1",
id: "tigerstyle",
name: "tigerstyle",
text: "skill instructions",
},
})
events.push({
id: "evt_skill_settled",
created: 0,
type: "session.execution.succeeded",
durable: durable("ses_1"),
data: { sessionID: "ses_1" },
})
await turn
expect(done).toBe(true)
await transport.close()
})
test("refreshes catalogs on connection and location-scoped invalidations", async () => {
const events = feed()
events.push(connected())