vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-08-19 10:23:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

feat(api): continue pending work after interrupt (#41888)

Browse source
This commit is contained in:
Dax 2026-08-11 17:56:54 -07:00 • committed by GitHub
parent c254ba8a7f
commit 5da7eaa6eb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 61 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/client/src/effect/api/api.ts
View file

@ -915,7 +915,7 @@ export type Endpoint5_31Output =
| EventLog.Synced
export type SessionLogOperation<E = never> = (input: Endpoint5_31Input) => Stream.Stream<Endpoint5_31Output, E>
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<E = never> = (input: Endpoint5_32Input) => Effect.Effect<Endpoint5_32Output, E>

5
packages/client/src/effect/generated/client.ts
View file

@ -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<Endpoint5_32Output>()(
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) =>

1
packages/client/src/promise/generated/client.ts
View file

@ -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,

5
packages/client/src/promise/generated/types.ts
View file

@ -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

4
packages/client/test/promise.test.ts
View file

@ -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

11
packages/core/src/session.ts
View file

@ -267,7 +267,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) => Effect.Effect<void>
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<void>
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)

31
packages/core/test/session-prompt.test.ts
View file

@ -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

4
packages/protocol/src/groups/session.ts
View file

@ -647,6 +647,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(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 = <I extends HttpApiMiddleware.AnyId, S>(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.",
}),
),
)

2
packages/server/src/handlers/session.ts
View file

@ -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()
}),
)

2
packages/tui/src/mini/runtime.ts
View file

@ -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(() => {

4
packages/tui/src/mini/stream-v2.transport.ts
View file

@ -1528,7 +1528,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
state.wait = active
const interrupt = () => {
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<Sessio
return
}
if (state.wait) state.wait.interrupted = true
await sdk.session.interrupt({ sessionID: input.sessionID }).catch(() => {})
await sdk.session.interrupt({ sessionID: input.sessionID, continue: true }).catch(() => {})
},
selectSubagent(sessionID) {
subagents.select(sdk, sessionID)

4
packages/tui/test/mini/stream-v2.transport.test.ts
View file

@ -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()
})