feat(conversations): Read background compaction events and check history version on replace

packages/junior/src/chat/conversations/fork.ts

@@ -45,6 +45,10 @@ export const UNCOPIED_EVENT_TYPES: ReadonlySet<string> = 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",

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<typeof historyReplacementSchema>;
 
+/**
+ * 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<Array<Pick<ConversationEvent, "seq" | "historyVersion">>>;
-  /** 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<void>;
+    options?: { expectedHistoryVersion?: number },
+  ): Promise<HistoryReplacementResult>;
   /** One event selected by its retry-stable key across history versions. */
   loadByIdempotencyKey(
     conversationId: string,

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<void> {
+    options: { expectedHistoryVersion?: number } = {},
+  ): Promise<HistoryReplacementResult> {
     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<number> {

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;

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." }] },
+    ]);
+  });
+});

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<void> {
+  ): Promise<HistoryReplacementResult> {
     throw new Error("not implemented");
   }