Skip to content

Commit b7b0c6b

Browse files
committed
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.
1 parent 88e4590 commit b7b0c6b

9 files changed

Lines changed: 209 additions & 10 deletions

‎src/session/assemble-runtime.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ import {
5858
applyRecordingPolicyToText,
5959
createCompactionArchive,
6060
createPrimaryDeliveryAdmission,
61+
hashAuthorizedBytes,
6162
wrapAuthorizeWithEvidenceArchive,
6263
wrapCompactorWithCompletenessGate,
6364
type CompactionArchive,
@@ -491,6 +492,16 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
491492
? storage
492493
: {
493494
...storage,
495+
async writeBlob(key, bytes, contentType, signal) {
496+
await storage.writeBlob(key, bytes, contentType, signal);
497+
if (!key.startsWith("img-")) return;
498+
await primaryArchive.recordExistingBlobReference({
499+
kind: "attachment",
500+
blobKey: key,
501+
contentHash: hashAuthorizedBytes(bytes),
502+
provenance: "persistBlobs:aged-image",
503+
});
504+
},
494505
async writeResponse(turn, signal) {
495506
const content = turn.content.map((block) => {
496507
if (block.type !== "text") return block;

‎src/session/attachment-store.test.ts‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,4 +119,31 @@ describe("ageImageBlocks / rehydrateAttachmentImages", () => {
119119
expect(defined(result.output[0]).content).toEqual([{ type: "text", text }]);
120120
expect(result.record.decisions.restoredImageCount).toBe(0);
121121
});
122+
123+
test("aging base64 images does not record before persistBlobs", async () => {
124+
const recorded: string[] = [];
125+
const archive = {
126+
recordAuthorizedPayload: async () => {
127+
recorded.push("payload");
128+
return {} as never;
129+
},
130+
recordExistingBlobReference: async () => {
131+
recorded.push("existing");
132+
return {} as never;
133+
},
134+
};
135+
const turn: ConversationTurn = {
136+
role: "user",
137+
content: [
138+
{
139+
type: "image",
140+
source: { kind: "base64", mimeType: "image/png", data: PNG_B64 },
141+
},
142+
],
143+
timestamp: 1,
144+
};
145+
const aged = await ageImageBlocks(turn, { archive: archive as never });
146+
expect(aged.blobs).toHaveLength(1);
147+
expect(recorded).toEqual([]);
148+
});
122149
});

‎src/session/compaction-archive.test.ts‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -680,4 +680,44 @@ describe("wrapCompactorWithCompletenessGate", () => {
680680
expect(result.output).toHaveLength(1);
681681
expect(result.record.reason).toBe("compact");
682682
});
683+
684+
test("image-only dropped turns require covering attachment evidence", async () => {
685+
const { wrapCompactorWithCompletenessGate } = await import("./compaction-archive.js");
686+
const { archive } = memoryArchive();
687+
const wrapped = wrapCompactorWithCompletenessGate(truncating("pruning-compactor"), archive);
688+
const png = "iVBORw0KGgo=";
689+
const turns: import("@intx/types/runtime").ConversationTurn[] = [
690+
{
691+
role: "user",
692+
content: [
693+
{
694+
type: "image",
695+
source: { kind: "base64", mimeType: "image/png", data: png },
696+
},
697+
],
698+
timestamp: 1,
699+
},
700+
{
701+
role: "user",
702+
content: [{ type: "text", text: "keep" }],
703+
timestamp: 2,
704+
},
705+
];
706+
707+
const blocked = await wrapped.apply(turns, ctx);
708+
expect(blocked.output).toBe(turns);
709+
expect(blocked.record.reason).toBe("incomplete-evidence-archive");
710+
711+
await archive.recordAuthorizedPayload({
712+
kind: "attachment",
713+
payload: png,
714+
});
715+
await archive.recordAuthorizedPayload({
716+
kind: "user_message",
717+
payload: "keep",
718+
});
719+
const allowed = await wrapped.apply(turns, ctx);
720+
expect(allowed.output).toHaveLength(1);
721+
expect(allowed.record.reason).toBe("compact");
722+
});
683723
});

‎src/session/compaction-archive.ts‎

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import {
2525
type ArchiveKind,
2626
type ToolRecordingLifecycle,
2727
} from "./compaction-archive-schema.js";
28+
import { parseAgedImageMarker } from "./attachment-uri.js";
2829

2930
const INDEX_DIR = "evidence-archive";
3031
const INDEX_FILE = "index.jsonl";
@@ -626,10 +627,12 @@ function textKindForRole(role: ConversationTurn["role"]): ArchiveKind {
626627

627628
function coveringOccurrence(
628629
units: readonly {
629-
kind: "text" | "tool_call" | "tool_result";
630+
kind: "text" | "tool_call" | "tool_result" | "image";
630631
text?: string;
631632
role?: ConversationTurn["role"];
632633
callId?: string;
634+
data?: string;
635+
blobKey?: string;
633636
}[],
634637
occurrences: readonly ArchiveOccurrence[],
635638
): { ids: string[]; unmatched: boolean } {
@@ -638,6 +641,12 @@ function coveringOccurrence(
638641
for (const unit of units) {
639642
const match = occurrences.find((occ) => {
640643
if (used.has(occ.occurrenceId) || occ.gap === true) return false;
644+
if (unit.kind === "image") {
645+
if (occ.kind !== "attachment") return false;
646+
if (unit.blobKey !== undefined) return occ.blobKey === unit.blobKey;
647+
if (unit.data === undefined) return false;
648+
return occ.contentHash === hashAuthorizedBytes(new TextEncoder().encode(unit.data));
649+
}
641650
if (unit.kind === "text") {
642651
if (unit.role === undefined || unit.text === undefined) return false;
643652
if (occ.kind !== textKindForRole(unit.role)) return false;
@@ -655,25 +664,40 @@ function coveringOccurrence(
655664
}
656665

657666
function droppedContentUnits(dropped: readonly ConversationTurn[]): {
658-
kind: "text" | "tool_call" | "tool_result";
667+
kind: "text" | "tool_call" | "tool_result" | "image";
659668
text?: string;
660669
role?: ConversationTurn["role"];
661670
callId?: string;
671+
data?: string;
672+
blobKey?: string;
662673
}[] {
663674
const units: {
664-
kind: "text" | "tool_call" | "tool_result";
675+
kind: "text" | "tool_call" | "tool_result" | "image";
665676
text?: string;
666677
role?: ConversationTurn["role"];
667678
callId?: string;
679+
data?: string;
680+
blobKey?: string;
668681
}[] = [];
669682
for (const turn of dropped) {
670683
for (const block of turn.content) {
671684
if (block.type === "text" && block.text.length > 0) {
672-
units.push({ kind: "text", role: turn.role, text: block.text });
685+
const marker = parseAgedImageMarker(block.text);
686+
if (marker !== undefined) {
687+
units.push({ kind: "image", blobKey: marker.id });
688+
} else {
689+
units.push({ kind: "text", role: turn.role, text: block.text });
690+
}
673691
} else if (block.type === "tool_call") {
674692
units.push({ kind: "tool_call", callId: block.id });
675693
} else if (block.type === "tool_result") {
676694
units.push({ kind: "tool_result", callId: block.callId });
695+
} else if (block.type === "image") {
696+
if (block.source.kind === "base64") {
697+
units.push({ kind: "image", data: block.source.data });
698+
} else {
699+
units.push({ kind: "image" });
700+
}
677701
}
678702
}
679703
}

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

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -782,6 +782,29 @@ describe("createOptimizedContextStore unpublished rewrite", () => {
782782
expect(turnTexts(loaded.turns)).toEqual(["old-a", "old-b"]);
783783
});
784784

785+
test("git commit failure after rewrite lands keeps load on HEAD", async () => {
786+
const dir = tempDir();
787+
const store = await createOptimizedContextStore(dir);
788+
const original = [turn("keep-a"), turn("keep-b"), turn("drop-me")];
789+
await store.writeTurns(original);
790+
await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
791+
const published = await store.commit({ message: "published original" });
792+
793+
await store.writeTurns([turn("[Compacted prior context]"), turn("keep-b")]);
794+
795+
const hookDir = path.join(dir, ".git", "hooks");
796+
fs.mkdirSync(hookDir, { recursive: true });
797+
const hook = path.join(hookDir, "commit-msg");
798+
fs.writeFileSync(hook, "#!/bin/sh\nexit 1\n");
799+
fs.chmodSync(hook, 0o755);
800+
801+
await expect(store.commit({ message: "publish compact" })).rejects.toThrow();
802+
803+
const loaded = await store.load();
804+
expect(turnTexts(loaded.turns)).toEqual(["keep-a", "keep-b", "drop-me"]);
805+
expect(turnTexts(await store.readAt(published.hash))).toEqual(["keep-a", "keep-b", "drop-me"]);
806+
});
807+
785808
test("append writeTurns is still visible before commit", async () => {
786809
const dir = tempDir();
787810
const store = await createOptimizedContextStore(dir);

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

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -438,7 +438,7 @@ export async function createSessionStores(
438438
const base = await createIsogitStore(dir, signer);
439439
const pendingBlobFilepaths = new Set<string>();
440440
const pendingSegmentPaths = new Set<string>();
441-
const writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
441+
let writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
442442
const writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE);
443443
let liveTurnRefs: readonly ConversationTurn[] | null = null;
444444
let unpublishedRewrite: ConversationTurn[] | null = null;
@@ -641,10 +641,9 @@ export async function createSessionStores(
641641
},
642642
async commit(options, signal) {
643643
return withResolvedDirLock(dir, async () => {
644-
if (unpublishedRewrite !== null) {
645-
await writeSegmented(writeTurnsSegmented, unpublishedRewrite);
646-
liveTurnRefs = unpublishedRewrite;
647-
unpublishedRewrite = null;
644+
const stagedRewrite = unpublishedRewrite;
645+
if (stagedRewrite !== null) {
646+
await writeSegmented(writeTurnsSegmented, stagedRewrite);
648647
}
649648
const toAdd: string[] = [];
650649
const toRemove: string[] = [];
@@ -686,9 +685,16 @@ export async function createSessionStores(
686685
const committed = await base.commit(options, signal);
687686
pendingBlobFilepaths.clear();
688687
pendingSegmentPaths.clear();
688+
if (stagedRewrite !== null) {
689+
liveTurnRefs = stagedRewrite;
690+
unpublishedRewrite = null;
691+
}
689692
return committed;
690693
} catch (cause) {
691694
await resetIndexPaths(dir, extraPaths);
695+
if (stagedRewrite !== null) {
696+
liveTurnRefs = null;
697+
}
692698
throw cause;
693699
}
694700
});

‎vendor/intx-inference/PATCHES.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -313,7 +313,9 @@ tracks the skip-unchanged-history patch.
313313
`reactor.ts` — `executeCompact` persists blobs, stages `writeTurns`, and
314314
leaves reactor memory on the old generation until `commitCycle` publishes.
315315
`replaceTurns` runs only after a successful commit. `commitCycle` must not
316-
`writeTurns` live (old) memory over that staging.
316+
`writeTurns` live (old) memory over that staging. A failed commit clears
317+
`pendingCompactOutput` so a later infer/tools cycle writes live history and
318+
does not `replaceTurns` with the unpublished compact.
317319

318320
**Disposition:** Promotion candidate. Compaction durability — an interrupt
319321
between stage and commit must resume the complete old generation, not a

‎vendor/intx-inference/src/reactor.test.ts‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5277,6 +5277,7 @@ function truncatingCompactor(name: string): Compactor {
52775277

52785278
function makeRecordingContextStore(opts?: {
52795279
failCommit?: boolean;
5280+
failCommitRemaining?: { n: number };
52805281
initialTurns?: ConversationTurn[];
52815282
}): {
52825283
store: ContextStore;
@@ -5313,6 +5314,10 @@ function makeRecordingContextStore(opts?: {
53135314
if (opts?.failCommit === true) {
53145315
throw new Error("commit failed");
53155316
}
5317+
if (opts?.failCommitRemaining !== undefined && opts.failCommitRemaining.n > 0) {
5318+
opts.failCommitRemaining.n -= 1;
5319+
throw new Error("commit failed");
5320+
}
53165321
commits.push({
53175322
message: options.message,
53185323
turns: [...lastWrittenTurns],
@@ -5674,6 +5679,64 @@ describe("createReactor — transform chain ordering and compact action", () =>
56745679
expect(seenLengths[1]).toBeGreaterThan(1);
56755680
});
56765681

5682+
test("failed compact commit does not replaceTurns stale output on a later infer cycle", async () => {
5683+
const seed: ConversationTurn[] = [
5684+
{ role: "user", content: [{ type: "text", text: "a" }], timestamp: 1 },
5685+
{ role: "user", content: [{ type: "text", text: "b" }], timestamp: 2 },
5686+
{ role: "user", content: [{ type: "text", text: "c" }], timestamp: 3 },
5687+
];
5688+
const recording = makeRecordingContextStore({
5689+
failCommitRemaining: { n: 1 },
5690+
initialTurns: seed,
5691+
});
5692+
let inspectLength = 0;
5693+
let messages = 0;
5694+
const director: ReactorDirector = {
5695+
async decide(event, state, caps) {
5696+
if (event.type === "message.received") {
5697+
messages++;
5698+
if (messages === 1) {
5699+
return caps.compact("tail-only", "explicit-test");
5700+
}
5701+
if (messages === 2) {
5702+
return caps.infer();
5703+
}
5704+
inspectLength = state.turns.length;
5705+
return caps.done();
5706+
}
5707+
if (event.type === "inference.done") {
5708+
return caps.wait();
5709+
}
5710+
return caps.done();
5711+
},
5712+
};
5713+
const { reactor, waitFor } = createDirectReactor({
5714+
contextStore: recording.store,
5715+
director,
5716+
compactors: { "tail-only": truncatingCompactor("tail-only") },
5717+
inferenceRunner: mockInferenceRunner("live-after-failed-compact"),
5718+
});
5719+
reactor.start();
5720+
reactor.deliver(makeInboundMessage());
5721+
setTimeout(() => reactor.deliver(makeInboundMessage()), 30);
5722+
setTimeout(() => reactor.deliver(makeInboundMessage()), 80);
5723+
await waitFor("reactor.done");
5724+
5725+
const liveWrites = recording.writeTurnsCalls.filter((turns) =>
5726+
turns.some(
5727+
(turn) =>
5728+
turn.role === "assistant" &&
5729+
turn.content.some((b) => b.type === "text" && b.text === "live-after-failed-compact"),
5730+
),
5731+
);
5732+
expect(liveWrites.length).toBeGreaterThan(0);
5733+
expect(liveWrites.some((turns) => turns.length === 1)).toBe(false);
5734+
expect(inspectLength).toBeGreaterThan(1);
5735+
const inferCommit = recording.commits.find((c) => c.message.startsWith("Cycle: inferred"));
5736+
expect(inferCommit).toBeDefined();
5737+
expect(inferCommit?.turns.length).toBeGreaterThan(1);
5738+
});
5739+
56775740
test("compact for an unknown name emits a fatal error and shuts down", async () => {
56785741
const recording = makeRecordingContextStore();
56795742
const { reactor, events, waitFor } = createDirectReactor({

‎vendor/intx-inference/src/reactor.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1098,6 +1098,9 @@ export function createReactor(config: ReactorConfig): Reactor {
10981098
`Cycle commit failed: ${cause instanceof Error ? cause.message : String(cause)}`,
10991099
false,
11001100
);
1101+
// A staged compact must not leak into a later infer/tools cycle: skip-write
1102+
// plus replaceTurns would publish stale compact output over live memory.
1103+
pendingCompactOutput = null;
11011104
resetCycleAccumulators();
11021105
return;
11031106
}

0 commit comments

Comments
 (0)