From 5da7eaa6eb8c5d3c81796ed7f086b41a113dd4f8 Mon Sep 17 00:00:00 2001 From: Dax Date: Tue, 11 Aug 2026 17:56:54 -0700 Subject: [PATCH] feat(api): continue pending work after interrupt (#41888) --- packages/client/src/effect/api/api.ts | 2 +- .../client/src/effect/generated/client.ts | 5 ++- .../client/src/promise/generated/client.ts | 1 + .../client/src/promise/generated/types.ts | 5 ++- packages/client/test/promise.test.ts | 4 +-- packages/core/src/session.ts | 11 +++++-- packages/core/test/session-prompt.test.ts | 31 +++++++++++++++++++ packages/protocol/src/groups/session.ts | 4 ++- packages/server/src/handlers/session.ts | 2 +- packages/tui/src/mini/runtime.ts | 2 +- packages/tui/src/mini/stream-v2.transport.ts | 4 +-- .../tui/test/mini/stream-v2.transport.test.ts | 4 +-- 12 files changed, 61 insertions(+), 14 deletions(-) diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 7691c2cde68..43c52806a33 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -915,7 +915,7 @@ export type Endpoint5_31Output = | EventLog.Synced export type SessionLogOperation = (input: Endpoint5_31Input) => Stream.Stream -export type Endpoint5_32Input = { readonly sessionID: Session.ID } +export type Endpoint5_32Input = { readonly sessionID: Session.ID; readonly continue?: boolean | undefined } export type Endpoint5_32Output = void export type SessionInterruptOperation = (input: Endpoint5_32Input) => Effect.Effect diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index cc334c97096..e2f35207de0 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -596,7 +596,10 @@ const Endpoint5_31 = (raw: RawClient["server.session"]) => (input: Endpoint5_31I const Endpoint5_32 = (raw: RawClient["server.session"]) => (input: Endpoint5_32Input) => preserveEffect()( - raw["session.interrupt"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)), + raw["session.interrupt"]({ + params: { sessionID: input["sessionID"] }, + query: { continue: input["continue"] }, + }).pipe(Effect.mapError(mapClientError)), ) const Endpoint5_33 = (raw: RawClient["server.session"]) => (input: Endpoint5_33Input) => diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index fbba0318823..de9dfe492b9 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -875,6 +875,7 @@ export function make(options: ClientOptions) { { method: "POST", path: `/api/session/${encodeURIComponent(input.sessionID)}/interrupt`, + query: { continue: input["continue"] }, successStatus: 204, declaredStatuses: [404, 400, 401], empty: true, diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 4625160b933..c5308381abd 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -3892,7 +3892,10 @@ export type SessionLogInput = { export type SessionLogOutput = SessionLogItem -export type SessionInterruptInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } +export type SessionInterruptInput = { + readonly sessionID: { readonly sessionID: string }["sessionID"] + readonly continue?: { readonly continue?: boolean | undefined }["continue"] +} export type SessionInterruptOutput = void diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 49368333754..82da4e9b466 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -543,7 +543,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" }) + 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") @@ -568,7 +568,7 @@ test("session methods use the public HTTP contract", async () => { ["POST", "http://localhost:3000/api/session/ses_test/wait"], ["GET", "http://localhost:3000/api/session/ses_test/context"], ["GET", "http://localhost:3000/api/experimental/session/ses_test/log?after=0"], - ["POST", "http://localhost:3000/api/session/ses_test/interrupt"], + ["POST", "http://localhost:3000/api/session/ses_test/interrupt?continue=true"], ["GET", "http://localhost:3000/api/session/ses_test/message/msg_model"], ]) const body = requests.find((request) => request.url.endsWith("/api/session/ses_test/prompt"))?.init?.body diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 9a44369d81c..bae924f9f43 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -267,7 +267,7 @@ export interface Interface { readonly active: Effect.Effect> readonly background: (sessionID: SessionSchema.ID) => Effect.Effect readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect - readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect + readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect readonly synthetic: (input: { id?: SessionMessage.ID sessionID: SessionSchema.ID @@ -835,7 +835,14 @@ const layer = Layer.effect( }), ), ), - interrupt: Effect.fn("Session.interrupt")((sessionID) => Effect.uninterruptible(execution.interrupt(sessionID))), + interrupt: Effect.fn("Session.interrupt")((sessionID, options) => + Effect.uninterruptible( + Effect.gen(function* () { + yield* execution.interrupt(sessionID) + if (options?.continue && (yield* SessionPending.has(db, sessionID, "any"))) yield* execution.wake(sessionID) + }), + ), + ), revert: { stage: Effect.fn("Session.revert.stage")(function* (input) { const session = yield* result.get(input.sessionID) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 7598d487e44..51059a2b237 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -168,13 +168,44 @@ describe("Session.prompt", () => { yield* setup const session = yield* Session.Service interruptCalls.length = 0 + wakeCalls.length = 0 yield* session.interrupt(sessionID) expect(interruptCalls).toEqual([sessionID]) + expect(wakeCalls).toEqual([]) expect(yield* session.messages({ sessionID })).toEqual([]) }), ) + it.effect("continues after interruption when pending work remains", () => + Effect.gen(function* () { + yield* setup + const session = yield* Session.Service + yield* session.synthetic({ sessionID, text: "Continue after interrupt", resume: false }) + interruptCalls.length = 0 + wakeCalls.length = 0 + + yield* session.interrupt(sessionID, { continue: true }) + + expect(interruptCalls).toEqual([sessionID]) + expect(wakeCalls).toEqual([sessionID]) + }), + ) + + it.effect("does not continue after interruption without pending work", () => + Effect.gen(function* () { + yield* setup + const session = yield* Session.Service + interruptCalls.length = 0 + wakeCalls.length = 0 + + yield* session.interrupt(sessionID, { continue: true }) + + expect(interruptCalls).toEqual([sessionID]) + expect(wakeCalls).toEqual([]) + }), + ) + it.effect("delegates interruption without requiring a recorded Session", () => Effect.gen(function* () { const session = yield* Session.Service diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 543442039f9..d6b8a328c2b 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -647,6 +647,7 @@ export const makeSessionGroup = (sessionLo .add( HttpApiEndpoint.post("session.interrupt", "/api/session/:sessionID/interrupt", { params: { sessionID: Session.ID }, + query: { continue: BooleanFromString.pipe(Schema.optional) }, success: HttpApiSchema.NoContent, error: SessionNotFoundError, }) @@ -655,7 +656,8 @@ export const makeSessionGroup = (sessionLo OpenApi.annotations({ identifier: "v2.session.interrupt", summary: "Interrupt session execution", - description: "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op.", + description: + "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes if durable pending work remains after interruption.", }), ), ) diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index 8691e480c12..39eba149be5 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -772,7 +772,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.interrupt", Effect.fn(function* (ctx) { - yield* session.interrupt(ctx.params.sessionID) + yield* session.interrupt(ctx.params.sessionID, { continue: ctx.query.continue }) return HttpApiSchema.NoContent.make() }), ) diff --git a/packages/tui/src/mini/runtime.ts b/packages/tui/src/mini/runtime.ts index 92b21476de9..b6053ee242e 100644 --- a/packages/tui/src/mini/runtime.ts +++ b/packages/tui/src/mini/runtime.ts @@ -374,7 +374,7 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep void ( state.stream ? state.stream.then((item) => item.handle.interruptActiveTurn()) - : state.sdk.session.interrupt({ sessionID: state.sessionID }) + : state.sdk.session.interrupt({ sessionID: state.sessionID, continue: true }) ) .catch(() => {}) .finally(() => { diff --git a/packages/tui/src/mini/stream-v2.transport.ts b/packages/tui/src/mini/stream-v2.transport.ts index 438993cf545..5bde6877f84 100644 --- a/packages/tui/src/mini/stream-v2.transport.ts +++ b/packages/tui/src/mini/stream-v2.transport.ts @@ -1528,7 +1528,7 @@ export async function createSessionTransport(input: StreamInput): Promise { active.interrupted = true - void sdk.session.interrupt({ sessionID: input.sessionID }).catch(() => {}) + void sdk.session.interrupt({ sessionID: input.sessionID, continue: true }).catch(() => {}) } next.signal?.addEventListener("abort", interrupt, { once: true }) try { @@ -1786,7 +1786,7 @@ export async function createSessionTransport(input: StreamInput): Promise {}) + await sdk.session.interrupt({ sessionID: input.sessionID, continue: true }).catch(() => {}) }, selectSubagent(sessionID) { subagents.select(sdk, sessionID) diff --git a/packages/tui/test/mini/stream-v2.transport.test.ts b/packages/tui/test/mini/stream-v2.transport.test.ts index 03f8b1322da..394e84b7a56 100644 --- a/packages/tui/test/mini/stream-v2.transport.test.ts +++ b/packages/tui/test/mini/stream-v2.transport.test.ts @@ -1497,7 +1497,7 @@ describe("V2 mini transport", () => { await transport.interruptActiveTurn() expect(prompt).toHaveBeenCalled() - expect(interrupt).toHaveBeenCalledWith({ sessionID: "ses_1" }) + expect(interrupt).toHaveBeenCalledWith({ sessionID: "ses_1", continue: true }) expect(firstPrompt).not.toHaveBeenCalled() expect(firstInterrupt).not.toHaveBeenCalled() await transport.close() @@ -2390,7 +2390,7 @@ describe("V2 mini transport", () => { idle.resolve() await turn - expect(interrupted).toHaveBeenCalledWith({ sessionID: "ses_1" }) + expect(interrupted).toHaveBeenCalledWith({ sessionID: "ses_1", continue: true }) await transport.close() })