refactor(session): effectify SessionCompaction service (#19459)

packages/opencode/src/session/compaction.ts

@@ -16,6 +16,8 @@ import { Config } from "@/config/config"
 import { NotFoundError } from "@/storage/db"
 import { ProviderTransform } from "@/provider/transform"
 import { ModelID, ProviderID } from "@/provider/schema"
+import { Effect, Layer, ServiceMap } from "effect"
+import { makeRuntime } from "@/effect/run-service"
 
 export namespace SessionCompaction {
   const log = Log.create({ service: "session.compaction" })
@@ -30,153 +32,194 @@ export namespace SessionCompaction {
   }
 
   const COMPACTION_BUFFER = 20_000
-
-  export async function isOverflow(input: { tokens: MessageV2.Assistant["tokens"]; model: Provider.Model }) {
-    const config = await Config.get()
-    if (config.compaction?.auto === false) return false
-    const context = input.model.limit.context
-    if (context === 0) return false
-
-    const count =
-      input.tokens.total ||
-      input.tokens.input + input.tokens.output + input.tokens.cache.read + input.tokens.cache.write
-
-    const reserved =
-      config.compaction?.reserved ?? Math.min(COMPACTION_BUFFER, ProviderTransform.maxOutputTokens(input.model))
-    const usable = input.model.limit.input
-      ? input.model.limit.input - reserved
-      : context - ProviderTransform.maxOutputTokens(input.model)
-    return count >= usable
-  }
-
   export const PRUNE_MINIMUM = 20_000
   export const PRUNE_PROTECT = 40_000
-
   const PRUNE_PROTECTED_TOOLS = ["skill"]
 
-  // goes backwards through parts until there are 40_000 tokens worth of tool
+  export interface Interface {
-  // calls. then erases output of previous tool calls. idea is to throw away old
+    readonly isOverflow: (input: {
-  // tool calls that are no longer relevant.
+      tokens: MessageV2.Assistant["tokens"]
-  export async function prune(input: { sessionID: SessionID }) {
+      model: Provider.Model
-    const config = await Config.get()
+    }) => Effect.Effect<boolean>
-    if (config.compaction?.prune === false) return
+    readonly prune: (input: { sessionID: SessionID }) => Effect.Effect<void>
-    log.info("pruning")
+    readonly process: (input: {
-    const msgs = await Session.messages({ sessionID: input.sessionID }).catch((err) => {
+      parentID: MessageID
-      if (NotFoundError.isInstance(err)) return undefined
+      messages: MessageV2.WithParts[]
-      throw err
+      sessionID: SessionID
-    })
+      abort: AbortSignal
-    if (!msgs) return
+      auto: boolean
-    let total = 0
+      overflow?: boolean
-    let pruned = 0
+    }) => Effect.Effect<"continue" | "stop">
-    const toPrune = []
+    readonly create: (input: {
-    let turns = 0
+      sessionID: SessionID
-
+      agent: string
-    loop: for (let msgIndex = msgs.length - 1; msgIndex >= 0; msgIndex--) {
+      model: { providerID: ProviderID; modelID: ModelID }
-      const msg = msgs[msgIndex]
+      auto: boolean
-      if (msg.info.role === "user") turns++
+      overflow?: boolean
-      if (turns < 2) continue
+    }) => Effect.Effect<void>
-      if (msg.info.role === "assistant" && msg.info.summary) break loop
-      for (let partIndex = msg.parts.length - 1; partIndex >= 0; partIndex--) {
-        const part = msg.parts[partIndex]
-        if (part.type === "tool")
-          if (part.state.status === "completed") {
-            if (PRUNE_PROTECTED_TOOLS.includes(part.tool)) continue
-
-            if (part.state.time.compacted) break loop
-            const estimate = Token.estimate(part.state.output)
-            total += estimate
-            if (total > PRUNE_PROTECT) {
-              pruned += estimate
-              toPrune.push(part)
-            }
-          }
-      }
-    }
-    log.info("found", { pruned, total })
-    if (pruned > PRUNE_MINIMUM) {
-      for (const part of toPrune) {
-        if (part.state.status === "completed") {
-          part.state.time.compacted = Date.now()
-          await Session.updatePart(part)
-        }
-      }
-      log.info("pruned", { count: toPrune.length })
-    }
   }
 
-  export async function process(input: {
+  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionCompaction") {}
-    parentID: MessageID
-    messages: MessageV2.WithParts[]
-    sessionID: SessionID
-    abort: AbortSignal
-    auto: boolean
-    overflow?: boolean
-  }) {
-    const userMessage = input.messages.findLast((m) => m.info.id === input.parentID)!.info as MessageV2.User
 
-    let messages = input.messages
+  export const layer: Layer.Layer<
-    let replay: MessageV2.WithParts | undefined
+    Service,
-    if (input.overflow) {
+    never,
-      const idx = input.messages.findIndex((m) => m.info.id === input.parentID)
+    Bus.Service | Config.Service | Session.Service | Agent.Service | Plugin.Service
-      for (let i = idx - 1; i >= 0; i--) {
+  > = Layer.effect(
-        const msg = input.messages[i]
+    Service,
-        if (msg.info.role === "user" && !msg.parts.some((p) => p.type === "compaction")) {
+    Effect.gen(function* () {
-          replay = msg
+      const bus = yield* Bus.Service
-          messages = input.messages.slice(0, i)
+      const config = yield* Config.Service
-          break
+      const session = yield* Session.Service
+      const agents = yield* Agent.Service
+      const plugin = yield* Plugin.Service
+
+      const isOverflow = Effect.fn("SessionCompaction.isOverflow")(function* (input: {
+        tokens: MessageV2.Assistant["tokens"]
+        model: Provider.Model
+      }) {
+        const cfg = yield* config.get()
+        if (cfg.compaction?.auto === false) return false
+        const context = input.model.limit.context
+        if (context === 0) return false
+
+        const count =
+          input.tokens.total ||
+          input.tokens.input + input.tokens.output + input.tokens.cache.read + input.tokens.cache.write
+
+        const reserved =
+          cfg.compaction?.reserved ?? Math.min(COMPACTION_BUFFER, ProviderTransform.maxOutputTokens(input.model))
+        const usable = input.model.limit.input
+          ? input.model.limit.input - reserved
+          : context - ProviderTransform.maxOutputTokens(input.model)
+        return count >= usable
+      })
+
+      // goes backwards through parts until there are PRUNE_PROTECT tokens worth of tool
+      // calls, then erases output of older tool calls to free context space
+      const prune = Effect.fn("SessionCompaction.prune")(function* (input: { sessionID: SessionID }) {
+        const cfg = yield* config.get()
+        if (cfg.compaction?.prune === false) return
+        log.info("pruning")
+
+        const msgs = yield* session
+          .messages({ sessionID: input.sessionID })
+          .pipe(Effect.catchIf(NotFoundError.isInstance, () => Effect.succeed(undefined)))
+        if (!msgs) return
+
+        let total = 0
+        let pruned = 0
+        const toPrune: MessageV2.ToolPart[] = []
+        let turns = 0
+
+        loop: for (let msgIndex = msgs.length - 1; msgIndex >= 0; msgIndex--) {
+          const msg = msgs[msgIndex]
+          if (msg.info.role === "user") turns++
+          if (turns < 2) continue
+          if (msg.info.role === "assistant" && msg.info.summary) break loop
+          for (let partIndex = msg.parts.length - 1; partIndex >= 0; partIndex--) {
+            const part = msg.parts[partIndex]
+            if (part.type === "tool")
+              if (part.state.status === "completed") {
+                if (PRUNE_PROTECTED_TOOLS.includes(part.tool)) continue
+                if (part.state.time.compacted) break loop
+                const estimate = Token.estimate(part.state.output)
+                total += estimate
+                if (total > PRUNE_PROTECT) {
+                  pruned += estimate
+                  toPrune.push(part)
+                }
+              }
+          }
         }
-      }
-      const hasContent =
-        replay && messages.some((m) => m.info.role === "user" && !m.parts.some((p) => p.type === "compaction"))
-      if (!hasContent) {
-        replay = undefined
-        messages = input.messages
-      }
-    }
 
-    const agent = await Agent.get("compaction")
+        log.info("found", { pruned, total })
-    const model = agent.model
+        if (pruned > PRUNE_MINIMUM) {
-      ? await Provider.getModel(agent.model.providerID, agent.model.modelID)
+          for (const part of toPrune) {
-      : await Provider.getModel(userMessage.model.providerID, userMessage.model.modelID)
+            if (part.state.status === "completed") {
-    const msg = (await Session.updateMessage({
+              part.state.time.compacted = Date.now()
-      id: MessageID.ascending(),
+              yield* session.updatePart(part)
-      role: "assistant",
+            }
-      parentID: input.parentID,
+          }
-      sessionID: input.sessionID,
+          log.info("pruned", { count: toPrune.length })
-      mode: "compaction",
+        }
-      agent: "compaction",
+      })
-      variant: userMessage.variant,
+
-      summary: true,
+      const processCompaction = Effect.fn("SessionCompaction.process")(function* (input: {
-      path: {
+        parentID: MessageID
-        cwd: Instance.directory,
+        messages: MessageV2.WithParts[]
-        root: Instance.worktree,
+        sessionID: SessionID
-      },
+        abort: AbortSignal
-      cost: 0,
+        auto: boolean
-      tokens: {
+        overflow?: boolean
-        output: 0,
+      }) {
-        input: 0,
+        const userMessage = input.messages.findLast((m) => m.info.id === input.parentID)!.info as MessageV2.User
-        reasoning: 0,
+
-        cache: { read: 0, write: 0 },
+        let messages = input.messages
-      },
+        let replay: MessageV2.WithParts | undefined
-      modelID: model.id,
+        if (input.overflow) {
-      providerID: model.providerID,
+          const idx = input.messages.findIndex((m) => m.info.id === input.parentID)
-      time: {
+          for (let i = idx - 1; i >= 0; i--) {
-        created: Date.now(),
+            const msg = input.messages[i]
-      },
+            if (msg.info.role === "user" && !msg.parts.some((p) => p.type === "compaction")) {
-    })) as MessageV2.Assistant
+              replay = msg
-    const processor = SessionProcessor.create({
+              messages = input.messages.slice(0, i)
-      assistantMessage: msg,
+              break
-      sessionID: input.sessionID,
+            }
-      model,
+          }
-      abort: input.abort,
+          const hasContent =
-    })
+            replay && messages.some((m) => m.info.role === "user" && !m.parts.some((p) => p.type === "compaction"))
-    // Allow plugins to inject context or replace compaction prompt
+          if (!hasContent) {
-    const compacting = await Plugin.trigger(
+            replay = undefined
-      "experimental.session.compacting",
+            messages = input.messages
-      { sessionID: input.sessionID },
+          }
-      { context: [], prompt: undefined },
+        }
-    )
+
-    const defaultPrompt = `Provide a detailed prompt for continuing our conversation above.
+        const agent = yield* agents.get("compaction")
+        const model = yield* Effect.promise(() =>
+          agent.model
+            ? Provider.getModel(agent.model.providerID, agent.model.modelID)
+            : Provider.getModel(userMessage.model.providerID, userMessage.model.modelID),
+        )
+        const msg = (yield* session.updateMessage({
+          id: MessageID.ascending(),
+          role: "assistant",
+          parentID: input.parentID,
+          sessionID: input.sessionID,
+          mode: "compaction",
+          agent: "compaction",
+          variant: userMessage.variant,
+          summary: true,
+          path: {
+            cwd: Instance.directory,
+            root: Instance.worktree,
+          },
+          cost: 0,
+          tokens: {
+            output: 0,
+            input: 0,
+            reasoning: 0,
+            cache: { read: 0, write: 0 },
+          },
+          modelID: model.id,
+          providerID: model.providerID,
+          time: {
+            created: Date.now(),
+          },
+        })) as MessageV2.Assistant
+        const processor = SessionProcessor.create({
+          assistantMessage: msg,
+          sessionID: input.sessionID,
+          model,
+          abort: input.abort,
+        })
+        // Allow plugins to inject context or replace compaction prompt.
+        const compacting = yield* plugin.trigger(
+          "experimental.session.compacting",
+          { sessionID: input.sessionID },
+          { context: [], prompt: undefined },
+        )
+        const defaultPrompt = `Provide a detailed prompt for continuing our conversation above.
 Focus on information that would be helpful for continuing the conversation, including what we did, what we're doing, which files we're working on, and what we're going to do next.
 The summary that you construct will be used so that another agent can read it and continue the work.
 
@@ -204,133 +247,179 @@ When constructing the summary, try to stick to this template:
 [Construct a structured list of relevant files that have been read, edited, or created that pertain to the task at hand. If all the files in a directory are relevant, include the path to the directory.]
 ---`
 
-    const promptText = compacting.prompt ?? [defaultPrompt, ...compacting.context].join("\n\n")
+        const prompt = compacting.prompt ?? [defaultPrompt, ...compacting.context].join("\n\n")
-    const msgs = structuredClone(messages)
+        const msgs = structuredClone(messages)
-    await Plugin.trigger("experimental.chat.messages.transform", {}, { messages: msgs })
+        yield* plugin.trigger("experimental.chat.messages.transform", {}, { messages: msgs })
-    const result = await processor.process({
+        const modelMessages = yield* Effect.promise(() => MessageV2.toModelMessages(msgs, model, { stripMedia: true }))
-      user: userMessage,
+        const result = yield* Effect.promise(() =>
-      agent,
+          processor.process({
-      abort: input.abort,
+            user: userMessage,
-      sessionID: input.sessionID,
+            agent,
-      tools: {},
+            abort: input.abort,
-      system: [],
-      messages: [
-        ...(await MessageV2.toModelMessages(msgs, model, { stripMedia: true })),
-        {
-          role: "user",
-          content: [
-            {
-              type: "text",
-              text: promptText,
-            },
-          ],
-        },
-      ],
-      model,
-    })
-
-    if (result === "compact") {
-      processor.message.error = new MessageV2.ContextOverflowError({
-        message: replay
-          ? "Conversation history too large to compact - exceeds model context limit"
-          : "Session too large to compact - context exceeds model limit even after stripping media",
-      }).toObject()
-      processor.message.finish = "error"
-      await Session.updateMessage(processor.message)
-      return "stop"
-    }
-
-    if (result === "continue" && input.auto) {
-      if (replay) {
-        const original = replay.info as MessageV2.User
-        const replayMsg = await Session.updateMessage({
-          id: MessageID.ascending(),
-          role: "user",
-          sessionID: input.sessionID,
-          time: { created: Date.now() },
-          agent: original.agent,
-          model: original.model,
-          format: original.format,
-          tools: original.tools,
-          system: original.system,
-          variant: original.variant,
-        })
-        for (const part of replay.parts) {
-          if (part.type === "compaction") continue
-          const replayPart =
-            part.type === "file" && MessageV2.isMedia(part.mime)
-              ? { type: "text" as const, text: `[Attached ${part.mime}: ${part.filename ?? "file"}]` }
-              : part
-          await Session.updatePart({
-            ...replayPart,
-            id: PartID.ascending(),
-            messageID: replayMsg.id,
             sessionID: input.sessionID,
-          })
+            tools: {},
+            system: [],
+            messages: [
+              ...modelMessages,
+              {
+                role: "user",
+                content: [{ type: "text", text: prompt }],
+              },
+            ],
+            model,
+          }),
+        )
+
+        if (result === "compact") {
+          processor.message.error = new MessageV2.ContextOverflowError({
+            message: replay
+              ? "Conversation history too large to compact - exceeds model context limit"
+              : "Session too large to compact - context exceeds model limit even after stripping media",
+          }).toObject()
+          processor.message.finish = "error"
+          yield* session.updateMessage(processor.message)
+          return "stop"
         }
-      } else {
+
-        const continueMsg = await Session.updateMessage({
+        if (result === "continue" && input.auto) {
+          if (replay) {
+            const original = replay.info as MessageV2.User
+            const replayMsg = yield* session.updateMessage({
+              id: MessageID.ascending(),
+              role: "user",
+              sessionID: input.sessionID,
+              time: { created: Date.now() },
+              agent: original.agent,
+              model: original.model,
+              format: original.format,
+              tools: original.tools,
+              system: original.system,
+              variant: original.variant,
+            })
+            for (const part of replay.parts) {
+              if (part.type === "compaction") continue
+              const replayPart =
+                part.type === "file" && MessageV2.isMedia(part.mime)
+                  ? { type: "text" as const, text: `[Attached ${part.mime}: ${part.filename ?? "file"}]` }
+                  : part
+              yield* session.updatePart({
+                ...replayPart,
+                id: PartID.ascending(),
+                messageID: replayMsg.id,
+                sessionID: input.sessionID,
+              })
+            }
+          }
+
+          if (!replay) {
+            const continueMsg = yield* session.updateMessage({
+              id: MessageID.ascending(),
+              role: "user",
+              sessionID: input.sessionID,
+              time: { created: Date.now() },
+              agent: userMessage.agent,
+              model: userMessage.model,
+            })
+            const text =
+              (input.overflow
+                ? "The previous request exceeded the provider's size limit due to large media attachments. The conversation was compacted and media files were removed from context. If the user was asking about attached images or files, explain that the attachments were too large to process and suggest they try again with smaller or fewer files.\n\n"
+                : "") +
+              "Continue if you have next steps, or stop and ask for clarification if you are unsure how to proceed."
+            yield* session.updatePart({
+              id: PartID.ascending(),
+              messageID: continueMsg.id,
+              sessionID: input.sessionID,
+              type: "text",
+              synthetic: true,
+              text,
+              time: {
+                start: Date.now(),
+                end: Date.now(),
+              },
+            })
+          }
+        }
+
+        if (processor.message.error) return "stop"
+        if (result === "continue") yield* bus.publish(Event.Compacted, { sessionID: input.sessionID })
+        return result
+      })
+
+      const create = Effect.fn("SessionCompaction.create")(function* (input: {
+        sessionID: SessionID
+        agent: string
+        model: { providerID: ProviderID; modelID: ModelID }
+        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() },
-          agent: userMessage.agent,
-          model: userMessage.model,
         })
-        const text =
+        yield* session.updatePart({
-          (input.overflow
-            ? "The previous request exceeded the provider's size limit due to large media attachments. The conversation was compacted and media files were removed from context. If the user was asking about attached images or files, explain that the attachments were too large to process and suggest they try again with smaller or fewer files.\n\n"
-            : "") +
-          "Continue if you have next steps, or stop and ask for clarification if you are unsure how to proceed."
-        await Session.updatePart({
           id: PartID.ascending(),
-          messageID: continueMsg.id,
+          messageID: msg.id,
-          sessionID: input.sessionID,
+          sessionID: msg.sessionID,
-          type: "text",
+          type: "compaction",
-          synthetic: true,
+          auto: input.auto,
-          text,
+          overflow: input.overflow,
-          time: {
-            start: Date.now(),
-            end: Date.now(),
-          },
         })
-      }
+      })
-    }
+
-    if (processor.message.error) return "stop"
+      return Service.of({
-    Bus.publish(Event.Compacted, { sessionID: input.sessionID })
+        isOverflow,
-    return "continue"
+        prune,
+        process: processCompaction,
+        create,
+      })
+    }),
+  )
+
+  export const defaultLayer = Layer.unwrap(
+    Effect.sync(() =>
+      layer.pipe(
+        Layer.provide(Session.defaultLayer),
+        Layer.provide(Agent.defaultLayer),
+        Layer.provide(Plugin.defaultLayer),
+        Layer.provide(Bus.layer),
+        Layer.provide(Config.defaultLayer),
+      ),
+    ),
+  )
+
+  const { runPromise } = makeRuntime(Service, defaultLayer)
+
+  export async function isOverflow(input: { tokens: MessageV2.Assistant["tokens"]; model: Provider.Model }) {
+    return runPromise((svc) => svc.isOverflow(input))
+  }
+
+  export async function prune(input: { sessionID: SessionID }) {
+    return runPromise((svc) => svc.prune(input))
+  }
+
+  export async function process(input: {
+    parentID: MessageID
+    messages: MessageV2.WithParts[]
+    sessionID: SessionID
+    abort: AbortSignal
+    auto: boolean
+    overflow?: boolean
+  }) {
+    return runPromise((svc) => svc.process(input))
   }
 
   export const create = fn(
     z.object({
       sessionID: SessionID.zod,
       agent: z.string(),
-      model: z.object({
+      model: z.object({ providerID: ProviderID.zod, modelID: ModelID.zod }),
-        providerID: ProviderID.zod,
-        modelID: ModelID.zod,
-      }),
       auto: z.boolean(),
       overflow: z.boolean().optional(),
     }),
-    async (input) => {
+    (input) => runPromise((svc) => svc.create(input)),
-      const msg = await Session.updateMessage({
-        id: MessageID.ascending(),
-        role: "user",
-        model: input.model,
-        sessionID: input.sessionID,
-        agent: input.agent,
-        time: {
-          created: Date.now(),
-        },
-      })
-      await Session.updatePart({
-        id: PartID.ascending(),
-        messageID: msg.id,
-        sessionID: msg.sessionID,
-        type: "compaction",
-        auto: input.auto,
-        overflow: input.overflow,
-      })
-    },
   )
 }

packages/opencode/test/session/compaction.test.ts

@@ -1,15 +1,30 @@
-import { describe, expect, test } from "bun:test"
+import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"
 import path from "path"
+import { Bus } from "../../src/bus"
 import { SessionCompaction } from "../../src/session/compaction"
 import { Token } from "../../src/util/token"
 import { Instance } from "../../src/project/instance"
 import { Log } from "../../src/util/log"
 import { tmpdir } from "../fixture/fixture"
 import { Session } from "../../src/session"
+import { MessageV2 } from "../../src/session/message-v2"
+import { MessageID, PartID, SessionID } from "../../src/session/schema"
+import { ModelID, ProviderID } from "../../src/provider/schema"
 import type { Provider } from "../../src/provider/provider"
+import * as ProviderModule from "../../src/provider/provider"
+import * as SessionProcessorModule from "../../src/session/processor"
 
 Log.init({ print: false })
 
+const ref = {
+  providerID: ProviderID.make("test"),
+  modelID: ModelID.make("test-model"),
+}
+
+afterEach(() => {
+  mock.restore()
+})
+
 function createModel(opts: {
   context: number
   output: number
@@ -40,6 +55,105 @@ function createModel(opts: {
   } as Provider.Model
 }
 
+async function user(sessionID: SessionID, text: string) {
+  const msg = await Session.updateMessage({
+    id: MessageID.ascending(),
+    role: "user",
+    sessionID,
+    agent: "build",
+    model: ref,
+    time: { created: Date.now() },
+  })
+  await Session.updatePart({
+    id: PartID.ascending(),
+    messageID: msg.id,
+    sessionID,
+    type: "text",
+    text,
+  })
+  return msg
+}
+
+async function assistant(sessionID: SessionID, parentID: MessageID, root: string) {
+  const msg: MessageV2.Assistant = {
+    id: MessageID.ascending(),
+    role: "assistant",
+    sessionID,
+    mode: "build",
+    agent: "build",
+    path: { cwd: root, root },
+    cost: 0,
+    tokens: {
+      output: 0,
+      input: 0,
+      reasoning: 0,
+      cache: { read: 0, write: 0 },
+    },
+    modelID: ref.modelID,
+    providerID: ref.providerID,
+    parentID,
+    time: { created: Date.now() },
+    finish: "end_turn",
+  }
+  await Session.updateMessage(msg)
+  return msg
+}
+
+async function tool(sessionID: SessionID, messageID: MessageID, tool: string, output: string) {
+  return Session.updatePart({
+    id: PartID.ascending(),
+    messageID,
+    sessionID,
+    type: "tool",
+    callID: crypto.randomUUID(),
+    tool,
+    state: {
+      status: "completed",
+      input: {},
+      output,
+      title: "done",
+      metadata: {},
+      time: { start: Date.now(), end: Date.now() },
+    },
+  })
+}
+
+function fake(
+  input: Parameters<(typeof SessionProcessorModule.SessionProcessor)["create"]>[0],
+  result: "continue" | "compact",
+): ReturnType<(typeof SessionProcessorModule.SessionProcessor)["create"]> {
+  const msg = input.assistantMessage
+  return {
+    get message() {
+      return msg
+    },
+    partFromToolCall() {
+      return {
+        id: PartID.ascending(),
+        messageID: msg.id,
+        sessionID: msg.sessionID,
+        type: "tool",
+        callID: "fake",
+        tool: "fake",
+        state: { status: "pending", input: {}, raw: "" },
+      }
+    },
+    process: async () => result,
+  }
+}
+
+function wait(ms = 50) {
+  return new Promise((resolve) => setTimeout(resolve, ms))
+}
+
+function defer() {
+  let resolve!: () => void
+  const promise = new Promise<void>((done) => {
+    resolve = done
+  })
+  return { promise, resolve }
+}
+
 describe("session.compaction.isOverflow", () => {
   test("returns true when token count exceeds usable context", async () => {
     await using tmp = await tmpdir()
@@ -227,6 +341,272 @@ describe("session.compaction.isOverflow", () => {
   })
 })
 
+describe("session.compaction.create", () => {
+  test("creates a compaction user message and part", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        const session = await Session.create({})
+
+        await SessionCompaction.create({
+          sessionID: session.id,
+          agent: "build",
+          model: ref,
+          auto: true,
+          overflow: true,
+        })
+
+        const msgs = await Session.messages({ sessionID: session.id })
+        expect(msgs).toHaveLength(1)
+        expect(msgs[0].info.role).toBe("user")
+        expect(msgs[0].parts).toHaveLength(1)
+        expect(msgs[0].parts[0]).toMatchObject({
+          type: "compaction",
+          auto: true,
+          overflow: true,
+        })
+      },
+    })
+  })
+})
+
+describe("session.compaction.prune", () => {
+  test("compacts old completed tool output", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        const session = await Session.create({})
+        const a = await user(session.id, "first")
+        const b = await assistant(session.id, a.id, tmp.path)
+        await tool(session.id, b.id, "bash", "x".repeat(200_000))
+        await user(session.id, "second")
+        await user(session.id, "third")
+
+        await SessionCompaction.prune({ sessionID: session.id })
+
+        const msgs = await Session.messages({ sessionID: session.id })
+        const part = msgs.flatMap((msg) => msg.parts).find((part) => part.type === "tool")
+        expect(part?.type).toBe("tool")
+        expect(part?.state.status).toBe("completed")
+        if (part?.type === "tool" && part.state.status === "completed") {
+          expect(part.state.time.compacted).toBeNumber()
+        }
+      },
+    })
+  })
+
+  test("skips protected skill tool output", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        const session = await Session.create({})
+        const a = await user(session.id, "first")
+        const b = await assistant(session.id, a.id, tmp.path)
+        await tool(session.id, b.id, "skill", "x".repeat(200_000))
+        await user(session.id, "second")
+        await user(session.id, "third")
+
+        await SessionCompaction.prune({ sessionID: session.id })
+
+        const msgs = await Session.messages({ sessionID: session.id })
+        const part = msgs.flatMap((msg) => msg.parts).find((part) => part.type === "tool")
+        expect(part?.type).toBe("tool")
+        if (part?.type === "tool" && part.state.status === "completed") {
+          expect(part.state.time.compacted).toBeUndefined()
+        }
+      },
+    })
+  })
+})
+
+describe("session.compaction.process", () => {
+  test("publishes compacted event on continue", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        spyOn(ProviderModule.Provider, "getModel").mockResolvedValue(createModel({ context: 100_000, output: 32_000 }))
+        spyOn(SessionProcessorModule.SessionProcessor, "create").mockImplementation((input) => fake(input, "continue"))
+
+        const session = await Session.create({})
+        const msg = await user(session.id, "hello")
+        const msgs = await Session.messages({ sessionID: session.id })
+        const done = defer()
+        let seen = false
+        const unsub = Bus.subscribe(SessionCompaction.Event.Compacted, (evt) => {
+          if (evt.properties.sessionID !== session.id) return
+          seen = true
+          done.resolve()
+        })
+
+        const result = await SessionCompaction.process({
+          parentID: msg.id,
+          messages: msgs,
+          sessionID: session.id,
+          abort: new AbortController().signal,
+          auto: false,
+        })
+
+        await Promise.race([
+          done.promise,
+          wait(500).then(() => {
+            throw new Error("timed out waiting for compacted event")
+          }),
+        ])
+        unsub()
+
+        expect(result).toBe("continue")
+        expect(seen).toBe(true)
+      },
+    })
+  })
+
+  test("marks summary message as errored on compact result", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        spyOn(ProviderModule.Provider, "getModel").mockResolvedValue(createModel({ context: 100_000, output: 32_000 }))
+        spyOn(SessionProcessorModule.SessionProcessor, "create").mockImplementation((input) => fake(input, "compact"))
+
+        const session = await Session.create({})
+        const msg = await user(session.id, "hello")
+        const result = await SessionCompaction.process({
+          parentID: msg.id,
+          messages: await Session.messages({ sessionID: session.id }),
+          sessionID: session.id,
+          abort: new AbortController().signal,
+          auto: false,
+        })
+
+        const summary = (await Session.messages({ sessionID: session.id })).find(
+          (msg) => msg.info.role === "assistant" && msg.info.summary,
+        )
+
+        expect(result).toBe("stop")
+        expect(summary?.info.role).toBe("assistant")
+        if (summary?.info.role === "assistant") {
+          expect(summary.info.finish).toBe("error")
+          expect(JSON.stringify(summary.info.error)).toContain("Session too large to compact")
+        }
+      },
+    })
+  })
+
+  test("adds synthetic continue prompt when auto is enabled", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        spyOn(ProviderModule.Provider, "getModel").mockResolvedValue(createModel({ context: 100_000, output: 32_000 }))
+        spyOn(SessionProcessorModule.SessionProcessor, "create").mockImplementation((input) => fake(input, "continue"))
+
+        const session = await Session.create({})
+        const msg = await user(session.id, "hello")
+
+        const result = await SessionCompaction.process({
+          parentID: msg.id,
+          messages: await Session.messages({ sessionID: session.id }),
+          sessionID: session.id,
+          abort: new AbortController().signal,
+          auto: true,
+        })
+
+        const msgs = await Session.messages({ sessionID: session.id })
+        const last = msgs.at(-1)
+
+        expect(result).toBe("continue")
+        expect(last?.info.role).toBe("user")
+        expect(last?.parts[0]).toMatchObject({
+          type: "text",
+          synthetic: true,
+        })
+        if (last?.parts[0]?.type === "text") {
+          expect(last.parts[0].text).toContain("Continue if you have next steps")
+        }
+      },
+    })
+  })
+
+  test("replays the prior user turn on overflow when earlier context exists", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        spyOn(ProviderModule.Provider, "getModel").mockResolvedValue(createModel({ context: 100_000, output: 32_000 }))
+        spyOn(SessionProcessorModule.SessionProcessor, "create").mockImplementation((input) => fake(input, "continue"))
+
+        const session = await Session.create({})
+        await user(session.id, "root")
+        const replay = await user(session.id, "image")
+        await Session.updatePart({
+          id: PartID.ascending(),
+          messageID: replay.id,
+          sessionID: session.id,
+          type: "file",
+          mime: "image/png",
+          filename: "cat.png",
+          url: "https://example.com/cat.png",
+        })
+        const msg = await user(session.id, "current")
+
+        const result = await SessionCompaction.process({
+          parentID: msg.id,
+          messages: await Session.messages({ sessionID: session.id }),
+          sessionID: session.id,
+          abort: new AbortController().signal,
+          auto: true,
+          overflow: true,
+        })
+
+        const last = (await Session.messages({ sessionID: session.id })).at(-1)
+
+        expect(result).toBe("continue")
+        expect(last?.info.role).toBe("user")
+        expect(last?.parts.some((part) => part.type === "file")).toBe(false)
+        expect(
+          last?.parts.some((part) => part.type === "text" && part.text.includes("Attached image/png: cat.png")),
+        ).toBe(true)
+      },
+    })
+  })
+
+  test("falls back to overflow guidance when no replayable turn exists", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        spyOn(ProviderModule.Provider, "getModel").mockResolvedValue(createModel({ context: 100_000, output: 32_000 }))
+        spyOn(SessionProcessorModule.SessionProcessor, "create").mockImplementation((input) => fake(input, "continue"))
+
+        const session = await Session.create({})
+        await user(session.id, "earlier")
+        const msg = await user(session.id, "current")
+
+        const result = await SessionCompaction.process({
+          parentID: msg.id,
+          messages: await Session.messages({ sessionID: session.id }),
+          sessionID: session.id,
+          abort: new AbortController().signal,
+          auto: true,
+          overflow: true,
+        })
+
+        const last = (await Session.messages({ sessionID: session.id })).at(-1)
+
+        expect(result).toBe("continue")
+        expect(last?.info.role).toBe("user")
+        if (last?.parts[0]?.type === "text") {
+          expect(last.parts[0].text).toContain("previous request exceeded the provider's size limit")
+        }
+      },
+    })
+  })
+})
+
 describe("util.token.estimate", () => {
   test("estimates tokens from text (4 chars per token)", () => {
     const text = "x".repeat(4000)