fix(cli): bound run setup RPCs and report startup progress

packages/cli/src/commands/handlers/run.ts

@@ -7,7 +7,14 @@ export default Runtime.handler(Commands.commands.run, (input) =>
   Effect.gen(function* () {
     const { runNonInteractive } = yield* Effect.promise(() => import("../../run/run"))
     const separator = process.argv.indexOf("--", 2)
-    const server = yield* ServerConnection.resolve()
+    const server = yield* ServerConnection.resolve({
+      onStart: (reason) =>
+        process.stderr.write(
+          reason === "version-mismatch"
+            ? "Restarting background server (version mismatch)...\n"
+            : "Starting background server...\n",
+        ),
+    })
     yield* Effect.promise(() =>
       runNonInteractive({
         server,

packages/cli/src/run/noninteractive.ts

@@ -12,6 +12,7 @@ import { EOL } from "node:os"
 import { readFile } from "node:fs/promises"
 import { nonEmptyToolContent, toolOutputText, type MiniToolPart } from "@opencode/tui/mini/tool"
 import { UI } from "./ui"
+import { RUN_ADMISSION_TIMEOUT_MS, runSetupRequest, RunSetupTimeoutError } from "./setup"
 
 type Model = {
   providerID: string
@@ -76,7 +77,13 @@ const PERMISSION_REJECTED_FEEDBACK =
 export async function runNonInteractivePrompt(input: Input) {
   const controller = new AbortController()
   const stream = input.client.event.subscribe({ signal: controller.signal })[Symbol.asyncIterator]()
-  const connected = await stream.next()
+  const connected = await runSetupRequest("Connecting event stream", async (signal) => {
+    signal.addEventListener("abort", () => controller.abort(), { once: true })
+    return stream.next()
+  }).catch((error) => {
+    controller.abort()
+    throw error
+  })
   if (connected.done) throw new Error("Event stream disconnected before prompt admission")
 
   const messageID = SessionMessage.ID.create()
@@ -688,51 +695,75 @@ export async function runNonInteractivePrompt(input: Input) {
   let completed: Promise<void> | undefined
   try {
     if (input.agent) {
-      await input.client.session.switchAgent({ sessionID: input.sessionID, agent: input.agent })
+      const agent = input.agent
+      await runSetupRequest("Selecting agent", (signal) =>
+        input.client.session.switchAgent({ sessionID: input.sessionID, agent }, { signal }),
+      )
     }
     const selected = input.model
       ? { providerID: input.model.providerID, id: input.model.modelID, variant: input.variant }
       : input.variant
-        ? await input.client.session
-            .get({ sessionID: input.sessionID })
+        ? await runSetupRequest("Resolving session", (signal) =>
+            input.client.session.get({ sessionID: input.sessionID }, { signal }),
+          )
             .then((result) => result.model)
             .then(async (model) => {
               if (model) return { ...model, variant: input.variant }
-              const result = await input.client.model.default()
+              const result = await runSetupRequest("Resolving model", (signal) =>
+                input.client.model.default(undefined, { signal }),
+              )
               const fallback = result.data
               return fallback ? { providerID: fallback.providerID, id: fallback.id, variant: input.variant } : undefined
             })
         : undefined
     if (input.variant && !selected) throw new Error("Cannot select a variant before selecting a model")
     if (selected) {
-      await input.client.session.switchModel({ sessionID: input.sessionID, model: selected })
+      await runSetupRequest("Selecting model", (signal) =>
+        input.client.session.switchModel({ sessionID: input.sessionID, model: selected }, { signal }),
+      )
     }
 
     const prepared = await Promise.all(input.files.map(prepareFile))
     if (interrupted) return
     submitted = true
     completed = consume()
     admission = new AbortController()
-    const response = await input.client.session
-      .prompt(
-        {
-          sessionID: input.sessionID,
-          id: messageID,
-          text: [input.message, ...prepared.flatMap((file) => (file.text ? [file.text] : []))].join("\n\n"),
-          files: prepared.flatMap((file) => (file.attachment ? [file.attachment] : [])),
-          delivery: "steer",
-        },
-        { signal: admission.signal },
-      )
-      .catch(async (error) => {
-        if (interrupted) {
-          await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {})
-        }
-        controller.abort()
+    const response = await runSetupRequest(
+      `Admitting prompt (session ${input.sessionID}, request ${messageID})`,
+      (signal) =>
+        input.client.session.prompt(
+          {
+            sessionID: input.sessionID,
+            id: messageID,
+            text: [input.message, ...prepared.flatMap((file) => (file.text ? [file.text] : []))].join("\n\n"),
+            files: prepared.flatMap((file) => (file.attachment ? [file.attachment] : [])),
+            delivery: "steer",
+          },
+          { signal },
+        ),
+      { signal: admission.signal },
+      RUN_ADMISSION_TIMEOUT_MS,
+    ).catch(async (error) => {
+      if (interrupted) {
+        await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {})
+      }
+      controller.abort()
+      if (interrupted) {
         await completed?.catch(() => {})
-        if (interrupted || emittedError) return undefined
-        throw error
-      })
+        return undefined
+      }
+      if (error instanceof RunSetupTimeoutError) {
+        // Event handling may itself be waiting on an RPC; do not let cleanup extend the admission deadline.
+        void completed?.catch(() => {})
+        throw new Error(
+          `${error.message}. Delivery is unknown; the prompt may already be admitted. Keep session ${input.sessionID} and request ${messageID}; inspect this session before resubmitting.`,
+          { cause: error },
+        )
+      }
+      await completed?.catch(() => {})
+      if (emittedError) return undefined
+      throw error
+    })
     admission = undefined
     if (!response) return
     if (interrupted) await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {})

packages/cli/src/run/run.ts

@@ -1,5 +1,5 @@
 import { Service, type Endpoint } from "@opencode/client/effect/service"
-import { OpenCode, type OpenCodeClient, type SessionMessageAssistantTool } from "@opencode/client/promise"
+import { OpenCode, type SessionMessageAssistantTool } from "@opencode/client/promise"
 import { FSUtil } from "@opencode/util/fs-util"
 import { open } from "node:fs/promises"
 import path from "node:path"
@@ -11,6 +11,7 @@ import { runNonInteractivePrompt } from "./noninteractive"
 import { UI } from "./ui"
 import { Env } from "../env"
 import { errorMessage } from "../util/error"
+import { withRunSetupDeadlines } from "./setup"
 
 export type RunCommandInput = {
   server: ServerConnection.Resolved
@@ -73,6 +74,7 @@ async function run(input: RunCommandInput, options: ExecutionOptions) {
   const root = options.root ?? process.env.PWD ?? process.cwd()
   const local = localDirectory(root)
   const directory = options.useServerDirectory ? undefined : (options.directory ?? local)
+  if (!process.stdin.isTTY) process.stderr.write("Reading stdin until EOF (close the pipe to continue)...\n")
   const message = mergeInput(formatMessage(input.message), process.stdin.isTTY ? undefined : await readStdin())
   if (!message?.trim()) fail("You must provide a message")
   const files = await Promise.all(input.file.map((file) => prepareFile(file, root, options)))
@@ -88,9 +90,10 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End
     fetch: ((request: RequestInfo | URL, init?: RequestInit) =>
       fetch(request, { ...init, timeout: false } as BunFetchRequestInit)) as typeof fetch,
   })
+  const setup = withRunSetupDeadlines(client)
   const explicit = parseRunModel(input.model)
   const target = await resolveSessionTarget({
-    client,
+    client: setup,
     location: prepared.directory ? { directory: prepared.directory } : undefined,
     continue: input.continue,
     session: input.session,
@@ -104,7 +107,7 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End
       const selected =
         next.model ??
         (options.variant
-          ? await client.model
+          ? await setup.model
               .default({ location: { directory: next.location.directory } })
               .then((result) => result.data)
           : undefined)
@@ -128,7 +131,7 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End
   const model = target.model ? { providerID: target.model.providerID, modelID: target.model.id } : undefined
   const variant = target.model?.variant
   if (!target.resume && input.title !== undefined) {
-    await client.session.update({
+    await setup.session.update({
       sessionID: target.session.id,
       title: input.title || prepared.message.slice(0, 50) + (prepared.message.length > 50 ? "..." : ""),
     })
@@ -247,6 +250,7 @@ async function renderToolError(part: SessionMessageAssistantTool, directory: str
 /** @internal Used by the V1 command boundary before a Session exists. */
 export function reportRunError(input: Pick<RunCommandInput, "format">, message: string, sessionID?: string) {
   process.exitCode = 1
+  UI.error(message)
   if (input.format === "json") {
     process.stdout.write(
       JSON.stringify({
@@ -258,7 +262,6 @@ export function reportRunError(input: Pick<RunCommandInput, "format">, message:
     )
     return
   }
-  UI.error(message)
 }
 
 function fail(message: string): never {

packages/cli/src/run/setup.ts

@@ -0,0 +1,69 @@
+import type { OpenCodeClient } from "@opencode/client/promise"
+
+type RequestOptions = NonNullable<Parameters<OpenCodeClient["session"]["get"]>[1]>
+
+// Subprocess tests shrink both deadlines proportionally instead of waiting them out.
+const scale = Number(process.env.OPENCODE_TEST_RUN_DEADLINE_SCALE) || 1
+
+export const RUN_SETUP_TIMEOUT_MS = 30_000 / scale
+
+/** Admission may wait on plugin activation, so it matches the service bootstrap deadline. */
+export const RUN_ADMISSION_TIMEOUT_MS = 120_000 / scale
+
+export class RunSetupTimeoutError extends Error {}
+
+/** Each setup call gets its own deadline; execution and its event stream have none. */
+export async function runSetupRequest<T>(
+  phase: string,
+  request: (signal: AbortSignal) => Promise<T>,
+  options?: RequestOptions,
+  timeout = RUN_SETUP_TIMEOUT_MS,
+) {
+  process.stderr.write(`${phase}...\n`)
+  const controller = new AbortController()
+  const signal = options?.signal ? AbortSignal.any([controller.signal, options.signal]) : controller.signal
+  const expired = new RunSetupTimeoutError(`${phase} timed out after ${timeout / 1000}s`)
+  const deadline = setTimeout(() => controller.abort(expired), timeout)
+  try {
+    const result = await request(signal).catch((error) => {
+      if (controller.signal.aborted) throw expired
+      throw error
+    })
+    if (controller.signal.aborted) throw expired
+    return result
+  } finally {
+    clearTimeout(deadline)
+  }
+}
+
+/** Bounds the session-target resolution RPCs; pass the unbounded client to execution. */
+export function withRunSetupDeadlines(client: OpenCodeClient) {
+  const bounded =
+    <A, B>(phase: string, request: (input: A, options?: RequestOptions) => Promise<B>) =>
+    (input: A, options?: RequestOptions) =>
+      runSetupRequest(phase, (signal) => request(input, { ...options, signal }), options)
+
+  const result: OpenCodeClient = {
+    ...client,
+    location: {
+      ...client.location,
+      get: (input, options) =>
+        runSetupRequest("Looking up location", (signal) => client.location.get(input, { ...options, signal }), options),
+    },
+    model: {
+      ...client.model,
+      default: (input, options) =>
+        runSetupRequest("Resolving model", (signal) => client.model.default(input, { ...options, signal }), options),
+    },
+    session: {
+      ...client.session,
+      get: bounded("Resolving session", client.session.get),
+      list: bounded("Looking up sessions", client.session.list),
+      create: bounded("Creating session", client.session.create),
+      fork: bounded("Forking session", client.session.fork),
+      environment: bounded("Setting session environment", client.session.environment),
+      update: bounded("Updating session", client.session.update),
+    },
+  }
+  return result
+}

packages/cli/test/run/noninteractive.test.ts

@@ -430,7 +430,10 @@ describe("runNonInteractivePrompt", () => {
     }
     expect(sdk.session.form.cancel).toHaveBeenCalledWith({ sessionID: "global", formID: "frm_live" }, globalOptions)
     expect(sdk.session.form.cancel).toHaveBeenCalledWith({ sessionID: "ses_1", formID: "frm_pending" })
-    expect(sdk.session.form.cancel).toHaveBeenCalledWith({ sessionID: "global", formID: "frm_pending_global" }, globalOptions)
+    expect(sdk.session.form.cancel).toHaveBeenCalledWith(
+      { sessionID: "global", formID: "frm_pending_global" },
+      globalOptions,
+    )
     expect(sdk.form.list).toHaveBeenCalledWith({
       location: { directory: "/work tree" },
     })
@@ -445,7 +448,10 @@ describe("runNonInteractivePrompt", () => {
     })
     expect(sdk.session.form.cancel).toHaveBeenCalledWith({ sessionID: "ses_1", formID: "frm_pending" })
     expect(sdk.form.list).not.toHaveBeenCalled()
-    expect(sdk.session.form.cancel).not.toHaveBeenCalledWith({ sessionID: "global", formID: "frm_live" }, expect.anything())
+    expect(sdk.session.form.cancel).not.toHaveBeenCalledWith(
+      { sessionID: "global", formID: "frm_live" },
+      expect.anything(),
+    )
     expect(sdk.session.form.cancel).not.toHaveBeenCalledWith(
       { sessionID: "global", formID: "frm_pending_global" },
       expect.anything(),
@@ -477,7 +483,9 @@ describe("runNonInteractivePrompt", () => {
         error: { type: "provider.transport", message: "Provider request failed" },
       }),
     ])
-    expect(output.stderr).toBe("")
+    expect(output.stderr).toMatch(
+      /^Connecting event stream\.\.\.\nAdmitting prompt \(session ses_1, request msg_[^)]+\)\.\.\.\n$/,
+    )
     const sdk = await run({ compatibility: "v1", turn: (messageID) => [prompted(messageID), settled()] })
     expect(sdk.session.wait).not.toHaveBeenCalled()
     expect(sdk.message.list).not.toHaveBeenCalled()
@@ -507,7 +515,11 @@ describe("runNonInteractivePrompt", () => {
       ],
     })
 
-    expect(output).toEqual({ stdout: "", stderr: "", exitCode: 0 })
+    expect(output.stdout).toBe("")
+    expect(output.exitCode).toBe(0)
+    expect(output.stderr).toMatch(
+      /^Connecting event stream\.\.\.\nAdmitting prompt \(session ses_1, request msg_[^)]+\)\.\.\.\n$/,
+    )
   })
 
   test("renders a native terminal failure snapshot when live progress was missed", async () => {
@@ -572,6 +584,8 @@ describe("runNonInteractivePrompt", () => {
     expect(events[0].part.state.output).toBeUndefined()
     expect(events[0].part.state.metadata.metadata).toBeUndefined()
     expect(events[0].part.state.metadata.content).toBeUndefined()
-    expect(output.stderr).toBe("")
+    expect(output.stderr).toMatch(
+      /^Connecting event stream\.\.\.\nAdmitting prompt \(session ses_1, request msg_[^)]+\)\.\.\.\n$/,
+    )
   })
 })