mason: stabilize marker advance wire replay

ualtinok · ualtinok · commit 8fac08a1d2fc · 2026-07-18T23:07:17.000+02:00
diff --git a/packages/plugin/src/hooks/magic-context/compaction-marker-manager.ts b/packages/plugin/src/hooks/magic-context/compaction-marker-manager.ts
@@ -55,6 +55,8 @@ export type MarkerUpdateOutcome =
           markerOrdinal: number;
           summaryMessageId: string;
           boundaryMessageId: string;
+          /** Summary source removed from the database during an advance, if any. */
+          removedSummaryMessageId: string | null;
       }
     | { kind: "already-current" }
     | {
@@ -255,6 +257,9 @@ export function applyDeferredCompactionMarker(
 
         // Remove old marker if present. `removeCompactionMarker` returns false
         // only when the DELETE transaction itself failed (e.g. SQLITE_BUSY).
+        // Keep the summary id so the transform can remove the old source from
+        // the in-memory drain representation before it serves the new marker.
+        const removedSummaryMessageId = existing?.summaryMessageId ?? null;
         // No-op success on already-missing rows is fine — that's why retry is
         // safe. False here means we couldn't even attempt the delete cleanly;
         // bail to retryable WITHOUT calling inject (avoids leaving two marker
@@ -309,6 +314,7 @@ export function applyDeferredCompactionMarker(
             markerOrdinal: pending.ordinal,
             summaryMessageId: result.summaryMessageId,
             boundaryMessageId: result.boundaryMessageId,
+            removedSummaryMessageId,
         };
     } catch (err) {
         // Thrown paths:
diff --git a/packages/plugin/src/hooks/magic-context/transform-postprocess-phase.test.ts b/packages/plugin/src/hooks/magic-context/transform-postprocess-phase.test.ts
@@ -20,7 +20,10 @@ import {
     setPendingCompactionMarkerState,
 } from "../../features/magic-context/storage";
 import { initializeDatabase } from "../../features/magic-context/storage-db";
-import { getPersistedCompactionMarkerState } from "../../features/magic-context/storage-meta-persisted";
+import {
+    getPersistedCompactionMarkerState,
+    setPersistedCompactionMarkerState,
+} from "../../features/magic-context/storage-meta-persisted";
 import { createTagger } from "../../features/magic-context/tagger";
 import { Database } from "../../shared/sqlite";
 import { MARKER_SUMMARY_TEXT } from "./compaction-marker-manager";
@@ -484,6 +487,182 @@ describe("deferred compaction marker representation", () => {
     });
 });
 
