diff --git a/packages/opencode/src/cli/cmd/run.ts b/packages/opencode/src/cli/cmd/run.ts index cccc2550179f..11fa9632987e 100644 --- a/packages/opencode/src/cli/cmd/run.ts +++ b/packages/opencode/src/cli/cmd/run.ts @@ -22,7 +22,7 @@ import { UI } from "../ui" import { effectCmd } from "../effect-cmd" import { EOL } from "os" import { Filesystem } from "@/util/filesystem" -import { createOpencodeClient, type OpencodeClient, type ToolPart } from "@opencode-ai/sdk/v2" +import { createOpencodeClient, type Event, type OpencodeClient, type ToolPart } from "@opencode-ai/sdk/v2" import { FormatError, FormatUnknownError } from "../error" import { INTERACTIVE_INPUT_ERROR, resolveInteractiveStdin } from "./run/runtime.stdin" @@ -690,12 +690,14 @@ export const RunCommand = effectCmd({ return false } + const toggles = new Map() + const printed = new Set() + // Consume one subscribed event stream for the active session and mirror it // to stdout/UI. `client` is passed explicitly because attach mode may // rebind the SDK to the session's directory after the subscription is // created, and replies issued from inside the loop must use that client. async function loop(client: OpencodeClient, events: Awaited>) { - const toggles = new Map() const sessions = new Set([sessionID]) let error: string | undefined @@ -720,6 +722,16 @@ export const RunCommand = effectCmd({ if (event.type === "message.part.updated") { const part = event.properties.part if (part.sessionID !== sessionID) continue + if ( + args.attach && + (part.type === "step-start" || + part.type === "step-finish" || + ((part.type === "text" || part.type === "reasoning") && part.time?.end) || + (part.type === "tool" && (part.state.status === "completed" || part.state.status === "error"))) + ) { + if (printed.has(part.id)) continue + printed.add(part.id) + } if (part.type === "tool" && (part.state.status === "completed" || part.state.status === "error")) { if (emit("tool_use", { part })) continue @@ -831,15 +843,41 @@ export const RunCommand = effectCmd({ await share(client, sessionID) if (!interactive) { - const events = await client.event.subscribe() + const controller = new AbortController() + const events = await client.event.subscribe(undefined, { signal: controller.signal }) const completed = loop(client, events).catch((e) => { console.error(e) process.exitCode = 1 }) - async function finish() { - if (args.attach) return + async function finish(parentID?: string) { + if (args.attach) controller.abort() const error = await completed if (error) process.exitCode = 1 + if (!args.attach || !parentID) return + + // The request has completed, but remote events may still be buffered + // or missing. Recover this prompt's output without waiting for idle. + const messages = await client.session.messages({ sessionID }, { throwOnError: true }) + async function* replay(): AsyncGenerator { + for (const message of messages.data) { + if (message.info.role !== "assistant" || message.info.parentID !== parentID) continue + yield { id: message.info.id, type: "message.updated", properties: { sessionID, info: message.info } } + for (const part of message.parts) + yield { + id: part.id, + type: "message.part.updated", + properties: { sessionID, part, time: message.info.time.completed ?? message.info.time.created }, + } + if (message.info.error && !error) + yield { + id: message.info.id, + type: "session.error", + properties: { sessionID, error: message.info.error }, + } + } + } + const replayError = await loop(client, { stream: replay() }) + if (replayError) process.exitCode = 1 } if (args.command) { @@ -856,7 +894,7 @@ export const RunCommand = effectCmd({ process.exitCode = 1 return } - await finish() + await finish(result.data?.info.parentID) return } @@ -873,7 +911,7 @@ export const RunCommand = effectCmd({ process.exitCode = 1 return } - await finish() + await finish(result.data?.info.parentID) return } diff --git a/packages/opencode/test/cli/run/run-process.test.ts b/packages/opencode/test/cli/run/run-process.test.ts index d2d4bae87921..ec44881a4313 100644 --- a/packages/opencode/test/cli/run/run-process.test.ts +++ b/packages/opencode/test/cli/run/run-process.test.ts @@ -7,6 +7,7 @@ import { describe, expect } from "bun:test" import { Effect } from "effect" import { reply } from "../../lib/llm-server" import { cliIt } from "../../lib/cli-process" +import { testProviderConfig } from "../../lib/test-provider" describe("opencode run (non-interactive subprocess)", () => { // Happy path: prompt completes, output reaches stdout, process exits 0. @@ -299,6 +300,7 @@ describe("opencode run (non-interactive subprocess)", () => { }) opencode.expectExit(result, 0) + expect(result.stdout).toBe("attachment received\n") const input = JSON.stringify(yield* llm.inputs) expect(input).toContain(sentinel) expect(input).not.toContain(`file://${source}`) @@ -306,6 +308,104 @@ describe("opencode run (non-interactive subprocess)", () => { 60_000, ) + for (const mode of ["stream", "missing-idle", "command"]) { + cliIt.live( + `attach preserves completed output (${mode})`, + ({ llm, opencode }) => + Effect.gen(function* () { + yield* llm.push( + reply().reason("thinking").text("before").tool("bash", { + command: "printf tool-output", + description: "Print deterministic output", + }), + ) + yield* llm.text("after") + const server = yield* opencode.serve({ + env: { + OPENCODE_CONFIG_CONTENT: JSON.stringify({ + ...testProviderConfig(llm.url), + command: { probe: { template: "exercise output" } }, + }), + }, + }) + // Keep SSE connected but withhold its events, including session idle. + const proxy = yield* Effect.acquireRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + const url = new URL(request.url) + if (url.pathname === "/event" && mode !== "stream") { + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode('data: {"type":"server.connected","properties":{}}\n\n'), + ) + }, + }), + { headers: { "Content-Type": "text/event-stream" } }, + ) + } + const forwarded = new Request(new URL(url.pathname + url.search, server.url), request) + forwarded.headers.set("Accept-Encoding", "identity") + return fetch(forwarded) + }, + }), + ), + (proxy) => Effect.sync(() => proxy.stop(true)), + ) + const result = yield* opencode.run("exercise output", { + command: mode === "command" ? "probe" : undefined, + format: "json", + extraArgs: ["--attach", proxy.url.origin, "--thinking", "--dangerously-skip-permissions"], + }) + opencode.expectExit(result, 0) + const events = opencode.parseJsonEvents(result.stdout) + expect(events.map((event) => event.type)).toEqual([ + "step_start", + "reasoning", + "text", + "tool_use", + "step_finish", + "step_start", + "text", + "step_finish", + ]) + expect(events.filter((event) => event.type === "text").map((event) => event.part)).toEqual([ + expect.objectContaining({ text: "before" }), + expect.objectContaining({ text: "after" }), + ]) + + yield* llm.text("next turn") + const resumed = yield* opencode.run("continue", { + format: "json", + extraArgs: ["--attach", proxy.url.origin, "--session", String(events[0].sessionID)], + }) + opencode.expectExit(resumed, 0) + expect(opencode.parseJsonEvents(resumed.stdout).map((event) => event.type)).toEqual([ + "step_start", + "text", + "step_finish", + ]) + expect(resumed.stdout).toContain("next turn") + expect(resumed.stdout).not.toContain('"before"') + + const rejected = yield* opencode.run("unknown model", { + model: "test/nonexistent-model", + format: "json", + extraArgs: ["--attach", proxy.url.origin], + }) + expect(rejected.exitCode).not.toBe(0) + const errors = opencode.parseJsonEvents(rejected.stdout) + expect(errors.length).toBeGreaterThan(0) + expect(errors.every((event) => event.type === "error")).toBe(true) + }), + 60_000, + ) + } + cliIt.concurrent( "attach mode rejects local directories before prompt admission", ({ home, opencode }) =>