Compare commits

...
4 Commits
24 changed files with 859 additions and 102 deletions
+1
View File
@@ -768,6 +768,7 @@ export function createData(config: CreateDataInput) {
match.status = event.data.shell.status
match.exit = event.data.shell.exit
match.output = event.data.output
if (event.data.shell.metadata.reason === "user") match.metadata = { ...match.metadata, reason: "user" }
match.time.completed = event.created
})
return
+31 -11
View File
@@ -723,8 +723,18 @@ test("ignores activity snapshots from an older connection", async () => {
}
})
test("projects background user shell metadata from durable shell data", () => {
test("projects user shell lifecycle metadata", () => {
const setup = activityFixture(() => Response.json({ data: {} }))
const shell = {
id: "sh_user",
status: "running" as const,
command: "pwd",
cwd: "/project",
shell: "/bin/sh",
file: "/project/shell.out",
metadata: { sessionID: "ses_refresh", background: true },
time: { started: 1 },
}
try {
setup.emit({
id: "evt_user_shell",
@@ -733,21 +743,31 @@ test("projects background user shell metadata from durable shell data", () => {
durable: { aggregateID: "ses_refresh", seq: 1, version: 1 },
data: {
sessionID: "ses_refresh",
shell: {
id: "sh_user",
status: "running",
command: "pwd",
cwd: "/project",
shell: "/bin/sh",
file: "/project/shell.out",
metadata: { sessionID: "ses_refresh", background: true },
time: { started: 1 },
},
shell,
},
})
expect(setup.data.session.message.list("ses_refresh")).toMatchObject([
{ type: "shell", shellID: "sh_user", status: "running", metadata: { background: true } },
])
setup.emit({
id: "evt_user_shell_stopped",
created: 2,
type: "session.shell.ended",
durable: { aggregateID: "ses_refresh", seq: 2, version: 1 },
data: {
sessionID: "ses_refresh",
shell: {
...shell,
status: "killed",
metadata: { ...shell.metadata, reason: "user" },
time: { started: 1, completed: 2 },
},
output: { output: "", size: 0, cursor: 0, truncated: false },
},
})
expect(setup.data.session.message.list("ses_refresh")).toMatchObject([
{ type: "shell", shellID: "sh_user", status: "killed", metadata: { background: true, reason: "user" } },
])
} finally {
setup.dispose()
}
+6 -2
View File
@@ -26,6 +26,7 @@ const Background = Schema.Struct({
}),
]),
status: Schema.Literals(["running", "completed", "error", "cancelled"]),
reason: Schema.optionalKey(Schema.Literal("user")),
output: Schema.optionalKey(Schema.String),
error: Schema.optionalKey(Schema.String),
})
@@ -42,6 +43,7 @@ export type Info = {
type: string
title?: string
status: Status
reason?: "user"
started_at: number
completed_at?: number
output?: string
@@ -129,7 +131,7 @@ export interface Interface {
readonly block: (input: BlockInput) => Effect.Effect<BlockResult | undefined>
readonly background: (id: string) => Effect.Effect<Info | undefined>
readonly backgroundAll: (input: BackgroundAllInput) => Effect.Effect<Info[]>
readonly cancel: (id: string) => Effect.Effect<Info | undefined>
readonly cancel: (id: string, options?: { reason?: "user" }) => Effect.Effect<Info | undefined>
readonly pendingBackground: Effect.Effect<readonly Background[]>
readonly completeBackground: (notificationID: SessionMessage.ID) => Effect.Effect<void>
}
@@ -179,6 +181,7 @@ export const make = Effect.gen(function* () {
notificationID: job.info.notificationID,
recovery: job.recovery,
status: job.info.status,
...(job.info.reason ? { reason: job.info.reason } : {}),
...(job.info.output !== undefined ? { output: job.info.output } : {}),
...(job.info.error !== undefined ? { error: job.info.error } : {}),
})
@@ -374,7 +377,7 @@ export const make = Effect.gen(function* () {
return result.map((item) => item.info)
})
const cancel: Interface["cancel"] = Effect.fn("Job.cancel")(function* (id) {
const cancel: Interface["cancel"] = Effect.fn("Job.cancel")(function* (id, options) {
const completed_at = yield* Clock.currentTimeMillis
const result = yield* SynchronizedRef.modifyEffect(
state.jobs,
@@ -388,6 +391,7 @@ export const make = Effect.gen(function* () {
info: {
...job.info,
status: "cancelled" as const,
...(options?.reason ? { reason: options.reason } : {}),
completed_at,
},
}
+1 -1
View File
@@ -88,7 +88,7 @@ export const layerWithCell = (cell: Cell): Layer.Layer<Service> =>
wait: (input) => require(cell, (runtime) => runtime.job.wait(input)),
block: (input) => require(cell, (runtime) => runtime.job.block(input)),
background: (id) => require(cell, (runtime) => runtime.job.background(id)),
cancel: (id) => require(cell, (runtime) => runtime.job.cancel(id)),
cancel: (id, options) => require(cell, (runtime) => runtime.job.cancel(id, options)),
completeBackground: (notificationID) =>
require(cell, (runtime) => runtime.job.completeBackground(notificationID)),
},
+1 -1
View File
@@ -122,7 +122,7 @@ export const layer = Layer.effect(
if (outcome.type === "interrupted") {
// A user cancel releases the claim: the turn must not resurrect at the next
// boot. Shutdown interruption keeps it for restart continuity.
if (outcome.reason === "user") yield* jobs.cancel(sessionID)
if (outcome.reason === "user") yield* jobs.cancel(sessionID, { reason: "user" })
yield* bus.publish(
SessionEvent.Execution.Interrupted,
{ sessionID, reason: outcome.reason },
+14 -9
View File
@@ -104,13 +104,15 @@ export const layer = (options?: Options) =>
) {
const state = background.status === "running" ? "cancelled" : background.status
const text =
background.status === "running"
? "Command cancelled because the server restarted"
: state === "completed"
? (background.output ?? "Command completed")
: state === "error"
? (background.error ?? "Command failed")
: "Command cancelled"
background.reason === "user"
? ShellResult.stopped
: background.status === "running"
? "Command cancelled because the server restarted"
: state === "completed"
? (background.output ?? "Command completed")
: state === "error"
? (background.error ?? "Command failed")
: "Command cancelled"
yield* sessions
.synthetic({
@@ -122,9 +124,10 @@ export const layer = (options?: Options) =>
shellID: recovery.shellID,
command: recovery.command,
state,
reason: background.reason,
text,
}),
...(suspended.has(recovery.sessionID) ? { resume: false } : {}),
...(background.reason === "user" || suspended.has(recovery.sessionID) ? { resume: false } : {}),
})
.pipe(
Effect.catchTag("Session.NotFoundError", () => Effect.void),
@@ -144,7 +147,9 @@ export const layer = (options?: Options) =>
return
}
const notify = Effect.fnUntraced(function* (result: Pick<Job.Background, "status" | "output" | "error">) {
const notify = Effect.fnUntraced(function* (
result: Pick<Job.Background, "status" | "output" | "error" | "reason">,
) {
yield* SubagentCompletion.deliver(sessions, jobs, {
...result,
recovery,
@@ -187,6 +187,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
draft.status = event.data.shell.status
draft.exit = event.data.shell.exit
draft.output = event.data.output
if (event.data.shell.metadata.reason === "user") draft.metadata = { ...draft.metadata, reason: "user" }
draft.time.completed = created
}),
)
+3 -1
View File
@@ -231,7 +231,9 @@ export const make = Effect.fn("Session.make")(function* () {
.pipe(Effect.catchTag("Shell.NotFoundError", () => Effect.succeed(ShellResult.unavailable)))
yield* bus.publish(SessionEvent.Shell.Ended, {
sessionID,
shell: terminal.info,
shell: terminal.reason
? { ...terminal.info, metadata: { ...terminal.info.metadata, reason: terminal.reason } }
: terminal.info,
output: preview,
})
yield* synthetic(sessionID, {
@@ -4,10 +4,12 @@ import { Effect } from "effect"
import type { Job } from "../job.js"
import type { Session } from "../session.js"
export const STOPPED_BY_USER = "Subagent stopped by user. Do not restart it unless the user asks."
export const deliver = Effect.fnUntraced(function* (
sessions: Pick<Session.Interface, "synthetic">,
jobs: Pick<Job.Interface, "completeBackground">,
input: Pick<Job.Info, "status" | "output" | "error" | "notificationID"> & {
input: Pick<Job.Info, "status" | "output" | "error" | "notificationID" | "reason"> & {
recovery: Extract<Job.Recovery, { kind: "subagent" }>
resume?: boolean
},
@@ -19,14 +21,22 @@ export const deliver = Effect.fnUntraced(function* (
? (input.output ?? "Subagent completed without a text response.")
: input.status === "error"
? (input.error ?? "Subagent failed")
: "Subagent cancelled"
: input.reason === "user"
? STOPPED_BY_USER
: "Subagent cancelled"
yield* sessions.synthetic({
...(input.notificationID ? { id: input.notificationID } : {}),
sessionID: recovery.parentSessionID,
...(input.resume === false ? { resume: false } : {}),
...(input.resume === false || input.reason === "user" ? { resume: false } : {}),
description: recovery.description,
text: `<subagent sessionID="${recovery.childSessionID}" state="${input.status}" description="${recovery.description}">\n${text}\n</subagent>`,
metadata: { source: "subagent", childID: recovery.childSessionID, agent: recovery.agent, state: input.status },
metadata: {
source: "subagent",
childID: recovery.childSessionID,
agent: recovery.agent,
state: input.status,
...(input.reason === "user" ? { reason: "user" } : {}),
},
})
if (input.notificationID) yield* jobs.completeBackground(input.notificationID)
})
+34 -13
View File
@@ -21,9 +21,12 @@ import { SessionSchema } from "./session/schema.js"
import { Config } from "./config.js"
import { ToolOutput } from "./tool-output.js"
import { ShellResult } from "./shell/result.js"
import { Job } from "./job.js"
export class NotFoundError extends Schema.TaggedError<NotFoundError>()("Shell.NotFoundError", {
id: Shell.ID,
// Explicit removal unblocks waiters; keep its intent separate from ordinary misses.
reason: Schema.optionalKey(Schema.Literal("user")),
}) {}
// Keep recent exited processes observable in memory, including their file-backed output.
@@ -68,16 +71,26 @@ export interface Interface {
// Resolves once the command reaches a terminal status, returning its final Info. Fails with
// NotFoundError if the command is unknown or is removed before it terminates.
readonly wait: (id: Shell.ID) => Effect.Effect<Shell.Info, NotFoundError>
// A known shell's terminal state and bounded tail. Missing capture remains distinct from its exit status.
// A created handle's terminal outcome survives removal; its output capture may no longer be available.
readonly result: (started: Shell.Info) => Effect.Effect<ShellResult.Result>
// Replaces the running command's timeout from now; zero clears it.
readonly timeout: (id: Shell.ID, duration: number) => Effect.Effect<Shell.Info, NotFoundError>
readonly output: (id: Shell.ID, input?: Shell.OutputInput) => Effect.Effect<Shell.Output, NotFoundError>
readonly remove: (id: Shell.ID) => Effect.Effect<void, NotFoundError>
readonly remove: (id: Shell.ID, options?: { reason?: "user" }) => Effect.Effect<void, NotFoundError>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/Shell") {}
/** User control: cancel the owning tool job before removing its process and capture. */
export const stop = Effect.fn("Shell.stop")(function* (id: Shell.ID) {
const shell = yield* Service
const jobs = yield* Job.Service
yield* shell.get(id)
yield* jobs.cancel(id, { reason: "user" })
// Cancelling a tool job also removes its shell through the interruption finalizer.
yield* shell.remove(id, { reason: "user" }).pipe(Effect.catchTag("Shell.NotFoundError", () => Effect.void))
})
export const cleanup = Effect.fn("Shell.cleanup")(function* () {
const fs = yield* FSUtil.Service
const global = yield* Global.Service
@@ -129,6 +142,7 @@ const layer = () =>
const context = yield* Effect.context()
const runFork = Effect.runForkWith(context)
const commands = new Map<Shell.ID, Active>()
const completions = new WeakMap<Info, Deferred.Deferred<Info, NotFoundError>>()
const exitOrder: Shell.ID[] = []
const outputDir = path.join(global.data, DIRECTORY, location.project.id)
@@ -154,7 +168,7 @@ const layer = () =>
return command
})
const removeCommand = Effect.fnUntraced(function* (id: Shell.ID) {
const removeCommand = Effect.fnUntraced(function* (id: Shell.ID, reason?: "user") {
const command = commands.get(id)
const index = exitOrder.indexOf(id)
if (index !== -1) exitOrder.splice(index, 1)
@@ -162,14 +176,14 @@ const layer = () =>
commands.delete(id)
if (command.timeoutFiber) yield* Fiber.interrupt(command.timeoutFiber)
// Unblock any wait still pending when the command is removed before it terminated.
yield* Deferred.fail(command.done, new NotFoundError({ id }))
yield* Deferred.fail(command.done, new NotFoundError({ id, ...(reason ? { reason } : {}) }))
yield* Effect.promise(() => unlink(command.file).catch(() => {}))
yield* bus.publish(Shell.Event.Deleted, { id })
})
const remove = Effect.fn("Shell.remove")(function* (id: Shell.ID) {
const remove: Interface["remove"] = Effect.fn("Shell.remove")(function* (id, options) {
yield* require(id)
yield* removeCommand(id)
yield* removeCommand(id, options?.reason)
})
const list = Effect.fn("Shell.list")(function* () {
@@ -224,25 +238,30 @@ const layer = () =>
})
const result = Effect.fn("Shell.result")(function* (started: Shell.Info) {
const info = yield* wait(started.id).pipe(
Effect.catchTag("Shell.NotFoundError", () =>
Effect.succeed({ ...started, status: "killed" as const, time: { ...started.time, completed: Date.now() } }),
const done = completions.get(started)
const terminal = yield* (done ? Deferred.await(done) : wait(started.id)).pipe(
Effect.map((info): Pick<ShellResult.Result, "info" | "reason"> => ({ info })),
Effect.catchTag("Shell.NotFoundError", (error) =>
Effect.succeed({
info: { ...started, status: "killed" as const, time: { ...started.time, completed: Date.now() } },
...(error.reason ? { reason: error.reason } : {}),
}),
),
)
const capture = yield* Effect.gen(function* () {
const limits = Config.latest(yield* config.entries(), "tool_output")
const maxLines = limits?.max_lines ?? ToolOutput.MAX_LINES
const maxBytes = limits?.max_bytes ?? ToolOutput.MAX_BYTES
const latest = yield* output(info.id, { cursor: Number.MAX_SAFE_INTEGER })
const page = yield* output(info.id, { cursor: Math.max(0, latest.size - maxBytes), limit: maxBytes })
const latest = yield* output(started.id, { cursor: Number.MAX_SAFE_INTEGER })
const page = yield* output(started.id, { cursor: Math.max(0, latest.size - maxBytes), limit: maxBytes })
const lines = page.output.split("\n")
if (page.output.endsWith("\n")) lines.pop()
const truncated = latest.size > maxBytes || lines.length > maxLines
const text = lines.length > maxLines ? lines.slice(-maxLines).join("\n") : page.output
const notice = truncated ? `\n\n[output truncated; full output saved to: ${info.file}]` : ""
const notice = truncated ? `\n\n[output truncated; full output saved to: ${started.file}]` : ""
return { output: `${text || "(no output)"}${notice}`, truncated }
}).pipe(Effect.catchTag("Shell.NotFoundError", () => Effect.succeed(undefined)))
return { info, capture }
return { ...terminal, capture }
})
const create = Effect.fn("Shell.create")(function* <E = never, R = never>(
@@ -413,6 +432,8 @@ const layer = () =>
)
const command = yield* Deferred.await(ready)
// The original handle retains its terminal signal even if removal precedes result().
completions.set(command.info, command.done)
return command.info
})
+11 -1
View File
@@ -5,6 +5,7 @@ import type { Shell } from "@opencode-ai/schema/shell"
export type Result = {
info: Shell.Info
capture: { output: string; truncated: boolean } | undefined
reason?: "user"
}
type Output = { output: string; truncated: boolean; exit?: number; timeout?: boolean }
@@ -17,6 +18,8 @@ export const unavailable: Shell.Output = {
truncated: false,
}
export const stopped = "Command stopped by user. Do not restart it unless the user asks."
export function output(result: Result): Output {
return {
output: result.capture?.output ?? unavailable.output,
@@ -44,6 +47,7 @@ export function notification(input: {
jobID?: string
command: string
state: "completed" | "cancelled" | "error"
reason?: "user"
text: string
output?: Output
}) {
@@ -54,6 +58,7 @@ export function notification(input: {
shellID: input.shellID,
...(input.jobID !== undefined ? { jobID: input.jobID } : {}),
state: input.state,
...(input.reason ? { reason: input.reason } : {}),
...(input.output ? metadata(input.output) : {}),
},
}
@@ -62,11 +67,16 @@ export function notification(input: {
export function userNotification(result: Result) {
const captured = output(result)
const status =
result.info.status === "killed" ? "Command cancelled." : (notice(captured) ?? "Command exited with code unknown.")
result.reason === "user"
? stopped
: result.info.status === "killed"
? "Command cancelled."
: (notice(captured) ?? "Command exited with code unknown.")
const message = notification({
shellID: result.info.id,
command: result.info.command,
state: result.info.status === "killed" ? "cancelled" : "completed",
reason: result.reason,
text: `${captured.output}\n\n${status}`,
output: captured,
})
+23 -8
View File
@@ -67,7 +67,8 @@ const StructuredOutput = Schema.Struct({
const Output = Schema.Struct({
...StructuredOutput.fields,
output: Schema.String,
status: Schema.optionalKey(Schema.Literals(["completed", "running"])),
status: Schema.optionalKey(Schema.Literals(["completed", "running", "cancelled"])),
reason: Schema.optionalKey(Schema.Literal("user")),
})
type Output = typeof Output.Type
@@ -83,6 +84,7 @@ const toolResult = (output: Output) => {
content: resultMessages(output).map((text) => ({ type: "text" as const, text })),
metadata: {
status: output.status,
...(output.reason ? { reason: output.reason } : {}),
...ShellResult.metadata(output),
...(output.shellID !== undefined ? { shellID: output.shellID } : {}),
},
@@ -170,20 +172,25 @@ export const Plugin = {
const info = (yield* runtime.job.wait({ id })).info
if (!info || info.status === "running") return
const output = info.status === "completed" ? yield* Deferred.await(settled) : undefined
const text = output
? resultMessages(output).join("\n\n")
: info.status === "error"
? (info.error ?? "Command failed")
: "Command cancelled"
const text =
info.reason === "user"
? ShellResult.stopped
: output
? resultMessages(output).join("\n\n")
: info.status === "error"
? (info.error ?? "Command failed")
: "Command cancelled"
yield* runtime.session.synthetic({
...(info.notificationID ? { id: info.notificationID } : {}),
sessionID,
...(info.reason === "user" ? { resume: false } : {}),
description: command,
...ShellResult.notification({
jobID: id,
shellID,
command,
state: info.status,
reason: info.reason,
text,
output,
}),
@@ -218,8 +225,6 @@ export const Plugin = {
finalTimeout = yield* prepare(invocation, context)
}),
)
yield* context.progress({ shellID: info.id })
const settled = yield* Deferred.make<Output>()
const run = Effect.gen(function* () {
const result = yield* shell.result(info)
@@ -251,6 +256,9 @@ export const Plugin = {
},
run,
})
yield* context
.progress({ shellID: info.id })
.pipe(Effect.onInterrupt(() => runtime.job.cancel(job.id).pipe(Effect.ignore)))
if (input.background === true) {
yield* runtime.job.background(job.id)
@@ -268,6 +276,13 @@ export const Plugin = {
}
if (result?.info.status === "error")
return yield* Effect.fail(new Error(result.info.error ?? "Command failed"))
if (result?.info.reason === "user")
return {
output: ShellResult.stopped,
status: "cancelled" as const,
reason: "user" as const,
truncated: false,
}
if (result?.info.status === "cancelled") return yield* Effect.fail(new Error("Command cancelled"))
return yield* Deferred.await(settled)
+17 -6
View File
@@ -40,7 +40,7 @@ export const Input = Schema.Struct({
export const Output = Schema.Struct({
sessionID: SessionSchema.ID,
status: Schema.Literals(["completed", "running"]),
status: Schema.Literals(["completed", "running", "cancelled"]),
output: Schema.String,
})
export const description = [
@@ -255,17 +255,28 @@ export const Plugin = {
return yield* new ToolFailure({
message: `Subagent failed (sessionID: ${child.id}): ${result.info.error ?? "unknown error"}`,
})
if (result?.info.status === "cancelled")
if (result?.info.status === "cancelled") {
if (result.info.reason === "user")
return {
sessionID: child.id,
status: "cancelled" as const,
output: SubagentCompletion.STOPPED_BY_USER,
}
return yield* new ToolFailure({ message: `Subagent cancelled (sessionID: ${child.id})` })
}
return { sessionID: child.id, status: "completed" as const, output: result?.info.output ?? NO_TEXT }
}).pipe(
Effect.map((output) => ({
output,
content:
output.status === "completed"
? `<subagent sessionID="${output.sessionID}" state="completed">\n${output.output}\n</subagent>`
: output.output,
metadata: { sessionID: output.sessionID, status: output.status },
output.status === "running"
? output.output
: `<subagent sessionID="${output.sessionID}" state="${output.status}">\n${output.output}\n</subagent>`,
metadata: {
sessionID: output.sessionID,
status: output.status,
...(output.status === "cancelled" ? { reason: "user" } : {}),
},
})),
),
}),
+3 -1
View File
@@ -209,11 +209,13 @@ describe("Job", () => {
expect(marker).toMatchObject({ id: job.id, recovery, status: "running" })
if (!marker) return yield* Effect.die("background marker missing")
yield* jobs.cancel(job.id)
yield* jobs.cancel(job.id, { reason: "user" })
expect((yield* jobs.pendingBackground).find((item) => item.id === job.id)).toMatchObject({
notificationID: marker.notificationID,
status: "cancelled",
reason: "user",
})
expect((yield* jobs.wait({ id: job.id })).info).toMatchObject({ status: "cancelled", reason: "user" })
yield* jobs.completeBackground(marker.notificationID)
}),
)
+47 -4
View File
@@ -196,7 +196,7 @@ describe("SessionExecution lifecycle", () => {
expect(yield* execution.interrupt(child)).toBeTrue()
yield* execution.awaitIdle(child)
expect((yield* jobs.wait({ id: child })).info?.status).toBe("cancelled")
expect(yield* jobs.pendingBackground).toMatchObject([{ id: child, status: "cancelled" }])
expect(yield* jobs.pendingBackground).toMatchObject([{ id: child, status: "cancelled", reason: "user" }])
expect((yield* claims(database))[child]).toBe(false)
yield* Scope.close(scope, Exit.void)
@@ -212,9 +212,14 @@ describe("SessionExecution lifecycle", () => {
)
yield* Context.get(restarted, SessionRestart.Service).resumeSuspendedSessions
yield* Context.get(restarted, SessionExecution.Service).awaitIdle(parent)
expect(drained).toEqual([parent])
expect(drained).toEqual([])
expect(yield* SessionInbox.list(database.db, parent)).toMatchObject([
{ payload: { text: expect.stringContaining("Subagent cancelled"), metadata: { state: "cancelled" } } },
{
payload: {
text: expect.stringContaining("Subagent stopped by user"),
metadata: { state: "cancelled", reason: "user" },
},
},
])
expect(yield* restartedJobs.pendingBackground).toEqual([])
}),
@@ -593,6 +598,42 @@ describe("SessionRestart background recovery", () => {
}),
)
it.effect("recovers a user-stopped shell without waking its idle session", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const jobs = yield* Job.Service
const sessionID = Session.ID.make("ses_user_stopped_shell")
yield* seedSessions(database, [sessionID])
yield* seedBackground(jobs, sessionID, [
{ id: "sh_user_stopped", shellID: "sh_user_stopped", command: "sleep 60" },
])
yield* jobs.cancel("sh_user_stopped", { reason: "user" })
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const restarted = yield* Job.make.pipe(Scope.provide(scope))
const drained: Session.ID[] = []
const context = yield* buildExecution(
scope,
({ sessionID }) => Effect.sync(() => void drained.push(sessionID)),
undefined,
restarted,
)
yield* Context.get(context, SessionRestart.Service).resumeSuspendedSessions
yield* Context.get(context, SessionExecution.Service).awaitIdle(sessionID)
expect(drained).toEqual([])
expect(yield* SessionInbox.list(database.db, sessionID)).toMatchObject([
{
payload: {
text: expect.stringContaining("Command stopped by user. Do not restart it unless the user asks."),
metadata: { source: "shell", state: "cancelled", reason: "user" },
},
},
])
expect(yield* restarted.pendingBackground).toEqual([])
}),
)
it.effect("delivers cancellation at the resumed parent's next step", () =>
Effect.gen(function* () {
const database = yield* Database.Service
@@ -1374,7 +1415,9 @@ function buildExecution(
Layer.provide(Layer.succeed(Job.Service, jobs)),
// Do not reuse the outer harness's selector with its already-captured Location map.
Layer.provide(
LayerNode.compile(Instance.byLocationNode, [[LocationServiceMap.node, locations]]).pipe(Layer.fresh),
LayerNode.compile(Instance.byLocationNode, {
replacements: [LocationServiceMap.node.replace(locations)],
}).pipe(Layer.fresh),
),
),
scope,
+53 -4
View File
@@ -3,6 +3,7 @@ import fs from "fs/promises"
import path from "path"
import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, Option, Schedule, Stream } from "effect"
import { Bus } from "@opencode-ai/core/bus"
import { Job } from "@opencode-ai/core/job"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { Location } from "@opencode-ai/core/location"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
@@ -53,10 +54,13 @@ const executionLayer = Layer.effect(
)
const it = testEffect(
AppNodeBuilder.build(LayerNode.group([Bus.node, Session.node, SessionExecution.node, LocationServiceMap.node]), [
Bus.node.replace(Bus.configured({ persist: true })),
SessionExecution.node.replace(executionLayer.pipe(Layer.provide(controlLayer))),
]).pipe(Layer.provideMerge(controlLayer)),
AppNodeBuilder.build(
LayerNode.group([Bus.node, Job.node, Session.node, SessionExecution.node, LocationServiceMap.node]),
[
Bus.node.replace(Bus.configured({ persist: true })),
SessionExecution.node.replace(executionLayer.pipe(Layer.provide(controlLayer))),
],
).pipe(Layer.provideMerge(controlLayer)),
)
const setup = Effect.gen(function* () {
@@ -272,6 +276,51 @@ describe("Session.shell", () => {
)
}
it.live("preserves user intent when stopping a user-entered shell without a tool job", () =>
Effect.gen(function* () {
const fixture = yield* setup
const command = yield* launch(fixture, "user-stop")
const jobs = yield* Job.Service
expect(yield* jobs.get(command.shellID)).toBeUndefined()
yield* Shell.stop(command.shellID).pipe(Effect.provideService(Shell.Service, fixture.shell))
yield* Fiber.join(command.caller).pipe(Effect.timeout("5 seconds"))
expect(yield* fixture.session.messages({ sessionID: fixture.created.id })).toMatchObject([
{ type: "shell", status: "killed", metadata: { background: true, reason: "user" } },
])
expect(
(yield* log(fixture.session, fixture.created.id).pipe(Stream.runCollect)).find(
(event) => event.type === "session.shell.ended",
),
).toMatchObject({ data: { shell: { metadata: { reason: "user" } } } })
expect(yield* fixture.session.inbox(fixture.created.id)).toMatchObject([
{
type: "synthetic",
payload: {
text: expect.stringContaining("Command stopped by user. Do not restart it unless the user asks."),
metadata: { source: "shell", shellID: command.shellID, state: "cancelled", reason: "user" },
},
},
])
expect(fixture.control.wakes).toEqual([])
}),
)
it.live("retains the stop result when the caller has not started waiting", () =>
Effect.gen(function* () {
const fixture = yield* setup
const started = yield* fixture.shell.create({
command: process.platform === "win32" ? "Start-Sleep -Seconds 60" : "sleep 60",
timeout: 0,
})
yield* Shell.stop(started.id).pipe(Effect.provideService(Shell.Service, fixture.shell))
expect(yield* fixture.shell.result(started)).toMatchObject({
info: { id: started.id, status: "killed" },
reason: "user",
capture: undefined,
})
}),
)
for (const outcome of [
{
status: "killed",
+105
View File
@@ -164,6 +164,13 @@ const productionIt = testEffect(AppNodeBuilder.build(nodes, replacements))
const it = testEffect(
AppNodeBuilder.build(nodes, [...replacements, PluginSupervisor.node.replace(shellPluginSupervisor)]),
)
const stopIt = testEffect(
AppNodeBuilder.build(nodes, [
Permission.node.replace(permission),
Global.node.replace(tempGlobalLayer),
PluginSupervisor.node.replace(shellPluginSupervisor),
]),
)
const permissionIt = testEffect(
AppNodeBuilder.build(LayerNode.group([nodes, PermissionSaved.node]), [
SessionExecution.node.replace(executionNode),
@@ -1339,6 +1346,104 @@ describe("ShellTool", () => {
{ timeout: 15_000 },
)
it.live("returns an intentional stop result for a foreground command", () =>
Effect.gen(function* () {
const tmp = yield* tmpdirScoped()
reset()
yield* withSession(tmp.path, (registry) =>
Effect.gen(function* () {
const ready = yield* Deferred.make<string>()
const running = yield* executeTool(registry, {
...call({ command: idleCommand }),
progress: (update) =>
typeof update.shellID === "string"
? Deferred.succeed(ready, update.shellID).pipe(Effect.asVoid)
: Effect.void,
}).pipe(Effect.forkScoped)
const id = ShellSchema.ID.make(yield* Deferred.await(ready))
yield* Shell.stop(id)
const result = yield* Fiber.join(running)
expect(result.metadata).toMatchObject({ status: "cancelled", reason: "user" })
expect(result.content).toEqual([
Expected.text("Command stopped by user. Do not restart it unless the user asks."),
])
const jobs = yield* Job.Service
expect((yield* jobs.get(id))?.status).toBe("cancelled")
}),
)
}),
)
it.live("cancels the shell job when interrupted during initial progress", () =>
Effect.gen(function* () {
const tmp = yield* tmpdirScoped()
reset()
yield* withSession(tmp.path, (registry) =>
Effect.gen(function* () {
const ready = yield* Deferred.make<string>()
const running = yield* executeTool(registry, {
...call({ command: idleCommand, background: true }),
progress: (update) =>
typeof update.shellID === "string"
? Deferred.succeed(ready, update.shellID).pipe(Effect.andThen(Effect.never))
: Effect.void,
}).pipe(Effect.forkScoped)
const id = yield* Deferred.await(ready)
yield* Fiber.interrupt(running)
const jobs = yield* Job.Service
const shell = yield* Shell.Service
expect(yield* jobs.get(id)).toMatchObject({ status: "cancelled" })
expect(yield* shell.list()).toEqual([])
expect(yield* jobs.pendingBackground).toEqual([])
}),
)
}),
)
stopIt.live("records a background user stop without waking the idle session", () =>
Effect.gen(function* () {
const tmp = yield* tmpdirScoped()
reset()
yield* withSession(tmp.path, (registry) =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const jobs = yield* Job.Service
const sessions = yield* Session.Service
const execution = yield* SessionExecution.Service
const started: Session.ID[] = []
yield* bus.project(SessionEvent.Execution.Started, (event) =>
Effect.sync(() => void started.push(event.data.sessionID)),
)
const admitted = yield* Deferred.make<Job.Background>()
yield* bus.project(SessionEvent.InboxEnqueued, (event) =>
Effect.gen(function* () {
if (event.data.sessionID !== sessionID || event.data.item.type !== "synthetic") return
const marker = (yield* jobs.pendingBackground).find((job) => job.notificationID === event.data.inboxID)
if (marker) yield* Deferred.succeed(admitted, marker)
}),
)
const result = yield* executeTool(registry, call({ command: idleCommand, background: true }))
const id = result.metadata?.shellID
if (typeof id !== "string") return yield* Effect.die("Expected shell ID")
yield* Shell.stop(ShellSchema.ID.make(id))
expect(yield* Deferred.await(admitted)).toMatchObject({ id, status: "cancelled", reason: "user" })
yield* jobs.pendingBackground.pipe(Effect.repeat({ until: (pending) => pending.length === 0 }))
yield* execution.awaitIdle(sessionID)
expect(started).toEqual([])
expect(yield* sessions.inbox(sessionID)).toMatchObject([
{
type: "synthetic",
payload: {
text: expect.stringContaining("Command stopped by user. Do not restart it unless the user asks."),
metadata: { source: "shell", state: "cancelled", reason: "user", shellID: id },
},
},
])
}),
)
}),
)
it.live("returns the shell id for a background command", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
+206 -4
View File
@@ -34,7 +34,7 @@ import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
import { Permission } from "@opencode-ai/core/permission"
import { SubagentTool } from "@opencode-ai/core/tool/plugin/subagent"
import { Tool } from "@opencode-ai/core/tool"
import { tmpdir } from "./fixture/tmpdir"
import { tmpdir, tmpdirScoped } from "./fixture/tmpdir"
import { tempGlobalLayer } from "./fixture/global"
import { testEffect } from "./lib/effect"
import { executeTool, registerToolPlugin, toolIdentity } from "./lib/tool"
@@ -129,11 +129,12 @@ const productionIt = testEffect(AppNodeBuilder.build(nodes, replacements))
const it = testEffect(
AppNodeBuilder.build(nodes, [...replacements, PluginSupervisor.node.replace(subagentPluginSupervisor)]),
)
const completionLLM = TestLLM.testLayer({ fallback: TestLLM.text(childText, "completion") })
const completionIt = testEffect(
AppNodeBuilder.build(LayerNode.group([nodes, SessionRestart.node, KV.node]), [
AppNodeBuilder.build(LayerNode.group([nodes, SessionRestart.node, SessionStore.node, KV.node]), [
Global.node.replace(tempGlobalLayer),
PluginSupervisor.node.replace(subagentPluginSupervisor),
LayerNodePlatform.llmClient.replace(TestLLM.testLayer({ fallback: TestLLM.text(childText, "completion") })),
LayerNodePlatform.llmClient.replace(completionLLM),
SessionRunnerModel.node.replace(
Layer.succeed(SessionRunnerModel.Service, {
resolve: () =>
@@ -149,7 +150,7 @@ const completionIt = testEffect(
),
}),
),
]),
]).pipe(Layer.provideMerge(completionLLM)),
)
const withSubagent = (location: Location.Ref) =>
@@ -178,6 +179,207 @@ const withSubagent = (location: Location.Ref) =>
})
describe("SubagentTool", () => {
completionIt.live("returns a successful cancelled result when the user stops a foreground child", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped()
const sessions = yield* Session.Service
const parent = yield* sessions.create({
location: Location.Ref.make({ directory: AbsolutePath.make(dir.path) }),
model: parentModel,
title: "Foreground parent",
})
yield* withSubagent(parent.location)
const locations = yield* LocationServiceMap.Service
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
const llm = yield* TestLLM.Test
yield* llm.push(TestLLM.hangAfter())
const running = yield* Deferred.make<Session.ID>()
const call = yield* executeTool(registry, {
sessionID: parent.id,
...toolIdentity,
progress: (update) => Deferred.succeed(running, outputSessionID(update)).pipe(Effect.asVoid),
call: {
type: "tool-call",
id: "call-user-stopped-subagent",
name: SubagentTool.name,
input: { agent: "reviewer", description: "foreground review", prompt: "review" },
},
}).pipe(Effect.forkScoped)
const childID = yield* Deferred.await(running)
yield* llm.wait(1)
const jobs = yield* Job.Service
yield* jobs.get(childID).pipe(Effect.repeat({ until: (info) => info?.status === "running" }))
expect(yield* sessions.interrupt(childID)).toBeTrue()
yield* sessions.wait(childID)
expect(yield* Fiber.join(call)).toEqual({
status: "completed",
output: {
sessionID: childID,
status: "cancelled",
output: "Subagent stopped by user. Do not restart it unless the user asks.",
},
content: [
{
type: "text",
text: `<subagent sessionID="${childID}" state="cancelled">\nSubagent stopped by user. Do not restart it unless the user asks.\n</subagent>`,
},
],
metadata: { sessionID: childID, status: "cancelled", reason: "user" },
})
expect(yield* jobs.get(childID)).toMatchObject({ status: "cancelled", reason: "user" })
expect(yield* llm.requests()).toHaveLength(1)
const resumed = yield* executeTool(registry, {
sessionID: parent.id,
...toolIdentity,
call: {
type: "tool-call",
id: "call-explicitly-resumed-subagent",
name: SubagentTool.name,
input: { agent: "reviewer", description: "continued review", prompt: "continue", sessionID: childID },
},
})
expect(resumed).toMatchObject({
status: "completed",
content: [{ type: "text", text: completedOutput(childID) }],
metadata: { sessionID: childID, status: "completed" },
})
expect((yield* jobs.get(childID))?.reason).toBeUndefined()
expect(yield* llm.requests()).toHaveLength(2)
}),
)
completionIt.live("keeps a non-user foreground cancellation as a tool error", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped()
const sessions = yield* Session.Service
const parent = yield* sessions.create({
location: Location.Ref.make({ directory: AbsolutePath.make(dir.path) }),
model: parentModel,
title: "Cancelled foreground parent",
})
yield* withSubagent(parent.location)
const locations = yield* LocationServiceMap.Service
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
const llm = yield* TestLLM.Test
yield* llm.push(TestLLM.hangAfter())
const running = yield* Deferred.make<Session.ID>()
const call = yield* executeTool(registry, {
sessionID: parent.id,
...toolIdentity,
progress: (update) => Deferred.succeed(running, outputSessionID(update)).pipe(Effect.asVoid),
call: {
type: "tool-call",
id: "call-cancelled-subagent",
name: SubagentTool.name,
input: { agent: "reviewer", description: "cancelled review", prompt: "review" },
},
}).pipe(Effect.forkScoped)
const childID = yield* Deferred.await(running)
yield* llm.wait(1)
const jobs = yield* Job.Service
yield* jobs.get(childID).pipe(Effect.repeat({ until: (info) => info?.status === "running" }))
yield* jobs.cancel(childID)
expect(yield* Fiber.join(call)).toEqual({
status: "error",
error: { type: "tool.execution", message: `Subagent cancelled (sessionID: ${childID})` },
})
expect((yield* jobs.get(childID))?.reason).toBeUndefined()
yield* sessions.interrupt(childID)
yield* sessions.wait(childID)
}),
)
completionIt.live("admits user-stopped background work without waking an idle parent, including restart replay", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped()
const sessions = yield* Session.Service
const parent = yield* sessions.create({
location: Location.Ref.make({ directory: AbsolutePath.make(dir.path) }),
model: parentModel,
title: "Idle notification recipient",
})
yield* withSubagent(parent.location)
const locations = yield* LocationServiceMap.Service
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
const llm = yield* TestLLM.Test
yield* llm.push(TestLLM.hangAfter())
const jobs = yield* Job.Service
const bus = yield* Bus.Service
const admitted = yield* Deferred.make<Job.Background>()
const notifications: SessionMessage.ID[] = []
yield* bus.project(SessionEvent.InboxEnqueued, (event) =>
Effect.gen(function* () {
if (event.data.sessionID !== parent.id || event.data.item.type !== "synthetic") return
notifications.push(event.data.inboxID)
const marker = (yield* jobs.pendingBackground).find((job) => job.notificationID === event.data.inboxID)
expect(marker).toMatchObject({ status: "cancelled", reason: "user" })
if (marker) yield* Deferred.succeed(admitted, marker)
}),
)
const result = yield* executeTool(registry, {
sessionID: parent.id,
...toolIdentity,
call: {
type: "tool-call",
id: "call-user-stopped-background",
name: SubagentTool.name,
input: { agent: "reviewer", description: "background review", prompt: "review", background: true },
},
})
const childID = outputSessionID(result.metadata)
yield* llm.wait(1)
expect(yield* sessions.interrupt(childID)).toBeTrue()
yield* sessions.wait(childID)
const marker = yield* Deferred.await(admitted)
yield* jobs.pendingBackground.pipe(Effect.repeat({ until: (pending) => pending.length === 0 }))
yield* sessions.wait(parent.id)
const inbox = yield* sessions.inbox(parent.id)
expect(inbox).toEqual([
expect.objectContaining({
id: marker.notificationID,
type: "synthetic",
payload: {
description: "background review",
text: `<subagent sessionID="${childID}" state="cancelled" description="background review">\nSubagent stopped by user. Do not restart it unless the user asks.\n</subagent>`,
metadata: { source: "subagent", childID, agent: "reviewer", state: "cancelled", reason: "user" },
},
}),
])
expect(yield* llm.requests()).toHaveLength(1)
const execution = yield* SessionExecution.Service
expect(yield* execution.isActive(parent.id)).toBeFalse()
expect(yield* execution.isActive(childID)).toBeFalse()
const store = yield* SessionStore.Service
expect(yield* store.listSuspended()).toEqual([])
// Replay the persisted terminal marker after a crash between admission and acknowledgment.
const kv = yield* KV.Service
yield* kv.set(`job.background/${marker.notificationID}`, marker)
const restart = yield* SessionRestart.Service
yield* restart.resumeSuspendedSessions
yield* sessions.wait(parent.id)
expect(yield* llm.requests()).toHaveLength(1)
expect(yield* sessions.inbox(parent.id)).toEqual(inbox)
expect(notifications).toEqual([marker.notificationID])
expect(yield* jobs.pendingBackground).toEqual([])
yield* sessions.prompt({ sessionID: parent.id, text: "Continue with other work" })
yield* sessions.wait(parent.id)
expect(yield* sessions.inbox(parent.id)).toEqual([])
expect((yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic")).toEqual([
expect.objectContaining({
id: marker.notificationID,
metadata: { source: "subagent", childID, agent: "reviewer", state: "cancelled", reason: "user" },
}),
])
expect(yield* llm.requests()).toHaveLength(2)
}),
)
completionIt.live("admits one durable completion across live delivery and restart replay", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
+9 -10
View File
@@ -1,4 +1,5 @@
import { Shell } from "@opencode-ai/core/shell"
import { Job } from "@opencode-ai/core/job"
import { Location } from "@opencode-ai/core/location"
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor-service"
import { Effect } from "effect"
@@ -9,6 +10,7 @@ import { response } from "../location"
export const ShellHandler = HttpApiBuilder.group(Api, "server.shell", (handlers) =>
Effect.gen(function* () {
const jobs = yield* Job.Service
return handlers
.handle(
"shell.list",
@@ -83,16 +85,13 @@ export const ShellHandler = HttpApiBuilder.group(Api, "server.shell", (handlers)
.handle(
"shell.remove",
Effect.fn(function* (ctx) {
const shell = yield* Shell.Service
yield* shell
.remove(ctx.params.id)
.pipe(
Effect.catchTag(
"Shell.NotFoundError",
() =>
new ShellNotFoundError({ id: ctx.params.id, message: `Shell command not found: ${ctx.params.id}` }),
),
)
yield* Shell.stop(ctx.params.id).pipe(
Effect.provideService(Job.Service, jobs),
Effect.catchTag(
"Shell.NotFoundError",
() => new ShellNotFoundError({ id: ctx.params.id, message: `Shell command not found: ${ctx.params.id}` }),
),
)
return HttpApiSchema.NoContent.make()
}),
)
+14 -13
View File
@@ -52,18 +52,19 @@ it.live(
const cell = PluginRuntime.makeCell()
// Host and private instances must reuse the same global layer identities.
const replacements: LayerNode.Replacements = [
[Global.node, tempGlobalLayer],
[Database.node, Database.node],
[Bus.node, Bus.node],
[App.node, App.node],
[ModelsDev.node, ModelsDev.configured({ fetch: false })],
[Watcher.node, Watcher.configured({ enabled: false })],
[PluginRuntime.node, PluginRuntime.layerWithCell(cell)],
[PluginRuntime.providerNode, PluginRuntime.providerNodeWithCell(cell)],
[llmClient, Layer.succeed(LLMClient.Service, llm)],
[SessionRunnerModel.node, Layer.succeed(SessionRunnerModel.Service, { resolve: () => Effect.succeed(model) })],
[
Instance.byLocationNode,
Global.node.replace(tempGlobalLayer),
Database.node.replace(Database.node),
Bus.node.replace(Bus.node),
App.node.replace(App.node),
ModelsDev.node.replace(ModelsDev.configured({ fetch: false })),
Watcher.node.replace(Watcher.configured({ enabled: false })),
PluginRuntime.node.replace(PluginRuntime.layerWithCell(cell)),
PluginRuntime.providerNode.replace(PluginRuntime.providerNodeWithCell(cell)),
llmClient.replace(Layer.succeed(LLMClient.Service, llm)),
SessionRunnerModel.node.replace(
Layer.succeed(SessionRunnerModel.Service, { resolve: () => Effect.succeed(model) }),
),
Instance.byLocationNode.replace(
Layer.effect(
Instance.Service,
Effect.gen(function* () {
@@ -151,7 +152,7 @@ it.live(
})
}),
),
],
),
]
const context = yield* Layer.build(
createEmbeddedRoutes({}, replacements).pipe(Layer.provide(HttpServer.layerServices)),
+13 -1
View File
@@ -172,7 +172,19 @@ export const { use: useSessionTabs, provider: SessionTabsProvider } = createSimp
: members.some((id) => (data.session.form.list(id)?.length ?? 0) > 0)
? ("question" as const)
: (false as const),
busy: members.some((id) => data.session.status(id) === "running" || data.session.pending.list(id).length > 0),
// Quiet user-stop notices are context, not pending execution.
busy: members.some(
(id) =>
data.session.status(id) === "running" ||
data.session.pending
.list(id)
.some(
(item) =>
item.type !== "synthetic" ||
item.payload.metadata?.state !== "cancelled" ||
item.payload.metadata?.reason !== "user",
),
),
renaming: data.session.title.pending(session),
}
}
+26 -8
View File
@@ -2119,6 +2119,7 @@ function SessionNoticeMessageV2(props: { message: SessionMessageInfo }) {
})
const completion = () => source() === "subagent" || source() === "shell"
const state = () => stringValue(metadata()?.state)
const stopped = () => state() === "cancelled" && metadata()?.reason === "user"
const actor = () => (source() === "shell" ? "Shell" : Locale.titlecase(stringValue(metadata()?.agent) ?? "Subagent"))
const text = () => {
if (props.message.type === "system") return props.message.description ?? "Instructions updated"
@@ -2127,14 +2128,16 @@ function SessionNoticeMessageV2(props: { message: SessionMessageInfo }) {
}
const description = () => (source() === "shell" ? text().replace(/\s+/g, " ").trim() : text())
const status = () => {
if (stopped()) return "stopped by user"
if (state() === "completed") return "finished"
if (state() === "error") return "failed"
return state() ?? "finished"
}
const heading = () => `${state() === "completed" ? "↳" : "!"} ${actor()} ${status()}`
const heading = () => `${state() === "completed" || stopped() ? "↳" : "!"} ${actor()} ${status()}`
const suffix = () => Locale.truncateWidth(` · ${description()}`, Math.max(0, ctx.width - 3 - stringWidth(heading())))
const color = () => {
if (hover()) return theme.text.action.secondary.hovered
if (stopped()) return theme.text.subdued
if (state() === "error") return theme.text.feedback.error.default
if (state() === "cancelled") return theme.text.feedback.warning.default
return theme.text.feedback.info.default
@@ -2334,7 +2337,9 @@ function RevertMessage(props: {
}
function ShellMessage(props: { message: Extract<SessionMessageInfo, { type: "shell" }> }) {
const stopped = () => props.message.status === "killed" && props.message.metadata?.reason === "user"
const error = createMemo(() => {
if (stopped()) return
if (props.message.status === "killed") return "Command cancelled"
if (props.message.status === "timeout") return "Command timed out"
if (props.message.exit !== undefined && props.message.exit !== 0)
@@ -2346,6 +2351,7 @@ function ShellMessage(props: { message: Extract<SessionMessageInfo, { type: "she
shellID={props.message.shellID}
command={props.message.command}
status={props.message.status === "running" ? "running" : "completed"}
stopped={stopped()}
output={props.message.output?.output}
error={error()}
/>
@@ -3181,6 +3187,8 @@ function BlockToolContent(props: BlockToolProps & { borderColor: RGBA }) {
const SHELL_DISPLAY_LIMIT = 1024 * 1024
function Shell(props: ToolProps) {
const stopped = () =>
props.part.state.status === "completed" && props.metadata.status === "cancelled" && props.metadata.reason === "user"
return (
<ShellDisplay
part={props.part}
@@ -3188,8 +3196,9 @@ function Shell(props: ToolProps) {
command={stringValue(props.input.command)}
workdir={stringValue(props.input.workdir)}
status={props.part.state.status}
background={Boolean(stringValue(props.metadata.shellID)) && props.part.state.status !== "running"}
output={stringValue(props.metadata.shellID) ? undefined : props.output}
stopped={stopped()}
background={!stopped() && Boolean(stringValue(props.metadata.shellID)) && props.part.state.status !== "running"}
output={!stopped() && stringValue(props.metadata.shellID) ? undefined : props.output}
/>
)
}
@@ -3200,6 +3209,7 @@ function ShellDisplay(props: {
command?: string
workdir?: string
status: SessionMessageAssistantTool["state"]["status"]
stopped?: boolean
background?: boolean
output?: string
error?: string
@@ -3217,7 +3227,7 @@ function ShellDisplay(props: {
const id = props.shellID
return Boolean(id && data.shell.get(id))
})
const isRunning = createMemo(() => props.status === "running" || backgroundRunning())
const isRunning = createMemo(() => !props.stopped && (props.status === "running" || backgroundRunning()))
const workdir = createMemo(() => pathFormatter.format(props.workdir))
const [expanded, setExpanded] = createSignal(false)
const [backgroundOutput, setBackgroundOutput] = createSignal("")
@@ -3332,6 +3342,9 @@ function ShellDisplay(props: {
<Show when={props.background}>
<StatusBadge>Background</StatusBadge>
</Show>
<Show when={props.stopped}>
<text fg={theme.text.subdued}>stopped by user</text>
</Show>
</box>
</BlockTool>
)
@@ -3452,19 +3465,22 @@ function WebSearch(props: ToolProps) {
function Subagent(props: ToolProps) {
const { navigate } = useRoute()
const data = useData()
const theme = useTheme()
const stopped = () =>
props.part.state.status === "completed" && props.metadata.status === "cancelled" && props.metadata.reason === "user"
const sessionID = createMemo(() => stringValue(props.metadata.sessionID) ?? stringValue(props.metadata.sessionId))
const description = createMemo(() => stringValue(props.input.description))
const continuation = createMemo(() => Boolean(stringValue(props.input.sessionID)))
const isRunning = createMemo(() => {
const id = sessionID()
return props.part.state.status === "running" || Boolean(id && data.session.status(id) === "running")
return !stopped() && (props.part.state.status === "running" || Boolean(id && data.session.status(id) === "running"))
})
return (
<InlineTool
icon={continuation() ? "↳" : isRunning() ? "│" : props.part.state.status === "completed" ? "✓" : "│"}
icon={continuation() || stopped() ? "↳" : isRunning() ? "│" : props.part.state.status === "completed" ? "✓" : "│"}
spinner={!continuation() && isRunning()}
complete={description()}
complete={stopped() || description()}
pending="Delegating…"
part={props.part}
onClick={() => {
@@ -3472,7 +3488,9 @@ function Subagent(props: ToolProps) {
if (id) navigate({ type: "session", sessionID: id })
}}
status={
isBackgroundSubagent(props.metadata, props.part.state.status) ? (
stopped() ? (
<text fg={theme.text.subdued}>stopped by user</text>
) : isBackgroundSubagent(props.metadata, props.part.state.status) ? (
<StatusBadge>Background</StatusBadge>
) : undefined
}
@@ -895,6 +895,47 @@ test("closing a tab is not undone by another TUI viewing the same session", asyn
}
})
test.each(["shell", "subagent"])("a quiet %s stop notice does not keep its tab busy", async (source) => {
const setup = await renderSessionTabs("parent")
try {
await wait(() => setup.tabs.current() === "parent")
setup.emit({
id: "evt_stopped",
created: 1,
type: "session.inbox.enqueued",
durable: { aggregateID: "parent", seq: 1, version: 1 },
data: {
sessionID: "parent",
inboxID: "msg_stopped",
item: {
type: "synthetic",
delivery: "steer",
payload: { text: "Stopped by user", metadata: { source, state: "cancelled", reason: "user" } },
},
},
})
await wait(() => setup.data.session.pending.list("parent").length === 1)
expect(setup.tabs.status("parent").busy).toBe(false)
for (const [index, type] of (["session.execution.started", "session.execution.succeeded"] as const).entries()) {
setup.emit({
id: `evt_execution_${index}`,
created: 2 + index,
type,
durable: { aggregateID: "parent", seq: 2 + index, version: 1 },
data: { sessionID: "parent" },
})
await wait(() => setup.tabs.status("parent").busy === (index === 0))
}
setup.emit(admitted("parent", "msg_4"))
await wait(() => setup.tabs.status("parent").busy)
} finally {
await setup.destroy()
}
})
test("user prompt admissions pulse an already-busy background tab", async () => {
const setup = await renderSessionTabs("background")
@@ -0,0 +1,185 @@
import { expect, test } from "bun:test"
import { createTestRenderer } from "@opentui/core/testing"
import type { SessionInboxInfo, SessionMessageInfo } from "@opencode-ai/client"
import { Effect, FileSystem } from "effect"
import { Global } from "@opencode-ai/util/global"
import { createEventStream, createFetch, directory, json } from "./fixture/tui-client"
import { tmpdir } from "./fixture/fixture"
test.each([
{ mode: "dark" as const, width: 100 },
{ mode: "light" as const, width: 50 },
])("renders user stops neutrally while the parent is idle ($mode, $width columns)", async ({ mode, width }) => {
await using state = await tmpdir()
const setup = await createTestRenderer({ width, height: 48, useThread: false })
setup.renderer.start()
const session = {
id: "session-user-stop",
title: "User stops",
projectID: "project",
location: { directory },
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: 0, updated: 0 },
}
const messages: SessionMessageInfo[] = [
{ id: "message-user", type: "user", text: "Work", time: { created: 0 } },
{
id: "message-user-shell",
type: "shell",
shellID: "shell-user",
command: "sleep 60",
status: "killed",
metadata: { background: true, reason: "user" },
time: { created: 0, completed: 1 },
},
{
id: "message-tools",
type: "assistant",
agent: "build",
model: { providerID: "demo", id: "demo-model" },
content: [
{
type: "tool",
id: "call-shell",
name: "shell",
state: {
status: "completed",
input: { command: "sleep 60" },
content: [{ type: "text", text: "Partial shell output" }],
metadata: { status: "cancelled", reason: "user" },
},
time: { created: 1, completed: 2 },
},
{
type: "tool",
id: "call-subagent",
name: "subagent",
state: {
status: "completed",
input: { agent: "explore", description: "Inspect files" },
content: [{ type: "text", text: "Subagent stopped by user." }],
metadata: { status: "cancelled", reason: "user", sessionID: "child-foreground" },
},
time: { created: 1, completed: 2 },
},
],
finish: "stop",
time: { created: 1, completed: 2 },
},
{
id: "message-instructions",
type: "system",
text: "Instructions",
description: "Instructions updated",
time: { created: 3 },
},
...["cancelled", "error"].map(
(status): SessionMessageInfo => ({
id: `message-${status}`,
type: "synthetic",
text: status,
description: "Other command",
metadata: { source: "shell", state: status },
time: { created: 4 },
}),
),
]
const pending: SessionInboxInfo[] = [
{
id: "message-shell-stop",
sessionID: session.id,
type: "synthetic",
delivery: "steer",
payload: {
text: "Shell stopped by user. Do not restart it unless the user asks.",
description: "sleep 60",
metadata: { source: "shell", state: "cancelled", reason: "user", shellID: "shell-background" },
},
timeCreated: 5,
},
]
const events = createEventStream()
const calls = createFetch((url) => {
if (url.pathname === "/api/session") return json({ data: [session], cursor: {} })
if (url.pathname === `/api/session/${session.id}`) return json({ data: session })
if (url.pathname === `/api/session/${session.id}/message`) return json({ data: messages.toReversed(), cursor: {} })
if (url.pathname === `/api/session/${session.id}/inbox`) return json({ data: pending })
if (url.pathname === `/api/session/${session.id}/permission`) return json({ data: [] })
return undefined
}, events)
const server = Bun.serve({ port: 0, idleTimeout: 0, fetch: (request) => calls.fetch(request) })
const { run } = await import("../src/app")
const task = Effect.runPromise(
run({
app: { name: "test", version: "test", channel: "test" },
server: { endpoint: { url: server.url.toString() } },
config: {
get: async () => ({ animations: false, tabs: { enabled: false }, theme: { name: "opencode", mode } }),
update: async () => ({}),
},
packages: { resolve: async () => undefined },
terminalHandoff: async () => ({ renderer: setup.renderer, mode, complete: () => {} }),
args: { sessionID: session.id },
log: () => {},
}).pipe(Effect.provide(Global.layerWith({ state: state.path })), Effect.provide(FileSystem.layerNoop({}))),
)
try {
// Inbox hydration must show the stop even though it is not in projected history.
await setup.waitForFrame((frame) => frame.includes("Shell stopped by user"))
expect(setup.captureCharFrame().match(/stopped by user/g)).toHaveLength(4)
expect(setup.captureCharFrame()).not.toContain("Command cancelled")
expect(setup.captureCharFrame()).toContain("\u21b3 Explore Subagent")
expect(setup.captureCharFrame()).not.toContain("\u2713 Explore Subagent")
// Admission alone is sufficient: the idle parent never receives a running event.
events.emit({
id: "evt_subagent_stopped",
type: "session.inbox.enqueued",
created: 6,
durable: { aggregateID: session.id, seq: 0, version: 1 },
data: {
sessionID: session.id,
inboxID: "message-subagent-stop",
item: {
type: "synthetic",
delivery: "steer",
payload: {
text: "Subagent stopped by user. Do not restart it unless the user asks.",
description: "Inspect source",
metadata: { source: "subagent", state: "cancelled", reason: "user", agent: "explore", childID: "child" },
},
},
},
})
await setup.waitForFrame((frame) => frame.includes("Explore stopped by user"))
expect(setup.captureCharFrame().match(/stopped by user/g)).toHaveLength(5)
expect(setup.captureCharFrame()).toContain("\u21b3 Shell stopped by user")
expect(setup.captureCharFrame()).toContain("\u21b3 Explore stopped by user")
expect(setup.captureCharFrame()).toContain("! Shell cancelled")
expect(setup.captureCharFrame()).toContain("! Shell failed")
const spans = setup.captureSpans().lines.flatMap((line) => line.spans)
const subdued = spans.find((span) => span.text.includes("Instructions updated"))!.fg.toInts()
const stopped = spans.filter((span) => span.text.includes("stopped by user"))
expect(stopped).toHaveLength(5)
stopped.forEach((span) => expect(span.fg.toInts()).toEqual(subdued))
for (const label of ["! Shell cancelled", "! Shell failed"])
expect(spans.find((span) => span.text.includes(label))!.fg.toInts()).not.toEqual(subdued)
events.emit({
id: "evt_subagent_delivered",
type: "session.inbox.delivered",
created: 7,
durable: { aggregateID: session.id, seq: 1, version: 1 },
data: { sessionID: session.id, inboxID: "message-subagent-stop" },
})
await setup.waitForFrame(
(frame) => frame.indexOf("Explore stopped by user") < frame.indexOf("Shell stopped by user"),
)
expect(setup.captureCharFrame().match(/Explore stopped by user/g)).toHaveLength(1)
} finally {
setup.renderer.destroy()
await task.finally(() => server.stop(true))
}
})