diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 9998568b229..b330bee8332 100644 --- a/packages/client/src/effect/api/api.ts +++ b/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 | undefined readonly cost?: Money.USD | undefined readonly tokens?: | { diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 7650139b5e7..ccbb2cbcd4d 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -1553,6 +1553,7 @@ export type SessionCompactionEnded = { model?: ModelRef native?: SessionMessageNative1 text: string + pendingUserIDs?: Array cost?: MoneyUSD tokens?: TokenUsageInfo } diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index fd204930a32..76d15a33d56 100644 --- a/packages/core/src/session/compaction.ts +++ b/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" } diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 24d9ce35280..d2d89edf022 100644 --- a/packages/core/src/session/message-updater.ts +++ b/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, diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 304237b180f..ec3bd41f818 100644 --- a/packages/core/src/session/projector.ts +++ b/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) }), ) diff --git a/packages/core/test/session-compaction.test.ts b/packages/core/test/session-compaction.test.ts index 316aadb55ea..d51654ddd93 100644 --- a/packages/core/test/session-compaction.test.ts +++ b/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* () { diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 0b6af19e6e4..5493d0fea33 100644 --- a/packages/schema/src/session-event.ts +++ b/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,