diff --git a/src/common/utils/messages/compactionBoundary.test.ts b/src/common/utils/messages/compactionBoundary.test.ts index 09e634eece..dc0230a882 100644 --- a/src/common/utils/messages/compactionBoundary.test.ts +++ b/src/common/utils/messages/compactionBoundary.test.ts @@ -214,6 +214,28 @@ describe("context boundary helpers", () => { ).toBe(false); expect(hasProviderEligibleMessages([createMuxMessage("u1", "user", "after")])).toBe(true); }); + + it("optionally counts reasoning while excluding rejected payloads and reset markers", () => { + const reasoning = createMuxMessage("thinking", "assistant", ""); + reasoning.parts = [{ type: "reasoning", text: "Preserved thinking" }]; + expect(hasProviderEligibleMessages([reasoning])).toBe(false); + expect(hasProviderEligibleMessages([reasoning], { preserveReasoningOnly: true })).toBe(true); + for (const metadata of [ + { contextBudgetRejected: true as const }, + { contextBoundaryKind: "reset" as const }, + ]) { + expect( + hasProviderEligibleMessages([{ ...reasoning, metadata }], { + preserveReasoningOnly: true, + }) + ).toBe(false); + } + expect( + hasProviderEligibleMessages([createMuxMessage("empty", "assistant", "")], { + preserveReasoningOnly: true, + }) + ).toBe(false); + }); }); describe("sliceMessagesFromLatestCompactionBoundary", () => { diff --git a/src/common/utils/messages/compactionBoundary.ts b/src/common/utils/messages/compactionBoundary.ts index b4a260bce2..69aa23e0d3 100644 --- a/src/common/utils/messages/compactionBoundary.ts +++ b/src/common/utils/messages/compactionBoundary.ts @@ -128,17 +128,23 @@ export function sliceMessagesFromLatestCompactionBoundary(messages: MuxMessage[] return sliced; } -export function isProviderEligibleMessage(message: MuxMessage): boolean { +export function isProviderEligibleMessage( + message: MuxMessage, + options?: Parameters[1] +): boolean { if (isDurableContextResetBoundaryMarker(message)) { return false; } - return hasProviderReplayableContent(message); + return hasProviderReplayableContent(message, options); } -export function hasProviderEligibleMessages(messages: MuxMessage[]): boolean { +export function hasProviderEligibleMessages( + messages: MuxMessage[], + options?: Parameters[1] +): boolean { assert(Array.isArray(messages), "hasProviderEligibleMessages requires a message array"); - return messages.some(isProviderEligibleMessage); + return messages.some((message) => isProviderEligibleMessage(message, options)); } /** diff --git a/src/node/services/historyService.test.ts b/src/node/services/historyService.test.ts index 3b2d4f7f97..3b86113ba3 100644 --- a/src/node/services/historyService.test.ts +++ b/src/node/services/historyService.test.ts @@ -4,12 +4,14 @@ import { CONTEXT_BOUNDARY_KINDS } from "@/common/constants/contextBoundary"; import { HistoryService } from "./historyService"; import type { Config } from "@/node/config"; import { createTestHistoryService } from "./testHistoryService"; +import type { ContinuousCompactionJournal } from "@/common/orpc/schemas/continuousCompaction"; +import { createContextBudgetRejectedMessage } from "@/common/utils/messages/contextBudgetRejection"; import { updateSubagentTranscriptArtifactsFile } from "./subagentTranscriptArtifacts"; import { createMuxMessage, type MuxMessage } from "@/common/types/message"; import assert from "node:assert"; import { createHash } from "node:crypto"; import * as fs from "fs/promises"; -import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import * as fileLock from "@/node/utils/concurrency/fileLock"; import { historyWriteLockPath, workspaceRemovalTombstonePath, @@ -388,7 +390,7 @@ describe("HistoryService", () => { // session-dir append lock: the batch's read+replace must wait, or its // replacement — built from contents read before the foreign append — // would silently delete the foreign row. - const foreign = await acquireProcessFileLock({ + const foreign = await fileLock.acquireProcessFileLock({ // r63: the history write lock lives outside the session directory so // removal can hold it across its tombstone+delete critical section. lockPath: historyWriteLockPath(config.rootDir, workspaceId), @@ -461,7 +463,7 @@ describe("HistoryService", () => { JSON.stringify({ finalArchiveHash: "in-flight", finalChatHash: "in-flight" }) ); - const foreign = await acquireProcessFileLock({ + const foreign = await fileLock.acquireProcessFileLock({ lockPath: historyWriteLockPath(config.rootDir, workspaceId), timeoutMs: 5_000, label: "test foreign truncation", @@ -991,6 +993,589 @@ describe("HistoryService", () => { }); }); + describe("destructive context publication fencing", () => { + const ws = "context-publication"; + const row = (id: string) => createMuxMessage(id, "user", `Context for ${id}`); + const reasoning = (id: string): MuxMessage => ({ + ...createMuxMessage(id, "assistant", "", { synthetic: true }), + parts: [{ type: "reasoning", text: "Provider-visible thinking" }], + }); + const display = () => + createMuxMessage("display", "user", "Workflow display", { + muxMetadata: { type: "workflow-trigger-display", rawCommand: "/wf", runId: "run" }, + }); + const boundary = () => + createMuxMessage("sealed", "assistant", "Compacted context", { + compacted: "user", + compactionBoundary: true, + compactionEpoch: 1, + }); + const reset = (rollover = false) => + createMuxMessage("reset", "assistant", "", { + contextBoundaryKind: CONTEXT_BOUNDARY_KINDS.RESET, + ...(rollover && { + muxMetadata: { + type: "context-window-rollover", + rolloverId: "rollover", + reason: "on-send", + previousWindowId: "w:0", + flushOpportunity: false, + contextTokens: 1000, + maxTokens: 1000, + } as const, + }), + }); + const prefix = [{ role: "user" as const, content: "Discarded provider context" }]; + + async function rejectLatestBudgetRequest() { + const latest = await service.getLastMessages(ws, 1); + assert(latest.success && latest.data.length === 1); + return service.rejectContextBudgetRequest(ws, latest.data[0]); + } + + async function capturePublication() { + const store = service.getContinuousCompactionJournal(ws); + const journal: ContinuousCompactionJournal = { + version: 1, + publicationGeneration: await store.captureGeneration(), + boundary: { ...boundary(), id: "compacted" }, + staticCopies: [], + liveTailCopySpec: { + sourceMessageId: "live", + sourceHistorySequence: 0, + copyId: "copy", + partIndex: 0, + metadataTemplate: { synthetic: true, rlmPreservedTailCopy: true }, + }, + postCompactionAttachments: [], + prefixSourceRows: [row("old")], + systemPrefix: [], + cacheEnabled: false, + preparation: { + modelString: "anthropic:claude-sonnet-4-5", + providerForMessages: "anthropic", + effectiveThinkingLevel: "off", + effectiveAgentId: "exec", + toolNamesForSentinel: [], + }, + providerFamily: "anthropic", + parentModel: "anthropic:claude-sonnet-4-5", + summaryModel: "anthropic:claude-sonnet-4-5", + headFingerprint: "head", + sourceFingerprint: "source", + headEnd: { id: "old", sequence: 0 }, + epoch: 0, + streamMessageId: "live", + streamHistorySequence: 0, + stepNumber: 0, + firstTailToolCallId: "tool", + }; + const receipt = await store.write(journal, prefix, () => true); + assert(receipt, "Expected a durable compaction publication"); + return { store, receipt }; + } + + const destructiveCases = [ + ...[false, true].flatMap((rollover) => + [false, true].map((batch) => ({ + name: `${rollover ? "rollover" : "reset"} ${batch ? "batch" : "single"}`, + rows: [row("old")], + mutate: () => + batch + ? service.appendManyToHistory(ws, [reset(rollover), row("fresh")]) + : service.appendToHistory(ws, reset(rollover)), + expected: batch ? ["old", "reset", "fresh"] : ["old", "reset"], + })) + ), + { + name: "matching-tail reset", + rows: [row("old")], + mutate: () => service.appendToHistoryIfTailMatches(ws, reset(), "old"), + expected: ["old", "reset"], + }, + { + name: "full clear", + rows: [row("old"), row("tail")], + mutate: () => service.clearHistory(ws), + expected: [], + }, + { + name: "display-only full clear", + rows: [display()], + mutate: () => service.clearHistory(ws), + expected: [], + }, + { + name: "rounded full clear", + rows: [row("old"), row("tail")], + mutate: () => service.truncateHistory(ws, 0.99), + expected: [], + }, + { + name: "active prefix", + rows: [row("old"), row("tail")], + mutate: () => service.truncateHistory(ws, 0.2), + expected: ["tail"], + }, + { + name: "active edit", + rows: [row("old"), row("tail")], + mutate: () => service.truncateAfterMessage(ws, "tail"), + expected: ["old"], + }, + { + name: "reasoning-only active edit", + rows: [row("old"), reasoning("tail")], + mutate: () => service.truncateAfterMessage(ws, "tail"), + expected: ["old"], + }, + { + name: "reasoning-only active prefix", + rows: [reasoning("old"), row("tail")], + mutate: () => service.truncateHistory(ws, 0.2), + expected: ["tail"], + }, + { + name: "reasoning-only context-budget rejection", + // A quarantined trigger isolates a still-visible owned reasoning prelude on retry. + rows: [ + row("old"), + reasoning("prelude"), + createContextBudgetRejectedMessage( + createMuxMessage("trigger", "user", "Rejected request", { + requestPreludeMessageIds: ["prelude"], + }) + ), + ], + mutate: rejectLatestBudgetRequest, + expected: ["old", "prelude", "trigger"], + }, + { + name: "archived edit", + rows: [row("old"), boundary(), row("tail")], + mutate: () => service.truncateAfterMessage(ws, "old", { keepTargetMessage: true }), + expected: ["old"], + }, + ...[ + { name: "reset", message: reset() }, + { name: "rollover", message: reset(true) }, + { name: "empty compaction", message: { ...boundary(), parts: [] } }, + ].flatMap(({ name, message }) => + ["active", "archived"].map((target) => ({ + name: `${target} edit removing only ${name} boundary`, + rows: [row("old"), message], + mutate: () => + target === "active" + ? service.truncateAfterMessage(ws, message.id) + : service.truncateAfterMessage(ws, "old", { keepTargetMessage: true }), + expected: ["old"], + })) + ), + { + name: "context-budget rejection", + rows: [ + row("old"), + createMuxMessage("prelude", "assistant", "Owned payload", { synthetic: true }), + createMuxMessage("trigger", "user", "Rejected request", { + requestPreludeMessageIds: ["prelude"], + }), + ], + mutate: async () => { + const result = await rejectLatestBudgetRequest(); + if (result.success) { + expect(result.data.map((message) => message.id)).toEqual(["prelude", "trigger"]); + expect(result.data.every((message) => message.parts.length === 0)).toBe(true); + } + return result; + }, + expected: ["old", "prelude", "trigger"], + }, + ]; + + it.each(destructiveCases)( + "$name prevents stale journal and history resurrection", + async (testCase) => { + for (const message of testCase.rows) { + assert((await service.appendToHistory(ws, structuredClone(message))).success); + } + const { store, receipt } = await capturePublication(); + const foreign = new HistoryService(config); + assert((await testCase.mutate()).success); + const settled = await collectFullHistory(foreign, ws); + expect(settled.map((message) => message.id)).toEqual(testCase.expected); + expect(await store.captureGeneration()).not.toBe(receipt.publicationGeneration); + + let committed = false; + const folded = await foreign.persistBoundaryWithTailCopies( + ws, + structuredClone(receipt.boundary), + [], + false, + () => true, + { + publication: { generation: receipt.publicationGeneration, journal: receipt }, + onCommitted: () => { + committed = true; + }, + } + ); + expect(folded.success).toBe(false); + expect(committed).toBe(false); + const foreignStore = foreign.getContinuousCompactionJournal(ws); + expect(await foreignStore.read()).toBeNull(); + // Rejection must survive cleanup of the old record, when the slot is empty. + expect(await foreignStore.write(receipt, prefix, () => true)).toBeNull(); + expect(await collectFullHistory(foreign, ws)).toEqual(settled); + expect( + await foreignStore.write( + { ...receipt, publicationGeneration: await foreignStore.captureGeneration() }, + prefix, + () => true + ) + ).not.toBeNull(); + } + ); + + it.each(destructiveCases)( + "$name leaves history intact if generation advancement fails", + async (testCase) => { + for (const message of testCase.rows) { + assert((await service.appendToHistory(ws, structuredClone(message))).success); + } + const { store, receipt } = await capturePublication(); + const before = await collectFullHistory(service, ws); + const failure = spyOn(store, "advanceGenerationUnderHistoryLock").mockRejectedValueOnce( + new Error("generation unavailable") + ); + try { + expect((await testCase.mutate()).success).toBe(false); + expect(await collectFullHistory(new HistoryService(config), ws)).toEqual(before); + expect(await store.read()).toEqual(receipt); + } finally { + failure.mockRestore(); + } + } + ); + + it("holds the history file lock from generation advancement through deletion", async () => { + assert((await service.appendToHistory(ws, row("old"))).success); + const store = service.getContinuousCompactionJournal(ws); + // Exercise an existing durable generation too, rather than only legacy absence. + await store.advanceGeneration(); + const { receipt } = await capturePublication(); + await store.clear(receipt); + const historyPath = path.join(config.sessionsDir, ws, "chat.jsonl"); + const before = await fs.readFile(historyPath, "utf8"); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const advance = store.advanceGenerationUnderHistoryLock.bind(store); + const advancing = spyOn(store, "advanceGenerationUnderHistoryLock").mockImplementationOnce( + async () => { + await fs.access(historyWriteLockPath(config.rootDir, ws)); + await advance(); + entered.resolve(); + await release.promise; + } + ); + const clearing = service.clearHistory(ws); + const foreign = new HistoryService(config).getContinuousCompactionJournal(ws); + const capture = spyOn(foreign, "captureGenerationUnderHistoryLock"); + const acquire = fileLock.acquireProcessFileLock; + const attempted = Promise.withResolvers(); + let acquiring: + | ReturnType> + | undefined; + let writing: ReturnType | undefined; + try { + await entered.promise; + acquiring = spyOn(fileLock, "acquireProcessFileLock").mockImplementation((options) => { + attempted.resolve(); + return acquire(options); + }); + writing = foreign.write(receipt, prefix, () => true); + await attempted.promise; + expect(capture).not.toHaveBeenCalled(); + expect(await fs.readFile(historyPath, "utf8")).toBe(before); + release.resolve(); + expect((await clearing).success).toBe(true); + expect(await writing).toBeNull(); + expect(await foreign.captureGeneration()).not.toBe(receipt.publicationGeneration); + expect(await collectFullHistory(new HistoryService(config), ws)).toEqual([]); + } finally { + release.resolve(); + await Promise.all([clearing, writing]); + acquiring?.mockRestore(); + capture.mockRestore(); + advancing.mockRestore(); + } + }); + + it.each([ + { + name: "zero prefix", + rows: [row("old")], + mutate: () => service.truncateHistory(ws, 0), + expected: ["old"], + success: true, + }, + { + name: "rounded zero", + rows: [row("old")], + mutate: () => service.truncateHistory(ws, 0.0001), + expected: ["old"], + success: true, + }, + { + name: "empty clear", + rows: [], + mutate: () => service.clearHistory(ws), + expected: [], + success: true, + }, + { + name: "declined full", + rows: [row("old")], + mutate: () => service.truncateHistory(ws, 0.9, { refuseFullDelete: true }), + expected: ["old"], + success: false, + }, + { + name: "declined removal", + rows: [row("old")], + mutate: () => service.truncateHistory(ws, 0.2, { refuseRowRemoval: true }), + expected: ["old"], + success: false, + }, + { + name: "declined partial", + rows: [row("old"), row("tail")], + mutate: () => service.truncateHistory(ws, 0.2, { requireFullDelete: true }), + expected: ["old", "tail"], + success: false, + }, + { + name: "missing edit", + rows: [row("old")], + mutate: () => service.truncateAfterMessage(ws, "missing"), + expected: ["old"], + success: false, + }, + { + name: "retained tail", + rows: [row("old")], + mutate: () => service.truncateAfterMessage(ws, "old", { keepTargetMessage: true }), + expected: ["old"], + success: true, + }, + ...[reset(), { ...boundary(), parts: [] }].map((message) => ({ + name: `retained ${message.id} boundary`, + rows: [row("old"), message], + mutate: () => service.truncateAfterMessage(ws, message.id, { keepTargetMessage: true }), + expected: ["old", message.id], + success: true, + })), + { + name: "invalid empty compaction marker", + rows: [ + row("old"), + createMuxMessage("invalid", "assistant", "", { + compactionBoundary: true, + compacted: "user", + compactionEpoch: 0, + }), + ], + mutate: () => service.truncateAfterMessage(ws, "invalid"), + expected: ["old"], + success: true, + }, + { + name: "display edit", + rows: [row("old"), display()], + mutate: () => service.truncateAfterMessage(ws, "display"), + expected: ["old"], + success: true, + }, + { + name: "display prefix", + rows: [display(), row("tail")], + mutate: () => service.truncateHistory(ws, 0.1), + expected: ["tail"], + success: true, + }, + { + name: "sealed prefix", + rows: [row("old"), boundary(), row("tail")], + mutate: () => service.truncateHistory(ws, 0.1), + expected: ["sealed", "tail"], + success: true, + }, + { + name: "sealed reasoning-only prefix", + rows: [reasoning("old"), boundary(), row("tail")], + mutate: () => service.truncateHistory(ws, 0.1), + expected: ["sealed", "tail"], + success: true, + }, + { + name: "mismatched reset", + rows: [row("old")], + mutate: () => service.appendToHistoryIfTailMatches(ws, reset(), "missing"), + expected: ["old"], + success: true, + }, + { + name: "compaction boundary", + rows: [row("old")], + mutate: () => service.appendToHistory(ws, boundary()), + expected: ["old", "sealed"], + success: true, + }, + { + name: "missing rejection trigger", + rows: [row("old")], + mutate: () => + service.rejectContextBudgetRequest( + ws, + createMuxMessage("missing", "user", "Gone request", { historySequence: 0 }) + ), + expected: ["old"], + success: false, + }, + { + name: "repeated rejection capsule", + rows: [createContextBudgetRejectedMessage(row("old"))], + mutate: rejectLatestBudgetRequest, + expected: ["old"], + success: true, + }, + ])("$name preserves a usable compaction publication", async (testCase) => { + for (const message of testCase.rows) { + assert((await service.appendToHistory(ws, message)).success); + } + const { store, receipt } = await capturePublication(); + expect((await testCase.mutate()).success).toBe(testCase.success); + expect((await collectFullHistory(service, ws)).map((message) => message.id)).toEqual([ + ...testCase.expected, + ]); + expect(await store.read()).toEqual(receipt); + expect(await store.captureGeneration()).toBe(receipt.publicationGeneration); + let committed = false; + const foreign = new HistoryService(config); + const folded = await foreign.persistBoundaryWithTailCopies( + ws, + structuredClone(receipt.boundary), + [], + false, + () => true, + { + publication: { generation: receipt.publicationGeneration, journal: receipt }, + onCommitted: () => { + committed = true; + }, + } + ); + expect(folded.success).toBe(true); + expect(committed).toBe(true); + const active = await foreign.getHistoryFromLatestBoundary(ws); + assert(active.success); + expect(active.data.map((message) => message.id)).toEqual([receipt.boundary.id]); + expect(await store.captureGeneration()).toBe(receipt.publicationGeneration); + }); + + it.each(["active", "archived"])( + "%s display-only edit retains raw reset evidence and publication", + async (target) => { + assert((await service.appendManyToHistory(ws, [row("old"), display()])).success); + assert((await service.getHistoryFromLatestBoundary(ws)).success); + const chatPath = path.join(config.sessionsDir, ws, "chat.jsonl"); + const archivePath = path.join(config.sessionsDir, ws, "chat-archive.jsonl"); + const [oldLine, displayLine] = (await fs.readFile(chatPath, "utf8")).trimEnd().split("\n"); + const raw = ' {\n"contextBoundaryKind"\n:\n"reset"\n}\n'; + if (target === "archived") await fs.writeFile(archivePath, oldLine + "\n"); + await fs.writeFile( + chatPath, + (target === "active" ? oldLine + "\n" : "") + raw + displayLine + "\n" + ); + const { store, receipt } = await capturePublication(); + const cut = await service.truncateAfterMessage( + ws, + target === "active" ? "display" : "old", + { + keepTargetMessage: target === "archived", + } + ); + assert(cut.success); + expect(cut.data.removedMessages.map((message) => message.id)).toEqual(["display"]); + expect(await fs.readFile(chatPath, "utf8")).toContain(raw); + const restarted = new HistoryService(config); + const active = await restarted.getHistoryFromLatestBoundary(ws); + assert(active.success); + expect(active.data).toEqual([]); + expect((await collectFullHistory(restarted, ws)).map((message) => message.id)).toEqual([ + "old", + ]); + expect(await store.captureGeneration()).toBe(receipt.publicationGeneration); + expect(await store.read()).toEqual(receipt); + } + ); + + it.each( + [ + '{"metadata":{"contextBoundaryKind":"reset"},broken\n', + ' {\n"contextBoundaryKind"\n:\n"reset"\n}\n', + '{"id":"reset","role":"assistant","parts":[],"metadata":{"contextBoundaryKind":"reset"},"metadata":{}}\n', + ].flatMap((raw) => + ["chat", "archive"].flatMap((artifact) => + [0, 0.1, 0.5].map((percentage) => ({ raw, artifact, percentage })) + ) + ) + )( + "prefix $percentage respects the retained raw reset floor in $artifact ($raw)", + async ({ raw, artifact, percentage }) => { + const source = [ + row("old"), + row("active"), + createMuxMessage("tail", "assistant", "Active reply", { + contextUsage: { inputTokens: 100, outputTokens: 10, totalTokens: 110 }, + }), + ]; + assert((await service.appendManyToHistory(ws, source)).success); + assert((await service.getHistoryFromLatestBoundary(ws)).success); + const chatPath = path.join(config.sessionsDir, ws, "chat.jsonl"); + const archivePath = path.join(config.sessionsDir, ws, "chat-archive.jsonl"); + const lines = (await fs.readFile(chatPath, "utf8")).trimEnd().split("\n"); + const sealed = lines[0] + "\n" + raw; + const active = lines.slice(1).join("\n") + "\n"; + if (artifact === "archive") await fs.writeFile(archivePath, sealed); + await fs.writeFile(chatPath, (artifact === "chat" ? sealed : "") + active); + const before = await service.getHistoryFromLatestBoundary(ws); + assert(before.success); + expect(before.data.map((message) => message.id)).toEqual(["active", "tail"]); + const { store, receipt } = await capturePublication(); + const truncated = await service.truncateHistory(ws, percentage); + assert(truncated.success); + const changed = percentage === 0.5; + expect(truncated.data).toEqual(percentage === 0 ? [] : changed ? [0, 1] : [0]); + const restarted = new HistoryService(config); + const after = await restarted.getHistoryFromLatestBoundary(ws); + assert(after.success); + if (changed) { + expect(after.data.map((message) => message.id)).toEqual(["tail"]); + expect(after.data[0].metadata?.contextUsage).toBeUndefined(); + expect(await store.read()).toBeNull(); + expect(await store.captureGeneration()).not.toBe(receipt.publicationGeneration); + } else { + expect(after.data).toEqual(before.data); + expect(await store.captureGeneration()).toBe(receipt.publicationGeneration); + expect(await store.read()).toEqual(receipt); + } + expect(await fs.readFile(artifact === "chat" ? chatPath : archivePath, "utf8")).toContain( + raw + ); + } + ); + }); + describe("clearHistory", () => { it("should delete chat.jsonl file", async () => { const workspaceId = "workspace1"; diff --git a/src/node/services/historyService.ts b/src/node/services/historyService.ts index c80965a79c..96101f5700 100644 --- a/src/node/services/historyService.ts +++ b/src/node/services/historyService.ts @@ -49,7 +49,6 @@ import { safeStringifyForCounting } from "@/common/utils/tokens/safeStringifyFor import { normalizeLegacyMuxMetadata } from "@/node/utils/messages/legacy"; import { CONTEXT_BOUNDARY_KINDS } from "@/common/constants/contextBoundary"; import { - findLatestContextBoundaryIndex, getContextBoundaryKind, hasProviderEligibleMessages, isDurableCompactedMarker, @@ -118,16 +117,15 @@ function hasDurableCompactionBoundary(metadata: MuxMetadata | undefined): boolea return isPositiveInteger(metadata.compactionEpoch); } -function prefixCutChangesActiveContext(messages: MuxMessage[], removeCount: number): boolean { - const boundaryIndex = findLatestContextBoundaryIndex(messages); - const activeStart = - boundaryIndex < 0 - ? 0 - : getContextBoundaryKind(messages[boundaryIndex]) === CONTEXT_BOUNDARY_KINDS.RESET - ? boundaryIndex + 1 - : boundaryIndex; - return hasProviderEligibleMessages( - filterWorkflowDisplayOnlyMessages(messages.slice(activeStart, removeCount)) +function tailCutChangesProviderContext(removedMessages: MuxMessage[]): boolean { + // Even an empty boundary can hide older context. Removing it changes the + // provider view; unreadable reset evidence is preserved outside this parsed tail. + return ( + removedMessages.some(isDurableContextBoundaryMarker) || + // Shared history may be replayed with Anthropic thinking enabled. + hasProviderEligibleMessages(filterWorkflowDisplayOnlyMessages(removedMessages), { + preserveReasoningOnly: true, + }) ); } @@ -2458,6 +2456,7 @@ export class HistoryService { `[HISTORY APPEND] Assigned historySequence=${message.metadata.historySequence ?? "unknown"} role=${message.role}` ); + await this.fenceContextResetUnderHistoryLock(workspaceId, [message]); await this.getAppendProvenance(workspaceId).appendChat( Buffer.from(JSON.stringify(historyEntry) + "\n") ); @@ -2468,6 +2467,20 @@ export class HistoryService { } } + private async fenceContextResetUnderHistoryLock( + workspaceId: string, + messages: readonly MuxMessage[] + ): Promise { + if ( + messages.some((message) => getContextBoundaryKind(message) === CONTEXT_BOUNDARY_KINDS.RESET) + ) { + // Reset/rollover discards context captured by foreign compactors too. Advance + // after admission but before appending under the same lock, so a failed + // generation write cannot leave a durable reset open to stale publication. + await this.getContinuousCompactionJournal(workspaceId).advanceGenerationUnderHistoryLock(); + } + } + /** Serialize messages as JSONL rows tagged with workspace context. */ private serializeHistoryEntries(messages: readonly MuxMessage[], workspaceId: string): string { return messages.map((msg) => JSON.stringify({ ...msg, workspaceId }) + "\n").join(""); @@ -2751,6 +2764,7 @@ export class HistoryService { // temp-and-rename helper the other history mutations use, under the // cross-process append lock (r50) so a foreign backend's row cannot // land between this read and the replace and be silently deleted. + await this.fenceContextResetUnderHistoryLock(workspaceId, messages); await this.getAppendProvenance(workspaceId).appendChat( Buffer.from(this.serializeHistoryEntries(messages, workspaceId)), true @@ -2857,6 +2871,7 @@ export class HistoryService { ) ); const rejected: MuxMessage[] = []; + let providerContextChanged = false; const earlier = new Set(messages.slice(0, triggerIndex)); const updated = this.serializeHistoryRewrite(rows, workspaceId, (row) => { const ownedPrelude = @@ -2866,10 +2881,21 @@ export class HistoryService { (isSyntheticSnapshotUserMessage(row) || (row.role === "assistant" && row.metadata?.synthetic === true)); if (row !== persisted && !ownedPrelude) return row; + providerContextChanged ||= hasProviderEligibleMessages( + filterWorkflowDisplayOnlyMessages([row]), + { preserveReasoningOnly: true } + ); const marked = createContextBudgetRejectedMessage(row); rejected.push(marked); return marked; }); + // Rejection removes provider context just like truncation. Retire foreign + // compactors before rewriting, but preserve publication on capsule retries. + if (providerContextChanged) { + await this.getContinuousCompactionJournal( + workspaceId + ).advanceGenerationUnderHistoryLock(); + } await writeFileAtomic(historyPath, updated); return Ok(rejected); } @@ -3334,6 +3360,13 @@ export class HistoryService { const archiveMaxSeq = await this.getArchiveTailMaxSequence(workspaceId); + // A real edit must retire captured provider context; missing targets, + // keep-target-at-tail no-ops and display-only cuts retain publication. + if (tailCutChangesProviderContext(removedMessages)) { + await this.getContinuousCompactionJournal( + workspaceId + ).advanceGenerationUnderHistoryLock(); + } // Atomic write prevents corruption if app crashes mid-write await writeFileAtomic(historyPath, historyEntries); @@ -3414,6 +3447,9 @@ export class HistoryService { if (lastArchiveRow && lastArchiveRow.raw.at(-1) !== 10 && activeEpochRows.length > 0) { archiveRows.push({ raw: Buffer.from("\n"), message: undefined }); } + if (tailCutChangesProviderContext(removedMessages)) { + await this.getContinuousCompactionJournal(workspaceId).advanceGenerationUnderHistoryLock(); + } await this.rewriteHistoryFilesUnlocked( workspaceId, null, @@ -3551,6 +3587,13 @@ export class HistoryService { .filter((s): s is number => isNonNegativeInteger(s)); if (percentage >= 1.0) { + // Explicit full-clear intent includes display-only or malformed history. + // An already-empty history remains a no-op for publication ownership. + if (archiveRows.length > 0 || chatRows.length > 0) { + await this.getContinuousCompactionJournal( + workspaceId + ).advanceGenerationUnderHistoryLock(); + } await this.rewriteHistoryFilesUnlocked(workspaceId, null, null); this.sequenceCounters.set(workspaceId, 0); return Ok(allSequences); @@ -3607,12 +3650,32 @@ export class HistoryService { "Truncation would remove every remaining message; retry to run it as a full clear." ); } + await this.getContinuousCompactionJournal( + workspaceId + ).advanceGenerationUnderHistoryLock(); await this.rewriteHistoryFilesUnlocked(workspaceId, null, null); this.sequenceCounters.set(workspaceId, 0); return Ok(allSequences); } - const activeContextChanged = prefixCutChangesActiveContext(messages, removeCount); + // The raw-aware reader returns the active suffix of these parsed rows. + // Reuse its privacy floor: retained unreadable reset evidence can seal + // rows that parsed boundary markers alone would misclassify as active. + const activeMessages = await readProviderHistoryFromLatestBoundary( + { + chat: this.getChatHistoryPath(workspaceId), + archive: this.getChatArchivePath(workspaceId), + }, + 0 + ); + const activeRemoveCount = Math.max( + 0, + removeCount - (messages.length - activeMessages.length) + ); + const activeContextChanged = hasProviderEligibleMessages( + filterWorkflowDisplayOnlyMessages(activeMessages.slice(0, activeRemoveCount)), + { preserveReasoningOnly: true } + ); const sanitize = activeContextChanged ? stripContextUsage : (message: MuxMessage) => message; @@ -3634,6 +3697,13 @@ export class HistoryService { retainedMessages, sanitize ); + // Trimming sealed or display-only rows does not change a compactor's + // active provider context, so only an active-context cut retires it. + if (activeContextChanged) { + await this.getContinuousCompactionJournal( + workspaceId + ).advanceGenerationUnderHistoryLock(); + } await this.rewriteHistoryFilesUnlocked( workspaceId, remainingArchive.length > 0 ? remainingArchive : null,