|
|
@@ -18,6 +18,7 @@ import { SessionRunner } from "@opencode-ai/core/session/runner"
|
|
|
import { 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"
|
|
|
import { testEffect } from "./lib/effect"
|
|
|
|
|
|
const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionStore.node])))
|
|
|
@@ -49,58 +50,90 @@ describe("SessionExecution lifecycle", () => {
|
|
|
})
|
|
|
})
|
|
|
|
|
|
- it.effect("atomically consumes each suspension at most once", () =>
|
|
|
+ it.effect("the sweep only lists claimed top-level Sessions", () =>
|
|
|
Effect.gen(function* () {
|
|
|
const database = yield* Database.Service
|
|
|
const store = yield* SessionStore.Service
|
|
|
- const first = Session.ID.make("ses_recover_first")
|
|
|
- const second = Session.ID.make("ses_recover_second")
|
|
|
- yield* seedSessions(database, [first, second], { time_suspended: Date.now() })
|
|
|
+ const parent = Session.ID.make("ses_recover_parent")
|
|
|
+ const child = Session.ID.make("ses_recover_child")
|
|
|
+ const idle = Session.ID.make("ses_recover_idle")
|
|
|
+ yield* seedSessions(database, [parent], { time_suspended: Date.now() })
|
|
|
+ yield* seedSessions(database, [idle])
|
|
|
+ // An orphaned child is never resumed: the resumed parent re-runs its
|
|
|
+ // tool call and spawns a fresh child instead.
|
|
|
+ yield* seedSessions(database, [child], { time_suspended: Date.now(), parent_id: parent })
|
|
|
+
|
|
|
+ expect(yield* store.listSuspended()).toEqual([parent])
|
|
|
|
|
|
- expect(yield* store.consumeSuspended(first)).toBe(true)
|
|
|
- expect(yield* store.consumeSuspended(first)).toBe(false)
|
|
|
- expect(yield* store.consumeSuspended(second)).toBe(true)
|
|
|
- expect(yield* suspensions(database)).toEqual({ [first]: false, [second]: false })
|
|
|
+ // The sweep clears orphaned child claims outright; parents keep theirs.
|
|
|
+ yield* store.releaseChildClaims
|
|
|
+ expect(yield* claims(database)).toEqual({ [parent]: true, [child]: false, [idle]: false })
|
|
|
}),
|
|
|
)
|
|
|
|
|
|
- it.effect("suspension survives teardown interruption and clears when a drain finishes on its own", () =>
|
|
|
+ it.effect("claims at execution start, releases on completion, and preserves through teardown", () =>
|
|
|
Effect.gen(function* () {
|
|
|
const database = yield* Database.Service
|
|
|
- const interrupted = Session.ID.make("ses_suspend_interrupted")
|
|
|
- const completed = Session.ID.make("ses_suspend_completed")
|
|
|
+ const interrupted = Session.ID.make("ses_claim_interrupted")
|
|
|
+ const completed = Session.ID.make("ses_claim_completed")
|
|
|
yield* seedSessions(database, [interrupted, completed])
|
|
|
|
|
|
- const draining = yield* Deferred.make<void>()
|
|
|
+ // Each drain signals once it runs; the claim commits before the drain starts.
|
|
|
+ const interruptedRunning = yield* Deferred.make<void>()
|
|
|
+ const completedRunning = yield* Deferred.make<void>()
|
|
|
const release = yield* Deferred.make<void>()
|
|
|
const scope = yield* Scope.make()
|
|
|
const context = yield* buildExecution(scope, ({ sessionID }) =>
|
|
|
sessionID === completed
|
|
|
- ? Deferred.await(release)
|
|
|
- : Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never)),
|
|
|
+ ? Deferred.succeed(completedRunning, undefined).pipe(Effect.andThen(Deferred.await(release)))
|
|
|
+ : Deferred.succeed(interruptedRunning, undefined).pipe(Effect.andThen(Effect.never)),
|
|
|
)
|
|
|
const execution = Context.get(context, SessionExecution.Service)
|
|
|
- const restart = Context.get(context, SessionRestart.Service)
|
|
|
yield* execution.resume(interrupted).pipe(Effect.forkScoped)
|
|
|
const completing = yield* execution.resume(completed).pipe(Effect.forkIn(scope))
|
|
|
- yield* Deferred.await(draining)
|
|
|
+ yield* Deferred.await(interruptedRunning)
|
|
|
+ yield* Deferred.await(completedRunning)
|
|
|
|
|
|
- yield* restart.suspendActiveSessions
|
|
|
- expect(yield* suspensions(database)).toEqual({ [interrupted]: true, [completed]: true })
|
|
|
+ // The write-ahead claim exists WHILE the turns run — no shutdown hook involved.
|
|
|
+ expect(yield* claims(database)).toEqual({ [interrupted]: true, [completed]: true })
|
|
|
|
|
|
- // A drain that finishes on its own after suspension clears its stale suspension.
|
|
|
+ // A drain that finishes on its own releases its claim.
|
|
|
yield* Deferred.succeed(release, undefined)
|
|
|
yield* Fiber.join(completing)
|
|
|
yield* execution.awaitIdle(completed)
|
|
|
- expect((yield* suspensions(database))[completed]).toBe(false)
|
|
|
+ expect((yield* claims(database))[completed]).toBe(false)
|
|
|
|
|
|
- // Teardown interruption preserves suspension for the next server start.
|
|
|
+ // Teardown interruption (graceful twin of an unclean death) preserves the claim
|
|
|
+ // for the next server start.
|
|
|
yield* Scope.close(scope, Exit.void)
|
|
|
- expect((yield* suspensions(database))[interrupted]).toBe(true)
|
|
|
+ expect((yield* claims(database))[interrupted]).toBe(true)
|
|
|
+ }),
|
|
|
+ )
|
|
|
+
|
|
|
+ it.effect("a user interrupt releases the claim so the turn never resurrects", () =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const database = yield* Database.Service
|
|
|
+ const sessionID = Session.ID.make("ses_claim_user_cancel")
|
|
|
+ yield* seedSessions(database, [sessionID])
|
|
|
+
|
|
|
+ const draining = yield* Deferred.make<void>()
|
|
|
+ const scope = yield* Scope.make()
|
|
|
+ yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
|
|
+ const context = yield* buildExecution(scope, () =>
|
|
|
+ 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)
|
|
|
+ expect((yield* claims(database))[sessionID]).toBe(true)
|
|
|
+
|
|
|
+ yield* execution.interrupt(sessionID)
|
|
|
+ yield* execution.awaitIdle(sessionID)
|
|
|
+ expect((yield* claims(database))[sessionID]).toBe(false)
|
|
|
}),
|
|
|
)
|
|
|
|
|
|
- it.effect("starts every suspended execution without waiting for earlier drains to finish", () =>
|
|
|
+ it.effect("starts every claimed execution without waiting for earlier drains to finish", () =>
|
|
|
Effect.gen(function* () {
|
|
|
const database = yield* Database.Service
|
|
|
const sessionIDs = Array.from({ length: 5 }, (_, index) => Session.ID.make(`ses_resume_concurrent_${index}`))
|
|
|
@@ -125,7 +158,7 @@ describe("SessionExecution lifecycle", () => {
|
|
|
}),
|
|
|
)
|
|
|
|
|
|
- it.effect("resumes each suspended Session at most once", () =>
|
|
|
+ it.effect("resumes each claimed Session at most once", () =>
|
|
|
Effect.gen(function* () {
|
|
|
const database = yield* Database.Service
|
|
|
const bus = yield* Bus.Service
|
|
|
@@ -134,14 +167,22 @@ describe("SessionExecution lifecycle", () => {
|
|
|
yield* seedSessions(database, [first, second], { time_suspended: Date.now() })
|
|
|
|
|
|
const drained: string[] = []
|
|
|
+ const bothDraining = yield* Deferred.make<void>()
|
|
|
const continued: SessionEvent.Synthetic[] = []
|
|
|
const scope = yield* Scope.make()
|
|
|
- const context = yield* buildExecution(scope, ({ sessionID }) => Effect.sync(() => void drained.push(sessionID)))
|
|
|
+ const context = yield* buildExecution(scope, ({ sessionID }) =>
|
|
|
+ Effect.sync(() => {
|
|
|
+ drained.push(sessionID)
|
|
|
+ if (drained.length === 2) Deferred.doneUnsafe(bothDraining, Effect.void)
|
|
|
+ }),
|
|
|
+ )
|
|
|
const execution = Context.get(context, SessionExecution.Service)
|
|
|
const restart = Context.get(context, SessionRestart.Service)
|
|
|
yield* bus.project(SessionEvent.Synthetic, (event) => Effect.sync(() => void continued.push(event)))
|
|
|
|
|
|
+ // The sweep forks resumed drains, so completion is observed through the executions.
|
|
|
yield* restart.resumeSuspendedSessions
|
|
|
+ yield* Deferred.await(bothDraining)
|
|
|
yield* Effect.forEach([first, second], execution.awaitIdle, { discard: true })
|
|
|
expect(drained.toSorted()).toEqual([first, second])
|
|
|
expect(continued.map((event) => event.data).toSorted((a, b) => a.sessionID.localeCompare(b.sessionID))).toEqual(
|
|
|
@@ -151,7 +192,9 @@ describe("SessionExecution lifecycle", () => {
|
|
|
description: "Continuing after restart",
|
|
|
})),
|
|
|
)
|
|
|
- expect(yield* suspensions(database)).toEqual({ [first]: false, [second]: false })
|
|
|
+ // Drains completed naturally, so claims are released and counters reset.
|
|
|
+ expect(yield* claims(database)).toEqual({ [first]: false, [second]: false })
|
|
|
+ expect(yield* attempts(database, first)).toBe(0)
|
|
|
|
|
|
yield* restart.resumeSuspendedSessions
|
|
|
expect(drained.length).toBe(2)
|
|
|
@@ -159,17 +202,106 @@ describe("SessionExecution lifecycle", () => {
|
|
|
yield* Scope.close(scope, Exit.void)
|
|
|
}),
|
|
|
)
|
|
|
+
|
|
|
+ it.effect("terminalizes a turn that exhausts its resume budget instead of crash-looping", () =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const database = yield* Database.Service
|
|
|
+ const bus = yield* Bus.Service
|
|
|
+ const sessionID = Session.ID.make("ses_resume_exhausted")
|
|
|
+ // A claim from a dead process, already resumed twice without completing.
|
|
|
+ yield* seedSessions(database, [sessionID], { time_suspended: Date.now(), resume_attempts: 2 })
|
|
|
+
|
|
|
+ const drained: string[] = []
|
|
|
+ const failures: SessionEvent.Execution.Failed[] = []
|
|
|
+ const scope = yield* Scope.make()
|
|
|
+ yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
|
|
+ const context = yield* buildExecution(
|
|
|
+ scope,
|
|
|
+ ({ sessionID: id }) => Effect.sync(() => void drained.push(id)),
|
|
|
+ { maxAttempts: 2 },
|
|
|
+ )
|
|
|
+ const restart = Context.get(context, SessionRestart.Service)
|
|
|
+ yield* bus.project(SessionEvent.Execution.Failed, (event) => Effect.sync(() => void failures.push(event)))
|
|
|
+
|
|
|
+ yield* restart.resumeSuspendedSessions
|
|
|
+ expect(drained).toEqual([])
|
|
|
+ expect(failures.map((event) => event.data.error.type)).toEqual(["aborted"])
|
|
|
+ // The terminal released the claim and reset the counter atomically.
|
|
|
+ expect(yield* claims(database)).toEqual({ [sessionID]: false })
|
|
|
+ expect(yield* attempts(database, sessionID)).toBe(0)
|
|
|
+ }),
|
|
|
+ )
|
|
|
+
|
|
|
+ it.effect("counts every resume durably and never consumes the claim it recovers", () =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const database = yield* Database.Service
|
|
|
+ const sessionID = Session.ID.make("ses_resume_counted")
|
|
|
+ yield* seedSessions(database, [sessionID], { time_suspended: Date.now() })
|
|
|
+
|
|
|
+ const draining = yield* Deferred.make<void>()
|
|
|
+ const scope = yield* Scope.make()
|
|
|
+ yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
|
|
+ // The drain never terminalizes (mirrors a process that will die mid-turn).
|
|
|
+ const context = yield* buildExecution(scope, () =>
|
|
|
+ Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never)),
|
|
|
+ )
|
|
|
+ const restart = Context.get(context, SessionRestart.Service)
|
|
|
+ yield* restart.resumeSuspendedSessions.pipe(Effect.forkIn(scope))
|
|
|
+ yield* Deferred.await(draining)
|
|
|
+
|
|
|
+ // The attempt is durable before the drain runs, and the claim is held
|
|
|
+ // throughout: a crash anywhere in the resume path leaves both intact.
|
|
|
+ expect(yield* attempts(database, sessionID)).toBe(1)
|
|
|
+ expect((yield* claims(database))[sessionID]).toBe(true)
|
|
|
+
|
|
|
+ // Teardown (a graceful shutdown's interrupt) preserves both, so the next
|
|
|
+ // boot counts attempt 2 against the same turn.
|
|
|
+ yield* Scope.close(scope, Exit.void)
|
|
|
+ expect((yield* claims(database))[sessionID]).toBe(true)
|
|
|
+ expect(yield* attempts(database, sessionID)).toBe(1)
|
|
|
+ }),
|
|
|
+ )
|
|
|
+
|
|
|
+ it.effect("the sweep leaves Sessions already draining in this process untouched", () =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const database = yield* Database.Service
|
|
|
+ const bus = yield* Bus.Service
|
|
|
+ const sessionID = Session.ID.make("ses_resume_local_active")
|
|
|
+ yield* seedSessions(database, [sessionID])
|
|
|
+
|
|
|
+ const draining = yield* Deferred.make<void>()
|
|
|
+ const continued: SessionEvent.Synthetic[] = []
|
|
|
+ const scope = yield* Scope.make()
|
|
|
+ yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
|
|
+ const context = yield* buildExecution(scope, () =>
|
|
|
+ Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never)),
|
|
|
+ )
|
|
|
+ const execution = Context.get(context, SessionExecution.Service)
|
|
|
+ const restart = Context.get(context, SessionRestart.Service)
|
|
|
+ yield* bus.project(SessionEvent.Synthetic, (event) => Effect.sync(() => void continued.push(event)))
|
|
|
+
|
|
|
+ // A live local turn holds a claim; the sweep must not count, continue, or terminalize it.
|
|
|
+ yield* execution.resume(sessionID).pipe(Effect.forkScoped)
|
|
|
+ yield* Deferred.await(draining)
|
|
|
+ yield* restart.resumeSuspendedSessions
|
|
|
+
|
|
|
+ expect(continued).toEqual([])
|
|
|
+ expect(yield* attempts(database, sessionID)).toBe(0)
|
|
|
+ expect((yield* claims(database))[sessionID]).toBe(true)
|
|
|
+ }),
|
|
|
+ )
|
|
|
})
|
|
|
|
|
|
function seedSessions(
|
|
|
database: Database.Service["Service"],
|
|
|
sessionIDs: ReadonlyArray<Session.ID>,
|
|
|
- values: { time_suspended?: number } = {},
|
|
|
+ values: Partial<Pick<typeof SessionTable.$inferInsert, "time_suspended" | "resume_attempts" | "parent_id">> = {},
|
|
|
) {
|
|
|
return Effect.gen(function* () {
|
|
|
yield* database.db
|
|
|
.insert(ProjectTable)
|
|
|
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
|
|
|
+ .onConflictDoNothing()
|
|
|
.run()
|
|
|
.pipe(Effect.orDie)
|
|
|
yield* database.db
|
|
|
@@ -190,19 +322,35 @@ function seedSessions(
|
|
|
})
|
|
|
}
|
|
|
|
|
|
-function suspensions(database: Database.Service["Service"]) {
|
|
|
+function claims(database: Database.Service["Service"]) {
|
|
|
return database.db
|
|
|
- .select({ id: SessionTable.id, suspended: SessionTable.time_suspended })
|
|
|
+ .select({ id: SessionTable.id, claimed: SessionTable.time_suspended })
|
|
|
.from(SessionTable)
|
|
|
.all()
|
|
|
.pipe(
|
|
|
Effect.orDie,
|
|
|
- Effect.map((rows) => Object.fromEntries(rows.map((row) => [row.id, row.suspended !== null]))),
|
|
|
+ Effect.map((rows) => Object.fromEntries(rows.map((row) => [row.id, row.claimed !== null]))),
|
|
|
+ )
|
|
|
+}
|
|
|
+
|
|
|
+function attempts(database: Database.Service["Service"], sessionID: Session.ID) {
|
|
|
+ return database.db
|
|
|
+ .select({ attempts: SessionTable.resume_attempts })
|
|
|
+ .from(SessionTable)
|
|
|
+ .where(eq(SessionTable.id, sessionID))
|
|
|
+ .get()
|
|
|
+ .pipe(
|
|
|
+ Effect.orDie,
|
|
|
+ Effect.map((row) => row?.attempts),
|
|
|
)
|
|
|
}
|
|
|
|
|
|
/** Builds the local execution layer plus the restart actions against the test harness services. */
|
|
|
-function buildExecution(scope: Scope.Closeable, drain: SessionRunner.Interface["drain"]) {
|
|
|
+function buildExecution(
|
|
|
+ scope: Scope.Closeable,
|
|
|
+ drain: SessionRunner.Interface["drain"],
|
|
|
+ options?: SessionRestart.Options,
|
|
|
+) {
|
|
|
return Effect.gen(function* () {
|
|
|
const database = yield* Database.Service
|
|
|
const bus = yield* Bus.Service
|
|
|
@@ -218,7 +366,7 @@ function buildExecution(scope: Scope.Closeable, drain: SessionRunner.Interface["
|
|
|
),
|
|
|
)
|
|
|
return yield* Layer.buildWithScope(
|
|
|
- SessionRestart.layer.pipe(
|
|
|
+ SessionRestart.layer(options).pipe(
|
|
|
Layer.provideMerge(SessionExecution.layer),
|
|
|
Layer.provide(Layer.succeed(Database.Service, database)),
|
|
|
Layer.provide(Layer.succeed(Bus.Service, bus)),
|
Kit Langton