diff --git a/packages/opencode/src/project/bootstrap.ts b/packages/opencode/src/project/bootstrap.ts index acf0caed3cdf..13bebe259f64 100644 --- a/packages/opencode/src/project/bootstrap.ts +++ b/packages/opencode/src/project/bootstrap.ts @@ -9,6 +9,7 @@ import { InstanceState } from "@/effect/instance-state" import { ShareNext } from "@/share/share-next" import { Effect, Layer } from "effect" import { Config } from "@/config/config" +import { SessionRecovery } from "@/session/recover" import { Service } from "./bootstrap-service" export { Service } from "./bootstrap-service" @@ -28,6 +29,7 @@ const layer = Layer.effect( const shareNext = yield* ShareNext.Service const snapshot = yield* Snapshot.Service const vcs = yield* Vcs.Service + const recovery = yield* SessionRecovery.Service const run = Effect.gen(function* () { const ctx = yield* InstanceState.context @@ -39,7 +41,7 @@ const layer = Layer.effect( // Each service self-manages its own slow work via Effect.forkScoped against // its per-instance state scope. We just await materialization here. yield* Effect.forEach( - [lsp, shareNext, format, vcs, snapshot, project], + [lsp, shareNext, format, vcs, snapshot, project, recovery], (s) => s.init().pipe(Effect.catchCause((cause) => Effect.logWarning("init failed", { cause }))), { concurrency: "unbounded", discard: true }, ).pipe(Effect.withSpan("InstanceBootstrap.init")) @@ -52,7 +54,17 @@ const layer = Layer.effect( export const node = makeGlobalNode({ service: Service, layer: layer, - deps: [Config.node, Format.node, LSP.node, Plugin.node, Project.node, ShareNext.node, Snapshot.node, Vcs.node], + deps: [ + Config.node, + Format.node, + LSP.node, + Plugin.node, + Project.node, + SessionRecovery.node, + ShareNext.node, + Snapshot.node, + Vcs.node, + ], }) export * as InstanceBootstrap from "./bootstrap" diff --git a/packages/opencode/src/session/recover.ts b/packages/opencode/src/session/recover.ts new file mode 100644 index 000000000000..f1bd1286fdf7 --- /dev/null +++ b/packages/opencode/src/session/recover.ts @@ -0,0 +1,94 @@ +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Database } from "@opencode-ai/core/database/database" +import { MessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { NamedError } from "@opencode-ai/core/util/error" +import { and, eq, sql } from "drizzle-orm" +import { Context, Effect, Layer } from "effect" +import { InstanceState } from "@/effect/instance-state" +import { MessageV2 } from "./message-v2" +import type { MessageID, SessionID } from "./schema" +import { Session } from "./session" + +export interface Interface { + readonly init: () => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/SessionRecovery") {} + +// Only one opencode process serves a sandbox and a turn's run loop lives in that process, so at +// instance boot every assistant message without a completion time is provably dead. Finish it the +// way an abort would. No session.error is published here: a client that keys "no reply yet" off +// the first assistant message can race it, and the coordinator already derives the terminal event +// from a completed errored message when it replays the session after a restart. +const layer = Layer.effect( + Service, + Effect.gen(function* () { + const database = yield* Database.Service + const sessions = yield* Session.Service + + const state = yield* InstanceState.make( + Effect.fn("SessionRecovery.state")(function* (ctx) { + const rows = yield* database.db + .select({ id: MessageTable.id, sessionID: MessageTable.session_id }) + .from(MessageTable) + .innerJoin(SessionTable, eq(SessionTable.id, MessageTable.session_id)) + .where( + and( + eq(SessionTable.project_id, ctx.project.id), + sql`json_extract(${MessageTable.data}, '$.role') = 'assistant'`, + sql`json_extract(${MessageTable.data}, '$.time.completed') IS NULL`, + ), + ) + .all() + .pipe(Effect.orDie) + + for (const row of rows) { + // One unreadable row must not strand every other orphan (REPL-29772 saw session-DB corruption). + yield* recover(row).pipe( + Effect.catchCause((cause) => Effect.logWarning("recovery failed", { messageID: row.id, cause })), + ) + } + }), + ) + + const recover = Effect.fn("SessionRecovery.recover")(function* (row: { id: MessageID; sessionID: SessionID }) { + const { info, parts } = yield* MessageV2.get({ sessionID: row.sessionID, messageID: row.id }).pipe( + Effect.provideService(Database.Service, database), + ) + if (info.role !== "assistant") return + const end = Date.now() + for (const part of parts) { + if (part.type !== "tool" || part.state.status === "completed" || part.state.status === "error") continue + const metadata = "metadata" in part.state && part.state.metadata ? part.state.metadata : {} + yield* sessions.updatePart({ + ...part, + state: { + ...part.state, + status: "error", + error: "Tool execution interrupted by a restart", + metadata: { ...metadata, interrupted: true }, + time: { start: "time" in part.state ? part.state.time.start : end, end }, + }, + }) + } + info.error = new NamedError.Unknown({ + message: "The agent restarted before it could finish this turn. Send your message again.", + }).toObject() + info.time.completed = end + yield* sessions.updateMessage(info) + yield* Effect.logInfo("recovered interrupted turn", { sessionID: info.sessionID, messageID: info.id }) + }) + + return Service.of({ + init: () => InstanceState.get(state).pipe(Effect.asVoid), + }) + }), +) + +export const node = LayerNode.make({ + service: Service, + layer, + deps: [Database.node, Session.node, MessageV2.node], +}) + +export * as SessionRecovery from "./recover" diff --git a/packages/opencode/test/session/recover.test.ts b/packages/opencode/test/session/recover.test.ts new file mode 100644 index 000000000000..b091eebf7858 --- /dev/null +++ b/packages/opencode/test/session/recover.test.ts @@ -0,0 +1,118 @@ +import { expect } from "bun:test" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { ModelV2 } from "@opencode-ai/core/model" +import { Effect } from "effect" +import { EventV2Bridge } from "@/event-v2-bridge" +import { Session } from "@/session/session" +import { SessionStatus } from "@/session/status" +import { SessionRecovery } from "@/session/recover" +import { MessageV2 } from "../../src/session/message-v2" +import { MessageID, PartID, type SessionID } from "../../src/session/schema" +import { testEffect } from "../lib/effect" + +const it = testEffect( + LayerNode.compile( + LayerNode.group([ + Session.node, + MessageV2.node, + SessionProjector.node, + SessionStatus.node, + EventV2Bridge.node, + SessionRecovery.node, + ]), + ), +) + +const seedTurn = Effect.fn("Test.seedTurn")(function* (sessionID: SessionID, opts?: { completed?: number }) { + const sessions = yield* Session.Service + const userID = MessageID.ascending() + yield* sessions.updateMessage({ + id: userID, + sessionID, + role: "user", + time: { created: Date.now() }, + agent: "test", + model: { providerID: "test", modelID: "test" }, + tools: {}, + mode: "", + } as unknown as SessionV1.Info) + const assistantID = MessageID.ascending() + yield* sessions.updateMessage({ + id: assistantID, + sessionID, + role: "assistant", + time: { created: Date.now(), completed: opts?.completed }, + parentID: userID, + modelID: ModelV2.ID.make("test"), + providerID: ProviderV2.ID.make("test"), + mode: "", + agent: "default", + path: { cwd: "/", root: "/" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + } as unknown as SessionV1.Info) + yield* sessions.updatePart({ + id: PartID.ascending(), + sessionID, + messageID: assistantID, + type: "tool", + callID: "call-1", + tool: "bash", + state: { status: "running", input: { command: "sleep 120" }, time: { start: 1 } }, + } as unknown as SessionV1.Part) + return assistantID +}) + +it.instance("finishes a restart-orphaned assistant message with an interrupted error", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const events = yield* EventV2Bridge.Service + const recovery = yield* SessionRecovery.Service + const chat = yield* sessions.create({}) + const assistantID = yield* seedTurn(chat.id) + const seen: string[] = [] + const off = yield* events.listen((event) => { + seen.push(event.type) + return Effect.void + }) + + yield* recovery.init() + yield* off + + const stored = yield* MessageV2.get({ sessionID: chat.id, messageID: assistantID }) + expect(stored.info.role).toBe("assistant") + if (stored.info.role === "assistant") { + expect(stored.info.error?.name).toBe("UnknownError") + expect(stored.info.time.completed).toBeNumber() + expect(stored.info.finish).toBeUndefined() + } + const tool = stored.parts.find((part) => part.type === "tool") + expect(tool?.type === "tool" && tool.state.status === "error" ? tool.state.metadata?.interrupted : undefined).toBe( + true, + ) + expect(seen).toContain(MessageV2.Event.PartUpdated.type) + expect(seen).toContain(MessageV2.Event.Updated.type) + // The coordinator derives the terminal event from the repaired row; publishing one here would race its marker. + expect(seen).not.toContain(Session.Event.Error.type) + expect(seen).not.toContain(SessionStatus.Event.Idle.type) + }), +) + +it.instance("leaves a completed assistant message alone", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const recovery = yield* SessionRecovery.Service + const chat = yield* sessions.create({}) + const assistantID = yield* seedTurn(chat.id, { completed: Date.now() }) + + yield* recovery.init() + + const stored = yield* MessageV2.get({ sessionID: chat.id, messageID: assistantID }) + expect(stored.info.role === "assistant" ? stored.info.error : "wrong role").toBeUndefined() + const tool = stored.parts.find((part) => part.type === "tool") + expect(tool?.type === "tool" ? tool.state.status : "missing").toBe("running") + }), +)