diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 5dafb157738..c6cfc6728a8 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -789,7 +789,10 @@ const layer = Layer.effect( return false }), ) - if (recovered) return + if (recovered) { + yield* execution.wakeActive(input.sessionID) + return + } yield* execution.wake(input.sessionID) }), compact: Effect.fn("Session.compact")(function* (input) { @@ -873,12 +876,7 @@ const layer = Layer.effect( ), ), interrupt: Effect.fn("Session.interrupt")((sessionID, options) => - Effect.uninterruptible( - Effect.gen(function* () { - yield* execution.interrupt(sessionID) - if (options?.continue && (yield* SessionInbox.has(db, sessionID, "any"))) yield* execution.wake(sessionID) - }), - ), + Effect.uninterruptible(execution.interrupt(sessionID, options)), ), revert: { stage: Effect.fn("Session.revert.stage")(function* (input) { diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 6ea2fc0779c..3da6165a527 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -1,7 +1,8 @@ export * as SessionExecution from "./execution.js" -import { Cause, Context, Effect, Exit, Layer, Stream } from "effect" +import { Cause, Context, Effect, Exit, Layer } from "effect" import { Bus } from "../bus.js" +import { Database } from "../database/database.js" import { LocationServiceMap } from "../location-service-map.js" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { SessionEvent } from "./event.js" @@ -11,6 +12,7 @@ import { SessionSchema } from "./schema.js" import { SessionStore } from "./store.js" import { toSessionError } from "./to-session-error.js" import { UserInterruptedError } from "./error.js" +import { SessionInbox } from "./inbox.js" export interface Interface { /** Snapshots active execution owned by this process. */ @@ -19,8 +21,10 @@ export interface Interface { readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect /** Registers newly recorded work. Repeated wakeups may coalesce. */ readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect + /** Wakes only an active execution, preserving its current input eligibility. */ + readonly wakeActive: (sessionID: SessionSchema.ID) => Effect.Effect /** Interrupt active work owned by this process. Idle interruption is a no-op. */ - readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect + readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect /** 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 } @@ -45,6 +49,7 @@ export const layer = Layer.effect( const store = yield* SessionStore.Service const locations = yield* LocationServiceMap.Service const bus = yield* Bus.Service + const db = (yield* Database.Service).db const reportLifecycle = (sessionID: SessionSchema.ID, effect: Effect.Effect) => effect.pipe( Effect.tapCause((cause) => @@ -71,12 +76,13 @@ export const layer = Layer.effect( sessionID: SessionSchema.ID, force: boolean, continuation?: SessionRunner.Continuation, + promotable: SessionInbox.Promotable = "input", ): Effect.Effect { return Effect.gen(function* () { const session = yield* store.get(sessionID) if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) const result = yield* SessionRunner.Service.use((runner) => - runner.drain({ sessionID, force, continuation }), + runner.drain({ sessionID, force, continuation, promotable }), ).pipe( Effect.provide(locations.get(session.location)), Effect.tapCause((cause) => @@ -86,7 +92,7 @@ export const layer = Layer.effect( ), ) if (result.type === "complete") return - return yield* drain(sessionID, false, result.continuation) + return yield* drain(sessionID, false, result.continuation, promotable) }) } const coordinator = yield* SessionRunCoordinator.make({ @@ -95,7 +101,7 @@ export const layer = Layer.effect( sessionID, bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)), ), - drain: (sessionID, force) => drain(sessionID, force), + drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable), // One terminal observation per busy period, covering every coalesced drain. settled: (sessionID, exit, reason) => reportLifecycle( @@ -127,16 +133,20 @@ export const layer = Layer.effect( }), ), }) - yield* bus.subscribe(SessionEvent.Moved).pipe( - Stream.runForEach((event) => coordinator.wake(event.data.sessionID)), - Effect.forkScoped, - ) return Service.of({ active: coordinator.active, - interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"), + interrupt: (sessionID, options) => + coordinator.interrupt( + sessionID, + "user", + options?.continue + ? { continue: { request: "steer", when: SessionInbox.has(db, sessionID, "steer") } } + : undefined, + ), resume: coordinator.run, wake: coordinator.wake, + wakeActive: coordinator.wakeActive, awaitIdle: coordinator.awaitIdle, }) }), @@ -145,7 +155,7 @@ export const layer = Layer.effect( export const node = makeGlobalNode({ service: Service, layer, - deps: [SessionStore.node, LocationServiceMap.node, Bus.node], + deps: [SessionStore.node, LocationServiceMap.node, Bus.node, Database.node], }) /** Low-level compatibility layer for callers that only need durable Session recording. */ @@ -155,6 +165,7 @@ export const noopLayer = Layer.succeed( active: Effect.succeed(new Set()), resume: () => Effect.void, wake: () => Effect.void, + wakeActive: () => Effect.void, interrupt: () => Effect.void, awaitIdle: () => Effect.void, }), diff --git a/packages/core/src/session/inbox.ts b/packages/core/src/session/inbox.ts index f324a77a458..a0412326b78 100644 --- a/packages/core/src/session/inbox.ts +++ b/packages/core/src/session/inbox.ts @@ -349,6 +349,14 @@ export const nextSteer = Effect.fn("SessionInbox.nextSteer")(function* ( return row ? fromRow(row) : undefined }) +export const nextPromotable = Effect.fn("SessionInbox.nextPromotable")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + promotable: Promotable, +) { + return (yield* nextSteer(db, sessionID)) ?? (promotable === "input" ? yield* nextQueued(db, sessionID) : undefined) +}) + /** * Which pending rows count: "any" counts every row, while "input" means any * item in either delivery mode. diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index d29ba0d9dd7..1c9b8c2ba94 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -1,6 +1,7 @@ export * as SessionRunCoordinator from "./run-coordinator.js" import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect" +import type { Promotable } from "./inbox.js" /** Serializes execution for each key while allowing different keys to run concurrently. */ export interface Coordinator { @@ -9,26 +10,41 @@ 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) => Effect.Effect + readonly wake: (key: Key, request?: Request) => Effect.Effect + /** Rings the current execution's doorbell with its existing request. 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) => Effect.Effect + readonly interrupt: ( + key: Key, + reason?: Reason, + options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect } }, + ) => 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, 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. + * execution rings it with its eligibility request, 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. */ type Execution = { readonly done: Deferred.Deferred owner?: Fiber.Fiber - pendingWake: boolean + request: Request + pendingWake?: Request stopping: boolean interruptionReason?: Reason + continuation?: { + readonly request: Request + readonly when: Effect.Effect + signaled: boolean + } } /** @@ -43,7 +59,7 @@ type Execution = { * ``` */ export const make = (options: { - readonly drain: (key: Key, force: boolean) => Effect.Effect + readonly drain: (key: Key, force: boolean, request: Request) => Effect.Effect /** Runs once when a process-local busy period begins, before its first drain. */ readonly started?: (key: Key) => Effect.Effect /** @@ -57,21 +73,22 @@ 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)).pipe( + Effect.suspend(() => options.drain(key, force, execution.request)).pipe( Effect.flatMap(() => Effect.suspend(() => { - if (execution.stopping || !execution.pendingWake) return Effect.void - execution.pendingWake = false + if (execution.stopping || execution.pendingWake === undefined) return Effect.void + execution.request = 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))) }), ), ) - const start = (key: Key, force: boolean) => { + const start = (key: Key, force: boolean, request: Request) => { const execution: Execution = { done: Deferred.makeUnsafe(), - pendingWake: false, + request, stopping: false, } executions.set(key, execution) @@ -87,7 +104,7 @@ export const make = (options: { execution.owner = undefined }).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)), ), - Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))), + Effect.onExit((exit) => finish(key, execution, exit)), Effect.exit, Effect.asVoid, ), @@ -97,12 +114,22 @@ 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) => { - if (execution.pendingWake) start(key, false) + 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) 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) @@ -111,26 +138,58 @@ export const make = (options: { if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key))) return Deferred.await(execution.done) } - return Deferred.await(start(key, true).done) + return Deferred.await(start(key, true, "input").done) }) - const wake = (key: Key) => + const wake = (key: Key, request: Request = "input") => Effect.sync(() => { const execution = executions.get(key) if (execution !== undefined) { - execution.pendingWake = true + 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 return } - start(key, false) + start(key, false, request) }) - const interrupt = (key: Key, reason?: Reason): Effect.Effect => + const wakeActive = (key: Key) => Effect.suspend(() => { const execution = executions.get(key) - if (execution?.owner === undefined || execution.stopping) return Effect.void + return execution ? wake(key, execution.request) : Effect.void + }) + + const interrupt = ( + key: Key, + reason?: Reason, + options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect } }, + ): 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) + } execution.stopping = true - execution.pendingWake = false + execution.pendingWake = undefined execution.interruptionReason = reason + if (options?.continue) execution.continuation = { ...options.continue, signaled: false } return Fiber.interrupt(execution.owner) }) @@ -143,5 +202,5 @@ export const make = (options: { return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key))) }) - return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle } + return { active: Effect.sync(() => new Set(executions.keys())), run, wake, wakeActive, interrupt, awaitIdle } }) diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index e744f9bb75c..4525918586d 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -3,6 +3,7 @@ export * as SessionRunner from "./index.js" import type { AIError } from "@opencode-ai/ai" import { Context, Effect } from "effect" import { SessionSchema } from "../schema.js" +import type { Promotable } from "../inbox.js" import type { AgentNotFoundError, MessageDecodeError, StepFailedError, UserInterruptedError } from "../error.js" import { SessionRunnerModel } from "./model.js" import type { Instructions } from "../../instructions/index.js" @@ -29,6 +30,8 @@ export interface Interface { readonly sessionID: SessionSchema.ID readonly force: boolean readonly continuation?: Continuation + /** "steer" settles the active intent without promoting queued next-turn work. */ + readonly promotable?: Promotable }) => Effect.Effect } diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 94ac3baf0e5..34577e86a68 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -128,22 +128,25 @@ const layer = Layer.effect( readonly sessionID: SessionSchema.ID readonly force: boolean readonly continuation?: Continuation + readonly promotable?: SessionInbox.Promotable }) { let force = input.force let continuation = input.continuation - if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "any"))) + const promotable = input.promotable ?? "input" + if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable))) return { type: "complete" as const } yield* settleStaleToolCalls(input.sessionID) while (true) { - if (yield* runPendingCompaction(input.sessionID)) { + if (yield* runPendingCompaction(input.sessionID, promotable)) { force = false continue } - if (yield* runPendingMove(input.sessionID, "input")) return { type: "moved" as const } - if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "input"))) + if (yield* runPendingMove(input.sessionID, promotable)) return { type: "moved" as const } + if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable))) return { type: "complete" as const } - const result = yield* runSteps(input.sessionID, continuation) + const result = yield* runSteps(input.sessionID, continuation, promotable) if (result.type === "moved") return result + if (promotable === "steer") return { type: "complete" as const } force = false continuation = undefined } @@ -155,14 +158,15 @@ const layer = Layer.effect( */ const runSteps = Effect.fn("SessionRunner.runSteps")(function* ( sessionID: SessionSchema.ID, - continuation?: Continuation, + continuation: Continuation | undefined, + drainPromotable: SessionInbox.Promotable, ) { - // Fresh work may promote queued input; later steps absorb steers only. - let promotable: SessionInbox.Promotable = continuation ? "steer" : "input" + // Fresh work may promote queued input; resumed turns and later steps absorb steers only. + let promotable: SessionInbox.Promotable = continuation ? "steer" : drainPromotable let step = continuation?.step ?? 1 let next = continuation while (true) { - if (yield* runPendingCompaction(sessionID)) continue + if (yield* runPendingCompaction(sessionID, "steer")) continue if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next } const result = yield* runStep(sessionID, promotable, step) next = result.needsContinuation ? { step: result.step + 1 } : undefined @@ -515,14 +519,14 @@ const layer = Layer.effect( /** Executes a previously admitted manual compaction request, if one is pending. */ const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* ( sessionID: SessionSchema.ID, + promotable: SessionInbox.Promotable, ) { return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { const pending = yield* SessionInbox.serialized( sessionID, Effect.gen(function* () { - const selected = - (yield* SessionInbox.nextSteer(db, sessionID)) ?? (yield* SessionInbox.nextQueued(db, sessionID)) + const selected = yield* SessionInbox.nextPromotable(db, sessionID, promotable) if (selected?.type !== "compaction") return yield* bus.publishAll([ [SessionEvent.InboxDelivered, { sessionID, inboxID: selected.id }], @@ -564,9 +568,7 @@ const layer = Layer.effect( return yield* SessionInbox.serialized( sessionID, Effect.gen(function* () { - const pending = - (yield* SessionInbox.nextSteer(db, sessionID)) ?? - (promotable === "input" ? yield* SessionInbox.nextQueued(db, sessionID) : undefined) + const pending = yield* SessionInbox.nextPromotable(db, sessionID, promotable) if (pending?.type !== "move") return false yield* modelTransport.close(sessionID) yield* bus.publishAll([ diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index aa60be29a0a..b5975d8227d 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -31,6 +31,7 @@ import { testEffect } from "./lib/effect" const executionCalls: Session.ID[] = [] const interruptCalls: Session.ID[] = [] +const interruptContinuations: Array = [] const wakeCalls: Session.ID[] = [] const activeSessions = new Set() const execution = Layer.succeed( @@ -41,14 +42,16 @@ const execution = Layer.succeed( Effect.sync(() => { executionCalls.push(sessionID) }), - interrupt: (sessionID) => + interrupt: (sessionID, options) => Effect.sync(() => { interruptCalls.push(sessionID) + interruptContinuations.push(options?.continue) }), wake: (sessionID) => Effect.sync(() => { wakeCalls.push(sessionID) }), + wakeActive: () => Effect.void, awaitIdle: () => Effect.void, }), ) @@ -177,31 +180,18 @@ describe("Session.prompt", () => { }), ) - 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", () => + it.effect("forwards interrupt continuation policy", () => Effect.gen(function* () { yield* setup const session = yield* Session.Service interruptCalls.length = 0 + interruptContinuations.length = 0 wakeCalls.length = 0 yield* session.interrupt(sessionID, { continue: true }) expect(interruptCalls).toEqual([sessionID]) + expect(interruptContinuations).toEqual([true]) expect(wakeCalls).toEqual([]) }), ) diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index c566c3f33c1..1c3995488ff 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -269,6 +269,35 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("replaces a settlement-window wake with a steer continuation", () => + Effect.scoped( + Effect.gen(function* () { + const settling = yield* Deferred.make() + const release = yield* Deferred.make() + const requests: SessionRunCoordinator.Request[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: (_key, _force, request) => Effect.sync(() => requests.push(request)), + settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))), + }) + + yield* coordinator.wake("session", "input") + 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"]) + }), + ), + ) + it.effect("interrupts active execution and clears its pending wake", () => Effect.scoped( Effect.gen(function* () { @@ -342,6 +371,193 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("coalesces drain requests with input taking precedence", () => + Effect.scoped( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make() + const release = 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 + yield* Deferred.succeed(firstStarted, undefined) + yield* Deferred.await(release) + }), + }) + + yield* coordinator.wake("session", "steer") + yield* Deferred.await(firstStarted) + yield* coordinator.wake("session", "steer") + yield* coordinator.wake("session", "input") + yield* Deferred.succeed(release, undefined) + yield* coordinator.awaitIdle("session") + + expect(requests).toEqual(["steer", "input"]) + }), + ), + ) + + it.effect("does not carry a completed input request into a steer drain", () => + Effect.scoped( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make() + const release = 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 + yield* Deferred.succeed(firstStarted, undefined) + yield* Deferred.await(release) + }), + }) + + yield* coordinator.wake("session", "input") + yield* Deferred.await(firstStarted) + yield* coordinator.wake("session", "steer") + yield* Deferred.succeed(release, undefined) + yield* coordinator.awaitIdle("session") + + expect(requests).toEqual(["input", "steer"]) + }), + ), + ) + + it.effect("an active wake inherits scope without starting idle work", () => + Effect.scoped( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make() + const release = 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 + yield* Deferred.succeed(firstStarted, undefined) + yield* Deferred.await(release) + }), + }) + + yield* coordinator.wakeActive("session") + yield* coordinator.wake("session", "steer") + yield* Deferred.await(firstStarted) + yield* coordinator.wakeActive("session") + yield* Deferred.succeed(release, undefined) + yield* coordinator.awaitIdle("session") + + expect(requests).toEqual(["steer", "steer"]) + }), + ), + ) + + it.effect("coalesces overlapping interrupt continuations into one steer successor", () => + 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 + 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 + yield* Deferred.succeed(firstStarted, undefined) + yield* Effect.never.pipe( + Effect.onInterrupt(() => + Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))), + ), + ) + }), + }) + + yield* coordinator.wake("session", "input") + yield* Deferred.await(firstStarted) + const plain = yield* coordinator.interrupt("session").pipe(Effect.forkChild) + yield* Deferred.await(cleanupStarted) + 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* 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"]) + }), + ), + ) + it.effect("starts a resume registered during interruption cleanup", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index 1533d8e20a1..522cf7a05c8 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -126,7 +126,8 @@ const execution = (llmClient: Layer.Layer) => active: coordinator.active, resume: coordinator.run, wake: coordinator.wake, - interrupt: coordinator.interrupt, + wakeActive: coordinator.wakeActive, + interrupt: (sessionID) => coordinator.interrupt(sessionID), awaitIdle: coordinator.awaitIdle, }) }), diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 773bc8558ce..b3a3adb030e 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -413,7 +413,8 @@ const execution = Layer.effect( active: coordinator.active, resume: coordinator.run, wake: coordinator.wake, - interrupt: coordinator.interrupt, + wakeActive: coordinator.wakeActive, + interrupt: (sessionID) => coordinator.interrupt(sessionID), awaitIdle: coordinator.awaitIdle, }) }), @@ -1383,6 +1384,44 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("keeps queued input parked across a mid-turn move", () => + Effect.gen(function* () { + const session = yield* setup + const bus = yield* Bus.Service + const { db } = yield* Database.Service + yield* admit(session, "Echo before moving") + yield* TestLLM.push( + TestLLM.tool("call-move", "echo", { text: "moving" }), + TestLLM.text("Done", "text-after-move"), + TestLLM.text("Handled queue", "text-after-queue"), + ) + const tools = yield* blockTools() + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* tools.started + yield* session.prompt({ sessionID, text: "Queued for later", delivery: "queue", resume: false }) + yield* SessionInbox.admit(db, bus, { + id: SessionMessage.ID.create(), + sessionID, + item: { + type: "move", + payload: { + location: Location.Ref.make({ directory: AbsolutePath.make("/project") }), + projectID: Project.ID.global, + }, + delivery: "steer", + }, + }) + + yield* tools.release + yield* Fiber.join(run) + + // The resumed turn absorbs steers only; queued input waits for the turn to end. + expect(requests).toHaveLength(3) + expect(userTexts(requests[1])).not.toContain("Queued for later") + expect(userTexts(requests[2])).toContain("Queued for later") + }), + ) + it.effect("seeds a fork with the parent's newest instruction values", () => Effect.gen(function* () { const session = yield* setup @@ -3088,6 +3127,24 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("stops a steer-scoped drain before queued input", () => + Effect.gen(function* () { + const session = yield* setup + const { db } = yield* Database.Service + yield* session.prompt({ sessionID, text: "Queue for later", delivery: "queue", resume: false }) + yield* session.prompt({ sessionID, text: "Steer now", resume: false }) + yield* TestLLM.push(TestLLM.stop()) + + const runner = yield* SessionRunner.Service + yield* runner.drain({ sessionID, force: false, promotable: "steer" }) + + expect(requests).toHaveLength(1) + expect(userTexts(requests[0])).toEqual(["Steer now"]) + expect(yield* SessionInbox.has(db, sessionID, "steer")).toBe(false) + expect(yield* SessionInbox.has(db, sessionID, "queue")).toBe(true) + }), + ) + it.effect("promotes queued input after steering continuation ends", () => Effect.gen(function* () { const session = yield* setup diff --git a/packages/core/test/tool-shell.test.ts b/packages/core/test/tool-shell.test.ts index dc58c62405a..4ba88464d7a 100644 --- a/packages/core/test/tool-shell.test.ts +++ b/packages/core/test/tool-shell.test.ts @@ -115,6 +115,7 @@ const executionNode = makeGlobalNode({ active: Effect.succeed(new Set()), resume: complete, wake: () => Effect.void, + wakeActive: () => Effect.void, interrupt: () => Effect.void, awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid), }) diff --git a/packages/core/test/tool-subagent.test.ts b/packages/core/test/tool-subagent.test.ts index 9e970eb7be2..3aeeabe774a 100644 --- a/packages/core/test/tool-subagent.test.ts +++ b/packages/core/test/tool-subagent.test.ts @@ -86,6 +86,7 @@ const executionNode = makeGlobalNode({ active: Effect.succeed(new Set()), resume: complete, wake: () => Effect.void, + wakeActive: () => Effect.void, interrupt: () => Effect.void, awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid), }) diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 30677d58414..6cd75d41d66 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -660,7 +660,7 @@ export const makeSessionGroup = (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 if durable inbox work remains after interruption.", + "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input while queued work remains parked.", }), ), )