test(session): clarify compaction test harness

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

@@ -1,8 +1,7 @@
 import { afterEach, describe, expect, mock, test } from "bun:test"
 import { APICallError } from "ai"
-import { Cause, Deferred, Effect, Exit, Layer, ManagedRuntime } from "effect"
+import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
 import * as Stream from "effect/Stream"
-import z from "zod"
 import { Bus } from "../../src/bus"
 import { Config } from "@/config/config"
 import { Image } from "@/image/image"
@@ -10,11 +9,10 @@ import { Agent } from "../../src/agent/agent"
 import { LLM } from "../../src/session/llm"
 import { SessionCompaction } from "../../src/session/compaction"
 import { Token } from "@/util/token"
-import { WithInstance } from "../../src/project/with-instance"
 import * as Log from "@opencode-ai/core/util/log"
 import { Permission } from "../../src/permission"
 import { Plugin } from "../../src/plugin"
-import { provideTmpdirInstance, TestInstance, tmpdir } from "../fixture/fixture"
+import { provideTmpdirInstance, TestInstance } from "../fixture/fixture"
 import { Session as SessionNs } from "@/session/session"
 import { MessageV2 } from "../../src/session/message-v2"
 import { MessageID, PartID, SessionID } from "../../src/session/schema"
@@ -32,26 +30,6 @@ import { TestConfig } from "../fixture/config"
 
 void Log.init({ print: false })
 
