Keep failed compact commits from replacing live history

A failed compact commit must not later replaceTurns stale output,
and an unpublished rewrite stays off the live generation until git
commit succeeds.

Author: TheGreatAxios

src/session/assemble-runtime.ts

@@ -54,6 +54,7 @@ import {
   applyRecordingPolicyToText,
   createCompactionArchive,
   createPrimaryDeliveryAdmission,
+  hashAuthorizedBytes,
   wrapAuthorizeWithEvidenceArchive,
   wrapCompactorWithCompletenessGate,
   type CompactionArchive,
@@ -454,6 +455,16 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
         ? storage
         : {
             ...storage,
+            async writeBlob(key, bytes, contentType, signal) {
+              await storage.writeBlob(key, bytes, contentType, signal);
+              if (!key.startsWith("img-")) return;
+              await primaryArchive.recordExistingBlobReference({
+                kind: "attachment",
+                blobKey: key,
+                contentHash: hashAuthorizedBytes(bytes),
+                provenance: "persistBlobs:aged-image",
+              });
+            },
             async writeResponse(turn, signal) {
               const content = turn.content.map((block) => {
                 if (block.type !== "text") return block;

src/session/attachment-store.test.ts

@@ -104,4 +104,31 @@ describe("ageImageBlocks / rehydrateAttachmentImages", () => {
     expect(result.output[0]!.content).toEqual([{ type: "text", text }]);
     expect(result.record.decisions.restoredImageCount).toBe(0);
   });
+
+  test("aging base64 images does not record before persistBlobs", async () => {
+    const recorded: string[] = [];
+    const archive = {
+      recordAuthorizedPayload: async () => {
+        recorded.push("payload");
+        return {} as never;
+      },
+      recordExistingBlobReference: async () => {
+        recorded.push("existing");
+        return {} as never;
+      },
+    };
+    const turn: ConversationTurn = {
+      role: "user",
+      content: [
+        {
+          type: "image",
+          source: { kind: "base64", mimeType: "image/png", data: PNG_B64 },
+        },
+      ],
+      timestamp: 1,
+    };
+    const aged = await ageImageBlocks(turn, { archive: archive as never });
+    expect(aged.blobs).toHaveLength(1);
+    expect(recorded).toEqual([]);
+  });
 });

src/session/compaction-archive.test.ts

@@ -680,4 +680,44 @@ describe("wrapCompactorWithCompletenessGate", () => {
     expect(result.output).toHaveLength(1);
     expect(result.record.reason).toBe("compact");
   });
+
+  test("image-only dropped turns require covering attachment evidence", async () => {
+    const { wrapCompactorWithCompletenessGate } = await import("./compaction-archive.js");
+    const { archive } = memoryArchive();
+    const wrapped = wrapCompactorWithCompletenessGate(truncating("pruning-compactor"), archive);
+    const png = "iVBORw0KGgo=";
+    const turns: import("@intx/types/runtime").ConversationTurn[] = [
+      {
+        role: "user",
+        content: [
+          {
+            type: "image",
+            source: { kind: "base64", mimeType: "image/png", data: png },
+          },
+        ],
+        timestamp: 1,
+      },
+      {
+        role: "user",
+        content: [{ type: "text", text: "keep" }],
+        timestamp: 2,
+      },
+    ];
+
+    const blocked = await wrapped.apply(turns, ctx);
+    expect(blocked.output).toBe(turns);
+    expect(blocked.record.reason).toBe("incomplete-evidence-archive");
+
+    await archive.recordAuthorizedPayload({
+      kind: "attachment",
+      payload: png,
+    });
+    await archive.recordAuthorizedPayload({
+      kind: "user_message",
+      payload: "keep",
+    });
+    const allowed = await wrapped.apply(turns, ctx);
+    expect(allowed.output).toHaveLength(1);
+    expect(allowed.record.reason).toBe("compact");
+  });
 });

src/session/compaction-archive.ts

@@ -25,6 +25,7 @@ import {
   type ArchiveKind,
   type ToolRecordingLifecycle,
 } from "./compaction-archive-schema.js";
+import { parseAgedImageMarker } from "./attachment-uri.js";
 
 const INDEX_DIR = "evidence-archive";
 const INDEX_FILE = "index.jsonl";
@@ -626,10 +627,12 @@ function textKindForRole(role: ConversationTurn["role"]): ArchiveKind {
 
 function coveringOccurrence(
   units: readonly {
-    kind: "text" | "tool_call" | "tool_result";
+    kind: "text" | "tool_call" | "tool_result" | "image";
     text?: string;
     role?: ConversationTurn["role"];
     callId?: string;
+    data?: string;
+    blobKey?: string;
   }[],
   occurrences: readonly ArchiveOccurrence[],
 ): { ids: string[]; unmatched: boolean } {
@@ -638,6 +641,12 @@ function coveringOccurrence(
   for (const unit of units) {
     const match = occurrences.find((occ) => {
       if (used.has(occ.occurrenceId) || occ.gap === true) return false;
+      if (unit.kind === "image") {
+        if (occ.kind !== "attachment") return false;
+        if (unit.blobKey !== undefined) return occ.blobKey === unit.blobKey;
+        if (unit.data === undefined) return false;
+        return occ.contentHash === hashAuthorizedBytes(new TextEncoder().encode(unit.data));
+      }
       if (unit.kind === "text") {
         if (unit.role === undefined || unit.text === undefined) return false;
         if (occ.kind !== textKindForRole(unit.role)) return false;
@@ -655,25 +664,40 @@ function coveringOccurrence(
 }
 
 function droppedContentUnits(dropped: readonly ConversationTurn[]): {
-  kind: "text" | "tool_call" | "tool_result";
+  kind: "text" | "tool_call" | "tool_result" | "image";
   text?: string;
   role?: ConversationTurn["role"];
   callId?: string;
+  data?: string;
+  blobKey?: string;
 }[] {
   const units: {
-    kind: "text" | "tool_call" | "tool_result";
+    kind: "text" | "tool_call" | "tool_result" | "image";
     text?: string;
     role?: ConversationTurn["role"];
     callId?: string;
+    data?: string;
+    blobKey?: string;
   }[] = [];
   for (const turn of dropped) {
     for (const block of turn.content) {
       if (block.type === "text" && block.text.length > 0) {
-        units.push({ kind: "text", role: turn.role, text: block.text });
+        const marker = parseAgedImageMarker(block.text);
+        if (marker !== undefined) {
+          units.push({ kind: "image", blobKey: marker.id });
+        } else {
+          units.push({ kind: "text", role: turn.role, text: block.text });
+        }
       } else if (block.type === "tool_call") {
         units.push({ kind: "tool_call", callId: block.id });
       } else if (block.type === "tool_result") {
         units.push({ kind: "tool_result", callId: block.callId });
+      } else if (block.type === "image") {
+        if (block.source.kind === "base64") {
+          units.push({ kind: "image", data: block.source.data });
+        } else {
+          units.push({ kind: "image" });
+        }
       }
     }
   }

src/session/optimized-context-store.test.ts

@@ -666,6 +666,29 @@ describe("createOptimizedContextStore unpublished rewrite", () => {
     expect(turnTexts(loaded.turns)).toEqual(["old-a", "old-b"]);
   });
 
+  test("git commit failure after rewrite lands keeps load on HEAD", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    const original = [turn("keep-a"), turn("keep-b"), turn("drop-me")];
+    await store.writeTurns(original);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    const published = await store.commit({ message: "published original" });
+
+    await store.writeTurns([turn("[Compacted prior context]"), turn("keep-b")]);
+
+    const hookDir = path.join(dir, ".git", "hooks");
+    fs.mkdirSync(hookDir, { recursive: true });
+    const hook = path.join(hookDir, "commit-msg");
+    fs.writeFileSync(hook, "#!/bin/sh\nexit 1\n");
+    fs.chmodSync(hook, 0o755);
+
+    await expect(store.commit({ message: "publish compact" })).rejects.toThrow();
+
+    const loaded = await store.load();
+    expect(turnTexts(loaded.turns)).toEqual(["keep-a", "keep-b", "drop-me"]);
+    expect(turnTexts(await store.readAt(published.hash))).toEqual(["keep-a", "keep-b", "drop-me"]);
+  });
+
   test("append writeTurns is still visible before commit", async () => {
     const dir = tempDir();
     const store = await createOptimizedContextStore(dir);

src/session/optimized-context-store.ts

@@ -381,6 +381,25 @@ async function extraSegmentNamesAtCommit(dir: string, hash: string): Promise<str
   return names;
 }
 
+async function restorePublishedTurnFiles(dir: string): Promise<void> {
+  const listing = await runGit(dir, ["ls-tree", "--name-only", "HEAD"]);
+  const present = new Set(listing.split("\n").filter((line) => line.length > 0));
+  const tracked: string[] = [];
+  if (present.has(TURNS_FILE)) tracked.push(TURNS_FILE);
+  for (let index = 1; ; index++) {
+    const name = segmentFileName(TURNS_FILE, index);
+    if (!present.has(name)) break;
+    tracked.push(name);
+  }
+  if (tracked.length > 0) {
+    await runGit(dir, ["checkout", "HEAD", "--", ...tracked]);
+  }
+  const disk = await listSegmentFiles(dir, TURNS_FILE);
+  for (const name of disk) {
+    if (!present.has(name)) await fs.promises.unlink(path.join(dir, name));
+  }
+}
+
 async function describeHead(dir: string, message: string): Promise<ContextCommit> {
   const [hash, seconds, parents] = (await runGit(dir, ["log", "-1", "--format=%H%n%ct%n%P"])).split(
     "\n",
@@ -446,7 +465,7 @@ export async function createOptimizedContextStore(
   const base = await createIsogitStore(dir);
   const pendingBlobFilepaths = new Set<string>();
   const pendingSegmentPaths = new Set<string>();
-  const writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
+  let writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
   const writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE);
   let liveTurnRefs: readonly ConversationTurn[] | null = null;
   let unpublishedRewrite: ConversationTurn[] | null = null;
@@ -620,50 +639,62 @@ export async function createOptimizedContextStore(
       pendingBlobFilepaths.add(`${TOOL_OUTPUT_DIR}/${filename}`);
     },
     async commit(options, _signal) {
-      if (unpublishedRewrite !== null) {
-        await writeSegmented(writeTurnsSegmented, unpublishedRewrite);
-        liveTurnRefs = unpublishedRewrite;
-        unpublishedRewrite = null;
+      const stagedRewrite = unpublishedRewrite;
+      if (stagedRewrite !== null) {
+        await writeSegmented(writeTurnsSegmented, stagedRewrite);
       }
 
-      const toAdd: string[] = [];
-      const toRemove: string[] = [];
+      try {
+        const toAdd: string[] = [];
+        const toRemove: string[] = [];
 
-      const rewrittenEachCycle = [RESPONSE_FILE, MANIFEST_FILE, METADATA_FILE];
-      for (const filepath of rewrittenEachCycle) {
-        if (await pathExists(path.join(dir, filepath))) toAdd.push(filepath);
-      }
+        const rewrittenEachCycle = [RESPONSE_FILE, MANIFEST_FILE, METADATA_FILE];
+        for (const filepath of rewrittenEachCycle) {
+          if (await pathExists(path.join(dir, filepath))) toAdd.push(filepath);
+        }
 
-      for (const filepath of [...pendingSegmentPaths, ...pendingBlobFilepaths]) {
-        if (await pathExists(path.join(dir, filepath))) toAdd.push(filepath);
-        else toRemove.push(filepath);
-      }
+        for (const filepath of [...pendingSegmentPaths, ...pendingBlobFilepaths]) {
+          if (await pathExists(path.join(dir, filepath))) toAdd.push(filepath);
+          else toRemove.push(filepath);
+        }
 
-      if (await pathExists(path.join(dir, EVIDENCE_ARCHIVE_DIR))) {
-        toAdd.push(EVIDENCE_ARCHIVE_DIR);
-      }
+        if (await pathExists(path.join(dir, EVIDENCE_ARCHIVE_DIR))) {
+          toAdd.push(EVIDENCE_ARCHIVE_DIR);
+        }
 
-      // Disk is source of truth for which turn/prompt segments should remain
-      // tracked after a rewrite or heal, even if pendingSegmentPaths was lost.
-      await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove);
-      await reconcileSegmentStaging(dir, PROMPT_FILE, toAdd, toRemove);
+        // Disk is source of truth for which turn/prompt segments should remain
+        // tracked after a rewrite or heal, even if pendingSegmentPaths was lost.
+        await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove);
+        await reconcileSegmentStaging(dir, PROMPT_FILE, toAdd, toRemove);
 
-      const add = [...new Set(toAdd)];
-      const remove = [...new Set(toRemove)].filter((p) => !add.includes(p));
+        const add = [...new Set(toAdd)];
+        const remove = [...new Set(toRemove)].filter((p) => !add.includes(p));
 
-      if (add.length > 0) await runGit(dir, ["add", "--", ...add]);
-      if (remove.length > 0) {
-        await runGit(dir, ["rm", "--cached", "--ignore-unmatch", "--", ...remove]);
+        if (add.length > 0) await runGit(dir, ["add", "--", ...add]);
+        if (remove.length > 0) {
+          await runGit(dir, ["rm", "--cached", "--ignore-unmatch", "--", ...remove]);
+        }
+        await runGit(
+          dir,
+          ["commit", "-m", options.message, `--author=${author.name} <${author.email}>`],
+          author,
+          gitEnv,
+        );
+        pendingBlobFilepaths.clear();
+        pendingSegmentPaths.clear();
+        if (stagedRewrite !== null) {
+          liveTurnRefs = stagedRewrite;
+          unpublishedRewrite = null;
+        }
+        return describeHead(dir, options.message);
+      } catch (cause) {
+        if (stagedRewrite !== null) {
+          await restorePublishedTurnFiles(dir);
+          writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
+          liveTurnRefs = null;
+        }
+        throw cause;
       }
-      await runGit(
-        dir,
-        ["commit", "-m", options.message, `--author=${author.name} <${author.email}>`],
-        author,
-        gitEnv,
-      );
-      pendingBlobFilepaths.clear();
-      pendingSegmentPaths.clear();
-      return describeHead(dir, options.message);
     },
   };
 }

vendor/intx-inference/PATCHES.md

@@ -299,7 +299,9 @@ tracks the skip-unchanged-history patch.
 `reactor.ts` — `executeCompact` persists blobs, stages `writeTurns`, and
 leaves reactor memory on the old generation until `commitCycle` publishes.
 `replaceTurns` runs only after a successful commit. `commitCycle` must not
-`writeTurns` live (old) memory over that staging.
+`writeTurns` live (old) memory over that staging. A failed commit clears
+`pendingCompactOutput` so a later infer/tools cycle writes live history and
+does not `replaceTurns` with the unpublished compact.
 
 **Disposition:** Promotion candidate. Compaction durability — an interrupt
 between stage and commit must resume the complete old generation, not a