feat(session): report interrupt result (#44766)

This commit is contained in:
Kit Langton
2026-08-24 17:34:37 -04:00
committed by GitHub
parent 7f5ea1889c
commit 34bd7c220c
22 changed files with 109 additions and 48 deletions
+5 -1
View File
@@ -37,7 +37,11 @@ export function createActiveComposerAdapter(input: {
current: () => data.session.get(id),
admitted: (messageID) => data.session.input.has(id, messageID) || !!data.session.message.get(id, messageID),
}),
interrupt: () => server.api.session.interrupt({ sessionID: id, continue: true }).catch(() => undefined),
interrupt: () =>
server.api.session
.interrupt({ sessionID: id, continue: true })
.then(() => undefined)
.catch(() => undefined),
}
return adapter
}
+2 -1
View File
@@ -601,6 +601,7 @@ describe("acp event behavior", () => {
},
onInterrupt({ sessionID, send }) {
send(durableEvent("session.execution.interrupted", { sessionID, reason: "user" }))
return true
},
})
const result = streamTurn({
@@ -624,7 +625,7 @@ describe("acp event behavior", () => {
await withTimeout(submitted.promise, "cancel test prompt was not admitted")
control.cancelled = true
control.admission.abort()
await fixture.client.session.interrupt({ sessionID: "ses_cancel" })
expect(await fixture.client.session.interrupt({ sessionID: "ses_cancel" })).toEqual({ interrupted: true })
const response = await withTimeout(result, "cancelled turn did not terminate")
expect(response).toMatchObject({ stopReason: "cancelled" })
+4 -3
View File
@@ -20,7 +20,7 @@ type FixtureOptions = {
readonly onInterrupt?: (input: {
readonly sessionID: string
readonly send: (event: unknown) => void
}) => void | Promise<void>
}) => boolean | Promise<boolean>
readonly onPermissionReply?: (input: {
readonly sessionID: string
readonly requestID: string
@@ -152,8 +152,9 @@ export function createSseFixture(options: FixtureOptions = {}) {
const interrupt = /^\/api\/session\/([^/]+)\/interrupt$/.exec(url.pathname)
if (interrupt?.[1]) {
await options.onInterrupt?.({ sessionID: decodeURIComponent(interrupt[1]), send })
return new Response(null, { status: 204 })
const interrupted =
(await options.onInterrupt?.({ sessionID: decodeURIComponent(interrupt[1]), send })) ?? false
return Response.json({ interrupted })
}
return new Response(null, { status: 404 })
+1 -1
View File
@@ -998,7 +998,7 @@ export type SessionLogOutput =
export type SessionLogOperation<E = never> = (input: SessionLogInput) => Stream.Stream<SessionLogOutput, E>
export type SessionInterruptInput = { readonly sessionID: Session.ID; readonly continue?: boolean | undefined }
export type SessionInterruptOutput = void
export type SessionInterruptOutput = { readonly interrupted: boolean }
export type SessionInterruptOperation<E = never> = (
input: SessionInterruptInput,
) => Effect.Effect<SessionInterruptOutput, E>
@@ -878,9 +878,9 @@ export function make(options: ClientOptions) {
method: "POST",
path: `/api/session/${encodeURIComponent(input.sessionID)}/interrupt`,
query: { continue: input["continue"] },
successStatus: 204,
successStatus: 200,
declaredStatuses: [404, 400, 401],
empty: true,
empty: false,
},
requestOptions,
),
@@ -73,6 +73,8 @@ export type ToolFileContent1 = { type: "file"; uri: string; mime: string; name?:
export type EventLogSynced = { type: "log.synced"; aggregateID: string; seq?: number }
export type SessionInterruptResponse = { interrupted: boolean }
export type ModelReasoningField = "reasoning" | "reasoning_content" | "reasoning_text" | (string & {})
export type ModelMaxTokensField = "max_completion_tokens" | "max_tokens"
@@ -3956,7 +3958,7 @@ export type SessionInterruptInput = {
readonly continue?: { readonly continue?: boolean | undefined }["continue"]
}
export type SessionInterruptOutput = void
export type SessionInterruptOutput = SessionInterruptResponse
export type SessionBackgroundInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] }
+11 -3
View File
@@ -172,7 +172,14 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(session)))
}
if (request.method === "POST") {
return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(null, { status: 204 })))
return Effect.succeed(
HttpClientResponse.fromWeb(
request,
request.url.includes("/interrupt")
? Response.json({ interrupted: true })
: new Response(null, { status: 204 }),
),
)
}
return Effect.succeed(
HttpClientResponse.fromWeb(request, Response.json({ data: [session.data], cursor: { next: "next" } })),
@@ -202,12 +209,12 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
const log = yield* client.session
.log({ sessionID: Session.ID.make("ses_test"), after: Event.Seq.make(0) })
.pipe(Stream.runCollect)
yield* client.session.interrupt({ sessionID: Session.ID.make("ses_test") })
const interrupted = yield* client.session.interrupt({ sessionID: Session.ID.make("ses_test") })
const message = yield* client.session.message({
sessionID: Session.ID.make("ses_test"),
messageID: SessionMessage.ID.make("msg_model"),
})
return { page, active, created, admitted, context, log, message }
return { page, active, created, admitted, context, log, interrupted, message }
}).pipe(Effect.provideService(HttpClient.HttpClient, httpClient), Effect.runPromise)
const listed = result.page.data[0]
@@ -216,6 +223,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
expect(DateTime.toEpochMillis(listed.time.idle)).toBe(1_717_171_717_002)
expect(DateTime.toEpochMillis(listed.time.viewed)).toBe(1_717_171_717_001)
expect(result.active).toEqual({ ses_test: { type: "running" } })
expect(result.interrupted).toEqual({ interrupted: true })
expect(Object.getPrototypeOf(result.page.data[0])).toBe(Object.prototype)
expect(Object.getPrototypeOf(result.created)).toBe(Object.prototype)
expect(result.created.id).toBe("ses_test")
+3 -1
View File
@@ -532,6 +532,7 @@ test("session methods use the public HTTP contract", async () => {
if (url.includes("/message/")) return Response.json({ data: modelSwitchedMessage })
if (url.endsWith("/api/session/active")) return Response.json({ data: { ses_test: { type: "running" } } })
if (init?.method === "POST" && url.endsWith("/api/session")) return Response.json(session)
if (url.includes("/interrupt")) return Response.json({ interrupted: true })
if (init?.method === "POST") return new Response(null, { status: 204 })
return Response.json({ data: [session.data], cursor: { next: "next" } })
},
@@ -563,7 +564,7 @@ test("session methods use the public HTTP contract", async () => {
const context = await client.session.context({ sessionID: "ses_test" })
const log = []
for await (const item of client.session.log({ sessionID: "ses_test", after: 0 })) log.push(item)
await client.session.interrupt({ sessionID: "ses_test", continue: true })
const interrupted = await client.session.interrupt({ sessionID: "ses_test", continue: true })
const message = await client.session.message({ sessionID: "ses_test", messageID: "msg_model" })
expect(page.cursor.next).toBe("next")
@@ -572,6 +573,7 @@ test("session methods use the public HTTP contract", async () => {
expect(created.id).toBe("ses_test")
expect(admitted.id).toBe("msg_test")
expect(generated.text).toBe("A transient answer")
expect(interrupted).toEqual({ interrupted: true })
expect(synthetic).toMatchObject({ type: "synthetic", data: { text: "Completed" }, delivery: "queue" })
expect(context).toEqual([])
expect(log).toEqual([modelSwitchedEvent, synced])
+4 -1
View File
@@ -402,7 +402,10 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: Interface, p
command: runtime.session.command,
rename: runtime.session.rename,
synthetic: runtime.session.synthetic,
interrupt: (input) => runtime.session.interrupt(input.sessionID),
interrupt: (input) =>
runtime.session
.interrupt(input.sessionID, { continue: input.continue })
.pipe(Effect.map((interrupted) => ({ interrupted }))),
wait: (input) => runtime.session.wait(input.sessionID),
},
} satisfies Plugin.Context
+1 -1
View File
@@ -272,7 +272,7 @@ export interface Interface {
readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
readonly background: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<void>
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<boolean>
readonly synthetic: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
+8 -6
View File
@@ -24,9 +24,10 @@ export interface Interface {
/**
* Interrupt active work owned by this process. Idle interruption is a no-op. Resolves once
* the interruption is accepted; cleanup settles asynchronously in the execution fiber.
* Compose with `awaitIdle` when settlement matters.
* Returns whether an active execution was interrupted. Compose with `awaitIdle` when
* settlement matters.
*/
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<void>
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<boolean>
/** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */
readonly awaitIdle: (sessionID: SessionSchema.ID) => Effect.Effect<void>
}
@@ -140,8 +141,8 @@ export const layer = Layer.effect(
active: coordinator.active,
interrupt: (sessionID, options) =>
Effect.gen(function* () {
yield* coordinator.interrupt(sessionID, "user")
if (!options?.continue) return
const interrupted = yield* coordinator.interrupt(sessionID, "user")
if (!options?.continue) return interrupted
// Resume steering input and between-turn control work from the interrupted
// intent. Queued next-turn prompts stay parked: a steer-scoped drain never
// promotes them, and a control item behind a queued prompt waits its turn.
@@ -151,9 +152,10 @@ export const layer = Layer.effect(
// rows inside uninterruptible publications, so a steer row is either still
// promotable here or was fully delivered and needs no resumption.
const next = yield* SessionInbox.nextPromotable(db, sessionID, "input")
if (next === undefined) return
if (next === undefined) return interrupted
if (next.delivery === "steer" || next.type === "compaction" || next.type === "move")
yield* coordinator.wake(sessionID, "steer")
return interrupted
}),
resume: coordinator.run,
wake: coordinator.wake,
@@ -175,7 +177,7 @@ export const noopLayer = Layer.succeed(
active: Effect.succeed(new Set()),
resume: () => Effect.void,
wake: () => Effect.void,
interrupt: () => Effect.void,
interrupt: () => Effect.succeed(false),
awaitIdle: () => Effect.void,
}),
)
+8 -7
View File
@@ -14,9 +14,10 @@ export interface Coordinator<Key, E, Reason = never> {
/**
* Stops the active execution and clears its doorbell. No-op when idle. Resolves once the
* interruption is accepted, not when cleanup settles: the execution fiber finishes its
* finalizers and settled hook on its own time. Compose with `awaitIdle` for settlement.
* finalizers and settled hook on its own time. Returns whether an active execution was
* interrupted. Compose with `awaitIdle` for settlement.
*/
readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<boolean>
/** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
readonly awaitIdle: (key: Key) => Effect.Effect<void>
}
@@ -134,16 +135,16 @@ export const make = <Key, E, Reason = never>(options: {
start(key, false, scope)
})
const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
Effect.suspend(() => {
const interrupt = (key: Key, reason?: Reason): Effect.Effect<boolean> =>
Effect.sync(() => {
const execution = executions.get(key)
if (execution === undefined || execution.stopping) return Effect.void
if (execution === undefined || execution.stopping) return false
if (execution.owner === undefined) {
// Settlement window: the owner exited but the settled hook has not finished. The
// terminal outcome is already decided, so no reason attaches — but the interrupt
// still claims the recorded wakes so settle does not start a dead-intent successor.
execution.pendingWake = undefined
return Effect.void
return false
}
execution.stopping = true
// Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
@@ -153,7 +154,7 @@ export const make = <Key, E, Reason = never>(options: {
// Fire and forget: nobody benefits from waiting out cleanup here, and callers like
// the interrupt endpoint must acknowledge immediately even when finalizers are slow.
fork(Fiber.interrupt(execution.owner))
return Effect.void
return true
})
// One execution's `done` already spans coalesced continuations; re-check after it
+5 -3
View File
@@ -121,7 +121,7 @@ describe("fromPromise", () => {
}),
)
it.effect("preserves no-content and rejected Promise behavior", () =>
it.effect("preserves interrupt results and rejected Promise behavior", () =>
Effect.gen(function* () {
const seen: unknown[] = []
const host = testHost({
@@ -131,7 +131,7 @@ describe("fromPromise", () => {
return Effect.fail(new Error("interrupt failed"))
}
expect(input.continue).toBe(true)
return Effect.void
return Effect.succeed({ interrupted: false })
},
switchAgent: (input) => Effect.sync(() => seen.push(input)),
switchModel: (input) => Effect.sync(() => seen.push(input)),
@@ -144,7 +144,9 @@ describe("fromPromise", () => {
define({
id: "promise-session-interrupt",
setup: async (ctx) => {
expect(await ctx.session.interrupt({ sessionID: "ses_success", continue: true })).toBeUndefined()
expect(await ctx.session.interrupt({ sessionID: "ses_success", continue: true })).toEqual({
interrupted: false,
})
await expect(ctx.session.interrupt({ sessionID: "ses_failure" })).rejects.toThrow("interrupt failed")
expect(await ctx.session.switchAgent({ sessionID: "ses_success", agent: "build" })).toBeUndefined()
expect(
+14 -1
View File
@@ -128,12 +128,25 @@ describe("SessionExecution lifecycle", () => {
yield* Deferred.await(draining)
expect((yield* claims(database))[sessionID]).toBe(true)
yield* execution.interrupt(sessionID)
expect(yield* execution.interrupt(sessionID)).toBeTrue()
yield* execution.awaitIdle(sessionID)
expect((yield* claims(database))[sessionID]).toBe(false)
}),
)
it.effect("reports an idle interrupt as a no-op", () =>
Effect.gen(function* () {
const sessionID = Session.ID.make("ses_idle_cancel")
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, () => Effect.never)
const execution = Context.get(context, SessionExecution.Service)
expect(yield* execution.interrupt(sessionID)).toBeFalse()
expect(yield* execution.active).not.toContain(sessionID)
}),
)
it.effect("starts every claimed execution without waiting for earlier drains to finish", () =>
Effect.gen(function* () {
const database = yield* Database.Service
+2 -1
View File
@@ -48,6 +48,7 @@ const execution = Layer.succeed(
Effect.sync(() => {
interruptCalls.push(sessionID)
interruptContinuations.push(options?.continue)
return activeSessions.delete(sessionID)
}),
wake: (sessionID) =>
Effect.sync(() => {
@@ -193,7 +194,7 @@ describe("Session.prompt", () => {
interruptCalls.length = 0
wakeCalls.length = 0
yield* session.interrupt(sessionID)
expect(yield* session.interrupt(sessionID)).toBeFalse()
expect(interruptCalls).toEqual([sessionID])
expect(wakeCalls).toEqual([])
expect(yield* session.messages({ sessionID })).toEqual([])
@@ -236,7 +236,7 @@ describe("SessionRunCoordinator", () => {
drain: () => Effect.void,
settled: (_key, _exit, reason) => Effect.sync(() => void reasons.push(reason)),
})
yield* coordinator.interrupt("session", "user")
expect(yield* coordinator.interrupt("session", "user")).toBeFalse()
yield* coordinator.run("session")
expect(reasons).toEqual([undefined])
}),
@@ -260,7 +260,7 @@ describe("SessionRunCoordinator", () => {
const run = yield* coordinator.run("session").pipe(Effect.forkChild)
yield* Deferred.await(settling)
yield* coordinator.interrupt("session", "user")
expect(yield* coordinator.interrupt("session", "user")).toBeFalse()
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(run)
yield* coordinator.run("session")
@@ -315,7 +315,7 @@ describe("SessionRunCoordinator", () => {
const idle = yield* coordinator.awaitIdle("session").pipe(Effect.forkChild)
yield* Effect.yieldNow
yield* coordinator.wake("session")
yield* coordinator.interrupt("session", "user")
expect(yield* coordinator.interrupt("session", "user")).toBeTrue()
yield* Deferred.await(interrupted)
const exits = yield* Fiber.awaitAll([first, second, idle])
+1 -1
View File
@@ -115,7 +115,7 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()),
resume: complete,
wake: () => Effect.void,
interrupt: () => Effect.void,
interrupt: () => Effect.succeed(false),
awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid),
})
}),
+1 -1
View File
@@ -88,7 +88,7 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()),
resume: complete,
wake: () => Effect.void,
interrupt: () => Effect.void,
interrupt: () => Effect.succeed(false),
awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
})
}),
+21 -3
View File
@@ -3753,8 +3753,15 @@
],
"security": [],
"responses": {
"204": {
"description": "<No Content>"
"200": {
"description": "SessionInterruptResponse",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SessionInterruptResponse"
}
}
}
},
"400": {
"description": "InvalidRequestError",
@@ -3794,7 +3801,7 @@
}
}
},
"description": "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input and next-in-line control items (manual compaction, moves) while queued prompts remain parked.",
"description": "Interrupt active execution owned by this OpenCode process. Returns interrupted=true when an active execution was interrupted and false for the idle no-op. When continue=true, execution resumes pending steering input and next-in-line control items (manual compaction, moves) while queued prompts remain parked.",
"summary": "Interrupt session execution"
}
},
@@ -16411,6 +16418,17 @@
"required": ["data"],
"additionalProperties": false
},
"SessionInterruptResponse": {
"type": "object",
"properties": {
"interrupted": {
"type": "boolean",
"description": "Whether an active execution owned by this OpenCode process was interrupted."
}
},
"required": ["interrupted"],
"additionalProperties": false
},
"SessionLogItemEncoded": {
"type": "string",
"contentMediaType": "application/json"
+6 -2
View File
@@ -664,7 +664,11 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
HttpApiEndpoint.post("session.interrupt", "/api/session/:sessionID/interrupt", {
params: { sessionID: Session.ID },
query: { continue: BooleanFromString.pipe(Schema.optional) },
success: HttpApiSchema.NoContent,
success: Schema.Struct({
interrupted: Schema.Boolean.annotate({
description: "Whether an active execution owned by this OpenCode process was interrupted.",
}),
}).annotate({ identifier: "SessionInterruptResponse" }),
error: SessionNotFoundError,
})
.middleware(sessionLocationMiddleware)
@@ -673,7 +677,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
identifier: "v2.session.interrupt",
summary: "Interrupt session execution",
description:
"Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input and next-in-line control items (manual compaction, moves) while queued prompts remain parked.",
"Interrupt active execution owned by this OpenCode process. Returns interrupted=true when an active execution was interrupted and false for the idle no-op. When continue=true, execution resumes pending steering input and next-in-line control items (manual compaction, moves) while queued prompts remain parked.",
}),
),
)
+1 -2
View File
@@ -606,8 +606,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
.handle(
"session.interrupt",
Effect.fn(function* (ctx) {
yield* session.interrupt(ctx.params.sessionID, { continue: ctx.query.continue })
return HttpApiSchema.NoContent.make()
return { interrupted: yield* session.interrupt(ctx.params.sessionID, { continue: ctx.query.continue }) }
}),
)
.handle(
@@ -1493,7 +1493,7 @@ describe("V2 mini transport", () => {
files: [],
includeFiles: true,
})
const interrupt = spyOn(second.session, "interrupt").mockImplementation(() => ok(undefined))
const interrupt = spyOn(second.session, "interrupt").mockImplementation(() => ok({ interrupted: true }))
await transport.interruptActiveTurn()
expect(prompt).toHaveBeenCalled()
@@ -2363,7 +2363,7 @@ describe("V2 mini transport", () => {
admitted = true
return ok({ data: promptAdmission(request) })
})
const interrupted = spyOn(client.session, "interrupt").mockImplementation(() => ok(undefined))
const interrupted = spyOn(client.session, "interrupt").mockImplementation(() => ok({ interrupted: true }))
const controller = new AbortController()
const turn = transport.runPromptTurn({
agent: undefined,
@@ -2500,7 +2500,7 @@ describe("V2 mini transport", () => {
})
}) as never,
)
const interrupted = spyOn(client.session, "interrupt").mockImplementation(() => ok(undefined))
const interrupted = spyOn(client.session, "interrupt").mockImplementation(() => ok({ interrupted: true }))
const turn = transport.runPromptTurn({
agent: undefined,