diff --git a/packages/junior/src/chat/conversations/fork.ts b/packages/junior/src/chat/conversations/fork.ts index 071f28e329..600b41ce83 100644 --- a/packages/junior/src/chat/conversations/fork.ts +++ b/packages/junior/src/chat/conversations/fork.ts @@ -45,6 +45,10 @@ export const UNCOPIED_EVENT_TYPES: ReadonlySet = new Set< >([ "authorization_completed", "authorization_requested", + // A copied request would look like a pending job that never runs for the + // fork. The fork requests its own compaction. + "compaction_prepared", + "compaction_requested", "guardian_action_reviewed", "mcp_provider_connected", "mcp_provider_connected_unowned", diff --git a/packages/junior/src/chat/conversations/history.ts b/packages/junior/src/chat/conversations/history.ts index 538c6416a3..dd395f9c7e 100644 --- a/packages/junior/src/chat/conversations/history.ts +++ b/packages/junior/src/chat/conversations/history.ts @@ -114,6 +114,14 @@ export const historyReplacementSchema = z /** One atomically persisted model-history replacement. */ export type HistoryReplacement = z.output; +/** + * Result of a history replacement. `historyVersion` is the new version when + * replaced, or the current version that did not match when in conflict. + */ +export type HistoryReplacementResult = + | { status: "replaced"; historyVersion: number } + | { status: "conflict"; historyVersion: number }; + const mcpProviderConnectedEventDataSchema = z .object({ type: z.literal("mcp_provider_connected"), @@ -398,6 +406,36 @@ const subagentEndedEventDataSchema = z }) .strict(); +// Background compaction bookkeeping. A request marks the history that a +// prepared summary covers. A later Turn swaps the summary in only when the +// History version and Model profile still match. Neither event is model context. +const compactionRequestedEventDataSchema = z + .object({ + type: z.literal("compaction_requested"), + // Last event seq that the summary covers. + markSeq: z.number().int().nonnegative(), + historyVersion: z.number().int().nonnegative(), + modelProfile: modelProfileSchema, + modelId: z.string().min(1), + }) + .strict(); + +const compactionPreparedFields = compactionRequestedEventDataSchema.extend({ + type: z.literal("compaction_prepared"), +}); + +const compactionPreparedEventDataSchema = z.discriminatedUnion("outcome", [ + compactionPreparedFields.extend({ + outcome: z.literal("ready"), + summary: z.string().min(1), + }), + compactionPreparedFields.extend({ + outcome: z.literal("failed"), + // A fixed label only. Do not store raw exception text. + errorType: z.string().min(1).max(64), + }), +]); + /** Prevent ordinary appends from bypassing history-replacement validation. */ const appendableConversationEventDataSchema = z.union([ messageEventDataSchema, @@ -419,6 +457,8 @@ const appendableConversationEventDataSchema = z.union([ turnFailedEventDataSchema, subagentStartedEventDataSchema, subagentEndedEventDataSchema, + compactionRequestedEventDataSchema, + compactionPreparedEventDataSchema, ]); /** Strict event-data contract reused by the SQL row codec. */ @@ -461,6 +501,8 @@ export const KNOWN_CONVERSATION_EVENT_TYPES = [ "subagent_ended", "handoff", "compaction", + "compaction_requested", + "compaction_prepared", ] as const; /** One known durable conversation event type name. */ @@ -609,11 +651,16 @@ export interface ConversationEventStore { events: NewConversationEvent[], options?: { activity?: "preserve" }, ): Promise>>; - /** Replace active model history with a compaction or handoff event. */ + /** + * Replace active model history with a compaction or handoff event. + * With `expectedHistoryVersion`, write nothing and return a conflict when + * the current History version differs. + */ replaceHistory( conversationId: string, replacement: HistoryReplacement, - ): Promise; + options?: { expectedHistoryVersion?: number }, + ): Promise; /** One event selected by its retry-stable key across history versions. */ loadByIdempotencyKey( conversationId: string, diff --git a/packages/junior/src/chat/conversations/sql/history.ts b/packages/junior/src/chat/conversations/sql/history.ts index d6f2bed61c..315aeac29c 100644 --- a/packages/junior/src/chat/conversations/sql/history.ts +++ b/packages/junior/src/chat/conversations/sql/history.ts @@ -22,6 +22,7 @@ import { type ConversationEventStore, type ConversationEventData, type HistoryReplacement, + type HistoryReplacementResult, type MessageHistory, type MessagesSummarizedEvent, type NewConversationEvent, @@ -295,24 +296,42 @@ class SqlConversationEventStore implements ConversationEventStore { async replaceHistory( conversationId: string, replacement: HistoryReplacement, - ): Promise { + options: { expectedHistoryVersion?: number } = {}, + ): Promise { const parsed = historyReplacementSchema.parse(replacement); - await withConversationEventLock(this.executor, conversationId, async () => { - await ensureConversationRow(this.executor, conversationId, Date.now()); - const cursor = await this.readCursor(conversationId); - const historyVersion = (cursor.maxHistoryVersion ?? 0) + 1; - await this.executor - .db() - .insert(juniorConversationEvents) - .values( - insertFromEvent( - conversationId, - cursor.nextSeq, - historyVersion, - parsed, - ), - ); - }); + return withConversationEventLock( + this.executor, + conversationId, + async () => { + const cursor = await this.readCursor(conversationId); + const currentHistoryVersion = cursor.maxHistoryVersion ?? 0; + // Checked under the event lock so a concurrent replacement or a newer + // compaction cannot slip in between the check and the write. + if ( + options.expectedHistoryVersion !== undefined && + options.expectedHistoryVersion !== currentHistoryVersion + ) { + return { + status: "conflict" as const, + historyVersion: currentHistoryVersion, + }; + } + await ensureConversationRow(this.executor, conversationId, Date.now()); + const historyVersion = currentHistoryVersion + 1; + await this.executor + .db() + .insert(juniorConversationEvents) + .values( + insertFromEvent( + conversationId, + cursor.nextSeq, + historyVersion, + parsed, + ), + ); + return { status: "replaced" as const, historyVersion }; + }, + ); } async loadCurrentHistoryVersion(conversationId: string): Promise { diff --git a/packages/junior/src/chat/pi/conversation-events.ts b/packages/junior/src/chat/pi/conversation-events.ts index 3d9086b62b..7efe02464d 100644 --- a/packages/junior/src/chat/pi/conversation-events.ts +++ b/packages/junior/src/chat/pi/conversation-events.ts @@ -131,6 +131,14 @@ export function projectConversationEvents( `Unsupported conversation event "${event.data.originalType}" at seq ${event.seq} (schema version ${event.schemaVersion})`, ); } + // Background compaction bookkeeping. A prepared summary enters model + // context only through the `compaction` event that swaps it in. + if ( + event.data.type === "compaction_requested" || + event.data.type === "compaction_prepared" + ) { + continue; + } if (event.data.type === "compaction" || event.data.type === "handoff") { modelProfile = event.data.modelProfile; replacementSeq = event.seq; diff --git a/packages/junior/tests/integration/background-compaction-events.test.ts b/packages/junior/tests/integration/background-compaction-events.test.ts new file mode 100644 index 0000000000..c6c238b775 --- /dev/null +++ b/packages/junior/tests/integration/background-compaction-events.test.ts @@ -0,0 +1,145 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { getConversationEventStore } from "@/chat/db"; +import type { HistoryReplacement } from "@/chat/conversations/history"; +import { openConversationProjection } from "@/chat/conversations/projection"; +import { contextProvenance } from "@/chat/conversations/provenance"; +import { closeConversationFixture } from "../fixtures/conversation"; + +const CONVERSATION_ID = "local:web:background-compaction"; + +function userMessage(text: string, timestamp: number) { + return { + type: "user_message" as const, + provenance: contextProvenance, + content: [{ type: "text" as const, text }], + timestamp, + }; +} + +function compaction(summary: string): HistoryReplacement { + return { + createdAtMs: 10, + data: { + type: "compaction", + modelProfile: "standard", + modelId: "test/model", + replacementHistory: [{ item: userMessage(summary, 10) }], + }, + }; +} + +describe("background compaction events", () => { + afterEach(closeConversationFixture); + + it("round-trips bookkeeping events without adding model context", async () => { + const events = getConversationEventStore(); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 1, data: userMessage("Remember the blue option.", 1) }, + { + createdAtMs: 2, + data: { + type: "compaction_requested", + markSeq: 0, + historyVersion: 0, + modelProfile: "standard", + modelId: "test/model", + }, + }, + { + createdAtMs: 3, + data: { + type: "compaction_prepared", + markSeq: 0, + historyVersion: 0, + modelProfile: "standard", + modelId: "test/model", + outcome: "ready", + summary: "The person picked the blue option.", + }, + }, + { + createdAtMs: 4, + data: { + type: "compaction_prepared", + markSeq: 0, + historyVersion: 0, + modelProfile: "standard", + modelId: "test/model", + outcome: "failed", + errorType: "timeout", + }, + }, + ]); + + const history = await events.loadCurrentHistory(CONVERSATION_ID); + expect(history.slice(1).map((event) => event.data)).toEqual([ + { + type: "compaction_requested", + markSeq: 0, + historyVersion: 0, + modelProfile: "standard", + modelId: "test/model", + }, + { + type: "compaction_prepared", + markSeq: 0, + historyVersion: 0, + modelProfile: "standard", + modelId: "test/model", + outcome: "ready", + summary: "The person picked the blue option.", + }, + { + type: "compaction_prepared", + markSeq: 0, + historyVersion: 0, + modelProfile: "standard", + modelId: "test/model", + outcome: "failed", + errorType: "timeout", + }, + ]); + + const projection = await openConversationProjection({ + conversationId: CONVERSATION_ID, + }); + expect(projection.messages).toEqual([ + { + role: "user", + content: [{ type: "text", text: "Remember the blue option." }], + timestamp: 1, + }, + ]); + }); + + it("writes nothing when the expected history version is stale", async () => { + const events = getConversationEventStore(); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 1, data: userMessage("First question.", 1) }, + ]); + await expect( + events.replaceHistory(CONVERSATION_ID, compaction("Foreground summary.")), + ).resolves.toEqual({ status: "replaced", historyVersion: 1 }); + + // A summary prepared for version 0 must not replace version 1. + await expect( + events.replaceHistory(CONVERSATION_ID, compaction("Stale summary."), { + expectedHistoryVersion: 0, + }), + ).resolves.toEqual({ status: "conflict", historyVersion: 1 }); + const afterConflict = await events.loadHistory(CONVERSATION_ID); + expect(afterConflict).toHaveLength(2); + + await expect( + events.replaceHistory(CONVERSATION_ID, compaction("Fresh summary."), { + expectedHistoryVersion: 1, + }), + ).resolves.toEqual({ status: "replaced", historyVersion: 2 }); + const projection = await openConversationProjection({ + conversationId: CONVERSATION_ID, + }); + expect(projection.messages).toMatchObject([ + { role: "user", content: [{ type: "text", text: "Fresh summary." }] }, + ]); + }); +}); diff --git a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts index 94fa8a188f..b09ee75d03 100644 --- a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts +++ b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts @@ -5,6 +5,7 @@ import type { ConversationEventQuery, ConversationEventStore, HistoryReplacement, + HistoryReplacementResult, MessageHistory, NewConversationEvent, } from "@/chat/conversations/history"; @@ -48,7 +49,7 @@ class MemoryConversationEventStore implements ConversationEventStore { async replaceHistory( _conversationId: string, _replacement: HistoryReplacement, - ): Promise { + ): Promise { throw new Error("not implemented"); }