From 2062a5a3a8d9ef847c4942749862753e9f9128e2 Mon Sep 17 00:00:00 2001
From: Kit Langton <kit.langton@gmail.com>
Date: Tue, 7 Jul 2026 20:44:34 -0400
Subject: [PATCH] fix(core): steer manual compaction

---
 packages/core/src/session.ts              |  5 +++--
 packages/core/src/session/input.ts        | 13 +++++++++----
 packages/core/test/session-runner.test.ts |  8 +++++---
 3 files changed, 17 insertions(+), 9 deletions(-)

diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts
index ea18ed3909..4282d6771a 100644
--- a/packages/core/src/session.ts
+++ b/packages/core/src/session.ts
@@ -637,7 +637,7 @@ const layer = Layer.effect(
       compact: Effect.fn("V2Session.compact")(function* (input) {
         yield* result.get(input.sessionID)
         const inputID = input.id ?? SessionMessage.ID.create()
-        const admitted = yield* SessionInput.admitCompaction(db, events, {
+        const admission = yield* SessionInput.admitCompaction(db, events, {
           id: inputID,
           sessionID: input.sessionID,
         }).pipe(
@@ -647,8 +647,9 @@ const layer = Layer.effect(
               : Effect.die(defect),
           ),
         )
+        if (admission.newlyAdmitted) yield* execution.interrupt(input.sessionID)
         yield* execution.wake(input.sessionID)
-        return admitted
+        return admission.entry
       }),
       wait: Effect.fn("V2Session.wait")(function* (sessionID) {
         yield* result.get(sessionID)
diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/input.ts
index c08439d0c1..bfc0095a95 100644
--- a/packages/core/src/session/input.ts
+++ b/packages/core/src/session/input.ts
@@ -142,11 +142,12 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
     Effect.gen(function* () {
       const exact = yield* find(db, input.id)
       if (exact) {
-        if (exact.type === "compaction" && exact.sessionID === input.sessionID) return exact
+        if (exact.type === "compaction" && exact.sessionID === input.sessionID)
+          return { entry: exact, newlyAdmitted: false }
         return yield* Effect.die(new LifecycleConflict({ id: input.id }))
       }
       const pending = yield* pendingCompaction(db, input.sessionID)
-      if (pending) return pending
+      if (pending) return { entry: pending, newlyAdmitted: false }
       return yield* events
         .publish(SessionEvent.Compaction.Admitted, {
           inputID: input.id,
@@ -158,13 +159,17 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
               return Effect.die(new Error("Compaction admission event is missing aggregate sequence"))
             return pendingCompaction(db, input.sessionID).pipe(
               Effect.flatMap((stored) =>
-                stored ? Effect.succeed(stored) : Effect.die(new LifecycleConflict({ id: input.id })),
+                stored
+                  ? Effect.succeed({ entry: stored, newlyAdmitted: stored.id === input.id })
+                  : Effect.die(new LifecycleConflict({ id: input.id })),
               ),
             )
           }),
           Effect.catchDefect((defect) =>
             pendingCompaction(db, input.sessionID).pipe(
-              Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect))),
+              Effect.flatMap((stored) =>
+                stored ? Effect.succeed({ entry: stored, newlyAdmitted: false }) : Effect.die(defect),
+              ),
             ),
           ),
         )
diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts
index b98c5ef106..487577a9f4 100644
--- a/packages/core/test/session-runner.test.ts
+++ b/packages/core/test/session-runner.test.ts
@@ -1346,7 +1346,7 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
-  it.effect("runs one durable compaction barrier before later steer and queued prompts", () =>
+  it.effect("interrupts active work to run one durable compaction barrier before later prompts", () =>
     Effect.gen(function* () {
       yield* setup
       requests.length = 0
@@ -1389,7 +1389,8 @@ describe("SessionRunnerLLM", () => {
       expect(yield* SessionInput.hasPending((yield* Database.Service).db, sessionID, "steer")).toBe(false)
 
       yield* Deferred.succeed(streamGate, undefined)
-      yield* Fiber.join(active)
+      expect(yield* Fiber.await(active)).toMatchObject({ _tag: "Failure" })
+      yield* session.wait(sessionID)
 
       expect(requests).toHaveLength(4)
       expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary")
@@ -1429,7 +1430,8 @@ describe("SessionRunnerLLM", () => {
         resume: false,
       })
       yield* Deferred.succeed(streamGate, undefined)
-      yield* Fiber.join(active)
+      expect(yield* Fiber.await(active)).toMatchObject({ _tag: "Failure" })
+      yield* session.wait(sessionID)
 
       expect(requests).toHaveLength(3)
       expect(userTexts(requests[2])).toContain("Continue after failure")