diff --git a/packages/cli/src/server-process.ts b/packages/cli/src/server-process.ts index 39c1e256d3..9f688fb16e 100644 --- a/packages/cli/src/server-process.ts +++ b/packages/cli/src/server-process.ts @@ -65,6 +65,7 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { hostname: options.hostname ?? config.hostname ?? "127.0.0.1", port: Option.fromNullishOr(options.port ?? config.port), password, + restartContinuity: options.mode === "service", }).pipe(Effect.provide(Logger.layer([], { mergeWithExisting: false }))) if (lockScope !== undefined) { yield* register(address, password) diff --git a/packages/cli/test/service.test.ts b/packages/cli/test/service.test.ts index ac1f9ac1cf..a793ce0257 100644 --- a/packages/cli/test/service.test.ts +++ b/packages/cli/test/service.test.ts @@ -1,8 +1,17 @@ import { NodeFileSystem } from "@effect/platform-node" import { Service } from "@opencode-ai/client/effect" +import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { EventTable } from "@opencode-ai/core/event/sql" import { Global } from "@opencode-ai/core/global" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionTable } from "@opencode-ai/core/session/sql" import { expect, test } from "bun:test" -import { Effect, Schema } from "effect" +import { Effect, Schedule, Schema } from "effect" import fs from "node:fs/promises" import os from "node:os" import path from "node:path" @@ -28,16 +37,42 @@ test("local channel stores service config with the local service filename", asyn test("concurrent service processes elect one server", async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-election-")) + const database = path.join(root, "opencode.db") const env = { ...process.env, HOME: root, - OPENCODE_DB: path.join(root, "opencode.db"), + OPENCODE_DB: database, OPENCODE_TEST_HOME: root, XDG_CACHE_HOME: path.join(root, "cache"), XDG_CONFIG_HOME: path.join(root, "config"), XDG_DATA_HOME: path.join(root, "data"), XDG_STATE_HOME: path.join(root, "state"), } + const sessionID = SessionV2.ID.make("ses_service_recovery") + await withDatabase( + database, + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make(root), sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "recovery", + directory: root, + title: "recovery", + version: "test", + time_suspended: Date.now(), + }) + .run() + .pipe(Effect.orDie) + }), + ) const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"] const first = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) const second = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) @@ -51,6 +86,20 @@ test("concurrent service processes elect one server", async () => { expect(exited).toBe(true) expect(winner.exitCode).toBe(null) + expect( + await withDatabase( + database, + Effect.gen(function* () { + const { db } = yield* Database.Service + return yield* db + .select({ timeSuspended: SessionTable.time_suspended }) + .from(SessionTable) + .get() + .pipe(Effect.orDie) + }), + ), + ).toEqual({ timeSuspended: null }) + expect(await waitForExecutionStart(database, sessionID)).toBe(1) } finally { first.kill("SIGTERM") second.kill("SIGTERM") @@ -59,6 +108,40 @@ test("concurrent service processes elect one server", async () => { } }) +function withDatabase(file: string, effect: Effect.Effect) { + return Effect.runPromise(effect.pipe(Effect.provide(Database.layerFromPath(file)), Effect.scoped)) +} + +function waitForExecutionStart(file: string, sessionID: SessionV2.ID) { + return withDatabase( + file, + Effect.gen(function* () { + const { db } = yield* Database.Service + return yield* db + .select({ id: EventTable.id, sessionID: EventTable.aggregate_id, type: EventTable.type }) + .from(EventTable) + .all() + .pipe( + Effect.orDie, + Effect.map((rows) => + rows.filter( + (row) => + row.sessionID === sessionID && + row.type === + EventV2.versionedType( + SessionEvent.Execution.Started.type, + SessionEvent.Execution.Started.durable.version, + ), + ), + ), + Effect.filterOrFail((rows) => rows.length > 0), + Effect.map((rows) => rows.length), + Effect.retry(Schedule.spaced("50 millis").pipe(Schedule.both(Schedule.recurs(200)))), + ) + }), + ) +} + async function waitForInfo(file: string) { for (let attempt = 0; attempt < 200; attempt++) { const value = await Bun.file(file) diff --git a/packages/core/schema.json b/packages/core/schema.json index 77884cc9ff..06cec4eaed 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,10 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "8c1748cc-f978-4df4-879d-fe7fda5c4c34", - "prevIds": [ - "beb91b32-23e2-405f-99b8-1e8763db94f8" - ], + "id": "01451b27-1e51-4657-b2d0-4b457dffa3ec", + "prevIds": ["8c1748cc-f978-4df4-879d-fe7fda5c4c34"], "ddl": [ { "name": "workspace", @@ -1442,6 +1440,16 @@ "entityType": "columns", "table": "session" }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_suspended", + "entityType": "columns", + "table": "session" + }, { "type": "text", "notNull": false, @@ -1503,13 +1511,9 @@ "table": "session_share" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1518,13 +1522,9 @@ "table": "workspace" }, { - "columns": [ - "active_account_id" - ], + "columns": ["active_account_id"], "tableTo": "account", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "SET NULL", "nameExplicit": false, @@ -1533,13 +1533,9 @@ "table": "account_state" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "tableTo": "event_sequence", - "columnsTo": [ - "aggregate_id" - ], + "columnsTo": ["aggregate_id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1548,13 +1544,9 @@ "table": "event" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1563,13 +1555,9 @@ "table": "permission" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1578,13 +1566,9 @@ "table": "project_directory" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1593,13 +1577,9 @@ "table": "instruction_checkpoint" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1608,13 +1588,9 @@ "table": "instruction_entry" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1623,13 +1599,9 @@ "table": "message" }, { - "columns": [ - "message_id" - ], + "columns": ["message_id"], "tableTo": "message", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1638,13 +1610,9 @@ "table": "part" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1653,13 +1621,9 @@ "table": "session_input" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1668,13 +1632,9 @@ "table": "session_message" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1683,13 +1643,9 @@ "table": "session" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1698,174 +1654,133 @@ "table": "session_share" }, { - "columns": [ - "email", - "url" - ], + "columns": ["email", "url"], "nameExplicit": false, "name": "control_account_pk", "entityType": "pks", "table": "control_account" }, { - "columns": [ - "project_id", - "directory" - ], + "columns": ["project_id", "directory"], "nameExplicit": false, "name": "project_directory_pk", "entityType": "pks", "table": "project_directory" }, { - "columns": [ - "session_id", - "key" - ], + "columns": ["session_id", "key"], "nameExplicit": false, "name": "instruction_entry_pk", "entityType": "pks", "table": "instruction_entry" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "workspace_pk", "table": "workspace", "entityType": "pks" }, { - "columns": [ - "name" - ], + "columns": ["name"], "nameExplicit": false, "name": "data_migration_pk", "table": "data_migration", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_state_pk", "table": "account_state", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_pk", "table": "account", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "credential_pk", "table": "credential", "entityType": "pks" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "nameExplicit": false, "name": "event_sequence_pk", "table": "event_sequence", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "event_pk", "table": "event", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "permission_pk", "table": "permission", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "project_pk", "table": "project", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "instruction_checkpoint_pk", "table": "instruction_checkpoint", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "message_pk", "table": "message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "part_pk", "table": "part", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_input_pk", "table": "session_input", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_message_pk", "table": "session_message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_pk", "table": "session", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "session_share_pk", "table": "session_share", @@ -2180,7 +2095,21 @@ "name": "session_parent_idx", "entityType": "indexes", "table": "session" + }, + { + "columns": [ + { + "value": "time_suspended", + "isExpression": false + } + ], + "isUnique": false, + "where": "\"session\".\"time_suspended\" is not null", + "origin": "manual", + "name": "session_time_suspended_idx", + "entityType": "indexes", + "table": "session" } ], "renames": [] -} \ No newline at end of file +} diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 70cbfca731..9f5c5b40a3 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -51,5 +51,6 @@ export const migrations = ( import("./migration/20260707120000_migrate_prelaunch_v2_state"), import("./migration/20260709013000_generic_session_input"), import("./migration/20260709025533_drop-todo"), + import("./migration/20260709163752_time_suspended"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260709163752_time_suspended.ts b/packages/core/src/database/migration/20260709163752_time_suspended.ts new file mode 100644 index 0000000000..70c8f136a8 --- /dev/null +++ b/packages/core/src/database/migration/20260709163752_time_suspended.ts @@ -0,0 +1,14 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260709163752_time_suspended", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`ALTER TABLE \`session\` ADD \`time_suspended\` integer;`) + yield* tx.run( + `CREATE INDEX \`session_time_suspended_idx\` ON \`session\` (\`time_suspended\`) WHERE "session"."time_suspended" is not null;`, + ) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index d175f7cf7a..27651d0c13 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -224,6 +224,7 @@ export default { \`time_updated\` integer NOT NULL, \`time_compacting\` integer, \`time_archived\` integer, + \`time_suspended\` integer, CONSTRAINT \`fk_session_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE ); `) @@ -273,6 +274,9 @@ export default { yield* tx.run(`CREATE INDEX \`session_project_idx\` ON \`session\` (\`project_id\`);`) yield* tx.run(`CREATE INDEX \`session_workspace_idx\` ON \`session\` (\`workspace_id\`);`) yield* tx.run(`CREATE INDEX \`session_parent_idx\` ON \`session\` (\`parent_id\`);`) + yield* tx.run( + `CREATE INDEX \`session_time_suspended_idx\` ON \`session\` (\`time_suspended\`) WHERE "session"."time_suspended" is not null;`, + ) }) }, } satisfies Omit diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 5dbba30776..3b45ba25c9 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -1,10 +1,16 @@ export * as SessionExecution from "./execution" -import { Context, Effect, Layer } from "effect" -import { LayerNode } from "../effect/layer-node" -import { Node } from "../effect/app-node" +import { Cause, Context, Effect, Exit, Layer } from "effect" +import { EventV2 } from "../event" +import { LocationServiceMap } from "../location-service-map" +import { makeGlobalNode } from "../effect/app-node" +import { SessionEvent } from "./event" +import { SessionRunCoordinator } from "./run-coordinator" import { SessionRunner } from "./runner/index" import { SessionSchema } from "./schema" +import { SessionStore } from "./store" +import { toSessionError } from "./to-session-error" +import { UserInterruptedError } from "./error" export interface Interface { /** Snapshots active execution owned by this process. */ @@ -22,7 +28,98 @@ export interface Interface { /** Routes execution from a Session ID to the runner owned by that Session's Location. */ export class Service extends Context.Service()("@opencode/v2/SessionExecution") {} -export const node = LayerNode.unbound(Service, Node.tags.values.global) +type InterruptReason = "user" | "shutdown" | "superseded" + +export function terminal(exit: Exit.Exit, reason?: InterruptReason) { + if (Exit.isSuccess(exit)) return { type: "succeeded" as const } + if (Cause.hasInterrupts(exit.cause)) return { type: "interrupted" as const, reason: reason ?? "shutdown" } + const failure = Cause.squash(exit.cause) + if (failure instanceof UserInterruptedError) return { type: "interrupted" as const, reason: "user" as const } + return { type: "failed" as const, error: toSessionError(failure) } +} + +/** Process-local execution: drains run in this process, routed through the Session's Location graph. */ +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const store = yield* SessionStore.Service + const locations = yield* LocationServiceMap.Service + const events = yield* EventV2.Service + const reportLifecycle = (sessionID: SessionSchema.ID, effect: Effect.Effect) => + effect.pipe( + Effect.tapCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.void + : Effect.logError("Failed to publish Session execution lifecycle", cause).pipe( + Effect.annotateLogs({ sessionID }), + ), + ), + 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)), + }) + const coordinator = yield* SessionRunCoordinator.make({ + started: (sessionID) => + reportLifecycle( + sessionID, + events.publish(SessionEvent.Execution.Started, { sessionID }, clearSuspensionOnCommit(sessionID)), + ), + drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) { + const session = yield* store.get(sessionID) + if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) + return yield* SessionRunner.Service.use((runner) => runner.drain({ sessionID, force })).pipe( + Effect.provide(locations.get(session.location)), + Effect.tapCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.void + : Effect.logError("Failed to drain Session", cause).pipe(Effect.annotateLogs({ sessionID })), + ), + ) + }), + // One terminal observation per busy period, covering every coalesced drain. + settled: (sessionID, exit, reason) => + reportLifecycle( + sessionID, + Effect.gen(function* () { + const outcome = terminal(exit, reason) + if (outcome.type === "succeeded") { + yield* events.publish(SessionEvent.Execution.Succeeded, { sessionID }, clearSuspensionOnCommit(sessionID)) + return + } + if (outcome.type === "interrupted") { + yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: outcome.reason }) + return + } + yield* events.publish( + SessionEvent.Execution.Failed, + { + sessionID, + error: outcome.error, + }, + clearSuspensionOnCommit(sessionID), + ) + }), + ), + }) + + return Service.of({ + active: coordinator.active, + interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"), + resume: coordinator.run, + wake: coordinator.wake, + awaitIdle: coordinator.awaitIdle, + }) + }), +) + +export const node = makeGlobalNode({ + service: Service, + layer, + deps: [SessionStore.node, LocationServiceMap.node, EventV2.node], +}) /** Low-level compatibility layer for callers that only need durable Session recording. */ export const noopLayer = Layer.succeed( diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts deleted file mode 100644 index c925ee655c..0000000000 --- a/packages/core/src/session/execution/local.ts +++ /dev/null @@ -1,96 +0,0 @@ -import { Cause, Effect, Exit, Layer } from "effect" -import { EventV2 } from "../../event" -import { LocationServiceMap } from "../../location-service-map" -import { makeGlobalNode } from "../../effect/app-node" -import { SessionEvent } from "../event" -import { SessionRunCoordinator } from "../run-coordinator" -import { SessionRunner } from "../runner" -import { SessionSchema } from "../schema" -import { SessionStore } from "../store" -import { SessionExecution } from "../execution" -import { toSessionError } from "../to-session-error" -import { UserInterruptedError } from "../error" - -export function terminal(exit: Exit.Exit, reason?: "user" | "shutdown" | "superseded") { - if (Exit.isSuccess(exit)) return { type: "succeeded" as const } - if (Cause.hasInterrupts(exit.cause)) return { type: "interrupted" as const, reason: reason ?? "shutdown" } - const failure = Cause.squash(exit.cause) - if (failure instanceof UserInterruptedError) return { type: "interrupted" as const, reason: "user" as const } - return { type: "failed" as const, error: toSessionError(failure) } -} - -/** Current-process routing for implicit-local Locations. Future remote placement belongs here. */ -const layer = Layer.effect( - SessionExecution.Service, - Effect.gen(function* () { - const store = yield* SessionStore.Service - const locations = yield* LocationServiceMap.Service - const events = yield* EventV2.Service - const reportLifecycle = (sessionID: SessionSchema.ID, effect: Effect.Effect) => - effect.pipe( - Effect.tapCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.void - : Effect.logError("Failed to publish Session execution lifecycle", cause).pipe( - Effect.annotateLogs({ sessionID }), - ), - ), - Effect.asVoid, - ) - const coordinator = yield* SessionRunCoordinator.make< - SessionSchema.ID, - SessionRunner.RunError, - "user" | "shutdown" | "superseded" - >({ - started: (sessionID) => reportLifecycle(sessionID, events.publish(SessionEvent.Execution.Started, { sessionID })), - drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) { - const session = yield* store.get(sessionID) - if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) - return yield* SessionRunner.Service.use((runner) => runner.drain({ sessionID, force })).pipe( - Effect.provide(locations.get(session.location)), - Effect.tapCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.void - : Effect.logError("Failed to drain Session", cause).pipe(Effect.annotateLogs({ sessionID })), - ), - ) - }), - // One terminal observation per busy period, covering every coalesced drain. - settled: (sessionID, exit, reason) => - reportLifecycle( - sessionID, - Effect.gen(function* () { - const outcome = terminal(exit, reason) - if (outcome.type === "succeeded") { - yield* events.publish(SessionEvent.Execution.Succeeded, { sessionID }) - return - } - if (outcome.type === "interrupted") { - yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: outcome.reason }) - return - } - yield* events.publish(SessionEvent.Execution.Failed, { - sessionID, - error: outcome.error, - }) - }), - ), - }) - - return SessionExecution.Service.of({ - active: coordinator.active, - interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"), - resume: coordinator.run, - wake: coordinator.wake, - awaitIdle: coordinator.awaitIdle, - }) - }), -) - -export const node = makeGlobalNode({ - service: SessionExecution.Service, - layer, - deps: [SessionStore.node, LocationServiceMap.node, EventV2.node], -}) - -export * as SessionExecutionLocal from "./local" diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts new file mode 100644 index 0000000000..eb75a7bf42 --- /dev/null +++ b/packages/core/src/session/execution/restart.ts @@ -0,0 +1,51 @@ +export * as SessionRestart from "./restart" + +import { Context, Effect, Layer } from "effect" +import { makeGlobalNode } from "../../effect/app-node" +import { SessionExecution } from "../execution" +import { SessionStore } from "../store" + +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. + */ + 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. + */ +export class Service extends Context.Service()("@opencode/v2/SessionRestart") {} + +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const store = yield* SessionStore.Service + const execution = yield* SessionExecution.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 + // Drain failures are already logged and durably recorded by the execution layer. + yield* Effect.ignore(execution.resume(sessionID)) + }), + // Each suspension is consumed atomically right before its drain; at most four drains run at once. + { concurrency: 4, discard: true }, + ) + }), + }) + }), +) + +export const node = makeGlobalNode({ service: Service, layer, deps: [SessionStore.node, SessionExecution.node] }) diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index bdc7b07967..f5e97aad76 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -61,11 +61,15 @@ export const SessionTable = sqliteTable( ...Timestamps, time_compacting: integer(), time_archived: integer(), + time_suspended: integer(), }, (table) => [ index("session_project_idx").on(table.project_id), index("session_workspace_idx").on(table.workspace_id), index("session_parent_idx").on(table.parent_id), + index("session_time_suspended_idx") + .on(table.time_suspended) + .where(sql`${table.time_suspended} is not null`), ], ) diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index 24c99cf515..516159c1e4 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 { eq } from "drizzle-orm" +import { and, eq, inArray, isNotNull, isNull } from "drizzle-orm" import { Context, Effect, Layer, Schema } from "effect" import { Database } from "../database/database" import { makeGlobalNode } from "../effect/app-node" @@ -17,6 +17,10 @@ export interface Interface { readonly message: ( messageID: SessionMessage.ID, ) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.Info } | undefined> + 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 } export class Service extends Context.Service()("@opencode/v2/SessionStore") {} @@ -49,6 +53,39 @@ const layer = Layer.effect( } : undefined }), + listSuspended: Effect.fn("SessionStore.listSuspended")(function* () { + return yield* db + .select({ sessionID: SessionTable.id }) + .from(SessionTable) + .where(isNotNull(SessionTable.time_suspended)) + .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. + yield* db + .update(SessionTable) + .set({ time_suspended: Date.now() }) + .where(and(inArray(SessionTable.id, ids), isNull(SessionTable.time_suspended))) + .run() + .pipe(Effect.orDie) + }), }) }), ) diff --git a/packages/core/test/database-migration.test.ts b/packages/core/test/database-migration.test.ts index 14b9983598..22a13a1c04 100644 --- a/packages/core/test/database-migration.test.ts +++ b/packages/core/test/database-migration.test.ts @@ -21,6 +21,7 @@ import migratePrelaunchV2StateMigration from "@opencode-ai/core/database/migrati import genericSessionInputMigration from "@opencode-ai/core/database/migration/20260709013000_generic_session_input" import renameInstructionsMigration from "@opencode-ai/core/database/migration/20260705180000_rename_instructions" import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork" +import timeSuspendedMigration from "@opencode-ai/core/database/migration/20260709163752_time_suspended" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2 } from "@opencode-ai/core/event" @@ -445,6 +446,28 @@ describe("DatabaseMigration", () => { ) }) + test("does not infer restart continuation from historical shutdown events", async () => { + await run( + Effect.gen(function* () { + const db = yield* makeDb + yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`) + yield* db.run( + sql`CREATE TABLE event (aggregate_id text NOT NULL, seq integer NOT NULL, type text NOT NULL, data text NOT NULL)`, + ) + yield* db.run(sql`INSERT INTO session VALUES ('ses_shutdown')`) + yield* db.run( + sql`INSERT INTO event VALUES ('ses_shutdown', 0, 'session.execution.interrupted.1', '{"reason":"shutdown"}')`, + ) + + yield* DatabaseMigration.applyOnly(db, [timeSuspendedMigration]) + + expect(yield* db.get(sql`SELECT time_suspended FROM session WHERE id = 'ses_shutdown'`)).toEqual({ + time_suspended: null, + }) + }), + ) + }) + test("renames instruction state without losing rows or durable updates", async () => { await run( Effect.gen(function* () { diff --git a/packages/core/test/session-execution-local.test.ts b/packages/core/test/session-execution-local.test.ts deleted file mode 100644 index be1c4aebea..0000000000 --- a/packages/core/test/session-execution-local.test.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { describe, expect, test } from "bun:test" -import { LLMError, TransportReason } from "@opencode-ai/llm" -import { terminal } from "@opencode-ai/core/session/execution/local" -import { UserInterruptedError } from "@opencode-ai/core/session/error" -import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" -import { Effect, Exit } from "effect" - -describe("SessionExecutionLocal lifecycle", () => { - test("classifies success and typed failure terminals", () => { - expect(terminal(Exit.succeed(undefined))).toEqual({ type: "succeeded" }) - expect( - terminal( - Exit.fail( - new LLMError({ - module: "test", - method: "stream", - reason: new TransportReason({ message: "Disconnected" }), - }), - ), - ), - ).toEqual({ type: "failed", error: { type: "provider.transport", message: "Disconnected" } }) - const storage = new ToolOutputStore.StorageError({ operation: "encode", cause: new Error("invalid output") }) - expect(terminal(Exit.fail(storage))).toEqual({ - type: "failed", - error: { type: "unknown", message: storage.message }, - }) - }) - - test("defaults owner-scope interruption to shutdown and preserves explicit reasons", () => { - const interrupted = Effect.runSyncExit(Effect.interrupt) - expect(terminal(interrupted)).toEqual({ type: "interrupted", reason: "shutdown" }) - expect(terminal(interrupted, "user")).toEqual({ type: "interrupted", reason: "user" }) - expect(terminal(interrupted, "superseded")).toEqual({ type: "interrupted", reason: "superseded" }) - expect(terminal(Exit.fail(new UserInterruptedError()))).toEqual({ type: "interrupted", reason: "user" }) - }) -}) diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts new file mode 100644 index 0000000000..2775df3564 --- /dev/null +++ b/packages/core/test/session-execution.test.ts @@ -0,0 +1,198 @@ +import { describe, expect, test } from "bun:test" +import { LLMError, TransportReason } from "@opencode-ai/llm" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2 } from "@opencode-ai/core/event" +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" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionRestart } from "@opencode-ai/core/session/execution/restart" +import { UserInterruptedError } from "@opencode-ai/core/session/error" +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 { ToolOutputStore } from "@opencode-ai/core/tool-output-store" +import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect" +import { testEffect } from "./lib/effect" + +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionStore.node]))) + +describe("SessionExecution lifecycle", () => { + test("classifies success and typed failure terminals", () => { + expect(SessionExecution.terminal(Exit.succeed(undefined))).toEqual({ type: "succeeded" }) + expect( + SessionExecution.terminal( + Exit.fail( + new LLMError({ + module: "test", + method: "stream", + reason: new TransportReason({ message: "Disconnected" }), + }), + ), + ), + ).toEqual({ type: "failed", error: { type: "provider.transport", message: "Disconnected" } }) + const storage = new ToolOutputStore.StorageError({ operation: "encode", cause: new Error("invalid output") }) + expect(SessionExecution.terminal(Exit.fail(storage))).toEqual({ + type: "failed", + error: { type: "unknown", message: storage.message }, + }) + }) + + test("defaults owner-scope interruption to shutdown and preserves explicit reasons", () => { + const interrupted = Effect.runSyncExit(Effect.interrupt) + expect(SessionExecution.terminal(interrupted)).toEqual({ type: "interrupted", reason: "shutdown" }) + expect(SessionExecution.terminal(interrupted, "user")).toEqual({ type: "interrupted", reason: "user" }) + expect(SessionExecution.terminal(interrupted, "superseded")).toEqual({ type: "interrupted", reason: "superseded" }) + expect(SessionExecution.terminal(Exit.fail(new UserInterruptedError()))).toEqual({ + type: "interrupted", + reason: "user", + }) + }) + + it.effect("atomically consumes each suspension at most once", () => + Effect.gen(function* () { + const database = yield* Database.Service + const store = yield* SessionStore.Service + const first = SessionV2.ID.make("ses_recover_first") + const second = SessionV2.ID.make("ses_recover_second") + yield* seedSessions(database, [first, second], { time_suspended: Date.now() }) + + 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 }) + }), + ) + + it.effect("suspension survives teardown interruption and clears when a drain finishes on its own", () => + Effect.gen(function* () { + const database = yield* Database.Service + const interrupted = SessionV2.ID.make("ses_suspend_interrupted") + const completed = SessionV2.ID.make("ses_suspend_completed") + yield* seedSessions(database, [interrupted, completed]) + + const draining = 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)), + ) + 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* restart.suspendActiveSessions + expect(yield* suspensions(database)).toEqual({ [interrupted]: true, [completed]: true }) + + // A drain that finishes on its own after suspension clears its stale suspension. + yield* Deferred.succeed(release, undefined) + yield* Fiber.join(completing) + yield* execution.awaitIdle(completed) + expect((yield* suspensions(database))[completed]).toBe(false) + + // Teardown interruption preserves suspension for the next server start. + yield* Scope.close(scope, Exit.void) + expect((yield* suspensions(database))[interrupted]).toBe(true) + }), + ) + + it.effect("resumes each suspended Session at most once", () => + Effect.gen(function* () { + const database = yield* Database.Service + const first = SessionV2.ID.make("ses_resume_first") + const second = SessionV2.ID.make("ses_resume_second") + yield* seedSessions(database, [first, second], { time_suspended: Date.now() }) + + const drained: string[] = [] + const scope = yield* Scope.make() + const context = yield* buildExecution(scope, ({ sessionID }) => Effect.sync(() => void drained.push(sessionID))) + const restart = Context.get(context, SessionRestart.Service) + + yield* restart.resumeSuspendedSessions + expect(drained.toSorted()).toEqual([first, second]) + expect(yield* suspensions(database)).toEqual({ [first]: false, [second]: false }) + + yield* restart.resumeSuspendedSessions + expect(drained.length).toBe(2) + yield* Scope.close(scope, Exit.void) + }), + ) +}) + +function seedSessions( + database: Database.Service["Service"], + sessionIDs: ReadonlyArray, + values: { time_suspended?: number } = {}, +) { + return Effect.gen(function* () { + yield* database.db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* database.db + .insert(SessionTable) + .values( + sessionIDs.map((id) => ({ + id, + project_id: Project.ID.global, + slug: id, + directory: "/project", + title: id, + version: "test", + ...values, + })), + ) + .run() + .pipe(Effect.orDie) + }) +} + +function suspensions(database: Database.Service["Service"]) { + return database.db + .select({ id: SessionTable.id, suspended: SessionTable.time_suspended }) + .from(SessionTable) + .all() + .pipe( + Effect.orDie, + Effect.map((rows) => Object.fromEntries(rows.map((row) => [row.id, row.suspended !== null]))), + ) +} + +/** Builds the local execution layer plus the restart actions against the test harness services. */ +function buildExecution(scope: Scope.Closeable, drain: SessionRunner.Interface["drain"]) { + return Effect.gen(function* () { + const database = yield* Database.Service + const events = yield* EventV2.Service + const store = yield* SessionStore.Service + const runner = Layer.succeed(SessionRunner.Service, SessionRunner.Service.of({ drain })) + const locations = Layer.effect( + LocationServiceMap.Service, + LayerMap.make( + () => + // The local execution test only needs the Session runner from the Location graph. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + runner as unknown as Layer.Layer, + ), + ) + return yield* Layer.buildWithScope( + SessionRestart.layer.pipe( + Layer.provideMerge(SessionExecution.layer), + Layer.provide(Layer.succeed(Database.Service, database)), + Layer.provide(Layer.succeed(EventV2.Service, events)), + Layer.provide(Layer.succeed(SessionStore.Service, store)), + Layer.provide(locations), + ), + scope, + ) + }) +} diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index 7cc3583dff..7ddfa9b94b 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -632,6 +632,43 @@ describe("SessionProjector", () => { }), ) + it.effect("does not infer restart continuation from lifecycle history", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "test", + directory: "/project", + title: "test", + version: "test", + }) + .run() + .pipe(Effect.orDie) + const events = yield* EventV2.Service + const suspended = () => + db + .select({ timeSuspended: SessionTable.time_suspended }) + .from(SessionTable) + .where(eq(SessionTable.id, sessionID)) + .get() + .pipe(Effect.orDie) + + yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: "shutdown" }) + expect((yield* suspended())?.timeSuspended).toBeNull() + + yield* events.publish(SessionEvent.Execution.Started, { sessionID }) + expect((yield* suspended())?.timeSuspended).toBeNull() + }), + ) + it.effect("updates only the newest incomplete assistant projection", () => Effect.gen(function* () { const { db } = yield* Database.Service diff --git a/packages/opencode/src/server/routes/instance/httpapi/server.ts b/packages/opencode/src/server/routes/instance/httpapi/server.ts index 471d9aa376..fad59961b1 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/server.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/server.ts @@ -63,8 +63,6 @@ import { PtyTicket } from "@opencode-ai/core/pty/ticket" import { Ripgrep } from "@opencode-ai/core/ripgrep" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionV2 } from "@opencode-ai/core/session" -import { SessionExecution } from "@opencode-ai/core/session/execution" -import * as SessionExecutionLocal from "@opencode-ai/core/session/execution/local" import { PluginRuntime } from "@opencode-ai/core/plugin/runtime" import { lazy } from "@/util/lazy" import { CorsConfig, isAllowedCorsOrigin, type CorsOptions } from "@opencode-ai/server/cors" @@ -302,7 +300,6 @@ export function createRoutes( Layer.provide( AppNodeBuilderV1.build(LayerNode.group([SessionV2.node, PluginRuntime.providerNode]), [ [LocationServiceMap.node, locationServiceMapV2], - [SessionExecution.node, SessionExecutionLocal.node], ]), ), Layer.provide(locationServiceMapV2), diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index 28e9a49117..d9bef69e10 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -6,6 +6,7 @@ import { Context, Effect, Layer, Option } from "effect" import { HttpClient, HttpClientRequest, HttpMiddleware, HttpRouter, HttpServer } from "effect/unstable/http" import { HttpApi, HttpApiClient } from "effect/unstable/httpapi" import { createServer } from "node:http" +import { SessionRestart } from "@opencode-ai/core/session/execution/restart" import { ServerAuth } from "./auth" import { createRoutes } from "./routes" import { ServerInfo } from "./server-info" @@ -14,6 +15,7 @@ export type Options = { readonly hostname: string readonly port: Option.Option readonly password: string + readonly restartContinuity?: boolean } const ReadinessApi = HttpApi.make("readiness").add(HealthGroup) @@ -38,30 +40,49 @@ export const start = Effect.fn("ServerProcess.start")(function* (options: Option }) function listen(options: Options) { - if (Option.isSome(options.port)) return bind(options.hostname, options.port.value, options.password) + if (Option.isSome(options.port)) return bind(options, options.port.value) const next = (port: number): ReturnType => - bind(options.hostname, port, options.password).pipe( - Effect.catch((error) => (port === 65_535 ? Effect.fail(error) : next(port + 1))), - ) + bind(options, port).pipe(Effect.catch((error) => (port === 65_535 ? Effect.fail(error) : next(port + 1)))) return next(4096) } -function bind(hostname: string, port: number, password: string) { +function bind(options: Options, port: number) { const server = createServer() return Layer.build( - createRoutes(password, () => { + createRoutes(options.password, () => { const address = server.address() if (address === null || typeof address === "string") return [] const host = address.family === "IPv6" ? `[${address.address}]` : address.address - return ServerInfo.connectionURLs(`http://${host}:${address.port}`, hostname) + return ServerInfo.connectionURLs(`http://${host}:${address.port}`, options.hostname) }).pipe( - Layer.flatMap((context) => - HttpServer.serve(Context.get(context, HttpRouter.HttpRouter).asHttpEffect(), HttpMiddleware.logger), - ), - Layer.provideMerge(NodeHttpServer.layer(() => server, { port, host: hostname })), + Layer.flatMap((context) => { + const serve = HttpServer.serve( + Context.get(context, HttpRouter.HttpRouter).asHttpEffect(), + HttpMiddleware.logger, + ) + if (!options.restartContinuity) return serve + const restart = Context.get(context, SessionRestart.Service) + return Layer.merge(serve, restartContinuity(restart)) + }), + Layer.provideMerge(NodeHttpServer.layer(() => server, { port, host: options.hostname })), ), ).pipe( Effect.tap(() => Effect.addFinalizer(() => Effect.sync(() => server.closeAllConnections()))), Effect.map((context) => Context.get(context, HttpServer.HttpServer).address), ) } + +/** + * 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. + */ +function restartContinuity(restart: SessionRestart.Interface) { + return Layer.effectDiscard( + Effect.gen(function* () { + yield* Effect.forkScoped(restart.resumeSuspendedSessions) + // Registered after the fork so suspension observes still-running resumed drains during teardown. + yield* Effect.addFinalizer(() => restart.suspendActiveSessions) + }), + ) +} diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index 0a398105ff..3f44ec3bf1 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -11,10 +11,9 @@ import { PtyTicket } from "@opencode-ai/core/pty/ticket" import { MoveSession } from "@opencode-ai/core/control-plane/move-session" import { Project } from "@opencode-ai/core/project" import { SessionV2 } from "@opencode-ai/core/session" -import { SessionExecution } from "@opencode-ai/core/session/execution" import { Job } from "@opencode-ai/core/job" import { LocationServiceMap } from "@opencode-ai/core/location-service-map" -import { SessionExecutionLocal } from "@opencode-ai/core/session/execution/local" +import { SessionRestart } from "@opencode-ai/core/session/execution/restart" import { PluginRuntime } from "@opencode-ai/core/plugin/runtime" import { SdkPlugins } from "@opencode-ai/core/plugin/sdk" import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" @@ -49,6 +48,7 @@ const applicationServices = LayerNode.group([ Credential.node, PtyEnvironment.node, LocationServiceMap.node, + SessionRestart.node, ]) export function createRoutes(password?: string, serviceURLs: () => ReadonlyArray = () => []) { @@ -70,7 +70,6 @@ function makeRoutes( ) { const pluginRuntimeCell = PluginRuntime.makeCell() const replacements: LayerNode.Replacements = [ - [SessionExecution.node, SessionExecutionLocal.node], [PluginRuntime.node, PluginRuntime.layerWithCell(pluginRuntimeCell)], [PluginRuntime.providerNode, PluginRuntime.providerNodeWithCell(pluginRuntimeCell)], ] diff --git a/specs/v2/session-restart-continuation.md b/specs/v2/session-restart-continuation.md new file mode 100644 index 0000000000..e2a7fb0960 --- /dev/null +++ b/specs/v2/session-restart-continuation.md @@ -0,0 +1,157 @@ +# RFC: Continue Sessions After Managed-Service Restart + +| Field | Value | +| -------------- | ------------------------------------------------------------ | +| Status | Proposed | +| Author | Kit Langton | +| Date | 2026-07-08 | +| Tracking issue | [#35646](https://github.com/anomalyco/opencode/issues/35646) | + +## Summary + +When the managed OpenCode server shuts down gracefully, active Sessions should continue automatically the next time the managed server starts. + +This RFC proposes one private nullable timestamp on the existing Session row: `time_suspended`. The managed server suspends its active Sessions on graceful shutdown and resumes suspended Sessions on startup. Both are explicit actions the managed server invokes; no other server ever suspends or auto-resumes. + +The field is not Session status. Live activity remains process-local. Hard-crash recovery and exactly-once provider or tool execution remain out of scope. + +## Decision + +Add this private Session field and partial index: + +```sql +ALTER TABLE session +ADD COLUMN time_suspended INTEGER; + +CREATE INDEX session_time_suspended_idx +ON session(time_suspended) +WHERE time_suspended IS NOT NULL; +``` + +A non-null `time_suspended` means: + +> A managed server suspended this Session during graceful shutdown, at this time. The next managed server may make one attempt to resume it. + +The name records the fact rather than one consumer's policy, and it follows the Session table's existing nullable-timestamp idiom (`time_compacting`, `time_archived`). The timestamp also gives operators suspension age for free, which later policy may use without a schema change. + +The field does not appear in public `Session.Info` and does not drive UI activity. + +## Status and Suspension Are Separate + +| Concept | Values | Source of truth | +| ----------------- | ------------------- | ------------------------------------- | +| Live activity | `inactive / active` | Process-local `SessionRunCoordinator` | +| Execution history | `started / settled` | Durable lifecycle events | +| Suspension | `null / timestamp` | Private Session-row `time_suspended` | + +A persisted status such as `idle / running / resumable` answers three different questions. `running` becomes stale after a crash, while `resumable` is pending work rather than current status. + +## The Managed Server Owns Restart Continuity + +Restart continuity is not layer configuration. `SessionRestart` is an inert core service exposing two actions, and only the managed server (`opencode serve --service`) calls them: + +```typescript +// ServerProcess, service mode only +yield * Effect.forkScoped(restart.resumeSuspendedSessions) +yield * Effect.addFinalizer(() => restart.suspendActiveSessions) +``` + +Default, embedded, and stdio servers build the same execution layer but never invoke the actions, so they never suspend or auto-resume. + +### Graceful shutdown suspends + +Teardown ordering makes suspension observe exactly the work a restart interrupts: + +1. The HTTP server closes all connections; no new work can arrive. +2. `suspendActiveSessions` snapshots `SessionExecution.active` and sets `time_suspended` for each. +3. Session execution teardown interrupts the still-running drains. + +A SIGKILL runs none of this: nothing is suspended, and the user resumes manually. That is deliberate — automatic post-crash continuation would retry ambiguous provider and tool work. + +### Ordinary lifecycle clears stale suspension + +The Session execution layer clears the field through EventV2 live `commit` hooks, with no knowledge of server mode: + +| Lifecycle event | `time_suspended` | +| --------------------------- | ------------------------------------- | +| Execution started | `NULL` | +| Execution succeeded | `NULL` | +| Execution failed | `NULL` | +| Execution interrupted (any) | unchanged — interruption preserves it | + +Interruption must preserve suspension because managed teardown interrupts drains immediately after suspending them. Every other transition clearing the field closes the races: a drain that finishes on its own between suspension and teardown clears its suspension, and an embedded server that completes a suspended Session during the gap clears it on start. + +Because the clears are `commit` hooks rather than projections, event replay preserves lifecycle history without recreating or destroying suspension. + +## Startup Consumes Each Suspension Atomically + +`resumeSuspendedSessions` reads pending Session IDs through the partial index. Immediately before resuming each Session, it performs a conditional clear: + +```sql +UPDATE session +SET time_suspended = NULL +WHERE id = ? AND time_suspended IS NOT NULL +RETURNING id; +``` + +Only the process receiving the returned row resumes that Session. A second consumer receives no row. Each suspension is consumed right before its own drain starts, and at most a handful of resumed drains run at once. + +The resume goes through the existing process-local coordinator, which joins duplicate same-process resumes and starts a forced drain while idle. + +## Failure Semantics + +The design provides at-most-once automatic scheduling, not guaranteed continuation. + +| Failure | Result | +| ------------------------------------------------------ | ------------------------------------------------------------------------------------- | +| Old server is killed before graceful closeout | Nothing is suspended; user resumes manually | +| Old server dies between suspension and teardown | Session stays suspended; next server resumes it | +| New server crashes before conditional clear | Session stays suspended | +| New server crashes after clear but before drain starts | Automatic continuation is lost | +| New server crashes after drain starts | No automatic hard-crash retry | +| Interrupted tool has uncertain side effects | Existing stale-tool reconciliation records failure rather than replaying the old call | + +Losing one automatic continuation is safer than repeatedly restarting ambiguous provider or tool work. + +## Migration Does Not Infer Historical Intent + +The migration adds the nullable column with no backfill. It does not scan historical shutdown events. + +An old shutdown event records what happened; it does not prove that a future process is authorized to start new work. The first upgrade may therefore require manual continuation for Sessions interrupted by the old binary. + +## Ranked Alternatives + +| Rank | Option | Verdict | Reason | +| ---: | --------------------------------------- | --------- | ------------------------------------------------------------------------------- | +| 1 | Daemon-invoked suspend/resume actions | Preferred | The restart authority acts explicitly; execution layer stays generic | +| 2 | Execution-layer configuration flag | Rejected | Threads a mode bit through server, routes, and layer construction | +| 3 | Dedicated continuation table | Reserve | Useful if continuation later needs metadata, leases, retries, or multiple rows | +| 4 | Leased continuation queue | Defer | Solves claimant failure but adds acknowledgement, expiry, and fencing semantics | +| 5 | Persisted general Session status | Reject | Conflates activity, history, and pending work | +| 6 | Scan lifecycle history on every startup | Reject | Repeats unbounded historical work and lacks direct atomic consumption | + +## Verification + +Required regression coverage: + +- A suspension can be consumed only once per Session. +- Generic lifecycle publication and replay do not infer suspension. +- Historical shutdown events remain unsuspended after migration. +- Concurrent managed-service candidates elect one process and produce one continued execution. +- Teardown interruption preserves suspension; a drain finishing on its own clears it. +- Only the managed server suspends and resumes; default, embedded, and stdio servers never invoke the actions. + +## Non-Goals + +- Recovering unmatched execution after a hard process or machine crash. +- Persisting authoritative live Session status. +- Coordinating Session execution across independent processes or a cluster. +- Guaranteeing exactly-once provider requests or tool side effects. +- Retrying a continuation after its suspension has been consumed. + +## Prior Work + +- [Issue #35646: auto-resume active Sessions after server restart](https://github.com/anomalyco/opencode/issues/35646) +- [Draft PR #35778: resume Sessions after restart](https://github.com/anomalyco/opencode/pull/35778) +- [Draft PR #35820: resume Sessions after restart](https://github.com/anomalyco/opencode/pull/35820) +- [Issue #35642: interrupted work remains spinning after machine restart](https://github.com/anomalyco/opencode/issues/35642)