fix(core): settle abandoned compactions before resuming sessions (#47178)

packages/core/src/session/runner/llm.ts

@@ -1,6 +1,7 @@
 export * as SessionRunnerLLM from "./llm.js"
 
 import { Message } from "@opencode-ai/ai"
+import { and, desc, eq, sql } from "drizzle-orm"
 import { Cause, Effect, Exit, FiberMap, Layer } from "effect"
 import { Database } from "../../database/database.js"
 import { Bus } from "../../bus.js"
@@ -15,6 +16,7 @@ import { SessionModelTransport } from "../model-transport.js"
 import { SessionMessage } from "../message.js"
 import { SessionSchema } from "../schema.js"
 import { SessionStore } from "../store.js"
+import { SessionMessageTable } from "../sql.js"
 import { SessionTitle } from "../title.js"
 import { DrainResult, Service, type Interface } from "./index.js"
 import { Snapshot } from "../../snapshot.js"
@@ -59,6 +61,7 @@ const layer = Layer.effect(
         if (promotable === "steer" && pending.delivery === "queue" && !control) return DrainResult.Complete()
       }
       yield* plugins.awaitActivation
+      yield* settleStaleCompactions(sessionID)
       yield* settleStaleToolCalls(sessionID)
 
       const advanceToStep = Effect.fn("SessionRunner.advanceToStep")(() =>
@@ -276,6 +279,36 @@ const layer = Layer.effect(
       }
     })
 
+    const settleStaleCompactions = Effect.fn("SessionRunner.settleStaleCompactions")(function* (
+      sessionID: SessionSchema.ID,
+    ) {
+      // A process death skips compaction finalizers. Include orphans behind a
+      // completed checkpoint, and settle newest first to match event projection.
+      const rows = yield* db
+        .select()
+        .from(SessionMessageTable)
+        .where(
+          and(
+            eq(SessionMessageTable.session_id, sessionID),
+            eq(SessionMessageTable.type, "compaction"),
+            sql`json_extract(${SessionMessageTable.data}, '$.status') = 'running'`,
+          ),
+        )
+        .orderBy(desc(SessionMessageTable.seq))
+        .all()
+        .pipe(Effect.orDie)
+      for (const row of rows) {
+        const message = yield* SessionHistory.decodeMessageRow(row)
+        if (message.type !== "compaction") continue
+        yield* bus.publish(SessionEvent.Compaction.Failed, {
+          sessionID,
+          reason: message.reason,
+          inputID: message.id,
+          error: { type: "compaction.interrupted", message: "Compaction was interrupted" },
+        })
+      }
+    })
+
     const settleStaleToolCalls = Effect.fn("SessionRunner.settleStaleToolCalls")(function* (
       sessionID: SessionSchema.ID,
     ) {

packages/core/test/session-runner.test.ts

@@ -3472,6 +3472,83 @@ describe("SessionRunnerLLM", () => {
     expect(userTexts(s.requests[1])).toEqual(["Start working", "Recover with this"])
   })
 
+  scenario("settles abandoned compactions before continuing after a process crash", function* (s) {
+    yield* s.runPrompt("History before the crash")
+    const first = SessionMessage.ID.create()
+    const completed = SessionMessage.ID.create()
+    const last = SessionMessage.ID.create()
+    yield* s.bus.publish(SessionEvent.Compaction.Started, {
+      sessionID,
+      reason: "manual",
+      inputID: first,
+      recent: "",
+    })
+    yield* s.bus.publish(SessionEvent.Compaction.Started, {
+      sessionID,
+      reason: "manual",
+      inputID: completed,
+      recent: "",
+    })
+    yield* s.bus.publish(SessionEvent.Compaction.Ended, {
+      sessionID,
+      reason: "manual",
+      text: "## Objective\n- Earlier completed checkpoint",
+      recent: "",
+    })
+    yield* s.bus.publish(SessionEvent.Compaction.Started, {
+      sessionID,
+      reason: "auto",
+      inputID: last,
+      recent: "",
+    })
+
+    // These starts have no terminal events, as after SIGKILL. The older orphan
+    // is outside model-visible history; recovery must settle it as well.
+    expect(
+      (yield* s.messages).filter((message) => message.type === "compaction" && message.status === "running"),
+    ).toHaveLength(2)
+    yield* s.llm.push(TestLLM.text("Recovered response", "recovered"))
+    const run = yield* s.resumePaused
+    expect((yield* s.messages).filter((message) => message.type === "compaction").toReversed()).toMatchObject([
+      { id: first, status: "failed", reason: "manual", error: { type: "compaction.interrupted" } },
+      { id: completed, status: "completed", summary: "## Objective\n- Earlier completed checkpoint" },
+      { id: last, status: "failed", reason: "auto", error: { type: "compaction.interrupted" } },
+    ])
+    yield* run.finish
+
+    yield* s.llm.push(TestLLM.text("## Objective\n- New checkpoint", "new-summary"))
+    const next = yield* s.session.compact({ sessionID })
+    yield* s.session.wait(sessionID)
+    expect((yield* s.messages).find((message) => message.id === next.id)).toMatchObject({
+      status: "completed",
+      summary: "## Objective\n- New checkpoint",
+    })
+    expect(
+      (yield* s.messages).filter((message) => message.type === "compaction" && message.status === "running"),
+    ).toHaveLength(0)
+  })
+
+  scenario("settles an abandoned compaction before delivering another manual compaction", function* (s) {
+    yield* s.runPrompt("History before the crash")
+    const previous = SessionMessage.ID.create()
+    yield* s.bus.publish(SessionEvent.Compaction.Started, {
+      sessionID,
+      reason: "manual",
+      inputID: previous,
+      recent: "",
+    })
+    yield* s.llm.push(TestLLM.text("## Objective\n- New checkpoint", "new-summary"))
+    const gate = yield* s.llm.gate
+    const next = yield* s.session.compact({ sessionID })
+    yield* gate.started
+    expect((yield* s.messages).filter((message) => message.type === "compaction").toReversed()).toMatchObject([
+      { id: previous, status: "failed", error: { type: "compaction.interrupted" } },
+      { id: next.id, status: "running" },
+    ])
+    yield* gate.release
+    yield* s.session.wait(sessionID)
+  })
+
   scenario("durably fails local tools left running by a prior process before continuing", function* (s) {
     yield* s.admit("Recover interrupted tool")
     yield* SessionInbox.promote(s.db, s.bus, sessionID, "steer")