From 57fb3e5cc585a4d26297db61635703a610abf99f Mon Sep 17 00:00:00 2001 From: Simon Klee Date: Sat, 4 Jul 2026 21:14:01 +0200 Subject: [PATCH] fix(run): align mini with current session contracts (#35354) --- packages/core/src/session.ts | 14 +- packages/core/test/session-skill.test.ts | 70 +++ .../opencode/src/cli/cmd/run/runtime.boot.ts | 4 + packages/opencode/src/cli/cmd/run/runtime.ts | 116 ++-- .../src/cli/cmd/run/stream-v2.subagent.ts | 49 +- .../src/cli/cmd/run/stream-v2.transport.ts | 88 ++- packages/opencode/src/cli/cmd/run/tool.ts | 9 + .../opencode/test/cli/run/entry.body.test.ts | 17 + .../opencode/test/cli/run/runtime.test.ts | 187 +++++- .../test/cli/run/stream-v2.transport.test.ts | 589 +++++++++++++++++- 10 files changed, 1063 insertions(+), 80 deletions(-) create mode 100644 packages/core/test/session-skill.test.ts diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 05d64c8efb..57ae67ce34 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -585,11 +585,15 @@ const layer = Layer.effect( const skills = yield* SkillV2.Service.pipe(Effect.provide(locations.get(session.location))) const skill = (yield* skills.list()).find((item) => item.name === input.skill) if (!skill) return yield* new SkillNotFoundError({ skill: input.skill }) - yield* events.publish(SessionEvent.Skill.Activated, { - sessionID: input.sessionID, - name: skill.name, - text: skill.content, - }) + yield* events.publish( + SessionEvent.Skill.Activated, + { + sessionID: input.sessionID, + name: skill.name, + text: skill.content, + }, + { id: input.id ? EventV2.ID.make(input.id.replace(/^msg_/, "evt_")) : undefined }, + ) if (input.resume !== false) yield* execution .resume(input.sessionID) diff --git a/packages/core/test/session-skill.test.ts b/packages/core/test/session-skill.test.ts new file mode 100644 index 0000000000..8f9803ef4a --- /dev/null +++ b/packages/core/test/session-skill.test.ts @@ -0,0 +1,70 @@ +import path from "path" +import { describe, expect } from "bun:test" +import { Effect, Layer, LayerMap } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2 } from "@opencode-ai/core/event" +import { Location } from "@opencode-ai/core/location" +import { LocationServiceMap } from "@opencode-ai/core/location-service-map" +import type { LocationServices } from "@opencode-ai/core/location-services" +import { ProjectV2 } from "@opencode-ai/core/project" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionStore } from "@opencode-ai/core/session/store" +import { SkillV2 } from "@opencode-ai/core/skill" +import { testEffect } from "./lib/effect" + +const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) +const projects = Layer.mock(ProjectV2.Service, { + resolve: (directory) => Effect.succeed({ id: ProjectV2.ID.global, directory }), +}) +const skills = Layer.mock(SkillV2.Service, { + list: () => + Effect.succeed([ + SkillV2.Info.make({ + name: "effect", + description: "Effect guidance", + location: AbsolutePath.make(path.resolve("/skills/effect/SKILL.md")), + content: "Use Effect", + }), + ]), +}) +const locations = Layer.effect( + LocationServiceMap.Service, + LayerMap.make( + () => + // The skill endpoint only needs the location-scoped Skill service. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + skills as unknown as Layer.Layer, + ), +) +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([Database.node, EventV2.node, SessionProjector.node, SessionStore.node, SessionV2.node]), + [ + [LocationServiceMap.node, locations], + [ProjectV2.node, projects], + [SessionExecution.node, SessionExecution.noopLayer], + ], + ), +) + +describe("SessionV2.skill", () => { + it.effect("projects the caller-supplied message ID", () => + Effect.gen(function* () { + const sessions = yield* SessionV2.Service + const session = yield* sessions.create({ location }) + const id = SessionMessage.ID.make("msg_caller_skill") + + yield* sessions.skill({ id, sessionID: session.id, skill: "effect", resume: false }) + + expect(yield* sessions.messages({ sessionID: session.id })).toContainEqual( + expect.objectContaining({ id, type: "skill", name: "effect", text: "Use Effect" }), + ) + }), + ) +}) diff --git a/packages/opencode/src/cli/cmd/run/runtime.boot.ts b/packages/opencode/src/cli/cmd/run/runtime.boot.ts index 4753adaae2..2933ce5fdd 100644 --- a/packages/opencode/src/cli/cmd/run/runtime.boot.ts +++ b/packages/opencode/src/cli/cmd/run/runtime.boot.ts @@ -174,6 +174,10 @@ export async function resolveModelInfo( return runtime.runPromise((svc) => svc.resolveModelInfo(sdk, directory, model)).catch(() => emptyModelInfo()) } +export function resolveModelInfoStrict(sdk: RunInput["sdk"], directory: string, model: RunInput["model"]) { + return runtime.runPromise((svc) => svc.resolveModelInfo(sdk, directory, model)) +} + // Fetches session messages to determine if this is the first turn and build prompt history. export async function resolveSessionInfo( sdk: RunInput["sdk"], diff --git a/packages/opencode/src/cli/cmd/run/runtime.ts b/packages/opencode/src/cli/cmd/run/runtime.ts index e340a89cc3..5c6ede1582 100644 --- a/packages/opencode/src/cli/cmd/run/runtime.ts +++ b/packages/opencode/src/cli/cmd/run/runtime.ts @@ -17,7 +17,7 @@ import { Flag } from "@opencode-ai/core/flag/flag" import { MessageID } from "@/session/schema" import { loadRunAgents, loadRunCommands, loadRunReferences } from "./catalog.shared" import { createRunDemo } from "./demo" -import { resolveModelInfo, resolveRunTuiConfig, resolveSessionInfo } from "./runtime.boot" +import { resolveModelInfo, resolveModelInfoStrict, resolveRunTuiConfig, resolveSessionInfo } from "./runtime.boot" import { createRuntimeLifecycle } from "./runtime.lifecycle" import { trace } from "./trace" import { cycleVariant, formatModelLabel, resolveSavedVariant, resolveVariant, saveVariant } from "./variant.shared" @@ -378,32 +378,89 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep state.localRows = [...state.localRows, { commit, after }].slice(-LOCAL_REPLAY_ROW_LIMIT) } - const loadCatalog = async (): Promise => { + const applyCatalog = (catalog: { + agents: Awaited> + references: Awaited> + commands: Awaited> + }) => { if (footer.isClosed) { return } - - const [agents, references, commands] = await Promise.all([ - loadRunAgents(ctx.sdk, ctx.directory).catch(() => []), - loadRunReferences(ctx.sdk, ctx.directory).catch(() => []), - loadRunCommands(ctx.sdk, ctx.directory).catch(() => []), - ]) - if (footer.isClosed) { - return - } - footer.event({ type: "catalog", - agents, - references, - commands, + agents: catalog.agents, + references: catalog.references, + commands: catalog.commands, }) } - void footer + const fetchCatalog = async () => { + const [agents, references, commands] = await Promise.all([ + loadRunAgents(ctx.sdk, ctx.directory), + loadRunReferences(ctx.sdk, ctx.directory), + loadRunCommands(ctx.sdk, ctx.directory), + ]) + return { agents, references, commands } + } + + const loadCatalog = async () => { + applyCatalog( + await Promise.all([ + loadRunAgents(ctx.sdk, ctx.directory).catch(() => []), + loadRunReferences(ctx.sdk, ctx.directory).catch(() => []), + loadRunCommands(ctx.sdk, ctx.directory).catch(() => []), + ]).then(([agents, references, commands]) => ({ agents, references, commands })), + ) + } + + const applyModelInfo = ( + info: Awaited>, + current: string | undefined, + boot = false, + ) => { + state.providers = info.providers + state.variants = variantsFor(state.providers, state.model) + state.limits = info.limits + state.activeVariant = boot + ? resolveVariant(ctx.variant, current, savedVariant, state.variants) + : current && !state.variants.includes(current) + ? undefined + : current + if (footer.isClosed) return + footer.event({ type: "models", providers: info.providers }) + footer.event({ type: "variants", variants: state.variants, current: state.activeVariant }) + if (state.model) + footer.event({ type: "model", model: formatModelLabel(state.model, state.activeVariant, state.providers) }) + } + + let catalogRefresh: Promise | undefined + let catalogRefreshQueued = false + const requestCatalogRefresh = () => { + catalogRefreshQueued = true + if (catalogRefresh || footer.isClosed) return + catalogRefresh = (async () => { + await Promise.all([modelTask, initialCatalog]) + while (catalogRefreshQueued && !footer.isClosed) { + catalogRefreshQueued = false + const [catalog, info] = await Promise.allSettled([ + fetchCatalog(), + resolveModelInfoStrict(ctx.sdk, ctx.directory, state.model), + ]) + if (catalog.status === "fulfilled") applyCatalog(catalog.value) + if (info.status === "fulfilled") applyModelInfo(info.value, state.activeVariant) + } + })().finally(() => { + catalogRefresh = undefined + if (catalogRefreshQueued) requestCatalogRefresh() + }) + void catalogRefresh.catch(() => {}) + } + + const initialCatalog = footer .idle() .then(loadCatalog) .catch(() => {}) + void initialCatalog if (Flag.OPENCODE_SHOW_TTFD) { footer.append({ @@ -428,31 +485,7 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep void Promise.resolve(input.afterPaint(ctx)).catch(() => {}) } - void modelTask.then((info) => { - state.providers = info.providers - state.variants = variantsFor(state.providers, state.model) - state.limits = info.limits - - const next = resolveVariant(ctx.variant, session.variant, savedVariant, state.variants) - if (next !== state.activeVariant) { - state.activeVariant = next - } - - if (footer.isClosed) { - return - } - - footer.event({ type: "models", providers: info.providers }) - footer.event({ type: "variants", variants: state.variants, current: state.activeVariant }) - if (!state.model) { - return - } - - footer.event({ - type: "model", - model: formatModelLabel(state.model, state.activeVariant, state.providers), - }) - }) + void modelTask.then((info) => applyModelInfo(info, session.variant, true)) const streamTask = deps.streamTransport ?? import("./stream-v2.transport") const ensureStream = () => { @@ -484,6 +517,7 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep providers: () => state.providers, footer, trace: log, + onCatalogRefresh: requestCatalogRefresh, }) if (footer.isClosed) { await handle.close() diff --git a/packages/opencode/src/cli/cmd/run/stream-v2.subagent.ts b/packages/opencode/src/cli/cmd/run/stream-v2.subagent.ts index 1e7b039fd8..00de6490c4 100644 --- a/packages/opencode/src/cli/cmd/run/stream-v2.subagent.ts +++ b/packages/opencode/src/cli/cmd/run/stream-v2.subagent.ts @@ -27,7 +27,7 @@ import type { FooterSubagentDetail, FooterSubagentState, FooterSubagentTab, Stre const CHILD_MESSAGE_LIMIT = 80 const CHILD_FRAME_LIMIT = 80 -const DISCOVERY_BUFFER_LIMIT = 64 +const CHILD_EVENT_BUFFER_LIMIT = 64 const FAMILY_LIST_LIMIT = 100 const FALLBACK_LABEL = "Subagent" @@ -160,6 +160,7 @@ type ChildState = { tools: Map finishedTools: Set messageIDs: Set + prompts: Map hydrated: boolean } @@ -223,6 +224,8 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac // Foreign events buffered while a session.get discovery is in flight, so a // fast child (including its settled event) is not lost mid-discovery. const pendingEvents = new Map() + const hydrationEvents = new Map() + const hydrationOverflow = new Set() const hydrations = new Map>() let selected: string | undefined @@ -244,6 +247,7 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac tools: new Map(), finishedTools: new Set(), messageIDs: new Set(), + prompts: new Map(), hydrated: false, } if (!existing) children.set(sessionID, child) @@ -332,6 +336,7 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac child.callIDs.clear() for (const message of messages) { if (message.type === "user") { + child.prompts.delete(message.id) userFrame(child, message.id, message.text) continue } @@ -382,16 +387,38 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac const hydrateChild = (child: ChildState): Promise => { const existing = hydrations.get(child.sessionID) if (existing) return existing + const pendingPrompts = new Map(child.prompts) + const pendingTools = new Map(child.tools) + let retry = false const task = input.sdk.v2.session .messages({ sessionID: child.sessionID, limit: CHILD_MESSAGE_LIMIT, order: "desc" }, { throwOnError: true }) .then((response) => { + const buffered = hydrationEvents.get(child.sessionID) ?? [] + hydrationEvents.delete(child.sessionID) + if (hydrationOverflow.delete(child.sessionID)) { + child.hydrated = false + retry = true + notifyDetail(child) + return + } + for (const [id, prompt] of pendingPrompts) { + if (!child.prompts.has(id)) child.prompts.set(id, prompt) + } rebuild(child, response.data.data.toReversed()) + for (const [id, tool] of pendingTools) { + if (!child.finishedTools.has(id) && !child.tools.has(id)) child.tools.set(id, tool) + } + for (const event of buffered) reduce(child, event) child.hydrated = true notifyDetail(child) }) - .catch(() => {}) + .catch(() => { + hydrationEvents.delete(child.sessionID) + hydrationOverflow.delete(child.sessionID) + }) .finally(() => { hydrations.delete(child.sessionID) + if (retry) queueMicrotask(() => void hydrateChild(child)) }) hydrations.set(child.sessionID, task) return task @@ -424,8 +451,14 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac } const reduce = (child: ChildState, event: V2Event) => { + if (event.type === "session.prompt.admitted") { + child.prompts.set(event.data.inputID, event.data.prompt.text) + return + } if (event.type === "session.prompt.promoted") { - if (userFrame(child, event.data.inputID, "")) { + const prompt = child.prompts.get(event.data.inputID) ?? "" + child.prompts.delete(event.data.inputID) + if (userFrame(child, event.data.inputID, prompt)) { touch(child, event.created) notifyDetail(child) } @@ -511,10 +544,12 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac return } if (event.type === "session.tool.input.started") { + if (child.finishedTools.has(event.data.callID)) return child.tools.set(event.data.callID, { name: event.data.name, input: {}, started: event.created }) return } if (event.type === "session.tool.called") { + if (child.finishedTools.has(event.data.callID)) return const current = child.tools.get(event.data.callID) child.tools.set(event.data.callID, { name: event.data.tool, @@ -640,12 +675,18 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac foreign(sessionID, event) { const child = children.get(sessionID) if (child) { + if (hydrations.has(sessionID)) { + const buffered = hydrationEvents.get(sessionID) ?? [] + if (buffered.length < CHILD_EVENT_BUFFER_LIMIT) buffered.push(event) + else hydrationOverflow.add(sessionID) + hydrationEvents.set(sessionID, buffered) + } reduce(child, event) return } discover(sessionID) const buffered = pendingEvents.get(sessionID) - if (buffered && buffered.length < DISCOVERY_BUFFER_LIMIT) buffered.push(event) + if (buffered && buffered.length < CHILD_EVENT_BUFFER_LIMIT) buffered.push(event) }, async hydrate(next) { for (const message of next.messages) { diff --git a/packages/opencode/src/cli/cmd/run/stream-v2.transport.ts b/packages/opencode/src/cli/cmd/run/stream-v2.transport.ts index 42e2c7bd53..8b80b0b64e 100644 --- a/packages/opencode/src/cli/cmd/run/stream-v2.transport.ts +++ b/packages/opencode/src/cli/cmd/run/stream-v2.transport.ts @@ -9,6 +9,7 @@ import type { SessionMessageAssistantTool, V2Event, } from "@opencode-ai/sdk/v2" +import { Event } from "@opencode-ai/schema/event" import { blockerStatus, pickBlockerView } from "./session-data" import { writeSessionOutput } from "./stream" import { createSubagentTracker, legacyTool, toolCommit } from "./stream-v2.subagent" @@ -41,6 +42,7 @@ type StreamInput = { footer: FooterApi trace?: Trace signal?: AbortSignal + onCatalogRefresh?: () => void } export type SessionTurnInput = { @@ -81,6 +83,8 @@ type Wait = { // callID correlates the live shell events once shell.started is observed, and // abort cancels the blocking request when the user interrupts the turn. type ShellWait = { + eventID: string + messageID: string callID?: string resolve: () => void abort: () => void @@ -232,7 +236,7 @@ function streamPartKey(messageID: string, partID: string) { function shellCommit( callID: string, command: string, - next: { text: string; phase: "start" | "progress"; toolState: "running" | "completed" }, + next: Pick, ): StreamCommit { return { kind: "tool", @@ -244,6 +248,41 @@ function shellCommit( } } +function shellTerminal( + callID: string, + command: string, + shell: { status: string; exit?: number | string }, + output: { output: string; cursor: number; size: number; truncated: boolean }, +) { + const incomplete = output.truncated || output.cursor < output.size + const text = `${output.output}${incomplete ? `${output.output.endsWith("\n") || !output.output ? "" : "\n"}[output truncated]` : ""}` + const error = + shell.status === "exited" && shell.exit === 0 + ? undefined + : shell.status === "exited" + ? `Shell exited with code ${shell.exit ?? "unknown"}` + : `Shell ${shell.status}` + if (!error) + return [shellCommit(callID, command, { text, phase: "progress", toolState: "completed" })] + return [ + ...(text ? [shellCommit(callID, command, { text, phase: "progress", toolState: "running" })] : []), + shellCommit(callID, command, { text: error, phase: "final", toolState: "error", toolError: error }), + ] +} + +function messageIDFromEvent(id: string) { + return id.replace(/^evt_/, "msg_") +} + +const catalogEvents = new Set([ + "catalog.updated", + "integration.updated", + "agent.updated", + "command.updated", + "skill.updated", + "reference.updated", +]) + // session.shell resolves after the command settled server-side; the matching // live shell.ended event usually lands within the same tick, but hold the turn // briefly so the output commit renders inside it. @@ -407,6 +446,7 @@ export async function createSessionTransport(input: StreamInput): Promise { + if (catalogEvents.has(event.type)) { + if (input.directory && event.location?.directory && event.location.directory !== input.directory) return + input.onCatalogRefresh?.() + return + } const source = sessionID(event) if (source !== input.sessionID) { if (source) subagents.foreign(source, event) @@ -540,8 +579,8 @@ export async function createSessionTransport(input: StreamInput): Promise {}) await hydrate({ render: state.initial ? input.replay === true : true, reuseVisibleWait: !state.initial }) + input.onCatalogRefresh?.() state.initial = false booting = false for (const event of buffered.splice(0)) apply(event) @@ -867,13 +899,19 @@ export async function createSessionTransport(input: StreamInput): Promise((resolve) => { rendered = resolve }) - const active: ShellWait = { resolve: rendered, abort: () => abort.abort() } + const eventID = Event.ID.create() + const active: ShellWait = { + eventID, + messageID: messageIDFromEvent(eventID), + resolve: rendered, + abort: () => abort.abort(), + } state.shellWait = active - input.trace?.write("send.shell", { sessionID: input.sessionID, command: next.prompt.text }) + input.trace?.write("send.shell", { sessionID: input.sessionID, id: eventID, command: next.prompt.text }) write([], { phase: "running", status: "running shell" }) try { await input.sdk.v2.session.shell( - { sessionID: input.sessionID, command: next.prompt.text }, + { sessionID: input.sessionID, id: eventID, command: next.prompt.text }, { throwOnError: true, signal: abort.signal }, ) await Promise.race([output, wait(SHELL_OUTPUT_GRACE_MS, abort.signal)]) diff --git a/packages/opencode/src/cli/cmd/run/tool.ts b/packages/opencode/src/cli/cmd/run/tool.ts index 9a717ba4e6..181ab7dc7c 100644 --- a/packages/opencode/src/cli/cmd/run/tool.ts +++ b/packages/opencode/src/cli/cmd/run/tool.ts @@ -671,6 +671,10 @@ function scrollBashProgress(p: ToolProps): string { } function scrollBashFinal(p: ToolProps): string { + if (p.frame.status === "error") { + return fail(p.frame) + } + const code = p.metadata.exit ?? num(p.frame.meta.exitCode) ?? num(p.frame.meta.exit_code) const time = span(p.frame.state) if (code === undefined) { @@ -1427,6 +1431,11 @@ export function toolEntryBody(commit: StreamCommit, raw: string): RunEntryBody | return textBody(shellOutput(commit.shell.command, raw) ?? "") } + if (commit.toolState === "error") { + const ctx = toolFrame(commit, raw) + return textBody(toolScroll("final", ctx)) + } + return undefined } diff --git a/packages/opencode/test/cli/run/entry.body.test.ts b/packages/opencode/test/cli/run/entry.body.test.ts index 17659113c5..f393350381 100644 --- a/packages/opencode/test/cli/run/entry.body.test.ts +++ b/packages/opencode/test/cli/run/entry.body.test.ts @@ -50,6 +50,23 @@ function structured(next: StreamCommit) { } describe("run entry body", () => { + test("renders a failed direct shell as an error instead of completed success", () => { + expect( + entryBody( + commit({ + kind: "tool", + text: "Shell exited with code 7", + phase: "final", + source: "tool", + tool: "bash", + toolState: "error", + toolError: "Shell exited with code 7", + shell: { callID: "sh_failed", command: "false" }, + }), + ), + ).toEqual({ type: "text", content: "✖ bash failed: Shell exited with code 7" }) + }) + test("renders assistant, reasoning, and user entries in their display formats", () => { expect( entryBody( diff --git a/packages/opencode/test/cli/run/runtime.test.ts b/packages/opencode/test/cli/run/runtime.test.ts index 9ad7e003b2..702d82d769 100644 --- a/packages/opencode/test/cli/run/runtime.test.ts +++ b/packages/opencode/test/cli/run/runtime.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, mock, spyOn, test } from "bun:test" import { OpencodeClient } from "@opencode-ai/sdk/v2" import { runInteractiveMode } from "@/cli/cmd/run/runtime" -import type { FooterApi, RunProvider } from "@/cli/cmd/run/types" +import type { FooterApi, FooterEvent, RunProvider } from "@/cli/cmd/run/types" const provider: RunProvider = { id: "openai", @@ -53,7 +53,7 @@ function ok(data: T) { }) } -function footer(): FooterApi { +function footer(events: FooterEvent[] = []): FooterApi { let closed = false const closes = new Set<() => void>() @@ -78,7 +78,9 @@ function footer(): FooterApi { closes.delete(fn) } }, - event() {}, + event(value) { + events.push(value) + }, append() {}, idle() { return Promise.resolve() @@ -296,4 +298,183 @@ describe("run interactive runtime", () => { expect(legacyAgents).not.toHaveBeenCalled() expect(legacyCommands).not.toHaveBeenCalled() }) + + test("retains last-known-good state across failed coalesced refreshes and retries later", async () => { + const sdk = new OpencodeClient() + const refreshGate = defer() + let providerCalls = 0 + let modelCalls = 0 + let agentCalls = 0 + let referenceCalls = 0 + const events: FooterEvent[] = [] + const api = footer(events) + spyOn(sdk.v2.provider, "list").mockImplementation(async () => { + providerCalls++ + if (providerCalls === 2) { + await refreshGate.promise + throw new Error("provider refresh failed") + } + return ok({ + location: { directory: "/tmp" }, + data: [ + { + id: "openai", + name: providerCalls >= 3 ? "OpenAI refreshed" : "OpenAI", + api: { type: "native", settings: {} }, + request: { headers: {}, body: {} }, + }, + ], + }) as never + }) + spyOn(sdk.v2.model, "list").mockImplementation(() => { + modelCalls++ + return ok({ + location: { directory: "/tmp" }, + data: [ + { + id: "gpt-5", + providerID: "openai", + name: "Little Frank", + api: { id: "openai", type: "native", settings: {} }, + capabilities: { tools: true, input: ["text"], output: ["text"] }, + request: { headers: {}, body: {} }, + variants: + modelCalls >= 4 + ? [] + : [{ id: modelCalls >= 3 ? "high" : "low", settings: {}, headers: {}, body: {} }], + time: { released: 1 }, + cost: [{ input: 0, output: 0, cache: { read: 0, write: 0 } }], + status: "active", + enabled: true, + limit: { context: modelCalls >= 3 ? 256000 : 128000, output: 8192 }, + }, + ], + }) as never + }) + spyOn(sdk.v2.agent, "list").mockImplementation(async () => { + agentCalls++ + if (agentCalls === 2) throw new Error("agent refresh failed") + return ok({ + location: { directory: "/tmp" }, + data: [{ id: "build", description: agentCalls >= 3 ? "Refreshed agent" : "Agent", mode: "primary" }], + }) as never + }) + spyOn(sdk.v2.reference, "list").mockImplementation(() => { + referenceCalls++ + return ok({ + location: { directory: "/tmp" }, + data: [ + { name: "effect", path: "/effect", description: referenceCalls >= 3 ? "Refreshed reference" : "Reference" }, + ], + }) as never + }) + spyOn(sdk.v2.command, "list").mockImplementation(() => + ok({ location: { directory: "/tmp" }, data: [{ name: "check", description: "Check" }] }) as never, + ) + spyOn(sdk.v2.skill, "list").mockImplementation(() => + ok({ location: { directory: "/tmp" }, data: [] }) as never, + ) + let finalProviders: RunProvider[] = [] + let finalLimits: Record = {} + let retainedProviders: RunProvider[] = [] + let retainedLimits: Record = {} + let retainedCatalog: FooterEvent | undefined + let selectedDefault: unknown + let selectDefault: (() => unknown) | undefined + let selectVariant: ((variant: string | undefined) => unknown) | undefined + let defaultRefreshVariants: FooterEvent | undefined + + await runInteractiveMode( + { + sdk, + directory: "/tmp", + sessionID: "ses-1", + sessionTitle: "Session", + resume: false, + agent: "build", + model: { providerID: "openai", modelID: "gpt-5" }, + variant: "low", + files: [], + thinking: false, + backgroundSubagents: false, + }, + { + createRuntimeLifecycle: async (input) => { + selectDefault = () => input.onVariantSelect?.(undefined) + selectVariant = (variant) => input.onVariantSelect?.(variant) + return { + footer: api, + onResize: () => () => {}, + refreshTheme: () => {}, + resetForReplay: () => Promise.resolve(), + close: () => Promise.resolve(), + } + }, + streamTransport: Promise.resolve({ + createSessionTransport: async (input) => { + while ( + !events.some( + (event) => event.type === "variants" && event.variants.includes("low") && event.current === "low", + ) + ) + await Bun.sleep(0) + selectedDefault = await Promise.resolve(selectDefault?.()) + input.onCatalogRefresh?.() + input.onCatalogRefresh?.() + input.onCatalogRefresh?.() + while (providerCalls < 2) await Bun.sleep(0) + refreshGate.resolve() + await new Promise((resolve) => setTimeout(resolve, 0)) + retainedProviders = input.providers?.() ?? [] + retainedLimits = input.limits() + retainedCatalog = events.filter((event) => event.type === "catalog").at(-1) + input.onCatalogRefresh?.() + input.onCatalogRefresh?.() + while (providerCalls < 3 || modelCalls < 3 || agentCalls < 3) await Bun.sleep(0) + await new Promise((resolve) => setTimeout(resolve, 0)) + defaultRefreshVariants = events.filter((event) => event.type === "variants").at(-1) + await Promise.resolve(selectVariant?.("high")) + input.onCatalogRefresh?.() + while (providerCalls < 4 || modelCalls < 4) await Bun.sleep(0) + await new Promise((resolve) => setTimeout(resolve, 0)) + finalProviders = input.providers?.() ?? [] + finalLimits = input.limits() + setTimeout(() => input.footer.close(), 0) + return { + runPromptTurn: async () => {}, + interruptActiveTurn: async () => {}, + selectSubagent: () => {}, + replayOnResize: async () => false, + close: async () => {}, + } + }, + formatUnknownError: (error: unknown) => (error instanceof Error ? error.message : String(error)), + }), + }, + ) + + expect(providerCalls).toBe(4) + expect(modelCalls).toBe(4) + expect(retainedProviders[0]?.name).toBe("OpenAI") + expect(retainedProviders[0]?.models["gpt-5"]?.variants).toEqual({ low: {} }) + expect(retainedLimits["openai/gpt-5"]).toBe(128000) + expect(retainedCatalog).toMatchObject({ + agents: [{ name: "build", description: "Agent" }], + references: [{ name: "effect", description: "Reference" }], + }) + expect(selectedDefault).toMatchObject({ variant: undefined }) + expect(defaultRefreshVariants).toMatchObject({ variants: ["high"], current: undefined }) + expect(finalProviders[0]?.name).toBe("OpenAI refreshed") + expect(finalProviders[0]?.models["gpt-5"]?.variants).toEqual({}) + expect(finalLimits["openai/gpt-5"]).toBe(256000) + expect(events.filter((event) => event.type === "variants").at(-1)).toMatchObject({ + variants: [], + current: undefined, + }) + expect(events.filter((event) => event.type === "catalog").at(-1)).toMatchObject({ + agents: [{ name: "build", description: "Refreshed agent" }], + references: [{ name: "effect", description: "Refreshed reference" }], + commands: [{ name: "check", description: "Check" }], + }) + }) }) diff --git a/packages/opencode/test/cli/run/stream-v2.transport.test.ts b/packages/opencode/test/cli/run/stream-v2.transport.test.ts index cd2d131642..cd43fda2de 100644 --- a/packages/opencode/test/cli/run/stream-v2.transport.test.ts +++ b/packages/opencode/test/cli/run/stream-v2.transport.test.ts @@ -1019,7 +1019,7 @@ describe("V2 mini transport", () => { request = input queueMicrotask(() => { events.push({ - id: "evt_shell_start", + id: input.id ?? "evt_missing", created: 0, type: "session.shell.started", durable: durable("ses_1"), @@ -1071,7 +1071,7 @@ describe("V2 mini transport", () => { includeFiles: true, }) - expect(request).toMatchObject({ sessionID: "ses_1", command: "ls" }) + expect(request).toMatchObject({ sessionID: "ses_1", command: "ls", id: expect.stringMatching(/^evt_/) }) expect(ui.commits.filter((item) => item.shell)).toMatchObject([ { phase: "start", tool: "bash", toolState: "running", shell: { callID: "sh_shell", command: "ls" } }, { phase: "progress", text: "file.txt", toolState: "completed", shell: { callID: "sh_shell", command: "ls" } }, @@ -1123,6 +1123,133 @@ describe("V2 mini transport", () => { 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, + limits: () => ({}), + footer: ui.api, + }) + let request: Parameters[0] | undefined + let complete!: () => void + spyOn(client.v2.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.shell?.callID === "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()) @@ -1190,6 +1317,91 @@ describe("V2 mini transport", () => { 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, + shell: { + id: "sh_failed", + status: "exited", + command: "false", + cwd: "/tmp", + shell: "/bin/sh", + file: "/tmp/failed", + exit: 7, + metadata: {}, + time: { started: 0, completed: 1 }, + }, + 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, + limits: () => ({}), + 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()) @@ -1363,6 +1575,17 @@ describe("V2 mini transport", () => { done = true }) while (!sent) await Bun.sleep(0) + events.push({ + id: "evt_other", + created: 0, + type: "session.skill.activated", + durable: durable("ses_1"), + data: { + sessionID: "ses_1", + name: "other", + text: "other instructions", + }, + }) events.push({ id: "evt_unrelated_settled", created: 0, @@ -1396,6 +1619,46 @@ describe("V2 mini transport", () => { 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, + directory: "/project", + sessionID: "ses_1", + thinking: false, + limits: () => ({}), + 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" }, data: {} }) + events.push({ + id: "evt_foreign_catalog", + created: 0, + type: "catalog.updated", + location: { directory: "/other" }, + 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()) @@ -1526,6 +1789,328 @@ describe("V2 mini transport", () => { 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, + limits: () => ({}), + 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.prompt.admitted", + durable: durable("ses_child"), + data: { + sessionID: "ses_child", + inputID: "msg_child_prompt", + prompt: { 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.prompt.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.v2.session, "messages").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, + limits: () => ({}), + 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.prompt.admitted", + durable: durable("ses_child"), + data: { + sessionID: "ses_child", + inputID: "msg_child_race", + prompt: { 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.prompt.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.v2.session, "messages").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, + limits: () => ({}), + 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", + textID: `txt_overflow_${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.v2.session, "messages").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: "bash", + state: { + status: "completed" as const, + input: { command: "projected" }, + content: [{ type: "text" as const, text: "projected result" }], + structured: {}, + }, + 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, + limits: () => ({}), + 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, tool: 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, + tool, + input, + provider: { executed: true }, + }, + }) + + inputStarted("call_terminal", "grep", 0) + called("call_terminal", "grep", { 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), + data: { + sessionID: "ses_child", + assistantMessageID: "msg_tool_projected", + callID: "call_terminal", + structured: {}, + content: [{ type: "text", text: "found" }], + provider: { executed: true }, + }, + }) + inputStarted("call_overlap", "bash", 3) + called("call_overlap", "bash", { 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: "bash", + 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())