Skip to content
Closed
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
9 changes: 9 additions & 0 deletions packages/core/src/session/runner/llm.ts
Original file line number Diff line number Diff line change
Expand Up @@ -514,6 +514,15 @@ const layer = Layer.effect(
if (steer) return { continue: true, step: step + 1, promotion: "steer" as SessionInput.Delivery }
const queue = yield* SessionInput.hasPending(db, sessionID, "queue")
if (queue) return { continue: true, step: 1, promotion: "queue" as SessionInput.Delivery }
// A step ending is not a turn ending, because a steer or a queued prompt continues the same
// turn through another step. This is the one place that knows the turn is over, for both
// modes. Only the ordinary ending: a turn the user stopped, or one a provider error ended,
// does not reach here, so a follower still needs its other reasons to stop waiting.
yield* events.publish(SessionEvent.Turn.Ended, {
sessionID,
timestamp: yield* DateTime.now,
finish: "stop",
})
return { continue: false, step: step + 1, promotion: undefined }
})

Expand Down
21 changes: 20 additions & 1 deletion packages/core/test/session-runner-model-call.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ import { eq } from "drizzle-orm"
import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat"
import { Auth } from "@opencode-ai/llm/route"
import { describe, expect, spyOn } from "bun:test"
import { Cause, DateTime, Deferred, Effect, Exit, Layer, Schema, Stream } from "effect"
import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect"
import { emptyStep, runnerHarness, seedSession } from "./lib/runner-harness"

// One tool call, then a clean finish: the shape a step that wants to keep going produces.
Expand Down Expand Up @@ -309,6 +309,25 @@ describe("SessionRunner model-only attempt", () => {
expect(closed(context)).toBe(true)
}),
)

// A step ending is not a turn ending: a steer or a queued prompt continues the same turn through
// another step, so the one place that decides publishes the turn's own ending.
runnerHarness(textOnly).effect("says the turn ended, once, when nothing follows it", () =>
Effect.gen(function* () {
yield* seedSession(sessionID)
const events = yield* EventV2.Service
const ended = yield* events
.subscribe(SessionEvent.Turn.Ended)
.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
const runner = yield* SessionRunner.Service

yield* runner.runStep({ sessionID, step: 2, promotion: undefined, first: false, force: false })

const seen = yield* Fiber.join(ended)
expect(seen.length).toBe(1)
expect(seen[0]?.data.sessionID).toBe(sessionID)
}),
)
})

// Dispatching one recorded call on its own. The policy under test is what happens when a dispatch
Expand Down
264 changes: 262 additions & 2 deletions packages/opencode/src/cli/cmd/detached.ts
Original file line number Diff line number Diff line change
@@ -1,9 +1,108 @@
// Commands about the deployment rather than about a session in front of you. `doctor` loads the
// Temporal driver only when it runs, so the rest of the `session` command never pays for it.
// Commands for a session nobody is sitting in front of: start one and walk away, ask the
// deployment what it is still running, and follow one from a machine that never had it.
//
// These are thin HTTP clients on purpose. In a durable deployment the serve processes are
// interchangeable (any of them reads the shared store and signals the same workflows), so a client
// needs an endpoint and a session id, never a particular host. That is the whole reason a session
// can outlive the process that started it, and it is why `start`, `running` and `watch` never talk
// to Temporal. `doctor` does, and loads the driver only when it runs, so the rest of the `session`
// command never pays for it.

import type { Argv } from "yargs"
import { cmd } from "./cmd"
import { UI } from "../ui"
import { ServerAuth } from "@/server/auth"

const DEFAULT_URL = "http://127.0.0.1:4096"

type Remote = { readonly url: string; readonly headers: Record<string, string> }

function remote(args: { attach?: string; password?: string; username?: string }): Remote {
const url = (args.attach ?? process.env["OPENCODE_SERVER"] ?? DEFAULT_URL).replace(/\/+$/, "")
// No password configured is a valid deployment, so absent auth is absent headers, not an error.
return { url, headers: ServerAuth.headers({ password: args.password, username: args.username }) ?? {} }
}

