fix(core): persist explicit compaction exclusions

packages/client/src/effect/api/api.ts

@@ -1302,6 +1302,7 @@ export type ExperimentalSessionLogOutput =
             readonly model?: Model.Ref | undefined
             readonly native?: SessionMessage.Native | undefined
             readonly text: string
+            readonly pendingUserIDs?: ReadonlyArray<SessionMessage.ID> | undefined
             readonly cost?: Money.USD | undefined
             readonly tokens?:
               | {

packages/client/src/promise/generated/types.ts

@@ -1553,6 +1553,7 @@ export type SessionCompactionEnded = {
     model?: ModelRef
     native?: SessionMessageNative1
     text: string
+    pendingUserIDs?: Array<SessionMessageID>
     cost?: MoneyUSD
     tokens?: TokenUsageInfo
   }

packages/core/src/session/compaction.ts

@@ -946,6 +946,13 @@ export const layer = Layer.effect(
         return { status: "failed", error: outcome.error }
       }
 
+      const metadata =
+        trigger.reason === "manual"
+          ? outcome.metadata
+          : {
+              pendingUserIDs: SessionHistory.pendingUsers(context.messages).map((message) => message.id),
+              ...outcome.metadata,
+            }
       yield* bus.publish(
         SessionEvent.Compaction.Ended,
         {
@@ -954,16 +961,12 @@ export const layer = Layer.effect(
           model: context.model.ref,
           native: outcome.native,
           text: outcome.text,
+          pendingUserIDs:
+            trigger.reason === "manual" ? undefined : SessionHistory.pendingUserIDs({ type: "compaction", metadata }),
           ...usage,
         },
         {
-          metadata:
-            trigger.reason === "manual"
-              ? outcome.metadata
-              : {
-                  pendingUserIDs: SessionHistory.pendingUsers(context.messages).map((message) => message.id),
-                  ...outcome.metadata,
-                },
+          metadata: outcome.metadata,
         },
       )
       return { status: "completed" }

packages/core/src/session/message-updater.ts

@@ -426,12 +426,19 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
         ),
       "session.compaction.ended": (event) =>
         Effect.gen(function* () {
+          const metadata =
+            event.data.pendingUserIDs === undefined
+              ? event.metadata
+              : {
+                  ...event.metadata,
+                  pendingUserIDs: event.data.pendingUserIDs,
+                }
           const current = yield* adapter.getCompaction()
           if (current?.status === "running") {
             yield* adapter.updateCompaction({
               ...current,
               status: "completed",
-              metadata: event.metadata ? { ...current.metadata, ...event.metadata } : current.metadata,
+              metadata: metadata ? { ...current.metadata, ...metadata } : current.metadata,
               reason: event.data.reason,
               model: event.data.model,
               native: event.data.native,
@@ -446,7 +453,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
               id: SessionMessage.ID.fromEvent(event.id),
               type: "compaction",
               status: "completed",
-              metadata: event.metadata,
+              metadata,
               reason: event.data.reason,
               model: event.data.model,
               native: event.data.native,

packages/core/src/session/projector.ts

@@ -871,36 +871,7 @@ const layer = Layer.effectDiscard(
     yield* bus.project(SessionEvent.Compaction.Started, (event) => run(db, event))
     yield* bus.project(SessionEvent.Compaction.Ended, (event) =>
       Effect.gen(function* () {
-        // Derive replay references from the ordered history so rebuilding the projection retains pending input too.
-        const derive = event.data.reason !== "manual" && event.metadata?.pendingUserIDs === undefined
-        const session = derive
-          ? yield* db
-              .select({ model: SessionTable.model })
-              .from(SessionTable)
-              .where(eq(SessionTable.id, event.data.sessionID))
-              .get()
-              .pipe(Effect.orDie)
-          : undefined
-        const pending = derive
-          ? SessionHistory.pendingUsers(
-              yield* SessionHistory.load(
-                db,
-                event.data.sessionID,
-                (event.data.native
-                  ? event.data.model
-                  : session?.model && Schema.decodeUnknownSync(Model.Ref)(session.model)) ?? "latest",
-              ).pipe(Effect.orDie),
-            )
-          : undefined
-        yield* run(
-          db,
-          pending
-            ? {
-                ...event,
-                metadata: { ...event.metadata, pendingUserIDs: pending.map((message) => message.id) },
-              }
-            : event,
-        )
+        yield* run(db, event)
         yield* InstructionState.advanceEpoch(db, event.data.sessionID, event.durable.seq)
       }),
     )

packages/core/test/session-compaction.test.ts

@@ -1243,40 +1243,45 @@ it.effect("automatic compaction keeps the last completed exchange beside pending
   }),
 )
 
-it.effect("compaction replay uses the session model boundary when a different model wrote the summary", () =>
-  Effect.gen(function* () {
-    const session = yield* insertSession(Session.ID.create(), { model: resolved.ref })
-    const old = textUser("Input already covered by a native checkpoint")
-    const pending = textUser("Only this request is pending")
-    yield* storeMessages(session.id, [
-      old,
-      SessionMessage.CompactionCompleted.make({
-        ...completedSummary(""),
-        model: resolved.ref,
-        native: { messages: [Message.user("Native context")] },
-      }),
-      pending,
-    ])
-    const bus = yield* Bus.Service
-    yield* bus.replay({
-      id: Event.ID.create(),
-      type: Bus.versionedType(SessionEvent.Compaction.Ended.type, 1),
-      aggregateID: session.id,
-      seq: 4,
-      created: 0,
-      data: {
+for (const preserve of [false, true]) {
+  it.effect(
+    `compaction replay preserves explicit exclusions without reinterpreting old summaries (preserve=${preserve})`,
+    () =>
+      Effect.gen(function* () {
+        const session = yield* insertSession(Session.ID.create(), { model: resolved.ref })
+        const old = textUser("Input already covered by a native checkpoint")
+        const pending = textUser("Only this request is pending")
+        yield* storeMessages(session.id, [
+          old,
+          SessionMessage.CompactionCompleted.make({
+            ...completedSummary(""),
+            model: resolved.ref,
+            native: { messages: [Message.user("Native context")] },
+          }),
+          pending,
+        ])
+      const bus = yield* Bus.Service
+      const data = {
         sessionID: session.id,
         reason: "auto",
         model: { id: "other-model", providerID: "other" },
         text: "## Objective\n- Compacted history",
-      },
-    })
-    const store = yield* SessionStore.Service
-    const history = yield* store.context(session.id)
-    expect(history[0].metadata?.pendingUserIDs).toEqual([pending.id])
-    expect(history.slice(1)).toEqual([pending])
-  }),
-)
+      }
+      yield* bus.replay({
+          id: Event.ID.create(),
+          type: Bus.versionedType(SessionEvent.Compaction.Ended.type, 1),
+          aggregateID: session.id,
+          seq: 4,
+          created: 0,
+        data: preserve ? { ...data, pendingUserIDs: [pending.id] } : data,
+        })
+        const store = yield* SessionStore.Service
+        const history = yield* store.context(session.id)
+        expect(history[0].metadata?.pendingUserIDs).toEqual(preserve ? [pending.id] : undefined)
+        expect(history.slice(1)).toEqual(preserve ? [pending] : [])
+      }),
+  )
+}
 
 it.effect("background compaction does not repeat the same prefix after its pending tail outgrows the limit", () =>
   Effect.gen(function* () {

packages/schema/src/session-event.ts

@@ -614,6 +614,8 @@ export namespace Compaction {
       model: SessionMessage.CompactionCompleted.fields.model,
       native: SessionMessage.CompactionCompleted.fields.native,
       text: Schema.String,
+      // These requests were excluded from the replacement context and must replay verbatim after it.
+      pendingUserIDs: Schema.Array(SessionMessage.ID).pipe(optional),
       // Repeats the internal `session.usage.recorded` figures: that event never reaches clients, and it
       // stays the accounting source for session totals and stats.
       cost: SessionMessage.CompactionCompleted.fields.cost,