fix(session): harden shell cancellation (#24553)
This commit is contained in:
Kit Langton
@@ -45,7 +45,7 @@ import { AppFileSystem } from "@opencode-ai/core/filesystem"
|
||||
import { Truncate } from "@/tool/truncate"
|
||||
import { decodeDataUrl } from "@/util/data-url"
|
||||
import { Process } from "@/util/process"
|
||||
import { Cause, Effect, Exit, Layer, Option, Scope, Context, Schema } from "effect"
|
||||
import { Cause, Effect, Exit, Latch, Layer, Option, Scope, Context, Schema } from "effect"
|
||||
import { zod } from "@/util/effect-zod"
|
||||
import { withStatics } from "@/util/schema"
|
||||
import * as EffectLogger from "@opencode-ai/core/effect/logger"
|
||||
@@ -720,143 +720,145 @@ NOTE: At any point in time through this workflow you should feel free to ask the
|
||||
} satisfies MessageV2.TextPart)
|
||||
})
|
||||
|
||||
const shellImpl = Effect.fn("SessionPrompt.shellImpl")(function* (input: ShellInput) {
|
||||
const ctx = yield* InstanceState.context
|
||||
const run = yield* runner()
|
||||
const session = yield* sessions.get(input.sessionID)
|
||||
if (session.revert) {
|
||||
yield* revert.cleanup(session)
|
||||
}
|
||||
const agent = yield* agents.get(input.agent)
|
||||
if (!agent) {
|
||||
const available = (yield* agents.list()).filter((a) => !a.hidden).map((a) => a.name)
|
||||
const hint = available.length ? ` Available agents: ${available.join(", ")}` : ""
|
||||
const error = new NamedError.Unknown({ message: `Agent not found: "${input.agent}".${hint}` })
|
||||
yield* bus.publish(Session.Event.Error, { sessionID: input.sessionID, error: error.toObject() })
|
||||
throw error
|
||||
}
|
||||
const model = input.model ?? agent.model ?? (yield* lastModel(input.sessionID))
|
||||
const userMsg: MessageV2.User = {
|
||||
id: input.messageID ?? MessageID.ascending(),
|
||||
sessionID: input.sessionID,
|
||||
time: { created: Date.now() },
|
||||
role: "user",
|
||||
agent: input.agent,
|
||||
model: { providerID: model.providerID, modelID: model.modelID },
|
||||
}
|
||||
yield* sessions.updateMessage(userMsg)
|
||||
const userPart: MessageV2.Part = {
|
||||
type: "text",
|
||||
id: PartID.ascending(),
|
||||
messageID: userMsg.id,
|
||||
sessionID: input.sessionID,
|
||||
text: "The following tool was executed by the user",
|
||||
synthetic: true,
|
||||
}
|
||||
yield* sessions.updatePart(userPart)
|
||||
|
||||
const msg: MessageV2.Assistant = {
|
||||
id: MessageID.ascending(),
|
||||
sessionID: input.sessionID,
|
||||
parentID: userMsg.id,
|
||||
mode: input.agent,
|
||||
agent: input.agent,
|
||||
cost: 0,
|
||||
path: { cwd: ctx.directory, root: ctx.worktree },
|
||||
time: { created: Date.now() },
|
||||
role: "assistant",
|
||||
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
|
||||
modelID: model.modelID,
|
||||
providerID: model.providerID,
|
||||
}
|
||||
yield* sessions.updateMessage(msg)
|
||||
const part: MessageV2.ToolPart = {
|
||||
type: "tool",
|
||||
id: PartID.ascending(),
|
||||
messageID: msg.id,
|
||||
sessionID: input.sessionID,
|
||||
tool: "bash",
|
||||
callID: ulid(),
|
||||
state: {
|
||||
status: "running",
|
||||
time: { start: Date.now() },
|
||||
input: { command: input.command },
|
||||
},
|
||||
}
|
||||
yield* sessions.updatePart(part)
|
||||
|
||||
const cfg = yield* config.get()
|
||||
const sh = Shell.preferred(cfg.shell)
|
||||
const cwd = ctx.directory
|
||||
const args = Shell.args(sh, input.command, cwd)
|
||||
const shellEnv = yield* plugin.trigger(
|
||||
"shell.env",
|
||||
{ cwd, sessionID: input.sessionID, callID: part.callID },
|
||||
{ env: {} },
|
||||
)
|
||||
|
||||
const cmd = ChildProcess.make(sh, args, {
|
||||
cwd,
|
||||
extendEnv: true,
|
||||
env: { ...shellEnv.env, TERM: "dumb" },
|
||||
stdin: "ignore",
|
||||
forceKillAfter: "3 seconds",
|
||||
})
|
||||
|
||||
let output = ""
|
||||
let aborted = false
|
||||
const finish = Effect.uninterruptible(
|
||||
const shellImpl = Effect.fn("SessionPrompt.shellImpl")(function* (input: ShellInput, ready?: Latch.Latch) {
|
||||
return yield* Effect.uninterruptibleMask((restore) =>
|
||||
Effect.gen(function* () {
|
||||
if (aborted) {
|
||||
output += "\n\n" + ["<metadata>", "User aborted the command", "</metadata>"].join("\n")
|
||||
}
|
||||
if (!msg.time.completed) {
|
||||
msg.time.completed = Date.now()
|
||||
const markReady = ready ? ready.open.pipe(Effect.asVoid) : Effect.void
|
||||
const { msg, part, cwd } = yield* Effect.gen(function* () {
|
||||
const ctx = yield* InstanceState.context
|
||||
const session = yield* sessions.get(input.sessionID)
|
||||
if (session.revert) {
|
||||
yield* revert.cleanup(session)
|
||||
}
|
||||
const agent = yield* agents.get(input.agent)
|
||||
if (!agent) {
|
||||
const available = (yield* agents.list()).filter((a) => !a.hidden).map((a) => a.name)
|
||||
const hint = available.length ? ` Available agents: ${available.join(", ")}` : ""
|
||||
const error = new NamedError.Unknown({ message: `Agent not found: "${input.agent}".${hint}` })
|
||||
yield* bus.publish(Session.Event.Error, { sessionID: input.sessionID, error: error.toObject() })
|
||||
throw error
|
||||
}
|
||||
const model = input.model ?? agent.model ?? (yield* lastModel(input.sessionID))
|
||||
const userMsg: MessageV2.User = {
|
||||
id: input.messageID ?? MessageID.ascending(),
|
||||
sessionID: input.sessionID,
|
||||
time: { created: Date.now() },
|
||||
role: "user",
|
||||
agent: input.agent,
|
||||
model: { providerID: model.providerID, modelID: model.modelID },
|
||||
}
|
||||
yield* sessions.updateMessage(userMsg)
|
||||
const userPart: MessageV2.Part = {
|
||||
type: "text",
|
||||
id: PartID.ascending(),
|
||||
messageID: userMsg.id,
|
||||
sessionID: input.sessionID,
|
||||
text: "The following tool was executed by the user",
|
||||
synthetic: true,
|
||||
}
|
||||
yield* sessions.updatePart(userPart)
|
||||
|
||||
const msg: MessageV2.Assistant = {
|
||||
id: MessageID.ascending(),
|
||||
sessionID: input.sessionID,
|
||||
parentID: userMsg.id,
|
||||
mode: input.agent,
|
||||
agent: input.agent,
|
||||
cost: 0,
|
||||
path: { cwd: ctx.directory, root: ctx.worktree },
|
||||
time: { created: Date.now() },
|
||||
role: "assistant",
|
||||
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
|
||||
modelID: model.modelID,
|
||||
providerID: model.providerID,
|
||||
}
|
||||
yield* sessions.updateMessage(msg)
|
||||
}
|
||||
if (part.state.status === "running") {
|
||||
part.state = {
|
||||
status: "completed",
|
||||
time: { ...part.state.time, end: Date.now() },
|
||||
input: part.state.input,
|
||||
title: "",
|
||||
metadata: { output, description: "" },
|
||||
output,
|
||||
const part: MessageV2.ToolPart = {
|
||||
type: "tool",
|
||||
id: PartID.ascending(),
|
||||
messageID: msg.id,
|
||||
sessionID: input.sessionID,
|
||||
tool: "bash",
|
||||
callID: ulid(),
|
||||
state: {
|
||||
status: "running",
|
||||
time: { start: Date.now() },
|
||||
input: { command: input.command },
|
||||
},
|
||||
}
|
||||
yield* sessions.updatePart(part)
|
||||
return { msg, part, cwd: ctx.directory }
|
||||
}).pipe(Effect.ensuring(markReady))
|
||||
|
||||
const cfg = yield* config.get()
|
||||
const sh = Shell.preferred(cfg.shell)
|
||||
const args = Shell.args(sh, input.command, cwd)
|
||||
let output = ""
|
||||
let aborted = false
|
||||
|
||||
const finish = Effect.uninterruptible(
|
||||
Effect.gen(function* () {
|
||||
if (aborted) {
|
||||
output += "\n\n" + ["<metadata>", "User aborted the command", "</metadata>"].join("\n")
|
||||
}
|
||||
if (!msg.time.completed) {
|
||||
msg.time.completed = Date.now()
|
||||
yield* sessions.updateMessage(msg)
|
||||
}
|
||||
if (part.state.status === "running") {
|
||||
part.state = {
|
||||
status: "completed",
|
||||
time: { ...part.state.time, end: Date.now() },
|
||||
input: part.state.input,
|
||||
title: "",
|
||||
metadata: { output, description: "" },
|
||||
output,
|
||||
}
|
||||
yield* sessions.updatePart(part)
|
||||
}
|
||||
}),
|
||||
)
|
||||
|
||||
const exit = yield* restore(
|
||||
Effect.gen(function* () {
|
||||
const shellEnv = yield* plugin.trigger(
|
||||
"shell.env",
|
||||
{ cwd, sessionID: input.sessionID, callID: part.callID },
|
||||
{ env: {} },
|
||||
)
|
||||
const cmd = ChildProcess.make(sh, args, {
|
||||
cwd,
|
||||
extendEnv: true,
|
||||
env: { ...shellEnv.env, TERM: "dumb" },
|
||||
stdin: "ignore",
|
||||
forceKillAfter: "3 seconds",
|
||||
})
|
||||
const handle = yield* spawner.spawn(cmd)
|
||||
yield* Stream.runForEach(Stream.decodeText(handle.all), (chunk) =>
|
||||
Effect.gen(function* () {
|
||||
output += chunk
|
||||
if (part.state.status === "running") {
|
||||
part.state.metadata = { output, description: "" }
|
||||
yield* sessions.updatePart(part)
|
||||
}
|
||||
}),
|
||||
)
|
||||
yield* handle.exitCode
|
||||
}).pipe(Effect.scoped, Effect.orDie),
|
||||
).pipe(Effect.exit)
|
||||
|
||||
if (Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause) && !Cause.hasDies(exit.cause)) {
|
||||
aborted = true
|
||||
}
|
||||
yield* finish
|
||||
|
||||
if (Exit.isFailure(exit) && !aborted && !Cause.hasInterruptsOnly(exit.cause)) {
|
||||
return yield* Effect.failCause(exit.cause)
|
||||
}
|
||||
|
||||
return { info: msg, parts: [part] }
|
||||
}),
|
||||
)
|
||||
|
||||
const exit = yield* Effect.gen(function* () {
|
||||
const handle = yield* spawner.spawn(cmd)
|
||||
yield* Stream.runForEach(Stream.decodeText(handle.all), (chunk) =>
|
||||
Effect.sync(() => {
|
||||
output += chunk
|
||||
if (part.state.status === "running") {
|
||||
part.state.metadata = { output, description: "" }
|
||||
void run.fork(sessions.updatePart(part))
|
||||
}
|
||||
}),
|
||||
)
|
||||
yield* handle.exitCode
|
||||
}).pipe(
|
||||
Effect.scoped,
|
||||
Effect.onInterrupt(() =>
|
||||
Effect.sync(() => {
|
||||
aborted = true
|
||||
}),
|
||||
),
|
||||
Effect.orDie,
|
||||
Effect.ensuring(finish),
|
||||
Effect.exit,
|
||||
)
|
||||
|
||||
if (Exit.isFailure(exit) && !Cause.hasInterruptsOnly(exit.cause)) {
|
||||
return yield* Effect.failCause(exit.cause)
|
||||
}
|
||||
|
||||
return { info: msg, parts: [part] }
|
||||
})
|
||||
|
||||
const getModel = Effect.fn("SessionPrompt.getModel")(function* (
|
||||
@@ -1507,7 +1509,8 @@ NOTE: At any point in time through this workflow you should feel free to ask the
|
||||
|
||||
const shell: (input: ShellInput) => Effect.Effect<MessageV2.WithParts> = Effect.fn("SessionPrompt.shell")(
|
||||
function* (input: ShellInput) {
|
||||
return yield* state.startShell(input.sessionID, lastAssistant(input.sessionID), shellImpl(input))
|
||||
const ready = yield* Latch.make()
|
||||
return yield* state.startShell(input.sessionID, lastAssistant(input.sessionID), shellImpl(input, ready), ready)
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { InstanceState } from "@/effect/instance-state"
|
||||
import { Runner } from "@/effect/runner"
|
||||
import { Effect, Layer, Scope, Context } from "effect"
|
||||
import { Effect, Latch, Layer, Scope, Context } from "effect"
|
||||
import * as Session from "./session"
|
||||
import { MessageV2 } from "./message-v2"
|
||||
import { SessionID } from "./schema"
|
||||
@@ -18,6 +18,7 @@ export interface Interface {
|
||||
sessionID: SessionID,
|
||||
onInterrupt: Effect.Effect<MessageV2.WithParts>,
|
||||
work: Effect.Effect<MessageV2.WithParts>,
|
||||
ready?: Latch.Latch,
|
||||
) => Effect.Effect<MessageV2.WithParts>
|
||||
}
|
||||
|
||||
@@ -95,8 +96,9 @@ export const layer = Layer.effect(
|
||||
sessionID: SessionID,
|
||||
onInterrupt: Effect.Effect<MessageV2.WithParts>,
|
||||
work: Effect.Effect<MessageV2.WithParts>,
|
||||
ready?: Latch.Latch,
|
||||
) {
|
||||
return yield* (yield* runner(sessionID, onInterrupt)).startShell(work)
|
||||
return yield* (yield* runner(sessionID, onInterrupt)).startShell(work, ready)
|
||||
})
|
||||
|
||||
return Service.of({ assertNotBusy, cancel, ensureRunning, startShell })
|
||||
|
||||
Reference in New Issue
Block a user