async function call<T>(r: Remote, path: string, init?: RequestInit): Promise<T> {
const response = await fetch(`${r.url}/api${path}`, {
...init,
headers: { ...r.headers, ...(init?.body ? { "content-type": "application/json" } : {}), ...init?.headers },
})
if (!response.ok) {
const detail = await response.text().catch(() => "")
throw new Error(`${init?.method ?? "GET"} /api${path} failed: ${response.status} ${detail.slice(0, 200)}`)
}
if (response.status === 204) return undefined as T
const body = (await response.json()) as { data: T }
return body.data
}

// The remote-facing options every command here shares. Kept in one builder so a second endpoint
// flag can never drift between them.
function remoteOptions(yargs: Argv) {
return yargs
.option("attach", {
type: "string",
describe: `server to talk to (default ${DEFAULT_URL}, or $OPENCODE_SERVER)`,
})
.option("password", { alias: "p", type: "string", describe: "basic auth password" })
.option("username", { alias: "u", type: "string", describe: "basic auth username" })
.option("json", { type: "boolean", describe: "print machine-readable output", default: false })
}

interface SessionInfo {
id: string
title?: string
time?: { created?: number; updated?: number }
location?: { directory?: string }
}

const stamp = (ms?: number) => (ms ? new Date(ms).toISOString().replace("T", " ").slice(0, 19) : "")

// UI.println writes to stderr, which is right for a person and wrong for a pipe. Anything a script
// is meant to read goes to stdout instead.
const emit = (line: string) => process.stdout.write(line + "\n")

export const SessionStartCommand = cmd({
command: "start <prompt>",
describe: "start a session, hand it a prompt, and return without waiting for it",
builder: (yargs: Argv) =>
remoteOptions(yargs)
.positional("prompt", { type: "string", describe: "what the agent should do", demandOption: true })
.option("dir", { type: "string", describe: "session working directory (default: this one)" })
.option("model", { type: "string", describe: "provider/model, e.g. openai/gpt-5-mini" }),
handler: async (args) => {
const r = remote(args)
try {
const directory = args.dir ?? process.cwd()
const session = await call<SessionInfo>(r, "/session", {
method: "POST",
body: JSON.stringify({ directory }),
})
if (args.model) {
const slash = args.model.indexOf("/")
if (slash < 1) throw new Error(`--model wants provider/model, got ${args.model}`)
const model = { providerID: args.model.slice(0, slash), id: args.model.slice(slash + 1) }
await call(r, `/session/${session.id}/model`, { method: "POST", body: JSON.stringify({ model }) })
}
// The prompt is admitted, not awaited. Whoever is polling the task queue runs the turn, and
// this process has nothing left to do with it.
await call(r, `/session/${session.id}/prompt`, {
method: "POST",
body: JSON.stringify({ prompt: { text: args.prompt } }),
})
if (args.json) {
emit(JSON.stringify({ id: session.id, directory, url: r.url }))
return
}
emit(session.id)
UI.println(` follow it with: opencode session watch ${session.id}`)
} catch (error) {
UI.error(error instanceof Error ? error.message : String(error))
process.exitCode = 1
}
},
})

// What this process resolved, and what is wrong with it. Deploying was a handful of variables that
// have to agree, with no way to ask whether they did: every mistake in them fails as something
Expand All @@ -29,3 +128,164 @@ export const SessionDoctorCommand = cmd({
UI.println("this deployment looks consistent")
},
})

