ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(core): resume suspended sessions after service restart (#36105)

Browse source
This commit is contained in:
Kit Langton 2026-07-09 13:19:15 -04:00 • committed by GitHub
commit 6524dfc818
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 820 additions and 299 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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<Service, Interface>()("@opencode/v2/SessionExecution") {}
export const node = LayerNode.unbound(Service, Node.tags.values.global)
type InterruptReason = "user" | "shutdown" | "superseded"
export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, 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 = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
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<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({
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(

96
packages/core/src/session/execution/local.ts
View file

@ -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<void, SessionRunner.RunError>, 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 = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
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"

51
packages/core/src/session/execution/restart.ts Normal file
View file

@ -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<void>
/** Resumes suspended Sessions. Each suspension is consumed atomically, so a Session resumes at most once. */
readonly resumeSuspendedSessions: Effect.Effect<void>
}
/**
* 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<Service, Interface>()("@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] })

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

@ -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`),
],
)

39
packages/core/src/session/store.ts
View file

@ -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<ReadonlyArray<Session.ID>>
/** Clears suspension, reporting whether this caller consumed it. At most one concurrent caller receives true. */
readonly consumeSuspended: (sessionID: Session.ID) => Effect.Effect<boolean>
readonly suspend: (sessionIDs: Iterable<Session.ID>) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@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)
}),
})
}),
)