+describe("deferred compaction marker advance representation", () => {
+    it("keeps the advance drain byte-identical with the next pass after removing the old marker", async () => {
+        db = new Database(":memory:");
+        initializeDatabase(db);
+        const sessionId = "ses-marker-advance-wire-stability";
+        const dataHome = mkdtempSync(join(tmpdir(), "postprocess-marker-advance-wire-"));
+        tempDirs.push(dataHome);
+        process.env.XDG_DATA_HOME = dataHome;
+        mkdirSync(join(dataHome, "opencode"), { recursive: true });
+        const opencodeDb = new Database(join(dataHome, "opencode", "opencode.db"));
+        opencodeDb.exec(
+            "CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, time_updated INTEGER, data TEXT)",
+        );
+        opencodeDb.exec(
+            "CREATE TABLE part (id TEXT PRIMARY KEY, message_id TEXT, session_id TEXT, time_created INTEGER, time_updated INTEGER, data TEXT)",
+        );
+        const insertMessage = opencodeDb.prepare(
+            "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)",
+        );
+        insertMessage.run(
+            "old-boundary",
+            sessionId,
+            1_000,
+            1_000,
+            JSON.stringify({ role: "user" }),
+        );
+        insertMessage.run(
+            "new-boundary",
+            sessionId,
+            2_000,
+            2_000,
+            JSON.stringify({ role: "user" }),
+        );
+        insertMessage.run(
+            "new-end",
+            sessionId,
+            3_000,
+            3_000,
+            JSON.stringify({ role: "assistant", finish: "stop" }),
+        );
+        insertMessage.run(
+            "old-summary",
+            sessionId,
+            1_001,
+            1_001,
+            JSON.stringify({ role: "assistant", summary: true, finish: "stop" }),
+        );
+        opencodeDb.prepare(
+            "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
+        ).run(
+            "old-compaction",
+            "old-boundary",
+            sessionId,
+            1_000,
+            1_000,
+            JSON.stringify({ type: "compaction", auto: true }),
+        );
+        opencodeDb.prepare(
+            "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
+        ).run(
+            "old-summary-part",
+            "old-summary",
+            sessionId,
+            1_001,
+            1_001,
+            JSON.stringify({ type: "text", text: MARKER_SUMMARY_TEXT }),
+        );
+        opencodeDb.close();
+
+        appendCompartments(db, sessionId, [
+            {
+                sequence: 0,
+                startMessage: 1,
+                endMessage: 20,
+                startMessageId: "old-boundary",
+                endMessageId: "new-end",
+                title: "marker advance",
+                content: "test content",
+            },
+        ]);
+        setPersistedCompactionMarkerState(db, sessionId, {
+            boundaryMessageId: "old-boundary",
+            summaryMessageId: "old-summary",
+            compactionPartId: "old-compaction",
+            summaryPartId: "old-summary-part",
+            boundaryOrdinal: 10,
+            targetEndMessageId: "old-end",
+        });
+        setPendingCompactionMarkerState(db, sessionId, {
+            ordinal: 20,
+            endMessageId: "new-end",
+            publishedAt: 2,
+        });
+
+        const tagger = createTagger();
+        const drainMessages = [
+            {
+                info: { role: "user", sessionID: sessionId },
+                parts: [{ type: "text", text: "<session-history>\\n\\n</session-history>" }],
+            },
+            {
+                info: {
+                    id: "old-summary",
+                    role: "assistant",
+                    sessionID: sessionId,
+                    summary: true,
+                    finish: "stop",
+                },
+                parts: [{ type: "text", text: MARKER_SUMMARY_TEXT }],
+            },
+            {
+                info: { id: "new-end", role: "assistant", sessionID: sessionId, finish: "stop" },
+                parts: [{ type: "text", text: "tail content" }],
+            },
+        ] as unknown as MessageLike[];
+        const taggedDrain = tagMessages(sessionId, drainMessages, tagger, db);
+        const deferredHistoryRefreshSessions = new Set<string>([sessionId]);
+
+        await runPostTransformPhase(
+            basePostTransformArgs(db, sessionId, drainMessages, {
+                fullFeatureMode: false,
+                tagger,
+                targets: taggedDrain.targets,
+                reasoningByMessage: taggedDrain.reasoningByMessage,
+                messageTagNumbers: taggedDrain.messageTagNumbers,
+                batch: taggedDrain.batch,
+                deferredHistoryWasPendingAtPassStart: true,
+                historyRebuiltThisPass: true,
+                canConsumeDeferredLate: true,
+                deferredHistoryRefreshSessions,
+                pendingCompartmentInjection: {
+                    block: "",
+                    compartmentEndMessage: 20,
+                    compartmentEndMessageId: "new-end",
+                    compartmentCount: 1,
+                    skippedVisibleMessages: 0,
+                    factCount: 0,
+                    memoryCount: 0,
+                    rebuiltFromDb: true,
+                },
+            }),
+        );
+
+        const marker = getPersistedCompactionMarkerState(db, sessionId);
+        expect(marker?.summaryMessageId).toBeString();
+        expect(drainMessages.some((message) => message.info.id === "old-summary")).toBe(false);
+
+        const rebuiltMessages = [
+            {
+                info: { role: "user", sessionID: sessionId },
+                parts: [{ type: "text", text: "<session-history>\\n\\n</session-history>" }],
+            },
+            {
+                info: {
+                    id: marker?.summaryMessageId,
+                    role: "assistant",
+                    sessionID: sessionId,
+                    summary: true,
+                    finish: "stop",
+                },
+                parts: [{ type: "text", text: MARKER_SUMMARY_TEXT }],
+            },
+            {
+                info: { id: "new-end", role: "assistant", sessionID: sessionId, finish: "stop" },
+                parts: [{ type: "text", text: "tail content" }],
+            },
+        ] as unknown as MessageLike[];
+        tagger.initFromDb(sessionId, db);
+        tagMessages(sessionId, rebuiltMessages, tagger, db);
+
+        expect(serializeAnthropicWireWithAdjacentAssistantMerge(drainMessages)).toBe(
+            serializeAnthropicWireWithAdjacentAssistantMerge(rebuiltMessages),
+        );
+    });
+});
+
 describe("deferred compaction marker CAS drain", () => {
     it("preserves the deferred-history signal when a newer pending blob exists", () => {
         db = new Database(":memory:");
diff --git a/packages/plugin/src/hooks/magic-context/transform-postprocess-phase.ts b/packages/plugin/src/hooks/magic-context/transform-postprocess-phase.ts
@@ -160,6 +160,22 @@ export function injectDeferredCompactionSummaryRepresentation(args: {
     return true;
 }
 
+/**
+ * Remove the prior marker source from the drain array when a boundary advances.
+ * The database mutation happens before the next transform pass, so serving the
+ * old summary during this pass would make the provider wire differ from replay.
+ */
+function removeDeferredCompactionSummaryRepresentation(
+    messages: MessageLike[],
+    summaryMessageId: string | null,
+): boolean {
+    if (summaryMessageId === null) return false;
+    const index = messages.findIndex((message) => message.info.id === summaryMessageId);
+    if (index < 0) return false;
+    messages.splice(index, 1);
+    return true;
+}
+
 function pendingMarkerCoveredByConsumedBoundary(
     pending: PendingCompactionMarker,
     injection: PreparedCompartmentInjection | null,
@@ -1363,6 +1379,10 @@ export async function runPostTransformPhase(
                     args.sessionDirectory,
                 );
                 if (outcome.kind === "applied") {
+                    removeDeferredCompactionSummaryRepresentation(
+                        args.messages,
+                        outcome.removedSummaryMessageId,
+                    );
                     injectDeferredCompactionSummaryRepresentation({
                         db: args.db,
                         sessionId: args.sessionId,
