mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-14 07:54:24 +00:00
fix(cli): rewrite noninteractive run and remove the V1 bridge (#61911)
This commit is contained in:
Aiden Cline
GitHub
parent
0dffb96c4b
commit
a950fa2741
12 changed files with 1077 additions and 1147 deletions
|
|
@ -13,7 +13,6 @@
|
|||
],
|
||||
"exports": {
|
||||
"./vite-host": "./dev/host.js",
|
||||
"./run": "./src/run/index.ts",
|
||||
"./server-process": "./src/server-process.ts"
|
||||
},
|
||||
"scripts": {
|
||||
|
|
|
|||
|
|
@ -12,23 +12,33 @@ export default Runtime.handler(Commands.commands.run, (input) =>
|
|||
Effect.catch((error) => Effect.sync(() => reportRunError(input, error.message))),
|
||||
)
|
||||
if (!server) return
|
||||
yield* Effect.promise(() =>
|
||||
runNonInteractive({
|
||||
server,
|
||||
message: [...input.message],
|
||||
continue: input.continue,
|
||||
session: Option.getOrUndefined(input.session),
|
||||
fork: input.fork,
|
||||
worktree: Option.getOrUndefined(input.worktree),
|
||||
model: Option.getOrUndefined(input.model),
|
||||
agent: Option.getOrUndefined(input.agent),
|
||||
format: input.format,
|
||||
file: [...input.file],
|
||||
title: Option.getOrUndefined(input.title),
|
||||
archived: input.archived,
|
||||
thinking: input.thinking,
|
||||
auto: input.auto || input.yolo || input.dangerouslySkipPermissions,
|
||||
}),
|
||||
)
|
||||
// SIGINT and SIGTERM interrupt this fiber. Interruption waits for the run to cancel its Session and
|
||||
// jobs, so the private server and the process shut down only after that.
|
||||
yield* Effect.callback<void>((resume) => {
|
||||
const controller = new AbortController()
|
||||
const done = runNonInteractive(
|
||||
{
|
||||
server,
|
||||
message: [...input.message],
|
||||
continue: input.continue,
|
||||
session: Option.getOrUndefined(input.session),
|
||||
fork: input.fork,
|
||||
worktree: Option.getOrUndefined(input.worktree),
|
||||
model: Option.getOrUndefined(input.model),
|
||||
agent: Option.getOrUndefined(input.agent),
|
||||
format: input.format,
|
||||
file: [...input.file],
|
||||
title: Option.getOrUndefined(input.title),
|
||||
archived: input.archived,
|
||||
thinking: input.thinking,
|
||||
auto: input.auto || input.yolo || input.dangerouslySkipPermissions,
|
||||
},
|
||||
controller.signal,
|
||||
).then(() => resume(Effect.void))
|
||||
return Effect.promise(() => {
|
||||
controller.abort()
|
||||
return done
|
||||
})
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
import { ensurePluginRuntime } from "@opencode/plugin/runtime"
|
||||
import { NodeRuntime, NodeServices } from "@effect/platform-node"
|
||||
import { Cause, Effect } from "effect"
|
||||
import { getErrorReported } from "effect/Runtime"
|
||||
import { defaultTeardown, getErrorReported } from "effect/Runtime"
|
||||
import { Commands } from "./commands/commands"
|
||||
import { Runtime } from "./framework/runtime"
|
||||
import { Observability } from "@opencode/util/observability"
|
||||
|
|
@ -39,6 +39,8 @@ if (process.env.OPENCODE_SSH_ASKPASS_PORT) {
|
|||
process.exit(await Effect.runPromise(askpass.pipe(Effect.provide(NodeServices.layer))))
|
||||
}
|
||||
|
||||
let terminated = false
|
||||
|
||||
const Handlers = Runtime.handlers(Commands, {
|
||||
$: () => import("./commands/handlers/default"),
|
||||
upgrade: () => import("./commands/handlers/upgrade"),
|
||||
|
|
@ -100,6 +102,10 @@ Effect.gen(function* () {
|
|||
yield* Heap.listen
|
||||
yield* CpuProfile.listen
|
||||
yield* ShutdownDeadline.install()
|
||||
// runMain reports every signal interruption as 130; remember a SIGTERM so the exit code is 143.
|
||||
process.once("SIGTERM", () => {
|
||||
terminated = true
|
||||
})
|
||||
const runFork = Effect.runForkWith(yield* Effect.context<never>())
|
||||
const uncaughtException = (cause: Error, origin: "uncaughtException" | "unhandledRejection") => {
|
||||
runFork(Effect.logError("uncaught exception", { cause, origin }))
|
||||
|
|
@ -152,7 +158,6 @@ Effect.gen(function* () {
|
|||
),
|
||||
Effect.provide(NodeServices.layer),
|
||||
Effect.scoped,
|
||||
Effect.tap(() => Effect.sync(() => process.exit(process.exitCode ?? 0))),
|
||||
// runMain's default reporter logs the fatal cause to stdout. Write it to stderr instead: the
|
||||
// desktop and `Service.ensure` only capture stderr from `serve --service`, so this is the only
|
||||
// channel through which a startup failure's reason reaches the user.
|
||||
|
|
@ -163,5 +168,16 @@ Effect.gen(function* () {
|
|||
process.stderr.write(Cause.pretty(cause) + "\n")
|
||||
}),
|
||||
),
|
||||
NodeRuntime.runMain({ disableErrorReporting: true }),
|
||||
NodeRuntime.runMain({
|
||||
disableErrorReporting: true,
|
||||
teardown: (exit) =>
|
||||
defaultTeardown(exit, (code) => {
|
||||
// Commands report failures through process.exitCode.
|
||||
const status = code === 0 ? (process.exitCode ?? 0) : code === 130 && terminated ? 143 : code
|
||||
// process.exit drops output still queued for a pipe; ending the streams waits for it.
|
||||
void Promise.all(
|
||||
[process.stdout, process.stderr].map((stream) => new Promise<void>((resolve) => stream.end(() => resolve()))),
|
||||
).then(() => process.exit(status))
|
||||
}),
|
||||
}),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,2 +0,0 @@
|
|||
export { runNonInteractive, type RunCommandInput } from "./run"
|
||||
export { runV1Bridge, type V1RunCommandInput } from "./v1"
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -4,7 +4,7 @@ import { FSUtil } from "@opencode/util/fs-util"
|
|||
import { noTimeoutFetch } from "@opencode/util/no-timeout-fetch"
|
||||
import { open } from "node:fs/promises"
|
||||
import path from "node:path"
|
||||
import { readStdin } from "../util/io"
|
||||
import { readStdin, readStdinIfReady } from "../util/io"
|
||||
import { ServerConnection } from "../services/server-connection"
|
||||
import { createWorktreeLocation, parseSessionTargetModel, resolveSessionTarget } from "../session-target"
|
||||
import { toolInlineInfo } from "@opencode/tui/mini/tool"
|
||||
|
|
@ -30,27 +30,14 @@ export type RunCommandInput = {
|
|||
auto?: boolean
|
||||
}
|
||||
|
||||
type FilePart = {
|
||||
url: string
|
||||
filename: string
|
||||
mime: string
|
||||
}
|
||||
type FilePart = { text: string } | { attachment: { uri: string; name: string } }
|
||||
|
||||
type Prepared = {
|
||||
directory?: string
|
||||
directory: string
|
||||
message: string
|
||||
files: FilePart[]
|
||||
}
|
||||
|
||||
type ExecutionOptions = {
|
||||
root?: string
|
||||
directory?: string
|
||||
useServerDirectory?: boolean
|
||||
variant?: string
|
||||
attached?: boolean
|
||||
compatibility?: "v1"
|
||||
}
|
||||
|
||||
class RunTargetError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
|
|
@ -61,31 +48,34 @@ class RunTargetError extends Error {
|
|||
}
|
||||
|
||||
const ATTACH_FILE_MAX_BYTES = 10 * 1024 * 1024
|
||||
const PIPED_INPUT_GRACE_MS = 1000
|
||||
|
||||
export function runNonInteractive(input: RunCommandInput) {
|
||||
return runNonInteractiveWithOptions(input, {})
|
||||
/** `signal` aborts on SIGINT or SIGTERM; the run then cancels its Session and background jobs and returns. */
|
||||
export function runNonInteractive(input: RunCommandInput, signal: AbortSignal) {
|
||||
return run(input, signal).catch((error) => reportRunError(input, errorMessage(error)))
|
||||
}
|
||||
|
||||
/** @internal Used only by the V1 command boundary. */
|
||||
export function runNonInteractiveWithOptions(input: RunCommandInput, options: ExecutionOptions) {
|
||||
return run(input, options).catch((error) => reportRunError(input, errorMessage(error)))
|
||||
}
|
||||
|
||||
async function run(input: RunCommandInput, options: ExecutionOptions) {
|
||||
async function run(input: RunCommandInput, signal: AbortSignal) {
|
||||
if (input.fork && !input.continue && !input.session) fail("--fork requires --continue or --session")
|
||||
if (input.worktree !== undefined && (input.continue || input.session))
|
||||
fail("--worktree starts a new session and cannot be combined with --continue or --session")
|
||||
const root = options.root ?? process.env.PWD ?? process.cwd()
|
||||
const local = localDirectory(root)
|
||||
const directory = options.useServerDirectory ? undefined : (options.directory ?? local)
|
||||
const message = mergeInput(formatMessage(input.message), process.stdin.isTTY ? undefined : await readStdin())
|
||||
const root = process.env.PWD ?? process.cwd()
|
||||
const directory = localDirectory(root)
|
||||
const argument = input.message.join(" ") || undefined
|
||||
// With a message argument, stdin is extra context: a shell pipe is read in full, but the open, silent
|
||||
// stdin that spawn's default stdio gives a child must not hold the run.
|
||||
const piped = process.stdin.isTTY
|
||||
? undefined
|
||||
: argument
|
||||
? await readStdinIfReady(PIPED_INPUT_GRACE_MS)
|
||||
: await readStdin()
|
||||
const message = mergeInput(argument, piped)
|
||||
if (!message?.trim()) fail("You must provide a message")
|
||||
const files = await Promise.all(input.file.map((file) => prepareFile(file, root, options)))
|
||||
const prepared = { directory, message, files }
|
||||
return execute(input, prepared, input.server.endpoint, options)
|
||||
const files = await Promise.all(input.file.map((file) => prepareFile(file, root)))
|
||||
return execute(input, { directory, message, files }, input.server.endpoint, signal)
|
||||
}
|
||||
|
||||
async function execute(input: RunCommandInput, prepared: Prepared, endpoint: Endpoint, options: ExecutionOptions) {
|
||||
async function execute(input: RunCommandInput, prepared: Prepared, endpoint: Endpoint, signal: AbortSignal) {
|
||||
const client = OpenCode.make({
|
||||
baseUrl: endpoint.url,
|
||||
headers: Service.headers(endpoint),
|
||||
|
|
@ -93,7 +83,7 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End
|
|||
fetch: noTimeoutFetch,
|
||||
})
|
||||
const explicit = parseRunModel(input.model)
|
||||
const location = prepared.directory ? { directory: prepared.directory } : undefined
|
||||
const location = { directory: prepared.directory }
|
||||
const worktree =
|
||||
input.worktree === undefined
|
||||
? undefined
|
||||
|
|
@ -115,23 +105,9 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End
|
|||
agent: input.agent,
|
||||
environment: input.server.service ? Env.session() : undefined,
|
||||
prepare: async (next) => {
|
||||
const selected =
|
||||
next.model ??
|
||||
(options.variant
|
||||
? await client.model
|
||||
.default({ location: { directory: next.location.directory } })
|
||||
.then((result) => result.data)
|
||||
: undefined)
|
||||
const model = selected
|
||||
? {
|
||||
providerID: selected.providerID,
|
||||
id: selected.id,
|
||||
variant: options.variant ?? ("variant" in selected ? selected.variant : undefined),
|
||||
}
|
||||
: undefined
|
||||
if ((options.variant ?? explicit?.variant) && !model)
|
||||
if (explicit?.variant && !next.model)
|
||||
throw new RunTargetError("Cannot select a variant before selecting a model", next.session?.id)
|
||||
return { model, agent: next.agent }
|
||||
return { model: next.model, agent: next.agent }
|
||||
},
|
||||
}).catch((error) => {
|
||||
if (!(error instanceof RunTargetError)) throw error
|
||||
|
|
@ -139,8 +115,6 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End
|
|||
return undefined
|
||||
})
|
||||
if (!target) return
|
||||
const model = target.model ? { providerID: target.model.providerID, modelID: target.model.id } : undefined
|
||||
const variant = target.model?.variant
|
||||
const title =
|
||||
!target.resume && input.title !== undefined
|
||||
? input.title || prepared.message.slice(0, 50) + (prepared.message.length > 50 ? "..." : "")
|
||||
|
|
@ -155,16 +129,15 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End
|
|||
client,
|
||||
sessionID: target.session.id,
|
||||
location: target.location,
|
||||
message: prepared.message,
|
||||
files: prepared.files,
|
||||
message: [prepared.message, ...prepared.files.flatMap((file) => ("text" in file ? [file.text] : []))].join("\n\n"),
|
||||
files: prepared.files.flatMap((file) => ("attachment" in file ? [file.attachment] : [])),
|
||||
agent: target.agent,
|
||||
model,
|
||||
variant,
|
||||
model: target.model,
|
||||
thinking: input.thinking ?? false,
|
||||
format: input.format,
|
||||
auto: input.auto ?? false,
|
||||
attached: options.attached ?? true,
|
||||
compatibility: options.compatibility,
|
||||
standalone: input.server.standalone ?? false,
|
||||
signal,
|
||||
renderTool: (part) => renderTool(part, target.location.directory),
|
||||
renderToolError: (part) => renderToolError(part, target.location.directory),
|
||||
}).catch((error) => reportRunError(input, errorMessage(error), target.session.id))
|
||||
|
|
@ -176,11 +149,6 @@ export function mergeInput(message: string | undefined, piped: string | undefine
|
|||
return message + "\n" + piped
|
||||
}
|
||||
|
||||
function formatMessage(message: string[]) {
|
||||
const value = message.map((part) => (part.includes(" ") ? `"${part.replace(/"/g, '\\"')}"` : part)).join(" ")
|
||||
return value || undefined
|
||||
}
|
||||
|
||||
function localDirectory(root: string) {
|
||||
try {
|
||||
process.chdir(root)
|
||||
|
|
@ -199,13 +167,11 @@ export function parseRunModel(value?: string) {
|
|||
}
|
||||
}
|
||||
|
||||
async function prepareFile(input: string, directory: string, options: ExecutionOptions): Promise<FilePart> {
|
||||
async function prepareFile(input: string, directory: string): Promise<FilePart> {
|
||||
const file = path.resolve(directory, input)
|
||||
const handle = await open(file, "r").catch(() => fail(`File not found: ${input}`))
|
||||
try {
|
||||
const stat = await handle.stat()
|
||||
if (options.compatibility === "v1" && options.attached && stat.isDirectory())
|
||||
fail(`Cannot attach local directory without a shared filesystem: ${input}`)
|
||||
if (!stat.isFile() || stat.size > ATTACH_FILE_MAX_BYTES)
|
||||
fail(`Cannot attach a directory, special file, or file larger than 10 MiB: ${input}`)
|
||||
const content = Buffer.alloc(Number(stat.size))
|
||||
|
|
@ -224,11 +190,9 @@ async function prepareFile(input: string, directory: string, options: ExecutionO
|
|||
: !isBinaryContent(bytes) && Buffer.from(text, "utf8").equals(bytes)
|
||||
? "text/plain"
|
||||
: detected
|
||||
return {
|
||||
url: `data:${mime};base64,${bytes.toString("base64")}`,
|
||||
filename: path.basename(file),
|
||||
mime,
|
||||
}
|
||||
const name = path.basename(file)
|
||||
if (mime === "text/plain") return { text: `<file name="${name}">\n${text}\n</file>` }
|
||||
return { attachment: { uri: `data:${mime};base64,${bytes.toString("base64")}`, name } }
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
|
|
@ -261,7 +225,6 @@ async function renderToolError(part: SessionMessageAssistantTool, directory: str
|
|||
UI.println(UI.Style.TEXT_NORMAL + "✗", UI.Style.TEXT_NORMAL + `${info.title} failed`)
|
||||
}
|
||||
|
||||
/** @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
|
||||
if (input.format === "json") {
|
||||
|
|
|
|||
|
|
@ -1,86 +0,0 @@
|
|||
import type { Endpoint } from "@opencode/client/effect/service"
|
||||
import { Effect } from "effect"
|
||||
import path from "node:path"
|
||||
import { Standalone } from "../services/standalone"
|
||||
import { errorMessage } from "../util/error"
|
||||
import { reportRunError, runNonInteractiveWithOptions, type RunCommandInput } from "./run"
|
||||
|
||||
export type V1RunCommandInput = {
|
||||
message: string[]
|
||||
continue?: boolean
|
||||
session?: string
|
||||
fork?: boolean
|
||||
model?: string
|
||||
agent?: string
|
||||
format: "default" | "json"
|
||||
file: string[]
|
||||
title?: string
|
||||
server?: string
|
||||
password?: string
|
||||
username?: string
|
||||
directory?: string
|
||||
variant?: string
|
||||
thinking?: boolean
|
||||
dangerouslySkipPermissions?: boolean
|
||||
standaloneCommand?: ReadonlyArray<string>
|
||||
}
|
||||
|
||||
export function runV1Bridge(input: V1RunCommandInput) {
|
||||
const root = process.env.PWD ?? process.cwd()
|
||||
const attached = input.server !== undefined
|
||||
const local = !attached && input.directory ? path.resolve(root, input.directory) : root
|
||||
try {
|
||||
process.chdir(local)
|
||||
} catch {
|
||||
reportRunError(input, `Failed to change directory to ${local}`)
|
||||
return Promise.resolve()
|
||||
}
|
||||
|
||||
return Effect.runPromise(
|
||||
Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const endpoint = attached
|
||||
? explicitEndpoint(input)
|
||||
: yield* Standalone.start({ command: input.standaloneCommand })
|
||||
yield* Effect.promise(() =>
|
||||
runNonInteractiveWithOptions(nativeInput(input, endpoint), {
|
||||
root: local,
|
||||
directory: attached ? input.directory : local,
|
||||
useServerDirectory: attached && input.directory === undefined,
|
||||
variant: input.variant,
|
||||
attached,
|
||||
compatibility: "v1",
|
||||
}),
|
||||
)
|
||||
}),
|
||||
),
|
||||
).catch((error) => reportRunError(input, errorMessage(error)))
|
||||
}
|
||||
|
||||
function nativeInput(input: V1RunCommandInput, endpoint: Endpoint): RunCommandInput {
|
||||
return {
|
||||
server: { endpoint },
|
||||
message: input.message,
|
||||
continue: input.continue,
|
||||
session: input.session,
|
||||
fork: input.fork,
|
||||
model: input.model,
|
||||
agent: input.agent,
|
||||
format: input.format,
|
||||
file: input.file,
|
||||
title: input.title,
|
||||
thinking: input.thinking,
|
||||
auto: input.dangerouslySkipPermissions,
|
||||
}
|
||||
}
|
||||
|
||||
function explicitEndpoint(input: V1RunCommandInput): Endpoint {
|
||||
const url = input.server
|
||||
if (!url) throw new Error("Missing V1 server URL")
|
||||
return {
|
||||
url,
|
||||
auth: input.password
|
||||
? { type: "basic", username: input.username ?? "opencode", password: input.password }
|
||||
: undefined,
|
||||
}
|
||||
}
|
||||
|
|
@ -20,6 +20,8 @@ export type Args = {
|
|||
export type Resolved = {
|
||||
readonly endpoint: Endpoint
|
||||
readonly service?: ReturnType<typeof managedService>
|
||||
/** A private server started for this command alone. */
|
||||
readonly standalone?: boolean
|
||||
}
|
||||
|
||||
export const resolve = Effect.fn("cli.server-connection.resolve")(function* (args: Args = {}) {
|
||||
|
|
@ -47,14 +49,14 @@ export const resolve = Effect.fn("cli.server-connection.resolve")(function* (arg
|
|||
return { endpoint } satisfies Resolved
|
||||
}
|
||||
if (standalone || (yield* ServiceConfig.read()).disabled === true) {
|
||||
return { endpoint: yield* Standalone.start() } satisfies Resolved
|
||||
return { endpoint: yield* Standalone.start(), standalone: true } satisfies Resolved
|
||||
}
|
||||
|
||||
const mismatch = args.mismatch ?? "ignore"
|
||||
const options = yield* ServiceConfig.options({ checkVersion: mismatch !== "ignore" })
|
||||
if (args.autostart === false) {
|
||||
const endpoint = yield* Service.discover({ ...options, version: undefined })
|
||||
if (!endpoint) return { endpoint: yield* Standalone.start() } satisfies Resolved
|
||||
if (!endpoint) return { endpoint: yield* Standalone.start(), standalone: true } satisfies Resolved
|
||||
return { endpoint, service: managedService(options) } satisfies Resolved
|
||||
}
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,24 @@
|
|||
import { once } from "node:events"
|
||||
import { fstatSync } from "node:fs"
|
||||
import { text } from "node:stream/consumers"
|
||||
|
||||
export function readStdin() {
|
||||
return text(process.stdin)
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads a shell pipe or redirected file in full, however slow its producer. Any other stdin (the socket
|
||||
* that spawn's default stdio gives a child) is read only if it has data or ends within `grace`
|
||||
* milliseconds; otherwise it is closed unread.
|
||||
*/
|
||||
export async function readStdinIfReady(grace: number) {
|
||||
const stat = fstatSync(0)
|
||||
if (stat.isFIFO() || stat.isFile()) return readStdin()
|
||||
const ready = await once(process.stdin, "readable", { signal: AbortSignal.timeout(grace) }).then(
|
||||
() => true,
|
||||
() => false,
|
||||
)
|
||||
if (ready) return readStdin()
|
||||
process.stdin.destroy()
|
||||
return undefined
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,12 +26,10 @@ describe("CLI frontend import boundaries", () => {
|
|||
})
|
||||
|
||||
test("exposes only the intentional package entrypoints", async () => {
|
||||
const run = await import("@opencode/cli/run")
|
||||
const mini = await import("@opencode/tui/mini")
|
||||
const tool = await import("@opencode/tui/mini/tool")
|
||||
const cli = await Bun.file(path.join(root, "packages/cli/package.json")).json()
|
||||
|
||||
expect(Object.keys(run).sort()).toEqual(["runNonInteractive", "runV1Bridge"])
|
||||
expect(Object.keys(mini).sort()).toEqual(["runMiniFrontend"])
|
||||
expect(Object.keys(tool).sort()).toEqual([
|
||||
"nonEmptyToolContent",
|
||||
|
|
|
|||
|
|
@ -1,8 +1,11 @@
|
|||
import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"
|
||||
import {
|
||||
ClientError,
|
||||
OpenCode,
|
||||
type EventSubscribeOutput,
|
||||
type ExperimentalJobListOutput,
|
||||
type SessionActiveOutput,
|
||||
type SessionInboxListOutput,
|
||||
type SessionMessageAssistantTool,
|
||||
type SessionMessageInfo,
|
||||
} from "@opencode/client/promise"
|
||||
|
|
@ -100,6 +103,50 @@ function executionFailed(message: string): V2Event {
|
|||
}
|
||||
}
|
||||
|
||||
function idleMessage(outcome: "succeeded" | "failed" | "interrupted", id = "msg_idle") {
|
||||
return {
|
||||
id,
|
||||
type: "idle",
|
||||
outcome,
|
||||
...(outcome === "failed" ? { error: { type: "provider.transport", message: "Execution failed" } } : {}),
|
||||
time: { created: 5 },
|
||||
} satisfies SessionMessageInfo
|
||||
}
|
||||
|
||||
// A step that failed and was retried, followed by the step that recovered.
|
||||
function recoveredMessages(outcome: "succeeded" | "failed" | "interrupted") {
|
||||
return (messageID: string): SessionMessageInfo[] => [
|
||||
idleMessage(outcome),
|
||||
{
|
||||
id: "msg_recovered",
|
||||
type: "assistant",
|
||||
agent: "build",
|
||||
model: { providerID: "test", id: "test-model" },
|
||||
content: [{ type: "text", text: "Recovered answer" }],
|
||||
finish: "stop",
|
||||
time: { created: 3, completed: 4 },
|
||||
},
|
||||
{
|
||||
id: "msg_assistant",
|
||||
type: "assistant",
|
||||
agent: "build",
|
||||
model: { providerID: "test", id: "test-model" },
|
||||
content: [],
|
||||
finish: "error",
|
||||
error: { type: "provider.transport", message: "Recovered transport error" },
|
||||
time: { created: 2, completed: 3 },
|
||||
},
|
||||
{ id: messageID, type: "user", text: "hello", time: { created: 1 } },
|
||||
]
|
||||
}
|
||||
|
||||
function jsonLines(stdout: string) {
|
||||
return stdout
|
||||
.split("\n")
|
||||
.filter(Boolean)
|
||||
.map((line) => JSON.parse(line))
|
||||
}
|
||||
|
||||
function failedTool(inboxID: string): V2Event[] {
|
||||
return [
|
||||
prompted(inboxID),
|
||||
|
|
@ -210,16 +257,18 @@ function successfulGrep(inboxID: string): V2Event[] {
|
|||
async function run(input: {
|
||||
turn: (inboxID: string) => V2Event[]
|
||||
pendingForms?: FormInfo[]
|
||||
attached?: boolean
|
||||
standalone?: boolean
|
||||
format?: "default" | "json"
|
||||
compatibility?: "v1"
|
||||
cancel?: (input: { sessionID: string; formID: string }) => Promise<void>
|
||||
renderTool?: (part: SessionMessageAssistantTool) => Promise<void>
|
||||
renderToolError?: (part: SessionMessageAssistantTool) => Promise<void>
|
||||
messages?: (inboxID: string) => SessionMessageInfo[]
|
||||
wait?: () => Promise<void>
|
||||
jobs?: () => ExperimentalJobListOutput
|
||||
inbox?: (sessionID: string) => SessionInboxListOutput
|
||||
active?: () => SessionActiveOutput
|
||||
terminalDelay?: number
|
||||
signal?: AbortSignal
|
||||
}) {
|
||||
const sdk = OpenCode.make({ baseUrl: "https://opencode.test" })
|
||||
const values: V2Event[] = [{ id: "evt_connected", type: "server.connected", data: {} }]
|
||||
|
|
@ -256,9 +305,14 @@ async function run(input: {
|
|||
spyOn(sdk.session.form, "cancel").mockImplementation((request) => (input.cancel?.(request) ?? ok(undefined)) as never)
|
||||
let promptID = "msg_prompt"
|
||||
spyOn(sdk.experimental.session, "wait").mockImplementation(() => input.wait?.() ?? wait.promise)
|
||||
spyOn(sdk.experimental.job, "list").mockImplementation(() => ok(input.jobs?.() ?? []) as never)
|
||||
spyOn(sdk.session, "active").mockImplementation(() => ok({}) as never)
|
||||
spyOn(sdk.session.inbox, "list").mockImplementation(() => ok([]) as never)
|
||||
spyOn(sdk.experimental.job, "list").mockImplementation((async () => input.jobs?.() ?? []) as never)
|
||||
spyOn(sdk.session, "active").mockImplementation((async () => input.active?.() ?? {}) as never)
|
||||
spyOn(sdk.session.inbox, "list").mockImplementation(
|
||||
(async (request: { sessionID: string }) => input.inbox?.(request.sessionID) ?? []) as never,
|
||||
)
|
||||
spyOn(sdk.session.inbox, "cancel").mockImplementation(() => ok(undefined))
|
||||
spyOn(sdk.session, "interrupt").mockImplementation(() => ok({ interrupted: true }) as never)
|
||||
spyOn(sdk.shell, "remove").mockImplementation(() => ok(undefined) as never)
|
||||
spyOn(sdk.session, "get").mockImplementation(
|
||||
({ sessionID }) => ok({ id: sessionID, parentID: sessionID === "ses_child" ? "ses_1" : undefined }) as never,
|
||||
)
|
||||
|
|
@ -285,8 +339,8 @@ async function run(input: {
|
|||
thinking: false,
|
||||
format: input.format ?? "default",
|
||||
auto: false,
|
||||
attached: input.attached ?? false,
|
||||
compatibility: input.compatibility,
|
||||
standalone: input.standalone ?? false,
|
||||
signal: input.signal ?? new AbortController().signal,
|
||||
renderTool: input.renderTool ?? (() => Promise.resolve()),
|
||||
renderToolError: input.renderToolError ?? (() => Promise.resolve()),
|
||||
})
|
||||
|
|
@ -370,6 +424,7 @@ describe("runNonInteractivePrompt", () => {
|
|||
turn: (messageID) => [prompted(messageID)],
|
||||
wait: () => idle.promise,
|
||||
messages: (messageID) => [
|
||||
idleMessage("succeeded"),
|
||||
{
|
||||
id: "msg_assistant",
|
||||
type: "assistant",
|
||||
|
|
@ -405,6 +460,7 @@ describe("runNonInteractivePrompt", () => {
|
|||
test("reports usage for a step that ends after the Session goes idle", async () => {
|
||||
const tokens = { input: 3, output: 5, reasoning: 7, cache: { read: 11, write: 13 } }
|
||||
const assistant = (messageID: string): SessionMessageInfo[] => [
|
||||
idleMessage("succeeded"),
|
||||
{
|
||||
id: "msg_assistant",
|
||||
type: "assistant",
|
||||
|
|
@ -483,105 +539,30 @@ describe("runNonInteractivePrompt", () => {
|
|||
}
|
||||
})
|
||||
|
||||
test("cancels session and global form blockers and exits on pre-promotion interrupt", async () => {
|
||||
const sdk = await run({
|
||||
pendingForms: [form("frm_pending", "ses_1"), form("frm_pending_global", "global")],
|
||||
// No prompted event: the execution settles interrupted before promotion,
|
||||
// which must not leave the consume loop waiting forever.
|
||||
turn: () => [formCreated(form("frm_live", "global")), settled("interrupted")],
|
||||
for (const standalone of [true, false]) {
|
||||
test(`${standalone ? "a standalone" : "an attached"} run ${standalone ? "cancels" : "leaves"} MCP input requests in its location`, async () => {
|
||||
const cancelled: { sessionID: string; formID: string }[] = []
|
||||
const output = await capture({
|
||||
format: "json",
|
||||
standalone,
|
||||
pendingForms: [form("frm_pending", "ses_1"), form("frm_pending_global", "global")],
|
||||
turn: (messageID) => [
|
||||
formCreated(form("frm_live", "global")),
|
||||
formCreated(form("frm_elsewhere", "global"), { ...location, directory: "/elsewhere" }),
|
||||
prompted(messageID),
|
||||
settled(),
|
||||
],
|
||||
cancel: (request) => {
|
||||
cancelled.push(request)
|
||||
return Promise.resolve()
|
||||
},
|
||||
})
|
||||
const global = standalone ? ["frm_live", "frm_pending_global"] : []
|
||||
expect(cancelled.map((request) => request.formID).sort()).toEqual(["frm_pending", ...global].sort())
|
||||
expect(output.exitCode).toBe(1)
|
||||
expect(jsonLines(output.stdout).filter((event) => event.type === "error")).toHaveLength(1)
|
||||
})
|
||||
const globalOptions = {
|
||||
headers: {
|
||||
"x-opencode-directory": "%2Fwork%20tree",
|
||||
},
|
||||
}
|
||||
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.form.list).toHaveBeenCalledWith({
|
||||
location: { directory: "/work tree" },
|
||||
})
|
||||
expect(process.exitCode).toBe(1)
|
||||
})
|
||||
|
||||
test("attach mode cancels only session-owned forms", async () => {
|
||||
const sdk = await run({
|
||||
attached: true,
|
||||
pendingForms: [form("frm_pending", "ses_1"), form("frm_pending_global", "global")],
|
||||
turn: (messageID) => [formCreated(form("frm_live", "global")), prompted(messageID), settled()],
|
||||
})
|
||||
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_pending_global" },
|
||||
expect.anything(),
|
||||
)
|
||||
expect(process.exitCode).toBe(1)
|
||||
})
|
||||
|
||||
test("V1 JSON output flushes step_start before an unrelated step failure", async () => {
|
||||
const output = await capture({
|
||||
compatibility: "v1",
|
||||
format: "json",
|
||||
turn: (messageID) => [
|
||||
prompted(messageID),
|
||||
stepStarted(),
|
||||
stepFailed("Provider request failed"),
|
||||
executionFailed("Provider request failed"),
|
||||
],
|
||||
})
|
||||
|
||||
expect(
|
||||
output.stdout
|
||||
.split("\n")
|
||||
.filter(Boolean)
|
||||
.map((line) => JSON.parse(line)),
|
||||
).toEqual([
|
||||
expect.objectContaining({ type: "step_start", part: expect.objectContaining({ type: "step-start" }) }),
|
||||
expect.objectContaining({
|
||||
type: "error",
|
||||
error: { type: "provider.transport", message: "Provider request failed" },
|
||||
}),
|
||||
])
|
||||
expect(output.stderr).toBe("")
|
||||
const sdk = await run({ compatibility: "v1", turn: (messageID) => [prompted(messageID), settled()] })
|
||||
expect(sdk.experimental.session.wait).not.toHaveBeenCalled()
|
||||
expect(sdk.message.list).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
test("V1 default output flushes step_start before an unrelated execution failure", async () => {
|
||||
const output = await capture({
|
||||
compatibility: "v1",
|
||||
turn: (messageID) => [prompted(messageID), stepStarted(), executionFailed("Execution failed")],
|
||||
})
|
||||
|
||||
expect(output.stdout).toBe("")
|
||||
expect(output.stderr).toContain("> build · test-model")
|
||||
expect(output.stderr).toContain("Error: \u001b[0mExecution failed")
|
||||
expect(output.stderr.indexOf("> build · test-model")).toBeLessThan(output.stderr.indexOf("Execution failed"))
|
||||
})
|
||||
|
||||
test("V1 preserves terminal-finish failure suppression before content", async () => {
|
||||
const output = await capture({
|
||||
compatibility: "v1",
|
||||
format: "json",
|
||||
turn: (messageID) => [
|
||||
prompted(messageID),
|
||||
stepStarted(),
|
||||
stepFailed("The provider response ended unexpectedly."),
|
||||
executionFailed("The provider response ended unexpectedly."),
|
||||
],
|
||||
})
|
||||
|
||||
expect(output).toEqual({ stdout: "", stderr: "", exitCode: 0 })
|
||||
})
|
||||
}
|
||||
|
||||
test("renders a native terminal failure snapshot when live progress was missed", async () => {
|
||||
const rendered: SessionMessageAssistantTool[] = []
|
||||
|
|
@ -621,8 +602,8 @@ describe("runNonInteractivePrompt", () => {
|
|||
])
|
||||
})
|
||||
|
||||
test("keeps failed tool partial output out of the explicit V1 JSON bridge shape", async () => {
|
||||
const output = await capture({ compatibility: "v1", format: "json", turn: failedTool })
|
||||
test("keeps failed tool partial output out of the JSON tool_use shape", async () => {
|
||||
const output = await capture({ format: "json", turn: failedTool })
|
||||
const events = output.stdout
|
||||
.split("\n")
|
||||
.filter(Boolean)
|
||||
|
|
@ -689,11 +670,191 @@ describe("runNonInteractivePrompt", () => {
|
|||
expect(lists).toEqual([])
|
||||
})
|
||||
|
||||
test("Ctrl-C while waiting on a background subagent exits with code 130", async () => {
|
||||
const running = run({ turn: (id) => [prompted(id), settled()], jobs: () => [subagentJob] })
|
||||
test("a recovered step failure exits 0, reports its retry once, and still waits for background work", async () => {
|
||||
const lists: ExperimentalJobListOutput[] = [[shellJob("sh_train", "python train.py")], []]
|
||||
const retry = { attempt: 1, error: { type: "provider.transport", message: "Recovered transport error" } }
|
||||
const output = await capture({
|
||||
format: "json",
|
||||
jobs: () => lists.shift() ?? [],
|
||||
messages: recoveredMessages("succeeded"),
|
||||
turn: (id) => [
|
||||
prompted(id),
|
||||
stepStarted(),
|
||||
stepFailed("Recovered transport error"),
|
||||
{
|
||||
id: "evt_retry",
|
||||
created: 3,
|
||||
type: "session.retry.scheduled",
|
||||
durable: { aggregateID: "ses_1", seq: 3, version: 1 },
|
||||
data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", at: 2003, ...retry },
|
||||
},
|
||||
stepStarted(),
|
||||
settled(),
|
||||
{
|
||||
id: "evt_text",
|
||||
created: 4,
|
||||
type: "session.text.ended",
|
||||
durable: { aggregateID: "ses_1", seq: 4, version: 1 },
|
||||
data: { sessionID: "ses_1", assistantMessageID: "msg_2", ordinal: 0, text: "Training finished." },
|
||||
},
|
||||
settled(),
|
||||
],
|
||||
})
|
||||
const events = jsonLines(output.stdout)
|
||||
expect(output.exitCode).toBe(0)
|
||||
expect(events.filter((event) => event.type === "error" || event.type === "retry")).toEqual([
|
||||
{ type: "retry", timestamp: 3, sessionID: "ses_1", messageID: "msg_assistant", nextRetryAt: 2003, ...retry },
|
||||
])
|
||||
expect(
|
||||
events.filter((event) => event.part?.messageID === "msg_assistant" && event.type === "step_start"),
|
||||
).toHaveLength(1)
|
||||
expect(events.find((event) => event.type === "background_wait")).toMatchObject({
|
||||
jobs: [{ id: "sh_train", kind: "shell" }],
|
||||
})
|
||||
expect(output.stdout).toContain("Training finished.")
|
||||
expect(lists).toEqual([])
|
||||
})
|
||||
|
||||
for (const outcome of ["failed", "interrupted"] as const) {
|
||||
test(`a final ${outcome} outcome exits 1 with one error when the live event is dropped`, async () => {
|
||||
const output = await capture({
|
||||
format: "json",
|
||||
turn: (id) => [prompted(id)],
|
||||
// Without the failed step, so only the idle marker can report the outcome.
|
||||
messages: (id) => recoveredMessages(outcome)(id).filter((message) => message.id !== "msg_assistant"),
|
||||
wait: () => Promise.resolve(),
|
||||
})
|
||||
expect(output.exitCode).toBe(1)
|
||||
expect(jsonLines(output.stdout).filter((event) => event.type === "error")).toHaveLength(1)
|
||||
})
|
||||
}
|
||||
|
||||
test("only the first idle marker after the prompt decides the outcome", async () => {
|
||||
const output = await capture({
|
||||
format: "json",
|
||||
turn: (id) => [prompted(id)],
|
||||
wait: () => Promise.resolve(),
|
||||
messages: (messageID) => [
|
||||
idleMessage("failed", "msg_later_idle"),
|
||||
{ id: "msg_other_prompt", type: "user", text: "another client", time: { created: 6 } },
|
||||
...recoveredMessages("succeeded")(messageID),
|
||||
idleMessage("failed", "msg_previous_idle"),
|
||||
],
|
||||
})
|
||||
expect(output.exitCode).toBe(0)
|
||||
expect(output.stdout).not.toContain('"type":"error"')
|
||||
})
|
||||
|
||||
test("a prompt left in the inbox by a failure before promotion does not hang", async () => {
|
||||
const output = await capture({
|
||||
format: "json",
|
||||
turn: () => [executionFailed("instructions unavailable")],
|
||||
messages: () => [],
|
||||
inbox: (sessionID) => (sessionID === "ses_1" ? [{ id: "msg_prompt" } as SessionInboxListOutput[number]] : []),
|
||||
})
|
||||
expect(output.exitCode).toBe(1)
|
||||
expect(jsonLines(output.stdout)).toEqual([
|
||||
expect.objectContaining({
|
||||
type: "error",
|
||||
error: expect.objectContaining({ message: "instructions unavailable" }),
|
||||
}),
|
||||
])
|
||||
})
|
||||
|
||||
test("keeps waiting when a woken reply moves from the inbox to running between reads", async () => {
|
||||
// The wake leaves the inbox as it is read, and its reply runs until the next check.
|
||||
let wake: "queued" | "running" | "done" = "queued"
|
||||
const output = await capture({
|
||||
format: "json",
|
||||
inbox: () => {
|
||||
if (wake === "queued") wake = "running"
|
||||
return []
|
||||
},
|
||||
active: (): SessionActiveOutput => {
|
||||
if (wake !== "running") return {}
|
||||
wake = "done"
|
||||
return { ses_1: { type: "running" } }
|
||||
},
|
||||
turn: (id) => [
|
||||
prompted(id),
|
||||
settled(),
|
||||
{
|
||||
id: "evt_text",
|
||||
created: 2,
|
||||
type: "session.text.ended",
|
||||
durable: { aggregateID: "ses_1", seq: 2, version: 1 },
|
||||
data: { sessionID: "ses_1", assistantMessageID: "msg_2", ordinal: 0, text: "Shell finished." },
|
||||
},
|
||||
settled(),
|
||||
],
|
||||
})
|
||||
expect(output.stdout).toContain("Shell finished.")
|
||||
})
|
||||
|
||||
test("fails the run when a background state read fails instead of treating it as done", async () => {
|
||||
await expect(
|
||||
run({
|
||||
turn: (id) => [prompted(id), settled()],
|
||||
active: () => {
|
||||
throw new Error("state unavailable")
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("state unavailable")
|
||||
})
|
||||
|
||||
test("treats a server without the job list endpoint as having no jobs", async () => {
|
||||
const output = await capture({
|
||||
turn: (id) => [prompted(id), settled()],
|
||||
jobs: () => {
|
||||
throw new ClientError("UnsupportedContentType", { detail: "text/plain" })
|
||||
},
|
||||
})
|
||||
expect(output.exitCode).toBe(0)
|
||||
expect(output.stderr).toBe("")
|
||||
})
|
||||
|
||||
test("does not wait on a job started before this prompt", async () => {
|
||||
// A dev server left running by an earlier turn, as with --continue.
|
||||
const output = await capture({
|
||||
turn: (id) => [prompted(id), settled()],
|
||||
jobs: () => [{ ...shellJob("sh_dev", "bun dev"), time: { started: 0 } }],
|
||||
})
|
||||
expect(output.exitCode).toBe(0)
|
||||
expect(output.stderr).not.toContain("waiting for")
|
||||
})
|
||||
|
||||
test("cancelling while waiting on background work stops the Session and this prompt's jobs", async () => {
|
||||
const controller = new AbortController()
|
||||
const old = { ...shellJob("sh_dev", "bun dev"), time: { started: 0 } }
|
||||
const all = [old, shellJob("sh_train", "python train.py"), subagentJob]
|
||||
// After the cancel reads and stops this prompt's jobs, only the older dev server is left.
|
||||
let reads = 0
|
||||
const running = run({
|
||||
signal: controller.signal,
|
||||
turn: (id) => [prompted(id), settled()],
|
||||
jobs: () => (controller.signal.aborted && reads++ > 0 ? [old] : all),
|
||||
})
|
||||
expect(await Promise.race([running.then(() => "exited"), Bun.sleep(20).then(() => "waiting")])).toBe("waiting")
|
||||
process.emit("SIGINT")
|
||||
await running
|
||||
expect(process.exitCode).toBe(130)
|
||||
controller.abort()
|
||||
const sdk = await running
|
||||
expect(sdk.session.interrupt).toHaveBeenCalledWith({ sessionID: "ses_1" })
|
||||
expect(sdk.session.interrupt).toHaveBeenCalledWith({ sessionID: "ses_child" })
|
||||
expect(sdk.shell.remove).toHaveBeenCalledWith({ id: "sh_train", location: { directory: "/work tree" } })
|
||||
expect(sdk.shell.remove).not.toHaveBeenCalledWith(expect.objectContaining({ id: "sh_dev" }))
|
||||
})
|
||||
|
||||
test("cancelling mid-turn interrupts the Session and ignores the transport errors that follow", async () => {
|
||||
const controller = new AbortController()
|
||||
const closed = Promise.withResolvers<void>()
|
||||
const output = capture({
|
||||
signal: controller.signal,
|
||||
turn: (id) => [prompted(id)],
|
||||
// The private server shuts down while the idle wait is open.
|
||||
wait: () => closed.promise,
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
controller.abort()
|
||||
closed.reject(new ClientError("Transport", { detail: "ECONNRESET" }))
|
||||
expect(await output).toMatchObject({ stdout: "", stderr: "", exitCode: 0 })
|
||||
})
|
||||
})
|
||||
|
|
|
|||
141
packages/cli/test/run/process.test.ts
Normal file
141
packages/cli/test/run/process.test.ts
Normal file
|
|
@ -0,0 +1,141 @@
|
|||
import { expect, test } from "bun:test"
|
||||
import path from "node:path"
|
||||
import { isolatedEnv } from "../fixture/environment"
|
||||
import { tmpdir } from "../fixture/tmpdir"
|
||||
|
||||
// A server that admits the prompt and then keeps the Session running until interrupted.
|
||||
async function serve(directory: string) {
|
||||
const requests: { method: string; path: string; body?: unknown }[] = []
|
||||
const prompted = Promise.withResolvers<void>()
|
||||
const interrupted = Promise.withResolvers<void>()
|
||||
const location = { directory, project: { id: "project", directory, canonical: directory } }
|
||||
const server = Bun.serve({
|
||||
hostname: "127.0.0.1",
|
||||
port: 0,
|
||||
idleTimeout: 0,
|
||||
async fetch(request) {
|
||||
const url = new URL(request.url)
|
||||
const text = await request.text()
|
||||
requests.push({ method: request.method, path: url.pathname, body: text ? JSON.parse(text) : undefined })
|
||||
if (url.pathname === "/api/info")
|
||||
return Response.json({ version: "test", pid: 0, urls: [], paths: { tmp: directory } })
|
||||
if (request.method === "GET" && url.pathname === "/api/location") return Response.json(location)
|
||||
if (request.method === "POST" && url.pathname === "/api/session")
|
||||
return Response.json({
|
||||
data: {
|
||||
id: "ses_1",
|
||||
projectID: "project",
|
||||
location: { directory },
|
||||
cost: 0,
|
||||
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
|
||||
time: { created: 1, updated: 1 },
|
||||
},
|
||||
})
|
||||
if (url.pathname === "/api/event")
|
||||
return new Response(
|
||||
new ReadableStream({
|
||||
start(controller) {
|
||||
controller.enqueue(
|
||||
`data: ${JSON.stringify({ id: "evt_connected", type: "server.connected", data: {} })}\n\n`,
|
||||
)
|
||||
},
|
||||
}),
|
||||
{ headers: { "content-type": "text/event-stream" } },
|
||||
)
|
||||
if (url.pathname === "/api/session/ses_1/prompt") {
|
||||
prompted.resolve()
|
||||
const body = JSON.parse(text)
|
||||
return Response.json({
|
||||
data: { id: body.id, sessionID: "ses_1", type: "user", time: { created: 1 }, payload: {}, delivery: "steer" },
|
||||
})
|
||||
}
|
||||
if (url.pathname === "/api/session/ses_1/interrupt") {
|
||||
interrupted.resolve()
|
||||
return Response.json({ interrupted: true })
|
||||
}
|
||||
if (url.pathname.endsWith("/wait")) return interrupted.promise.then(() => new Response(null, { status: 204 }))
|
||||
if (request.method === "GET") return Response.json({ data: [] })
|
||||
return new Response(null, { status: 404 })
|
||||
},
|
||||
})
|
||||
return { server, requests, prompted: prompted.promise }
|
||||
}
|
||||
|
||||
function spawnRun(url: string, directory: string, message: string[], stdin: "pipe" | "ignore") {
|
||||
return Bun.spawn([process.execPath, "run", "src/index.ts", "run", "--server", url, ...message], {
|
||||
cwd: path.join(import.meta.dir, "../.."),
|
||||
env: isolatedEnv(directory, { PWD: directory }),
|
||||
signal: AbortSignal.timeout(25_000),
|
||||
stdin,
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
})
|
||||
}
|
||||
|
||||
test("run sends the message as typed", async () => {
|
||||
await using directory = await tmpdir()
|
||||
const fake = await serve(directory.path)
|
||||
const child = spawnRun(fake.server.url.toString(), directory.path, ["say hello world", `it's "quoted"`], "ignore")
|
||||
try {
|
||||
await fake.prompted
|
||||
expect(fake.requests.find((item) => item.path.endsWith("/prompt"))?.body).toMatchObject({
|
||||
text: `say hello world it's "quoted"`,
|
||||
})
|
||||
} finally {
|
||||
child.kill("SIGKILL")
|
||||
await child.exited
|
||||
await fake.server.stop(true)
|
||||
}
|
||||
}, 30_000)
|
||||
|
||||
const posix = test.skipIf(process.platform === "win32")
|
||||
|
||||
// Windows has no SIGTERM delivery: kill() terminates the process outright.
|
||||
posix(
|
||||
"run does not wait on spawn's open stdin, and SIGTERM interrupts the Session with 143",
|
||||
async () => {
|
||||
await using directory = await tmpdir()
|
||||
const fake = await serve(directory.path)
|
||||
// A stdin pipe that stays open and silent, as with spawn's default stdio.
|
||||
const child = spawnRun(fake.server.url.toString(), directory.path, ["hello"], "pipe")
|
||||
try {
|
||||
await fake.prompted
|
||||
await Bun.sleep(200)
|
||||
child.kill("SIGTERM")
|
||||
expect(await child.exited).toBe(143)
|
||||
expect(fake.requests.some((item) => item.path === "/api/session/ses_1/interrupt")).toBe(true)
|
||||
} finally {
|
||||
child.kill("SIGKILL")
|
||||
await child.exited
|
||||
await fake.server.stop(true)
|
||||
}
|
||||
},
|
||||
30_000,
|
||||
)
|
||||
|
||||
posix(
|
||||
"run reads a slow shell pipe in full alongside a message argument",
|
||||
async () => {
|
||||
await using directory = await tmpdir()
|
||||
const fake = await serve(directory.path)
|
||||
const command = `(sleep 4; echo slow context) | "${process.execPath}" run src/index.ts run --server ${fake.server.url} review`
|
||||
const child = Bun.spawn(["sh", "-c", command], {
|
||||
cwd: path.join(import.meta.dir, "../.."),
|
||||
env: isolatedEnv(directory.path, { PWD: directory.path }),
|
||||
signal: AbortSignal.timeout(25_000),
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
})
|
||||
try {
|
||||
await fake.prompted
|
||||
expect(fake.requests.find((item) => item.path.endsWith("/prompt"))?.body).toMatchObject({
|
||||
text: "review\nslow context\n",
|
||||
})
|
||||
} finally {
|
||||
child.kill("SIGKILL")
|
||||
await child.exited
|
||||
await fake.server.stop(true)
|
||||
}
|
||||
},
|
||||
30_000,
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue