Visitar URL original
refactor(core): stop an interrupted shell call through cooperative ca… · anomalyco/opencode@9c7c5f0 · GitHub
Skip to content

Commit 9c7c5f0

Browse files
jlongsterjlongster
committed
refactor(core): stop an interrupted shell call through cooperative cancel
The user's stop now reaches tools through a new `cancel` effect on Tool.Context instead of the ToolInterrupted side channel. The step gives each tool call a cancel gate: on a user stop it completes the gates, lets every tool that was awaiting its gate settle its own call (failing normally, through execute.after hooks and failTool), interrupts every other tool at once, and interrupts stragglers after 5 s. A server shutdown or a closed Location interrupts every tool as before. The step still ends interrupted and starts no new request. The shell tool races its job against `cancel`: on a stop it stops the command, reads the output so far (waiting at most 500 ms) and fails with a ToolFailure carrying the tail, the output file and the shell ID. The interruption reason now travels explicitly from the session execution through the runner to the step, like isLocationClosed. CodeMode never hands `cancel` to nested calls. tool/interrupted.ts and session/interruption.ts are gone. User decisions: - An interrupted shell call is a failed call carrying its output, not a success - Use cooperative cancel instead of the side-channel slot - execute.after hooks must still run and be able to replace the error
1 parent 95ccf88 commit 9c7c5f0

16 files changed

Lines changed: 274 additions & 161 deletions

File tree

‎packages/core/src/codemode/tool.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,8 @@ export const create = (
132132
...context,
133133
// Progress replaces the execute call's own, so an inner tool's updates would hide the call rows.
134134
progress: () => Effect.void,
135+
// A stop is the execute call's to handle: a nested call never fails it with its own error.
136+
cancel: undefined,
135137
}).pipe(
136138
Effect.mapError((error) => new ToolFailure({ message: toSessionError(error).message, error })),
137139
)

‎packages/core/src/session/execution.ts‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,6 @@ import { SessionSchema } from "./schema.js"
1313
import { SessionStore } from "./store.js"
1414
import { toSessionError } from "./to-session-error.js"
1515
import { SessionInbox } from "./inbox.js"
16-
import { SessionInterruption } from "./interruption.js"
1716

1817
export interface Interface {
1918
/** Snapshots active execution owned by this process. */
@@ -82,6 +81,8 @@ export const layer = Layer.effect(
8281
const releaseOnCommit = (sessionID: SessionSchema.ID) => ({
8382
commit: () => store.release(sessionID),
8483
})
84+
// Why a Session's execution is being interrupted, read from the coordinator once it exists.
85+
let interruption: (sessionID: SessionSchema.ID) => InterruptReason | undefined = () => undefined
8586
const drain = Effect.fnUntraced(function* (
8687
sessionID: SessionSchema.ID,
8788
force: boolean,
@@ -92,7 +93,14 @@ export const layer = Layer.effect(
9293
const session = yield* store.get(sessionID)
9394
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
9495
const result = yield* SessionRunner.Service.use((runner) =>
95-
runner.drain({ sessionID, force, continuation, promotable, stepped }),
96+
runner.drain({
97+
sessionID,
98+
force,
99+
continuation,
100+
promotable,
101+
stepped,
102+
interruption: () => interruption(sessionID),
103+
}),
96104
).pipe(
97105
instances.provide(session),
98106
Effect.tapCause((cause) =>
@@ -113,10 +121,7 @@ export const layer = Layer.effect(
113121
sessionID,
114122
bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)),
115123
),
116-
drain: (sessionID, force, promotable): Effect.Effect<void, SessionRunner.RunError> =>
117-
drain(sessionID, force, undefined, promotable).pipe(
118-
Effect.provideService(SessionInterruption.Reason, () => coordinator.interruption(sessionID)),
119-
),
124+
drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable),
120125
// One terminal observation per busy period, covering every coalesced drain.
121126
settled: (sessionID, exit, reason) =>
122127
reportLifecycle(
@@ -149,6 +154,7 @@ export const layer = Layer.effect(
149154
),
150155
})
151156

157+
interruption = coordinator.interruption
152158
return Service.of({
153159
active: coordinator.active,
154160
isActive: coordinator.isActive,

‎packages/core/src/session/interruption.ts‎

Lines changed: 0 additions & 12 deletions
This file was deleted.

‎packages/core/src/session/runner/index.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@ export interface Interface {
3838
readonly promotable?: Promotable
3939
/** The previous drain ran a step before handing off, so reaching idle still runs the idle hook. */
4040
readonly stepped?: boolean
41+
/** Why the execution is being interrupted, once it is: `"user"` when someone stopped it. */
42+
readonly interruption?: () => "user" | "shutdown" | undefined
4143
}) => Effect.Effect<DrainResult, RunError>
4244
}
4345

‎packages/core/src/session/runner/llm.ts‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,9 @@ const layer = Layer.effect(
178178
// The Session would go idle; a plugin may admit input to keep this execution working.
179179
yield* restore(
180180
SessionBackground.outstanding(jobs, store, sessionID).pipe(
181-
Effect.flatMap((background) => hooks.trigger("session", "experimental.idle", { sessionID, background })),
181+
Effect.flatMap((background) =>
182+
hooks.trigger("session", "experimental.idle", { sessionID, background }),
183+
),
182184
),
183185
)
184186
if (!(yield* SessionInbox.has(db, sessionID, "input"))) return DrainResult.Complete()
@@ -212,7 +214,7 @@ const layer = Layer.effect(
212214
while (true) {
213215
const next = yield* advanceToStep()
214216
if (next._tag !== "Ready") return next
215-
continuing = yield* runStep(next.context, step)
217+
continuing = yield* runStep(next.context, step, input.interruption)
216218
stepped = true
217219
step++
218220
force = false
@@ -230,7 +232,11 @@ const layer = Layer.effect(
230232
})
231233

232234
/** Owns logical Step policy; each attempt owns its streaming, tools, and durable settlement. */
233-
const runStep = Effect.fn("SessionRunner.runStep")(function* (first: SessionContext.Loaded, step: number) {
235+
const runStep = Effect.fn("SessionRunner.runStep")(function* (
236+
first: SessionContext.Loaded,
237+
step: number,
238+
interruption: DrainInput["interruption"],
239+
) {
234240
const sessionID = first.session.id
235241
let assistantMessageID = SessionMessage.ID.create()
236242
const retry = yield* SessionRunnerRetry.make(bus, sessionID)
@@ -286,6 +292,7 @@ const layer = Layer.effect(
286292
)
287293
const outcome = yield* steps.attempt({
288294
isLocationClosed: lifecycle.isClosed,
295+
interruption,
289296
sessionID,
290297
assistantMessageID,
291298
agent: loaded.agent.id,

‎packages/core/src/session/runner/step.ts‎

Lines changed: 53 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,14 @@ import {
88
type ToolCall,
99
} from "@opencode/ai/effect"
1010
import type { Agent } from "@opencode/schema/agent"
11-
import { Cause, Clock, Data, Effect, Exit, Fiber, Option, Stream } from "effect"
11+
import { Cause, Clock, Data, Deferred, Effect, Exit, Fiber, Option, Stream } from "effect"
1212
import { SessionError } from "@opencode/schema/session-error"
1313
import { Bus } from "../../bus.js"
1414
import { Permission } from "../../permission.js"
1515
import { Snapshot } from "../../snapshot.js"
1616
import { Tool } from "../../tool.js"
1717
import { ToolOutput } from "../../tool-output.js"
1818
import { QuestionTool } from "../../tool/plugin/question.js"
19-
import { ToolInterrupted } from "../../tool/interrupted.js"
2019
import { StepFailedError } from "../error.js"
2120
import { SessionEvent } from "../event.js"
2221
import { SessionMessage } from "../message.js"
@@ -43,6 +42,8 @@ export const Outcome = Data.taggedEnum<Outcome>()
4342

4443
interface Input {
4544
readonly isLocationClosed: () => boolean
45+
/** Why the execution is being interrupted, once it is: `"user"` when someone stopped it. */
46+
readonly interruption?: () => "user" | "shutdown" | undefined
4647
readonly sessionID: SessionSchema.ID
4748
readonly assistantMessageID: SessionMessage.ID
4849
readonly agent: Agent.ID
@@ -98,9 +99,20 @@ export const make = Effect.gen(function* () {
9899
const toolRuns: Array<{
99100
readonly call: ToolCall
100101
readonly fiber: Fiber.Fiber<void, Permission.DeclinedError | QuestionTool.CancelledError>
102+
readonly cancel: Cancel
101103
}> = []
102104
const interruptTools = Effect.suspend(() => Fiber.interruptAll(toolRuns.map((run) => run.fiber)))
103-
const executeTool = (call: ToolCall) => {
105+
// A user's stop asks each tool awaiting `cancel` to stop and settle its own call, and interrupts every other
106+
// tool. A tool that does not settle in time is interrupted too. Teardown interrupts every tool.
107+
const stopTools = Effect.suspend(() => {
108+
const user = input.interruption?.() === "user" && !input.isLocationClosed()
109+
const immediate = toolRuns.flatMap((run) => (user && run.cancel.trigger() ? [] : [run.fiber]))
110+
return Fiber.interruptAll(immediate).pipe(
111+
Effect.andThen(Fiber.awaitAll(toolRuns.map((run) => run.fiber)).pipe(Effect.timeoutOption(COOPERATIVE_STOP))),
112+
Effect.flatMap((settled) => (Option.isSome(settled) ? Effect.void : interruptTools)),
113+
)
114+
})
115+
const executeTool = (call: ToolCall, cancel: Cancel) => {
104116
if (input.prepared.request.toolChoice?.type === "none")
105117
return new Tool.Error({ message: "Tools are disabled after the maximum agent steps" })
106118
return input.prepared.executeTool({
@@ -109,6 +121,7 @@ export const make = Effect.gen(function* () {
109121
messageID: input.assistantMessageID,
110122
call,
111123
progress: (update) => publisher.progress(call.id, update),
124+
cancel: cancel.effect,
112125
})
113126
}
114127

@@ -127,18 +140,12 @@ export const make = Effect.gen(function* () {
127140
if (firstOutputAt === undefined && publisher.record().outputStarted)
128141
firstOutputAt = yield* Clock.currentTimeMillis
129142
if (event.type !== "tool-call" || event.providerExecuted) return
130-
const interrupted: ToolInterrupted.Slot = {}
143+
const cancel = makeCancel()
131144
toolRuns.push({
132145
call: event,
146+
cancel,
133147
fiber: yield* Effect.uninterruptibleMask((restore) =>
134-
restore(executeTool(event).pipe(Effect.provideService(ToolInterrupted.Slot, interrupted))).pipe(
135-
// An interrupted tool may still settle its call with its own failure. The step is interrupted
136-
// either way, so it starts no other request.
137-
Effect.catchCause((cause) =>
138-
interrupted.failure && Cause.hasInterruptsOnly(cause)
139-
? Effect.fail(interrupted.failure)
140-
: Effect.failCause(cause),
141-
),
148+
restore(executeTool(event, cancel)).pipe(
142149
Effect.flatMap(toolOutput.truncate),
143150
Effect.flatMap((outcome) => publisher.toolExecution(event.id, event.name, outcome)),
144151
Effect.catchTag("Tool.Error", (error) =>
@@ -167,9 +174,9 @@ export const make = Effect.gen(function* () {
167174
)
168175
return yield* Effect.failCause(stream.cause)
169176
if (publisher.hasStarted()) yield* publisher.streamed()
170-
if (streamInterrupted) yield* interruptTools
177+
if (streamInterrupted) yield* stopTools
171178
const joined = yield* restore(Fiber.awaitAll(toolRuns.map((run) => run.fiber))).pipe(Effect.exit)
172-
if (Exit.isFailure(joined)) yield* interruptTools
179+
if (!streamInterrupted && Exit.isFailure(joined)) yield* stopTools
173180
const tools = classifyToolExits(joined, toolRuns)
174181

175182
if (
@@ -350,3 +357,35 @@ const classifyToolExits = (
350357
.at(0)
351358
return { interrupted: causes.some(Cause.hasInterrupts), declines, failure }
352359
}
360+
361+
// How long a user's stop waits for the tools it asked to stop before interrupting them.
362+
const COOPERATIVE_STOP = "5 seconds"
363+
364+
type Cancel = ReturnType<typeof makeCancel>
365+
366+
// A tool call's `cancel`. `trigger` completes it, and reports whether the tool was awaiting it, so would stop on its
367+
// own; a tool that was not is interrupted instead.
368+
const makeCancel = () => {
369+
let listeners = 0
370+
let cancelled = false
371+
const gate = Deferred.makeUnsafe<void>()
372+
return {
373+
effect: Effect.suspend(() => {
374+
if (cancelled) return Effect.void
375+
listeners++
376+
return Deferred.await(gate).pipe(
377+
Effect.ensuring(
378+
Effect.sync(() => {
379+
listeners--
380+
}),
381+
),
382+
)
383+
}),
384+
trigger: () => {
385+
cancelled = true
386+
if (listeners === 0) return false
387+
Deferred.doneUnsafe(gate, Exit.void)
388+
return true
389+
},
390+
}
391+
}

‎packages/core/src/tool.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@ import { CodeModeCatalog } from "./codemode/catalog.js"
1212
import { CodeModeTool } from "./codemode/tool.js"
1313
import { Flag } from "./flag.js"
1414
import { Image } from "./image.js"
15-
import { ToolInterrupted } from "./tool/interrupted.js"
1615
import { Permission } from "./permission.js"
1716
import { PluginHooks } from "./plugin/hooks.js"
1817
import { SessionMessage } from "./session/message.js"
@@ -76,6 +75,8 @@ export interface Snapshot {
7675
readonly messageID: SessionMessage.ID
7776
readonly call: ToolCall
7877
readonly progress?: (update: Tool.Metadata) => Effect.Effect<void>
78+
/** Completes when the user asks the call to stop; see `Tool.Context.cancel`. */
79+
readonly cancel?: Effect.Effect<void>
7980
/** Surviving request definitions, keyed by the names advertised after session context hooks. */
8081
readonly definitions?: ReadonlyMap<string, ToolDefinition>
8182
}) => Effect.Effect<NormalizedResult, Tool.Error>
@@ -276,9 +277,7 @@ const layer = Layer.effect(
276277
const codeModeTool = codeModeEnabled
277278
? CodeModeTool.create(
278279
codeModeInventory,
279-
// A child call cannot settle the outer CodeMode call with its own failure.
280-
(name, tool, input, context) =>
281-
run(tool, name, input, context).pipe(Effect.provideService(ToolInterrupted.Slot, undefined)),
280+
(name, tool, input, context) => run(tool, name, input, context),
282281
permission,
283282
flag,
284283
)
@@ -310,6 +309,7 @@ const layer = Layer.effect(
310309
messageID: input.messageID,
311310
id: Tool.CallID.make(input.call.id),
312311
progress: input.progress ?? (() => Effect.void),
312+
cancel: input.cancel,
313313
}
314314
const event = yield* beforeExecute(input.call.name, input.call.input, context)
315315
const requested = input.definitions?.get(event.tool)

‎packages/core/src/tool/interrupted.ts‎

Lines changed: 0 additions & 23 deletions
This file was deleted.

0 commit comments

Comments
 (0)