fix(opencode): finish restart-orphaned turns at instance boot

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"

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<void>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@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"

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")
+  }),
+)