🤖 feat: record durable history acceptance for compaction replacement

src/common/orpc/schemas/message.ts

@@ -162,6 +162,7 @@ export const MuxMessageSchema = z.object({
   metadata: z
     .object({
       historySequence: z.number().optional(),
+      compactionReplacementNonce: z.string().min(1).optional().catch(undefined),
       // Step cuts are an optimization; malformed legacy metadata must not block chat replay.
       stepStartPartIndices: z.array(z.number()).optional().catch(undefined),
       timestamp: z.number().optional(),

src/common/types/message.ts

@@ -961,6 +961,8 @@ export interface ContextBudgetRejectedMessage {
 
 // Our custom metadata type
 export interface MuxMetadata {
+  /** Stop replaced by this durably accepted row; its enclosing id/sequence identify the receipt. */
+  compactionReplacementNonce?: string;
   /** Highest persisted history sequence included in the provider request that produced this assistant. */
   requestHistorySequence?: number;
   historySequence?: number; // Assigned by backend for global message ordering (required when writing to history)

src/common/utils/messages/contextBudgetRejection.ts

@@ -22,6 +22,8 @@ export function createContextBudgetRejectedMessage(message: MuxMessage): MuxMess
     parts: [],
     metadata: {
       historySequence: message.metadata?.historySequence,
+      // Quarantining this same occurrence must preserve its already-committed acceptance.
+      compactionReplacementNonce: message.metadata?.compactionReplacementNonce,
       timestamp: message.metadata?.timestamp,
       synthetic: true,
       uiVisible: false,

src/node/services/compactionCancellation.storage.test.ts

@@ -631,26 +631,25 @@ describe("inactive real cancellation storage", () => {
       /not configured/
     );
     const refusing = new FileCompactionCancellationStorage(foreign, workspaceId, () =>
-      Promise.resolve(false)
+      Promise.resolve(() => Promise.resolve(false))
     );
     await assert.rejects(
       refusing.mutate(mutation, () => true, mutationCommitted),
       /not verified/
     );
     expect(await storage.read()).toEqual(retained);
     let current = true;
-    const verifying = new FileCompactionCancellationStorage(
-      foreign,
-      workspaceId,
-      async (witness) => {
-        expect(witness.nonce).toBe(retained.nonce);
+    const verifying = new FileCompactionCancellationStorage(foreign, workspaceId, (witness) => {
+      expect(witness.nonce).toBe(retained.nonce);
+      expect(nodeFs.existsSync(historyWriteLockPath(h.config.rootDir, workspaceId))).toBe(false);
+      return Promise.resolve(async () => {
         expect(
           await fs.readFile(historyWriteLockPath(h.config.rootDir, workspaceId), "utf8")
         ).not.toBe("");
         current = false;
         return true;
-      }
-    );
+      });
+    });
     expect(await verifying.mutate(mutation, () => current, mutationCommitted)).toBe("superseded");
     expect(await storage.read()).toEqual(retained);
     // This injected authority exercises the seam only; real accepted-row proof belongs to H2b.
@@ -791,13 +790,26 @@ describe("inactive real cancellation storage", () => {
     expect(await storage.read()).toEqual(failed);
   });
 
+  it("overlapping Stop during generation staging returns superseded and preserves its successor", async () => {
+    let successor: ReturnType<typeof state.cancel> | undefined;
+    afterCompactionStaging(path.join(sessionDir, CONTINUOUS_COMPACTION_GENERATION_FILE), () => {
+      successor ??= state.cancel();
+    });
+    const first = await state.cancel().catch((error: unknown) => error);
+    expect(first).toBe("superseded");
+    expect(await successor).toBe("applied");
+    expect(await storage.read()).toEqual(await state.read());
+    expect((await storage.read())?.nonce).toBeDefined();
+  });
+
   it("a locally superseded publication cannot rename its staged cancellation", async () => {
     let current = true;
     const journal = h.historyService.getContinuousCompactionJournal(workspaceId);
     const advance = journal.advanceGenerationUnderHistoryLock.bind(journal);
     spyOn(journal, "advanceGenerationUnderHistoryLock").mockImplementationOnce(async (...args) => {
-      await advance(...args);
+      const result = await advance(...args);
       current = false;
+      return result;
     });
     expect(await storage.mutate(publication("stale"), () => current, mutationCommitted)).toBe(
       "superseded"
@@ -1007,8 +1019,9 @@ describe("inactive real cancellation storage", () => {
     const journal = h.historyService.getContinuousCompactionJournal(workspaceId);
     const advance = journal.advanceGenerationUnderHistoryLock.bind(journal);
     spyOn(journal, "advanceGenerationUnderHistoryLock").mockImplementationOnce(async (...args) => {
-      await advance(...args);
+      const result = await advance(...args);
       current = false;
+      return result;
     });
     expect(await storage.repair(() => current, committed)).toBeNull();
     expect(await fs.readFile(storage.path, "utf8")).toBe("{bad again");

src/node/services/compactionCancellation.ts

@@ -9,6 +9,25 @@ import { isPlainObject } from "@/common/utils/isPlainObject";
 import type { HistoryService } from "./historyService";
 import { publishCompactionFile } from "./continuousCompactionJournal";
 import { hasAmbiguousResetKeys } from "./historyScanner";
+import type { MuxMessage } from "@/common/types/message";
+
+export interface CompactionReplacementCapture {
+  nonce: string | null;
+  generation: string | undefined;
+}
+
+export type CompactionReplacementOperation =
+  | {
+      kind: "append";
+      messages: MuxMessage[];
+      /** Record ordinary input against this capture without replacing its Stop. */
+      preserveCancellation?: true;
+    }
+  | { kind: "resume"; message: MuxMessage };
+
+export type CompactionReplacementOutcome =
+  | { kind: "accepted"; witness: CompactionCancellationReplacementWitness | null }
+  | { kind: "superseded" | "skipped" };
 
 export interface CompactionCancellationSummary {
   id: string;
@@ -119,11 +138,11 @@ export class FileCompactionCancellationStorage implements CompactionCancellation
   constructor(
     private readonly history: HistoryService,
     private readonly workspaceId: string,
-    // This verifier runs under both history locks and must not re-enter them.
-    // No default authority: a caller-provided nonce alone cannot retire retention.
-    private readonly verifyReplacementUnderHistoryLock?: (
+    // Prepare expensive history evidence outside both locks; the returned verifier only
+    // revalidates it under the lock. A caller-provided nonce alone grants no authority.
+    private readonly prepareVerification?: (
       witness: CompactionCancellationReplacementWitness
-    ) => Promise<boolean>
+    ) => Promise<() => Promise<boolean>>
   ) {
     this.path = path.join(
       path.dirname(history.getContinuousCompactionJournal(workspaceId).path),
@@ -172,11 +191,16 @@ export class FileCompactionCancellationStorage implements CompactionCancellation
     }
   }
 
-  mutate(
+  async mutate(
     mutation: CompactionCancellationMutation,
     isCurrent: () => boolean,
     onCommitted: (record: CompactionCancellationRecord | null) => undefined
   ): Promise<CompactionCancellationMutationOutcome> {
+    if (!isCurrent()) return "superseded";
+    const verifyReplacementUnderHistoryLock =
+      mutation.kind === "retire" && mutation.replacementWitness
+        ? await this.prepareVerification?.(mutation.replacementWitness)
+        : undefined;
     return this.history.withCompactionStorageLock(this.workspaceId, async (_dir, checkLock) => {
       if (!isCurrent()) return "superseded";
       if (
@@ -211,9 +235,14 @@ export class FileCompactionCancellationStorage implements CompactionCancellation
         const { contents, record: committed } = serializeCancellation(mutation.record);
         // Record admission before advancing, and advancement at its commit point.
         // Unobserved failures remain blocking until a new explicit Stop captures a frontier.
-        await journal.advanceGenerationUnderHistoryLock((advanced) => {
-          frontier.generation = advanced;
-        }, checkLock);
+        const advanced = await journal.advanceGenerationUnderHistoryLock(
+          (advanced) => {
+            frontier.generation = advanced;
+          },
+          checkLock,
+          isCurrent
+        );
+        if (!advanced) return "superseded";
         return (await publishCompactionFile(
           this.path,
           contents,
@@ -252,9 +281,9 @@ export class FileCompactionCancellationStorage implements CompactionCancellation
       }
       const witness = mutation.replacementWitness;
       if (witness) {
-        if (!this.verifyReplacementUnderHistoryLock)
+        if (!verifyReplacementUnderHistoryLock)
           throw new Error("Replacement witness verification is not configured");
-        if (witness.nonce !== nonce || !(await this.verifyReplacementUnderHistoryLock(witness)))
+        if (witness.nonce !== nonce || !(await verifyReplacementUnderHistoryLock()))
           throw new Error("Replacement witness was not verified");
       } else if (current.retainUntilReplacement) return "superseded";
       await checkLock();

src/node/services/continuousCompactionJournal.ts

@@ -159,22 +159,26 @@ export class ContinuousCompactionJournalStore {
   /** Caller already holds the history lock; never re-enter the journal queue here. */
   async advanceGenerationUnderHistoryLock(
     onCommitted?: (generation: string) => undefined,
-    assertStillOwned?: () => Promise<void>
-  ): Promise<void> {
+    assertStillOwned?: () => Promise<void>,
+    isCurrent: () => boolean = () => true
+  ): Promise<boolean> {
     const generation = randomUUID();
-    await publishCompactionFile(
+    const committed = await publishCompactionFile(
       path.join(path.dirname(this.path), CONTINUOUS_COMPACTION_GENERATION_FILE),
       generation,
-      () => true,
+      isCurrent,
       // The cancellation retry must learn its exact frontier at rename, before
       // cleanup or lock release can fail or admit a foreign generation.
       () => onCommitted?.(createHash("sha256").update(generation).digest("hex")),
       assertStillOwned
     );
+    return committed;
   }
 
   advanceGeneration(): Promise<void> {
-    return this.enqueue(() => this.advanceGenerationUnderHistoryLock());
+    return this.enqueue(async () => {
+      await this.advanceGenerationUnderHistoryLock();
+    });
   }
 
   private async readUnderHistoryLock(): Promise<ContinuousCompactionJournal | null> {

src/node/services/historyAppendProvenance.test.ts

@@ -94,6 +94,24 @@ afterEach(async () => {
 });
 
 describe("history append provenance", () => {
+  test("single replacement acceptance preserves the append cursor's fixed snapshot", async () => {
+    const cursor = await startCursor();
+    const capture = await fixture.historyService.captureCompactionReplacement(ws);
+    assert(capture.success);
+    const accepted = await fixture.historyService.acceptCompactionReplacement(
+      ws,
+      capture.data,
+      {
+        kind: "append",
+        messages: [createMuxMessage("accepted", "user", "new input")],
+      },
+      { isCurrent: () => true, onCommitted: () => undefined }
+    );
+    expect(accepted).toEqual({ success: true, data: { kind: "accepted", witness: null } });
+    expect((await resume(cursor)).map((row) => row.id)).toEqual(["row-1", "row-2"]);
+    expect((await store.read()).receipt?.epoch).toBe(cursor.provenanceEpoch);
+  });
+
   test("bootstraps privately without an existing receipt and uses exact bigint stamps", async () => {
     await fs.rm(store.receiptPath);
     const cursor = await startCursor();

src/node/services/historyAppendProvenance.ts

@@ -280,7 +280,8 @@ export class HistoryAppendProvenance {
   async appendChat(
     bytes: Buffer,
     atomic = false,
-    publishAtomic?: (filePath: string, bytes: Buffer) => Promise<void>
+    publishAtomic?: (filePath: string, bytes: Buffer) => Promise<void>,
+    publishAppend?: (filePath: string, bytes: Buffer, createsFile: boolean) => Promise<void>
   ): Promise<void> {
     const transaction = transactions.getStore();
     assert(
@@ -330,7 +331,8 @@ export class HistoryAppendProvenance {
             await tailHandle.close();
           }
         }
-        await fs.appendFile(this.chatPath, bytes);
+        if (publishAppend) await publishAppend(this.chatPath, bytes, before.chat == null);
+        else await fs.appendFile(this.chatPath, bytes);
         published = true;
         const handle = await fs.open(this.chatPath, "r");
         try {

src/node/services/historyService.contextReset.test.ts

@@ -34,8 +34,9 @@ describe("empty context reset transactions", () => {
         async (...args) => {
           entered.resolve();
           await release.promise;
-          await advance(...args);
+          const result = await advance(...args);
           order.push("fenced");
+          return result;
         }
       );
       const fencing = h.historyService.fenceEmptyContext(workspaceId);