diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index 08d8bf4011..54e3a4e080 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -2,9 +2,14 @@ export * as SessionRestart from "./restart" import { Context, Effect, Layer } from "effect" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" +import { Bus } from "../../bus" +import { SessionEvent } from "../event" import { SessionExecution } from "../execution" 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." + export interface Interface { /** * Marks every execution active in this process for resumption by the next server start. @@ -26,6 +31,7 @@ export const layer = Layer.effect( 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) @@ -37,6 +43,11 @@ export const layer = Layer.effect( (sessionID) => Effect.gen(function* () { if (!(yield* store.consumeSuspended(sessionID))) return + yield* bus.publish(SessionEvent.Synthetic, { + sessionID, + text: CONTINUE_AFTER_SERVER_RESTART, + description: "Server restarted", + }) // Drain failures are already logged and durably recorded by the execution layer. yield* Effect.ignore(execution.resume(sessionID)) }), @@ -47,4 +58,8 @@ export const layer = Layer.effect( }), ) -export const node = makeGlobalNode({ service: Service, layer, deps: [SessionStore.node, SessionExecution.node] }) +export const node = makeGlobalNode({ + service: Service, + layer, + deps: [SessionStore.node, SessionExecution.node, Bus.node], +}) diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index b51881cf38..e8adaab6cc 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -13,6 +13,7 @@ import { Session } 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 { SessionEvent } from "@opencode-ai/core/session/event" import { SessionRunner } from "@opencode-ai/core/session/runner" import { SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" @@ -127,23 +128,34 @@ describe("SessionExecution lifecycle", () => { it.effect("resumes each suspended Session at most once", () => Effect.gen(function* () { const database = yield* Database.Service + const bus = yield* Bus.Service const first = Session.ID.make("ses_resume_first") const second = Session.ID.make("ses_resume_second") yield* seedSessions(database, [first, second], { time_suspended: Date.now() }) const drained: string[] = [] + const continued: SessionEvent.Synthetic[] = [] const scope = yield* Scope.make() const context = yield* buildExecution(scope, ({ sessionID }) => Effect.sync(() => void drained.push(sessionID))) 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))) yield* restart.resumeSuspendedSessions 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( + [first, second].map((sessionID) => ({ + sessionID, + text: "The server restarted while you were working. Continue from where you left off without repeating completed work.", + description: "Server restarted", + })), + ) expect(yield* suspensions(database)).toEqual({ [first]: false, [second]: false }) yield* restart.resumeSuspendedSessions expect(drained.length).toBe(2) + expect(continued.length).toBe(2) yield* Scope.close(scope, Exit.void) }), )