diff --git a/.changeset/steer-resume-simplified.md b/.changeset/steer-resume-simplified.md new file mode 100644 index 00000000000..efd98411e2b --- /dev/null +++ b/.changeset/steer-resume-simplified.md @@ -0,0 +1,5 @@ +--- +"@opencode-ai/core": patch +--- + +Simplify interrupt continuation: the steer-scoped resume decision now lives in SessionExecution as a post-cleanup inbox check, and the run coordinator drops its continuation state machine. Wakes arriving during cancellation cleanup now restart a normal full drain, and interrupting an idle session with continue now resumes pending steering input. diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 3da6165a527..aef322abfaa 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -137,13 +137,13 @@ export const layer = Layer.effect( return Service.of({ active: coordinator.active, interrupt: (sessionID, options) => - coordinator.interrupt( - sessionID, - "user", - options?.continue - ? { continue: { request: "steer", when: SessionInbox.has(db, sessionID, "steer") } } - : undefined, - ), + Effect.gen(function* () { + yield* coordinator.interrupt(sessionID, "user") + if (!options?.continue) return + // Resume only steering input from the interrupted intent. Queued next-turn work + // stays parked: a steer-scoped drain never promotes queue-delivery rows. + if (yield* SessionInbox.has(db, sessionID, "steer")) yield* coordinator.wake(sessionID, "steer") + }), resume: coordinator.run, wake: coordinator.wake, wakeActive: coordinator.wakeActive, diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 1c9b8c2ba94..2a0d3c321f9 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -10,25 +10,19 @@ export interface Coordinator { /** Starts an execution while idle, or joins the active execution and returns its exit. */ readonly run: (key: Key) => Effect.Effect /** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */ - readonly wake: (key: Key, request?: Request) => Effect.Effect - /** Rings the current execution's doorbell with its existing request. Idle keys remain idle. */ + readonly wake: (key: Key, scope?: Promotable) => Effect.Effect + /** Rings the current execution's doorbell with its existing scope. Idle keys remain idle. */ readonly wakeActive: (key: Key) => Effect.Effect /** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */ - readonly interrupt: ( - key: Key, - reason?: Reason, - options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect } }, - ) => Effect.Effect + readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */ readonly awaitIdle: (key: Key) => Effect.Effect } -export type Request = Promotable - /** * One execution is a busy period for one key: one fiber that drains from the first wake * until the key would stay idle. `pendingWake` is the doorbell: work recorded during the - * execution rings it with its eligibility request, and the execution loop drains again + * execution rings it with the scope that work needs, and the execution loop drains again * instead of ending. The doorbell closes the gap between a drain's last eligibility check * and the idle transition, since those cannot be one atomic step. `done` resolves joiners * with this execution's exit. @@ -36,15 +30,10 @@ export type Request = Promotable type Execution = { readonly done: Deferred.Deferred owner?: Fiber.Fiber - request: Request - pendingWake?: Request + scope: Promotable + pendingWake?: Promotable stopping: boolean interruptionReason?: Reason - continuation?: { - readonly request: Request - readonly when: Effect.Effect - signaled: boolean - } } /** @@ -59,7 +48,7 @@ type Execution = { * ``` */ export const make = (options: { - readonly drain: (key: Key, force: boolean, request: Request) => Effect.Effect + readonly drain: (key: Key, force: boolean, scope: Promotable) => Effect.Effect /** Runs once when a process-local busy period begins, before its first drain. */ readonly started?: (key: Key) => Effect.Effect /** @@ -73,11 +62,11 @@ export const make = (options: { const fork = yield* FiberSet.makeRuntime() const loop = (key: Key, execution: Execution, force: boolean): Effect.Effect => - Effect.suspend(() => options.drain(key, force, execution.request)).pipe( + Effect.suspend(() => options.drain(key, force, execution.scope)).pipe( Effect.flatMap(() => Effect.suspend(() => { if (execution.stopping || execution.pendingWake === undefined) return Effect.void - execution.request = execution.pendingWake + execution.scope = execution.pendingWake execution.pendingWake = undefined // Trampoline so drains that complete synchronously cannot grow the stack. return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false))) @@ -85,10 +74,10 @@ export const make = (options: { ), ) - const start = (key: Key, force: boolean, request: Request) => { + const start = (key: Key, force: boolean, scope: Promotable) => { const execution: Execution = { done: Deferred.makeUnsafe(), - request, + scope, stopping: false, } executions.set(key, execution) @@ -104,7 +93,7 @@ export const make = (options: { execution.owner = undefined }).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)), ), - Effect.onExit((exit) => finish(key, execution, exit)), + Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))), Effect.exit, Effect.asVoid, ), @@ -114,22 +103,12 @@ export const make = (options: { // A doorbell that survives the execution loop (rung after the loop decided to end, or // during failure or interruption cleanup) starts a fresh execution for the remaining work. - const settle = (key: Key, execution: Execution, exit: Exit.Exit, resume: boolean) => { - if (resume && execution.continuation) start(key, false, execution.continuation.request) - else if (execution.pendingWake) start(key, false, execution.pendingWake) + const settle = (key: Key, execution: Execution, exit: Exit.Exit) => { + if (execution.pendingWake) start(key, false, execution.pendingWake) else executions.delete(key) Deferred.doneUnsafe(execution.done, exit) } - const finish = (key: Key, execution: Execution, exit: Exit.Exit) => { - if (!execution.continuation) return Effect.sync(() => settle(key, execution, exit, false)) - return execution.continuation.when.pipe( - Effect.flatMap((ready) => - Effect.sync(() => settle(key, execution, exit, ready || execution.continuation?.signaled === true)), - ), - ) - } - const run = (key: Key): Effect.Effect => Effect.suspend(() => { const execution = executions.get(key) @@ -141,55 +120,32 @@ export const make = (options: { return Deferred.await(start(key, true, "input").done) }) - const wake = (key: Key, request: Request = "input") => + const wake = (key: Key, scope: Promotable = "input") => Effect.sync(() => { const execution = executions.get(key) if (execution !== undefined) { - if (execution.stopping) { - if (execution.continuation) execution.continuation.signaled = true - else execution.continuation = { request, when: Effect.succeed(true), signaled: true } - return - } - // Coalesced wakes keep the widest request: "input" subsumes "steer". - execution.pendingWake = execution.pendingWake === "input" ? "input" : request + // Coalesced wakes keep the widest scope: "input" subsumes "steer". + execution.pendingWake = execution.pendingWake === "input" ? "input" : scope return } - start(key, false, request) + start(key, false, scope) }) const wakeActive = (key: Key) => Effect.suspend(() => { const execution = executions.get(key) - return execution ? wake(key, execution.request) : Effect.void + return execution ? wake(key, execution.scope) : Effect.void }) - const interrupt = ( - key: Key, - reason?: Reason, - options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect } }, - ): Effect.Effect => + const interrupt = (key: Key, reason?: Reason): Effect.Effect => Effect.suspend(() => { const execution = executions.get(key) - if (execution === undefined) return Effect.void - if (execution.stopping) { - if (options?.continue) - execution.continuation = { - ...options.continue, - signaled: execution.continuation?.signaled ?? false, - } - return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid) - } - if (execution.owner === undefined) { - if (!options?.continue) return Effect.void - execution.stopping = true - execution.pendingWake = undefined - execution.continuation = { ...options.continue, signaled: false } - return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid) - } + if (execution?.owner === undefined || execution.stopping) return Effect.void execution.stopping = true + // Wakes recorded so far belong to the interrupted intent; the interrupt claims them. + // Wakes arriving during cleanup are new admissions and restart normally at settle. execution.pendingWake = undefined execution.interruptionReason = reason - if (options?.continue) execution.continuation = { ...options.continue, signaled: false } return Fiber.interrupt(execution.owner) }) diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index d41f1416809..b5acc837461 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -14,8 +14,10 @@ import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionRestart } from "@opencode-ai/core/session/execution/restart" import { UserInterruptedError } from "@opencode-ai/core/session/error" import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionInbox } from "@opencode-ai/core/session/inbox" +import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionRunner } from "@opencode-ai/core/session/runner/index" -import { SessionTable } from "@opencode-ai/core/session/sql" +import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect" import { eq } from "drizzle-orm" @@ -290,6 +292,113 @@ describe("SessionExecution lifecycle", () => { ) }) +describe("SessionExecution interrupt continuation", () => { + it.effect("resumes only steering input after an interrupt with continue", () => + Effect.gen(function* () { + const database = yield* Database.Service + const sessionID = Session.ID.make("ses_continue_steer") + yield* seedSessions(database, [sessionID]) + yield* seedInbox(database, sessionID, ["steer", "queue"]) + + const draining = yield* Deferred.make() + const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = [] + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) + const context = yield* buildExecution(scope, (input) => + Effect.suspend(() => { + drains.push({ force: input.force, promotable: input.promotable }) + if (drains.length > 1) return Effect.void + return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never)) + }), + ) + const execution = Context.get(context, SessionExecution.Service) + yield* execution.resume(sessionID).pipe(Effect.forkScoped) + yield* Deferred.await(draining) + + yield* execution.interrupt(sessionID, { continue: true }) + yield* execution.awaitIdle(sessionID) + + // The successor drain is steer-scoped: queued next-turn work stays parked. + expect(drains).toEqual([ + { force: true, promotable: "input" }, + { force: false, promotable: "steer" }, + ]) + }), + ) + + it.effect("stays parked after an interrupt with continue when only queued work remains", () => + Effect.gen(function* () { + const database = yield* Database.Service + const sessionID = Session.ID.make("ses_continue_parked") + yield* seedSessions(database, [sessionID]) + yield* seedInbox(database, sessionID, ["queue"]) + + const draining = yield* Deferred.make() + const drains: Array = [] + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) + const context = yield* buildExecution(scope, (input) => + Effect.suspend(() => { + drains.push(input.promotable) + return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never)) + }), + ) + const execution = Context.get(context, SessionExecution.Service) + yield* execution.resume(sessionID).pipe(Effect.forkScoped) + yield* Deferred.await(draining) + + yield* execution.interrupt(sessionID, { continue: true }) + yield* execution.awaitIdle(sessionID) + + expect(drains).toEqual(["input"]) + expect(yield* execution.active).toEqual(new Set()) + }), + ) + + it.effect("an idle interrupt with continue resumes pending steers", () => + Effect.gen(function* () { + const database = yield* Database.Service + const sessionID = Session.ID.make("ses_continue_idle") + yield* seedSessions(database, [sessionID]) + yield* seedInbox(database, sessionID, ["steer"]) + + const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = [] + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) + const context = yield* buildExecution(scope, (input) => + Effect.sync(() => void drains.push({ force: input.force, promotable: input.promotable })), + ) + const execution = Context.get(context, SessionExecution.Service) + + yield* execution.interrupt(sessionID, { continue: true }) + yield* execution.awaitIdle(sessionID) + + expect(drains).toEqual([{ force: false, promotable: "steer" }]) + }), + ) +}) + +function seedInbox( + database: Database.Service["Service"], + sessionID: Session.ID, + deliveries: ReadonlyArray, +) { + return database.db + .insert(SessionInboxTable) + .values( + deliveries.map((delivery, index) => ({ + id: SessionMessage.ID.create(), + session_id: sessionID, + type: "compaction" as const, + payload: {}, + delivery, + enqueued_seq: index + 1, + })), + ) + .run() + .pipe(Effect.orDie) +} + function seedSessions( database: Database.Service["Service"], sessionIDs: ReadonlyArray, diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index 1c3995488ff..053d16f58d1 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -1,5 +1,6 @@ import { describe, expect } from "bun:test" import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect" +import { SessionInbox } from "@opencode-ai/core/session/inbox" import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator" import { testEffect } from "./lib/effect" @@ -269,31 +270,24 @@ describe("SessionRunCoordinator", () => { ), ) - it.effect("replaces a settlement-window wake with a steer continuation", () => + it.effect("a settlement-window wake starts a fresh execution with its own scope", () => Effect.scoped( Effect.gen(function* () { const settling = yield* Deferred.make() const release = yield* Deferred.make() - const requests: SessionRunCoordinator.Request[] = [] + const scopes: SessionInbox.Promotable[] = [] const coordinator = yield* SessionRunCoordinator.make({ - drain: (_key, _force, request) => Effect.sync(() => requests.push(request)), + drain: (_key, _force, scope) => Effect.sync(() => scopes.push(scope)), settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))), }) - yield* coordinator.wake("session", "input") + yield* coordinator.wake("session", "steer") yield* Deferred.await(settling) yield* coordinator.wake("session", "input") - const interrupted = yield* coordinator - .interrupt("session", undefined, { - continue: { request: "steer", when: Effect.succeed(true) }, - }) - .pipe(Effect.forkChild) - yield* Effect.yieldNow yield* Deferred.succeed(release, undefined) - yield* Fiber.join(interrupted) yield* coordinator.awaitIdle("session") - expect(requests).toEqual(["input", "steer"]) + expect(scopes).toEqual(["steer", "input"]) }), ), ) @@ -371,17 +365,17 @@ describe("SessionRunCoordinator", () => { ), ) - it.effect("coalesces drain requests with input taking precedence", () => + it.effect("coalesces drain scopes with input taking precedence", () => Effect.scoped( Effect.gen(function* () { const firstStarted = yield* Deferred.make() const release = yield* Deferred.make() - const requests: SessionRunCoordinator.Request[] = [] + const scopes: SessionInbox.Promotable[] = [] const coordinator = yield* SessionRunCoordinator.make({ - drain: (_key, _force, request) => + drain: (_key, _force, scope) => Effect.gen(function* () { - requests.push(request) - if (requests.length !== 1) return + scopes.push(scope) + if (scopes.length !== 1) return yield* Deferred.succeed(firstStarted, undefined) yield* Deferred.await(release) }), @@ -394,22 +388,22 @@ describe("SessionRunCoordinator", () => { yield* Deferred.succeed(release, undefined) yield* coordinator.awaitIdle("session") - expect(requests).toEqual(["steer", "input"]) + expect(scopes).toEqual(["steer", "input"]) }), ), ) - it.effect("does not carry a completed input request into a steer drain", () => + it.effect("does not carry a completed input scope into a steer drain", () => Effect.scoped( Effect.gen(function* () { const firstStarted = yield* Deferred.make() const release = yield* Deferred.make() - const requests: SessionRunCoordinator.Request[] = [] + const scopes: SessionInbox.Promotable[] = [] const coordinator = yield* SessionRunCoordinator.make({ - drain: (_key, _force, request) => + drain: (_key, _force, scope) => Effect.gen(function* () { - requests.push(request) - if (requests.length !== 1) return + scopes.push(scope) + if (scopes.length !== 1) return yield* Deferred.succeed(firstStarted, undefined) yield* Deferred.await(release) }), @@ -421,7 +415,7 @@ describe("SessionRunCoordinator", () => { yield* Deferred.succeed(release, undefined) yield* coordinator.awaitIdle("session") - expect(requests).toEqual(["input", "steer"]) + expect(scopes).toEqual(["input", "steer"]) }), ), ) @@ -431,12 +425,12 @@ describe("SessionRunCoordinator", () => { Effect.gen(function* () { const firstStarted = yield* Deferred.make() const release = yield* Deferred.make() - const requests: SessionRunCoordinator.Request[] = [] + const scopes: SessionInbox.Promotable[] = [] const coordinator = yield* SessionRunCoordinator.make({ - drain: (_key, _force, request) => + drain: (_key, _force, scope) => Effect.gen(function* () { - requests.push(request) - if (requests.length !== 1) return + scopes.push(scope) + if (scopes.length !== 1) return yield* Deferred.succeed(firstStarted, undefined) yield* Deferred.await(release) }), @@ -449,61 +443,23 @@ describe("SessionRunCoordinator", () => { yield* Deferred.succeed(release, undefined) yield* coordinator.awaitIdle("session") - expect(requests).toEqual(["steer", "steer"]) + expect(scopes).toEqual(["steer", "steer"]) }), ), ) - it.effect("coalesces overlapping interrupt continuations into one steer successor", () => + it.effect("a cleanup-era wake starts a successor with its own scope", () => Effect.scoped( Effect.gen(function* () { const firstStarted = yield* Deferred.make() const cleanupStarted = yield* Deferred.make() const cleanupGate = yield* Deferred.make() - const requests: SessionRunCoordinator.Request[] = [] + const scopes: SessionInbox.Promotable[] = [] const coordinator = yield* SessionRunCoordinator.make({ - drain: (_key, _force, request) => + drain: (_key, _force, scope) => Effect.gen(function* () { - requests.push(request) - if (requests.length !== 1) return - yield* Deferred.succeed(firstStarted, undefined) - yield* Effect.never.pipe( - Effect.onInterrupt(() => - Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))), - ), - ) - }), - }) - const continuation = { continue: { request: "steer" as const, when: Effect.succeed(false) } } - - yield* coordinator.wake("session") - yield* Deferred.await(firstStarted) - const first = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild) - yield* Deferred.await(cleanupStarted) - const second = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild) - yield* Effect.yieldNow - yield* coordinator.wake("session", "input") - yield* Deferred.succeed(cleanupGate, undefined) - yield* Effect.all([Fiber.join(first), Fiber.join(second)]) - yield* coordinator.awaitIdle("session") - - expect(requests).toEqual(["input", "steer"]) - }), - ), - ) - - it.effect("a continuing interrupt replaces a cleanup-era input wake", () => - Effect.scoped( - Effect.gen(function* () { - const firstStarted = yield* Deferred.make() - const cleanupStarted = yield* Deferred.make() - const cleanupGate = yield* Deferred.make() - const requests: SessionRunCoordinator.Request[] = [] - const coordinator = yield* SessionRunCoordinator.make({ - drain: (_key, _force, request) => - Effect.gen(function* () { - requests.push(request) - if (requests.length !== 1) return + scopes.push(scope) + if (scopes.length !== 1) return yield* Deferred.succeed(firstStarted, undefined) yield* Effect.never.pipe( Effect.onInterrupt(() => @@ -515,45 +471,16 @@ describe("SessionRunCoordinator", () => { yield* coordinator.wake("session", "input") yield* Deferred.await(firstStarted) - const plain = yield* coordinator.interrupt("session").pipe(Effect.forkChild) + const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild) yield* Deferred.await(cleanupStarted) + // A new admission during cancellation restarts normally: interruption only + // claims the wakes recorded before it. yield* coordinator.wake("session", "input") - const continuing = yield* coordinator - .interrupt("session", undefined, { - continue: { request: "steer", when: Effect.succeed(false) }, - }) - .pipe(Effect.forkChild) - yield* Effect.yieldNow yield* Deferred.succeed(cleanupGate, undefined) - yield* Effect.all([Fiber.join(plain), Fiber.join(continuing)]) + yield* Fiber.join(interrupt) yield* coordinator.awaitIdle("session") - expect(requests).toEqual(["input", "steer"]) - }), - ), - ) - - it.effect("does not start a conditional continuation without eligible work", () => - Effect.scoped( - Effect.gen(function* () { - const started = yield* Deferred.make() - const requests: SessionRunCoordinator.Request[] = [] - const coordinator = yield* SessionRunCoordinator.make({ - drain: (_key, _force, request) => - Effect.sync(() => requests.push(request)).pipe( - Effect.andThen(Deferred.succeed(started, undefined)), - Effect.andThen(Effect.never), - ), - }) - - yield* coordinator.wake("session") - yield* Deferred.await(started) - yield* coordinator.interrupt("session", undefined, { - continue: { request: "steer", when: Effect.succeed(false) }, - }) - yield* coordinator.awaitIdle("session") - - expect(requests).toEqual(["input"]) + expect(scopes).toEqual(["input", "input"]) }), ), )