diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 2f3738709f8..dc8e242d84f 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -21,7 +21,11 @@ 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 - /** Interrupt active work owned by this process. Idle interruption is a no-op. */ + /** + * 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. + */ 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 @@ -141,6 +145,11 @@ export const layer = Layer.effect( // 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. + // Interruption acknowledges before cleanup settles, so this wake usually lands + // on the stopping execution's doorbell and starts the successor at settle. + // Reading the inbox concurrently with the dying drain is safe: delivery consumes + // 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.delivery === "steer" || next.type === "compaction" || next.type === "move") diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index e78280fc7c2..75d81557a10 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -11,7 +11,11 @@ export interface Coordinator { 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, scope?: Promotable) => Effect.Effect - /** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */ + /** + * 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. + */ 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 @@ -26,7 +30,13 @@ export interface Coordinator { * with this execution's exit. */ type Execution = { - readonly done: Deferred.Deferred + /** + * Resolves with the execution's exit as a success value. Success-valued on purpose: + * completing a Deferred with an interrupted exit interrupts suspended waiters as it + * resumes them, and can starve later waiters of their resume entirely + * (Effect-TS/effect#7364). Joiners flatten the exit; idleness waiters just await. + */ + readonly done: Deferred.Deferred> owner?: Fiber.Fiber scope: Promotable pendingWake?: Promotable @@ -74,7 +84,7 @@ export const make = (options: { const start = (key: Key, force: boolean, scope: Promotable) => { const execution: Execution = { - done: Deferred.makeUnsafe(), + done: Deferred.makeUnsafe>(), scope, stopping: false, } @@ -104,7 +114,7 @@ export const make = (options: { 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) + Deferred.doneUnsafe(execution.done, Exit.succeed(exit)) } const run = (key: Key): Effect.Effect => @@ -113,9 +123,9 @@ export const make = (options: { if (execution !== undefined) { // A stopping execution refuses joiners: wait out its cleanup, then run fresh. if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key))) - return Deferred.await(execution.done) + return Deferred.await(execution.done).pipe(Effect.flatten) } - return Deferred.await(start(key, true, "input").done) + return Deferred.await(start(key, true, "input").done).pipe(Effect.flatten) }) const wake = (key: Key, scope: Promotable = "input") => @@ -132,13 +142,23 @@ export const make = (options: { const interrupt = (key: Key, reason?: Reason): Effect.Effect => Effect.suspend(() => { const execution = executions.get(key) - if (execution?.owner === undefined || execution.stopping) return Effect.void + if (execution === undefined || execution.stopping) return Effect.void + 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 + } 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 - return Fiber.interrupt(execution.owner) + // 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 }) // One execution's `done` already spans coalesced continuations; re-check after it @@ -147,7 +167,7 @@ export const make = (options: { Effect.suspend(() => { const execution = executions.get(key) if (execution === undefined) return Effect.void - return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key))) + return Deferred.await(execution.done).pipe(Effect.andThen(awaitIdle(key))) }) return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle } diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index ec97a093cf4..415de1185ba 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -622,6 +622,7 @@ describe("SessionRunCoordinator", () => { yield* coordinator.wake("session") yield* Deferred.await(started) yield* coordinator.interrupt("session") + yield* coordinator.awaitIdle("session") expect(settled).toHaveLength(1) expect(settled[0] !== undefined && Exit.isFailure(settled[0]) && Cause.hasInterrupts(settled[0].cause)).toBe( @@ -632,6 +633,69 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("acknowledges interruption before cleanup settles", () => + Effect.scoped( + Effect.gen(function* () { + const started = yield* Deferred.make() + const cleanupStarted = yield* Deferred.make() + const cleanupGate = yield* Deferred.make() + const settled: Array = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => + Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))), + ), + ), + settled: (_key, _exit, reason) => Effect.sync(() => void settled.push(reason)), + }) + + yield* coordinator.wake("session") + yield* Deferred.await(started) + // Interrupt resolves while the cleanup gate is still closed: acceptance, not settlement. + yield* coordinator.interrupt("session", "user") + expect(settled).toHaveLength(0) + expect(Array.from(yield* coordinator.active)).toEqual(["session"]) + // Repeating the interrupt during cleanup stays an immediate no-op. + yield* coordinator.interrupt("session", "user") + + yield* Deferred.await(cleanupStarted) + yield* Deferred.succeed(cleanupGate, undefined) + yield* coordinator.awaitIdle("session") + + expect(settled).toEqual(["user"]) + expect(yield* coordinator.active).toEqual(new Set()) + }), + ), + ) + + it.effect("an interrupt during terminal settlement claims the recorded wake", () => + Effect.scoped( + Effect.gen(function* () { + const settling = yield* Deferred.make() + const release = yield* Deferred.make() + let drains = 0 + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => Effect.sync(() => void drains++), + settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))), + }) + + yield* coordinator.wake("session") + yield* Deferred.await(settling) + // The owner has exited; this wake lands on the settling execution's doorbell. + yield* coordinator.wake("session") + // The interrupt claims it: settle must not start a successor for the dead intent. + yield* coordinator.interrupt("session") + yield* Deferred.succeed(release, undefined) + yield* coordinator.awaitIdle("session") + + expect(drains).toBe(1) + expect(yield* coordinator.active).toEqual(new Set()) + }), + ), + ) + it.effect("trampolines synchronous self-waking execution", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/tui/src/mini/runtime.ts b/packages/tui/src/mini/runtime.ts index 0a4dcf31c2f..5770c388dde 100644 --- a/packages/tui/src/mini/runtime.ts +++ b/packages/tui/src/mini/runtime.ts @@ -115,7 +115,6 @@ type ResolvedSession = { type RuntimeState = { sdk: RunInput["sdk"] shown: boolean - aborting: boolean model: RunInput["model"] defaultModel: RunInput["model"] providers: RunProvider[] @@ -213,7 +212,6 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep const state: RuntimeState = { sdk: ctx.sdk, shown: !session.first, - aborting: false, model: ctx.model ?? session.model, defaultModel: undefined, providers: [], @@ -368,20 +366,17 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep } }, onInterrupt: () => { - if (!state.sessionID || state.aborting) { + if (!state.sessionID) { return false } - state.aborting = true + // No in-flight guard: interruption acknowledges immediately server-side and repeating + // it is an idempotent no-op, so repeated presses are never swallowed. void ( state.stream ? state.stream.then((item) => item.handle.interruptActiveTurn()) : state.sdk.session.interrupt({ sessionID: state.sessionID, continue: true }) - ) - .catch(() => {}) - .finally(() => { - state.aborting = false - }) + ).catch(() => {}) return true }, onBackground: () => { diff --git a/packages/tui/src/mini/stream-v2.transport.ts b/packages/tui/src/mini/stream-v2.transport.ts index 8756bd05457..4ac952b4c1b 100644 --- a/packages/tui/src/mini/stream-v2.transport.ts +++ b/packages/tui/src/mini/stream-v2.transport.ts @@ -143,6 +143,8 @@ type State = { closed: boolean initial: boolean rootActive: boolean + /** Bumped on every root execution lifecycle event; guards paints against stale acks. */ + executionEpoch: number buffered?: ReplayBuffer errors: Set pending: Map @@ -474,6 +476,7 @@ export async function createSessionTransport(input: StreamInput): Promise subagents.close(), { once: true }) + // The one "go idle" transition, shared by settlement, terminal events, and the + // interrupt ack so the flag and the paint cannot drift apart. + const paintIdle = (status: string) => { + state.rootActive = false + write([], { phase: "idle", status }) + } + const write = (commits: StreamCommit[], patch?: { phase?: "idle" | "running"; status?: string; usage?: string }) => { if (state.closed || controller.signal.aborted || input.footer.isClosed) return if (!state.initial && state.buffered === undefined) @@ -790,8 +800,7 @@ export async function createSessionTransport(input: StreamInput): Promise { await client.session.wait({ sessionID: input.sessionID }, { signal: controller.signal }) for (const message of await projectedMessages(client, controller.signal)) renderMessage(message, true, true) - state.rootActive = false - write([], { phase: "idle", status: blockerStatus(state.view) }) + paintIdle(blockerStatus(state.view)) await input.footer.idle() } @@ -1297,6 +1306,7 @@ export async function createSessionTransport(input: StreamInput): Promise {}) + // Paint idle at the ack, not at settlement: the server accepts interruption immediately + // while cleanup finishes asynchronously, and the terminal execution event re-confirms. + // A failed request paints nothing, so the two-press gesture stays available for retry, + // and a lifecycle event racing the ack wins via the epoch guard. + const epoch = state.executionEpoch + await sdk.session.interrupt({ sessionID: input.sessionID, continue: true }).then( + () => { + if (state.executionEpoch === epoch) paintIdle(blockerStatus(state.view)) + }, + () => {}, + ) }, selectSubagent(sessionID) { subagents.select(sdk, sessionID)