🤖 refactor: coordinate compaction and durable continuations

src/common/constants/compactionCancellation.ts

@@ -0,0 +1 @@
+export const COMPACTION_CANCELLATION_FILE = "compaction-cancellation.json";

src/common/orpc/schemas/continuousCompaction.ts

@@ -75,6 +75,8 @@ const attachment: z.ZodType<PostCompactionAttachment> = z.discriminatedUnion("ty
 export const ContinuousCompactionJournalSchema = z
   .object({
     version: z.literal(1),
+    // Legacy absence is valid only before the first durable repair generation.
+    publicationGeneration: z.string().min(1).optional(),
     boundary: row,
     staticCopies: z.array(row),
     liveTailCopySpec: z.object({

src/common/orpc/schemas/message.ts

@@ -161,6 +161,7 @@ export const MuxMessageSchema = z.object({
   metadata: z
     .object({
       historySequence: z.number().optional(),
+      compactionCancellationNonce: z.string().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

@@ -939,6 +939,8 @@ export interface ContextBudgetRejectedMessage {
 
 // Our custom metadata type
 export interface MuxMetadata {
+  /** Durable explicit replacement of a stopped compaction intent, safe across sidecar retirement crashes. */
+  compactionCancellationNonce?: 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/constants/continuousCompaction.ts

@@ -4,3 +4,6 @@ export const TAIL_MIN_TOKENS = 4_000;
 export const TAIL_MAX_TOKENS = 60_000;
 export const MIN_HEAD_TOKENS = 8_000;
 export const SUMMARIZER_INPUT_FRACTION = 0.7;
+
+export const CONTINUOUS_COMPACTION_JOURNAL_FILE = "continuous-compaction.json";
+export const CONTINUOUS_COMPACTION_GENERATION_FILE = "continuous-compaction-generation.json";

src/node/services/agentSession.admissionGates.test.ts

@@ -211,6 +211,7 @@ describe("AgentSession.sendMessage (admission gates)", () => {
     expect(result).toEqual({
       success: false,
       error: { type: "unknown", raw: CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE },
+      superseded: true,
     });
     // Accepted, then notified so delivered-state bookkeeping can revert
     // (terminal-attention outbox contract, r41) — and the stale snapshot

src/node/services/agentSession.compactionAcceptance.test.ts

@@ -0,0 +1,316 @@
+import { afterEach, expect, mock, spyOn, test } from "bun:test";
+import { createMuxMessage } from "@/common/types/message";
+import { Err } from "@/common/types/result";
+import type { TurnCoordinator } from "./turnCoordinator";
+import * as branchSummary from "./branchSummary";
+import { createAgentSessionHarness } from "./agentSession.testHarness";
+
+afterEach(() => mock.restore());
+
+test.each([
+  ["before", false],
+  ["after", false],
+  ["before", true],
+  ["after", true],
+] as const)(
+  "Stop %s handoff append commits clears only its own pending continuation (successor=%s)",
+  async (commitPoint, hasSuccessor) => {
+    const workspaceId = "stopped-appending-compaction";
+    const h = await createAgentSessionHarness({ workspaceId });
+    const options = { model: "openai:gpt-4o", agentId: "exec" };
+    const summary = createMuxMessage("summary", "assistant", "Earlier work", {
+      muxMetadata: {
+        type: "compaction-summary",
+        pendingFollowUp: { text: "Continue", ...options },
+      },
+    });
+    await h.historyService.appendToHistory(workspaceId, summary);
+    const internals = h.session as unknown as {
+      coordinator: TurnCoordinator;
+      dispatchPendingFollowUp(): Promise<boolean>;
+    };
+    const entered = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    const append = h.historyService.appendToHistory.bind(h.historyService);
+    spyOn(h.historyService, "appendToHistory").mockImplementation(async (...args) => {
+      const isHandoff =
+        args[1].role === "user" &&
+        args[1].parts.some((part) => part.type === "text" && part.text === "Continue");
+      if (!isHandoff) return append(...args);
+      if (commitPoint === "before") {
+        entered.resolve();
+        await release.promise;
+        return append(...args);
+      }
+      const result = await append(...args);
+      entered.resolve();
+      await release.promise;
+      return result;
+    });
+    const stream = spyOn(h.aiService, "streamMessage");
+    const pending = internals.dispatchPendingFollowUp();
+    let restarted: Awaited<ReturnType<typeof createAgentSessionHarness>> | undefined;
+    try {
+      await entered.promise;
+      await h.session.interruptStream({ abandonPartial: true });
+      let successor: ReturnType<TurnCoordinator["beginCompactionObservation"]>;
+      if (hasSuccessor) {
+        expect((await h.session.sendMessage("manual replacement", options)).success).toBe(true);
+        successor = internals.coordinator.beginCompactionObservation("continuous");
+        expect(successor).toBeDefined();
+        // A's late rollback must not retire B's compaction work or erase a
+        // replacement summary, even when its history row reuses the source ID.
+        await h.historyService.updateHistory(
+          workspaceId,
+          createMuxMessage("summary", "assistant", "Replacement boundary", {
+            ...summary.metadata,
+            muxMetadata: {
+              type: "compaction-summary",
+              pendingFollowUp: { text: "Replacement follow-up", ...options },
+            },
+          })
+        );
+      }
+      release.resolve();
+      expect(await pending).toBe(false);
+      expect(stream).toHaveBeenCalledTimes(hasSuccessor ? 1 : 0);
+      const history = await h.historyService.getHistoryFromLatestBoundary(workspaceId);
+      if (!history.success) throw new Error(history.error);
+      expect(
+        history.data
+          .filter((message) => message.role === "user")
+          .map((message) =>
+            message.parts
+              .filter((part) => part.type === "text")
+              .map((part) => part.text)
+              .join("")
+          )
+      ).toEqual(hasSuccessor ? ["manual replacement"] : []);
+      const source = history.data.find((message) => message.id === summary.id);
+      if (hasSuccessor) {
+        if (!successor) throw new Error("Expected successor compaction");
+        expect(internals.coordinator.isCurrentCompaction(successor)).toBe(true);
+        expect(source?.metadata?.muxMetadata).toHaveProperty(
+          "pendingFollowUp.text",
+          "Replacement follow-up"
+        );
+      } else {
+        expect(source?.metadata?.muxMetadata).not.toHaveProperty("pendingFollowUp");
+        await h.session.dispose();
+        restarted = await createAgentSessionHarness({
+          workspaceId,
+          config: h.config,
+          historyService: h.historyService,
+        });
+        const recovered = restarted.session as unknown as {
+          dispatchPendingFollowUp(): Promise<boolean>;
+        };
+        const resumed = spyOn(restarted.aiService, "streamMessage");
+        expect(await recovered.dispatchPendingFollowUp()).toBe(false);
+        expect(resumed).not.toHaveBeenCalled();
+      }
+    } finally {
+      release.resolve();
+      await pending.catch(() => undefined);
+      await restarted?.session.dispose();
+      await h.session.dispose();
+      await h.cleanup();
+    }
+  }
+);
+
+test.each(["before preparation", "during provider startup"] as const)(
+  "an accepted handoff stays continued when replaced %s before send returns an error",
+  async (replacementPoint) => {
+    const workspaceId = "replaced-accepted-compaction";
+    const h = await createAgentSessionHarness({ workspaceId });
+    const options = { model: "openai:gpt-4o", agentId: "exec" };
+    const internals = h.session as unknown as {
+      coordinator: TurnCoordinator;
+      dispatchPendingFollowUp(): Promise<boolean>;
+    };
+    const send = h.session.sendMessage.bind(h.session);
+    const entered = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    const stream = spyOn(h.aiService, "streamMessage");
+    if (replacementPoint === "before preparation") {
+      spyOn(h.session, "sendMessage").mockImplementationOnce((message, sendOptions, internal) =>
+        send(message, sendOptions, {
+          ...internal,
+          onAccepted: async () => {
+            await internal?.onAccepted?.();
+            entered.resolve();
+            await release.promise;
+          },
+        })
+      );
+    } else {
+      stream.mockImplementationOnce(async () => {
+        entered.resolve();
+        await release.promise;
+        return Err({ type: "runtime_start_failed", message: "retired startup failed" });
+      });
+    }
+    await h.historyService.appendToHistory(
+      workspaceId,
+      createMuxMessage("summary", "assistant", "Earlier work", {
+        muxMetadata: {
+          type: "compaction-summary",
+          pendingFollowUp: { text: "Continue", ...options },
+        },
+      })
+    );
+    const pending = internals.dispatchPendingFollowUp();
+    try {
+      await entered.promise;
+      // A blocked startup can be retired before its engine call returns; its
+      // durable continuation still belongs to the predecessor's completed handoff.
+      if (replacementPoint === "during provider startup")
+        internals.coordinator.preemptPreparation();
+      expect((await send("manual replacement", options)).success).toBe(true);
+      release.resolve();
+      expect(await pending).toBe(true);
+      expect(stream).toHaveBeenCalledTimes(replacementPoint === "before preparation" ? 1 : 2);
+      const history = await h.historyService.getHistoryFromLatestBoundary(workspaceId);
+      expect(
+        history.success &&
+          history.data.filter((message) => message.role === "user").map((message) => message.parts)
+      ).toMatchObject([
+        [{ type: "text", text: "Continue" }],
+        [{ type: "text", text: "manual replacement" }],
+      ]);
+    } finally {
+      release.resolve();
+      await pending.catch(() => undefined);
+      await h.session.dispose();
+      await h.cleanup();
+    }
+  }
+);
+
+test.each([false, true])(
+  "an accepted handoff still reports a current provider startup failure (later replacement=%s)",
+  async (replaceAfterFailure) => {
+    const workspaceId = "failed-accepted-compaction";
+    const h = await createAgentSessionHarness({ workspaceId });
+    spyOn(h.aiService, "streamMessage").mockResolvedValueOnce(
+      Err({ type: "runtime_start_failed", message: "provider startup failed" })
+    );
+    if (replaceAfterFailure) {
+      const send = h.session.sendMessage.bind(h.session);
+      spyOn(h.session, "sendMessage").mockImplementationOnce(async (...args) => {
+        const result = await send(...args);
+        expect(result.success).toBe(false);
+        expect(
+          (await send("manual replacement", { model: "openai:gpt-4o", agentId: "exec" })).success
+        ).toBe(true);
+        return result;
+      });
+    }
+    await h.historyService.appendToHistory(
+      workspaceId,
+      createMuxMessage("summary", "assistant", "Earlier work", {
+        muxMetadata: {
+          type: "compaction-summary",
+          pendingFollowUp: { text: "Continue", model: "openai:gpt-4o", agentId: "exec" },
+        },
+      })
+    );
+    try {
+      const dispatch = h.session as unknown as { dispatchPendingFollowUp(): Promise<boolean> };
+      const failure = await dispatch.dispatchPendingFollowUp().catch((error: unknown) => error);
+      expect(failure).toBeInstanceOf(Error);
+      expect(failure).toHaveProperty("message", expect.stringContaining("provider startup failed"));
+      const history = await h.historyService.getHistoryFromLatestBoundary(workspaceId);
+      expect(
+        history.success && history.data.find((message) => message.role === "user")?.parts
+      ).toMatchObject([{ type: "text", text: "Continue" }]);
+    } finally {
+      await h.session.dispose();
+      await h.cleanup();
+    }
+  }
+);
+
+test("disposal during real pre-acceptance preparation does not report a continued handoff", async () => {
+  const workspaceId = "unaccepted-compaction";
+  const h = await createAgentSessionHarness({ workspaceId });
+  const entered = Promise.withResolvers<void>();
+  const release = Promise.withResolvers<void>();
+  const stream = spyOn(h.aiService, "streamMessage");
+  spyOn(branchSummary, "awaitPendingBranchSummary").mockImplementationOnce(async () => {
+    entered.resolve();
+    await release.promise;
+    return null;
+  });
+  await h.historyService.appendToHistory(
+    workspaceId,
+    createMuxMessage("summary", "assistant", "Earlier work", {
+      muxMetadata: {
+        type: "compaction-summary",
+        pendingFollowUp: {
+          text: "Continue",
+          model: "openai:gpt-4o",
+          agentId: "exec",
+        },
+      },
+    })
+  );
+  const dispatch = h.session as unknown as { dispatchPendingFollowUp(): Promise<boolean> };
+  const pending = dispatch.dispatchPendingFollowUp();
+  let disposal: Promise<void> | undefined;
+  try {
+    await entered.promise;
+    disposal = h.session.dispose();
+    release.resolve();
+    expect(await pending).toBe(false);
+    await disposal;
+    expect(stream).not.toHaveBeenCalled();
+    const history = await h.historyService.getHistoryFromLatestBoundary(workspaceId);
+    expect(history.success && history.data.some((message) => message.role === "user")).toBe(false);
+  } finally {
+    release.resolve();
+    await pending;
+    await disposal;
+    await h.session.dispose();
+    await h.cleanup();
+  }
+});
+
+test("an accepted real handoff stays continued after a subsequent manual replacement", async () => {
+  const workspaceId = "accepted-compaction";
+  const h = await createAgentSessionHarness({ workspaceId });
+  const options = { model: "openai:gpt-4o", agentId: "exec" };
+  await h.historyService.appendToHistory(
+    workspaceId,
+    createMuxMessage("summary", "assistant", "Earlier work", {
+      muxMetadata: {
+        type: "compaction-summary",
+        pendingFollowUp: { text: "Continue", ...options },
+      },
+    })
+  );
+  const send = h.session.sendMessage.bind(h.session);
+  spyOn(h.session, "sendMessage").mockImplementationOnce(async (...args) => {
+    const result = await send(...args);
+    expect(result.success).toBe(true);
+    await h.session.interruptStream();
+    expect((await send("manual replacement", options)).success).toBe(true);
+    return result;
+  });
+  try {
+    const dispatch = h.session as unknown as { dispatchPendingFollowUp(): Promise<boolean> };
+    expect(await dispatch.dispatchPendingFollowUp()).toBe(true);
+    const history = await h.historyService.getHistoryFromLatestBoundary(workspaceId);
+    expect(
+      history.success &&
+        history.data.filter((message) => message.role === "user").map((message) => message.parts)
+    ).toMatchObject([
+      [{ type: "text", text: "Continue" }],
+      [{ type: "text", text: "manual replacement" }],
+    ]);
+  } finally {
+    await h.session.dispose();
+    await h.cleanup();
+  }
+});