diff --git a/src/browser/stories/App.tokenBudget.stories.tsx b/src/browser/stories/App.tokenBudget.stories.tsx
index 283ae0e446..288b43496b 100644
--- a/src/browser/stories/App.tokenBudget.stories.tsx
+++ b/src/browser/stories/App.tokenBudget.stories.tsx
@@ -133,7 +133,7 @@ export const Rollover: AppStory = {
     await expect(canvas.queryByText(WARNING)).not.toBeInTheDocument();
     const warning = await canvas.findByRole("button", { name: /Context budget warning/ });
     await userEvent.click(warning);
-    await expect(canvas.getByText(WARNING)).toBeVisible();
+    await waitFor(() => expect(canvas.getByText(WARNING)).toBeVisible());
     await userEvent.click(warning);
     const tool = await canvas.findByText("session_history", { exact: true });
     await userEvent.click(tool);
diff --git a/src/node/services/compactionCancellation.test.ts b/src/node/services/compactionCancellation.test.ts
new file mode 100644
index 0000000000..0ab8cb38c5
--- /dev/null
+++ b/src/node/services/compactionCancellation.test.ts
@@ -0,0 +1,1098 @@
+import { describe, expect, it, mock, spyOn } from "bun:test";
+import assert from "node:assert/strict";
+import {
+  CompactionCancellation,
+  MalformedCompactionCancellationError,
+  CompactionCancellationReadRefusedError,
+  matchesCompactionCancellation,
+  type CompactionCancellationMutation,
+  type CompactionCancellationMutationOutcome,
+  type CompactionCancellationPublication,
+  type CompactionCancellationRecord,
+  type CompactionCancellationStorage,
+  type CompactionCancellationSummary,
+} from "./compactionCancellation";
+
+const summary: CompactionCancellationSummary = {
+  id: "summary-a",
+  sequence: 3,
+  pendingFollowUp: { text: "Continue", options: { model: "test" } },
+};
+
+function cancellation(nonce: string): CompactionCancellationRecord {
+  return { version: 1, nonce, scope: { kind: "unresolved" } };
+}
+
+function harness() {
+  const shared: { record: CompactionCancellationRecord | null } = { record: null };
+  // Scripted adapter outcomes exercise the core's response to committed, failed and
+  // superseded writes. This is not a simulation/proof of filesystem CAS or history repair.
+  const storage = {
+    read: mock(() => Promise.resolve(structuredClone(shared.record))),
+    mutate: mock(
+      (
+        mutation: CompactionCancellationMutation,
+        isCurrent: () => boolean,
+        onCommitted: (record: CompactionCancellationRecord | null) => undefined
+      ): Promise<CompactionCancellationMutationOutcome> => {
+        if (!isCurrent()) return Promise.resolve("superseded");
+        shared.record = mutation.kind === "retire" ? null : structuredClone(mutation.record);
+        onCommitted(shared.record);
+        return Promise.resolve("applied");
+      }
+    ),
+    repair: mock(
+      (
+        _isCurrent: () => boolean,
+        _onCommitted: () => void
+      ): Promise<CompactionCancellationRecord | null> => {
+        return Promise.reject(new Error("Unexpected repair"));
+      }
+    ),
+  } satisfies CompactionCancellationStorage;
+  return { shared, storage, state: new CompactionCancellation(storage) };
+}
+
+describe("inactive cancellation state core", () => {
+  it("retains a failed publication's nonce and advanced frontier through an exact retry", async () => {
+    const { state, storage, shared } = harness();
+    const apply = storage.mutate.getMockImplementation()!;
+    const frontier = { nonce: null, generation: "advanced-a" };
+    let publication: CompactionCancellationPublication | undefined;
+    storage.mutate.mockImplementationOnce((mutation) => {
+      assert(mutation.kind === "publish");
+      publication = mutation.publication;
+      publication.predecessor = frontier;
+      return Promise.reject(new Error("sidecar publication failed after advancement"));
+    });
+    await assert.rejects(state.cancel(), /sidecar publication failed/);
+    const first = await state.read();
+    expect(state.blocksRecovery).toBe(true);
+    expect(storage.read).not.toHaveBeenCalled();
+    await assert.rejects(state.narrow(first!.nonce, summary), /sidecar publication failed/);
+    storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => {
+      assert(mutation.kind === "publish");
+      expect(mutation.record.nonce).toBe(first!.nonce);
+      expect(mutation.publication).toBe(publication!);
+      expect(mutation.publication.predecessor).toBe(frontier);
+      expect(mutation.publication.attempts).toBe(2);
+      return apply(mutation, current, onCommitted);
+    });
+    expect(await state.retry()).toBe("applied");
+    expect(shared.record?.nonce).toBe(first!.nonce);
+    expect(state.needsPersistence).toBe(false);
+  });
+
+  it("refreshes a foreign successor after an adapter rejects the retry frontier", async () => {
+    const { state, storage, shared } = harness();
+    storage.mutate.mockRejectedValueOnce(new Error("failed Stop"));
+    await assert.rejects(state.cancel(), /failed Stop/);
+    shared.record = cancellation("foreign-b");
+    storage.mutate.mockResolvedValueOnce("superseded");
+    expect(await state.retry()).toBe("superseded");
+    expect(state.needsPersistence).toBe(false);
+    expect(await state.readForReplacement()).toEqual(shared.record);
+    expect(storage.mutate).toHaveBeenCalledTimes(2);
+  });
+
+  it("queued local Stops cannot publish or acknowledge an obsolete nonce", async () => {
+    const { state, storage } = harness();
+    const first = state.cancel();
+    const firstRecord = state.read();
+    const second = state.cancel();
+    const secondRecord = state.read();
+    expect((await firstRecord)?.nonce).not.toBe((await secondRecord)?.nonce);
+    expect(await first).toBe("superseded");
+    expect(await second).toBe("applied");
+    expect(await state.read()).toEqual(await secondRecord);
+    expect(storage.mutate).toHaveBeenCalledTimes(1);
+    expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({
+      kind: "publish",
+      record: await secondRecord,
+    });
+  });
+
+  it("a Stop admitted during an old write invalidates that write's final guard", async () => {
+    const { state, storage } = harness();
+    const entered = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    const apply = storage.mutate.getMockImplementation()!;
+    storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => {
+      entered.resolve();
+      await release.promise;
+      expect(current()).toBe(false);
+      return apply(mutation, current, onCommitted);
+    });
+    const first = state.cancel();
+    await entered.promise;
+    const second = state.cancel();
+    const expected = await state.read();
+    release.resolve();
+    expect(await first).toBe("superseded");
+    expect(await second).toBe("applied");
+    expect(await state.read()).toEqual(expected);
+  });
+
+  it("failed retirement retries deletion while maintaining conservative exclusion", async () => {
+    const { state, storage } = harness();
+    await state.cancel();
+    const record = (await state.read())!;
+    storage.mutate.mockRejectedValueOnce(new Error("unlink failed"));
+    await assert.rejects(state.retire(record.nonce), /unlink failed/);
+    expect(state.blocksRecovery).toBe(true);
+    expect(await state.read()).toEqual(record);
+    await assert.rejects(state.flush(), /unlink failed/);
+    expect(await state.retry()).toBe("applied");
+    expect(storage.mutate.mock.calls.at(-1)?.[0]).toEqual({ kind: "retire", nonce: record.nonce });
+    expect(await state.read()).toBeNull();
+  });
+
+  it("late acknowledgment of a committed Stop cannot settle its pending successor", async () => {
+    const { state, storage } = harness();
+    const committed = Promise.withResolvers<void>();
+    const acknowledge = Promise.withResolvers<void>();
+    const successorEntered = Promise.withResolvers<void>();
+    const releaseSuccessor = Promise.withResolvers<void>();
+    const apply = storage.mutate.getMockImplementation()!;
+    storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => {
+      const outcome = await apply(mutation, current, onCommitted);
+      committed.resolve();
+      await acknowledge.promise;
+      return outcome;
+    });
+    storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => {
+      successorEntered.resolve();
+      await releaseSuccessor.promise;
+      return apply(mutation, current, onCommitted);
+    });
+    const first = state.cancel();
+    await committed.promise;
+    const second = state.cancel();
+    const expected = await state.read();
+    acknowledge.resolve();
+    await successorEntered.promise;
+    expect(await first).toBe("applied");
+    expect(state.blocksRecovery).toBe(true);
+    expect(await state.read()).toEqual(expected);
+    releaseSuccessor.resolve();
+    expect(await second).toBe("applied");
+    expect(state.needsPersistence).toBe(false);
+  });
+
+  it("failed narrowing retains unresolved exclusion until the exact retry commits", async () => {
+    const { state, storage } = harness();
+    await state.cancel();
+    const record = (await state.read())!;
+    storage.mutate.mockRejectedValueOnce(new Error("narrowing failed"));
+    await assert.rejects(state.narrow(record.nonce, summary), /narrowing failed/);
+    expect(await state.read()).toEqual(record);
+    expect(state.blocksRecovery).toBe(true);
+    await state.retry();
+    expect((await state.read())?.scope).toEqual({ kind: "summary", ...summary });
+  });
+
+  it.each([
+    { witnessed: false, failed: false },
+    { witnessed: true, failed: false },
+    { witnessed: false, failed: true },
+    { witnessed: true, failed: true },
+  ])(
+    "narrowing cannot replace a concurrently queued retirement (witnessed=$witnessed, failed=$failed)",
+    async ({ witnessed, failed }) => {
+      const { state, storage, shared } = harness();
+      await state.cancel();
+      const record = (await state.read())!;
+      if (failed) storage.mutate.mockRejectedValueOnce(new Error("retirement failed"));
+
+      // narrow yields at its publication join. Retirement must keep ownership when
+      // that continuation resumes, on both sides of the adapter's deletion result.
+      const narrowing = state.narrow(record.nonce, summary);
+      const retirement = witnessed
+        ? state.retireReplacement({ nonce: record.nonce })
+        : state.retire(record.nonce);
+      const results = await Promise.allSettled([narrowing, retirement]);
+      expect(results[0]).toEqual({ status: "fulfilled", value: undefined });
+      expect(results[1].status).toBe(failed ? "rejected" : "fulfilled");
+      expect(storage.mutate.mock.calls.map(([mutation]) => mutation.kind)).toEqual([
+        "publish",
+        "retire",
+      ]);
+      expect(state.needsPersistence).toBe(failed);
+      expect(state.blocksRecovery).toBe(failed && !witnessed);
+      expect(shared.record).toEqual(failed ? record : null);
+
+      if (failed) {
+        expect(await state.retry()).toBe("applied");
+        expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({
+          kind: "retire",
+          nonce: record.nonce,
+        });
+      }
+      expect(shared.record).toBeNull();
+      expect(state.needsPersistence).toBe(false);
+    }
+  );
+
+  it("requires a matching replacement witness to retire retained cancellation", async () => {
+    const { state, storage } = harness();
+    await state.cancel({ retainUntilReplacement: true });
+    const record = (await state.read())!;
+    await state.narrow(record.nonce, summary);
+    await state.retire(record.nonce);
+    await state.retireReplacement({ nonce: "unrelated" });
+    expect(storage.mutate).toHaveBeenCalledTimes(1);
+    expect(await state.read()).toEqual(record);
+    expect(await state.retireReplacement({ nonce: record.nonce })).toBe("applied");
+    expect(await state.read()).toBeNull();
+  });
+
+  it("a subsequent Stop carries a retained full-clear obligation", async () => {
+    const { state } = harness();
+    await state.cancel({ retainUntilReplacement: true });
+    const first = (await state.read())!;
+    await state.cancel();
+    const second = (await state.read())!;
+    expect(second.nonce).not.toBe(first.nonce);
+    expect(second.retainUntilReplacement).toBe(true);
+    await state.retireReplacement({ nonce: first.nonce });
+    expect(await state.read()).toEqual(second);
+  });
+
+  it("preserves retention inherited by the adapter without first reading the predecessor", async () => {
+    const { state, storage, shared } = harness();
+    shared.record = { ...cancellation("retained-predecessor"), retainUntilReplacement: true };
+    const apply = storage.mutate.getMockImplementation()!;
+    storage.mutate.mockImplementationOnce((mutation, current, onCommitted) => {
+      assert(mutation.kind === "publish");
+      expect(mutation.record.retainUntilReplacement).toBeUndefined();
+      // Inheritance happens under the adapter lock, without mutating the submitted record.
+      return apply(
+        { ...mutation, record: { ...mutation.record, retainUntilReplacement: true } },
+        current,
+        onCommitted
+      );
+    });
+    await state.cancel();
+    const committed = structuredClone(shared.record);
+    assert(committed);
+    await state.narrow(committed.nonce, summary);
+    await state.retire(committed.nonce);
+    expect(shared.record).toEqual(committed);
+    expect(storage.mutate).toHaveBeenCalledTimes(1);
+    expect(storage.read).not.toHaveBeenCalled();
+    await state.cancel();
+    expect(shared.record?.retainUntilReplacement).toBe(true);
+  });
+
+  it.each(["record", "malformed", "I/O"])(
+    "a read spanning committed witnessed retirement cannot restore its retention or error (%s)",
+    async (outcome) => {
+      const { state, storage, shared } = harness();
+      await state.cancel({ retainUntilReplacement: true });
+      const record = structuredClone(shared.record);
+      assert(record);
+      const snapshot = Promise.withResolvers<CompactionCancellationRecord | null>();
+      storage.read.mockReturnValueOnce(snapshot.promise);
+      const retirement = state.retireReplacement({ nonce: record.nonce });
+      const reading = state.read();
+      await retirement;
+      expect(shared.record).toBeNull();
+      if (outcome === "record") snapshot.resolve(record);
+      else
+        snapshot.reject(
+          outcome === "malformed"
+            ? new MalformedCompactionCancellationError("old bytes")
+            : new Error("old I/O failure")
+        );
+      expect(await reading).toBeNull();
+      expect(storage.repair).not.toHaveBeenCalled();
+      await state.cancel();
+      expect(shared.record?.retainUntilReplacement).toBeUndefined();
+    }
+  );
+
+  it.each([
+    { outcome: "failed", completion: "before" },
+    { outcome: "failed", completion: "after" },
+    { outcome: "superseded", completion: "before" },
+    { outcome: "superseded", completion: "after" },
+  ])(
+    "uncommitted retirement preserves a foreign read ($outcome, completion=$completion)",
+    async ({ outcome, completion }) => {
+      const { state, storage, shared } = harness();
+      await state.cancel({ retainUntilReplacement: true });
+      const record = shared.record;
+      assert(record);
+      const snapshot = Promise.withResolvers<CompactionCancellationRecord | null>();
+      storage.read.mockReturnValueOnce(snapshot.promise);
+      const acknowledge = Promise.withResolvers<CompactionCancellationMutationOutcome>();
+      storage.mutate.mockReturnValueOnce(acknowledge.promise);
+      const retirement = state.retireReplacement({ nonce: record.nonce });
+      const reading = state.read();
+      const foreign = { ...cancellation("foreign-b"), retainUntilReplacement: true };
+      if (completion === "before") {
+        snapshot.resolve(foreign);
+        expect(await reading).toEqual(foreign);
+      }
+      if (outcome === "failed") acknowledge.reject(new Error("unlink failed"));
+      else acknowledge.resolve("superseded");
+      if (outcome === "failed") await assert.rejects(retirement, /unlink failed/);
+      else await retirement;
+      if (completion === "after") {
+        snapshot.resolve(foreign);
+        expect(await reading).toEqual(foreign);
+      }
+      await state.cancel();
+      expect(shared.record?.retainUntilReplacement).toBe(true);
+    }
+  );
+
+  it.each([
+    { completion: "before", failedCleanup: false },
+    { completion: "after", failedCleanup: false },
+    { completion: "before", failedCleanup: true },
+    { completion: "after", failedCleanup: true },
+  ])(
+    "preserves a post-deletion foreign read (completion=$completion, failed cleanup=$failedCleanup)",
+    async ({ completion, failedCleanup }) => {
+      const { state, storage, shared } = harness();
+      await state.cancel({ retainUntilReplacement: true });
+      const record = shared.record;
+      assert(record);
+      const deleted = Promise.withResolvers<void>();
+      const acknowledge = Promise.withResolvers<void>();
+      storage.mutate.mockImplementationOnce(async (_mutation, _current, onCommitted) => {
+        shared.record = null;
+        onCommitted(null);
+        deleted.resolve();
+        await acknowledge.promise;
+        if (failedCleanup) throw new Error("cleanup failed after deletion");
+        return "applied";
+      });
+      const retirement = state.retireReplacement({ nonce: record.nonce });
+      await deleted.promise;
+      const foreign = { ...cancellation("foreign-b"), retainUntilReplacement: true };
+      shared.record = foreign;
+      const snapshot = Promise.withResolvers<CompactionCancellationRecord | null>();
+      storage.read.mockReturnValueOnce(snapshot.promise);
+      const reading = state.read();
+      if (completion === "before") {
+        snapshot.resolve(foreign);
+        expect(await reading).toEqual(foreign);
+      }
+      acknowledge.resolve();
+      if (failedCleanup) await assert.rejects(retirement, /cleanup failed/);
+      else await retirement;
+      expect(state.needsPersistence).toBe(failedCleanup);
+      if (completion === "after") {
+        snapshot.resolve(foreign);
+        expect(await reading).toEqual(foreign);
+      }
+      // No refresh is needed for the next Stop to carry B's full-clear obligation.
+      await state.cancel();
+      expect(shared.record?.retainUntilReplacement).toBe(true);
+    }
+  );
+
+  it("witnessed deletion debt allows fresh reads without adopting a foreign Stop for retry", async () => {
+    const { state, storage, shared } = harness();
+    await state.cancel({ retainUntilReplacement: true });
+    const record = (await state.read())!;
+    storage.mutate.mockRejectedValueOnce(new Error("witnessed unlink failed"));
+    await assert.rejects(state.retireReplacement({ nonce: record.nonce }), /unlink failed/);
+    expect(state.needsPersistence).toBe(true);
+    expect(state.blocksRecovery).toBe(false);
+    await state.flush();
+    expect(await state.readForReplacement()).toBeNull();
+    shared.record = cancellation("foreign-b");
+    expect(await state.readForReplacement()).toEqual(shared.record);
+    expect(state.needsPersistence).toBe(true);
+    storage.mutate.mockResolvedValueOnce("superseded");
+    await state.retry();
+    expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({
+      kind: "retire",
+      nonce: record.nonce,
+    });
+    expect(await state.read()).toEqual(shared.record);
+    expect(state.needsPersistence).toBe(false);
+  });
+
+  it.each([false, true])(
+    "foreign narrowing is independent of witnessed deletion debt (narrow failure=%s)",
+    async (failed) => {
+      const { state, storage, shared } = harness();
+      await state.cancel();
+      const first = (await state.read())!;
+      storage.mutate.mockRejectedValueOnce(new Error("old unlink failed"));
+      await assert.rejects(state.retireReplacement({ nonce: first.nonce }), /old unlink failed/);
+      shared.record = cancellation("foreign-b");
+      expect(await state.readForReplacement()).toEqual(shared.record);
+
+      if (failed) storage.mutate.mockRejectedValueOnce(new Error("new narrowing failed"));
+      const narrowing = state.narrow("foreign-b", summary);
+      if (failed) {
+        await assert.rejects(narrowing, /new narrowing failed/);
+        expect(state.blocksRecovery).toBe(true);
+        expect(await state.read()).toEqual(cancellation("foreign-b"));
+        expect(await state.retry()).toBe("applied");
+      } else expect(await narrowing).toBe("applied");
+
+      expect(shared.record).toEqual({
+        ...cancellation("foreign-b"),
+        scope: { kind: "summary", ...summary },
+      });
+      expect(storage.mutate.mock.calls.slice(2).map(([mutation]) => mutation.kind)).toEqual(
+        failed ? ["narrow", "narrow"] : ["narrow"]
+      );
+      expect(state.needsPersistence).toBe(false);
+    }
+  );
+
+  it("a stale witness cannot make a newer failed Stop non-blocking", async () => {
+    const { state, storage } = harness();
+    await state.cancel();
+    const first = (await state.read())!;
+    await state.retireReplacement({ nonce: first.nonce });
+    storage.mutate.mockRejectedValueOnce(new Error("new Stop failed"));
+    await assert.rejects(state.cancel(), /new Stop failed/);
+    const second = await state.read();
+    await state.retireReplacement({ nonce: first.nonce });
+    expect(state.blocksRecovery).toBe(true);
+    expect(await state.read()).toEqual(second);
+    await assert.rejects(state.flush(), /new Stop failed/);
+  });
+
+  it("ordinary cleanup cannot downgrade witnessed deletion debt", async () => {
+    const { state, storage } = harness();
+    await state.cancel();
+    const record = (await state.read())!;
+    storage.mutate.mockRejectedValue(new Error("unlink still failed"));
+    await assert.rejects(state.retireReplacement({ nonce: record.nonce }), /unlink/);
+    await assert.rejects(state.retire(record.nonce), /unlink/);
+    expect(state.needsPersistence).toBe(true);
+    expect(state.blocksRecovery).toBe(false);
+    await state.flush();
+    expect(await state.readForReplacement()).toBeNull();
+  });
+
+  it("readers joining an active failed publication report it without retrying", async () => {
+    const { state, storage } = harness();
+    const entered = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    storage.mutate.mockImplementationOnce(async () => {
+      entered.resolve();
+      await release.promise;
+      throw new Error("active publication failed");
+    });
+    const stopping = state.cancel();
+    await entered.promise;
+    const readers = [state.readForReplacement(), state.readForReplacement()];
+    release.resolve();
+    const results = await Promise.allSettled([stopping, ...readers]);
+    expect(results.map((result) => result.status)).toEqual(["rejected", "rejected", "rejected"]);
+    expect(storage.mutate).toHaveBeenCalledTimes(1);
+    expect(state.blocksRecovery).toBe(true);
+  });
+
+  it.each([false, true])(
+    "replacement readers share a retry and its outcome (failure=%s)",
+    async (failed) => {
+      const { state, storage } = harness();
+      storage.mutate.mockRejectedValueOnce(new Error("initial failure"));
+      await assert.rejects(state.cancel(), /initial failure/);
+      const entered = Promise.withResolvers<void>();
+      const release = Promise.withResolvers<void>();
+      const apply = storage.mutate.getMockImplementation()!;
+      storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => {
+        entered.resolve();
+        await release.promise;
+        if (failed) throw new Error("retry failure");
+        return apply(mutation, current, onCommitted);
+      });
+      const readers = [state.readForReplacement(), state.readForReplacement()];
+      await entered.promise;
+      const retryA = state.retry();
+      expect(state.retry()).toBe(retryA);
+      release.resolve();
+      const results = await Promise.allSettled(readers);
+      expect(results.map((result) => result.status)).toEqual(
+        failed ? ["rejected", "rejected"] : ["fulfilled", "fulfilled"]
+      );
+      expect(storage.mutate).toHaveBeenCalledTimes(2);
+      expect(state.needsPersistence).toBe(failed);
+      if (!failed) expect(await readers[0]).toEqual(await readers[1]);
+    }
+  );
+
+  it.each(["read", "replacement"] as const)(
+    "a pending newer read cannot hide an earlier durable cancellation (%s)",
+    async (reader) => {
+      const { state, storage } = harness();
+      const older = cancellation("older-a");
+      const newer = cancellation("newer-b");
+      const firstRead = Promise.withResolvers<CompactionCancellationRecord | null>();
+      const secondRead = Promise.withResolvers<CompactionCancellationRecord | null>();
+      storage.read.mockReturnValueOnce(firstRead.promise).mockReturnValueOnce(secondRead.promise);
+      const earlier = reader === "read" ? state.read() : state.readForReplacement();
+      const later = state.read();
+      firstRead.resolve(older);
+      try {
+        expect(await earlier).toEqual(older);
+        expect(storage.mutate).not.toHaveBeenCalled();
+      } finally {
+        secondRead.resolve(newer);
+        expect(await later).toEqual(newer);
+      }
+    }
+  );
+
+  it.each([false, true])(
+    "a pending newer read does not suppress repair or its failure (repair failure=%s)",
+    async (failed) => {
+      const { state, storage } = harness();
+      const entered = Promise.withResolvers<void>();
+      const release = Promise.withResolvers<void>();
+      const secondRead = Promise.withResolvers<CompactionCancellationRecord | null>();
+      storage.read
+        .mockRejectedValueOnce(new MalformedCompactionCancellationError())
+        .mockReturnValueOnce(secondRead.promise);
+      storage.repair.mockImplementationOnce(async (current, committed) => {
+        entered.resolve();
+        await release.promise;
+        if (failed) throw new Error("repair failed");
+        if (current()) committed();
+        return null;
+      });
+      const earlier = state.read();
+      await entered.promise;
+      const later = state.read();
+      release.resolve();
+      try {
+        if (failed) await assert.rejects(earlier, /repair failed/);
+        else expect(await earlier).toBeNull();
+        expect(state.repairRevision).toBe(failed ? 0 : 1);
+      } finally {
+        secondRead.resolve(null);
+        await later;
+      }
+    }
+  );
+
+  it.each(["old record", "absence", "malformed", "I/O"] as const)(
+    "a late read cannot replace a newer accepted read (%s)",
+    async (result) => {
+      const { state, storage, shared } = harness();
+      const entered = Promise.withResolvers<void>();
+      const release = Promise.withResolvers<void>();
+      storage.read.mockImplementationOnce(async () => {
+        entered.resolve();
+        await release.promise;
+        if (result === "malformed") throw new MalformedCompactionCancellationError();
+        if (result === "I/O") throw new Error("obsolete read failure");
+        return result === "absence" ? null : cancellation("older-a");
+      });
+      const earlier = state.read();
+      await entered.promise;
+      shared.record = cancellation("newer-b");
+      expect(await state.read()).toEqual(shared.record);
+      release.resolve();
+      expect(await earlier).toEqual(shared.record);
+      expect(storage.repair).not.toHaveBeenCalled();
+      expect(storage.mutate).not.toHaveBeenCalled();
+    }
+  );
+
+  it.each([false, true])(
+    "a newer read invalidates an in-flight repair (repair failure=%s)",
+    async (failed) => {
+      const { state, storage, shared } = harness();
+      const entered = Promise.withResolvers<void>();
+      const release = Promise.withResolvers<void>();
+      storage.read.mockRejectedValueOnce(new MalformedCompactionCancellationError());
+      storage.repair.mockImplementationOnce(async (current, committed) => {
+        entered.resolve();
+        await release.promise;
+        if (failed) throw new Error("obsolete repair failure");
+        if (current()) {
+          shared.record = null;
+          committed();
+        }
+        return null;
+      });
+      const earlier = state.read();
+      await entered.promise;
+      const newer = cancellation("newer-b");
+      shared.record = newer;
+      expect(await state.read()).toEqual(newer);
+      release.resolve();
+      expect(await earlier).toEqual(newer);
+      expect(shared.record).toEqual(newer);
+      expect(state.repairRevision).toBe(0);
+    }
+  );
+
+  it.each([
+    { completion: "before", successor: "cancel" },
+    { completion: "after", successor: "cancel" },
+    { completion: "before", successor: "read" },
+    { completion: "after", successor: "read" },
+  ])(
+    "repair receipt fences retained reads before acknowledgment (read=$completion, successor=$successor)",
+    async ({ completion, successor }) => {
+      const { state, storage, shared } = harness();
+      const retained = { ...cancellation("older-a"), retainUntilReplacement: true };
+      const olderSnapshot = Promise.withResolvers<CompactionCancellationRecord | null>();
+      const entered = Promise.withResolvers<void>();
+      const commit = Promise.withResolvers<void>();
+      const committed = Promise.withResolvers<void>();
+      const acknowledge = Promise.withResolvers<void>();
+      storage.read
+        .mockReturnValueOnce(olderSnapshot.promise)
+        .mockRejectedValueOnce(new MalformedCompactionCancellationError());
+      storage.repair.mockImplementationOnce(async (current, onCommitted) => {
+        entered.resolve();
+        await commit.promise;
+        assert(current());
+        shared.record = null;
+        onCommitted();
+        committed.resolve();
+        await acknowledge.promise;
+        return null;
+      });
+      const older = state.read();
+      const repairing = state.read();
+      await entered.promise;
+      if (completion === "before") {
+        olderSnapshot.resolve(retained);
+        expect(await older).toEqual(retained);
+      }
+      commit.resolve();
+      await committed.promise;
+      expect(state.repairRevision).toBe(1);
+      try {
+        if (completion === "after") {
+          olderSnapshot.resolve(retained);
+          expect(await older).toBeNull();
+        }
+        if (successor === "cancel") {
+          await state.cancel();
+          expect(shared.record?.retainUntilReplacement).toBeUndefined();
+        } else {
+          shared.record = cancellation("newer-b");
+          expect(await state.read()).toEqual(shared.record);
+        }
+      } finally {
+        acknowledge.resolve();
+        await repairing;
+      }
+      expect(await repairing).toEqual(shared.record);
+    }
+  );
+
+  it("out-of-order reads preserve the original witnessed retirement retry", async () => {
+    const { state, storage, shared } = harness();
+    await state.cancel();
+    const first = (await state.read())!;
+    storage.mutate.mockRejectedValueOnce(new Error("unlink failed"));
+    await assert.rejects(state.retireReplacement({ nonce: first.nonce }), /unlink failed/);
+    const entered = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    storage.read.mockImplementationOnce(async () => {
+      entered.resolve();
+      await release.promise;
+      return first;
+    });
+    const earlier = state.read();
+    await entered.promise;
+    shared.record = cancellation("newer-b");
+    expect(await state.readForReplacement()).toEqual(shared.record);
+    release.resolve();
+    expect(await earlier).toEqual(shared.record);
+    expect(state.needsPersistence).toBe(true);
+    expect(state.blocksRecovery).toBe(false);
+    storage.mutate.mockResolvedValueOnce("superseded");
+    await state.retry();
+    expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({
+      kind: "retire",
+      nonce: first.nonce,
+    });
+    expect(shared.record).toEqual(cancellation("newer-b"));
+    expect(state.needsPersistence).toBe(false);
+  });
+
+  it.each(["old record", "malformed", "I/O"] as const)(
+    "ignores an obsolete read after Stop (%s)",
+    async (result) => {
+      const { state, storage } = harness();
+      const entered = Promise.withResolvers<void>();
+      const release = Promise.withResolvers<void>();
+      storage.read.mockImplementationOnce(async () => {
+        entered.resolve();
+        await release.promise;
+        if (result === "malformed") throw new MalformedCompactionCancellationError();
+        if (result === "I/O") throw new Error("read failed");
+        return cancellation("obsolete");
+      });
+      const reading = state.read();
+      await entered.promise;
+      await state.cancel();
+      const expected = await state.read();
+      release.resolve();
+      expect(await reading).toEqual(expected);
+      expect(storage.repair).not.toHaveBeenCalled();
+    }
+  );
+
+  it.each([false, true])(
+    "repairs malformed state with guarded commit evidence (superseded=%s)",
+    async (superseded) => {
+      const { state, storage, shared } = harness();
+      storage.read.mockRejectedValueOnce(new MalformedCompactionCancellationError());
+      const entered = Promise.withResolvers<void>();
+      const release = Promise.withResolvers<void>();
+      storage.repair.mockImplementationOnce(async (current, committed) => {
+        entered.resolve();
+        await release.promise;
+        if (!current()) return null;
+        committed();
+        shared.record = null;
+        return null;
+      });
+      const reading = state.read();
+      await entered.promise;
+      if (superseded) await state.cancel();
+      release.resolve();
+      expect(await reading).toEqual(shared.record);
+      expect(state.repairRevision).toBe(superseded ? 0 : 1);
+    }
+  );
+
+  it("accepts a newer valid record returned by repair without claiming a history rewrite", async () => {
+    const { state, storage } = harness();
+    storage.read.mockRejectedValueOnce(new MalformedCompactionCancellationError());
+    storage.repair.mockResolvedValueOnce(cancellation("newer-valid"));
+    expect(await state.read()).toEqual(cancellation("newer-valid"));
+    expect(state.repairRevision).toBe(0);
+  });
+
+  it("replacement propagates a read refusal without repair or fallback publication", async () => {
+    const { state, storage } = harness();
+    const refusal = new CompactionCancellationReadRefusedError("Read refused");
+    storage.read.mockRejectedValueOnce(refusal);
+    await assert.rejects(state.readForReplacement(), (error) => error === refusal);
+    expect(storage.repair).not.toHaveBeenCalled();
+    expect(storage.mutate).not.toHaveBeenCalled();
+    expect(state.needsPersistence).toBe(false);
+  });
+
+  it("ordinary read failure never repairs; explicit replacement publishes a retained fence", async () => {
+    const { state, storage } = harness();
+    storage.read.mockRejectedValueOnce(new Error("permission denied"));
+    await assert.rejects(state.read(), /permission denied/);
+    expect(storage.repair).not.toHaveBeenCalled();
+    expect(storage.mutate).not.toHaveBeenCalled();
+    storage.read.mockRejectedValueOnce(new Error("permission denied"));
+    expect(await state.readForReplacement()).toMatchObject({ retainUntilReplacement: true });
+    expect(storage.repair).not.toHaveBeenCalled();
+    expect(storage.mutate).toHaveBeenCalledTimes(1);
+  });
+
+  it.each([
+    { completion: "before", freshFailure: false },
+    { completion: "after", freshFailure: false },
+    { completion: "before", freshFailure: true },
+    { completion: "after", freshFailure: true },
+  ])(
+    "replacement refreshes unknown state (error=$completion supersession, fresh failure=$freshFailure)",
+    async ({ completion, freshFailure }) => {
+      const { state, storage, shared } = harness();
+      shared.record = { ...cancellation("foreign-b"), retainUntilReplacement: true };
+      const failure = Promise.withResolvers<CompactionCancellationRecord | null>();
+      const superseded = Promise.withResolvers<CompactionCancellationMutationOutcome>();
+      const readFinished = Promise.withResolvers<void>();
+      const deliver = Promise.withResolvers<void>();
+      storage.read.mockReturnValueOnce(failure.promise);
+      if (freshFailure) storage.read.mockRejectedValueOnce(new Error("fresh read unavailable"));
+      storage.mutate.mockReturnValueOnce(superseded.promise);
+      const read = state.read.bind(state);
+      const checkedRead = spyOn(state, "read").mockImplementationOnce(async () => {
+        try {
+          const record = await read();
+          readFinished.resolve();
+          if (completion === "before") await deliver.promise;
+          return record;
+        } finally {
+          readFinished.resolve();
+        }
+      });
+      try {
+        const replacement = state.readForReplacement();
+        const stopping = state.cancel();
+        if (completion === "before") {
+          failure.reject(new Error("old read failed"));
+          await readFinished.promise;
+        }
+        superseded.resolve("superseded");
+        expect(await stopping).toBe("superseded");
+        deliver.resolve();
+        if (completion === "after") failure.reject(new Error("old read failed"));
+        if (freshFailure) await assert.rejects(replacement, /fresh read unavailable/);
+        else expect(await replacement).toEqual(shared.record);
+        expect(storage.read).toHaveBeenCalledTimes(2);
+        expect(storage.mutate).toHaveBeenCalledTimes(1);
+      } finally {
+        checkedRead.mockRestore();
+      }
+    }
+  );
+
+  it.each([
+    { overlap: false, readFailure: true },
+    { overlap: true, readFailure: true },
+    { overlap: true, readFailure: false },
+  ])(
+    "replacement bounds its refresh (another overlap=$overlap, read failure=$readFailure)",
+    async ({ overlap, readFailure }) => {
+      const { state, storage, shared } = harness();
+      shared.record = cancellation("foreign-b");
+      const failure = Promise.withResolvers<CompactionCancellationRecord | null>();
+      const refresh = Promise.withResolvers<CompactionCancellationRecord | null>();
+      const refreshEntered = Promise.withResolvers<void>();
+      storage.read.mockReturnValueOnce(failure.promise).mockImplementationOnce(() => {
+        refreshEntered.resolve();
+        return refresh.promise;
+      });
+      storage.mutate.mockResolvedValue("superseded");
+      const replacement = state.readForReplacement();
+      await state.cancel();
+      failure.reject(new Error("old read failed"));
+      // The race allows no absence decision before the fresh authoritative read starts.
+      await Promise.race([refreshEntered.promise, replacement]);
+      expect(storage.read).toHaveBeenCalledTimes(2);
+      if (overlap) await state.cancel();
+      if (readFailure) {
+        refresh.reject(new Error("fresh read unavailable"));
+        await assert.rejects(replacement, /fresh read unavailable/);
+      } else {
+        refresh.resolve(null);
+        await assert.rejects(replacement);
+      }
+      expect(storage.read).toHaveBeenCalledTimes(2);
+      expect(storage.mutate).toHaveBeenCalledTimes(overlap ? 2 : 1);
+      expect(shared.record).toEqual(cancellation("foreign-b"));
+    }
+  );
+
+  it("joining a superseded Stop surfaces a failed authoritative read without fallback", async () => {
+    const { state, storage, shared } = harness();
+    shared.record = cancellation("foreign-b");
+    storage.read.mockRejectedValue(new Error("fresh read unavailable"));
+    storage.mutate.mockResolvedValueOnce("superseded");
+    storage.mutate.mockRejectedValue(new Error("unexpected fallback"));
+    const stopping = state.cancel();
+    const replacement = state.readForReplacement();
+    expect(await stopping).toBe("superseded");
+    await assert.rejects(replacement, /fresh read unavailable/);
+    expect(storage.read).toHaveBeenCalledTimes(1);
+    expect(storage.mutate).toHaveBeenCalledTimes(1);
+  });
+
+  it.each([false, true])(
+    "a bounded refresh refuses another Stop overlapping its read (publication settled=%s)",
+    async (settled) => {
+      const { state, storage, shared } = harness();
+      shared.record = cancellation("foreign-a");
+      const failedRead = Promise.withResolvers<CompactionCancellationRecord | null>();
+      const refresh = Promise.withResolvers<CompactionCancellationRecord | null>();
+      const refreshEntered = Promise.withResolvers<void>();
+      const publish = Promise.withResolvers<void>();
+      storage.read.mockReturnValueOnce(failedRead.promise).mockImplementationOnce(() => {
+        refreshEntered.resolve();
+        return refresh.promise;
+      });
+      const apply = storage.mutate.getMockImplementation()!;
+      storage.mutate
+        .mockResolvedValueOnce("superseded")
+        .mockImplementationOnce(async (mutation, current, onCommitted) => {
+          await publish.promise;
+          return apply(mutation, current, onCommitted);
+        });
+      const replacement = state.readForReplacement();
+      await state.cancel();
+      failedRead.reject(new Error("old read unavailable"));
+      await refreshEntered.promise;
+      const stopping = state.cancel();
+      if (settled) {
+        publish.resolve();
+        expect(await stopping).toBe("applied");
+      }
+      refresh.resolve(shared.record);
+      await assert.rejects(replacement);
+      expect(state.blocksRecovery).toBe(!settled);
+      expect(state.needsPersistence).toBe(!settled);
+      expect(storage.read).toHaveBeenCalledTimes(2);
+      publish.resolve();
+      expect(await stopping).toBe("applied");
+      expect(await state.readForReplacement()).toEqual(shared.record);
+      expect(state.blocksRecovery).toBe(false);
+    }
+  );
+
+  it.each(["record", "absence"])(
+    "replacement returns the committed Stop admitted after a successful read (%s)",
+    async (snapshot) => {
+      const { state, storage, shared } = harness();
+      shared.record = snapshot === "record" ? cancellation("older-a") : null;
+      const read = state.read.bind(state);
+      const checkedRead = spyOn(state, "read").mockImplementationOnce(() =>
+        read().then(async (record) => {
+          await state.cancel();
+          return record;
+        })
+      );
+      try {
+        expect(await state.readForReplacement()).toEqual(shared.record);
+        expect(shared.record).not.toBeNull();
+        expect(state.blocksRecovery).toBe(false);
+        expect(storage.read).toHaveBeenCalledTimes(1);
+      } finally {
+        checkedRead.mockRestore();
+      }
+    }
+  );
+
+  it("fallback cannot replace a Stop admitted after the read's final error check", async () => {
+    const { state, storage } = harness();
+    storage.read.mockRejectedValueOnce(new Error("read unavailable"));
+    const read = state.read.bind(state);
+    let stopping: ReturnType<CompactionCancellation["cancel"]> | undefined;
+    let newerRecord: ReturnType<CompactionCancellation["read"]> | undefined;
+    // Enter the promise boundary after read() has checked ownership and rejected,
+    // before readForReplacement() receives that rejection and considers fallback.
+    const checkedRead = spyOn(state, "read").mockImplementationOnce(() =>
+      read().catch((error: unknown) => {
+        stopping = state.cancel();
+        newerRecord = read();
+        throw error;
+      })
+    );
+    try {
+      const replacement = await state.readForReplacement();
+      expect(await stopping).toBe("applied");
+      assert(newerRecord);
+      expect(replacement).toEqual(await newerRecord);
+      expect(storage.mutate).toHaveBeenCalledTimes(1);
+    } finally {
+      checkedRead.mockRestore();
+    }
+  });
+
+  it("fallback cannot replace a same-mutation retry started after the checked read fails", async () => {
+    const { state, storage, shared } = harness();
+    await state.cancel();
+    const first = (await state.read())!;
+    storage.mutate.mockRejectedValueOnce(new Error("unlink failed"));
+    await assert.rejects(state.retireReplacement({ nonce: first.nonce }), /unlink failed/);
+    storage.read.mockRejectedValueOnce(new Error("read unavailable"));
+    const read = state.read.bind(state);
+    let retry: ReturnType<CompactionCancellation["retry"]> | undefined;
+    const checkedRead = spyOn(state, "read").mockImplementationOnce(() =>
+      read().catch((error: unknown) => {
+        retry = state.retry();
+        throw error;
+      })
+    );
+    try {
+      expect(await state.readForReplacement()).toBeNull();
+      expect(await retry).toBe("applied");
+      expect(shared.record).toBeNull();
+      expect(storage.mutate.mock.calls.map(([mutation]) => mutation.kind)).toEqual([
+        "publish",
+        "retire",
+        "retire",
+      ]);
+    } finally {
+      checkedRead.mockRestore();
+    }
+  });
+
+  it("fallback cannot replace a newer read accepted after the checked read fails", async () => {
+    const { state, storage, shared } = harness();
+    storage.read.mockRejectedValueOnce(new Error("read unavailable"));
+    const newer = cancellation("newer-b");
+    const read = state.read.bind(state);
+    const checkedRead = spyOn(state, "read").mockImplementationOnce(() =>
+      read().catch(async (error: unknown) => {
+        shared.record = newer;
+        expect(await read()).toEqual(newer);
+        throw error;
+      })
+    );
+    try {
+      expect(await state.readForReplacement()).toEqual(newer);
+      expect(storage.mutate).not.toHaveBeenCalled();
+    } finally {
+      checkedRead.mockRestore();
+    }
+  });
+
+  it("failed explicit fence publication stays blocking and is reported", async () => {
+    const { state, storage } = harness();
+    storage.read.mockRejectedValueOnce(new Error("read failed"));
+    storage.mutate.mockRejectedValueOnce(new Error("write failed"));
+    await assert.rejects(state.readForReplacement(), /write failed/);
+    expect(state.blocksRecovery).toBe(true);
+    expect(await state.read()).toMatchObject({ retainUntilReplacement: true });
+  });
+
+  it("narrowing snapshots inputs before awaits and read results cannot mutate state", async () => {
+    const { state } = harness();
+    await state.cancel();
+    const record = (await state.read())!;
+    const input = structuredClone(summary);
+    const narrowing = state.narrow(record.nonce, input);
+    input.pendingFollowUp.text = "changed";
+    await narrowing;
+    const narrowed = (await state.read())!;
+    expect(matchesCompactionCancellation(narrowed, summary)).toBe(true);
+    assert(narrowed.scope.kind === "summary");
+    narrowed.scope.pendingFollowUp.text = "also changed";
+    expect(matchesCompactionCancellation((await state.read())!, summary)).toBe(true);
+  });
+
+  it.each(["id", "sequence", "request"] as const)(
+    "summary cancellation matches exact identity (%s changes)",
+    (field) => {
+      const record: CompactionCancellationRecord = {
+        ...cancellation("narrowed"),
+        scope: { kind: "summary", ...summary },
+      };
+      const changed = structuredClone(summary);
+      if (field === "id") changed.id = "other";
+      if (field === "sequence") changed.sequence = 4;
+      if (field === "request") changed.pendingFollowUp.options = { model: "other" };
+      expect(matchesCompactionCancellation(record, changed)).toBe(false);
+      expect(matchesCompactionCancellation(cancellation("unresolved"), changed)).toBe(true);
+    }
+  );
+
+  it("flush follows the latest Stop through an obsolete publication failure", async () => {
+    const { state, storage } = harness();
+    const entered = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    storage.mutate.mockImplementationOnce(async () => {
+      entered.resolve();
+      await release.promise;
+      throw new Error("obsolete failure");
+    });
+    const first = state.cancel();
+    await entered.promise;
+    const flushing = state.flush();
+    const second = state.cancel();
+    release.resolve();
+    await assert.rejects(first, /obsolete failure/);
+    await flushing;
+    expect(await second).toBe("applied");
+    expect(state.needsPersistence).toBe(false);
+  });
+});
diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts
new file mode 100644
index 0000000000..9e906f20a8
--- /dev/null
+++ b/src/node/services/compactionCancellation.ts
@@ -0,0 +1,353 @@
+import { randomUUID } from "node:crypto";
+import { isDeepStrictEqual } from "node:util";
+
+export interface CompactionCancellationSummary {
+  id: string;
+  sequence?: number;
+  pendingFollowUp: Record<string, unknown>;
+}
+
+export interface CompactionCancellationRecord {
+  version: 1;
+  nonce: string;
+  retainUntilReplacement?: boolean;
+  scope: { kind: "unresolved" } | ({ kind: "summary" } & CompactionCancellationSummary);
+}
+
+export interface CompactionCancellationPublication {
+  attempts: number;
+  // The adapter records each admitted/advanced frontier BEFORE any subsequent failing await.
+  // Retries reuse this object; an unobserved attempt must never adopt a foreign frontier.
+  predecessor?: { nonce: string | null | undefined; generation: string | undefined };
+}
+
+/** Issued only after an exact replacement row is durably committed or found in history. */
+export interface CompactionCancellationReplacementWitness {
+  readonly nonce: string;
+}
+
+export type CompactionCancellationMutation =
+  | {
+      kind: "publish";
+      record: CompactionCancellationRecord;
+      publication: CompactionCancellationPublication;
+    }
+  | { kind: "narrow"; record: CompactionCancellationRecord }
+  | {
+      kind: "retire";
+      nonce: string;
+      replacementWitness?: CompactionCancellationReplacementWitness;
+    };
+
+export type CompactionCancellationMutationOutcome = "applied" | "superseded";
+
+/** Only successfully read bytes with invalid JSON/schema authorize automatic repair. */
+export class MalformedCompactionCancellationError extends Error {}
+
+/** Unsupported or oversized records must be preserved instead of repaired or overwritten. */
+export class CompactionCancellationReadRefusedError extends Error {}
+
+export interface CompactionCancellationStorage {
+  /** Fresh shared state; absence and unreadable I/O must remain distinguishable. */
+  read(): Promise<CompactionCancellationRecord | null>;
+  /**
+   * Atomically compare nonce/frontier and mutate under the shared history lock, checking
+   * isCurrent immediately before publication. Preserve inherited retention, including an
+   * unreadable predecessor. Retirement requires the exact nonce and, for retained records,
+   * a verified replacement witness. Superseded means no authority to apply this mutation.
+   * Call onCommitted synchronously at the durable commit/confirmation, before releasing
+   * the lock or awaiting cleanup, with inherited retention or null after retirement.
+   * Every applied outcome requires this receipt; later failure cannot undo the commit.
+   */
+  mutate(
+    mutation: CompactionCancellationMutation,
+    isCurrent: () => boolean,
+    onCommitted: (record: CompactionCancellationRecord | null) => undefined
+  ): Promise<CompactionCancellationMutationOutcome>;
+  /**
+   * Re-read under the lock; preserve newer valid records. Neutralize obsolete recovery
+   * before removing malformed bytes, preserve privacy floors, and call onCommitted
+   * synchronously when repair commits. Never repair an ordinary read/I/O failure.
+   */
+  repair(
+    isCurrent: () => boolean,
+    onCommitted: () => void
+  ): Promise<CompactionCancellationRecord | null>;
+}
+
+/**
+ * Inactive cancellation state core. Stop's retry identity outlives failed turn preparation;
+ * a later delivery must supply the storage adapter and activate every recovery consumer.
+ * Injected-adapter tests establish state invariants, not filesystem or cross-process CAS.
+ */
+export class CompactionCancellation {
+  private current?: CompactionCancellationRecord | null;
+  private replacementNonce?: string;
+  private mutation?: CompactionCancellationMutation;
+  private pending: Promise<CompactionCancellationMutationOutcome | undefined> =
+    Promise.resolve(undefined);
+  private unsettled = false;
+  private inFlight = false;
+  private readGeneration = 0;
+  private acceptedReadGeneration = 0;
+  private repairedHistoryRevision = 0;
+
+  constructor(private readonly storage: CompactionCancellationStorage) {}
+
+  get needsPersistence(): boolean {
+    return this.unsettled;
+  }
+
+  get blocksRecovery(): boolean {
+    return this.unsettled && !this.isWitnessedRetirement();
+  }
+
+  get repairRevision(): number {
+    return this.repairedHistoryRevision;
+  }
+
+  private isWitnessedRetirement(): boolean {
+    return (
+      this.mutation?.kind === "retire" &&
+      this.mutation.replacementWitness?.nonce === this.mutation.nonce
+    );
+  }
+
+  private effectiveRecord(): CompactionCancellationRecord | null {
+    // Callers cannot mutate a captured cancellation or its exact pending-request identity.
+    return structuredClone(
+      this.current?.nonce === this.replacementNonce ? null : (this.current ?? null)
+    );
+  }
+
+  async read(): Promise<CompactionCancellationRecord | null> {
+    if (this.blocksRecovery) return this.effectiveRecord();
+    const mutation = this.mutation;
+    const pending = this.pending;
+    // Only an accepted newer read or committed mutation displaces a snapshot/error/repair.
+    // A pending successor is not evidence of absence and cannot hide a valid Stop.
+    const generation = ++this.readGeneration;
+    const isCurrent = () =>
+      generation >= this.acceptedReadGeneration &&
+      mutation === this.mutation &&
+      pending === this.pending;
+    try {
+      const record = await this.storage.read().catch((error: unknown) => {
+        if (!(error instanceof MalformedCompactionCancellationError) || !isCurrent()) throw error;
+        return this.storage.repair(isCurrent, () => {
+          this.repairedHistoryRevision++;
+          if (!isCurrent()) return;
+          // Removal is already committed; pre-repair reads must not restore retention
+          // while the adapter is still finishing lock cleanup.
+          this.current = null;
+          this.acceptedReadGeneration = ++this.readGeneration;
+        });
+      });
+      if (isCurrent()) {
+        this.current = structuredClone(record);
+        this.acceptedReadGeneration = generation;
+      }
+    } catch (error) {
+      // A stale read/repair cannot hide a newer local Stop or trigger its replacement.
+      if (isCurrent() || this.current === undefined) throw error;
+    }
+    // Supersession without a receipt leaves unknown state, never evidence of absence.
+    if (this.current === undefined) throw new Error("Cancellation state changed during read");
+    return this.effectiveRecord();
+  }
+
+  cancel(options?: {
+    retainUntilReplacement?: boolean;
+  }): Promise<CompactionCancellationMutationOutcome> {
+    this.current = {
+      version: 1,
+      nonce: randomUUID(),
+      scope: { kind: "unresolved" },
+      ...(options?.retainUntilReplacement || this.current?.retainUntilReplacement
+        ? { retainUntilReplacement: true }
+        : {}),
+    };
+    return this.persist({ kind: "publish", record: this.current, publication: { attempts: 0 } });
+  }
+
+  async readForReplacement(): Promise<CompactionCancellationRecord | null> {
+    for (;;) {
+      if (this.blocksRecovery) {
+        const pending = this.pending;
+        const retryFailed = !this.inFlight;
+        try {
+          await pending;
+        } catch (error) {
+          // Retry already-failed debt; readers joining an in-flight attempt share its
+          // outcome instead of turning one failure into a chain of additional retries.
+          if (pending !== this.pending) continue;
+          if (!retryFailed) throw error;
+          const retried = this.retry();
+          try {
+            await retried;
+          } catch (error) {
+            if (retried === this.pending) throw error;
+          }
+        }
+        if (this.current === undefined) return this.refreshForReplacement();
+        continue;
+      }
+      const mutation = this.mutation;
+      const pending = this.pending;
+      const reading = this.read();
+      const generation = this.readGeneration;
+      try {
+        await reading;
+        if (this.current === undefined) return this.refreshForReplacement();
+        // A Stop or newer read can commit after reading resolves but before we resume.
+        if (!this.blocksRecovery) return this.effectiveRecord();
+      } catch (error) {
+        if (error instanceof CompactionCancellationReadRefusedError) throw error;
+        // Refresh unknown state once; propagate that read's failure instead of repeatedly
+        // publishing fallback Stops that a foreign cancellation keeps superseding.
+        if (this.current === undefined && (this.mutation !== mutation || this.pending !== pending))
+          return this.refreshForReplacement();
+        // read() checks before rejecting, but a newer Stop/retry/read can enter before
+        // this rejection resumes. Fallback must still own that exact failed read.
+        if (
+          this.acceptedReadGeneration > generation ||
+          this.mutation !== mutation ||
+          this.pending !== pending
+        )
+          continue;
+        // Explicit intervention may replace unreadable state, but cannot lose an unknown
+        // full-clear obligation. Failed publication remains blocking and visible.
+        await this.cancel({ retainUntilReplacement: true });
+      }
+    }
+  }
+
+  private async refreshForReplacement(): Promise<CompactionCancellationRecord | null> {
+    const pending = this.pending;
+    await this.read();
+    // One refresh cannot turn another Stop's tentative state into replacement authority.
+    // Further overlap requires a new request rather than an unbounded refresh/retry loop.
+    if (this.pending !== pending || this.blocksRecovery || this.current === undefined)
+      throw new Error("Cancellation changed during replacement refresh");
+    return this.effectiveRecord();
+  }
+
+  async narrow(nonce: string, summary: CompactionCancellationSummary) {
+    const captured = structuredClone(summary);
+    const mutation = this.mutation;
+    const pending = this.pending;
+    try {
+      await pending;
+    } catch (error) {
+      if (this.mutation !== mutation || this.pending !== pending) return;
+      // An old witnessed unlink is ancillary once a fresh read discovers B.
+      // Its failure cannot block B's narrowing; B's own failed writes still do.
+      if (
+        mutation?.kind !== "retire" ||
+        mutation.replacementWitness?.nonce !== mutation.nonce ||
+        mutation.nonce === nonce
+      )
+        throw error;
+    }
+    // Retirement can claim the same nonce during this join. Narrowing must not
+    // supersede its deletion or discard witnessed cleanup debt on resumption.
+    if (
+      this.mutation !== mutation ||
+      this.pending !== pending ||
+      this.replacementNonce === nonce ||
+      this.current?.nonce !== nonce ||
+      this.current.scope.kind !== "unresolved" ||
+      this.current.retainUntilReplacement
+    )
+      return;
+    // Failed narrowing must retain the broader exclusion until persistence succeeds.
+    return this.persist({
+      kind: "narrow",
+      record: { ...this.current, scope: { kind: "summary", ...captured } },
+    });
+  }
+
+  retire(nonce: string) {
+    if (this.current?.nonce !== nonce) return Promise.resolve(undefined);
+    // A later cleanup request cannot downgrade already-witnessed deletion debt.
+    if (this.replacementNonce === nonce) return this.retireReplacement({ nonce });
+    if (this.current.retainUntilReplacement) return Promise.resolve(undefined);
+    return this.persist({ kind: "retire", nonce });
+  }
+
+  retireReplacement(witness: CompactionCancellationReplacementWitness) {
+    if (this.current?.nonce !== witness.nonce) return Promise.resolve(undefined);
+    this.replacementNonce = witness.nonce;
+    return this.persist({
+      kind: "retire",
+      nonce: witness.nonce,
+      replacementWitness: { ...witness },
+    });
+  }
+
+  retry(): Promise<CompactionCancellationMutationOutcome | undefined> {
+    return this.unsettled && !this.inFlight && this.mutation
+      ? this.persist(this.mutation)
+      : this.pending;
+  }
+
+  async flush(): Promise<void> {
+    for (;;) {
+      const pending = this.pending;
+      try {
+        await pending;
+      } catch (error) {
+        if (pending !== this.pending) continue;
+        if (!this.isWitnessedRetirement()) throw error;
+      }
+      if (pending === this.pending) return;
+    }
+  }
+
+  private persist(
+    mutation: CompactionCancellationMutation
+  ): Promise<CompactionCancellationMutationOutcome> {
+    this.mutation = mutation;
+    this.unsettled = true;
+    this.inFlight = true;
+    const generation = this.acceptedReadGeneration;
+    const isCurrent = () => this.mutation === mutation;
+    const result = this.pending
+      .catch(() => undefined)
+      .then(async (): Promise<CompactionCancellationMutationOutcome> => {
+        if (!isCurrent()) return "superseded";
+        if (mutation.kind === "publish") mutation.publication.attempts++;
+        const outcome = await this.storage.mutate(mutation, isCurrent, (record) => {
+          if (!isCurrent()) return;
+          this.current = structuredClone(record);
+          // Commit invalidates pre-deletion reads before lock release. A later foreign
+          // read must survive acknowledgment delayed by adapter cleanup.
+          this.acceptedReadGeneration = ++this.readGeneration;
+        });
+        if (isCurrent()) {
+          this.unsettled = false;
+          if (outcome === "superseded" && this.acceptedReadGeneration === generation)
+            this.current = undefined;
+        }
+        return outcome;
+      });
+    this.pending = result;
+    const settled = () => {
+      if (this.pending === result) this.inFlight = false;
+    };
+    result.then(settled, settled);
+    return result;
+  }
+}
+
+export function matchesCompactionCancellation(
+  record: CompactionCancellationRecord,
+  summary: CompactionCancellationSummary
+): boolean {
+  return (
+    record.scope.kind === "unresolved" ||
+    (record.scope.id === summary.id &&
+      record.scope.sequence === summary.sequence &&
+      isDeepStrictEqual(record.scope.pendingFollowUp, summary.pendingFollowUp))
+  );
+}