vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-09-02 12:15:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(core): recover successors deferred by shutdown

Browse source 
Stop scheduling before the execution FiberSet closes. Preserve a pending successor claim after the previous terminal settles, without forking into a closed set or leaving ghost ownership.

Cover user-interruption cleanup, terminal settlement, sequential and parallel scope closure, and restart recovery of an admitted move.
This commit is contained in:
Kit Langton 2026-08-31 12:43:35 -04:00
parent afd7492018
commit d1ca3089ce
4 changed files with 227 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/core/src/session/execution.ts
View file

@ -109,6 +109,10 @@ export const layer = Layer.effect(
bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)),
),
drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable),
// Claim after the old terminal: user cancellation must release its claim and retry budget
// before a deferred successor becomes fresh recovery intent, without reporting a start.
// This retains ID-only recovery: restart does not preserve the drain's promotable scope.
suspended: (sessionID) => reportLifecycle(sessionID, store.claim(sessionID)),
// One terminal observation per busy period, covering every coalesced drain.
settled: (sessionID, exit, reason) =>
reportLifecycle(

31
packages/core/src/session/run-coordinator.ts
View file

@ -9,9 +9,9 @@ export interface Coordinator<Key, E, Reason = never> {
readonly active: Effect.Effect<ReadonlySet<Key>>
/** Checks ownership for one key, including cleanup and terminal settlement. */
readonly isActive: (key: Key) => Effect.Effect<boolean>
/** Starts an execution while idle, or joins the active execution and returns its exit. */
/** Starts while idle or joins the active execution. Interrupts new runs once shutdown begins. */
readonly run: (key: Key) => Effect.Effect<void, E>
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
/** Rings the doorbell: starts while idle or drains again before settling. No-op once shutdown begins. */
readonly wake: (key: Key, scope?: Promotable) => Effect.Effect<void>
/**
* Stops the active execution and clears its doorbell. No-op when idle. Resolves once the
@ -61,10 +61,19 @@ export const make = <Key, E, Reason = never>(options: {
* drain and before the execution settles (waiters resolve after it completes).
*/
readonly settled?: (key: Key, exit: Exit.Exit<void, E>, reason?: Reason) => Effect.Effect<void>
/** Preserves a pending successor when shutdown prevents starting it, after the terminal hook completes. */
readonly suspended?: (key: Key) => Effect.Effect<void>
}): Effect.Effect<Coordinator<Key, E, Reason>, never, Scope.Scope> =>
Effect.gen(function* () {
const executions = new Map<Key, Execution<E, Reason>>()
const fork = yield* FiberSet.makeRuntime<never, void, never>()
let closing = false
// Finalizers run in reverse order: stop scheduling before FiberSet closes and interrupts its owners.
yield* Effect.addFinalizer(() =>
Effect.sync(() => {
closing = true
}),
)
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
Effect.suspend(() => options.drain(key, force, execution.scope)).pipe(
@ -98,7 +107,16 @@ export const make = <Key, E, Reason = never>(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) =>
Effect.suspend(() => {
if (closing && execution.pendingWake)
return (options.suspended?.(key) ?? Effect.void).pipe(
Effect.ensuring(Effect.sync(() => settle(key, execution, exit))),
)
settle(key, execution, exit)
return Effect.void
}),
),
Effect.exit,
Effect.asVoid,
),
@ -107,9 +125,9 @@ export const make = <Key, E, Reason = never>(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.
// during failure or interruption cleanup) starts fresh work, unless shutdown suspends it.
const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
if (execution.pendingWake) start(key, false, execution.pendingWake)
if (execution.pendingWake && !closing) start(key, false, execution.pendingWake)
else executions.delete(key)
Deferred.doneUnsafe(execution.done, exit)
}
@ -118,6 +136,7 @@ export const make = <Key, E, Reason = never>(options: {
const run = (key: Key): Effect.Effect<void, E> =>
Effect.suspend(() => {
if (closing) return Effect.interrupt
const execution = executions.get(key)
if (execution !== undefined) {
// A stopping execution refuses joiners: wait out its cleanup, then run fresh.
@ -130,6 +149,7 @@ export const make = <Key, E, Reason = never>(options: {
const wake = (key: Key, scope: Promotable = "input") =>
Effect.sync(() => {
if (closing) return
const execution = executions.get(key)
if (execution !== undefined) {
// Coalesced wakes keep the widest scope: "input" subsumes "steer".
@ -141,6 +161,7 @@ export const make = <Key, E, Reason = never>(options: {
const interrupt = (key: Key, reason?: Reason): Effect.Effect<boolean> =>
Effect.sync(() => {
if (closing) return false
const execution = executions.get(key)
if (execution === undefined || execution.stopping) return false
if (execution.owner === undefined) {

110
packages/core/test/session-execution.test.ts
View file

@ -6,6 +6,7 @@ import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Bus } from "@opencode-ai/core/bus"
import { Job } from "@opencode-ai/core/job"
import { KV } from "@opencode-ai/core/kv"
import { Location } from "@opencode-ai/core/location"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import type { LocationServices } from "@opencode-ai/core/location-services"
import { Project } from "@opencode-ai/core/project"
@ -18,12 +19,15 @@ 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 { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionRunner } from "@opencode-ai/core/session/runner/index"
import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql"
import { SessionStore } from "@opencode-ai/core/session/store"
import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect"
import { eq } from "drizzle-orm"
import { testEffect } from "./lib/effect"
import { globalProjectNode } from "./lib/project"
import { tmpdirScoped } from "./fixture/tmpdir"
const it = testEffect(
AppNodeBuilder.build(
@ -159,6 +163,112 @@ describe("SessionExecution lifecycle", () => {
}),
)
it.live("recovers a move admitted during user-interruption cleanup when shutdown prevents its successor", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const bus = yield* Bus.Service
const store = yield* SessionStore.Service
const admission = yield* SessionInbox.Service
const jobs = yield* Job.Service
const destination = AbsolutePath.make((yield* tmpdirScoped()).path)
const sessionID = Session.ID.make("ses_shutdown_successor")
yield* seedSessions(database, [sessionID], { resume_attempts: 2 })
const lifecycle: string[] = []
yield* bus.project(SessionEvent.Execution.Started, () => Effect.sync(() => void lifecycle.push("started")))
yield* bus.project(SessionEvent.Execution.Interrupted, (event) =>
Effect.sync(() => void lifecycle.push(event.data.reason)),
)
const draining = yield* Deferred.make<void>()
const cleanup = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() =>
Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))),
)
const drains: string[] = []
const context = yield* buildExecution(scope, () =>
Effect.sync(() => void drains.push("original")).pipe(
Effect.andThen(Deferred.succeed(draining, undefined)),
Effect.andThen(Effect.never),
Effect.onInterrupt(() => Deferred.succeed(cleanup, undefined).pipe(Effect.andThen(Deferred.await(release)))),
),
)
const execution = Context.get(context, SessionExecution.Service)
const sessions = Context.get(
yield* Layer.buildWithScope(
AppNodeBuilder.build(Session.node, [
[Database.node, Layer.succeed(Database.Service, database)],
[Bus.node, Layer.succeed(Bus.Service, bus)],
[SessionStore.node, Layer.succeed(SessionStore.Service, store)],
[SessionInbox.node, Layer.succeed(SessionInbox.Service, admission)],
[Job.node, Layer.succeed(Job.Service, jobs)],
// The shared Bus already has the production Session projectors.
[SessionProjector.node, Layer.empty],
[Project.node, globalProjectNode],
[SessionExecution.node, Layer.succeed(SessionExecution.Service, execution)],
[
LocationServiceMap.node,
Layer.effect(
LocationServiceMap.Service,
LayerMap.make(
(ref: Location.Ref) =>
// Move validation only needs Location from the destination graph.
// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
LayerNode.compile(Location.boundNode(ref), [
[Project.node, globalProjectNode],
]) as unknown as Layer.Layer<LocationServices>,
),
),
],
]).pipe(Layer.fresh),
scope,
),
Session.Service,
)
yield* execution.wake(sessionID)
yield* Deferred.await(draining)
yield* sessions.interrupt(sessionID)
yield* Deferred.await(cleanup)
yield* sessions.move({ sessionID, directory: destination })
expect(yield* sessions.inbox(sessionID)).toMatchObject([{ type: "move" }])
const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true }))
yield* Effect.yieldNow
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(closing)
expect({ active: yield* execution.isActive(sessionID), claimed: (yield* claims(database))[sessionID] }).toEqual({
active: false,
claimed: true,
})
yield* execution.awaitIdle(sessionID)
expect(drains).toEqual(["original"])
expect(lifecycle).toEqual(["started", "user"])
// The interrupted intent releases its old recovery budget; the pending move is new work.
expect(yield* attempts(database, sessionID)).toBe(0)
const restartedScope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(restartedScope, Exit.void))
const resumed = yield* Deferred.make<void>()
const pending: SessionInbox.Item["type"][] = []
const restarted = yield* buildExecution(restartedScope, () =>
Effect.gen(function* () {
drains.push("restarted")
pending.push(...(yield* SessionInbox.list(database.db, sessionID)).map((item) => item.type))
yield* Deferred.succeed(resumed, undefined)
}),
)
yield* Context.get(restarted, SessionRestart.Service).resumeSuspendedSessions
yield* Deferred.await(resumed)
yield* Context.get(restarted, SessionExecution.Service).awaitIdle(sessionID)
expect(drains).toEqual(["original", "restarted"])
expect(pending).toEqual(["move"])
expect((yield* claims(database))[sessionID]).toBe(false)
}),
)
it.effect("does not resume a user-cancelled background child whose notification was not admitted", () =>
Effect.gen(function* () {
const database = yield* Database.Service

88
packages/core/test/session-run-coordinator.test.ts
View file

@ -1,5 +1,5 @@
import { describe, expect } from "bun:test"
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
import { Cause, Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
import { testEffect } from "./lib/effect"
@ -158,6 +158,92 @@ describe("SessionRunCoordinator", () => {
}),
)
for (const strategy of ["sequential", "parallel"] as const) {
for (const pending of [false, true]) {
it.effect(
`${strategy} shutdown ${pending ? "suspends a cleanup-era wake" : "leaves cancelled work stopped"}`,
() =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const cleanup = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scope = yield* Scope.make(strategy)
yield* Effect.addFinalizer(() =>
Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))),
)
const lifecycle: Array<string | undefined> = []
const coordinator = yield* SessionRunCoordinator.make<string, never, string>({
drain: () =>
Deferred.succeed(started, undefined).pipe(
Effect.andThen(Effect.never),
Effect.onInterrupt(() =>
Deferred.succeed(cleanup, undefined).pipe(Effect.andThen(Deferred.await(release))),
),
),
started: () => Effect.sync(() => void lifecycle.push("started")),
settled: (_key, _exit, reason) => Effect.sync(() => void lifecycle.push(reason)),
suspended: () => Effect.sync(() => void lifecycle.push("suspended")),
}).pipe(Scope.provide(scope))
yield* coordinator.wake("session")
yield* Deferred.await(started)
yield* coordinator.interrupt("session", "user")
yield* Deferred.await(cleanup)
if (pending) yield* coordinator.wake("session")
const idle = yield* coordinator.awaitIdle("session").pipe(Effect.forkChild)
const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true }))
yield* Effect.yieldNow
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(closing)
expect(yield* coordinator.active).toEqual(new Set())
yield* Fiber.join(idle)
expect(lifecycle).toEqual(pending ? ["started", "user", "suspended"] : ["started", "user"])
// A retained reference to the closed coordinator must not create ghost ownership.
yield* coordinator.wake("session")
expect(yield* coordinator.interrupt("session", "user")).toBe(false)
const exit = yield* coordinator.run("session").pipe(Effect.exit)
expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true)
expect(yield* coordinator.active).toEqual(new Set())
}),
)
}
}
it.effect("suspends a settlement-window wake after its terminal hook completes", () =>
Effect.gen(function* () {
const settling = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() =>
Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))),
)
const lifecycle: string[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: () => Effect.sync(() => void lifecycle.push("drained")),
settled: () =>
Deferred.succeed(settling, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Effect.andThen(Effect.sync(() => void lifecycle.push("settled"))),
),
suspended: () => Effect.sync(() => void lifecycle.push("suspended")),
}).pipe(Scope.provide(scope))
yield* coordinator.wake("session")
yield* Deferred.await(settling)
yield* coordinator.wake("session")
const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true }))
yield* Effect.yieldNow
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(closing)
expect(lifecycle).toEqual(["drained", "settled", "suspended"])
expect(yield* coordinator.active).toEqual(new Set())
yield* coordinator.awaitIdle("session")
}),
)
it.effect("coalesces wakes received during active execution", () =>
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()