fix(session): dedupe compaction execution

packages/opencode/src/session/compaction.ts

@@ -22,6 +22,8 @@ import { RuntimeFlags } from "@/effect/runtime-flags"
 import { EventV2 } from "@opencode-ai/core/event"
 import { EventV2Bridge } from "@/event-v2-bridge"
 import { SessionEvent } from "@opencode-ai/core/session-event"
+import { Database } from "@/storage/db"
+import { MessageTable, PartTable } from "./session.sql"
 
 const log = Log.create({ service: "session.compaction" })
 
@@ -122,6 +124,23 @@ function completedCompactions(messages: MessageV2.WithParts[]) {
   })
 }
 
+function activeCompactionMarker(messages: MessageV2.WithParts[]) {
+  const completed = new Set<MessageID>()
+  for (const msg of messages) {
+    if (msg.info.role !== "assistant") continue
+    if (msg.info.agent !== "compaction") continue
+    if (!msg.info.summary || !msg.info.finish) continue
+    completed.add(msg.info.parentID)
+  }
+
+  return messages.find(
+    (msg) =>
+      msg.info.role === "user" &&
+      msg.parts.some((part) => part.type === "compaction") &&
+      !completed.has(msg.info.id),
+  )
+}
+
 function buildPrompt(input: { previousSummary?: string; context: string[] }) {
   const anchor = input.previousSummary
     ? [
@@ -354,31 +373,60 @@ export const layer = Layer.effect(
       if (!parent || parent.info.role !== "user") {
         throw new Error(`Compaction parent must be a user message: ${input.parentID}`)
       }
+      const existing = Array.from(MessageV2.stream(input.sessionID)).find(
+        (msg) =>
+          msg.info.role === "assistant" &&
+          msg.info.agent === "compaction" &&
+          msg.info.summary &&
+          msg.info.parentID === input.parentID &&
+          !msg.info.error,
+      )
+      let inputMessages = input.messages
+      if (existing?.info.role === "assistant") {
+        if (existing.info.finish) return "continue"
+        const interrupted: MessageV2.Assistant = {
+          ...existing.info,
+          finish: "error",
+          error: MessageV2.fromError(new DOMException("Compaction interrupted before completion", "AbortError"), {
+            providerID: existing.info.providerID,
+            aborted: true,
+          }),
+          time: {
+            ...existing.info.time,
+            completed: existing.info.time.completed ?? Date.now(),
+          },
+        }
+        yield* session.updateMessage(interrupted)
+        inputMessages = input.messages.map((msg) =>
+          msg.info.id === interrupted.id ? { ...msg, info: interrupted } : msg,
+        )
+      }
+
       const userMessage = parent.info
       const compactionPart = parent.parts.find((part): part is MessageV2.CompactionPart => part.type === "compaction")
 
-      let messages = input.messages
+      let messages = inputMessages
       let replay:
         | {
             info: MessageV2.User
             parts: MessageV2.Part[]
           }
         | undefined
       if (input.overflow) {
-        const idx = input.messages.findIndex((m) => m.info.id === input.parentID)
+        const idx = inputMessages.findIndex((m) => m.info.id === input.parentID)
         for (let i = idx - 1; i >= 0; i--) {
-          const msg = input.messages[i]
+          const msg = inputMessages[i]
           if (msg.info.role === "user" && !msg.parts.some((p) => p.type === "compaction")) {
             replay = { info: msg.info, parts: msg.parts }
-            messages = input.messages.slice(0, i)
+            messages = inputMessages.slice(0, i)
             break
           }
         }
         const hasContent =
           replay && messages.some((m) => m.info.role === "user" && !m.parts.some((p) => p.type === "compaction"))
         if (!hasContent) {
           replay = undefined
-          messages = input.messages
+          messages = inputMessages
         }
       }
 
@@ -590,22 +638,43 @@ export const layer = Layer.effect(
       auto: boolean
       overflow?: boolean
     }) {
-      const msg = yield* session.updateMessage({
-        id: MessageID.ascending(),
-        role: "user",
-        model: input.model,
-        sessionID: input.sessionID,
-        agent: input.agent,
-        time: { created: Date.now() },
-      })
-      yield* session.updatePart({
-        id: PartID.ascending(),
-        messageID: msg.id,
-        sessionID: msg.sessionID,
-        type: "compaction",
-        auto: input.auto,
-        overflow: input.overflow,
-      })
+      const created = yield* Effect.sync(() =>
+        Database.transaction(
+          (tx) => {
+            const messages = Array.from(MessageV2.stream(input.sessionID)).reverse()
+            if (activeCompactionMarker(messages)) return undefined
+
+            const now = Date.now()
+            const msg: MessageV2.User = {
+              id: MessageID.ascending(),
+              role: "user",
+              model: input.model,
+              sessionID: input.sessionID,
+              agent: input.agent,
+              time: { created: now },
+            }
+            const part: MessageV2.CompactionPart = {
+              id: PartID.ascending(),
+              messageID: msg.id,
+              sessionID: msg.sessionID,
+              type: "compaction",
+              auto: input.auto,
+              overflow: input.overflow,
+            }
+            const { id, sessionID, ...info } = msg
+            const { id: partID, messageID, sessionID: partSessionID, ...partData } = part
+            tx.insert(MessageTable).values({ id, session_id: sessionID, time_created: now, data: info }).run()
+            tx.insert(PartTable)
+              .values({ id: partID, message_id: messageID, session_id: partSessionID, time_created: now, data: partData })
+              .run()
+            return { msg, part, time: now }
+          },
+          { behavior: "immediate" },
+        ),
+      )
+      if (!created) return
+      yield* session.updateMessage(created.msg)
+      yield* session.updatePart(created.part)
       if (flags.experimentalEventSystem) {
         yield* events.publish(SessionEvent.Compaction.Started, {
           sessionID: input.sessionID,

packages/opencode/src/session/prompt.ts

@@ -1698,7 +1698,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
           if (task?.type === "compaction") {
             const result = yield* compaction.process({
               messages: msgs,
-              parentID: lastUser.id,
+              parentID: task.messageID,
               sessionID,
               auto: task.auto,
               overflow: task.overflow,