diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 6645a5a86ccf..36e303824943 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -1905,6 +1905,13 @@ export type ShellOutputOutput = { } export type ShellOutputOperation = (input: ShellOutputInput) => Effect.Effect +export type ShellStopInput = { + readonly id: Shell.ID + readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined +} +export type ShellStopOutput = { readonly location: Location.Info; readonly data: Shell.Info } +export type ShellStopOperation = (input: ShellStopInput) => Effect.Effect + export type ShellRemoveInput = { readonly id: Shell.ID readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined @@ -1918,6 +1925,7 @@ export interface ShellApi { readonly get: ShellGetOperation readonly timeout: ShellTimeoutOperation readonly output: ShellOutputOperation + readonly stop: ShellStopOperation readonly remove: ShellRemoveOperation } diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index dafd6cda741c..74a71add5dbe 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -234,6 +234,8 @@ import type { ShellTimeoutOutput, ShellOutputInput, ShellOutputOutput, + ShellStopInput, + ShellStopOutput, ShellRemoveInput, ShellRemoveOutput, ReferenceListInput, @@ -1428,6 +1430,13 @@ const EndpointShellOutput = (raw: RawClient["server.shell"]) => (input: ShellOut }).pipe(Effect.mapError(mapClientError)), ) +const EndpointShellStop = (raw: RawClient["server.shell"]) => (input: ShellStopInput) => + preserveEffect()( + raw["shell.stop"]({ params: { id: input["id"] }, query: { location: input["location"] } }).pipe( + Effect.mapError(mapClientError), + ), + ) + const EndpointShellRemove = (raw: RawClient["server.shell"]) => (input: ShellRemoveInput) => preserveEffect()( raw["shell.remove"]({ params: { id: input["id"] }, query: { location: input["location"] } }).pipe( @@ -1441,6 +1450,7 @@ const adaptGroupShell = (raw: RawClient["server.shell"]) => ({ get: EndpointShellGet(raw), timeout: EndpointShellTimeout(raw), output: EndpointShellOutput(raw), + stop: EndpointShellStop(raw), remove: EndpointShellRemove(raw), }) diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 33f55761c798..008027ecf5e0 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -230,6 +230,8 @@ import type { ShellTimeoutOutput, ShellOutputInput, ShellOutputOutput, + ShellStopInput, + ShellStopOutput, ShellRemoveInput, ShellRemoveOutput, ReferenceListInput, @@ -1940,6 +1942,18 @@ export function make(options: ClientOptions) { }, requestOptions, ), + stop: (input: ShellStopInput, requestOptions?: RequestOptions) => + request( + { + method: "POST", + path: `/api/shell/${encodeURIComponent(input.id)}/stop`, + query: { location: input["location"] }, + successStatus: 200, + declaredStatuses: [400, 401, 404], + empty: false, + }, + requestOptions, + ), remove: (input: ShellRemoveInput, requestOptions?: RequestOptions) => request( { diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 3638435548af..18ec00bd684d 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -106,7 +106,7 @@ export type SessionMessageShell = { type: "shell" shellID: string command: string - status: "running" | "exited" | "timeout" | "killed" + status: "running" | "exited" | "timeout" | "killed" | "unavailable" exit?: number | "Infinity" | "-Infinity" | "NaN" output?: { output: string; cursor: number; size: number; truncated: boolean } } @@ -154,7 +154,7 @@ export type SessionInboxSyntheticPayload1 = { text: string; description?: string export type ShellInfo = { id: string - status: "running" | "exited" | "timeout" | "killed" + status: "running" | "exited" | "timeout" | "killed" | "unavailable" command: string cwd: string shell: string @@ -386,7 +386,7 @@ export type PersistentPtyHandoff = { directory: string; instanceID: string; tick export type ShellInfo1 = { id: string - status: "running" | "exited" | "timeout" | "killed" + status: "running" | "exited" | "timeout" | "killed" | "unavailable" command: string cwd: string shell: string @@ -1065,7 +1065,7 @@ export type ShellExited = { metadata?: { [x: string]: any } type: "shell.exited" location?: LocationRef - data: { id: string; exit?: number; status: "running" | "exited" | "timeout" | "killed" } + data: { id: string; exit?: number; status: "running" | "exited" | "timeout" | "killed" | "unavailable" } } export type ShellDeleted = { @@ -2945,7 +2945,7 @@ export type SessionImportInput = { readonly type: "shell" readonly shellID: string readonly command: string - readonly status: "running" | "exited" | "timeout" | "killed" + readonly status: "running" | "exited" | "timeout" | "killed" | "unavailable" readonly exit?: number | "Infinity" | "-Infinity" | "NaN" readonly output?: { readonly output: string @@ -3224,7 +3224,7 @@ export type SessionImportInput = { readonly type: "shell" readonly shellID: string readonly command: string - readonly status: "running" | "exited" | "timeout" | "killed" + readonly status: "running" | "exited" | "timeout" | "killed" | "unavailable" readonly exit?: number | "Infinity" | "-Infinity" | "NaN" readonly output?: { readonly output: string @@ -3503,7 +3503,7 @@ export type SessionImportInput = { readonly type: "shell" readonly shellID: string readonly command: string - readonly status: "running" | "exited" | "timeout" | "killed" + readonly status: "running" | "exited" | "timeout" | "killed" | "unavailable" readonly exit?: number | "Infinity" | "-Infinity" | "NaN" readonly output?: { readonly output: string @@ -6057,6 +6057,18 @@ export type ShellOutputOutput = { data: { output: string; cursor: number; size: number; truncated: boolean } } +export type ShellStopInput = { + readonly id: { readonly id: string }["id"] + readonly location?: { + readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined + }["location"] +} + +export type ShellStopOutput = { + location: { directory: string; workspaceID?: string; project: { id: string; directory: string; canonical: string } } + data: ShellInfo1 +} + export type ShellRemoveInput = { readonly id: { readonly id: string }["id"] readonly location?: { diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 989900e83f5b..5cbc5e466b7e 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -53,7 +53,7 @@ test("exposes every standard HTTP API group", () => { expect(Object.keys(client.pty.connect)).toEqual(["token"]) expect(Object.keys(client.experimental)).toEqual(["persistentPty"]) expect(client.experimental.persistentPty.read).toBeFunction() - expect(Object.keys(client.shell)).toEqual(["list", "create", "get", "timeout", "output", "remove"]) + expect(Object.keys(client.shell)).toEqual(["list", "create", "get", "timeout", "output", "stop", "remove"]) expect(Object.keys(client.project)).toEqual(["list", "update", "current"]) expect(Object.keys(client.worktree)).toEqual(["list", "create", "remove", "refresh"]) }) diff --git a/packages/core/src/config/plugin/command.ts b/packages/core/src/config/plugin/command.ts index f7cf3647b363..f2cff8de6a6e 100644 --- a/packages/core/src/config/plugin/command.ts +++ b/packages/core/src/config/plugin/command.ts @@ -118,7 +118,7 @@ export const Plugin = define({ description: command.description ?? name, } yield* subagents.start(recovery) - yield* subagents.background(recovery) + yield* subagents.background(child.id) return } if (agent !== undefined) { diff --git a/packages/core/src/job.ts b/packages/core/src/job.ts index 08faea74e1e3..bd06dc29f489 100644 --- a/packages/core/src/job.ts +++ b/packages/core/src/job.ts @@ -6,6 +6,17 @@ import { Identifier } from "./id/id.js" import { KV } from "./kv.js" import { SessionMessage } from "./session/message.js" import { SessionSchema } from "./session/schema.js" +import { ShellResult } from "./shell/result.js" +import { SubagentOutcome } from "./session/subagent-outcome.js" + +/** + * The producer's typed account of how its work ended. A job never classifies the work itself: + * `status` says whether the run reported an outcome (`completed`), died (`error`), or was + * abandoned before reporting (`cancelled`). A user stop, a timeout, or a nonzero exit are all + * `completed` runs whose outcome says so. + */ +export const Outcome = Schema.Union([ShellResult.Outcome, SubagentOutcome.Outcome]) +export type Outcome = typeof Outcome.Type const Background = Schema.Struct({ id: Schema.String, @@ -26,6 +37,8 @@ const Background = Schema.Struct({ }), ]), status: Schema.Literals(["running", "completed", "error", "cancelled"]), + result: Schema.optionalKey(Outcome), + // Read-only compatibility with markers written before typed outcomes. Never reconstruct facts from this text. output: Schema.optionalKey(Schema.String), error: Schema.optionalKey(Schema.String), }) @@ -33,6 +46,8 @@ const Background = Schema.Struct({ export type Background = typeof Background.Type export type Recovery = Background["recovery"] export type Status = Background["status"] +/** One job's terminal facts, shared by live Info and the durable background marker. */ +export type Terminal = Pick const decodeBackground = Schema.decodeUnknownResult(Background) const backgroundPrefix = "job.background/" @@ -44,10 +59,11 @@ export type Info = { status: Status started_at: number completed_at?: number - output?: string + result?: Outcome error?: string metadata?: Record notificationID?: SessionMessage.ID + recovery?: Recovery } type Active = { @@ -57,7 +73,6 @@ type Active = { scope: Scope.Closeable blockingSessions: Map isBackgrounded: boolean - recovery?: Recovery } type State = { @@ -96,7 +111,7 @@ export type StartInput = { metadata?: Record recovery?: Recovery notificationID?: SessionMessage.ID - run: Effect.Effect + run: Effect.Effect } export type WaitInput = { @@ -172,18 +187,18 @@ export const make = Effect.gen(function* () { } const persistBackground = Effect.fnUntraced(function* (job: Active) { - if (!job.recovery || !job.info.notificationID) return + if (!job.info.recovery || !job.info.notificationID) return yield* kv.set(`${backgroundPrefix}${job.info.notificationID}`, { id: job.info.id, notificationID: job.info.notificationID, - recovery: job.recovery, + recovery: job.info.recovery, status: job.info.status, - ...(job.info.output !== undefined ? { output: job.info.output } : {}), + ...(job.info.result !== undefined ? { result: job.info.result } : {}), ...(job.info.error !== undefined ? { error: job.info.error } : {}), }) }) - const settle = Effect.fnUntraced(function* (id: string, scope: Scope.Closeable, exit: Exit.Exit) { + const settle = Effect.fnUntraced(function* (id: string, scope: Scope.Closeable, exit: Exit.Exit) { const completed_at = yield* Clock.currentTimeMillis const result = yield* SynchronizedRef.modifyEffect( state.jobs, @@ -204,7 +219,7 @@ export const make = Effect.gen(function* () { ...job.info, status, completed_at, - ...(Exit.isSuccess(exit) ? { output: exit.value } : {}), + ...(Exit.isSuccess(exit) ? { result: exit.value } : {}), ...(Exit.isFailure(exit) ? { error: errorText(Cause.squash(exit.cause)) } : {}), }, } @@ -248,6 +263,7 @@ export const make = Effect.gen(function* () { status: "running" as const, started_at, metadata: input.metadata, + recovery: input.recovery, ...(input.notificationID ? { notificationID: input.notificationID } : {}), }, done, @@ -255,7 +271,6 @@ export const make = Effect.gen(function* () { scope, blockingSessions: new Map(), isBackgrounded: false, - recovery: input.recovery, } return [{ info: snapshot(job), scope }, new Map(jobs).set(id, job)] }), @@ -324,7 +339,7 @@ export const make = Effect.gen(function* () { blockingSessions: new Map(), info: { ...job.info, - ...(job.recovery ? { notificationID: job.info.notificationID ?? SessionMessage.ID.create() } : {}), + ...(job.info.recovery ? { notificationID: job.info.notificationID ?? SessionMessage.ID.create() } : {}), }, } yield* persistBackground(next) @@ -337,7 +352,7 @@ export const make = Effect.gen(function* () { Effect.fnUntraced(function* (jobs): Effect.fn.Return]> { const job = jobs.get(id) // Recoverable work may finish before the caller backgrounds it. - if (!job || (job.info.status !== "running" && !job.recovery)) return [{}, jobs] + if (!job || (job.info.status !== "running" && !job.info.recovery)) return [{}, jobs] if (job.isBackgrounded) return [{ info: snapshot(job) }, jobs] const next = yield* markBackground(job) return [{ info: snapshot(next), backgrounded: job.backgrounded }, new Map(jobs).set(id, next)] diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 3e8fdb577779..b960aa468c91 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -196,7 +196,9 @@ export interface Interface { readonly wait: (id: SessionSchema.ID) => Effect.Effect readonly active: Effect.Effect> readonly background: (sessionID: SessionSchema.ID) => Effect.Effect - readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect + readonly resume: ( + sessionID: SessionSchema.ID, + ) => Effect.Effect readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect readonly synthetic: ( input: Parameters[0] & { sessionID: SessionSchema.ID }, diff --git a/packages/core/src/session/background-notice.ts b/packages/core/src/session/background-notice.ts new file mode 100644 index 000000000000..299337cb0cb0 --- /dev/null +++ b/packages/core/src/session/background-notice.ts @@ -0,0 +1,89 @@ +export * as BackgroundNotice from "./background-notice.js" + +import { Effect } from "effect" +import type { Job } from "../job.js" +import type { Session } from "../session.js" +import { ShellResult } from "../shell/result.js" +import type { SessionMessage } from "./message.js" +import { SubagentOutcome } from "./subagent-outcome.js" + +export type Input = Job.Terminal & { + id: string + recovery: Job.Recovery + notificationID?: SessionMessage.ID + resume?: boolean +} + +/** + * Admits one background job's terminal as a synthetic notice to the Session that owns it, then + * acknowledges the durable marker. Live observers and restart recovery share this path, so a + * notice renders the same from the same terminal whenever it is delivered. A user stop is a quiet + * notice: recorded for the model's next step, never a wake of an idle Session. + */ +export const deliver = Effect.fnUntraced(function* ( + sessions: Pick, + jobs: Pick, + input: Input, +) { + const notice = input.recovery.kind === "shell" ? shell(input, input.recovery) : subagent(input, input.recovery) + yield* sessions.synthetic({ + ...(input.notificationID ? { id: input.notificationID } : {}), + sessionID: input.recovery.kind === "shell" ? input.recovery.sessionID : input.recovery.parentSessionID, + ...(input.resume === false || notice.state === "stopped" ? { resume: false } : {}), + description: input.recovery.kind === "shell" ? input.recovery.command : input.recovery.description, + text: notice.text, + metadata: notice.metadata, + }) + if (input.notificationID) yield* jobs.completeBackground(input.notificationID) +}) + +// A marker still `running` at delivery means the process died with the work: report it as cancelled. +function state(input: Job.Terminal, stopped: boolean): ShellResult.State { + if (input.status === "completed") return stopped ? "stopped" : "completed" + if (input.status === "error") return "error" + return "cancelled" +} + +function shell(input: Input, recovery: Extract) { + const outcome = input.result?.kind === "shell" ? input.result : undefined + const ended = outcome ? ShellResult.state(outcome) : state(input, false) + const text = outcome + ? ShellResult.text(outcome) + : input.status === "completed" + ? (input.output ?? "Command completed") + : input.status === "error" + ? (input.error ?? "Command failed") + : input.status === "running" + ? "Command cancelled because the server restarted" + : "Command cancelled" + return { + state: ended, + ...ShellResult.notification({ + jobID: input.id, + shellID: recovery.shellID, + command: recovery.command, + state: ended, + text, + outcome, + }), + } +} + +function subagent(input: Input, recovery: Extract) { + const outcome = input.result?.kind === "subagent" ? input.result : undefined + const ended = state(input, outcome?.status === "interrupted") + const text = outcome + ? outcome.status === "completed" + ? outcome.text + : SubagentOutcome.stopped + : input.status === "completed" + ? (input.output ?? "Subagent completed without a text response.") + : input.status === "error" + ? (input.error ?? "Subagent failed") + : "Subagent cancelled" + return { + state: ended, + text: `\n${text}\n`, + metadata: { source: "subagent", childID: recovery.childSessionID, agent: recovery.agent, state: ended }, + } +} diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 9ff89f5113f1..9b620162f80d 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -3,7 +3,6 @@ export * as SessionExecution from "./execution.js" import { Cause, Context, Effect, Exit, Layer } from "effect" import { Bus } from "../bus.js" import { Database } from "../database/database.js" -import { Job } from "../job.js" import { Instance } from "../instance/service.js" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { SessionEvent } from "./event.js" @@ -15,13 +14,20 @@ import { toSessionError } from "./to-session-error.js" import { UserInterruptedError } from "./error.js" import { SessionInbox } from "./inbox.js" +/** + * How a joined execution ended. A user interruption is an outcome the joiner learns about; + * shutdown interruption is not: it relinquishes local execution while the durable claim keeps + * restart continuity, so it interrupts process-local joiners instead of resolving them. + */ +export type Terminal = { readonly type: "succeeded" } | { readonly type: "interrupted"; readonly reason: "user" } + export interface Interface { /** Snapshots active execution owned by this process. */ readonly active: Effect.Effect> /** Checks process-local ownership, including interruption cleanup and terminal settlement. */ readonly isActive: (sessionID: SessionSchema.ID) => Effect.Effect - /** Starts execution while idle or joins the active execution. */ - readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect + /** Starts execution while idle or joins the active execution, and returns how it ended. */ + readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect /** Registers newly recorded work. Repeated wakeups may coalesce. */ readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect /** @@ -42,7 +48,7 @@ type InterruptReason = "user" | "shutdown" export function terminal(exit: Exit.Exit, reason?: InterruptReason) { if (Exit.isSuccess(exit)) return { type: "succeeded" as const } - if (Cause.hasInterrupts(exit.cause)) return { type: "interrupted" as const, reason: reason ?? "shutdown" } + if (Cause.hasInterruptsOnly(exit.cause)) return { type: "interrupted" as const, reason: reason ?? "shutdown" } const failure = Cause.squash(exit.cause) if (failure instanceof UserInterruptedError) return { type: "interrupted" as const, reason: "user" as const } return { type: "failed" as const, error: toSessionError(failure) } @@ -55,7 +61,6 @@ export const layer = Layer.effect( const store = yield* SessionStore.Service const instances = yield* Instance.Service const bus = yield* Bus.Service - const jobs = yield* Job.Service const db = (yield* Database.Service).db const reportLifecycle = (sessionID: SessionSchema.ID, effect: Effect.Effect) => effect.pipe( @@ -122,7 +127,6 @@ 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) yield* bus.publish( SessionEvent.Execution.Interrupted, { sessionID, reason: outcome.reason }, @@ -163,7 +167,14 @@ export const layer = Layer.effect( yield* coordinator.wake(sessionID, "steer") return interrupted }), - resume: coordinator.run, + resume: (sessionID) => + coordinator.run(sessionID).pipe( + Effect.flatMap((terminal): Effect.Effect => { + if (terminal.type === "succeeded") return Effect.succeed(terminal) + if (terminal.reason === "user") return Effect.succeed({ type: "interrupted", reason: "user" }) + return Effect.interrupt + }), + ), wake: coordinator.wake, awaitIdle: coordinator.awaitIdle, }) @@ -173,7 +184,7 @@ export const layer = Layer.effect( export const node = makeGlobalNode({ service: Service, layer, - deps: [SessionStore.node, Instance.node, Bus.node, Database.node, Job.node], + deps: [SessionStore.node, Instance.node, Bus.node, Database.node], }) /** Low-level compatibility layer for callers that only need durable Session recording. */ @@ -182,7 +193,7 @@ export const noopLayer = Layer.succeed( Service.of({ active: Effect.succeed(new Set()), isActive: () => Effect.succeed(false), - resume: () => Effect.void, + resume: () => Effect.succeed({ type: "succeeded" }), wake: () => Effect.void, interrupt: () => Effect.succeed(false), awaitIdle: () => Effect.void, diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index af048a1e9da0..a246243c1b78 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -9,8 +9,8 @@ import { SessionEvent } from "../event.js" import { SessionExecution } from "../execution.js" import { SessionSchema } from "../schema.js" import { SessionStore } from "../store.js" -import { ShellResult } from "../../shell/result.js" -import { SubagentCompletion } from "../subagent-completion.js" +import { BackgroundNotice } from "../background-notice.js" +import { SubagentOutcome } from "../subagent-outcome.js" const CONTINUE_AFTER_SERVER_RESTART = "The server restarted while you were working. Continue from where you left off without repeating completed work." @@ -97,40 +97,24 @@ export const layer = (options?: Options) => return true }) - const recoverShell = Effect.fnUntraced(function* ( + const notify = Effect.fnUntraced(function* ( background: Job.Background, - recovery: Extract, + terminal: Job.Terminal, suspended: ReadonlySet, ) { - 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" - - yield* sessions - .synthetic({ - id: background.notificationID, - sessionID: recovery.sessionID, - description: recovery.command, - ...ShellResult.notification({ - jobID: background.id, - shellID: recovery.shellID, - command: recovery.command, - state, - text, - }), - ...(suspended.has(recovery.sessionID) ? { resume: false } : {}), - }) - .pipe( - Effect.catchTag("Session.NotFoundError", () => Effect.void), - Effect.orDie, - ) - yield* jobs.completeBackground(background.notificationID) + const recipient = + background.recovery.kind === "shell" ? background.recovery.sessionID : background.recovery.parentSessionID + yield* BackgroundNotice.deliver(sessions, jobs, { + ...terminal, + id: background.id, + recovery: background.recovery, + notificationID: background.notificationID, + resume: suspended.has(recipient) ? false : undefined, + }).pipe( + // A deleted recipient has nothing to be told; drop its marker with the notice. + Effect.catchTag("Session.NotFoundError", () => jobs.completeBackground(background.notificationID)), + Effect.orDie, + ) }) const recoverSubagent = Effect.fnUntraced(function* ( @@ -143,23 +127,13 @@ export const layer = (options?: Options) => yield* jobs.completeBackground(background.notificationID) return } - - const notify = Effect.fnUntraced(function* (result: Pick) { - yield* SubagentCompletion.deliver(sessions, jobs, { - ...result, - recovery, - notificationID: background.notificationID, - resume: suspended.has(recovery.parentSessionID) ? false : undefined, - }).pipe(Effect.orDie) - }) - if (background.status !== "running") { - yield* notify(background) + yield* notify(background, background, suspended) return } if (yield* execution.isActive(recovery.childSessionID)) return if (!(yield* prepareResume(recovery.childSessionID))) { - yield* notify({ status: "error", error: RESUME_EXHAUSTED.message }) + yield* notify(background, { status: "error", error: RESUME_EXHAUSTED.message }, suspended) return } @@ -169,20 +143,13 @@ export const layer = (options?: Options) => title: recovery.description, notificationID: background.notificationID, recovery, - run: execution.resume(recovery.childSessionID).pipe( - Effect.andThen(store.context(recovery.childSessionID)), - Effect.map((messages) => { - const assistant = messages.findLast( - (message) => - message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, - ) - return SubagentCompletion.text(assistant) - }), - ), + run: SubagentOutcome.run(sessions, recovery.childSessionID), }) yield* jobs.background(background.id) yield* jobs.wait({ id: background.id }).pipe( - Effect.flatMap((result) => (result.info ? notify(result.info) : Effect.void)), + Effect.flatMap((result) => + result.info && result.info.status !== "running" ? notify(background, result.info, suspended) : Effect.void, + ), Effect.forkIn(scope), ) }) @@ -207,8 +174,10 @@ export const layer = (options?: Options) => Effect.fnUntraced(function* (background) { if ((yield* jobs.get(background.id))?.status === "running") return const recovery = background.recovery + // A shell cannot be rerun after its process died: deliver the marker's terminal as-is, + // which the notice reports as cancelled when the command was still running. yield* recovery.kind === "shell" - ? recoverShell(background, recovery, suspended) + ? notify(background, background, suspended) : recoverSubagent(background, recovery, suspended) }), { discard: true }, diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 7130f5ed5e2c..698ce17759f6 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -1,16 +1,24 @@ export * as SessionRunCoordinator from "./run-coordinator.js" -import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect" import type { Promotable } from "./inbox.js" +/** + * How an execution ended, as observed by joiners. Interruption is reported as a value: the + * execution was interrupted, not the fiber asking how it went. Failures still fail with `E`. + */ +export type Terminal = + | { readonly type: "succeeded" } + | { readonly type: "interrupted"; readonly reason?: Reason } + /** Serializes execution for each key while allowing different keys to run concurrently. */ export interface Coordinator { /** Snapshots keys with an execution owned by this coordinator. */ readonly active: Effect.Effect> /** Checks ownership for one key, including cleanup and terminal settlement. */ readonly isActive: (key: Key) => Effect.Effect - /** Starts an execution while idle, or joins the active execution and returns its exit. */ - readonly run: (key: Key) => Effect.Effect + /** Starts an execution while idle, or joins the active execution, and returns how it ended. */ + readonly run: (key: Key) => Effect.Effect, E> /** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */ readonly wake: (key: Key, scope?: Promotable) => Effect.Effect /** @@ -30,10 +38,10 @@ export interface Coordinator { * execution rings it with the scope that work needs, and the execution loop drains again * instead of ending. The doorbell closes the gap between a drain's last eligibility check * and the idle transition, since those cannot be one atomic step. `done` resolves joiners - * with this execution's exit. + * with how this execution ended. */ type Execution = { - readonly done: Deferred.Deferred + readonly done: Deferred.Deferred, E> owner?: Fiber.Fiber scope: Promotable pendingWake?: Promotable @@ -81,7 +89,7 @@ export const make = (options: { const start = (key: Key, force: boolean, scope: Promotable) => { const execution: Execution = { - done: Deferred.makeUnsafe(), + done: Deferred.makeUnsafe, E>(), scope, stopping: false, } @@ -111,12 +119,17 @@ export const make = (options: { const settle = (key: Key, execution: Execution, exit: Exit.Exit) => { if (execution.pendingWake) start(key, false, execution.pendingWake) else executions.delete(key) - Deferred.doneUnsafe(execution.done, exit) + const ended: Effect.Effect, E> = Exit.isSuccess(exit) + ? Effect.succeed({ type: "succeeded" }) + : Cause.hasInterruptsOnly(exit.cause) + ? Effect.succeed({ type: "interrupted", reason: execution.interruptionReason }) + : Effect.failCause(exit.cause) + Deferred.doneUnsafe(execution.done, ended) } const isActive = (key: Key) => Effect.sync(() => executions.has(key)) - const run = (key: Key): Effect.Effect => + const run = (key: Key): Effect.Effect, E> => Effect.suspend(() => { const execution = executions.get(key) if (execution !== undefined) { diff --git a/packages/core/src/session/session.ts b/packages/core/src/session/session.ts index b67366ce0044..a5fd9ee0c6f5 100644 --- a/packages/core/src/session/session.ts +++ b/packages/core/src/session/session.ts @@ -277,7 +277,7 @@ export const make = Effect.fn("Session.make")(function* () { }) const resume = Effect.fn("Session.resume")(function* (sessionID: SessionSchema.ID) { yield* get(sessionID) - yield* execution.resume(sessionID) + return yield* execution.resume(sessionID) }) const synthetic = Effect.fn("Session.synthetic")( ( diff --git a/packages/core/src/session/subagent-completion.ts b/packages/core/src/session/subagent-completion.ts deleted file mode 100644 index 746bc5170ee5..000000000000 --- a/packages/core/src/session/subagent-completion.ts +++ /dev/null @@ -1,45 +0,0 @@ -export * as SubagentCompletion from "./subagent-completion.js" - -import { Effect } from "effect" -import type { Job } from "../job.js" -import type { Session } from "../session.js" -import type { SessionMessage } from "./message.js" - -export const NO_TEXT = "Subagent completed without a text response." - -export function text(message: SessionMessage.Info | undefined) { - if (message?.type !== "assistant") return NO_TEXT - return ( - message.content - .filter((part) => part.type === "text") - .map((part) => part.text) - .join("") || NO_TEXT - ) -} - -export const deliver = Effect.fnUntraced(function* ( - sessions: Pick, - jobs: Pick, - input: Pick & { - recovery: Extract - resume?: boolean - }, -) { - if (input.status === "running") return - const recovery = input.recovery - const text = - input.status === "completed" - ? (input.output ?? NO_TEXT) - : input.status === "error" - ? (input.error ?? "Subagent failed") - : "Subagent cancelled" - yield* sessions.synthetic({ - ...(input.notificationID ? { id: input.notificationID } : {}), - sessionID: recovery.parentSessionID, - ...(input.resume === false ? { resume: false } : {}), - description: recovery.description, - text: `\n${text}\n`, - metadata: { source: "subagent", childID: recovery.childSessionID, agent: recovery.agent, state: input.status }, - }) - if (input.notificationID) yield* jobs.completeBackground(input.notificationID) -}) diff --git a/packages/core/src/session/subagent-job.ts b/packages/core/src/session/subagent-job.ts index f8e7503fbd0a..2983be4d5b84 100644 --- a/packages/core/src/session/subagent-job.ts +++ b/packages/core/src/session/subagent-job.ts @@ -3,14 +3,16 @@ export * as SubagentJob from "./subagent-job.js" import { Effect, Scope } from "effect" import { Job } from "../job.js" import { Session } from "../session.js" -import { SubagentCompletion } from "./subagent-completion.js" +import { BackgroundNotice } from "./background-notice.js" +import { SubagentOutcome } from "./subagent-outcome.js" +import type { SessionSchema } from "./schema.js" type Recovery = Extract interface Runner { start: (recovery: Recovery) => Effect.Effect - background: (recovery: Recovery) => Effect.Effect - notify: (recovery: Recovery, startedAt: number) => Effect.Effect + background: (childID: SessionSchema.ID) => Effect.Effect + notify: (job: Job.Info) => Effect.Effect } export const make: Effect.Effect = Effect.gen(function* () { @@ -20,13 +22,14 @@ export const make: Effect.Effect() - const notify = Effect.fn("SubagentJob.notify")(function* (recovery: Recovery, startedAt: number) { - const key = `${recovery.childSessionID}:${startedAt}` + const notify = Effect.fn("SubagentJob.notify")(function* (job: Job.Info) { + const key = `${job.id}:${job.started_at}` if (notifications.has(key)) return notifications.add(key) yield* Effect.gen(function* () { - const info = (yield* jobs.wait({ id: recovery.childSessionID })).info - if (info) yield* SubagentCompletion.deliver(sessions, jobs, { ...info, recovery }) + const info = (yield* jobs.wait({ id: job.id })).info + if (info?.recovery && info.status !== "running") + yield* BackgroundNotice.deliver(sessions, jobs, { ...info, recovery: info.recovery }) }).pipe( Effect.ensuring(Effect.sync(() => notifications.delete(key))), Effect.forkIn(scope, { startImmediately: true }), @@ -41,19 +44,11 @@ export const make: Effect.Effect - message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, - ) - return SubagentCompletion.text(assistant) - }), + run: SubagentOutcome.run(sessions, recovery.childSessionID), }), - background: Effect.fn("SubagentJob.background")(function* (recovery: Recovery) { - const info = yield* jobs.background(recovery.childSessionID) - if (info) yield* notify(recovery, info.started_at) + background: Effect.fn("SubagentJob.background")(function* (childID: SessionSchema.ID) { + const info = yield* jobs.background(childID) + if (info) yield* notify(info) }), notify, } diff --git a/packages/core/src/session/subagent-outcome.ts b/packages/core/src/session/subagent-outcome.ts new file mode 100644 index 000000000000..363d94c76e82 --- /dev/null +++ b/packages/core/src/session/subagent-outcome.ts @@ -0,0 +1,48 @@ +export * as SubagentOutcome from "./subagent-outcome.js" + +import { Effect, Schema } from "effect" +import type { Session } from "../session.js" +import type { SessionSchema } from "./schema.js" + +const NO_TEXT = "Subagent completed without a text response." +export const stopped = "Subagent stopped by user. Do not restart it unless the user asks." + +/** + * How one child execution ended, as the parent's job saw it. Shutdown never settles here: it + * interrupts the joining run so its durable marker stays `running` and restart resumes the child. + */ +export const Outcome = Schema.Union([ + Schema.Struct({ + kind: Schema.Literal("subagent"), + status: Schema.Literal("completed"), + text: Schema.String, + }), + Schema.Struct({ + kind: Schema.Literal("subagent"), + status: Schema.Literal("interrupted"), + }), +]) +export type Outcome = typeof Outcome.Type + +/** Runs or joins the child's execution and reports how it ended, with its final assistant text. */ +export const run = Effect.fnUntraced(function* ( + sessions: Pick, + childID: SessionSchema.ID, +) { + const terminal = yield* sessions.resume(childID) + if (terminal.type === "interrupted") return { kind: "subagent", status: "interrupted" } as const + // Concatenate the child's final completed assistant text. "Completed with no text" is a + // completion; a failed run is the job's error, not an outcome. + const messages = yield* sessions.messages({ sessionID: childID, order: "desc", limit: 20 }) + const assistant = messages.find( + (message) => message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, + ) + const text = + assistant?.type === "assistant" + ? assistant.content + .filter((part): part is Extract => part.type === "text") + .map((part) => part.text) + .join("") + : "" + return { kind: "subagent", status: "completed", text: text.length > 0 ? text : NO_TEXT } as const +}) diff --git a/packages/core/src/shell.ts b/packages/core/src/shell.ts index bf147ab41a3f..a4b83fadab39 100644 --- a/packages/core/src/shell.ts +++ b/packages/core/src/shell.ts @@ -44,9 +44,10 @@ type Active = { size: number // Resolves with the terminal Info once the command exits, times out, or is killed. A wait // started after termination resolves immediately from the already-completed deferred. - done: Deferred.Deferred + done: Deferred.Deferred timeoutFiber?: Fiber.Fiber timeout?: (duration: number) => Effect.Effect + stop?: Effect.Effect } /** @@ -67,12 +68,16 @@ export interface Interface { readonly get: (id: Shell.ID) => Effect.Effect // 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 - // A known shell's terminal state and bounded tail. Missing capture remains distinct from its exit status. + readonly wait: (id: Shell.ID) => Effect.Effect + // A known terminal state and bounded tail, or `unavailable` if observation outlives retention/removal. readonly result: (started: Shell.Info) => Effect.Effect // Replaces the running command's timeout from now; zero clears it. readonly timeout: (id: Shell.ID, duration: number) => Effect.Effect readonly output: (id: Shell.ID, input?: Shell.OutputInput) => Effect.Effect + // Kills a running command. It ends as `killed` and stays observable like any exited command, so + // waiters see the real terminal Info and its captured output remains readable. Stopping is not + // removal: `remove` forgets the command and its capture. Resolves once the command is terminal. + readonly stop: (id: Shell.ID) => Effect.Effect readonly remove: (id: Shell.ID) => Effect.Effect } @@ -193,6 +198,14 @@ const layer = () => return command.info }) + const stop = Effect.fn("Shell.stop")(function* (id: Shell.ID) { + const command = yield* require(id) + // Once accepted, a client's cancellation cannot abandon killing or capture settlement. + // Concurrent callers wait on the same completion, not the early status change. + if (command.info.status === "running" && command.stop) yield* command.stop + return yield* Deferred.await(command.done) + }, Effect.uninterruptible) + const output = Effect.fnUntraced(function* (id: Shell.ID, input?: Shell.OutputInput) { const command = yield* require(id) const cursor = input?.cursor ?? 0 @@ -225,9 +238,7 @@ 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() } }), - ), + Effect.catchTag("Shell.NotFoundError", () => Effect.succeed({ ...started, status: "unavailable" as const })), ) const capture = yield* Effect.gen(function* () { const limits = Config.latest(yield* config.entries(), "tool_output") @@ -309,7 +320,7 @@ const layer = () => }), file, size: 0, - done: Deferred.makeUnsafe(), + done: Deferred.makeUnsafe(), } commands.set(id, command) @@ -343,21 +354,23 @@ const layer = () => }), ) - const finish = (status: Info["status"], exit?: number, beforeWait = Effect.void) => + const finish = (status: ShellResult.TerminalInfo["status"], exit?: number, beforeWait = Effect.void) => Effect.gen(function* () { if (command.info.status !== "running") return - command.info = produce(command.info, (draft) => { - draft.status = status - if (exit !== undefined) draft.exit = exit - draft.time.completed = Date.now() - }) + const info = { + ...command.info, + status, + ...(exit !== undefined ? { exit } : {}), + time: { ...command.info.time, completed: Date.now() }, + } + command.info = info yield* beforeWait yield* outputDone.await // Resolve waiters with the terminal Info before any retention eviction, so an evicted // command still reports success rather than the removal NotFoundError. This runs before // the timeout-fiber interrupt below, which on the timeout path would otherwise cancel // this very fiber (finish is invoked by the timeout fiber) before waiters are resolved. - yield* Deferred.succeed(command.done, command.info) + yield* Deferred.succeed(command.done, info) yield* bus.publish(Shell.Event.Exited, { id, ...(exit !== undefined ? { exit } : {}), @@ -372,6 +385,7 @@ const layer = () => // Keep exited history data-only. Interrupt last because finish may run on the timeout fiber. const timeoutFiber = command.timeoutFiber command.timeout = undefined + command.stop = undefined command.timeoutFiber = undefined if (timeoutFiber) yield* Fiber.interrupt(timeoutFiber) }) @@ -395,6 +409,11 @@ const layer = () => }) yield* command.timeout(invocation.timeout) + command.stop = finish( + "killed", + undefined, + handle.kill({ forceKillAfter: Duration.seconds(3) }).pipe(Effect.catch(() => Effect.void)), + ) runFork( handle.exitCode.pipe( @@ -416,7 +435,7 @@ const layer = () => return command.info }) - return Service.of({ create, list, get, wait, result, timeout, output, remove }) + return Service.of({ create, list, get, wait, result, timeout, output, stop, remove }) }), ) diff --git a/packages/core/src/shell/result.ts b/packages/core/src/shell/result.ts index 4c377e779b75..49c902a23566 100644 --- a/packages/core/src/shell/result.ts +++ b/packages/core/src/shell/result.ts @@ -1,13 +1,28 @@ export * as ShellResult from "./result.js" import type { Shell } from "@opencode-ai/schema/shell" +import { Schema } from "effect" + +export type TerminalInfo = Shell.Info & { status: Exclude } export type Result = { - info: Shell.Info + info: TerminalInfo capture: { output: string; truncated: boolean } | undefined } -type Output = { output: string; truncated: boolean; exit?: number; timeout?: boolean } +/** + * How one shell command ended, as the producer saw it. This is the shell job's typed result: + * the foreground tool response, the background notice, and restart recovery all render from it. + * `killed` records an explicit stop; removal and expired results are `unavailable`. + */ +export const Outcome = Schema.Struct({ + kind: Schema.Literal("shell"), + status: Schema.Literals(["exited", "timeout", "killed", "unavailable"]), + exit: Schema.optionalKey(Schema.Number), + output: Schema.String, + truncated: Schema.Boolean, +}) +export type Outcome = typeof Outcome.Type const missing = "Shell command output is no longer available." export const unavailable: Shell.Output = { @@ -17,35 +32,53 @@ export const unavailable: Shell.Output = { truncated: false, } -export function output(result: Result): Output { +export const stopped = "Command stopped by user. Do not restart it unless the user asks." + +export function outcome(result: Result): Outcome { return { + kind: "shell", + status: result.info.status, + ...(result.info.exit !== undefined ? { exit: result.info.exit } : {}), output: result.capture?.output ?? unavailable.output, truncated: result.capture?.truncated ?? false, - ...(result.info.exit !== undefined ? { exit: result.info.exit } : {}), - ...(result.info.status === "timeout" ? { timeout: true } : {}), } } -export function notice(output: Pick) { - if (output.timeout) return "Command timed out before completion." - if (output.exit !== undefined) return `Command exited with code ${output.exit}.` +export function notice(outcome: Pick) { + if (outcome.status === "killed") return stopped + if (outcome.status === "unavailable") return missing + if (outcome.status === "timeout") return "Command timed out before completion." + return `Command exited with code ${outcome.exit ?? "unknown"}.` } -export function metadata(output: Output) { +/** Model-visible text: the bounded output followed by how the command ended. */ +export function text(outcome: Outcome) { + if (outcome.status === "unavailable") return missing + return `${outcome.output}\n\n${notice(outcome)}` +} + +export function metadata(outcome: Pick) { return { - truncated: output.truncated, - ...(output.exit !== undefined ? { exit: output.exit } : {}), - ...(output.timeout !== undefined ? { timeout: output.timeout } : {}), + truncated: outcome.truncated, + ...(outcome.exit !== undefined ? { exit: outcome.exit } : {}), + ...(outcome.status === "timeout" ? { timeout: true } : {}), } } +export type State = "completed" | "stopped" | "cancelled" | "error" + +export function state(outcome: Pick): State { + if (outcome.status === "unavailable") return "error" + return outcome.status === "killed" ? "stopped" : "completed" +} + export function notification(input: { shellID: string jobID?: string command: string - state: "completed" | "cancelled" | "error" + state: State text: string - output?: Output + outcome?: Outcome }) { return { text: `\n${input.text}\n`, @@ -54,21 +87,19 @@ export function notification(input: { shellID: input.shellID, ...(input.jobID !== undefined ? { jobID: input.jobID } : {}), state: input.state, - ...(input.output ? metadata(input.output) : {}), + ...(input.outcome ? metadata(input.outcome) : {}), }, } } export function userNotification(result: Result) { - const captured = output(result) - const status = - result.info.status === "killed" ? "Command cancelled." : (notice(captured) ?? "Command exited with code unknown.") + const ended = outcome(result) const message = notification({ shellID: result.info.id, command: result.info.command, - state: result.info.status === "killed" ? "cancelled" : "completed", - text: `${captured.output}\n\n${status}`, - output: captured, + state: state(ended), + text: text(ended), + outcome: ended, }) return { ...message, text: `The following shell command was executed by the user:\n${message.text}` } } diff --git a/packages/core/src/tool/plugin/shell.ts b/packages/core/src/tool/plugin/shell.ts index bb1c0a440945..b73ccb1c2cdd 100644 --- a/packages/core/src/tool/plugin/shell.ts +++ b/packages/core/src/tool/plugin/shell.ts @@ -4,7 +4,7 @@ import { ToolFailure } from "@opencode-ai/ai" import type { Context } from "@opencode-ai/plugin/effect/plugin" import type { ShellCreateBefore } from "@opencode-ai/plugin/effect/shell" import type { Tool } from "@opencode-ai/schema/tool" -import { Deferred, Effect, Schema, Scope } from "effect" +import { Effect, Schema, Scope } from "effect" import { Config } from "../../config.js" import { Environment } from "../../environment/index.js" import { Job } from "../../job.js" @@ -12,7 +12,7 @@ import { LocationMutation } from "../../location-mutation.js" import { Permission } from "../../permission.js" import { NonNegativeInt } from "../../schema.js" import { Session } from "../../session.js" -import { SessionSchema } from "../../session/schema.js" +import { BackgroundNotice } from "../../session/background-notice.js" import { Shell } from "../../shell.js" import { ShellParse } from "../../shell/parse.js" import { ShellSelect } from "../../shell/select.js" @@ -68,34 +68,45 @@ 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", "stopped"])), }) type Output = typeof Output.Type -const resultMessages = (output: Output) => { - const notice = output.status === "running" ? BACKGROUND_INSTRUCTION : ShellResult.notice(output) - return [output.output, ...(notice ? [notice] : [])] -} - -const toolResult = (output: Output) => { +const toolResult = (output: Output, notice: string) => { return { output, - content: resultMessages(output).map((text) => ({ type: "text" as const, text })), + content: [output.output, notice].map((text) => ({ type: "text" as const, text })), metadata: { status: output.status, - ...ShellResult.metadata(output), + truncated: output.truncated, + ...(output.exit !== undefined ? { exit: output.exit } : {}), + ...(output.timeout !== undefined ? { timeout: output.timeout } : {}), ...(output.shellID !== undefined ? { shellID: output.shellID } : {}), }, } } -const backgroundResult = (shellID: string, file: string) => ({ - output: `Command moved to the background (shell ID: ${shellID}).\nOutput is streaming to: ${file}`, - shellID, - truncated: false, - status: "running" as const, -}) +const completedResult = (outcome: ShellResult.Outcome) => + toolResult( + { + output: outcome.output, + ...ShellResult.metadata(outcome), + status: outcome.status === "killed" ? "stopped" : "completed", + }, + ShellResult.notice(outcome), + ) + +const backgroundResult = (shellID: string, file: string) => + toolResult( + { + output: `Command moved to the background (shell ID: ${shellID}).\nOutput is streaming to: ${file}`, + shellID, + truncated: false, + status: "running", + }, + BACKGROUND_INSTRUCTION, + ) export const Plugin = { id: "opencode.tool.shell", @@ -162,35 +173,10 @@ export const Plugin = { }) const notifyWhenDone = Effect.fn("ShellTool.notifyWhenDone")( - function* ( - sessionID: SessionSchema.ID, - id: string, - shellID: string, - command: string, - settled: Deferred.Deferred, - ) { + function* (id: string) { const info = (yield* jobs.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" - yield* sessions.synthetic({ - ...(info.notificationID ? { id: info.notificationID } : {}), - sessionID, - description: command, - ...ShellResult.notification({ - jobID: id, - shellID, - command, - state: info.status, - text, - output, - }), - }) - if (info.notificationID) yield* jobs.completeBackground(info.notificationID) + if (!info?.recovery || info.status === "running") return + yield* BackgroundNotice.deliver(sessions, jobs, { ...info, recovery: info.recovery }) }, Effect.forkIn(scope, { startImmediately: true }), ) @@ -220,23 +206,22 @@ export const Plugin = { finalTimeout = yield* prepare(invocation, context) }), ) - yield* context.progress({ shellID: info.id }) - - const settled = yield* Deferred.make() - const run = Effect.gen(function* () { - const result = yield* shell.result(info) - if (!result.capture) return yield* new Shell.NotFoundError({ id: info.id }) - const output = ShellResult.output(result) - return { - ...output, - output: output.timeout - ? `${output.output}\n\nCommand exceeded timeout of ${finalTimeout} ms. Retry with a larger timeout if the command is expected to take longer.` - : output.output, - status: "completed" as const, - } - }).pipe( - Effect.tap((output) => Deferred.succeed(settled, output)), - Effect.map((output) => resultMessages(output).join("\n\n")), + const recovery = { + kind: "shell" as const, + sessionID: context.sessionID, + shellID: info.id, + command: info.command, + } + // The shell's own terminal state is the job's result; nothing here reclassifies it. + const run = shell.result(info).pipe( + Effect.map((result) => { + const outcome = ShellResult.outcome(result) + if (outcome.status !== "timeout") return outcome + return { + ...outcome, + output: `${outcome.output}\n\nCommand exceeded timeout of ${finalTimeout} ms. Retry with a larger timeout if the command is expected to take longer.`, + } + }), Effect.onInterrupt(() => shell.remove(info.id).pipe(Effect.ignore)), ) const job = yield* jobs.start({ @@ -245,18 +230,17 @@ export const Plugin = { type: name, title: info.command, metadata: { sessionID: context.sessionID, shellID: info.id }, - recovery: { - kind: "shell", - sessionID: context.sessionID, - shellID: info.id, - command: info.command, - }, + recovery, run, }) + // Once the job owns the shell, interruption anywhere before block/background cancels through it. + yield* context + .progress({ shellID: info.id }) + .pipe(Effect.onInterrupt(() => jobs.cancel(job.id).pipe(Effect.ignore))) if (input.background === true) { yield* jobs.background(job.id) - yield* notifyWhenDone(context.sessionID, job.id, info.id, info.command, settled) + yield* notifyWhenDone(job.id) return backgroundResult(info.id, info.file) } @@ -264,17 +248,17 @@ export const Plugin = { .block({ id: job.id, sessionID: context.sessionID }) .pipe(Effect.onInterrupt(() => jobs.cancel(job.id).pipe(Effect.ignore))) if (result?.type === "backgrounded") { - yield* shell.timeout(info.id, 0) - yield* notifyWhenDone(context.sessionID, job.id, info.id, info.command, settled) + yield* shell.timeout(info.id, 0).pipe(Effect.ignore) + yield* notifyWhenDone(job.id) return backgroundResult(info.id, info.file) } if (result?.info.status === "error") return yield* Effect.fail(new Error(result.info.error ?? "Command failed")) - if (result?.info.status === "cancelled") return yield* Effect.fail(new Error("Command cancelled")) - - return yield* Deferred.await(settled) + if (result?.info.result?.kind !== "shell") return yield* Effect.fail(new Error("Command cancelled")) + if (result.info.result.status === "unavailable") + return yield* Effect.fail(new Error(ShellResult.unavailable.output)) + return completedResult(result.info.result) }).pipe( - Effect.map(toolResult), Effect.mapError( (error) => new ToolFailure({ message: `Unable to execute command: ${input.command}`, error }), ), diff --git a/packages/core/src/tool/plugin/subagent.ts b/packages/core/src/tool/plugin/subagent.ts index c12eaeccbe3e..a0fc91a1f75b 100644 --- a/packages/core/src/tool/plugin/subagent.ts +++ b/packages/core/src/tool/plugin/subagent.ts @@ -9,8 +9,8 @@ import { Job } from "../../job.js" import { Permission } from "../../permission.js" import { Session } from "../../session.js" import { SessionSchema } from "../../session/schema.js" -import { SubagentCompletion } from "../../session/subagent-completion.js" import { SubagentJob } from "../../session/subagent-job.js" +import { SubagentOutcome } from "../../session/subagent-outcome.js" export const name = "subagent" @@ -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", "stopped"]), output: Schema.String, }) export const description = [ @@ -166,8 +166,8 @@ export const Plugin = { const background = input.background === true yield* context.progress({ sessionID: child.id, status: "running" }) - // Standard prompt admission outside the job: Job.start joining a running child skips - // its run effect, and the default wake starts an idle child or steers a running one. + // A new child starts only through its Job. Existing children still need the prompt's + // wake: joining an active Job skips run, but new input must still steer the child. yield* sessions .prompt({ sessionID: child.id, @@ -175,7 +175,7 @@ export const Plugin = { existing === undefined ? ["You are a subagent spawned by another session.", input.prompt].join("\n") : input.prompt, - ...(background && existing === undefined ? { resume: false } : {}), + ...(existing === undefined ? { resume: false } : {}), }) .pipe( Effect.mapError( @@ -193,7 +193,7 @@ export const Plugin = { yield* subagents.start(recovery) if (background) { - yield* subagents.background(recovery) + yield* subagents.background(child.id) return backgroundResult(child.id) } @@ -205,7 +205,7 @@ export const Plugin = { ), ) if (result?.type === "backgrounded") { - yield* subagents.notify(recovery, result.info.started_at) + yield* subagents.notify(result.info) return backgroundResult(child.id) } // Failure surfaces keep the sessionID visible so the model can continue the child. @@ -213,20 +213,20 @@ export const Plugin = { return yield* new ToolFailure({ message: `Subagent failed (sessionID: ${child.id}): ${result.info.error ?? "unknown error"}`, }) - if (result?.info.status === "cancelled") + const outcome = result?.info.result?.kind === "subagent" ? result.info.result : undefined + if (outcome === undefined) return yield* new ToolFailure({ message: `Subagent cancelled (sessionID: ${child.id})` }) - return { - sessionID: child.id, - status: "completed" as const, - output: result?.info.output ?? SubagentCompletion.NO_TEXT, - } + // A user stop is a successful answer: the work ended and the parent should not redo it. + if (outcome.status === "interrupted") + return { sessionID: child.id, status: "stopped" as const, output: SubagentOutcome.stopped } + return { sessionID: child.id, status: "completed" as const, output: outcome.text } }).pipe( Effect.map((output) => ({ output, content: - output.status === "completed" - ? `\n${output.output}\n` - : output.output, + output.status === "running" + ? output.output + : `\n${output.output}\n`, metadata: { sessionID: output.sessionID, status: output.status }, })), ), diff --git a/packages/core/test/job.test.ts b/packages/core/test/job.test.ts index 62302facd2ed..87abc36bc445 100644 --- a/packages/core/test/job.test.ts +++ b/packages/core/test/job.test.ts @@ -9,6 +9,15 @@ import { testEffect } from "./lib/effect" const it = testEffect(AppNodeBuilder.build(LayerNode.group([Job.node, KV.node]))) +// Jobs transport the producer's typed outcome; these tests only need a distinguishable one. +const outcome = (output: string): Job.Outcome => ({ + kind: "shell", + status: "exited", + exit: 0, + output, + truncated: false, +}) + describe("Job", () => { it.live("tracks process-local work through explicit observation", () => Effect.gen(function* () { @@ -17,7 +26,7 @@ describe("Job", () => { const job = yield* jobs.start({ type: "test", metadata: { durable: false }, - run: Deferred.await(latch).pipe(Effect.as("done")), + run: Deferred.await(latch).pipe(Effect.as(outcome("done"))), }) expect(job).toMatchObject({ type: "test", status: "running", metadata: { durable: false } }) @@ -29,7 +38,7 @@ describe("Job", () => { yield* Deferred.succeed(latch, undefined) expect(yield* jobs.wait({ id: job.id })).toMatchObject({ timedOut: false, - info: { status: "completed", output: "done" }, + info: { status: "completed", result: { output: "done" } }, }) }), ) @@ -49,7 +58,7 @@ describe("Job", () => { .pipe( Effect.flatMap((info) => info?.status === "running" - ? Effect.succeed(`done-${index}`) + ? Effect.succeed(outcome(`done-${index}`)) : Effect.fail("job started before publish"), ), ), @@ -57,7 +66,7 @@ describe("Job", () => { expect(yield* jobs.wait({ id: job.id })).toMatchObject({ timedOut: false, - info: { status: "completed", output: `done-${index}` }, + info: { status: "completed", result: { output: `done-${index}` } }, }) }) }) @@ -67,18 +76,18 @@ describe("Job", () => { it.live("reuses running work when started again with the same ID", () => Effect.gen(function* () { const jobs = yield* Job.Service - const output = yield* Deferred.make() + const output = yield* Deferred.make() const job = yield* jobs.start({ id: "job_reused", type: "test", run: Deferred.await(output) }) expect( yield* jobs.start({ id: job.id, type: "duplicate", run: Effect.die("Duplicate work must not run") }), ).toEqual(job) - yield* Deferred.succeed(output, "original output") + yield* Deferred.succeed(output, outcome("original output")) expect((yield* jobs.wait({ id: job.id })).info).toMatchObject({ type: "test", status: "completed", - output: "original output", + result: { output: "original output" }, }) }), ) @@ -87,15 +96,15 @@ describe("Job", () => { Effect.gen(function* () { const jobs = yield* Job.Service const callback = yield* Deferred.make<() => void>() - const output = yield* Deferred.make() + const output = yield* Deferred.make() const finalized = yield* Deferred.make() const job = yield* jobs.start({ id: "job_replaced", type: "test", - run: Effect.callback((resume) => { + run: Effect.callback((resume) => { Deferred.doneUnsafe( callback, - Effect.succeed(() => resume(Effect.succeed("obsolete output"))), + Effect.succeed(() => resume(Effect.succeed(outcome("obsolete output")))), ) }), }) @@ -119,11 +128,11 @@ describe("Job", () => { expect(yield* jobs.get(job.id)).toMatchObject({ type: "replacement", status: "running" }) expect(yield* Deferred.isDone(finalized)).toBe(false) - yield* Deferred.succeed(output, "replacement output") + yield* Deferred.succeed(output, outcome("replacement output")) expect((yield* jobs.wait({ id: job.id })).info).toMatchObject({ type: "replacement", status: "completed", - output: "replacement output", + result: { output: "replacement output" }, }) expect(yield* Deferred.isDone(finalized)).toBe(true) }), @@ -133,7 +142,7 @@ describe("Job", () => { Effect.gen(function* () { const jobs = yield* Job.Service const latch = yield* Deferred.make() - const job = yield* jobs.start({ type: "test", run: Deferred.await(latch).pipe(Effect.as("done")) }) + const job = yield* jobs.start({ type: "test", run: Deferred.await(latch).pipe(Effect.as(outcome("done"))) }) const waiting = yield* jobs .block({ id: job.id, sessionID: SessionSchema.ID.make("ses_parent") }) .pipe(Effect.forkIn(yield* Scope.Scope, { startImmediately: true })) @@ -142,7 +151,7 @@ describe("Job", () => { expect(yield* Fiber.join(waiting)).toMatchObject({ type: "finished", - info: { status: "completed", output: "done" }, + info: { status: "completed", result: { output: "done" } }, }) expect(yield* jobs.background(job.id)).toBeUndefined() }), @@ -152,7 +161,7 @@ describe("Job", () => { Effect.gen(function* () { const jobs = yield* Job.Service const latch = yield* Deferred.make() - const job = yield* jobs.start({ type: "test", run: Deferred.await(latch).pipe(Effect.as("done")) }) + const job = yield* jobs.start({ type: "test", run: Deferred.await(latch).pipe(Effect.as(outcome("done"))) }) const waiting = yield* jobs .block({ id: job.id, sessionID: SessionSchema.ID.make("ses_parent") }) .pipe(Effect.forkIn(yield* Scope.Scope, { startImmediately: true })) @@ -166,7 +175,7 @@ describe("Job", () => { yield* Deferred.succeed(latch, undefined) expect(yield* jobs.wait({ id: job.id })).toMatchObject({ timedOut: false, - info: { status: "completed", output: "done" }, + info: { status: "completed", result: { output: "done" } }, }) }), ) @@ -180,17 +189,17 @@ describe("Job", () => { const first = yield* jobs.start({ id: "job_first", type: "test", - run: Deferred.await(latch).pipe(Effect.as("first")), + run: Deferred.await(latch).pipe(Effect.as(outcome("first"))), }) const second = yield* jobs.start({ id: "job_second", type: "test", - run: Deferred.await(latch).pipe(Effect.as("second")), + run: Deferred.await(latch).pipe(Effect.as(outcome("second"))), }) const third = yield* jobs.start({ id: "job_third", type: "other", - run: Deferred.await(latch).pipe(Effect.as("third")), + run: Deferred.await(latch).pipe(Effect.as(outcome("third"))), }) const scope = yield* Scope.Scope const firstWait = yield* jobs @@ -222,7 +231,11 @@ describe("Job", () => { shellID: "shell_background", command: "echo done", } - const job = yield* jobs.start({ type: "shell", recovery, run: Deferred.await(latch).pipe(Effect.as("done")) }) + const job = yield* jobs.start({ + type: "shell", + recovery, + run: Deferred.await(latch).pipe(Effect.as(outcome("done"))), + }) expect((yield* jobs.pendingBackground).find((item) => item.id === job.id)).toBeUndefined() const background = yield* jobs.background(job.id) @@ -241,7 +254,7 @@ describe("Job", () => { notificationID: running?.notificationID, recovery, status: "completed", - output: "done", + result: { output: "done" }, }) if (!completed) return yield* Effect.die("background marker missing") @@ -262,7 +275,11 @@ describe("Job", () => { agent: "explore", description: "Explore background recovery", } - const job = yield* jobs.start({ type: "subagent", recovery, run: Deferred.await(latch).pipe(Effect.as("done")) }) + const job = yield* jobs.start({ + type: "subagent", + recovery, + run: Deferred.await(latch).pipe(Effect.as(outcome("done"))), + }) const waiting = yield* jobs .block({ id: job.id, sessionID: parentSessionID }) .pipe(Effect.forkIn(yield* Scope.Scope, { startImmediately: true })) diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index 3d34e97f2c08..58c40a41b169 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -56,6 +56,11 @@ describe("SessionExecution lifecycle", () => { }) }) + test("does not classify interruption mixed with a defect as a user stop", () => { + const exit = Effect.runSyncExit(Effect.interrupt.pipe(Effect.ensuring(Effect.die(new Error("cleanup failed"))))) + expect(SessionExecution.terminal(exit, "user")).toMatchObject({ type: "failed" }) + }) + it.effect("the sweep only lists claimed top-level Sessions", () => Effect.gen(function* () { const database = yield* Database.Service @@ -160,7 +165,7 @@ describe("SessionExecution lifecycle", () => { }), ) - it.effect("does not resume a user-cancelled background child whose notification was not admitted", () => + it.effect("records a user-stopped background child quietly instead of resuming it", () => Effect.gen(function* () { const database = yield* Database.Service const parent = Session.ID.make("ses_cancelled_background_parent") @@ -189,14 +194,24 @@ describe("SessionExecution lifecycle", () => { agent: "general", description: "Cancelled inspection", }, - run: execution.resume(child).pipe(Effect.as("unused")), + run: execution + .resume(child) + .pipe( + Effect.map( + (ended): Job.Outcome => + ended.type === "interrupted" + ? { kind: "subagent", status: "interrupted" } + : { kind: "subagent", status: "completed", text: "unused" }, + ), + ), }) yield* jobs.background(child) yield* Deferred.await(running) 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" }]) + const stopped = { kind: "subagent", status: "interrupted" } + expect((yield* jobs.wait({ id: child })).info).toMatchObject({ status: "completed", result: stopped }) + expect(yield* jobs.pendingBackground).toMatchObject([{ id: child, status: "completed", result: stopped }]) expect((yield* claims(database))[child]).toBe(false) yield* Scope.close(scope, Exit.void) @@ -212,9 +227,10 @@ describe("SessionExecution lifecycle", () => { ) yield* Context.get(restarted, SessionRestart.Service).resumeSuspendedSessions yield* Context.get(restarted, SessionExecution.Service).awaitIdle(parent) - expect(drained).toEqual([parent]) + // The notice is recorded for the parent's next step without waking it. + 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: "stopped" } } }, ]) expect(yield* restartedJobs.pendingBackground).toEqual([]) }), @@ -480,7 +496,7 @@ describe("SessionRestart background recovery", () => { const jobs = yield* Job.Service const sessionID = Session.ID.make("ses_background_completed_shell") yield* seedSessions(database, [sessionID]) - const complete = yield* Deferred.make() + const complete = yield* Deferred.make() yield* jobs.start({ id: "call-completed-shell", type: "shell", @@ -493,7 +509,13 @@ describe("SessionRestart background recovery", () => { run: Deferred.await(complete), }) yield* jobs.background("call-completed-shell") - yield* Deferred.succeed(complete, "(no output)\n\nCommand exited with code 7.") + yield* Deferred.succeed(complete, { + kind: "shell", + status: "exited", + exit: 7, + output: "(no output)", + truncated: false, + }) yield* jobs.wait({ id: "call-completed-shell" }) const scope = yield* Scope.make() @@ -524,11 +546,105 @@ describe("SessionRestart background recovery", () => { jobID: "call-completed-shell", shellID: "sh_completed", state: "completed", + truncated: false, + exit: 7, + }) + expect(yield* restarted.pendingBackground).toEqual([]) + }), + ) + + it.effect("recovers a user-stopped shell as a quiet notice 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]) + const stopped = yield* Deferred.make() + yield* jobs.start({ + id: "sh_user_stopped", + type: "shell", + recovery: { kind: "shell", sessionID, shellID: "sh_user_stopped", command: "sleep 60" }, + run: Deferred.await(stopped), }) + yield* jobs.background("sh_user_stopped") + yield* Deferred.succeed(stopped, { kind: "shell", status: "killed", output: "partial", truncated: false }) + yield* jobs.wait({ id: "sh_user_stopped" }) + + 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: '\npartial\n\nCommand stopped by user. Do not restart it unless the user asks.\n', + metadata: { source: "shell", shellID: "sh_user_stopped", state: "stopped", truncated: false }, + }, + }, + ]) expect(yield* restarted.pendingBackground).toEqual([]) }), ) + for (const kind of ["shell", "subagent"] as const) { + it.effect(`replays a pre-outcome ${kind} completion without losing its saved text`, () => + Effect.gen(function* () { + const database = yield* Database.Service + const kv = yield* KV.Service + const parent = Session.ID.make(`ses_legacy_${kind}_parent`) + const child = Session.ID.make(`ses_legacy_${kind}_child`) + const notificationID = SessionMessage.ID.make(`msg_legacy_${kind}`) + yield* seedSessions(database, [parent]) + yield* seedSessions(database, [child], { parent_id: parent }) + // Literal old writer format: no typed result and no inferred exit/capture metadata. + const output = kind === "shell" ? "saved output\n\nCommand exited with code 7." : "The saved child answer." + yield* kv.set(`job.background/${notificationID}`, { + id: kind === "shell" ? "sh_legacy" : child, + notificationID, + recovery: + kind === "shell" + ? { kind, sessionID: parent, shellID: "sh_legacy", command: "exit 7" } + : { kind, parentSessionID: parent, childSessionID: child, agent: "reviewer", description: "Review" }, + status: "completed", + output, + }) + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) + const jobs = yield* Job.make.pipe(Scope.provide(scope)) + const drained: Session.ID[] = [] + const context = yield* buildExecution( + scope, + ({ sessionID }) => Effect.sync(() => void drained.push(sessionID)), + undefined, + jobs, + ) + yield* Context.get(context, SessionRestart.Service).resumeSuspendedSessions + yield* Context.get(context, SessionExecution.Service).awaitIdle(parent) + const inbox = yield* SessionInbox.list(database.db, parent) + expect(drained).toEqual([parent]) + expect(inbox).toMatchObject([ + { + id: notificationID, + type: "synthetic", + payload: { text: expect.stringContaining(output), metadata: { state: "completed" } }, + }, + ]) + expect(inbox[0]).not.toHaveProperty("payload.metadata.exit") + expect(inbox[0]).not.toHaveProperty("payload.metadata.truncated") + expect(yield* jobs.pendingBackground).toEqual([]) + }), + ) + } + for (const delivered of [false, true]) { it.effect(`does not duplicate a shell notification already ${delivered ? "delivered" : "admitted"}`, () => Effect.gen(function* () { @@ -863,7 +979,7 @@ describe("SessionRestart background recovery", () => { const child = Session.ID.make("ses_subagent_completed_child") yield* seedSessions(database, [parent]) yield* seedSessions(database, [child], { parent_id: parent }) - const complete = yield* Deferred.make() + const complete = yield* Deferred.make() yield* jobs.start({ id: child, type: "subagent", @@ -877,7 +993,7 @@ describe("SessionRestart background recovery", () => { run: Deferred.await(complete), }) yield* jobs.background(child) - yield* Deferred.succeed(complete, "Recovered result") + yield* Deferred.succeed(complete, { kind: "subagent", status: "completed", text: "Recovered result" }) yield* jobs.wait({ id: child }) const parentWoken = yield* Deferred.make() @@ -925,7 +1041,7 @@ describe("SessionRestart background recovery", () => { agent: "explore", description: "Completed inspection", }, - run: Effect.succeed("Recovered result"), + run: Effect.succeed({ kind: "subagent", status: "completed", text: "Recovered result" }), }) yield* jobs.wait({ id: child }) yield* jobs.background(child) @@ -961,7 +1077,7 @@ describe("SessionRestart background recovery", () => { ] yield* seedSessions(database, [parent], { time_suspended: Date.now(), resume_attempts: resumeAttempts }) yield* seedSessions(database, children, { parent_id: parent }) - const complete = yield* Deferred.make() + const complete = yield* Deferred.make() for (const child of children) { yield* jobs.start({ id: child, @@ -977,7 +1093,7 @@ describe("SessionRestart background recovery", () => { }) yield* jobs.background(child) } - yield* Deferred.succeed(complete, "Recovered result") + yield* Deferred.succeed(complete, { kind: "subagent", status: "completed", text: "Recovered result" }) yield* Effect.forEach(children, (id) => jobs.wait({ id }), { discard: true }) const draining = yield* Deferred.make() @@ -1340,8 +1456,10 @@ function buildExecution( Session.Service, Effect.gen(function* () { const execution = yield* SessionExecution.Service + // Route execution through this fresh coordinator, not the outer harness's. return Session.Service.of({ ...sessions, + resume: execution.resume, synthetic: (input) => sessions .synthetic({ ...input, resume: false }) diff --git a/packages/core/test/session-message-update.test.ts b/packages/core/test/session-message-update.test.ts index 25076956662e..6af4211eae7a 100644 --- a/packages/core/test/session-message-update.test.ts +++ b/packages/core/test/session-message-update.test.ts @@ -38,7 +38,7 @@ const it = testEffect( SessionExecution.Service.of({ active: Effect.sync(() => active), isActive: (sessionID) => Effect.sync(() => active.has(sessionID)), - resume: () => Effect.void, + resume: () => Effect.succeed({ type: "succeeded" as const }), wake: () => Effect.void, interrupt: () => Effect.succeed(false), awaitIdle: () => Effect.void, diff --git a/packages/core/test/session-owned.test.ts b/packages/core/test/session-owned.test.ts index 64ec38edb356..1e2ff0393337 100644 --- a/packages/core/test/session-owned.test.ts +++ b/packages/core/test/session-owned.test.ts @@ -102,6 +102,7 @@ const setup = Effect.fnUntraced(function* (options?: { resume: (id) => Effect.sync(() => { resumes.push(id) + return { type: "succeeded" as const } }), awaitIdle: () => Effect.void, interrupt: () => Effect.succeed(false), @@ -449,6 +450,7 @@ describe("Session-owned handles", () => { Effect.gen(function* () { calls.push(`resume:${id}`) yield* Effect.never + return { type: "succeeded" as const } }).pipe( Effect.onInterrupt(() => Effect.sync(() => { @@ -523,7 +525,7 @@ describe("Session-owned handles", () => { Deferred.succeed(blocked, undefined).pipe( Effect.andThen(Deferred.await(release)), Effect.as({ - info: Info.make({ ...started, status: "exited", exit: 0, time: { started: 0, completed: 1 } }), + info: { ...started, status: "exited" as const, exit: 0, time: { started: 0, completed: 1 } }, capture: { output: "owned", truncated: false }, }), ), @@ -634,6 +636,7 @@ describe("Session-owned handles", () => { resumes.push(id) if (resumes.length === 2) yield* Deferred.succeed(joining, undefined) yield* coordinator.run(id) + return { type: "succeeded" as const } }), wake: coordinator.wake, awaitIdle: coordinator.awaitIdle, diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index ba6784e40f4f..ec95a6644e3f 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -47,6 +47,7 @@ const execution = Layer.succeed( resume: (sessionID) => Effect.sync(() => { executionCalls.push(sessionID) + return { type: "succeeded" as const } }), interrupt: (sessionID, options) => Effect.sync(() => { diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index e04ea8e16bdd..86792f60132a 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -139,6 +139,34 @@ describe("SessionRunCoordinator", () => { }), ) + for (const where of ["drain", "settlement"]) { + it.effect(`preserves ${where} defects combined with user interruption`, () => + Effect.gen(function* () { + const started = yield* Deferred.make() + const defect = new Error(`${where} failed`) + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.never), + Effect.ensuring(where === "drain" ? Effect.die(defect) : Effect.void), + ), + settled: () => (where === "settlement" ? Effect.die(defect) : Effect.void), + }) + const joining = yield* coordinator.run("session").pipe(Effect.forkChild) + yield* Deferred.await(started) + yield* coordinator.interrupt("session", "user") + const exit = yield* Fiber.await(joining) + expect(Exit.isFailure(exit)).toBeTrue() + if (Exit.isFailure(exit)) { + expect(Cause.hasInterrupts(exit.cause)).toBeTrue() + expect(Cause.hasDies(exit.cause)).toBeTrue() + expect(Cause.pretty(exit.cause)).toContain(defect.message) + } + expect(yield* coordinator.active).toEqual(new Set()) + }), + ) + } + it.effect("cleans active executions when its scope closes", () => Effect.gen(function* () { const started = yield* Deferred.make() @@ -306,7 +334,11 @@ describe("SessionRunCoordinator", () => { yield* Deferred.await(interrupted) const exits = yield* Fiber.awaitAll([first, second, idle]) - expect(exits.slice(0, 2).every((exit) => Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause))).toBeTrue() + // Joiners learn how the execution ended; they are not interrupted themselves. + expect(exits.slice(0, 2)).toEqual([ + Exit.succeed({ type: "interrupted", reason: "user" }), + Exit.succeed({ type: "interrupted", reason: "user" }), + ]) expect(exits.slice(2).every(Exit.isSuccess)).toBeTrue() expect(Array.from(yield* coordinator.active)).toEqual([]) expect(runs).toBe(1) diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index 08a64ae014c2..5fc46bf46b7c 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -126,7 +126,14 @@ const execution = (llmClient: Layer.Layer) => return SessionExecution.Service.of({ active: coordinator.active, isActive: coordinator.isActive, - resume: coordinator.run, + resume: (id) => + coordinator + .run(id) + .pipe( + Effect.map((ended) => + ended.type === "succeeded" ? ended : { type: "interrupted" as const, reason: "user" as const }, + ), + ), wake: coordinator.wake, interrupt: (sessionID) => coordinator.interrupt(sessionID), awaitIdle: coordinator.awaitIdle, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index f8bc581d617a..641cd1845fe3 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -441,7 +441,14 @@ const layer = Layer.unwrap( return SessionExecution.Service.of({ active: coordinator.active, isActive: coordinator.isActive, - resume: coordinator.run, + resume: (id) => + coordinator + .run(id) + .pipe( + Effect.map((ended) => + ended.type === "succeeded" ? ended : { type: "interrupted" as const, reason: "user" as const }, + ), + ), wake: coordinator.wake, interrupt: (sessionID) => coordinator.interrupt(sessionID), awaitIdle: coordinator.awaitIdle, @@ -486,6 +493,8 @@ const layer = Layer.unwrap( ).pipe(Layer.provideMerge(Layer.sync(RunnerState, makeRunnerState)), Layer.provideMerge(testLLM)) const it = testEffect(layer) const sessionID = Session.ID.make("ses_runner_test") +// A joined execution reports the user's interruption as its outcome instead of interrupting the joiner. +const interruptedByUser = Exit.succeed({ type: "interrupted", reason: "user" }) const otherSessionID = Session.ID.make("ses_runner_other") const insertSession = (id: Session.ID) => @@ -1237,7 +1246,7 @@ describe("SessionRunnerLLM", () => { yield* s.admit("Second") const exit = yield* s.resume.pipe(Effect.exit) - expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true) + expect(exit).toEqual(interruptedByUser) expect(s.requests).toHaveLength(1) expect(yield* SessionInbox.has(s.db, sessionID, "steer")).toBe(true) }, @@ -2642,7 +2651,7 @@ describe("SessionRunnerLLM", () => { yield* s.session.interrupt(sessionID) const exit = yield* Fiber.await(run) - expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue() + expect(exit).toEqual(interruptedByUser) expect(yield* s.context).toContainEqual( expect.objectContaining({ type: "compaction", @@ -3268,7 +3277,7 @@ describe("SessionRunnerLLM", () => { delivery: "queue", }) yield* s.session.interrupt(sessionID) - expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + expect(yield* Fiber.await(run)).toEqual(interruptedByUser) expect(s.requests).toHaveLength(1) expect(yield* SessionInbox.has(s.db, sessionID, "queue")).toBe(true) const resumed = yield* s.resume.pipe(Effect.forkChild) @@ -3294,7 +3303,7 @@ describe("SessionRunnerLLM", () => { text: "Steer after interrupt", }) yield* s.session.interrupt(sessionID) - expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + expect(yield* Fiber.await(run)).toEqual(interruptedByUser) expect(s.requests).toHaveLength(1) expect(yield* SessionInbox.has(s.db, sessionID, "steer")).toBe(true) @@ -3914,8 +3923,7 @@ describe("SessionRunnerLLM", () => { const exit = yield* s.resume.pipe(Effect.exit) - expect(exit._tag).toBe("Failure") - if (exit._tag === "Failure") expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true) + expect(exit).toEqual(interruptedByUser) expect(s.requests).toHaveLength(1) expect(yield* s.context).toMatchObject([ Expected.user("Call declined"), @@ -4015,8 +4023,7 @@ describe("SessionRunnerLLM", () => { const run = yield* s.resume.pipe(Effect.exit, Effect.forkChild) const exit = yield* Fiber.join(run) - expect(exit._tag).toBe("Failure") - if (exit._tag === "Failure") expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true) + expect(exit).toEqual(interruptedByUser) expect(s.requests).toHaveLength(1) expect(yield* s.context).toMatchObject([ Expected.user("Ask then stop"), @@ -4077,7 +4084,7 @@ describe("SessionRunnerLLM", () => { yield* tools.started yield* s.session.interrupt(sessionID) - expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + expect(yield* Fiber.await(run)).toEqual(interruptedByUser) yield* s.session.interrupt(sessionID) const context = yield* s.context expect(context).toMatchObject([ @@ -4118,7 +4125,7 @@ describe("SessionRunnerLLM", () => { yield* s.session.interrupt(sessionID) const exit = yield* Fiber.await(run) - expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue() + expect(exit).toEqual(interruptedByUser) expect(s.requests).toHaveLength(1) expect(yield* s.context).toMatchObject([ Expected.user("Interrupt provider"), @@ -4405,7 +4412,7 @@ describe("SessionRunnerLLM", () => { yield* Queue.take(scheduled) yield* s.session.interrupt(sessionID) const exit = yield* Fiber.await(run) - expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true) + expect(exit).toEqual(interruptedByUser) yield* TestClock.adjust("1 minute") expect(s.requests).toHaveLength(1) const events = yield* recordedEventTypes(sessionID) @@ -5140,7 +5147,7 @@ describe("SessionRunnerLLM", () => { yield* Effect.yieldNow yield* s.session.interrupt(sessionID) - expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + expect(yield* Fiber.await(run)).toEqual(interruptedByUser) expect(s.requests).toHaveLength(1) expect(yield* s.context).toMatchObject([ Expected.user("Interrupt malformed recovery"), diff --git a/packages/core/test/session-shell.test.ts b/packages/core/test/session-shell.test.ts index ecff6a276936..ae47e2cb624b 100644 --- a/packages/core/test/session-shell.test.ts +++ b/packages/core/test/session-shell.test.ts @@ -12,6 +12,8 @@ import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator" import { Shell } from "@opencode-ai/core/shell" +import { ShellResult } from "@opencode-ai/core/shell/result" +import { ID } from "@opencode-ai/schema/shell" import { LayerNode } from "@opencode-ai/util/effect/layer-node" import { location } from "./fixture/location" import { offlineModels } from "./fixture/models" @@ -47,7 +49,14 @@ const executionLayer = Layer.effect( return SessionExecution.Service.of({ active: coordinator.active, isActive: coordinator.isActive, - resume: coordinator.run, + resume: (id) => + coordinator + .run(id) + .pipe( + Effect.map((ended) => + ended.type === "succeeded" ? ended : { type: "interrupted" as const, reason: "user" as const }, + ), + ), interrupt: (sessionID) => coordinator.interrupt(sessionID), awaitIdle: coordinator.awaitIdle, wake: (sessionID) => @@ -277,22 +286,38 @@ describe("Session.shell", () => { ) } + // Stop preserves the terminal and capture; removal only tells us that the result is unavailable. for (const outcome of [ { status: "killed", - state: "cancelled", - text: "Command cancelled", + via: "stop", + end: (shell: Shell.Interface, id: ID) => shell.stop(id), + state: "stopped", + text: "Command stopped by user. Do not restart it unless the user asks.", + output: "killed started", + }, + { + status: "unavailable", + via: "remove", + end: (shell: Shell.Interface, id: ID) => shell.remove(id), + state: "error", + text: "Shell command output is no longer available.", output: "Shell command output is no longer available.", }, - { status: "timeout", state: "completed", text: "Command timed out", output: "timeout started" }, + { + status: "timeout", + via: "timeout", + end: (shell: Shell.Interface, id: ID) => shell.timeout(id, 1), + state: "completed", + text: "Command timed out", + output: "timeout started", + }, ]) { - it.live(`records a ${outcome.status} shell and admits its completion without waking the model`, () => + it.live(`records a ${outcome.status} shell via ${outcome.via} without waking the model`, () => Effect.gen(function* () { const fixture = yield* setup const command = yield* launch(fixture, outcome.status) - yield* outcome.status === "killed" - ? fixture.shell.remove(command.shellID) - : fixture.shell.timeout(command.shellID, 1) + yield* outcome.end(fixture.shell, command.shellID) yield* Fiber.join(command.caller).pipe(Effect.timeout("5 seconds")) expect(yield* fixture.session.messages({ sessionID: fixture.created.id })).toMatchObject([ { @@ -320,6 +345,74 @@ describe("Session.shell", () => { ) } + it.live("reports a stop that lands before anyone waits, with its output still readable", () => + Effect.gen(function* () { + const fixture = yield* setup + const started = yield* fixture.shell.create({ + command: process.platform === "win32" ? "Write-Output early; Start-Sleep -Seconds 60" : "echo early; sleep 60", + timeout: 0, + }) + yield* fixture.shell + .output(started.id) + .pipe(Effect.repeat({ until: (page) => page.size > 0, schedule: Schedule.spaced("10 millis") })) + expect(yield* fixture.shell.stop(started.id)).toMatchObject({ id: started.id, status: "killed" }) + // A second stop is a no-op on an already terminal command. + expect(yield* fixture.shell.stop(started.id)).toMatchObject({ status: "killed" }) + expect(yield* fixture.shell.list()).toEqual([]) + expect(yield* fixture.shell.result(started)).toMatchObject({ + info: { id: started.id, status: "killed" }, + capture: { output: expect.stringContaining("early"), truncated: false }, + }) + expect(ShellResult.outcome(yield* fixture.shell.result(started))).toMatchObject({ + kind: "shell", + status: "killed", + output: expect.stringContaining("early"), + }) + }), + ) + + const posix = process.platform === "win32" ? it.live.skip : it.live + posix("finishes an accepted stop despite caller interruption and joins overlapping stops", () => + Effect.gen(function* () { + const fixture = yield* setup + const started = yield* fixture.shell.create({ + shell: "/bin/sh", + command: 'trap "" TERM; printf ready; while :; do sleep 60; done', + timeout: 0, + }) + const pid = started.pid + if (pid === undefined) return yield* Effect.die("Expected shell process ID") + yield* Effect.addFinalizer(() => + Effect.try(() => process.kill(-pid, "SIGKILL")).pipe( + Effect.ignore, + Effect.andThen(fixture.shell.remove(started.id).pipe(Effect.ignore)), + ), + ) + yield* fixture.shell + .output(started.id) + .pipe( + Effect.repeat({ until: (page) => page.size > 0, schedule: Schedule.spaced("5 millis") }), + Effect.timeout("5 seconds"), + ) + const first = yield* fixture.shell.stop(started.id).pipe(Effect.forkScoped) + yield* fixture.shell + .get(started.id) + .pipe( + Effect.repeat({ until: (info) => info.status === "killed", schedule: Schedule.spaced("1 millis") }), + Effect.timeout("5 seconds"), + ) + const second = yield* fixture.shell.stop(started.id).pipe(Effect.forkScoped({ startImmediately: true })) + // SIGTERM is ignored, so both stops must wait for escalation and capture completion. + expect(second.pollUnsafe()).toBeUndefined() + yield* Fiber.interrupt(first) + const terminal = yield* fixture.shell.wait(started.id).pipe(Effect.timeout("5 seconds")) + expect(terminal.status).toBe("killed") + expect(yield* Fiber.join(second)).toEqual(terminal) + expect(yield* fixture.shell.result(started)).toMatchObject({ capture: { output: "ready", truncated: false } }) + expect(yield* Effect.try(() => process.kill(pid, 0)).pipe(Effect.result)).toMatchObject({ _tag: "Failure" }) + }), + ) + it.live("admits a spawn failure without waking the model before failing the caller", () => Effect.gen(function* () { const fixture = yield* setup diff --git a/packages/core/test/shell-retention.test.ts b/packages/core/test/shell-retention.test.ts index 33bddfa9cc45..dfab687594e1 100644 --- a/packages/core/test/shell-retention.test.ts +++ b/packages/core/test/shell-retention.test.ts @@ -7,6 +7,7 @@ import { Config } from "@opencode-ai/core/config" import { Environment } from "@opencode-ai/core/environment/index" import { Location } from "@opencode-ai/core/location" import { Shell } from "@opencode-ai/core/shell" +import { ShellResult } from "@opencode-ai/core/shell/result" import { Global } from "@opencode-ai/util/global" import { hostEnvironmentLayer } from "./fixture/environment" import { tempGlobalLayer } from "./fixture/global" @@ -66,7 +67,7 @@ it.live("eviction makes progress past an already-removed shell", () => const removed = yield* shell.create({ shell: "sh", command: "removed", timeout: 0 }) const finishRemoved = yield* Queue.take(completions) yield* shell.remove(removed.id) - expect((yield* shell.result(removed)).capture).toBeUndefined() + expect(yield* shell.result(removed)).toMatchObject({ info: { status: "unavailable" }, capture: undefined }) yield* finishRemoved const complete = Effect.gen(function* () { @@ -79,6 +80,12 @@ it.live("eviction makes progress past an already-removed shell", () => // Exceed the 25-entry retention cap with the removed ID at the head of exitOrder. yield* Effect.forEach(Array.from({ length: 25 }), () => complete, { discard: true }) expect(yield* shell.get(oldest.id).pipe(Effect.flip)).toBeInstanceOf(Shell.NotFoundError) + const expired = yield* shell.result(oldest) + expect(expired.info.status).toBe("unavailable") + expect(expired.info.time.completed).toBeUndefined() + expect(expired.capture).toBeUndefined() + expect(ShellResult.userNotification(expired)).toMatchObject({ metadata: { state: "error" } }) + expect(ShellResult.userNotification(expired).text).not.toContain("stopped by user") const survivor = yield* complete expect(yield* shell.result(survivor)).toMatchObject({ diff --git a/packages/core/test/tool-shell.test.ts b/packages/core/test/tool-shell.test.ts index 85f368ae79e2..a999fed275d0 100644 --- a/packages/core/test/tool-shell.test.ts +++ b/packages/core/test/tool-shell.test.ts @@ -3,7 +3,7 @@ import { realpathSync, watch } from "node:fs" import os from "os" import path from "path" import { describe, expect } from "bun:test" -import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Queue, Scope, Stream } from "effect" +import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Queue, Schedule, Scope, Stream } from "effect" import { Money } from "@opencode-ai/schema/money" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/util/effect/layer-node" @@ -115,7 +115,7 @@ const executionNode = makeGlobalNode({ return SessionExecution.Service.of({ active: Effect.succeed(new Set()), isActive: () => Effect.succeed(false), - resume: complete, + resume: (id) => complete(id).pipe(Effect.as({ type: "succeeded" as const })), wake: () => Effect.void, interrupt: () => Effect.succeed(false), awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid), @@ -170,6 +170,14 @@ const permissionIt = testEffect( offlineModels, ]), ) +// Real SessionExecution: the fake above fabricates a step on wake, so "nothing started" is only observable here. +const executionIt = testEffect( + AppNodeBuilder.build(nodes, [ + Permission.node.replace(permission), + Global.node.replace(tempGlobalLayer), + PluginSupervisor.node.replace(shellPluginSupervisor), + ]), +) const call = (input: typeof ShellTool.Input.Type, id = "call-shell") => ({ sessionID, @@ -1435,6 +1443,115 @@ describe("ShellTool", () => { { timeout: 15_000 }, ) + it.live("returns a stopped result with the captured output when the user kills a foreground command", () => + Effect.gen(function* () { + const tmp = yield* tmpdirScoped() + reset() + yield* withSession(tmp.path, (registry) => + Effect.gen(function* () { + const shell = yield* Shell.Service + const ready = yield* Deferred.make() + const running = yield* executeTool(registry, { + ...call({ + command: isWindows ? "Write-Output started; Start-Sleep -Seconds 60" : "echo started; sleep 60", + }), + progress: (update) => + typeof update.shellID === "string" + ? Deferred.succeed(ready, update.shellID).pipe(Effect.asVoid) + : Effect.void, + }).pipe(Effect.forkScoped) + const id = ID.make(yield* Deferred.await(ready)) + yield* shell + .output(id) + .pipe(Effect.repeat({ until: (page) => page.size > 0, schedule: Schedule.spaced("10 millis") })) + expect(yield* shell.stop(id)).toMatchObject({ id, status: "killed" }) + + const result = yield* Fiber.join(running) + expect(result.metadata).toMatchObject({ status: "stopped", truncated: false }) + expect(result.metadata).not.toHaveProperty("exit") + expect(result.content).toEqual([ + Expected.text(expect.stringContaining("started")), + Expected.text("Command stopped by user. Do not restart it unless the user asks."), + ]) + const jobs = yield* Job.Service + expect(yield* jobs.get(id)).toMatchObject({ + status: "completed", + result: { kind: "shell", status: "killed" }, + }) + // Stopping keeps the command readable; only removal forgets it. + expect(yield* shell.get(id)).toMatchObject({ status: "killed" }) + }), + ) + }), + ) + + 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() + 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([]) + }), + ) + }), + ) + + executionIt.live("records a background user stop as a quiet notice 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 shell = yield* Shell.Service + const started: Session.ID[] = [] + yield* bus.project(SessionEvent.Execution.Started, (event) => + Effect.sync(() => void started.push(event.data.sessionID)), + ) + 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(ID.make(id)) + yield* jobs.pendingBackground.pipe( + Effect.repeat({ until: (pending) => pending.length === 0, schedule: Schedule.spaced("10 millis") }), + Effect.timeout("5 seconds"), + ) + 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: "stopped", shellID: id, jobID: id }, + }, + }, + ]) + // The stop reached the model-facing notice, so the shell must not have been forgotten. + expect(yield* shell.get(ID.make(id))).toMatchObject({ status: "killed" }) + }), + ) + }), + ) + it.live("returns the shell id for a background command", () => Effect.acquireUseRelease( Effect.promise(() => tmpdir()), @@ -1557,7 +1674,7 @@ describe("ShellTool", () => { { id: settled.metadata?.shellID, status: "completed", - output: "(no output)\n\nCommand exited with code 7.", + result: { kind: "shell", status: "exited", exit: 7, output: "(no output)", truncated: false }, }, ]) }), diff --git a/packages/core/test/tool-subagent.test.ts b/packages/core/test/tool-subagent.test.ts index 6b800e1c7a8b..5df542f9b584 100644 --- a/packages/core/test/tool-subagent.test.ts +++ b/packages/core/test/tool-subagent.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { Deferred, Effect, Fiber, Layer, Schema, Stream } from "effect" +import { Deferred, Effect, Fiber, Layer, Schedule, Schema, Stream } from "effect" import { LanguageModel } from "@opencode-ai/ai" import { OpenAIChat } from "@opencode-ai/ai/protocols" import { TestLLM } from "@opencode-ai/ai/testing" @@ -32,9 +32,11 @@ import { SessionStore } from "@opencode-ai/core/session/store" import { Plugin } from "@opencode-ai/core/plugin" import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor" import { Permission } from "@opencode-ai/core/permission" +import { SubagentOutcome } from "@opencode-ai/core/session/subagent-outcome" +import { SubagentJob } from "@opencode-ai/core/session/subagent-job" 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 { offlineModels } from "./fixture/models" import { testEffect } from "./lib/effect" @@ -94,7 +96,7 @@ const executionNode = makeGlobalNode({ return SessionExecution.Service.of({ active: Effect.succeed(new Set()), isActive: () => Effect.succeed(false), - resume: complete, + resume: (sessionID) => complete(sessionID).pipe(Effect.as({ type: "succeeded" as const })), wake: () => Effect.void, interrupt: () => Effect.succeed(false), awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid), @@ -127,12 +129,14 @@ const productionIt = testEffect(AppNodeBuilder.build(nodes, replacements)) const it = testEffect( AppNodeBuilder.build(nodes, [...replacements, PluginSupervisor.node.replace(subagentPluginSupervisor)]), ) +// Merged back in so tests can drive the child's model through TestLLM.Test. +const completionLLM = TestLLM.testLayer({ fallback: TestLLM.text(childText, "completion") }) const completionIt = testEffect( AppNodeBuilder.build(LayerNode.group([nodes, SessionRestart.node, KV.node]), [ Global.node.replace(tempGlobalLayer), offlineModels, 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: () => @@ -148,7 +152,7 @@ const completionIt = testEffect( ), }), ), - ]), + ]).pipe(Layer.provideMerge(completionLLM)), ) const withSubagent = (location: Location.Ref) => @@ -176,7 +180,231 @@ const withSubagent = (location: Location.Ref) => ).pipe(Effect.provide(locations.get(location))) }) +// A parent with a registry, plus one child whose first model request never completes. +const hangingChild = Effect.fn(function* (title: string, callID: string, input?: { background?: boolean }) { + 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, + }) + 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() + const jobs = yield* Job.Service + const bus = yield* Bus.Service + const launches: Array = [] + yield* bus.project(SessionEvent.Execution.Started, (event) => + jobs.get(event.data.sessionID).pipe(Effect.tap((job) => Effect.sync(() => void launches.push(job)))), + ) + const call = yield* executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + progress: (update) => Deferred.succeed(running, outputSessionID(update)).pipe(Effect.asVoid), + call: { + type: "tool-call", + id: callID, + name: SubagentTool.name, + input: { agent: "reviewer", description: title, prompt: "review", ...input }, + }, + }).pipe(Effect.forkScoped) + const childID = yield* Deferred.await(running) + yield* llm.wait(1) + // Both foreground and background new children have a Job before execution begins. + expect(launches).toMatchObject([{ id: childID, status: "running" }]) + yield* jobs + .get(childID) + .pipe( + Effect.repeat({ until: (info) => info?.status === "running", schedule: Schedule.spaced("5 millis") }), + Effect.timeout("5 seconds"), + ) + return { parent, registry, llm, call, childID } +}) + describe("SubagentTool", () => { + completionIt.live("uses the original job metadata when a joining call backgrounds the child", () => + Effect.gen(function* () { + const dir = yield* tmpdirScoped() + const sessions = yield* Session.Service + const jobs = yield* Job.Service + const subagents = yield* SubagentJob.make + const parent = yield* sessions.create({ + location: Location.Ref.make({ directory: AbsolutePath.make(dir.path) }), + model: parentModel, + }) + const child = yield* sessions.create({ parentID: parent.id, model: childModel }) + const completed = yield* Deferred.make() + const recovery = { + kind: "subagent" as const, + parentSessionID: parent.id, + childSessionID: child.id, + agent: "reviewer", + description: "Original review", + } + yield* jobs.start({ id: child.id, type: "subagent", recovery, run: Deferred.await(completed) }) + const joined = yield* subagents.start({ ...recovery, agent: "explorer", description: "Follow-up review" }) + expect(joined.recovery).toEqual(recovery) + yield* subagents.background(child.id) + expect(yield* jobs.pendingBackground).toMatchObject([{ recovery }]) + // A stopped result keeps the notice pending, making the actual live admission inspectable. + yield* Deferred.succeed(completed, { kind: "subagent", status: "interrupted" }) + yield* jobs.pendingBackground.pipe( + Effect.repeat({ until: (pending) => pending.length === 0, schedule: Schedule.spaced("5 millis") }), + Effect.timeout("5 seconds"), + ) + expect(yield* sessions.inbox(parent.id)).toMatchObject([ + { payload: { description: recovery.description, metadata: { agent: recovery.agent, state: "stopped" } } }, + ]) + }), + ) + + completionIt.live("answers a user stop of a foreground child as a stopped result the parent can continue", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const jobs = yield* Job.Service + const child = yield* hangingChild("foreground review", "call-user-stopped-subagent") + + expect(yield* sessions.interrupt(child.childID)).toBeTrue() + yield* sessions.wait(child.childID) + expect(yield* Fiber.join(child.call)).toEqual({ + status: "completed", + output: { sessionID: child.childID, status: "stopped", output: SubagentOutcome.stopped }, + content: [ + { + type: "text", + text: `\n${SubagentOutcome.stopped}\n`, + }, + ], + metadata: { sessionID: child.childID, status: "stopped" }, + }) + expect(yield* jobs.get(child.childID)).toMatchObject({ + status: "completed", + result: { kind: "subagent", status: "interrupted" }, + }) + expect(yield* child.llm.requests()).toHaveLength(1) + + // Explicit continuation is a fresh job generation with its own outcome. + const resumed = yield* executeTool(child.registry, { + sessionID: child.parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-explicitly-resumed-subagent", + name: SubagentTool.name, + input: { agent: "reviewer", description: "continued review", prompt: "continue", sessionID: child.childID }, + }, + }) + expect(resumed).toMatchObject({ + status: "completed", + content: [{ type: "text", text: completedOutput(child.childID) }], + metadata: { sessionID: child.childID, status: "completed" }, + }) + expect(yield* jobs.get(child.childID)).toMatchObject({ result: { status: "completed", text: childText } }) + expect(yield* child.llm.requests()).toHaveLength(2) + }), + ) + + completionIt.live("keeps an abandoned foreground job a tool error", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const jobs = yield* Job.Service + const child = yield* hangingChild("cancelled review", "call-cancelled-subagent") + + // Cancelling the job abandons the observation; the child itself was not stopped. + yield* jobs.cancel(child.childID) + expect(yield* Fiber.join(child.call)).toEqual({ + status: "error", + error: { type: "tool.execution", message: `Subagent cancelled (sessionID: ${child.childID})` }, + }) + expect(yield* jobs.get(child.childID)).toMatchObject({ status: "cancelled" }) + expect(yield* jobs.get(child.childID)).not.toHaveProperty("result") + yield* sessions.interrupt(child.childID) + yield* sessions.wait(child.childID) + }), + ) + + completionIt.live("records a user stop of a background child quietly, once, across restart replay", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const jobs = yield* Job.Service + const bus = yield* Bus.Service + const notifications: SessionMessage.ID[] = [] + const child = yield* hangingChild("background review", "call-user-stopped-background", { background: true }) + yield* bus.project(SessionEvent.InboxEnqueued, (event) => + Effect.sync(() => { + if (event.data.sessionID === child.parent.id && event.data.item.type === "synthetic") + notifications.push(event.data.inboxID) + }), + ) + expect(yield* Fiber.join(child.call)).toMatchObject({ metadata: { status: "running" } }) + + expect(yield* sessions.interrupt(child.childID)).toBeTrue() + yield* sessions.wait(child.childID) + yield* jobs.pendingBackground.pipe( + Effect.repeat({ until: (pending) => pending.length === 0, schedule: Schedule.spaced("5 millis") }), + Effect.timeout("5 seconds"), + ) + yield* sessions.wait(child.parent.id) + const inbox = yield* sessions.inbox(child.parent.id) + expect(inbox).toEqual([ + expect.objectContaining({ + type: "synthetic", + payload: { + description: "background review", + text: `\n${SubagentOutcome.stopped}\n`, + metadata: { source: "subagent", childID: child.childID, agent: "reviewer", state: "stopped" }, + }, + }), + ]) + const notificationID = inbox[0]?.id + if (!notificationID) return yield* Effect.die("Expected a notice") + expect(notifications).toEqual([notificationID]) + // Quiet: the idle parent made no request because of the stop. + expect(yield* child.llm.requests()).toHaveLength(1) + const execution = yield* SessionExecution.Service + expect(yield* execution.isActive(child.parent.id)).toBeFalse() + expect(yield* execution.isActive(child.childID)).toBeFalse() + + // Replay the persisted terminal after a crash between admission and acknowledgment. + const kv = yield* KV.Service + yield* kv.set(`job.background/${notificationID}`, { + id: child.childID, + notificationID, + recovery: { + kind: "subagent", + parentSessionID: child.parent.id, + childSessionID: child.childID, + agent: "reviewer", + description: "background review", + }, + status: "completed", + result: { kind: "subagent", status: "interrupted" }, + }) + yield* SessionRestart.Service.use((restart) => restart.resumeSuspendedSessions) + yield* sessions.wait(child.parent.id) + expect(yield* child.llm.requests()).toHaveLength(1) + expect(yield* sessions.inbox(child.parent.id)).toEqual(inbox) + expect(notifications).toEqual([notificationID]) + expect(yield* jobs.pendingBackground).toEqual([]) + + // The recorded notice enters context at the parent's next step. + yield* sessions.prompt({ sessionID: child.parent.id, text: "Continue with other work" }) + yield* sessions.wait(child.parent.id) + expect(yield* sessions.inbox(child.parent.id)).toEqual([]) + expect((yield* sessions.context(child.parent.id)).filter((message) => message.type === "synthetic")).toEqual([ + expect.objectContaining({ + id: notificationID, + metadata: { source: "subagent", childID: child.childID, agent: "reviewer", state: "stopped" }, + }), + ]) + expect(yield* child.llm.requests()).toHaveLength(2) + }), + ) + completionIt.live("admits one durable completion across live delivery and restart replay", () => Effect.acquireRelease( Effect.promise(() => tmpdir()), diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index 54fa594848c9..b281dfeeda76 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -11901,6 +11901,118 @@ "summary": "Read shell output" } }, + "/api/shell/{id}/stop": { + "post": { + "tags": ["shell"], + "operationId": "v2.shell.stop", + "parameters": [ + { + "name": "id", + "in": "path", + "schema": { + "type": "string", + "pattern": "^sh_" + }, + "required": true + }, + { + "name": "location", + "in": "query", + "schema": { + "anyOf": [ + { + "type": "object", + "properties": { + "directory": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "workspace": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + } + }, + "additionalProperties": false + }, + { + "type": "null" + } + ] + }, + "required": false, + "style": "deepObject", + "explode": true + } + ], + "security": [], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "location": { + "$ref": "#/components/schemas/Location.InfoEncoded" + }, + "data": { + "$ref": "#/components/schemas/Shell.Info" + } + }, + "required": ["location", "data"], + "additionalProperties": false + } + } + } + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedErrorEncoded" + } + } + } + }, + "404": { + "description": "ShellNotFoundError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ShellNotFoundErrorEncoded" + } + } + } + } + }, + "description": "Kill a running shell command. It ends as killed and stays readable with its captured output; an already finished command is returned unchanged.", + "summary": "Stop shell command" + } + }, "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/api/reference": { "get": { "tags": ["reference"], @@ -18221,6 +18333,12 @@ "type": "string", "enum": ["auto", "manual"] }, + "model": { + "$ref": "#/components/schemas/Model.Ref" + }, + "providerState": { + "$ref": "#/components/schemas/Session.Message.ProviderState_5" + }, "summary": { "type": "string" }, @@ -18448,6 +18566,9 @@ "Session.Message.ProviderState_4": { "type": "object" }, + "Session.Message.ProviderState_5": { + "type": "object" + }, "Session.Message.Shell": { "type": "object", "properties": { @@ -18484,7 +18605,7 @@ }, "status": { "type": "string", - "enum": ["running", "exited", "timeout", "killed"] + "enum": ["running", "exited", "timeout", "killed", "unavailable"] }, "exit": { "anyOf": [ @@ -19205,7 +19326,7 @@ }, "status": { "type": "string", - "enum": ["running", "exited", "timeout", "killed"] + "enum": ["running", "exited", "timeout", "killed", "unavailable"] }, "command": { "type": "string" diff --git a/packages/protocol/src/groups/shell.ts b/packages/protocol/src/groups/shell.ts index 0b4cc081f699..5a1c17065216 100644 --- a/packages/protocol/src/groups/shell.ts +++ b/packages/protocol/src/groups/shell.ts @@ -90,6 +90,23 @@ export const ShellGroup = HttpApiGroup.make("server.shell") }), ), ) + .add( + HttpApiEndpoint.post("shell.stop", "/api/shell/:id/stop", { + params: { id: Shell.ID }, + query: LocationQuery, + success: Location.response(Shell.Info), + error: ShellNotFoundError, + }) + .annotateMerge(locationQueryOpenApi) + .annotateMerge( + OpenApi.annotations({ + identifier: "v2.shell.stop", + summary: "Stop shell command", + description: + "Kill a running shell command. It ends as killed and stays readable with its captured output; an already finished command is returned unchanged.", + }), + ), + ) .add( HttpApiEndpoint.delete("shell.remove", "/api/shell/:id", { params: { id: Shell.ID }, diff --git a/packages/schema/src/shell.ts b/packages/schema/src/shell.ts index bcd09cd901ac..1afc759399b1 100644 --- a/packages/schema/src/shell.ts +++ b/packages/schema/src/shell.ts @@ -19,7 +19,8 @@ export const ID = IDSchema.pipe( ) export type ID = typeof ID.Type -export const Status = Schema.Literals(["running", "exited", "timeout", "killed"]) +// `unavailable` records a lost result (removal or retention), not a process exit or a user stop. +export const Status = Schema.Literals(["running", "exited", "timeout", "killed", "unavailable"]) export type Status = typeof Status.Type export const Time = Schema.Struct({ diff --git a/packages/server/src/handlers/shell.ts b/packages/server/src/handlers/shell.ts index 96d7530cd816..0d7bb75b71d8 100644 --- a/packages/server/src/handlers/shell.ts +++ b/packages/server/src/handlers/shell.ts @@ -77,6 +77,23 @@ export const ShellHandler = HttpApiBuilder.group(Api, "server.shell", (handlers) ) }), ) + .handle( + "shell.stop", + Effect.fn(function* (ctx) { + const shell = yield* Shell.Service + return yield* response( + shell + .stop(ctx.params.id) + .pipe( + Effect.catchTag( + "Shell.NotFoundError", + () => + new ShellNotFoundError({ id: ctx.params.id, message: `Shell command not found: ${ctx.params.id}` }), + ), + ), + ) + }), + ) .handle( "shell.remove", Effect.fn(function* (ctx) { diff --git a/packages/server/test/fetch.test.ts b/packages/server/test/fetch.test.ts index 5f34a1ee9fff..fab88559d549 100644 --- a/packages/server/test/fetch.test.ts +++ b/packages/server/test/fetch.test.ts @@ -355,6 +355,42 @@ it.live("serves the session view operation and missing-session error", () => }), ) +it.live("stops a shell command over HTTP without removing it", () => + Effect.gen(function* () { + const handler = yield* ServerFetch.make(options) + const json = (input: Request) => Effect.promise(() => handler(input).then((response) => response.json())) + const created = yield* json( + new Request("http://opencode.local/api/shell", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + command: process.platform === "win32" ? "Start-Sleep -Seconds 60" : "sleep 60", + timeout: 0, + }), + }), + ) + const id = Schema.decodeUnknownSync(Schema.Struct({ data: Schema.Struct({ id: Schema.String }) }))(created).data.id + + const stopped = yield* Effect.promise(() => + handler(new Request(`http://opencode.local/api/shell/${id}/stop`, { method: "POST" })), + ) + expect(stopped.status).toBe(200) + expect(yield* Effect.promise(() => stopped.json())).toMatchObject({ data: { id, status: "killed" } }) + // Still readable after the stop; a second stop is idempotent. + expect(yield* json(new Request(`http://opencode.local/api/shell/${id}`))).toMatchObject({ + data: { id, status: "killed" }, + }) + expect(yield* json(new Request(`http://opencode.local/api/shell/${id}/stop`, { method: "POST" }))).toMatchObject({ + data: { id, status: "killed" }, + }) + + const missing = yield* Effect.promise(() => + handler(new Request("http://opencode.local/api/shell/sh_missing/stop", { method: "POST" })), + ) + expect(missing.status).toBe(404) + }), +) + it.live("routes pending requests through the Session's instance", () => Effect.gen(function* () { const config = yield* Effect.acquireDisposable(Effect.promise(() => tmpdir("opencode-pending-read-"))) diff --git a/packages/server/test/session-message-update.test.ts b/packages/server/test/session-message-update.test.ts index 91f4c64c7229..53906d5e4a33 100644 --- a/packages/server/test/session-message-update.test.ts +++ b/packages/server/test/session-message-update.test.ts @@ -30,7 +30,7 @@ it.live("updates completed assistant message content through the session HTTP AP return SessionExecution.Service.of({ active: Effect.sync(() => state.active), isActive: (sessionID) => Effect.sync(() => state.active.has(sessionID)), - resume: () => Effect.void, + resume: () => Effect.succeed({ type: "succeeded" as const }), wake: (sessionID) => Effect.gen(function* () { yield* bus.publish(SessionEvent.InboxDelivered, { sessionID, inboxID: state.user }) diff --git a/packages/tui/src/context/session-tabs.tsx b/packages/tui/src/context/session-tabs.tsx index 01762c08bfee..e4b20b986efe 100644 --- a/packages/tui/src/context/session-tabs.tsx +++ b/packages/tui/src/context/session-tabs.tsx @@ -169,7 +169,14 @@ 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), + // A pending user-stop notice is context for the next step, not work that will start on its own. + busy: members.some( + (id) => + data.session.status(id) === "running" || + data.session.pending + .list(id) + .some((item) => item.type !== "synthetic" || item.payload.metadata?.state !== "stopped"), + ), renaming: data.session.title.pending(session), } } diff --git a/packages/tui/src/routes/session/composer/shell-tab.tsx b/packages/tui/src/routes/session/composer/shell-tab.tsx index 81a8acd0882b..e6a433add4df 100644 --- a/packages/tui/src/routes/session/composer/shell-tab.tsx +++ b/packages/tui/src/routes/session/composer/shell-tab.tsx @@ -81,7 +81,7 @@ export function ShellTab(props: { sessionID: string }) { run() { const entry = selectedEntry() if (!entry) return - void client.api.shell.remove({ + void client.api.shell.stop({ id: entry.id, location: { directory: entry.location.directory, workspace: entry.location.workspaceID }, }) diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index 8d7d1d4605f1..19193b7c812b 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -2137,13 +2137,16 @@ function SessionNoticeMessageV2(props: { message: SessionMessageInfo }) { const description = () => (source() === "shell" ? text().replace(/\s+/g, " ").trim() : text()) const status = () => { if (state() === "completed") return "finished" + if (state() === "stopped") return "stopped by user" if (state() === "error") return "failed" return state() ?? "finished" } - const heading = () => `${state() === "completed" ? "↳" : "!"} ${actor()} ${status()}` + const heading = () => `${state() === "completed" || state() === "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 + // A user stop is the user's own action, not feedback about the work. + if (state() === "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 @@ -2344,7 +2347,7 @@ function RevertMessage(props: { function ShellMessage(props: { message: Extract }) { const error = createMemo(() => { - if (props.message.status === "killed") return "Command cancelled" + if (props.message.status === "unavailable") return "Command result is no longer available" if (props.message.status === "timeout") return "Command timed out" if (props.message.exit !== undefined && props.message.exit !== 0) return `Command exited with code ${props.message.exit}` @@ -2355,6 +2358,7 @@ function ShellMessage(props: { message: Extract @@ -3199,6 +3203,14 @@ function BlockToolContent(props: BlockToolProps & { borderColor: RGBA }) { const SHELL_DISPLAY_LIMIT = 1024 * 1024 function Shell(props: ToolProps) { + const stopped = () => props.metadata.status === "stopped" + const output = () => { + if (stringValue(props.metadata.shellID)) return undefined + if (!stopped()) return props.output + // Show the captured output alone; the trailing sentence tells the model not to restart it. + const first = toolDisplayContent(props.part.state)[0] + return first?.type === "text" ? first.text : undefined + } return ( ) } @@ -3218,6 +3231,8 @@ function ShellDisplay(props: { command?: string workdir?: string status: SessionMessageAssistantTool["state"]["status"] + // The user killed the command: rendered as a neutral note, not as failure. + stopped?: boolean background?: boolean output?: string error?: string @@ -3235,7 +3250,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("") @@ -3328,7 +3343,12 @@ function ShellDisplay(props: { } return ( - + stringValue(props.metadata.sessionID) ?? stringValue(props.metadata.sessionId)) const description = createMemo(() => stringValue(props.input.description)) const continuation = createMemo(() => Boolean(stringValue(props.input.sessionID))) + const stopped = () => props.part.state.status === "completed" && props.metadata.status === "stopped" 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 ( stopped by user + ) : isBackgroundSubagent(props.metadata, props.part.state.status) ? ( Background ) : undefined } diff --git a/packages/tui/test/cli/tui/composer-keymap.test.tsx b/packages/tui/test/cli/tui/composer-keymap.test.tsx index 0c473d95d7cd..29f6adf6134a 100644 --- a/packages/tui/test/cli/tui/composer-keymap.test.tsx +++ b/packages/tui/test/cli/tui/composer-keymap.test.tsx @@ -30,7 +30,7 @@ async function renderComposer( ) { const events = createEventStream() const interrupted: string[] = [] - const removed: string[] = [] + const stopped: string[] = [] const ready = Promise.withResolvers() let closed = 0 let dispatch!: ReturnType["dispatch"] @@ -52,10 +52,14 @@ async function renderComposer( data: shells, }) } - const shellID = url.pathname.match(/^\/api\/shell\/([^/]+)$/)?.[1] - if (shellID && request.method === "DELETE") { - removed.push(shellID) - return new Response(null, { status: 204 }) + const shellID = url.pathname.match(/^\/api\/shell\/([^/]+)\/stop$/)?.[1] + if (shellID && request.method === "POST") { + stopped.push(shellID) + const shell = shells.find((item) => item.id === shellID) + return json({ + location: { directory, project: { id: "proj_test", directory } }, + data: { ...shell, status: "killed" }, + }) } }, events) @@ -118,7 +122,7 @@ async function renderComposer( return { app, interrupted, - removed, + stopped, route: () => route.data, dispatch: (command: string) => dispatch(command), closed: () => closed, @@ -162,12 +166,12 @@ test("disabled shell bindings have no component fallbacks", async () => { composer.app.mockInput.pressKey("d", { ctrl: true }) await composer.app.renderOnce() expect(composer.closed()).toBe(0) - expect(composer.removed).toEqual([]) + expect(composer.stopped).toEqual([]) composer.app.mockInput.pressArrow("down") composer.dispatch("composer.shell.kill") - await wait(() => composer.removed.length === 1) - expect(composer.removed).toEqual(["sh-a"]) + await wait(() => composer.stopped.length === 1) + expect(composer.stopped).toEqual(["sh-a"]) } finally { composer.app.renderer.destroy() } @@ -180,8 +184,8 @@ test("configured composer bindings work with a focused textarea", async () => { await composer.app.renderOnce() expect(composer.app.captureCharFrame()).toContain("bun test") composer.app.mockInput.pressKey("u", { ctrl: true }) - await wait(() => composer.removed.length === 1) - expect(composer.removed).toEqual(["sh-a"]) + await wait(() => composer.stopped.length === 1) + expect(composer.stopped).toEqual(["sh-a"]) } finally { composer.app.renderer.destroy() } diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 7776514ea54b..aca9a7ed83fe 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -1984,11 +1984,23 @@ test("keeps shell state scoped to location", async () => { const events = createEventStream() const other = "/tmp/opencode/other" const workspace = "ws_other" - let removed: URL | undefined + let stopped: URL | undefined const calls = createFetch((url, request) => { - if (url.pathname === "/api/shell/sh_other" && request.method === "DELETE") { - removed = url - return new Response(null, { status: 204 }) + if (url.pathname === "/api/shell/sh_other/stop" && request.method === "POST") { + stopped = url + return json({ + location: { directory: other, workspaceID: workspace, project: { id: "proj_test", directory: other } }, + data: { + id: "sh_other", + status: "killed", + command: "pnpm dev", + cwd: other, + shell: "/bin/sh", + file: "/tmp/opencode-shell", + metadata: { sessionID: "ses_shared" }, + time: { started: 1, completed: 2 }, + }, + }) } if (url.pathname !== "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/api/shell") return const requestDirectory = url.searchParams.get("location[directory]") @@ -2054,9 +2066,9 @@ test("keeps shell state scoped to location", async () => { await app.waitForFrame((frame) => frame.includes("pnpm dev")) app.mockInput.pressArrow("down") app.mockInput.pressKey("d", { ctrl: true }) - await wait(() => removed !== undefined) - expect(removed?.searchParams.get("location[directory]")).toBe(other) - expect(removed?.searchParams.get("location[workspace]")).toBe(workspace) + await wait(() => stopped !== undefined) + expect(stopped?.searchParams.get("location[directory]")).toBe(other) + expect(stopped?.searchParams.get("location[workspace]")).toBe(workspace) events.emit({ id: "evt_shell_created", diff --git a/packages/tui/test/context/session-tabs.test.tsx b/packages/tui/test/context/session-tabs.test.tsx index 4b31c15ee5a3..65273cd35cee 100644 --- a/packages/tui/test/context/session-tabs.test.tsx +++ b/packages/tui/test/context/session-tabs.test.tsx @@ -898,6 +898,49 @@ test("closing a tab is not undone by another TUI viewing the same session", asyn } }) +test("a pending user-stop notice does not keep its tab busy", async () => { + const setup = await renderSessionTabs("parent", { persisted: ["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: "shell", state: "stopped" } }, + }, + }, + }) + await wait(() => setup.data.session.pending.list("parent").length === 1) + expect(setup.tabs.status("parent").busy).toBe(false) + + // Real execution still shows busy while the notice is pending. + 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)) + } + + // As does any other pending input. + 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", { persisted: ["background"] }) diff --git a/packages/tui/test/session-user-stop.test.tsx b/packages/tui/test/session-user-stop.test.tsx new file mode 100644 index 000000000000..428c83b87daf --- /dev/null +++ b/packages/tui/test/session-user-stop.test.tsx @@ -0,0 +1,207 @@ +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" + +const STOPPED_SHELL = "Command stopped by user. Do not restart it unless the user asks." +const STOPPED_SUBAGENT = "Subagent stopped by user. Do not restart it unless the user asks." + +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: 56, 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 }, + output: { output: "user shell output", cursor: 17, size: 17, truncated: false }, + time: { created: 0, completed: 1 }, + }, + { + id: "message-unavailable-shell", + type: "shell", + shellID: "shell-unavailable", + command: "echo expired", + status: "unavailable", + 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" }, + // The real tool response: captured output first, then the model-facing sentence. + content: [ + { type: "text", text: "Partial shell output" }, + { type: "text", text: STOPPED_SHELL }, + ], + metadata: { status: "stopped", truncated: false }, + }, + 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: STOPPED_SUBAGENT }], + metadata: { status: "stopped", 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: STOPPED_SHELL, + description: "sleep 60", + metadata: { source: "shell", state: "stopped", shellID: "shell-background" }, + }, + timeCreated: 5, + }, + ] + const events = createEventStream() + const calls = createFetch((url) => { + if (url.pathname === "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/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: { prepare: async () => ({ directory: "" }) }, + 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")) + const frame = setup.captureCharFrame() + // `!` shell, shell tool, subagent tool, pending shell notice. + expect(frame.match(/stopped by user/g)).toHaveLength(4) + expect(frame).not.toContain("Do not restart it") + expect(frame).toContain("Partial shell output") + expect(frame).toContain("user shell output") + expect(frame).toContain("Command result is no longer available") + expect(frame).not.toContain("Command cancelled") + expect(frame).toContain("\u21b3 Explore Subagent") + expect(frame).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: STOPPED_SUBAGENT, + description: "Inspect source", + metadata: { source: "subagent", state: "stopped", 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)) + } +}) diff --git a/packages/www/openapi.json b/packages/www/openapi.json index 54fa594848c9..b281dfeeda76 100644 --- a/packages/www/openapi.json +++ b/packages/www/openapi.json @@ -11901,6 +11901,118 @@ "summary": "Read shell output" } }, + "/api/shell/{id}/stop": { + "post": { + "tags": ["shell"], + "operationId": "v2.shell.stop", + "parameters": [ + { + "name": "id", + "in": "path", + "schema": { + "type": "string", + "pattern": "^sh_" + }, + "required": true + }, + { + "name": "location", + "in": "query", + "schema": { + "anyOf": [ + { + "type": "object", + "properties": { + "directory": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "workspace": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + } + }, + "additionalProperties": false + }, + { + "type": "null" + } + ] + }, + "required": false, + "style": "deepObject", + "explode": true + } + ], + "security": [], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "location": { + "$ref": "#/components/schemas/Location.InfoEncoded" + }, + "data": { + "$ref": "#/components/schemas/Shell.Info" + } + }, + "required": ["location", "data"], + "additionalProperties": false + } + } + } + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedErrorEncoded" + } + } + } + }, + "404": { + "description": "ShellNotFoundError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ShellNotFoundErrorEncoded" + } + } + } + } + }, + "description": "Kill a running shell command. It ends as killed and stays readable with its captured output; an already finished command is returned unchanged.", + "summary": "Stop shell command" + } + }, "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/api/reference": { "get": { "tags": ["reference"], @@ -18221,6 +18333,12 @@ "type": "string", "enum": ["auto", "manual"] }, + "model": { + "$ref": "#/components/schemas/Model.Ref" + }, + "providerState": { + "$ref": "#/components/schemas/Session.Message.ProviderState_5" + }, "summary": { "type": "string" }, @@ -18448,6 +18566,9 @@ "Session.Message.ProviderState_4": { "type": "object" }, + "Session.Message.ProviderState_5": { + "type": "object" + }, "Session.Message.Shell": { "type": "object", "properties": { @@ -18484,7 +18605,7 @@ }, "status": { "type": "string", - "enum": ["running", "exited", "timeout", "killed"] + "enum": ["running", "exited", "timeout", "killed", "unavailable"] }, "exit": { "anyOf": [ @@ -19205,7 +19326,7 @@ }, "status": { "type": "string", - "enum": ["running", "exited", "timeout", "killed"] + "enum": ["running", "exited", "timeout", "killed", "unavailable"] }, "command": { "type": "string" diff --git a/packages/www/public/openapi.json b/packages/www/public/openapi.json index 54fa594848c9..b281dfeeda76 100644 --- a/packages/www/public/openapi.json +++ b/packages/www/public/openapi.json @@ -11901,6 +11901,118 @@ "summary": "Read shell output" } }, + "/api/shell/{id}/stop": { + "post": { + "tags": ["shell"], + "operationId": "v2.shell.stop", + "parameters": [ + { + "name": "id", + "in": "path", + "schema": { + "type": "string", + "pattern": "^sh_" + }, + "required": true + }, + { + "name": "location", + "in": "query", + "schema": { + "anyOf": [ + { + "type": "object", + "properties": { + "directory": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "workspace": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + } + }, + "additionalProperties": false + }, + { + "type": "null" + } + ] + }, + "required": false, + "style": "deepObject", + "explode": true + } + ], + "security": [], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "location": { + "$ref": "#/components/schemas/Location.InfoEncoded" + }, + "data": { + "$ref": "#/components/schemas/Shell.Info" + } + }, + "required": ["location", "data"], + "additionalProperties": false + } + } + } + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedErrorEncoded" + } + } + } + }, + "404": { + "description": "ShellNotFoundError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ShellNotFoundErrorEncoded" + } + } + } + } + }, + "description": "Kill a running shell command. It ends as killed and stays readable with its captured output; an already finished command is returned unchanged.", + "summary": "Stop shell command" + } + }, "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/api/reference": { "get": { "tags": ["reference"], @@ -18221,6 +18333,12 @@ "type": "string", "enum": ["auto", "manual"] }, + "model": { + "$ref": "#/components/schemas/Model.Ref" + }, + "providerState": { + "$ref": "#/components/schemas/Session.Message.ProviderState_5" + }, "summary": { "type": "string" }, @@ -18448,6 +18566,9 @@ "Session.Message.ProviderState_4": { "type": "object" }, + "Session.Message.ProviderState_5": { + "type": "object" + }, "Session.Message.Shell": { "type": "object", "properties": { @@ -18484,7 +18605,7 @@ }, "status": { "type": "string", - "enum": ["running", "exited", "timeout", "killed"] + "enum": ["running", "exited", "timeout", "killed", "unavailable"] }, "exit": { "anyOf": [ @@ -19205,7 +19326,7 @@ }, "status": { "type": "string", - "enum": ["running", "exited", "timeout", "killed"] + "enum": ["running", "exited", "timeout", "killed", "unavailable"] }, "command": { "type": "string" diff --git a/specs/v2/session.md b/specs/v2/session.md index 2774863477ea..d91038536388 100644 --- a/specs/v2/session.md +++ b/specs/v2/session.md @@ -40,6 +40,12 @@ References share operation implementations, host services, and the existing exec User shell commands start immediately as background work without waiting for model execution or other user shells. They do not suppress prompt wakeups. The shell operation forks completion recording into the captured host Scope with `startImmediately: true` and joins that fiber, so caller cancellation does not cancel recording; closing the host Scope does. Shell started and ended events retain output in one shell entry. Completion and startup-failure notifications are admitted as synthetic input with `resume: false`, without waking execution. Shell services are resolved at startup, while Session events remain outside that Location context so movement does not pin them to the old Location. +Background tool work (shell commands and subagents) is tracked by `Job`, which transports the producer's typed outcome: a job is `completed` when its run reported how the work ended, `error` when the run died, and `cancelled` when it was abandoned before reporting. New subagents admit their prompt without waking and start through their Job; continuations still wake to steer an existing execution. A running Job retains its original recovery description when a later caller joins it. Live and recovered notices use that same descriptor. + +The Shell service owns a command's terminal state. An explicit stop ends as `killed`, keeps the command and its captured output readable, and completes settlement even if the requesting caller is interrupted. Overlapping stops await the same completion. A removed or expired result is `unavailable`, never an inferred user stop or exit time. `SessionExecution.resume` reports an interruption-only user terminal as a value; mixed causes preserve their failures. Shutdown still interrupts process-local joiners so the durable claim can resume the Session after restart. + +One `BackgroundNotice` path renders live and recovered terminals into synthetic notices. A user stop is admitted with `resume: false` so an idle owner records it for its next step without waking. Persisted stopped Job outcomes replay quietly; older markers containing rendered `output` replay that text without inventing new outcome fields. There is no atomic commit across the Session terminal and Job outcome: a process death after the child records its interruption but before the Job persists its result can leave a running background marker that recovery resumes. Stronger reconciliation across that handoff is not provided here. + Location services are acquired only when an operation needs them. In particular, retry reconciliation happens before prompt preparation, so an already-admitted input skips hooks and attachment resolution. Execution continues to resolve placement independently at drain start and after movement. `servicesFor` selects instance services from the saved placement. Each instance constructs `SessionPrompt.Service`, whose `prepare` method turns submitted input into a user inbox item without admitting it, committing a revert, or waking execution. It captures FSUtil, PluginSupervisor, PluginHooks, Image, and Skill; readiness is checked before hooks on every call. Prompt keeps early retry reconciliation outside preparation and invokes preparation interruptibly in the current instance. Lower Session does not depend directly on Database or FSUtil; its Location requirements are SessionPrompt, SessionRevert, Shell, and the PluginSupervisor still used by manual shell startup.