diff --git a/packages/core/schema.json b/packages/core/schema.json index 9b774734a7..8388ce9cd2 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,8 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "15060ec5-05f7-4b86-b2a5-9108609432b3", - "prevIds": ["1551a157-8959-4ba9-a52b-4ea3b7b28cae"], + "id": "00924d88-1842-4d71-ac74-5682ddc47e1c", + "prevIds": ["15060ec5-05f7-4b86-b2a5-9108609432b3"], "ddl": [ { "name": "account_state", @@ -1302,6 +1302,16 @@ "entityType": "columns", "table": "session_v2" }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "resume_attempts", + "entityType": "columns", + "table": "session_v2" + }, { "type": "text", "notNull": false, diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 3a5257e50f..ad32ddafdb 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -40,6 +40,7 @@ import m37 from "./migration/20260622202450_simplify_session_input" import m38 from "./migration/20260804233008_loose_psylocke" import m39 from "./migration/20260805200742_import_legacy_credentials" import m40 from "./migration/20260808023530_workspace_domain" +import m41 from "./migration/20260811161259_execution_claim_attempts" export const migrations = [ m00, @@ -83,4 +84,5 @@ export const migrations = [ m38, m39, m40, + m41, ] satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260811161259_execution_claim_attempts.ts b/packages/core/src/database/migration/20260811161259_execution_claim_attempts.ts new file mode 100644 index 0000000000..2b025ee3fd --- /dev/null +++ b/packages/core/src/database/migration/20260811161259_execution_claim_attempts.ts @@ -0,0 +1,13 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +const migration: DatabaseMigration.Migration = { + id: "20260811161259_execution_claim_attempts", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`ALTER TABLE \`session_v2\` ADD \`resume_attempts\` integer DEFAULT 0 NOT NULL;`) + }) + }, +} + +export default migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 537a5c304f..7a1376fe9a 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -200,6 +200,7 @@ const schema: Omit = { \`time_compacting\` integer, \`time_archived\` integer, \`time_suspended\` integer, + \`resume_attempts\` integer DEFAULT 0 NOT NULL, CONSTRAINT \`fk_session_v2_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE ); `) diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 44a90a8492..acbbb5925a 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -56,16 +56,22 @@ export const layer = Layer.effect( ), Effect.asVoid, ) - // Starting or finishing on its own clears stale suspension; interruption preserves it because - // managed-server teardown suspends active Sessions immediately before interrupting their drains. - const clearSuspensionOnCommit = (sessionID: SessionSchema.ID) => ({ - commit: () => Effect.asVoid(store.consumeSuspended(sessionID)), + // Write-ahead claim: starting records the durable intent that a turn is in flight, in the same + // transaction as the started event. Terminals release it — except shutdown interruption, which + // preserves the claim so the next server start resumes the turn. A claim that survives with no + // terminal is the signature of a process that died without teardown (crash, SIGKILL, eviction); + // recovery is a property of the database, never of a shutdown hook that may not run. + const claimOnCommit = (sessionID: SessionSchema.ID) => ({ + commit: () => store.claim(sessionID), + }) + const releaseOnCommit = (sessionID: SessionSchema.ID) => ({ + commit: () => store.release(sessionID), }) const coordinator = yield* SessionRunCoordinator.make({ started: (sessionID) => reportLifecycle( sessionID, - bus.publish(SessionEvent.Execution.Started, { sessionID }, clearSuspensionOnCommit(sessionID)), + bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)), ), drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) { const session = yield* store.get(sessionID) @@ -86,11 +92,17 @@ export const layer = Layer.effect( Effect.gen(function* () { const outcome = terminal(exit, reason) if (outcome.type === "succeeded") { - yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID }, clearSuspensionOnCommit(sessionID)) + yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID }, releaseOnCommit(sessionID)) return } if (outcome.type === "interrupted") { - yield* bus.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: outcome.reason }) + // A user cancel (or a superseding execution) releases the claim: the turn must not + // resurrect at the next boot. Shutdown interruption keeps it for restart continuity. + yield* bus.publish( + SessionEvent.Execution.Interrupted, + { sessionID, reason: outcome.reason }, + outcome.reason === "shutdown" ? undefined : releaseOnCommit(sessionID), + ) return } yield* bus.publish( @@ -99,7 +111,7 @@ export const layer = Layer.effect( sessionID, error: outcome.error, }, - clearSuspensionOnCommit(sessionID), + releaseOnCommit(sessionID), ) }), ), diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index b46cf3dfdf..22a8912514 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -5,61 +5,111 @@ import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { Bus } from "../../bus" import { SessionEvent } from "../event" import { SessionExecution } from "../execution" +import { SessionSchema } from "../schema" import { SessionStore } from "../store" const CONTINUE_AFTER_SERVER_RESTART = "The server restarted while you were working. Continue from where you left off without repeating completed work." +const RESUME_EXHAUSTED = { + type: "aborted", + message: "Execution was interrupted repeatedly and will not be resumed automatically.", +} as const + +export interface Options { + /** + * Times a single turn may be resumed before it is terminalized instead. + * The counter is durable and only a terminal event resets it, so a turn + * that keeps dying cannot crash-loop across restarts. Turns that complete + * never accumulate: the budget is per-turn, not per-session. + */ + readonly maxAttempts?: number +} + +const DEFAULT_MAX_ATTEMPTS = 10 + export interface Interface { /** - * Marks every execution active in this process for resumption by the next server start. - * Call once new work has stopped arriving and before teardown interrupts the drains. + * Resumes Sessions whose execution claim was never released — turns orphaned + * by a process that died without teardown, or interrupted by a graceful + * shutdown (which preserves the claim on purpose). The claim is never + * cleared here: only a terminal event releases it, so a death anywhere in + * the resume path leaves the same orphaned claim for the next boot. */ - readonly suspendActiveSessions: Effect.Effect - /** Resumes suspended Sessions. Each suspension is consumed atomically, so a Session resumes at most once. */ readonly resumeSuspendedSessions: Effect.Effect } /** - * Restart continuity actions for the managed server. The service is inert until called: only the - * managed server invokes it, so default, embedded, and stdio servers never suspend or auto-resume. + * Recovery for orphaned executions. Claims are written at turn start by + * SessionExecution, so this sweep needs no cooperation from the previous + * process: crash, SIGKILL, isolate eviction, and graceful restart all leave + * the same durable signature. + * + * The sweep assumes every orphaned claim's owner is dead. The managed-server + * protocol guarantees this: a successor is only spawned after the previous + * process is confirmed dead (client service `kill`/`evict` poll the PID), the + * registration lock admits one managed server at a time, and unregistered + * servers sharing the database never sweep. The service is inert until called + * — the managed server invokes it at boot; embedders may call it from their + * own start-up. */ export class Service extends Context.Service()("@opencode/SessionRestart") {} -export const layer = Layer.effect( - Service, - Effect.gen(function* () { - const store = yield* SessionStore.Service - const execution = yield* SessionExecution.Service - const bus = yield* Bus.Service - return Service.of({ - suspendActiveSessions: Effect.gen(function* () { - yield* store.suspend(yield* execution.active) - }), - resumeSuspendedSessions: Effect.gen(function* () { - const sessions = yield* store.listSuspended() - yield* Effect.forEach( - sessions, - (sessionID) => - Effect.gen(function* () { - if (!(yield* store.consumeSuspended(sessionID))) return - yield* bus.publish(SessionEvent.Synthetic, { - sessionID, - text: CONTINUE_AFTER_SERVER_RESTART, - description: "Continuing after restart", - }) - // Drain failures are already logged and durably recorded by the execution layer. - yield* Effect.ignore(execution.resume(sessionID)) - }), - { concurrency: "unbounded", discard: true }, - ) - }), - }) - }), -) +export const layer = (options?: Options) => + Layer.effect( + Service, + Effect.gen(function* () { + const store = yield* SessionStore.Service + const execution = yield* SessionExecution.Service + const bus = yield* Bus.Service + const scope = yield* Effect.scope + const maxAttempts = options?.maxAttempts ?? DEFAULT_MAX_ATTEMPTS + + const resumeOne = Effect.fnUntraced(function* (sessionID: SessionSchema.ID) { + // Durable before the resume runs, so a crash inside the resumed turn is + // counted by the next sweep and the budget cannot be dodged. + const attempts = yield* store.countResume(sessionID) + if (attempts === undefined) return // the Session was deleted since listing + if (attempts > maxAttempts) { + // Terminalize instead: the release hook clears the claim and resets the + // counter atomically with the terminal event. + yield* bus.publish( + SessionEvent.Execution.Failed, + { sessionID, error: RESUME_EXHAUSTED }, + { commit: () => store.release(sessionID) }, + ) + return + } + yield* bus.publish(SessionEvent.Synthetic, { + sessionID, + text: CONTINUE_AFTER_SERVER_RESTART, + description: "Continuing after restart", + }) + // Forked into the service scope so boot never waits on resumed turns; + // resuming an already-live Session joins its execution. Drain failures + // are logged and durably recorded by the execution layer. + yield* execution.resume(sessionID).pipe(Effect.ignore, Effect.forkIn(scope)) + }) + + return Service.of({ + resumeSuspendedSessions: Effect.gen(function* () { + // Child claims never drive recovery (children are not resumed), so a + // dead child's claim is noise no terminal will ever release. Clearing + // is safe even against a live child: claims are recovery markers, not + // locks, and children are excluded from that recovery. + yield* store.releaseChildClaims + const active = yield* execution.active + // Sessions already draining in this process keep their claim; resuming + // them would only inject a stray continuation into a live turn. + const orphaned = (yield* store.listSuspended()).filter((sessionID) => !active.has(sessionID)) + yield* Effect.forEach(orphaned, resumeOne, { concurrency: "unbounded", discard: true }) + }), + }) + }), + ) export const node = makeGlobalNode({ service: Service, - layer, + layer: layer(), deps: [SessionStore.node, SessionExecution.node, Bus.node], }) diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 331325b1d3..11a1a3fa7c 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -58,7 +58,9 @@ export const SessionTable = sqliteTable( ...Timestamps, time_compacting: integer(), time_archived: integer(), + /** The execution claim timestamp (historical column name; see SessionStore.claim). */ time_suspended: integer(), + resume_attempts: integer().notNull().default(0), }, (table) => [ index("session_v2_project_idx").on(table.project_id), diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index 8a6058ca2a..1101b950ad 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -1,6 +1,6 @@ export * as SessionStore from "./store" -import { and, eq, inArray, isNotNull, isNull } from "drizzle-orm" +import { and, eq, isNotNull, isNull, sql } from "drizzle-orm" import { Context, Effect, Layer, Schema } from "effect" import { Database } from "../database/database" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" @@ -17,10 +17,32 @@ export interface Interface { readonly message: ( messageID: SessionMessage.ID, ) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.Info } | undefined> + /** + * Top-level Sessions holding an execution claim. Child (subagent) Sessions + * are excluded: a resumed parent re-runs its tool call and spawns fresh + * children, so resuming orphaned children would duplicate their work. + */ readonly listSuspended: () => Effect.Effect> - /** Clears suspension, reporting whether this caller consumed it. At most one concurrent caller receives true. */ - readonly consumeSuspended: (sessionID: Session.ID) => Effect.Effect - readonly suspend: (sessionIDs: Iterable) => Effect.Effect + /** + * Records the execution claim: the durable write-ahead intent that a turn is + * (or was) in flight. Set when execution starts; a claim that survives to the + * next boot marks a turn that never completed — its process crashed or shut + * down mid-turn. + */ + readonly claim: (sessionID: Session.ID) => Effect.Effect + /** Releases the claim and resets resume accounting. Terminal events call this on commit. */ + readonly release: (sessionID: Session.ID) => Effect.Effect + /** + * Clears orphaned child (subagent) claims. Children are never resumed + * independently, so a dead child's claim is noise no terminal will ever + * release. + */ + readonly releaseChildClaims: Effect.Effect + /** + * Durably counts one more resume of an orphaned claim, returning the new + * total — or undefined when the Session no longer exists. + */ + readonly countResume: (sessionID: Session.ID) => Effect.Effect } export class Service extends Context.Service()("@opencode/SessionStore") {} @@ -57,35 +79,52 @@ const layer = Layer.effect( return yield* db .select({ sessionID: SessionTable.id }) .from(SessionTable) - .where(isNotNull(SessionTable.time_suspended)) + .where(and(isNotNull(SessionTable.time_suspended), isNull(SessionTable.parent_id))) .all() .pipe( Effect.orDie, Effect.map((rows) => rows.map((row) => row.sessionID)), ) }), - consumeSuspended: Effect.fn("SessionStore.consumeSuspended")(function* (sessionID) { - return ( - (yield* db - .update(SessionTable) - .set({ time_suspended: null }) - .where(and(eq(SessionTable.id, sessionID), isNotNull(SessionTable.time_suspended))) - .returning({ sessionID: SessionTable.id }) - .get() - .pipe(Effect.orDie)) !== undefined - ) - }), - suspend: Effect.fn("SessionStore.suspend")(function* (sessionIDs) { - const ids = Array.from(sessionIDs) - if (ids.length === 0) return - // The null guard preserves the original suspension time if a Session is somehow suspended twice. + claim: Effect.fn("SessionStore.claim")(function* (sessionID) { + // The null guard makes re-claiming a still-claimed Session a zero-row + // no-op (a resumed turn re-claims through the same started hook). + // Claim bookkeeping never counts as user activity: time_updated is + // pinned so session ordering only moves on real changes. yield* db .update(SessionTable) - .set({ time_suspended: Date.now() }) - .where(and(inArray(SessionTable.id, ids), isNull(SessionTable.time_suspended))) + .set({ time_suspended: Date.now(), time_updated: sql`${SessionTable.time_updated}` }) + .where(and(eq(SessionTable.id, sessionID), isNull(SessionTable.time_suspended))) .run() .pipe(Effect.orDie) }), + release: Effect.fn("SessionStore.release")(function* (sessionID) { + yield* db + .update(SessionTable) + .set({ time_suspended: null, resume_attempts: 0, time_updated: sql`${SessionTable.time_updated}` }) + .where(eq(SessionTable.id, sessionID)) + .run() + .pipe(Effect.orDie) + }), + releaseChildClaims: db + .update(SessionTable) + .set({ time_suspended: null, resume_attempts: 0, time_updated: sql`${SessionTable.time_updated}` }) + .where(and(isNotNull(SessionTable.time_suspended), isNotNull(SessionTable.parent_id))) + .run() + .pipe(Effect.orDie, Effect.asVoid, Effect.withSpan("SessionStore.releaseChildClaims")), + countResume: Effect.fn("SessionStore.countResume")(function* (sessionID) { + const row = yield* db + .update(SessionTable) + .set({ + resume_attempts: sql`${SessionTable.resume_attempts} + 1`, + time_updated: sql`${SessionTable.time_updated}`, + }) + .where(eq(SessionTable.id, sessionID)) + .returning({ attempts: SessionTable.resume_attempts }) + .get() + .pipe(Effect.orDie) + return row?.attempts + }), }) }), ) diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index acdc4fc965..ce43f65d32 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -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.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 }) + expect(yield* store.listSuspended()).toEqual([parent]) + + // 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() + // Each drain signals once it runs; the claim commits before the drain starts. + const interruptedRunning = yield* Deferred.make() + const completedRunning = yield* Deferred.make() const release = yield* Deferred.make() 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("starts every suspended execution without waiting for earlier drains to finish", () => + 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() + 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 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() 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() + 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() + 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, - values: { time_suspended?: number } = {}, + values: Partial> = {}, ) { 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)), diff --git a/packages/core/test/v1-migration.test.ts b/packages/core/test/v1-migration.test.ts index b3ec583f69..d503ecf045 100644 --- a/packages/core/test/v1-migration.test.ts +++ b/packages/core/test/v1-migration.test.ts @@ -63,6 +63,7 @@ const session = ( time_compacting: 3, time_archived: null, time_suspended: null, + resume_attempts: 0, ...overrides, }) diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index 128685ed15..ece570ce66 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -247,12 +247,10 @@ function unavailable(status: Status.State) { } /** - * The managed server owns restart continuity: it resumes Sessions the previous server suspended and - * suspends its own active Sessions on graceful shutdown. Suspension runs while the drains are still - * alive: connections close first, this finalizer runs next, and Session execution teardown follows. + * The managed server owns restart continuity: at boot it resumes Sessions whose execution claim was + * never released. Claims are written when execution starts (see SessionExecution), so recovery covers + * graceful restarts and unclean deaths alike — no shutdown hook participates. */ const installRestartContinuity = Effect.fnUntraced(function* (restart: SessionRestart.Interface) { yield* Effect.forkScoped(restart.resumeSuspendedSessions) - // Registered after the fork so suspension observes still-running resumed drains during teardown. - yield* Effect.addFinalizer(() => restart.suspendActiveSessions) })