export const SessionRunningCommand = cmd({
command: "running",
describe: "list the sessions this deployment is executing right now",
builder: (yargs: Argv) => remoteOptions(yargs),
handler: async (args) => {
const r = remote(args)
try {
// Which sessions are running is the executor's answer, not a guess from the transcript: a
// durable deployment reads it from the running workflows, so it survives a restart of
// whichever process happens to be answering this call.
const active = await call<Record<string, { type: string }>>(r, "/session/active")
const ids = Object.keys(active)
if (args.json) {
emit(JSON.stringify(ids.map((id) => ({ id, status: active[id]?.type }))))
return
}
if (ids.length === 0) {
UI.println("nothing running")
return
}
const all = await call<SessionInfo[]>(r, "/session").catch(() => [] as SessionInfo[])
const byId = new Map(all.map((s) => [s.id, s]))
for (const id of ids) {
const session = byId.get(id)
const cells = [id, active[id]?.type ?? "?", stamp(session?.time?.updated), session?.title ?? ""]
emit(cells.join(" "))
}
} catch (error) {
UI.error(error instanceof Error ? error.message : String(error))
process.exitCode = 1
}
},
})

// What a follower prints. The stream carries far more than a person watching wants to read, so this
// keeps the events that say the work moved and drops the token-level ones.
const INTERESTING: Record<string, (data: any) => string | undefined> = {
"session.next.prompted": () => "prompted",
"session.next.step.started": () => "step",
"session.next.tool.called": (d) => `tool ${d.tool}: ${JSON.stringify(d.input ?? {}).slice(0, 120)}`,
"session.next.tool.success": (d) => `tool ok ${firstText(d.content).slice(0, 200)}`,
"session.next.tool.failed": (d) => `tool failed ${firstText(d.content).slice(0, 200)}`,
"session.next.text.ended": (d) => (d.text ? `said: ${String(d.text).slice(0, 400)}` : undefined),
"session.next.step.failed": (d) => `step failed: ${d.error?.message ?? ""}`,
}

function firstText(content: unknown): string {
if (!Array.isArray(content)) return ""
const part = content.find((c) => c && typeof c === "object" && (c as any).type === "text") as any
return part?.text ? String(part.text).trim() : ""
}

