Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 45 additions & 7 deletions packages/opencode/src/cli/cmd/run.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -690,12 +690,14 @@ export const RunCommand = effectCmd({
return false
}

const toggles = new Map<string, boolean>()
const printed = new Set<string>()

// 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<ReturnType<typeof sdk.event.subscribe>>) {
const toggles = new Map<string, boolean>()
const sessions = new Set([sessionID])
let error: string | undefined

Expand All @@ -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
Expand Down Expand Up @@ -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<Event> {
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) {
Expand All @@ -856,7 +894,7 @@ export const RunCommand = effectCmd({
process.exitCode = 1
return
}
await finish()
await finish(result.data?.info.parentID)
return
}

Expand All @@ -873,7 +911,7 @@ export const RunCommand = effectCmd({
process.exitCode = 1
return
}
await finish()
await finish(result.data?.info.parentID)
return
}

Expand Down
100 changes: 100 additions & 0 deletions packages/opencode/test/cli/run/run-process.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -299,13 +300,112 @@ 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}`)
}),
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 }) =>
Expand Down
Loading