import { afterEach, describe, expect, mock, spyOn, test } from "bun:test" import fs from "fs/promises" import path from "path" import { pathToFileURL } from "node:url" import { OpenCode, type EventSubscribeOutput, type FormInfo, type MessageListOutput, type OpenCodeClient, type PermissionRequest, } from "@opencode-ai/client/promise" import { createSessionTransport } from "../../src/mini/stream-v2.transport" import type { StreamCommit } from "../../src/mini/types" import { createFooterApiFixture } from "./fixture/footer-api" import { canonicalToolPart } from "./fixture/tool-part" import { tmpdir } from "../fixture/fixture" type RunV2Event = EventSubscribeOutput function feed() { const values: RunV2Event[] = [] let closed = false let wake: (() => void) | undefined const stream = (async function* (): AsyncGenerator { while (!closed || values.length > 0) { if (values.length === 0) { await new Promise((resolve) => { wake = resolve }) continue } const value = values.shift() if (value) yield value } })() return { stream, push(value: RunV2Event) { values.push(value) wake?.() wake = undefined }, close() { closed = true wake?.() wake = undefined }, } } function ok(data: T) { return Promise.resolve(data) } function defer() { let resolve!: (value: T | PromiseLike) => void const promise = new Promise((done) => { resolve = done }) return { promise, resolve } } function connected(id = "evt_connected") { return { id, type: "server.connected", data: {} } satisfies RunV2Event } function durable(sessionID: string, seq?: number): { aggregateID: string; seq: number; version: 1 } function durable( sessionID: string, seq: number, version: Version, ): { aggregateID: string; seq: number; version: Version } function durable(sessionID: string, seq = 0, version: 1 | 2 = 1) { return { aggregateID: sessionID, seq, version } } function promptAdmission(input: Parameters[0], sessionID = "ses_1") { return { admittedSeq: 1, id: input.id ?? "msg_prompt", sessionID, type: "user" as const, data: { text: input.text, files: input.files, agents: input.agents, metadata: input.metadata, }, delivery: input.delivery ?? ("steer" as const), timeCreated: 2, } } function footer() { return createFooterApiFixture() } type SessionMessages = MessageListOutput["data"] function compaction(status: "running" | "completed", summary: string): SessionMessages[number] { const message = { id: "msg_compaction", type: "compaction" as const, reason: "auto" as const, summary, recent: "", time: { created: 1 }, } if (status === "running") return { ...message, status } return { ...message, status } } function form(id: string, sessionID: string, title = id): FormInfo { return { id, sessionID, title, fields: [ { key: "answer", type: "string", options: [{ value: "yes", label: "Yes" }], custom: true, }, ], } } function eventForm(info: FormInfo): Extract["data"]["form"] { return info as Extract["data"]["form"] } function sdk(input: { streams: ReturnType[] active?: () => Record messages?: Record sessions?: Array<{ id: string; parentID?: string; title?: string; agent?: string; time: { updated: number } }> forms?: Record globals?: FormInfo[] globalLocation?: { directory: string; workspaceID?: string } permissions?: Record pending?: Record>> wait?: () => Promise }) { const client = OpenCode.make({ baseUrl: "https://opencode.test" }) let subscription = 0 spyOn(client.event, "subscribe").mockImplementation(() => input.streams[subscription++]?.stream ?? feed().stream) spyOn(client.message, "list").mockImplementation((request) => ok({ data: input.messages?.[request.sessionID] ?? [ { id: "msg_old", type: "user" as const, text: "previous prompt", files: [], agents: [], time: { created: 1 }, }, ], cursor: {}, }), ) spyOn(client.permission, "list").mockImplementation((request) => ok(input.permissions?.[request.sessionID] ?? [])) spyOn(client.form, "list").mockImplementation((request) => ok(input.forms?.[request.sessionID] ?? [])) spyOn(client.form.request, "list").mockImplementation(() => ok({ location: { directory: input.globalLocation?.directory ?? "/tmp", workspaceID: input.globalLocation?.workspaceID, project: { id: "proj_1", directory: input.globalLocation?.directory ?? "/tmp" }, }, data: input.globals ?? [], }), ) 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}`)) }) spyOn(client.session, "switchAgent").mockImplementation(() => ok(undefined)) spyOn(client.session, "switchModel").mockImplementation(() => ok(undefined)) // The generated methods have conditional return types for throwOnError; the // minimal shapes below are enough for family discovery and model fallback. spyOn(client.session, "list").mockImplementation((request) => { const parentID = request?.parentID return ok({ location: { directory: "/tmp", project: { id: "proj_1", directory: "/tmp" } }, data: input.sessions?.filter((session) => parentID === undefined ? true : parentID === null ? session.parentID === undefined : session.parentID === parentID, ) ?? [], }) as never }) spyOn(client.model, "default").mockImplementation( () => ok({ location: { directory: "/tmp", project: { id: "proj_1", directory: "/tmp" } }, data: undefined, }) as never, ) return client } afterEach(() => { mock.restore() }) describe("V2 mini transport", () => { test("renders projected compactions as labeled transcript boundaries", async () => { const events = feed() events.push(connected()) const ui = footer() const transport = await createSessionTransport({ sdk: sdk({ streams: [events], messages: { ses_1: [compaction("completed", "## Transport")], }, }), sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, }) expect(ui.commits).toMatchObject([ { text: "Compaction", compaction: true, messageID: "msg_compaction" }, { text: "## Transport", phase: "progress", messageID: "msg_compaction" }, { text: "", phase: "final", messageID: "msg_compaction" }, ]) await transport.close() }) test("shows an active compaction boundary before live summary output without history replay", async () => { const events = feed() events.push(connected()) const ui = footer() const transport = await createSessionTransport({ sdk: sdk({ streams: [events], active: () => ({ ses_1: { type: "running" } }), messages: { ses_1: [compaction("running", "")], }, }), sessionID: "ses_1", thinking: false, footer: ui.api, }) events.push({ id: "evt_compaction_delta", created: 2, type: "session.compaction.delta", data: { sessionID: "ses_1", text: "Transport" }, }) events.push({ id: "evt_compaction_ended", created: 3, type: "session.compaction.ended", durable: durable("ses_1", 3), data: { sessionID: "ses_1", reason: "auto", text: "Transport", recent: "" }, }) while (!ui.commits.some((commit) => commit.phase === "final")) await Bun.sleep(0) expect(ui.commits).toMatchObject([ { text: "Compaction", compaction: true, messageID: "msg_compaction" }, { text: "Transport", phase: "progress", messageID: "msg_compaction" }, { text: "", phase: "final", messageID: "msg_compaction" }, ]) await transport.close() }) test("reports session title changes", async () => { const events = feed() events.push(connected()) const titles: string[] = [] const transport = await createSessionTransport({ sdk: sdk({ streams: [events] }), sessionID: "ses_1", thinking: false, footer: footer().api, onSessionTitle: (title) => titles.push(title), }) events.push({ id: "evt_renamed", created: 1, type: "session.renamed", durable: durable("ses_1", 1), data: { sessionID: "ses_1", title: "Greeting" }, }) while (titles.length === 0) await Bun.sleep(0) expect(titles).toEqual(["Greeting"]) await transport.close() }) test("formats footer usage with compact tokens and context percentage", async () => { const events = feed() events.push(connected()) const ui = footer() const transport = await createSessionTransport({ sdk: sdk({ streams: [events] }), sessionID: "ses_1", thinking: false, footer: ui.api, contextLimit: (model) => (model.providerID === "test" && model.modelID === "model" ? 160_000 : undefined), }) events.push({ id: "evt_step_started", created: 1, type: "session.step.started", durable: durable("ses_1", 1), data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", agent: "build", model: { providerID: "test", id: "model" }, }, }) events.push({ id: "evt_step_ended", created: 2, type: "session.step.ended", durable: durable("ses_1", 2), data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", finish: "stop", cost: 0, tokens: { input: 7_000, output: 500, reasoning: 8, cache: { read: 0, write: 0 } }, }, }) while (!ui.events.some((event) => event.type === "stream.patch" && event.patch.usage)) await Bun.sleep(0) expect(ui.events).toContainEqual({ type: "stream.patch", patch: { usage: "7.5K (5%)" } }) await transport.close() }) test("recursively hydrates blockers for direct and transitive descendants", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], sessions: [ { id: "ses_child", parentID: "ses_1", title: "Child", time: { updated: 2 } }, { id: "ses_grandchild", parentID: "ses_child", title: "Grandchild", time: { updated: 1 } }, ], forms: { ses_child: [form("frm_child", "ses_child")], ses_grandchild: [form("frm_grandchild", "ses_grandchild")], }, }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const snapshots = ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) expect(snapshots.at(-1)?.tabs.map((item) => item.sessionID)).toEqual(["ses_child", "ses_grandchild"]) expect(snapshots.at(-1)?.forms.map((item) => item.id)).toEqual(["frm_child", "frm_grandchild"]) expect( ui.events.find( (event) => event.type === "stream.view" && event.view.type === "form" && event.view.request.id === "frm_child", ), ).toMatchObject({ type: "stream.view", view: { type: "form", request: { id: "frm_child", sessionID: "ses_child" } }, }) transport.settleForm?.("ses_child", "frm_child") expect(ui.events.at(-1)).toMatchObject({ type: "stream.view", view: { type: "form", request: { id: "frm_grandchild", sessionID: "ses_grandchild" } }, }) await transport.close() }) test("resolves a pre-existing child permission from its exact source message at startup", async () => { const events = feed() events.push(connected()) const sourceMessage = { id: "msg_child_source", type: "assistant" as const, agent: "build", model: { providerID: "test", id: "model" }, content: [ canonicalToolPart( "shell", { status: "running" as const, input: { command: "git status --short" }, metadata: {}, }, "call_child_source", ), ], time: { created: 1 }, } const permission: PermissionRequest = { id: "per_child_startup", sessionID: "ses_child", action: "shell", resources: ["git status --short"], source: { type: "tool", messageID: "msg_child_source", callID: "call_child_source" }, } const client = sdk({ streams: [events], sessions: [{ id: "ses_child", parentID: "ses_1", title: "Child", time: { updated: 1 } }], permissions: { ses_child: [permission] }, messages: { ses_child: [sourceMessage], }, }) const releaseSource = defer() let sourceLookups = 0 spyOn(client.session, "message").mockImplementation(async () => { sourceLookups++ if (sourceLookups === 1) throw new Error("source temporarily unavailable") await releaseSource.promise return sourceMessage as never }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) while (sourceLookups < 2) await Bun.sleep(0) expect( ui.events.some( (event) => event.type === "stream.view" && event.view.type === "permission" && event.view.request.id === permission.id, ), ).toBe(false) releaseSource.resolve() while ( !ui.events.some( (event) => event.type === "stream.view" && event.view.type === "permission" && event.view.request.id === permission.id, ) ) await Bun.sleep(0) expect(client.session.message).toHaveBeenCalledWith( { sessionID: "ses_child", messageID: "msg_child_source" }, { signal: expect.any(AbortSignal) }, ) expect( ui.events.find( (event) => event.type === "stream.view" && event.view.type === "permission" && event.view.request.id === permission.id, ), ).toMatchObject({ view: { request: { tool: { id: "call_child_source", name: "shell", state: { status: "running", input: { command: "git status --short" } }, }, }, }, }) expect(client.message.list).not.toHaveBeenCalledWith( expect.objectContaining({ sessionID: "ses_child" }), expect.anything(), ) await transport.close() }) test("reduces nested form owners idempotently and filters global events by complete location", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], sessions: [{ id: "ses_child", parentID: "ses_1", title: "Child", time: { updated: 1 } }], globalLocation: { directory: "/work", workspaceID: "wrk_1" }, }) const ui = footer() const transport = await createSessionTransport({ sdk: client, location: { directory: "/work", workspaceID: "wrk_1" }, sessionID: "ses_1", thinking: false, footer: ui.api, }) const child = form("frm_child_live", "ses_child") events.push({ id: "evt_child_form", created: 1, type: "form.created", data: { form: eventForm(child) } }) events.push({ id: "evt_child_form_retry", created: 2, type: "form.created", data: { form: eventForm(child) } }) while ( !ui.events.some( (event) => event.type === "stream.view" && event.view.type === "form" && event.view.request.id === child.id, ) ) await Bun.sleep(0) const childSnapshots = ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) expect(childSnapshots.at(-1)?.forms.filter((item) => item.id === child.id)).toHaveLength(1) events.push({ id: "evt_child_form_done", created: 3, type: "form.replied", data: { id: child.id, sessionID: "ses_child", answer: { answer: "yes" } }, }) const global = form("frm_global_live", "global") events.push({ id: "evt_global_wrong", created: 4, type: "form.created", location: { directory: "/work", workspaceID: "wrk_other" }, data: { form: eventForm(global) }, }) await Bun.sleep(0) expect( ui.events.some( (event) => event.type === "stream.view" && event.view.type === "form" && event.view.request.id === global.id, ), ).toBe(false) events.push({ id: "evt_global_right", created: 5, type: "form.created", location: { directory: "/work", workspaceID: "wrk_1" }, data: { form: eventForm(global) }, }) while ( !ui.events.some( (event) => event.type === "stream.view" && event.view.type === "form" && event.view.request.id === global.id, ) ) await Bun.sleep(0) expect(ui.events.at(-1)).toMatchObject({ type: "stream.view", view: { type: "form", request: { id: "frm_global_live", location: { directory: "/work", workspaceID: "wrk_1" } }, }, }) const beforeCancel = ui.events.filter((event) => event.type === "stream.view").length events.push({ id: "evt_global_done", created: 6, type: "form.cancelled", location: { directory: "/work", workspaceID: "wrk_1" }, data: { id: global.id, sessionID: "global" }, }) while (ui.events.filter((event) => event.type === "stream.view").length === beforeCancel) await Bun.sleep(0) expect(ui.events.filter((event) => event.type === "stream.view").at(-1)).toEqual({ type: "stream.view", view: { type: "prompt" }, }) await transport.close() }) 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: messages }, wait: () => settled.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({ data: promptAdmission(request) }) as never }) 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) events.push({ id: "evt_prompted", created: 0, type: "session.input.promoted", durable: durable("ses_1"), data: { sessionID: "ses_1", inputID: "msg_prompt", }, }) events.push({ id: "evt_text", created: 0, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: "ans", }, }) events.push({ id: "evt_settled", created: 0, type: "session.execution.succeeded", 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(["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: "review", model: undefined, variant: undefined, prompt: { messageID: "msg_next", text: "another", parts: [] }, files: [], includeFiles: false, }) expect(client.session.switchAgent).toHaveBeenCalledWith({ sessionID: "ses_1", agent: "review" }, expect.anything()) 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() }) test("sends local file and directory mentions as structured prompt files", async () => { await using tmp = await tmpdir() const filePath = path.join(tmp.path, "note.ts") const contextPath = path.join(tmp.path, "context.txt") const directoryPath = path.join(tmp.path, "docs") await Bun.write(filePath, "export const answer = 42\n") await Bun.write(contextPath, "context body") await fs.mkdir(directoryPath) await Bun.write(path.join(directoryPath, "README.md"), "# hello\n") const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) const ui = footer() const transport = await createSessionTransport({ sdk: client, readTextFile: (url) => fs.readFile(new URL(url), "utf8"), sessionID: "ses_1", thinking: false, footer: ui.api, }) let request: Parameters[0] | undefined spyOn(client.session, "prompt").mockImplementation((input) => { request = input queueMicrotask(() => { events.push({ id: "evt_prompted", created: 0, type: "session.input.promoted", durable: durable("ses_1"), data: { sessionID: "ses_1", inputID: "msg_prompt", }, }) events.push({ id: "evt_settled", created: 0, type: "session.execution.succeeded", durable: durable("ses_1"), data: { sessionID: "ses_1" }, }) }) return ok({ data: promptAdmission(input) }) as never }) await transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_prompt", text: "Review @note.ts and @docs", parts: [ { type: "file", url: pathToFileURL(filePath).href, mime: "text/plain", filename: "note.ts", source: { type: "file", path: "note.ts", text: { start: 7, end: 15, value: "@note.ts" } }, }, { type: "file", url: pathToFileURL(`${directoryPath}${path.sep}`).href, mime: "application/x-directory", filename: "docs", source: { type: "file", path: "docs/", text: { start: 20, end: 25, value: "@docs" } }, }, ], }, files: [ { type: "file", url: pathToFileURL(contextPath).href, filename: "context.txt", mime: "text/plain" }, { type: "file", url: "file:///tmp/image.png", filename: "image.png", mime: "image/png" }, ], includeFiles: true, }) expect(request?.text).toBe('Review @note.ts and @docs\n\n\ncontext body\n') expect(request?.files).toEqual([ { uri: "file:///tmp/image.png", name: "image.png" }, { uri: pathToFileURL(filePath).href, name: "note.ts", mention: { start: 7, end: 15, text: "@note.ts" }, }, { uri: pathToFileURL(`${directoryPath}${path.sep}`).href, name: "docs", mention: { start: 20, end: 25, text: "@docs" }, }, ]) await transport.close() }) test("sends attached file mentions as structured prompt files without reading them", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) const ui = footer() const remoteRead = spyOn(client.file, "read") const remoteList = spyOn(client.file, "list") const transport = await createSessionTransport({ sdk: client, location: { directory: "/remote/project" }, sessionID: "ses_1", thinking: false, footer: ui.api, }) let request: Parameters[0] | undefined // 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((input) => { request = input queueMicrotask(() => { events.push({ id: "evt_prompted", created: 0, type: "session.input.promoted", durable: durable("ses_1"), data: { sessionID: "ses_1", inputID: "msg_prompt", }, }) events.push({ id: "evt_settled", created: 0, type: "session.execution.succeeded", durable: durable("ses_1"), data: { sessionID: "ses_1" }, }) }) return ok({ data: promptAdmission(input) }) }) await transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_prompt", text: "Review @note.ts and @docs", parts: [ { type: "file", url: "file:///remote/project/note.ts", mime: "text/plain", filename: "note.ts", source: { type: "file", path: "note.ts", text: { start: 7, end: 15, value: "@note.ts" } }, }, { type: "file", url: "file:///remote/project/docs", mime: "application/x-directory", filename: "docs", source: { type: "file", path: "docs", text: { start: 20, end: 25, value: "@docs" } }, }, ], }, files: [], includeFiles: true, }) expect(remoteRead).not.toHaveBeenCalled() expect(remoteList).not.toHaveBeenCalled() expect(request?.text).toBe("Review @note.ts and @docs") expect(request?.files).toEqual([ { uri: "file:///remote/project/note.ts", name: "note.ts", mention: { start: 7, end: 15, text: "@note.ts" }, }, { uri: "file:///remote/project/docs", name: "docs", mention: { start: 20, end: 25, text: "@docs" }, }, ]) await transport.close() }) test("sends local media mentions as structured prompt files", async () => { await using tmp = await tmpdir() const filePath = path.join(tmp.path, "diagram.png") await Bun.write(filePath, Uint8Array.of(0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00)) 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 request: Parameters[0] | undefined // 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((input) => { request = input queueMicrotask(() => { events.push({ id: "evt_prompted", created: 0, type: "session.input.promoted", durable: durable("ses_1"), data: { sessionID: "ses_1", inputID: "msg_prompt", }, }) events.push({ id: "evt_settled", created: 0, type: "session.execution.succeeded", durable: durable("ses_1"), data: { sessionID: "ses_1" }, }) }) return ok({ data: promptAdmission(input) }) }) await transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_prompt", text: "Review @diagram.png", parts: [ { type: "file", url: pathToFileURL(filePath).href, mime: "text/plain", filename: "diagram.png", source: { type: "file", path: "diagram.png", text: { start: 7, end: 19, value: "@diagram.png" } }, }, ], }, files: [], includeFiles: true, }) expect(request?.text).toBe("Review @diagram.png") expect(request?.files).toEqual([ { name: "diagram.png", uri: pathToFileURL(filePath).href, mention: { start: 7, end: 19, text: "@diagram.png" }, }, ]) await transport.close() }) test("shows V2 blockers and replies through the runtime-owned session API", 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, }) events.push({ id: "evt_permission", created: 0, type: "permission.asked", data: { id: "per_1", sessionID: "ses_1", action: "read", resources: ["/tmp/file"] }, }) await Bun.sleep(0) expect(ui.events).toContainEqual({ type: "stream.view", view: { type: "permission", request: { id: "per_1", sessionID: "ses_1", action: "read", resources: ["/tmp/file"], }, }, }) await transport.close() }) 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], active: () => { const active: Record = {} if (running) active.ses_1 = { type: "running" } return active }, wait: () => idle.promise, }) let projected = false spyOn(client.message, "list").mockImplementation(() => ok({ data: projected ? [ { id: "msg_prompt", type: "user", text: "hello", files: [], agents: [], time: { created: 2 }, }, ] : [], cursor: {}, }), ) 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 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) 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 await transport.close() }) test("does not duplicate the optimistic user row when reconnect hydration recovers a missed prompt", async () => { const first = feed() const second = feed() first.push(connected("evt_connected_1")) second.push(connected("evt_connected_2")) let running = true let projected = false const client = sdk({ streams: [first, second], active: () => { const active: Record = {} if (running) active.ses_1 = { type: "running" } return active }, }) spyOn(client.message, "list").mockImplementation(() => ok({ data: projected ? [ { id: "msg_prompt", type: "user", text: "hello", files: [], agents: [], time: { created: 2 }, }, ] : [], cursor: {}, }), ) const ui = footer() ui.commits.push({ kind: "user", source: "system", text: "hello", phase: "start", messageID: "msg_prompt" }) 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 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) projected = true running = false first.close() await turn expect(ui.commits.filter((item) => item.kind === "user" && item.messageID === "msg_prompt")).toHaveLength(1) await transport.close() }) test("replaces the client for buffered hydration, descendants, turns, and interrupts", async () => { const firstEvents = feed() const secondEvents = feed() firstEvents.push(connected("evt_connected_1")) secondEvents.push(connected("evt_connected_2")) const first = sdk({ streams: [firstEvents] }) const second = sdk({ streams: [secondEvents], sessions: [{ id: "ses_child", parentID: "ses_1", title: "Child", time: { updated: 2 } }], forms: { ses_child: [form("frm_child", "ses_child")] }, }) const firstPrompt = spyOn(first.session, "prompt") const firstInterrupt = spyOn(first.session, "interrupt") spyOn(first.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_assistant", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [{ type: "text", text: "partial" }], time: { created: 1 }, }, ], cursor: {}, }), ) let releaseHydration!: () => void let replacementHydrating = false const hydration = new Promise((resolve) => { releaseHydration = resolve }) let releaseCatalog!: () => void let refreshes = 0 const catalog = new Promise((resolve) => { releaseCatalog = resolve }) spyOn(second.message, "list").mockImplementation(async (request) => { if (request.sessionID !== "ses_1") return ok({ data: [], cursor: {} }) replacementHydrating = true await hydration return ok({ data: [ { id: "msg_assistant", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [{ type: "text", text: "partial replacement" }], time: { created: 1 }, }, ], cursor: {}, }) }) const current: OpenCodeClient[] = [] const ui = footer() const transport = await createSessionTransport({ sdk: first, reconnect: async () => second, onClient: (client) => current.push(client), sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, onCatalogRefresh: () => { refreshes++ if (refreshes === 2) return catalog }, }) firstEvents.close() while (!replacementHydrating) await Bun.sleep(0) await expect( transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_blocked", text: "blocked", parts: [] }, files: [], includeFiles: true, }), ).rejects.toThrow("Event stream is reconnecting") secondEvents.push({ id: "evt_buffered_text", created: 2, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: " replacement", }, }) let resized = false const resize = transport.replayOnResize({ localRows: () => [], reset: async () => { resized = true }, }) releaseHydration() while ( !ui.events.some( (event) => event.type === "stream.view" && event.view.type === "form" && event.view.request.id === "frm_child", ) ) await Bun.sleep(0) while (refreshes < 2) await Bun.sleep(0) await resize expect(resized).toBe(false) await expect( transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_catalog_blocked", text: "blocked", parts: [] }, files: [], includeFiles: true, }), ).rejects.toThrow("Event stream is reconnecting") releaseCatalog() await Bun.sleep(0) expect(current).toEqual([second]) expect(first.event.subscribe).toHaveBeenCalledTimes(1) expect(second.event.subscribe).toHaveBeenCalledTimes(1) expect(second.session.list).toHaveBeenCalled() expect(second.form.list).toHaveBeenCalledWith({ sessionID: "ses_child" }, { signal: expect.any(AbortSignal) }) expect(ui.commits.filter((commit) => commit.messageID === "msg_assistant").map((commit) => commit.text)).toEqual([ "partial", " replacement", ]) const prompt = spyOn(second.session, "prompt").mockImplementation((request) => { queueMicrotask(() => { secondEvents.push({ id: "evt_replacement_prompt", created: 3, type: "session.input.promoted", durable: durable("ses_1", 1), data: { sessionID: "ses_1", inputID: "msg_replacement" }, }) secondEvents.push({ id: "evt_replacement_settled", created: 4, type: "session.execution.succeeded", durable: durable("ses_1", 2), data: { sessionID: "ses_1" }, }) }) return ok({ data: promptAdmission(request) }) as never }) await transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_replacement", text: "replacement prompt", parts: [] }, files: [], includeFiles: true, }) const interrupt = spyOn(second.session, "interrupt").mockImplementation(() => ok(undefined)) await transport.interruptActiveTurn() expect(prompt).toHaveBeenCalled() expect(interrupt).toHaveBeenCalledWith({ sessionID: "ses_1" }) expect(firstPrompt).not.toHaveBeenCalled() expect(firstInterrupt).not.toHaveBeenCalled() await transport.close() }) test("reconciles buffered deltas already present in a resize snapshot", 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, replay: true, footer: ui.api, }) spyOn(client.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_assistant", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [{ type: "text", text: "the answer" }], time: { created: 2, completed: 3 }, }, ], cursor: {}, }), ) let reset!: () => void const resetting = new Promise((resolve) => { reset = resolve }) const replay = transport.replayOnResize({ localRows: () => [], reset: () => resetting }) events.push({ id: "evt_text_started", created: 0, type: "session.text.started", durable: durable("ses_1"), data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, }, }) events.push({ id: "evt_text", created: 0, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: "answer", }, }) await Bun.sleep(0) reset() await replay expect(ui.commits.filter((item) => item.text === "the answer")).toHaveLength(1) expect(ui.commits.some((item) => item.text === "answer")).toBe(false) await transport.close() }) test("replays live assistant text missing from the resize projection", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) spyOn(client.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_assistant", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [{ type: "text", text: "partial" }], time: { created: 2, completed: 3 }, }, ], cursor: {}, }), ) const ui = footer() const live: StreamCommit[] = [] const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, onCommit: (commit) => live.push(commit), }) events.push({ id: "evt_text", created: 0, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: " suffix", }, }) await Bun.sleep(0) expect(live.map((commit) => commit.text)).toEqual(["partial suffix"]) await transport.replayOnResize({ localRows: () => [ { commit: live[0]! }, { commit: { ...live[0]!, partID: "text:1", text: "entirely local", }, }, ], reset: async () => {}, }) expect(ui.commits.filter((commit) => commit.messageID === "msg_assistant").map((commit) => commit.text)).toEqual([ "partial", " suffix", "partial", " suffix", "entirely local", ]) await transport.close() }) test("does not replay a resize-buffered suffix twice", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) spyOn(client.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_assistant", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [{ type: "text", text: "partial" }], time: { created: 2, completed: 3 }, }, ], cursor: {}, }), ) const ui = footer() const live: StreamCommit[] = [] const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, onCommit: (commit) => live.push(commit), }) let reset!: () => void const resetting = new Promise((resolve) => { reset = resolve }) const replay = transport.replayOnResize({ localRows: () => live.map((commit) => ({ commit })), reset: () => resetting, }) events.push({ id: "evt_text", created: 0, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: " suffix", }, }) await Bun.sleep(0) reset() await replay expect(ui.commits.filter((commit) => commit.messageID === "msg_assistant").map((commit) => commit.text)).toEqual([ "partial", "partial", " suffix", ]) expect(live.map((commit) => commit.text)).toEqual(["partial suffix"]) await transport.close() }) test("preserves active text and reasoning across resize before terminal projection", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) spyOn(client.message, "list").mockImplementation(() => ok({ data: [], cursor: {} })) const ui = footer() const live: StreamCommit[] = [] const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: true, replay: true, footer: ui.api, onCommit: (commit) => live.push(commit), }) events.push({ id: "evt_text", created: 0, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: "hello", }, }) events.push({ id: "evt_reasoning", created: 0, type: "session.reasoning.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: "thought", }, }) await Bun.sleep(0) expect(live.map((commit) => commit.text)).toEqual(["hello", "Thinking: thought"]) await transport.replayOnResize({ localRows: () => live.map((commit) => ({ commit })), reset: async () => {}, }) expect(ui.commits.slice(-2).map((commit) => commit.text)).toEqual(["hello", "Thinking: thought"]) await transport.close() }) test("serializes and coalesces overlapping resize replays", 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, replay: true, footer: ui.api, }) let release!: () => void const blocked = new Promise((resolve) => { release = resolve }) const order: string[] = [] const first = transport.replayOnResize({ localRows: () => [], reset: async () => { order.push("first:start") await blocked order.push("first:end") }, }) await Bun.sleep(0) const second = transport.replayOnResize({ localRows: () => [], reset: async () => { order.push("second") }, }) release() await Promise.all([first, second]) expect(second).toBe(first) expect(order).toEqual(["first:start", "first:end", "second"]) await transport.close() }) test("restores local output and drains buffered events when resize hydration fails", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) const ui = footer() const live: StreamCommit[] = [] const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, onCommit: (commit) => live.push(commit), }) events.push({ id: "evt_text_1", created: 0, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: "hello", }, }) await Bun.sleep(0) spyOn(client.message, "list").mockImplementation(() => Promise.reject(new Error("projection failed"))) const replay = transport.replayOnResize({ localRows: () => live.map((commit) => ({ commit })), reset: async () => {}, }) events.push({ id: "evt_text_2", created: 0, type: "session.text.delta", data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, delta: " world", }, }) await expect(replay).rejects.toThrow("projection failed") expect(ui.commits.slice(-2).map((commit) => commit.text)).toEqual(["hello", " world"]) expect(live.at(-1)?.text).toBe("hello world") await transport.close() }) test("dedupes a projected step failure from live redelivery", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) spyOn(client.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_assistant", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [], error: { type: "provider.transport", message: "provider failed" }, time: { created: 2, completed: 3 }, }, ], cursor: {}, }), ) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, }) events.push({ id: "evt_step_failed", created: 2, type: "session.step.failed", durable: durable("ses_1", 1), data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", error: { type: "provider.transport", message: "provider failed" }, }, }) await Bun.sleep(0) expect(ui.commits.filter((commit) => commit.kind === "error" && commit.text === "provider failed")).toHaveLength(1) await transport.close() }) test("dedupes a retained live step failure from resize projection", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) const ui = footer() const live: StreamCommit[] = [] const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, onCommit: (commit) => live.push(commit), }) events.push({ id: "evt_step_failed", created: 2, type: "session.step.failed", durable: durable("ses_1", 1), data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", error: { type: "provider.transport", message: "provider failed" }, }, }) await Bun.sleep(0) expect(live[0]?.messageID).toBe("msg_assistant") spyOn(client.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_assistant", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [], error: { type: "provider.transport", message: "provider failed" }, time: { created: 2, completed: 3 }, }, ], cursor: {}, }), ) await transport.replayOnResize({ localRows: () => live.map((commit) => ({ commit })), reset: async () => {}, }) expect(ui.commits.filter((commit) => commit.kind === "error" && commit.text === "provider failed")).toHaveLength(2) await transport.close() }) test("preserves an execution-only local error beside its projected prompt", 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, replay: true, footer: ui.api, }) spyOn(client.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_prompt", type: "user", text: "hello", files: [], agents: [], time: { created: 2 }, }, ], cursor: {}, }), ) await transport.replayOnResize({ localRows: () => [ { commit: { kind: "error", source: "system", text: "model unavailable", phase: "start", messageID: "msg_prompt", }, }, ], reset: async () => {}, }) expect(ui.commits.some((commit) => commit.kind === "error" && commit.text === "model unavailable")).toBe(true) await transport.close() }) test("scopes text and reasoning ordinals by assistant message", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) spyOn(client.message, "list").mockImplementation(() => ok({ data: [ { id: "msg_b", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [ { type: "reasoning", text: "second thought" }, { type: "text", text: "second answer" }, ], time: { created: 4, completed: 5 }, }, { id: "msg_a", type: "assistant", agent: "build", model: { providerID: "test", id: "model" }, content: [ { type: "reasoning", text: "first thought" }, { type: "text", text: "first answer" }, ], time: { created: 2, completed: 3 }, }, ], cursor: {}, }), ) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: true, replay: true, footer: ui.api, }) expect(ui.commits.map((item) => item.text)).toEqual([ "Thinking: first thought", "first answer", "Thinking: second thought", "second answer", ]) await transport.close() }) test("renders full reasoning when only the ended event 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: true, footer: ui.api, }) events.push({ id: "evt_reasoning", created: 0, type: "session.reasoning.ended", durable: durable("ses_1"), data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", ordinal: 0, text: "considering", }, }) await Bun.sleep(0) expect(ui.commits.at(-1)?.text).toBe("Thinking: considering") await transport.close() }) test("tracks repeated root call IDs independently across assistant messages", 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, }) for (const [index, messageID] of ["msg_tool_one", "msg_tool_two"].entries()) { events.push({ id: `evt_repeated_input_${index}`, created: index * 3 + 1, type: "session.tool.input.started", durable: durable("ses_1", index * 3), data: { sessionID: "ses_1", assistantMessageID: messageID, callID: "call_repeated", name: "read" }, }) events.push({ id: `evt_repeated_called_${index}`, created: index * 3 + 2, type: "session.tool.called", durable: durable("ses_1", index * 3 + 1), data: { sessionID: "ses_1", assistantMessageID: messageID, callID: "call_repeated", input: { path: `${index + 1}.txt` }, executed: true, }, }) events.push({ id: `evt_repeated_success_${index}`, created: index * 3 + 3, type: "session.tool.success", durable: durable("ses_1", index * 3 + 2, 2), data: { sessionID: "ses_1", assistantMessageID: messageID, callID: "call_repeated", metadata: {}, content: [{ type: "text", text: "" }], executed: true, }, }) } await Bun.sleep(0) const commits = ui.commits.filter((item) => item.part?.id === "call_repeated") expect(commits.map((item) => [item.messageID, item.phase])).toEqual([ ["msg_tool_one", "start"], ["msg_tool_one", "final"], ["msg_tool_two", "start"], ["msg_tool_two", "final"], ]) expect( commits .filter((item) => item.phase === "final") .map((item) => (item.part?.state.status === "streaming" ? undefined : item.part?.state.input)), ).toEqual([{ path: "1.txt" }, { path: "2.txt" }]) await transport.close() }) test("reduces root tool progress and preserves it on failure", 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, }) events.push({ id: "evt_progress_input", created: 1, type: "session.tool.input.started", durable: durable("ses_1"), data: { sessionID: "ses_1", assistantMessageID: "msg_progress", callID: "call_progress", name: "shell", }, }) events.push({ id: "evt_progress_called", created: 2, type: "session.tool.called", durable: durable("ses_1", 1), data: { sessionID: "ses_1", assistantMessageID: "msg_progress", callID: "call_progress", input: { command: "printf partial && false" }, executed: true, }, }) events.push({ id: "evt_progress", created: 3, type: "session.tool.progress", data: { sessionID: "ses_1", assistantMessageID: "msg_progress", callID: "call_progress", metadata: { checkpoint: 1 }, }, }) events.push({ id: "evt_progress_failed", created: 4, type: "session.tool.failed", durable: durable("ses_1", 3, 2), data: { sessionID: "ses_1", assistantMessageID: "msg_progress", callID: "call_progress", error: { type: "unknown", message: "boom" }, metadata: { checkpoint: 1 }, content: [{ type: "text", text: "partial" }], executed: true, }, }) await Bun.sleep(0) const commits = ui.commits.filter((item) => item.part?.id === "call_progress") expect(commits.map((item) => [item.phase, item.text, item.toolState])).toEqual([ ["start", "running shell", "running"], ["progress", "partial", "running"], ["final", "boom", "error"], ]) expect(commits.at(-1)?.part?.state).toMatchObject({ status: "error", metadata: { checkpoint: 1 }, content: [{ type: "text", text: "partial" }], }) 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()) const client = sdk({ streams: [events] }) const ui = footer() const transport = await createSessionTransport({ sdk: client, location: { directory: "/project", workspaceID: "wrk_1" }, sessionID: "ses_1", thinking: false, footer: ui.api, }) spyOn(client.session, "get").mockImplementation(() => ok({ model: undefined }) as never) const defaultModel = spyOn(client.model, "default").mockImplementation( () => ok({ location: { directory: "/tmp", project: { id: "proj_1", directory: "/tmp" } }, data: { id: "gpt-5", providerID: "openai" }, }) as never, ) const switched = spyOn(client.session, "switchModel").mockImplementation(() => ok(undefined)) 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 turn = transport.runPromptTurn({ agent: undefined, model: undefined, variant: "high", prompt: { messageID: "msg_prompt", text: "hello", parts: [] }, files: [], includeFiles: true, }) while (!admitted) await Bun.sleep(0) events.push({ id: "evt_prompted", created: 0, type: "session.input.promoted", durable: durable("ses_1"), data: { sessionID: "ses_1", inputID: "msg_prompt", }, }) events.push({ id: "evt_settled", created: 0, type: "session.execution.succeeded", durable: durable("ses_1"), data: { sessionID: "ses_1" }, }) await turn expect(switched).toHaveBeenCalledWith( { sessionID: "ses_1", model: { providerID: "openai", id: "gpt-5", variant: "high" } }, { signal: undefined }, ) expect(defaultModel).toHaveBeenCalledWith( { location: { directory: "/project", workspace: "wrk_1" } }, { signal: undefined }, ) await transport.close() }) test("interrupts the current Session when an active turn is aborted", async () => { const events = feed() events.push(connected()) const idle = defer() const client = sdk({ streams: [events], wait: () => idle.promise }) 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 controller = new AbortController() const turn = transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_prompt", text: "hello", parts: [] }, files: [], includeFiles: true, signal: controller.signal, }) while (!admitted) await Bun.sleep(0) events.push({ id: "evt_prompted", created: 0, type: "session.input.promoted", durable: durable("ses_1"), data: { sessionID: "ses_1", inputID: "msg_prompt", }, }) await Bun.sleep(0) controller.abort() idle.resolve() await turn expect(interrupted).toHaveBeenCalledWith({ sessionID: "ses_1" }) await transport.close() }) test("runs a shell turn through v2.session.shell and renders live output", 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 request: Parameters[0] | undefined spyOn(client.session, "shell").mockImplementation((input) => { request = input queueMicrotask(() => { events.push({ id: input.id ?? "evt_missing", created: 0, type: "session.shell.started", durable: durable("ses_1"), data: { sessionID: "ses_1", shell: { id: "sh_shell", status: "running", command: "ls", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/opencode-shell", metadata: {}, time: { started: 0 }, }, }, }) events.push({ id: "evt_shell_end", created: 0, type: "session.shell.ended", durable: durable("ses_1", 1), data: { sessionID: "ses_1", shell: { id: "sh_shell", status: "exited", command: "ls", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/opencode-shell", exit: 0, metadata: {}, time: { started: 0, completed: 1 }, }, output: { output: "file.txt", cursor: 8, size: 8, truncated: false }, }, }) }) return ok(undefined) as never }) await transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { text: "ls", parts: [], mode: "shell" }, files: [], includeFiles: true, }) expect(request).toMatchObject({ sessionID: "ses_1", command: "ls", id: expect.stringMatching(/^evt_/) }) expect(ui.commits.filter((item) => item.shell)).toMatchObject([ { phase: "start", partID: "shell:sh_shell", tool: "shell", toolState: "running", shell: { command: "ls" } }, { phase: "progress", partID: "shell:sh_shell", text: "file.txt", toolState: "completed", shell: { command: "ls" }, }, ]) expect(ui.events).toContainEqual({ type: "stream.patch", patch: { phase: "running", status: "running shell" } }) await transport.close() }) test("aborts an active shell turn without interrupting the session", 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 started = false let aborted = false spyOn(client.session, "shell").mockImplementation( (_input, options) => new Promise((_, reject) => { started = true options?.signal?.addEventListener("abort", () => { aborted = true reject(new Error("aborted")) }) }) as never, ) const interrupted = spyOn(client.session, "interrupt").mockImplementation(() => ok(undefined)) const turn = transport.runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { text: "sleep 100", parts: [], mode: "shell" }, files: [], includeFiles: true, }) while (!started) await Bun.sleep(0) await transport.interruptActiveTurn() await turn expect(aborted).toBe(true) expect(interrupted).not.toHaveBeenCalled() await transport.close() }) test("does not resolve an owned shell output wait from an unrelated shell", 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 request: Parameters[0] | undefined let complete!: () => void spyOn(client.session, "shell").mockImplementation((input) => { request = input return new Promise((resolve) => { complete = resolve }) as never }) let done = false const turn = transport .runPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { text: "pwd", parts: [], mode: "shell" }, files: [], includeFiles: true, }) .then(() => { done = true }) while (!request) await Bun.sleep(0) events.push({ id: "evt_unrelated_shell", created: 0, type: "session.shell.started", durable: durable("ses_1"), data: { sessionID: "ses_1", shell: { id: "sh_unrelated", status: "running", command: "other", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/unrelated", metadata: {}, time: { started: 0 }, }, }, }) events.push({ id: "evt_unrelated_end", created: 0, type: "session.shell.ended", durable: durable("ses_1", 1), data: { sessionID: "ses_1", shell: { id: "sh_unrelated", status: "exited", command: "other", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/unrelated", exit: 0, metadata: {}, time: { started: 0, completed: 1 }, }, output: { output: "wrong", cursor: 5, size: 5, truncated: false }, }, }) await Bun.sleep(0) complete() await Bun.sleep(0) expect(done).toBe(false) events.push({ id: request.id ?? "evt_missing", created: 0, type: "session.shell.started", durable: durable("ses_1", 2), data: { sessionID: "ses_1", shell: { id: "sh_owned", status: "running", command: "pwd", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/owned", metadata: {}, time: { started: 0 }, }, }, }) events.push({ id: "evt_owned_end", created: 0, type: "session.shell.ended", durable: durable("ses_1", 3), data: { sessionID: "ses_1", shell: { id: "sh_owned", status: "exited", command: "pwd", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/owned", exit: 0, metadata: {}, time: { started: 0, completed: 1 }, }, output: { output: "/tmp", cursor: 4, size: 4, truncated: false }, }, }) await turn expect(request.id).toMatch(/^evt_/) expect(ui.commits.some((item) => item.partID === "shell:sh_owned" && item.text === "/tmp")).toBe(true) await transport.close() }) test("hydrates projected shell transcripts once and dedupes live redelivery", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_1: [ { id: "msg_shell", type: "shell" as const, shellID: "sh_1", status: "exited", command: "ls", exit: 0, output: { output: "file.txt", cursor: 8, size: 8, truncated: false }, time: { created: 1, completed: 2 }, }, ], }, }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, }) events.push({ id: "evt_shell_end", created: 0, type: "session.shell.ended", durable: durable("ses_1", 1), data: { sessionID: "ses_1", shell: { id: "sh_1", status: "exited", command: "ls", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/opencode-shell", exit: 0, metadata: {}, time: { started: 0, completed: 1 }, }, output: { output: "file.txt", cursor: 8, size: 8, truncated: false }, }, }) await Bun.sleep(0) await Bun.sleep(0) expect(ui.commits.filter((item) => item.shell)).toMatchObject([ { phase: "start", partID: "shell:sh_1", shell: { command: "ls" } }, { phase: "progress", partID: "shell:sh_1", text: "file.txt", toolState: "completed" }, ]) await transport.close() }) test("renders failed projected shells as errors and marks truncated live output", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_1: [ { id: "msg_failed_shell", type: "shell" as const, shellID: "sh_failed", status: "exited", command: "false", exit: 7, output: { output: "failure output", cursor: 14, size: 14, truncated: false }, time: { created: 1, completed: 2 }, }, ], }, }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, }) events.push({ id: "evt_truncated_start", created: 0, type: "session.shell.started", durable: durable("ses_1"), data: { sessionID: "ses_1", shell: { id: "sh_truncated", status: "running", command: "long", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/truncated", metadata: {}, time: { started: 0 }, }, }, }) events.push({ id: "evt_truncated_end", created: 0, type: "session.shell.ended", durable: durable("ses_1", 1), data: { sessionID: "ses_1", shell: { id: "sh_truncated", status: "exited", command: "long", cwd: "/tmp", shell: "/bin/sh", file: "/tmp/truncated", exit: 0, metadata: {}, time: { started: 0, completed: 1 }, }, output: { output: "partial", cursor: 7, size: 20, truncated: false }, }, }) await Bun.sleep(0) expect(ui.commits).toContainEqual( expect.objectContaining({ toolState: "error", toolError: "Shell exited with code 7" }), ) expect(ui.commits).toContainEqual(expect.objectContaining({ text: "partial\n[output truncated]" })) await transport.close() }) test("routes command prompts through v2.session.command", 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 request: Parameters[0] | undefined spyOn(client.session, "command").mockImplementation((input) => { request = input queueMicrotask(() => { events.push({ id: "evt_prompted", created: 0, type: "session.input.promoted", durable: durable("ses_1"), data: { sessionID: "ses_1", inputID: "msg_cmd", }, }) events.push({ id: "evt_settled", created: 0, type: "session.execution.succeeded", durable: durable("ses_1"), data: { sessionID: "ses_1" }, }) }) return ok({ admittedSeq: 1, id: input.id ?? "msg_cmd", sessionID: "ses_1", type: "user" as const, data: { text: "evaluated template" }, delivery: "steer" as const, timeCreated: 2, }) }) await transport.runPromptTurn({ agent: "build", model: { providerID: "test", modelID: "model" }, variant: undefined, prompt: { messageID: "msg_cmd", text: "/deploy prod", parts: [ { type: "file", url: "file:///tmp/mentioned.txt", filename: "mentioned.txt", source: { type: "file", text: { start: 8, end: 12, value: "prod" } }, }, ], command: { name: "deploy", arguments: "prod" }, }, files: [{ type: "file", url: "file:///tmp/context.txt", filename: "context.txt", mime: "text/plain" }], includeFiles: true, }) expect(request).toMatchObject({ sessionID: "ses_1", id: "msg_cmd", command: "deploy", arguments: "prod", agent: "build", model: { providerID: "test", id: "model" }, files: [ { uri: "file:///tmp/context.txt", name: "context.txt" }, { uri: "file:///tmp/mentioned.txt", name: "mentioned.txt", mention: { start: 8, end: 12, text: "prod" }, }, ], delivery: "steer", }) // Selection rides the command payload; no separate client-side switch. expect(client.session.switchAgent).not.toHaveBeenCalled() expect(client.session.switchModel).not.toHaveBeenCalled() await transport.close() }) test("routes skill prompts through v2.session.skill and settles without promotion", 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 request: Parameters[0] | undefined const command = spyOn(client.session, "command") const prompt = spyOn(client.session, "prompt") spyOn(client.session, "skill").mockImplementation((input) => { request = input queueMicrotask(() => { events.push({ id: "evt_skill", created: 0, type: "session.skill.activated", durable: durable("ses_1"), data: { sessionID: "ses_1", id: input.skill ?? "tigerstyle", name: input.skill ?? "tigerstyle", text: "skill instructions", }, }) events.push({ id: "evt_settled", created: 0, type: "session.execution.succeeded", durable: durable("ses_1"), data: { sessionID: "ses_1" }, }) }) return ok(undefined) as never }) await transport.runPromptTurn({ agent: "review", model: undefined, variant: undefined, prompt: { messageID: "msg_skill", text: "/tigerstyle", parts: [], command: { name: "tigerstyle", arguments: "", source: "skill" }, }, files: [], includeFiles: true, }) expect(client.session.switchAgent).toHaveBeenCalledWith({ sessionID: "ses_1", agent: "review" }, expect.anything()) expect(request).toMatchObject({ sessionID: "ses_1", id: "msg_skill", skill: "tigerstyle" }) expect(command).not.toHaveBeenCalled() expect(prompt).not.toHaveBeenCalled() expect(ui.commits).toContainEqual( expect.objectContaining({ kind: "system", text: '→ Skill "tigerstyle"', messageID: "msg_skill" }), ) await transport.close() }) test("refreshes catalogs on connection and location-scoped invalidations", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) const ui = footer() let refreshes = 0 const transport = await createSessionTransport({ sdk: client, location: { directory: "/project", workspaceID: "work-1", }, sessionID: "ses_1", thinking: false, footer: ui.api, onCatalogRefresh: () => refreshes++, }) expect(refreshes).toBe(1) for (const type of [ "catalog.updated", "integration.updated", "agent.updated", "command.updated", "skill.updated", "reference.updated", ] as const) events.push({ id: `evt_${type}`, created: 0, type, location: { directory: "/project", workspaceID: "work-1" }, data: {}, }) events.push({ id: "evt_foreign_catalog", created: 0, type: "catalog.updated", location: { directory: "/other" }, data: {}, }) events.push({ id: "evt_foreign_workspace_catalog", created: 0, type: "catalog.updated", location: { directory: "/project", workspaceID: "work-2" }, data: {}, }) while (refreshes < 7) await Bun.sleep(0) await Bun.sleep(0) expect(refreshes).toBe(7) await transport.close() }) test("hydrates skill activation messages once and dedupes live redelivery", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_1: [ { id: "msg_skill", type: "skill" as const, skill: "tigerstyle", name: "tigerstyle", text: "skill instructions", time: { created: 2 }, }, ], }, }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, replay: true, footer: ui.api, }) 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", }, }) await Bun.sleep(0) await Bun.sleep(0) expect(ui.commits.filter((item) => item.text === '→ Skill "tigerstyle"')).toHaveLength(1) await transport.close() }) test("discovers a subagent from its terminal failure snapshot", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_child_failed: [] } }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) events.push({ id: "evt_failed_subagent_input", created: 1, type: "session.tool.input.started", durable: durable("ses_1"), data: { sessionID: "ses_1", assistantMessageID: "msg_failed_subagent", callID: "call_failed_subagent", name: "subagent", }, }) events.push({ id: "evt_failed_subagent_called", created: 2, type: "session.tool.called", durable: durable("ses_1", 1), data: { sessionID: "ses_1", assistantMessageID: "msg_failed_subagent", callID: "call_failed_subagent", input: { agent: "explore", description: "Inspect failure", prompt: "inspect" }, executed: true, }, }) events.push({ id: "evt_failed_subagent", created: 3, type: "session.tool.failed", durable: durable("ses_1", 2, 2), data: { sessionID: "ses_1", assistantMessageID: "msg_failed_subagent", callID: "call_failed_subagent", error: { type: "unknown", message: "subagent failed" }, metadata: { sessionID: "ses_child_failed", status: "running" }, executed: true, }, }) while (!states().some((state) => state.tabs.some((tab) => tab.sessionID === "ses_child_failed"))) await Bun.sleep(0) expect(states().at(-1)?.tabs).toMatchObject([ { sessionID: "ses_child_failed", label: "Explore", description: "Inspect failure", }, ]) await transport.close() }) test("discovers current subagents from progress and reduces descendant tool state", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_child_progress: [] } }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) events.push({ id: "evt_subagent_input", created: 1, type: "session.tool.input.started", durable: durable("ses_1"), data: { sessionID: "ses_1", assistantMessageID: "msg_subagent", callID: "call_subagent", name: "subagent", }, }) events.push({ id: "evt_subagent_called", created: 2, type: "session.tool.called", durable: durable("ses_1", 1), data: { sessionID: "ses_1", assistantMessageID: "msg_subagent", callID: "call_subagent", input: { agent: "explore", description: "Inspect progress", prompt: "inspect" }, executed: true, }, }) events.push({ id: "evt_subagent_progress", created: 3, type: "session.tool.progress", data: { sessionID: "ses_1", assistantMessageID: "msg_subagent", callID: "call_subagent", metadata: { sessionID: "ses_child_progress", status: "running" }, }, }) while (!states().some((state) => state.tabs.some((tab) => tab.sessionID === "ses_child_progress"))) await Bun.sleep(0) expect(states().at(-1)?.tabs).toMatchObject([ { sessionID: "ses_child_progress", label: "Explore", description: "Inspect progress", status: "running", background: undefined, }, ]) transport.selectSubagent("ses_child_progress") while (!states().at(-1)?.details.ses_child_progress) await Bun.sleep(0) events.push({ id: "evt_child_tool_input", created: 4, type: "session.tool.input.started", durable: durable("ses_child_progress"), data: { sessionID: "ses_child_progress", assistantMessageID: "msg_child_tool", callID: "call_child_shell", name: "shell", }, }) events.push({ id: "evt_child_tool_called", created: 5, type: "session.tool.called", durable: durable("ses_child_progress", 1), data: { sessionID: "ses_child_progress", assistantMessageID: "msg_child_tool", callID: "call_child_shell", input: { command: "printf child && false" }, executed: true, }, }) events.push({ id: "evt_child_tool_progress", created: 6, type: "session.tool.progress", data: { sessionID: "ses_child_progress", assistantMessageID: "msg_child_tool", callID: "call_child_shell", metadata: { checkpoint: "child" }, }, }) events.push({ id: "evt_child_permission", created: 7, type: "permission.asked", data: { id: "per_child", sessionID: "ses_child_progress", action: "shell", resources: ["printf child && false"], source: { type: "tool", messageID: "msg_child_tool", callID: "call_child_shell" }, }, }) events.push({ id: "evt_child_tool_failed", created: 8, type: "session.tool.failed", durable: durable("ses_child_progress", 3, 2), data: { sessionID: "ses_child_progress", assistantMessageID: "msg_child_tool", callID: "call_child_shell", error: { type: "unknown", message: "child boom" }, metadata: { checkpoint: "child" }, content: [{ type: "text", text: "child partial" }], executed: true, }, }) while ( !states() .at(-1) ?.details.ses_child_progress?.commits.some( (item) => item.part?.id === "call_child_shell" && item.toolState === "error", ) ) await Bun.sleep(0) const commits = states().at(-1)?.details.ses_child_progress?.commits ?? [] expect( commits .filter((item) => item.part?.id === "call_child_shell") .map((item) => [item.phase, item.text, item.toolState]), ).toEqual([ ["progress", "child partial", "running"], ["final", "child boom", "error"], ]) expect( commits.find((item) => item.part?.id === "call_child_shell" && item.toolState === "error")?.part?.state, ).toMatchObject({ status: "error", metadata: { checkpoint: "child" }, content: [{ type: "text", text: "child partial" }], }) expect( ui.events.find( (event) => event.type === "stream.view" && event.view.type === "permission" && event.view.request.id === "per_child", ), ).toMatchObject({ view: { request: { sessionID: "ses_child_progress", tool: { id: "call_child_shell", state: { input: { command: "printf child && false" } } }, }, }, }) await transport.close() }) test("discovers a live child session and tracks its tab and selected detail", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_child: [ { id: "msg_task", type: "user" as const, text: "task prompt", files: [], agents: [], time: { created: 1 }, }, { id: "msg_child_a", type: "assistant" as const, agent: "explore", model: { providerID: "test", id: "model" }, content: [{ type: "text" as const, text: "child answer" }], time: { created: 2 }, }, ], }, }) spyOn(client.session, "get").mockImplementation( () => ok({ id: "ses_child", parentID: "ses_1", projectID: "proj_1", agent: "explore", cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, time: { created: 1, updated: 1 }, title: "Find files", location: { directory: "/tmp" }, }) as never, ) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) transport.selectSubagent("ses_child") events.push({ id: "evt_child_step", created: 0, type: "session.step.started", durable: durable("ses_child"), data: { sessionID: "ses_child", assistantMessageID: "msg_child_a", agent: "explore", model: { providerID: "test", id: "model" }, }, }) while (!states().some((state) => state.details.ses_child?.commits.some((item) => item.text === "task prompt"))) await Bun.sleep(0) expect(states().at(-1)?.tabs).toMatchObject([ { sessionID: "ses_child", label: "Explore", title: "Find files", status: "running" }, ]) expect( states() .at(-1) ?.details.ses_child?.commits.filter((item) => item.text === "child answer"), ).toHaveLength(1) events.push({ id: "evt_child_text_replayed", created: 0, type: "session.text.delta", data: { sessionID: "ses_child", assistantMessageID: "msg_child_a", ordinal: 0, delta: "answer", }, }) await Bun.sleep(0) expect( states() .at(-1) ?.details.ses_child?.commits.filter((item) => item.text === "child answer"), ).toHaveLength(1) events.push({ id: "evt_child_text_suffix", created: 0, type: "session.text.delta", data: { sessionID: "ses_child", assistantMessageID: "msg_child_a", ordinal: 0, delta: " suffix", }, }) while ( !states().some((state) => state.details.ses_child?.commits.some((item) => item.text === "child answer suffix")) ) await Bun.sleep(0) events.push({ id: "evt_child_settled", created: 0, type: "session.execution.succeeded", durable: durable("ses_child"), data: { sessionID: "ses_child" }, }) while (!states().some((state) => state.tabs.some((tab) => tab.status === "completed"))) await Bun.sleep(0) await transport.close() }) test("reveals an admitted child prompt only when it is promoted after hydration", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_child: [] }, sessions: [{ id: "ses_child", parentID: "ses_1", time: { updated: 1 } }], }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) transport.selectSubagent("ses_child") while (!states().some((state) => state.details.ses_child)) await Bun.sleep(0) events.push({ id: "evt_child_admitted", created: 1, type: "session.input.admitted", durable: durable("ses_child"), data: { sessionID: "ses_child", inputID: "msg_child_prompt", input: { type: "user", data: { text: "actual child prompt" }, delivery: "steer" }, }, }) await Bun.sleep(0) expect( states() .at(-1) ?.details.ses_child?.commits.some((item) => item.messageID === "msg_child_prompt"), ).toBe(false) events.push({ id: "evt_child_promoted", created: 2, type: "session.input.promoted", durable: durable("ses_child", 1), data: { sessionID: "ses_child", inputID: "msg_child_prompt" }, }) while ( !states() .at(-1) ?.details.ses_child?.commits.some( (item) => item.messageID === "msg_child_prompt" && item.text === "actual child prompt", ) ) await Bun.sleep(0) await transport.close() }) test("preserves a pre-hydration admission promoted during stale hydration", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], sessions: [{ id: "ses_child", parentID: "ses_1", time: { updated: 1 } }], }) let childHydrating = false let releaseHydration!: () => void const hydration = new Promise((resolve) => { releaseHydration = resolve }) spyOn(client.message, "list").mockImplementation(async (request) => { if (request.sessionID === "ses_child") { childHydrating = true await hydration } return ok({ data: [], cursor: {} }) }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) events.push({ id: "evt_child_admitted_race", created: 1, type: "session.input.admitted", durable: durable("ses_child"), data: { sessionID: "ses_child", inputID: "msg_child_race", input: { type: "user", data: { text: "prompt admitted before hydration" }, delivery: "steer" }, }, }) await Bun.sleep(0) transport.selectSubagent("ses_child") while (!childHydrating) await Bun.sleep(0) events.push({ id: "evt_child_promoted_race", created: 2, type: "session.input.promoted", durable: durable("ses_child", 1), data: { sessionID: "ses_child", inputID: "msg_child_race" }, }) await Bun.sleep(0) releaseHydration() await Bun.sleep(0) await Bun.sleep(0) while ( !states() .at(-1) ?.details.ses_child?.commits.some( (item) => item.messageID === "msg_child_race" && item.text === "prompt admitted before hydration", ) ) await Bun.sleep(0) await transport.close() }) test("retries child hydration after a bounded live-event overflow", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], sessions: [{ id: "ses_child", parentID: "ses_1", time: { updated: 1 } }], }) let childRequests = 0 let releaseStale!: () => void let releaseRetry!: () => void const stale = new Promise((resolve) => { releaseStale = resolve }) const retry = new Promise((resolve) => { releaseRetry = resolve }) spyOn(client.message, "list").mockImplementation(async (request) => { if (request.sessionID !== "ses_child") return ok({ data: [], cursor: {} }) childRequests++ if (childRequests === 1) { await stale return ok({ data: [], cursor: {} }) } await retry return ok({ data: [ { id: "msg_overflow_assistant", type: "assistant" as const, agent: "explore", model: { providerID: "test", id: "model" }, content: [{ type: "text" as const, id: "txt_overflow_64", text: "live 64" }], time: { created: 2, completed: 3 }, }, { id: "msg_overflow_baseline", type: "user" as const, text: "baseline history", files: [], agents: [], time: { created: 1 }, }, ], cursor: {}, }) }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) transport.selectSubagent("ses_child") while (childRequests < 1) await Bun.sleep(0) for (let index = 0; index < 65; index++) events.push({ id: `evt_overflow_${index}`, created: index, type: "session.text.delta", data: { sessionID: "ses_child", assistantMessageID: "msg_overflow_assistant", ordinal: index, delta: `live ${index}`, }, }) while ( !states() .at(-1) ?.details.ses_child?.commits.some((item) => item.text === "live 64") ) await Bun.sleep(0) releaseStale() while (childRequests < 2) await Bun.sleep(0) expect( states() .at(-1) ?.details.ses_child?.commits.some((item) => item.text === "live 64"), ).toBe(true) releaseRetry() while ( !states() .at(-1) ?.details.ses_child?.commits.some((item) => item.text === "baseline history") ) await Bun.sleep(0) expect( states() .at(-1) ?.details.ses_child?.commits.some((item) => item.text === "live 64"), ).toBe(true) expect(childRequests).toBe(2) await transport.close() }) test("reconciles pre-hydration tool metadata without downgrading projected completion", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], sessions: [{ id: "ses_child", parentID: "ses_1", time: { updated: 1 } }], }) let childHydrating = false let releaseHydration!: () => void const hydration = new Promise((resolve) => { releaseHydration = resolve }) spyOn(client.message, "list").mockImplementation(async (request) => { if (request.sessionID !== "ses_child") return ok({ data: [], cursor: {} }) childHydrating = true await hydration return ok({ data: [ { id: "msg_tool_projected", type: "assistant" as const, agent: "explore", model: { providerID: "test", id: "model" }, content: [ { type: "tool" as const, id: "call_overlap", name: "shell", state: { status: "completed" as const, input: { command: "projected" }, content: [{ type: "text" as const, text: "projected result" }], metadata: {}, }, time: { created: 1, ran: 1, completed: 2 }, }, ], time: { created: 1, completed: 2 }, }, ], cursor: {}, }) }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) const inputStarted = (callID: string, name: string, seq: number) => events.push({ id: `evt_started_${callID}`, created: seq, type: "session.tool.input.started", durable: durable("ses_child", seq), data: { sessionID: "ses_child", assistantMessageID: "msg_tool_projected", callID, name }, }) const called = (callID: string, input: Record, seq: number) => events.push({ id: `evt_called_${callID}`, created: seq, type: "session.tool.called", durable: durable("ses_child", seq), data: { sessionID: "ses_child", assistantMessageID: "msg_tool_projected", callID, input, executed: true, }, }) inputStarted("call_terminal", "grep", 0) called("call_terminal", { pattern: "needle" }, 1) await Bun.sleep(0) transport.selectSubagent("ses_child") while (!childHydrating) await Bun.sleep(0) events.push({ id: "evt_success_terminal", created: 2, type: "session.tool.success", durable: durable("ses_child", 2, 2), data: { sessionID: "ses_child", assistantMessageID: "msg_tool_projected", callID: "call_terminal", metadata: {}, content: [{ type: "text", text: "found" }], executed: true, }, }) inputStarted("call_overlap", "shell", 3) called("call_overlap", { command: "stale" }, 4) await Bun.sleep(0) const beforeHydration = states().length releaseHydration() while (states().length === beforeHydration) await Bun.sleep(0) await Bun.sleep(0) const commits = states().at(-1)?.details.ses_child?.commits ?? [] expect(commits.find((item) => item.partID === "prt_call_terminal")).toMatchObject({ tool: "grep", toolState: "completed", part: { state: { input: { pattern: "needle" } } }, }) expect(commits.find((item) => item.partID === "prt_call_overlap")).toMatchObject({ tool: "shell", toolState: "completed", part: { state: { input: { command: "projected" } } }, }) await transport.close() }) test("keeps child terminal state observed during discovery", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) let resolveGet: (() => void) | undefined const gate = new Promise((resolve) => { resolveGet = resolve }) spyOn(client.session, "get").mockImplementation(async () => { await gate return ok({ id: "ses_child", parentID: "ses_1", projectID: "proj_1", agent: "explore", cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, time: { created: 1, updated: 1 }, title: "Find files", location: { directory: "/tmp" }, }) as never }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) // Both events arrive while session.get is still in flight. events.push({ id: "evt_child_step", created: 0, type: "session.step.started", durable: durable("ses_child"), data: { sessionID: "ses_child", assistantMessageID: "msg_child_a", agent: "explore", model: { providerID: "test", id: "model" }, }, }) events.push({ id: "evt_child_settled", created: 0, type: "session.execution.interrupted", durable: durable("ses_child"), data: { sessionID: "ses_child", reason: "user" }, }) await Bun.sleep(0) resolveGet?.() while (!states().some((state) => state.tabs.some((tab) => tab.status === "cancelled"))) await Bun.sleep(0) await transport.close() }) test("does not resurrect a settled child from stale discovery buffer", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events] }) let resolveGet: (() => void) | undefined const gate = new Promise((resolve) => { resolveGet = resolve }) spyOn(client.session, "get").mockImplementation(async () => { await gate return ok({ id: "ses_child", parentID: "ses_1", projectID: "proj_1", agent: "explore", cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, time: { created: 1, updated: 1 }, title: "Find files", location: { directory: "/tmp" }, }) as never }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) // Child event arrives first and gets buffered behind the gated session.get. events.push({ id: "evt_child_step", created: 0, type: "session.step.started", durable: durable("ses_child"), data: { sessionID: "ses_child", assistantMessageID: "msg_child_a", agent: "explore", model: { providerID: "test", id: "model" }, }, }) // Parent's background subagent tool.success adopts the child mid-discovery. events.push({ id: "evt_parent_input", created: 0, type: "session.tool.input.started", durable: durable("ses_1"), data: { sessionID: "ses_1", assistantMessageID: "msg_parent_a", callID: "call_sub", name: "subagent", }, }) events.push({ id: "evt_parent_call", created: 0, type: "session.tool.called", durable: durable("ses_1"), data: { sessionID: "ses_1", assistantMessageID: "msg_parent_a", callID: "call_sub", input: { agent: "explore", description: "Find things", prompt: "go", background: true }, executed: true, }, }) events.push({ id: "evt_parent_success", created: 0, type: "session.tool.success", durable: durable("ses_1", 1, 2), data: { sessionID: "ses_1", assistantMessageID: "msg_parent_a", callID: "call_sub", metadata: { sessionID: "ses_child", status: "running", output: "" }, content: [{ type: "text", text: "" }], executed: true, }, }) // The settled event arrives after adoption, so it applies directly. events.push({ id: "evt_child_settled", created: 0, type: "session.execution.interrupted", durable: durable("ses_child"), data: { sessionID: "ses_child", reason: "shutdown" }, }) while (!states().some((state) => state.tabs.some((tab) => tab.status === "cancelled"))) await Bun.sleep(0) // Resolving discovery must not replay the buffered step.started over the // terminal status. const before = states().length resolveGet?.() while (states().length === before) await Bun.sleep(0) await Bun.sleep(0) await Bun.sleep(0) expect(states().at(-1)?.tabs).toMatchObject([{ sessionID: "ses_child", status: "cancelled" }]) await transport.close() }) test("adopts historical children from the session family list", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], sessions: [ { id: "ses_child_old", parentID: "ses_1", title: "Earlier subagent", agent: "explore", time: { updated: 9 } }, { id: "ses_unrelated", title: "Different session", time: { updated: 5 } }, { id: "ses_sibling", parentID: "ses_2", title: "Someone else's child", time: { updated: 4 } }, ], }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) expect(client.session.list).toHaveBeenCalledWith( { parentID: "ses_1", limit: 100, order: "desc" }, { signal: expect.any(AbortSignal) }, ) expect(states.at(-1)?.tabs).toMatchObject([ { sessionID: "ses_child_old", label: "Explore", title: "Earlier subagent", status: "completed", }, ]) await transport.close() }) test("hydrates completed subagent children from projected tool output", async () => { const events = feed() events.push(connected()) const client = sdk({ streams: [events], messages: { ses_1: [ { id: "msg_parent", type: "assistant" as const, agent: "build", model: { providerID: "test", id: "model" }, time: { created: 1, completed: 3 }, content: [ { type: "tool" as const, id: "call_sub", name: "subagent", state: { status: "completed" as const, input: { agent: "explore", description: "Find things", prompt: "go" }, content: [{ type: "text" as const, text: "done" }], metadata: { sessionID: "ses_child", status: "completed", output: "done" }, }, time: { created: 1, ran: 1, completed: 2 }, }, ], }, ], }, }) const ui = footer() const transport = await createSessionTransport({ sdk: client, sessionID: "ses_1", thinking: false, footer: ui.api, }) const states = ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : [])) expect(states.at(-1)?.tabs).toMatchObject([ { sessionID: "ses_child", label: "Explore", description: "Find things", status: "completed", }, ]) await transport.close() }) })