-function run<A, E>(fx: Effect.Effect<A, E, SessionNs.Service>) {
-  return Effect.runPromise(fx.pipe(Effect.provide(SessionNs.defaultLayer)))
-}
-
-const svc = {
-  ...SessionNs,
-  create(input?: SessionNs.CreateInput) {
-    return run(SessionNs.Service.use((svc) => svc.create(input)))
-  },
-  messages(input: z.output<typeof SessionNs.MessagesInput.zod>) {
-    return run(SessionNs.Service.use((svc) => svc.messages(input)))
-  },
-  updateMessage<T extends MessageV2.Info>(msg: T) {
-    return run(SessionNs.Service.use((svc) => svc.updateMessage(msg)))
-  },
-  updatePart<T extends MessageV2.Part>(part: T) {
-    return run(SessionNs.Service.use((svc) => svc.updatePart(part)))
-  },
-}
-
 const summary = Layer.succeed(
   SessionSummary.Service,
   SessionSummary.Service.of({
@@ -102,50 +80,6 @@ function createModel(opts: {
 
 const wide = () => ProviderTest.fake({ model: createModel({ context: 100_000, output: 32_000 }) })
 
-async function user(sessionID: SessionID, text: string) {
-  const msg = await svc.updateMessage({
-    id: MessageID.ascending(),
-    role: "user",
-    sessionID,
-    agent: "build",
-    model: ref,
-    time: { created: Date.now() },
-  })
-  await svc.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 svc.updateMessage(msg)
-  return msg
-}
-
 function createUserMessage(sessionID: SessionID, text: string) {
   return Effect.gen(function* () {
     const ssn = yield* SessionNs.Service
@@ -193,8 +127,10 @@ function createAssistantMessage(sessionID: SessionID, parentID: MessageID, root:
   )
 }
 
-async function summaryAssistant(sessionID: SessionID, parentID: MessageID, root: string, text: string) {
+function createSummaryAssistantMessage(sessionID: SessionID, parentID: MessageID, root: string, text: string) {
-  const msg: MessageV2.Assistant = {
+  return SessionNs.Service.use((ssn) =>
+    Effect.gen(function* () {
+      const msg = yield* ssn.updateMessage({
         id: MessageID.ascending(),
         role: "assistant",
         sessionID,
@@ -214,9 +150,8 @@ async function summaryAssistant(sessionID: SessionID, parentID: MessageID, root:
         summary: true,
         time: { created: Date.now() },
         finish: "end_turn",
-  }
+      })
-  await svc.updateMessage(msg)
+      yield* ssn.updatePart({
-  await svc.updatePart({
         id: PartID.ascending(),
         messageID: msg.id,
         sessionID,
@@ -224,6 +159,8 @@ async function summaryAssistant(sessionID: SessionID, parentID: MessageID, root:
         text,
       })
       return msg
+    }),
+  )
 }
 
 function createCompactionMarker(sessionID: SessionID) {
@@ -248,10 +185,6 @@ function createCompactionMarker(sessionID: SessionID) {
   )
 }
 
-async function createCompactionMarkerAsync(sessionID: SessionID) {
-  return run(createCompactionMarker(sessionID))
-}
-
 function fake(
   input: Parameters<SessionProcessorModule.SessionProcessor.Interface["create"]>[0],
   result: "continue" | "compact",
@@ -283,26 +216,6 @@ function cfg(compaction?: Config.Info["compaction"]) {
   })
 }
 
-function runtime(
-  result: "continue" | "compact",
-  plugin = Plugin.defaultLayer,
-  provider = ProviderTest.fake(),
-  config = Config.defaultLayer,
-) {
-  const bus = Bus.layer
-  return ManagedRuntime.make(
-    Layer.mergeAll(SessionCompaction.layer, bus).pipe(
-      Layer.provide(provider.layer),
-      Layer.provide(SessionNs.defaultLayer),
-      Layer.provide(layer(result)),
-      Layer.provide(Agent.defaultLayer),
-      Layer.provide(plugin),
-      Layer.provide(bus),
-      Layer.provide(config),
-    ),
-  )
-}
-
 const deps = Layer.mergeAll(
   wide().layer,
   layer("continue"),
@@ -320,16 +233,22 @@ const env = Layer.mergeAll(
 
 const it = testEffect(env)
 
-const processEnv = Layer.mergeAll(SessionNs.defaultLayer, CrossSpawnSpawner.defaultLayer)
+const compactionEnv = Layer.mergeAll(SessionNs.defaultLayer, CrossSpawnSpawner.defaultLayer)
-const itProcess = testEffect(processEnv)
+const itCompaction = testEffect(compactionEnv)
 
-function compactionProcessLayer(options?: {
+type CompactionProcessOptions = {
   result?: "continue" | "compact"
   llm?: Layer.Layer<LLM.Service>
   plugin?: Layer.Layer<Plugin.Service>
   provider?: ReturnType<typeof ProviderTest.fake>
   config?: Layer.Layer<Config.Service>
-}) {
+}
+
+function withCompaction(options?: CompactionProcessOptions) {
+  return Effect.provide(compactionProcessLayer(options))
+}
+
+function compactionProcessLayer(options?: CompactionProcessOptions) {
   const bus = Bus.layer
   const status = SessionStatus.layer.pipe(Layer.provide(bus))
   const processor = options?.llm
@@ -365,10 +284,6 @@ function readCompactionPart(sessionID: SessionID) {
   )
 }
 
-async function lastCompactionPart(sessionID: SessionID) {
-  return run(readCompactionPart(sessionID))
-}
-
 function llm() {
   const queue: Array<
     Stream.Stream<LLM.Event, unknown> | ((input: LLM.StreamInput) => Stream.Stream<LLM.Event, unknown>)
@@ -391,29 +306,6 @@ function llm() {
   }
 }
 
-function liveRuntime(layer: Layer.Layer<LLM.Service>, provider = ProviderTest.fake(), config = Config.defaultLayer) {
-  const bus = Bus.layer
-  const status = SessionStatus.layer.pipe(Layer.provide(bus))
-  const processor = SessionProcessorModule.SessionProcessor.layer.pipe(
-    Layer.provide(summary),
-    Layer.provide(Image.defaultLayer),
-  )
-  return ManagedRuntime.make(
-    Layer.mergeAll(SessionCompaction.layer.pipe(Layer.provide(processor)), processor, bus, status).pipe(
-      Layer.provide(provider.layer),
-      Layer.provide(SessionNs.defaultLayer),
-      Layer.provide(Snapshot.defaultLayer),
-      Layer.provide(layer),
-      Layer.provide(Permission.defaultLayer),
-      Layer.provide(Agent.defaultLayer),
-      Layer.provide(Plugin.defaultLayer),
-      Layer.provide(status),
-      Layer.provide(bus),
-      Layer.provide(config),
-    ),
-  )
-}
-
 function reply(
   text: string,
   capture?: (input: LLM.StreamInput) => void,
@@ -469,23 +361,14 @@ function reply(
   }
 }
 
-function wait(ms = 50) {
+function plugin(ready: Deferred.Deferred<void>) {
-  return new Promise((resolve) => setTimeout(resolve, ms))
-}
-
-function defer() {
-  let resolve!: () => void
-  const promise = new Promise<void>((done) => {
-    resolve = done
-  })
-  return { promise, resolve }
-}
-
-function plugin(ready: ReturnType<typeof defer>) {
   return Layer.mock(Plugin.Service)({
     trigger: <Name extends string, Input, Output>(name: Name, _input: Input, output: Output) => {
       if (name !== "experimental.session.compacting") return Effect.succeed(output)
-      return Effect.sync(() => ready.resolve()).pipe(Effect.andThen(Effect.never), Effect.as(output))
+      return Effect.sync(() => Deferred.doneUnsafe(ready, Effect.void)).pipe(
+        Effect.andThen(Effect.never),
+        Effect.as(output),
+      )
     },
     list: () => Effect.succeed([]),
     init: () => Effect.void,
@@ -980,7 +863,7 @@ describe("session.compaction.process", () => {
     }),
   )
 
-  itProcess.instance(
+  itCompaction.instance(
     "marks summary message as errored on compact result",
     Effect.gen(function* () {
       const ssn = yield* SessionNs.Service
@@ -1005,7 +888,7 @@ describe("session.compaction.process", () => {
         expect(summary.info.finish).toBe("error")
         expect(JSON.stringify(summary.info.error)).toContain("Session too large to compact")
       }
-    }).pipe(Effect.provide(compactionProcessLayer({ result: "compact" }))),
+    }).pipe(withCompaction({ result: "compact" })),
   )
 
   it.instance(
@@ -1039,7 +922,7 @@ describe("session.compaction.process", () => {
     }),
   )
 
-  itProcess.instance(
+  itCompaction.instance(
     "persists tail_start_id for retained recent turns",
     Effect.gen(function* () {
       const ssn = yield* SessionNs.Service
@@ -1062,10 +945,10 @@ describe("session.compaction.process", () => {
       const part = yield* readCompactionPart(session.id)
       expect(part?.type).toBe("compaction")
       expect(part?.tail_start_id).toBe(keep.id)
-    }).pipe(Effect.provide(compactionProcessLayer({ config: cfg({ tail_turns: 2, preserve_recent_tokens: 10_000 }) }))),
+    }).pipe(withCompaction({ config: cfg({ tail_turns: 2, preserve_recent_tokens: 10_000 }) })),
   )
 
-  itProcess.instance(
+  itCompaction.instance(
     "shrinks retained tail to fit preserve token budget",
     Effect.gen(function* () {
       const ssn = yield* SessionNs.Service
@@ -1088,10 +971,10 @@ describe("session.compaction.process", () => {
       const part = yield* readCompactionPart(session.id)
       expect(part?.type).toBe("compaction")
       expect(part?.tail_start_id).toBe(keep.id)
-    }).pipe(Effect.provide(compactionProcessLayer({ config: cfg({ tail_turns: 2, preserve_recent_tokens: 100 }) }))),
+    }).pipe(withCompaction({ config: cfg({ tail_turns: 2, preserve_recent_tokens: 100 }) })),
   )
 
-  itProcess.instance(
+  itCompaction.instance(
     "falls back to full summary when even one recent turn exceeds preserve token budget",
     () => {
       const stub = llm()
@@ -1113,16 +996,12 @@ describe("session.compaction.process", () => {
         expect(part?.type).toBe("compaction")
         expect(part?.tail_start_id).toBeUndefined()
         expect(captured).toContain("yyyy")
-      }).pipe(
+      }).pipe(withCompaction({ llm: stub.layer, config: cfg({ tail_turns: 1, preserve_recent_tokens: 20 }) }))
-        Effect.provide(
-          compactionProcessLayer({ llm: stub.layer, config: cfg({ tail_turns: 1, preserve_recent_tokens: 20 }) }),
-        ),
-      )
     },
     { git: true },
   )
 
-  itProcess.instance(
+  itCompaction.instance(
     "falls back to full summary when retained tail media exceeds preserve token budget",
     () => {
       const stub = llm()
@@ -1154,16 +1033,12 @@ describe("session.compaction.process", () => {
         expect(part?.tail_start_id).toBeUndefined()
         expect(captured).toContain("recent image turn")
         expect(captured).toContain("Attached image/png: big.png")
-      }).pipe(
+      }).pipe(withCompaction({ llm: stub.layer, config: cfg({ tail_turns: 1, preserve_recent_tokens: 100 }) }))
-        Effect.provide(
-          compactionProcessLayer({ llm: stub.layer, config: cfg({ tail_turns: 1, preserve_recent_tokens: 100 }) }),
-        ),
-      )
     },
     { git: true },
   )
 
-  itProcess.instance(
+  itCompaction.instance(
     "retains a split turn suffix when a later message fits the preserve token budget",
     () => {
       const stub = llm()
@@ -1209,16 +1084,12 @@ describe("session.compaction.process", () => {
         expect(filtered[1]?.info.role).toBe("assistant")
         expect(filtered[1]?.info.role === "assistant" ? filtered[1].info.summary : false).toBe(true)
         expect(filtered.map((msg) => msg.info.id)).not.toContain(large.id)
-      }).pipe(
+      }).pipe(withCompaction({ llm: stub.layer, config: cfg({ tail_turns: 1, preserve_recent_tokens: 100 }) }))
-        Effect.provide(
-          compactionProcessLayer({ llm: stub.layer, config: cfg({ tail_turns: 1, preserve_recent_tokens: 100 }) }),
-        ),
-      )
     },
     { git: true },
   )
 
-  itProcess.instance(
+  itCompaction.instance(
     "allows plugins to disable synthetic continue prompt",
     Effect.gen(function* () {
       const ssn = yield* SessionNs.Service
@@ -1247,7 +1118,7 @@ describe("session.compaction.process", () => {
             ),
         ),
       ).toBe(false)
-    }).pipe(Effect.provide(compactionProcessLayer({ plugin: autocontinue(false) }))),
+    }).pipe(withCompaction({ plugin: autocontinue(false) })),
   )
 
   it.instance(
@@ -1315,9 +1186,10 @@ describe("session.compaction.process", () => {
     }),
   )
 
-  test("stops quickly when aborted during retry backoff", async () => {
+  itCompaction.instance(
+    "stops quickly when aborted during retry backoff",
+    () => {
       const stub = llm()
-    const ready = defer()
       stub.push(
         Stream.fromAsyncIterable(
           {
@@ -1338,133 +1210,77 @@ describe("session.compaction.process", () => {
         ),
       )
 
-    await using tmp = await tmpdir({ git: true })
+      return Effect.gen(function* () {
-    await WithInstance.provide({
+        const ssn = yield* SessionNs.Service
-      directory: tmp.path,
+        const bus = yield* Bus.Service
-      fn: async () => {
+        const ready = yield* Deferred.make<void>()
-        const session = await svc.create({})
+        const session = yield* ssn.create({})
-        const msg = await user(session.id, "hello")
+        const msg = yield* createUserMessage(session.id, "hello")
-        const msgs = await svc.messages({ sessionID: session.id })
+        const msgs = yield* ssn.messages({ sessionID: session.id })
-        const abort = new AbortController()
+        const off = yield* bus.subscribeCallback(SessionStatus.Event.Status, (evt) => {
-        const rt = liveRuntime(stub.layer, wide())
-        let off: (() => void) | undefined
-        let run: Promise<"continue" | "stop"> | undefined
-        try {
-          off = await rt.runPromise(
-            Bus.Service.use((svc) =>
-              svc.subscribeCallback(SessionStatus.Event.Status, (evt) => {
           if (evt.properties.sessionID !== session.id) return
           if (evt.properties.status.type !== "retry") return
-                ready.resolve()
+          Deferred.doneUnsafe(ready, Effect.void)
-              }),
+        })
-            ),
+        yield* Effect.addFinalizer(() => Effect.sync(off))
-          )
 
-          run = rt
+        const fiber = yield* SessionCompaction.use
-            .runPromiseExit(
+          .process({
-              SessionCompaction.Service.use((svc) =>
-                svc.process({
             parentID: msg.id,
             messages: msgs,
             sessionID: session.id,
             auto: false,
-                }),
-              ),
-              { signal: abort.signal },
-            )
-            .then((exit) => {
-              if (Exit.isFailure(exit)) {
-                if (Cause.hasInterrupts(exit.cause) && abort.signal.aborted) return "stop"
-                throw Cause.squash(exit.cause)
-              }
-              return exit.value
           })
+          .pipe(Effect.forkChild)
 
-          await Promise.race([
+        yield* Deferred.await(ready).pipe(Effect.timeout("1 second"))
-            ready.promise,
-            wait(1000).then(() => {
-              throw new Error("timed out waiting for retry status")
-            }),
-          ])
-
         const start = Date.now()
-          abort.abort()
+        yield* Fiber.interrupt(fiber)
-          const result = await Promise.race([
+        const exit = yield* Fiber.await(fiber).pipe(Effect.timeout("250 millis"))
-            run.then((value) => ({ kind: "done" as const, value, ms: Date.now() - start })),
-            wait(250).then(() => ({ kind: "timeout" as const })),
-          ])
 
-          expect(result.kind).toBe("done")
+        expect(Exit.isFailure(exit)).toBe(true)
-          if (result.kind === "done") {
+        if (Exit.isFailure(exit)) {
-            expect(result.value).toBe("stop")
+          expect(Cause.hasInterrupts(exit.cause)).toBe(true)
-            expect(result.ms).toBeLessThan(250)
+          expect(Date.now() - start).toBeLessThan(250)
-          }
-        } finally {
-          off?.()
-          abort.abort()
-          await rt.dispose()
-          await run?.catch(() => undefined)
         }
+      }).pipe(withCompaction({ llm: stub.layer }))
     },
-    })
+    { git: true },
-  })
+  )
 
-  test("does not leave a summary assistant when aborted before processor setup", async () => {
+  itCompaction.instance(
-    const ready = defer()
+    "does not leave a summary assistant when aborted before processor setup",
-
+    () =>
-    await using tmp = await tmpdir({ git: true })
+      Effect.gen(function* () {
-    await WithInstance.provide({
+        const ready = yield* Deferred.make<void>()
-      directory: tmp.path,
+        return yield* Effect.gen(function* () {
-      fn: async () => {
+          const ssn = yield* SessionNs.Service
-        const session = await svc.create({})
+          const session = yield* ssn.create({})
-        const msg = await user(session.id, "hello")
+          const msg = yield* createUserMessage(session.id, "hello")
-        const msgs = await svc.messages({ sessionID: session.id })
+          const msgs = yield* ssn.messages({ sessionID: session.id })
-        const abort = new AbortController()
+          const fiber = yield* SessionCompaction.use
-        const rt = runtime("continue", plugin(ready), wide())
+            .process({
-        let run: Promise<"continue" | "stop"> | undefined
-        try {
-          run = rt
-            .runPromiseExit(
-              SessionCompaction.Service.use((svc) =>
-                svc.process({
               parentID: msg.id,
               messages: msgs,
               sessionID: session.id,
               auto: false,
-                }),
-              ),
-              { signal: abort.signal },
-            )
-            .then((exit) => {
-              if (Exit.isFailure(exit)) {
-                if (Cause.hasInterrupts(exit.cause) && abort.signal.aborted) return "stop"
-                throw Cause.squash(exit.cause)
-              }
-              return exit.value
             })
+            .pipe(Effect.forkChild)
 
-          await Promise.race([
+          yield* Deferred.await(ready).pipe(Effect.timeout("1 second"))
-            ready.promise,
+          yield* Fiber.interrupt(fiber)
-            wait(1000).then(() => {
+          const exit = yield* Fiber.await(fiber).pipe(Effect.timeout("250 millis"))
-              throw new Error("timed out waiting for compaction hook")
+          const all = yield* ssn.messages({ sessionID: session.id })
-            }),
-          ])
 
-          abort.abort()
+          expect(Exit.isFailure(exit)).toBe(true)
-          expect(await run).toBe("stop")
+          if (Exit.isFailure(exit)) expect(Cause.hasInterrupts(exit.cause)).toBe(true)
-
-          const all = await svc.messages({ sessionID: session.id })
           expect(all.some((msg) => msg.info.role === "assistant" && msg.info.summary)).toBe(false)
-        } finally {
+        }).pipe(withCompaction({ plugin: plugin(ready) }))
-          abort.abort()
+      }),
-          await rt.dispose()
+    { git: true },
-          await run?.catch(() => undefined)
+  )
-        }
-      },
-    })
-  })
 
-  itProcess.instance(
+  itCompaction.instance(
     "does not allow tool calls while generating the summary",
     () => {
       const stub = llm()
@@ -1528,12 +1344,14 @@ describe("session.compaction.process", () => {
 
         expect(summary?.info.role).toBe("assistant")
         expect(summary?.parts.some((part) => part.type === "tool")).toBe(false)
-      }).pipe(Effect.provide(compactionProcessLayer({ llm: stub.layer })))
+      }).pipe(withCompaction({ llm: stub.layer }))
     },
     { git: true },
   )
 
-  test("summarizes only the head while keeping recent tail out of summary input", async () => {
+  itCompaction.instance(
+    "summarizes only the head while keeping recent tail out of summary input",
+    () => {
       const stub = llm()
       let captured = ""
       stub.push(
@@ -1541,45 +1359,36 @@ describe("session.compaction.process", () => {
           captured = JSON.stringify(input.messages)
         }),
       )
+      return Effect.gen(function* () {
+        const ssn = yield* SessionNs.Service
+        const session = yield* ssn.create({})
+        yield* createUserMessage(session.id, "older context")
+        yield* createUserMessage(session.id, "keep this turn")
+        yield* createUserMessage(session.id, "and this one too")
+        yield* createCompactionMarker(session.id)
 
-    await using tmp = await tmpdir({ git: true })
+        const msgs = yield* ssn.messages({ sessionID: session.id })
-    await WithInstance.provide({
-      directory: tmp.path,
-      fn: async () => {
-        const session = await svc.create({})
-        await user(session.id, "older context")
-        await user(session.id, "keep this turn")
-        await user(session.id, "and this one too")
-        await createCompactionMarkerAsync(session.id)
-
-        const rt = liveRuntime(stub.layer, wide())
-        try {
-          const msgs = await svc.messages({ sessionID: session.id })
         const parent = msgs.at(-1)?.info.id
         expect(parent).toBeTruthy()
-          await rt.runPromise(
+        yield* SessionCompaction.use.process({
-            SessionCompaction.Service.use((svc) =>
-              svc.process({
           parentID: parent!,
           messages: msgs,
           sessionID: session.id,
           auto: false,
-              }),
+        })
-            ),
-          )
 
         expect(captured).toContain("older context")
         expect(captured).not.toContain("keep this turn")
         expect(captured).not.toContain("and this one too")
         expect(captured).not.toContain("What did we do so far?")
-        } finally {
+      }).pipe(withCompaction({ llm: stub.layer }))
-          await rt.dispose()
-        }
     },
-    })
+    { git: true },
-  })
+  )
 
-  test("anchors repeated compactions with the previous summary", async () => {
+  itCompaction.instance(
+    "anchors repeated compactions with the previous summary",
+    () => {
       const stub = llm()
       let captured = ""
       stub.push(reply("summary one"))
@@ -1589,106 +1398,61 @@ describe("session.compaction.process", () => {
         }),
       )
 
-    await using tmp = await tmpdir({ git: true })
+      return Effect.gen(function* () {
-    await WithInstance.provide({
+        const ssn = yield* SessionNs.Service
-      directory: tmp.path,
+        const session = yield* ssn.create({})
-      fn: async () => {
+        yield* createUserMessage(session.id, "older context")
-        const session = await svc.create({})
+        yield* createUserMessage(session.id, "keep this turn")
-        await user(session.id, "older context")
+        yield* createCompactionMarker(session.id)
-        await user(session.id, "keep this turn")
-        await createCompactionMarkerAsync(session.id)
 
-        const rt = liveRuntime(stub.layer, wide())
+        let msgs = yield* ssn.messages({ sessionID: session.id })
-        try {
-          let msgs = await svc.messages({ sessionID: session.id })
         let parent = msgs.at(-1)?.info.id
         expect(parent).toBeTruthy()
-          await rt.runPromise(
+        yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
-            SessionCompaction.Service.use((svc) =>
-              svc.process({
-                parentID: parent!,
-                messages: msgs,
-                sessionID: session.id,
-                auto: false,
-              }),
-            ),
-          )
 
-          await user(session.id, "latest turn")
+        yield* createUserMessage(session.id, "latest turn")
-          await createCompactionMarkerAsync(session.id)
+        yield* createCompactionMarker(session.id)
 
         msgs = MessageV2.filterCompacted(MessageV2.stream(session.id))
         parent = msgs.at(-1)?.info.id
         expect(parent).toBeTruthy()
-          await rt.runPromise(
+        yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
-            SessionCompaction.Service.use((svc) =>
-              svc.process({
-                parentID: parent!,
-                messages: msgs,
-                sessionID: session.id,
-                auto: false,
-              }),
-            ),
-          )
 
         expect(captured).toContain("<previous-summary>")
         expect(captured).toContain("summary one")
         expect(captured.match(/summary one/g)?.length).toBe(1)
         expect(captured).toContain("## Constraints & Preferences")
         expect(captured).toContain("## Progress")
-        } finally {
+      }).pipe(withCompaction({ llm: stub.layer }))
-          await rt.dispose()
-        }
     },
-    })
+    { git: true },
-  })
+  )
 
-  test("keeps recent pre-compaction turns across repeated compactions", async () => {
+  itCompaction.instance("keeps recent pre-compaction turns across repeated compactions", () => {
     const stub = llm()
     stub.push(reply("summary one"))
     stub.push(reply("summary two"))
-    await using tmp = await tmpdir()
-    await WithInstance.provide({
-      directory: tmp.path,
-      fn: async () => {
-        const session = await svc.create({})
-        const u1 = await user(session.id, "one")
-        const u2 = await user(session.id, "two")
-        const u3 = await user(session.id, "three")
-        await createCompactionMarkerAsync(session.id)
 
-        const rt = liveRuntime(stub.layer, wide(), cfg({ tail_turns: 2, preserve_recent_tokens: 10_000 }))
+    return Effect.gen(function* () {
-        try {
+      const ssn = yield* SessionNs.Service
-          let msgs = await svc.messages({ sessionID: session.id })
+      const session = yield* ssn.create({})
+      const u1 = yield* createUserMessage(session.id, "one")
+      const u2 = yield* createUserMessage(session.id, "two")
+      const u3 = yield* createUserMessage(session.id, "three")
+      yield* createCompactionMarker(session.id)
+
+      let msgs = yield* ssn.messages({ sessionID: session.id })
       let parent = msgs.at(-1)?.info.id
       expect(parent).toBeTruthy()
-          await rt.runPromise(
+      yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
-            SessionCompaction.Service.use((svc) =>
-              svc.process({
-                parentID: parent!,
-                messages: msgs,
-                sessionID: session.id,
-                auto: false,
-              }),
-            ),
-          )
 
-          const u4 = await user(session.id, "four")
+      const u4 = yield* createUserMessage(session.id, "four")
-          await createCompactionMarkerAsync(session.id)
+      yield* createCompactionMarker(session.id)
 
       msgs = MessageV2.filterCompacted(MessageV2.stream(session.id))
       parent = msgs.at(-1)?.info.id
       expect(parent).toBeTruthy()
-          await rt.runPromise(
+      yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
-            SessionCompaction.Service.use((svc) =>
-              svc.process({
-                parentID: parent!,
-                messages: msgs,
-                sessionID: session.id,
-                auto: false,
-              }),
-            ),
-          )
 
       const filtered = MessageV2.filterCompacted(MessageV2.stream(session.id))
       const ids = filtered.map((msg) => msg.info.id)
@@ -1701,23 +1465,19 @@ describe("session.compaction.process", () => {
       expect(
         filtered.some((msg) => msg.info.role === "user" && msg.parts.some((part) => part.type === "compaction")),
       ).toBe(true)
-        } finally {
+    }).pipe(withCompaction({ llm: stub.layer, config: cfg({ tail_turns: 2, preserve_recent_tokens: 10_000 }) }))
-          await rt.dispose()
-        }
-      },
-    })
   })
 
-  test("ignores previous summaries when sizing the retained tail", async () => {
+  itCompaction.instance(
-    await using tmp = await tmpdir()
+    "ignores previous summaries when sizing the retained tail",
-    await WithInstance.provide({
+    Effect.gen(function* () {
-      directory: tmp.path,
+      const ssn = yield* SessionNs.Service
-      fn: async () => {
+      const test = yield* TestInstance
-        const session = await svc.create({})
+      const session = yield* ssn.create({})
-        await user(session.id, "older")
+      yield* createUserMessage(session.id, "older")
-        const keep = await user(session.id, "keep this turn")
+      const keep = yield* createUserMessage(session.id, "keep this turn")
-        const keepReply = await assistant(session.id, keep.id, tmp.path)
+      const keepReply = yield* createAssistantMessage(session.id, keep.id, test.directory)
-        await svc.updatePart({
+      yield* ssn.updatePart({
         id: PartID.ascending(),
         messageID: keepReply.id,
         sessionID: session.id,
@@ -1725,14 +1485,14 @@ describe("session.compaction.process", () => {
         text: "keep reply",
       })
 
-        await createCompactionMarkerAsync(session.id)
+      yield* createCompactionMarker(session.id)
-        const firstCompaction = (await svc.messages({ sessionID: session.id })).at(-1)?.info.id
+      const firstCompaction = (yield* ssn.messages({ sessionID: session.id })).at(-1)?.info.id
       expect(firstCompaction).toBeTruthy()
-        await summaryAssistant(session.id, firstCompaction!, tmp.path, "summary ".repeat(800))
+      yield* createSummaryAssistantMessage(session.id, firstCompaction!, test.directory, "summary ".repeat(800))
 
-        const recent = await user(session.id, "recent turn")
+      const recent = yield* createUserMessage(session.id, "recent turn")
-        const recentReply = await assistant(session.id, recent.id, tmp.path)
+      const recentReply = yield* createAssistantMessage(session.id, recent.id, test.directory)
-        await svc.updatePart({
+      yield* ssn.updatePart({
         id: PartID.ascending(),
         messageID: recentReply.id,
         sessionID: session.id,
@@ -1740,33 +1500,17 @@ describe("session.compaction.process", () => {
         text: "recent reply",
       })
 
-        await createCompactionMarkerAsync(session.id)
+      yield* createCompactionMarker(session.id)
-
+      const msgs = yield* ssn.messages({ sessionID: session.id })
-        const rt = runtime("continue", Plugin.defaultLayer, wide(), cfg({ tail_turns: 2, preserve_recent_tokens: 500 }))
-        try {
-          const msgs = await svc.messages({ sessionID: session.id })
       const parent = msgs.at(-1)?.info.id
       expect(parent).toBeTruthy()
-          await rt.runPromise(
+      yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
-            SessionCompaction.Service.use((svc) =>
-              svc.process({
-                parentID: parent!,
-                messages: msgs,
-                sessionID: session.id,
-                auto: false,
-              }),
-            ),
-          )
 
-          const part = await lastCompactionPart(session.id)
+      const part = yield* readCompactionPart(session.id)
       expect(part?.type).toBe("compaction")
       expect(part?.tail_start_id).toBe(keep.id)
-        } finally {
+    }).pipe(withCompaction({ config: cfg({ tail_turns: 2, preserve_recent_tokens: 500 }) })),
-          await rt.dispose()
+  )
-        }
-      },
-    })
-  })
 })
 
 describe("util.token.estimate", () => {