fix(run): align mini with current session contracts (#35354)

2026-07-04 21:14:01 +02:00
parent ba07481b59
commit 57fb3e5cc5
10 changed files with 1063 additions and 80 deletions
@@ -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"],
@@ -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<void> => {
+  const applyCatalog = (catalog: {
+    agents: Awaited<ReturnType<typeof loadRunAgents>>
+    references: Awaited<ReturnType<typeof loadRunReferences>>
+    commands: Awaited<ReturnType<typeof loadRunCommands>>
+  }) => {
    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<ReturnType<typeof resolveModelInfo>>,
+    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<void> | 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()
@@ -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<string, ToolTrack>
  finishedTools: Set<string>
  messageIDs: Set<string>
+  prompts: Map<string, string>
  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<string, V2Event[]>()
+  const hydrationEvents = new Map<string, V2Event[]>()
+  const hydrationOverflow = new Set<string>()
  const hydrations = new Map<string, Promise<void>>()
  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<void> => {
    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) {
@@ -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, "text" | "phase" | "toolState" | "toolError">,
 ): 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<Sessio
    }
    if (message.type === "shell") {
      state.shellCommands.set(message.shell.id, message.shell.command)
+      if (state.shellWait?.messageID === message.id) state.shellWait.callID = message.shell.id
      const completed = message.time.completed !== undefined
      if (!render) {
        // Suppressed history: mark settled shells rendered so live redelivery
@@ -430,13 +470,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
      }
      if (completed && message.output && !state.shellEnded.has(message.shell.id)) {
        state.shellEnded.add(message.shell.id)
-        write([
-          shellCommit(message.shell.id, message.shell.command, {
-            text: message.output.output,
-            phase: "progress",
-            toolState: "completed",
-          }),
-        ])
+        write(shellTerminal(message.shell.id, message.shell.command, message.shell, message.output))
      }
      if (completed && state.shellWait?.callID === message.shell.id) state.shellWait.resolve()
      return
@@ -522,6 +556,11 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
  }

  const apply = (event: RunV2Event) => {
+    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<Sessio
      return
    }
    if (event.type === "session.skill.activated") {
-      const messageID = event.id.replace(/^evt_/, "msg_")
-      if (state.wait) state.wait.promoted = true
+      const messageID = messageIDFromEvent(event.id)
+      if (state.wait?.messageID === messageID) state.wait.promoted = true
      if (state.skillMessages.has(messageID)) return
      state.skillMessages.add(messageID)
      write([skillCommit(messageID, event.data.name)])
@@ -550,7 +589,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
    if (event.type === "session.shell.started") {
      state.shellCommands.set(event.data.shell.id, event.data.shell.command)
      const wait = state.shellWait
-      if (wait && wait.callID === undefined) wait.callID = event.data.shell.id
+      if (wait?.eventID === event.id) wait.callID = event.data.shell.id
      if (state.shellStarted.has(event.data.shell.id)) return
      state.shellStarted.add(event.data.shell.id)
      write(
@@ -580,19 +619,11 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
      }
      if (!state.shellEnded.has(event.data.shell.id)) {
        state.shellEnded.add(event.data.shell.id)
-        commits.push(
-          shellCommit(event.data.shell.id, command, {
-            text: event.data.output.output,
-            phase: "progress",
-            toolState: "completed",
-          }),
-        )
+        commits.push(...shellTerminal(event.data.shell.id, command, event.data.shell, event.data.output))
      }
      const wait = state.shellWait
-      // An unset callID means shell.started has not been observed yet (event
-      // delivery lag); mini serializes its own shells, so adopt this ended.
-      const owned = wait !== undefined && (wait.callID === undefined || wait.callID === event.data.shell.id)
-      write(commits, owned || state.wait ? undefined : { phase: "idle", status: "" })
+      const owned = wait?.callID === event.data.shell.id
+      write(commits, owned || state.wait || state.shellWait ? undefined : { phase: "idle", status: "" })
      if (owned) wait.resolve()
      return
    }
@@ -828,6 +859,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
          })()
          void consume.catch(() => {})
          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<Sessio
    const output = new Promise<void>((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)])
@@ -671,6 +671,10 @@ function scrollBashProgress(p: ToolProps<typeof BashTool>): string {
 }

 function scrollBashFinal(p: ToolProps<typeof BashTool>): 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
  }