export const SessionWatchCommand = cmd({
command: "watch <sessionID>",
describe: "follow a running session from anywhere, and exit when its turn ends",
builder: (yargs: Argv) =>
remoteOptions(yargs)
.positional("sessionID", { type: "string", describe: "session to follow", demandOption: true })
.option("wait", { type: "boolean", default: true, describe: "keep following until the turn ends" }),
handler: async (args) => {
const r = remote(args)
const sessionID = args.sessionID

// The turn saying so itself. A step ending is not a turn ending, because a steer or a queued
// prompt continues the same turn, so this is the one event that can end a follow on its own.
const turnEnded = (event: { type?: string }) => event.type === "session.next.turn.ended"

// The event above cannot answer every case. A turn the user stopped, or one a provider error
// ended, publishes no ordinary ending, and the running set holds a session for its whole idle
// timeout after the work is done, so it is slow. It is also the one reading that only ever says
// "finished", which is why it can end a watch without being able to hang one.
const POLL_MS = Number(process.env.OPENCODE_WATCH_POLL_MS ?? 30_000)
const inactive = async () => {
const active = await call<Record<string, unknown>>(r, "/session/active").catch(() => undefined)
// Unreachable is not finished, which is the whole point of this command.
return active !== undefined && !(sessionID in active)
}

// The stream ends when the serve this is attached to restarts, which is the event this command
// exists for. Exiting 0 there reports a turn that is still running as done.
const attempts = 30
try {
for (let attempt = 0; ; attempt++) {
const response = await fetch(`${r.url}/api/session/${sessionID}/event`, {
headers: r.headers,
}).catch(() => undefined)
if (!response?.ok || !response.body) {
if (attempt >= attempts) throw new Error(`cannot follow ${sessionID}: ${response?.status ?? "unreachable"}`)
await new Promise((resolve) => setTimeout(resolve, 1000))
continue
}

const reader = response.body.getReader()
const decoder = new TextDecoder()
let buffer = ""
let ended = false
// Held across iterations rather than started fresh each time: a read that loses the race is
// still queued on the stream, and dropping it drops whatever it goes on to deliver.
let pending: ReturnType<typeof reader.read> | undefined
for (;;) {
const next = (pending ??= reader.read())
let timer: ReturnType<typeof setTimeout> | undefined
const chunk = args.wait
? await Promise.race([
next,
new Promise<"quiet">((resolve) => {
timer = setTimeout(() => resolve("quiet"), POLL_MS)
}),
])
: await next
if (timer) clearTimeout(timer)
if (chunk === "quiet") {
if (await inactive()) {
ended = true
break
}
continue
}
pending = undefined
const { done, value } = chunk
if (done) break
buffer += decoder.decode(value, { stream: true })
const lines = buffer.split("\n")
buffer = lines.pop() ?? ""
for (const raw of lines) {
const line = raw.startsWith("data:") ? raw.slice(5).trim() : raw.trim()
if (!line.startsWith("{")) continue
let event: { type?: string; data?: any }
try {
event = JSON.parse(line)
} catch {
continue
}
if (args.json) {
emit(line)
} else {
const render = event.type ? INTERESTING[event.type] : undefined
const text = render?.(event.data ?? {})
if (text) UI.println(`${stamp(event.data?.timestamp)} ${text}`)
}
if (args.wait && turnEnded(event)) {
ended = true
break
}
}
if (ended) break
}
await reader.cancel().catch(() => {})
// Without `--wait` the stream itself is the whole command, so its end is this one's too.
if (ended || !args.wait) return
// The stream dropped with the turn still going. Reconnect and keep following.
if (attempt >= attempts) throw new Error(`lost the stream for ${sessionID}`)
await new Promise((resolve) => setTimeout(resolve, 1000))
}
} catch (error) {
UI.error(error instanceof Error ? error.message : String(error))
process.exitCode = 1
}
},
})
11 changes: 9 additions & 2 deletions packages/opencode/src/cli/cmd/session.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import type { Argv } from "yargs"
import { Effect } from "effect"
import { cmd } from "./cmd"
import { SessionDoctorCommand } from "./detached"
import { SessionDoctorCommand, SessionRunningCommand, SessionStartCommand, SessionWatchCommand } from "./detached"
import { effectCmd, fail } from "../effect-cmd"
import { Session } from "@/session/session"
import { SessionID } from "../../session/schema"
Expand Down Expand Up @@ -46,7 +46,14 @@ export const SessionCommand = cmd({
command: "session",
describe: "manage sessions",
builder: (yargs: Argv) =>
yargs.command(SessionListCommand).command(SessionDeleteCommand).command(SessionDoctorCommand).demandCommand(),
yargs
.command(SessionListCommand)
.command(SessionDeleteCommand)
.command(SessionStartCommand)
.command(SessionRunningCommand)
.command(SessionWatchCommand)
.command(SessionDoctorCommand)
.demandCommand(),
async handler() {},
})

Expand Down
19 changes: 19 additions & 0 deletions packages/schema/src/session-event.ts
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,24 @@ export namespace Shell {
export type Ended = typeof Ended.Type
}

// The turn, as opposed to the steps it was made of. A step ending is not a turn ending: a steer or
// a queued prompt continues the same turn through another step, so a follower cannot tell the two
// apart from a finish reason. Durable, because the follower that needs it is another process
// reading the replayable stream, which is built from the rows and nothing else.
export namespace Turn {
export const Ended = Event.define({
type: "session.next.turn.ended",
...options,
schema: {
...Base,
// What the last step of the turn came to, so a follower can say why it stopped rather than
// only that it did.
finish: Schema.String,
},
})
export type Ended = typeof Ended.Type
}

export namespace Step {
export const Started = Event.define({
type: "session.next.step.started",
Expand Down Expand Up @@ -458,6 +476,7 @@ export const DurableDefinitions = Event.inventory(
Step.Started,
Step.Ended,
Step.Failed,
Turn.Ended,
Text.Started,
Text.Ended,
Tool.Input.Started,
Expand Down
Loading
Loading