From c9d240704d6eefc88b63a1eca2cb933b3eb70ed3 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 3 Sep 2026 21:02:29 -0400 Subject: [PATCH] fix(core): settle abandoned compactions before resuming sessions (#47178) --- packages/core/src/session/runner/llm.ts | 33 ++++++++++ packages/core/test/session-runner.test.ts | 77 +++++++++++++++++++++++ 2 files changed, 110 insertions(+) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 0ef2e5ea634..643dc82b419 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/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, ) { diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index f8bc581d617..26e4494bb30 100644 --- a/packages/core/test/session-runner.test.ts +++ b/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")