From e4228d9a50fbd515ad8cff2bf92ba3fcf5477312 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 16:45:36 +0200 Subject: [PATCH 1/4] =?UTF-8?q?=F0=9F=A4=96=20tests:=20await=20token-budge?= =?UTF-8?q?t=20warning=20visibility?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wait for the warning's visible state after opening it so the assertion can tolerate the app entrance transition without accepting persistent invisibility. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ieeb9c6b96af089dab27156d43a7f54fcb9b38dbe --- src/browser/stories/App.tokenBudget.stories.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/browser/stories/App.tokenBudget.stories.tsx b/src/browser/stories/App.tokenBudget.stories.tsx index 283ae0e4466..288b43496b0 100644 --- a/src/browser/stories/App.tokenBudget.stories.tsx +++ b/src/browser/stories/App.tokenBudget.stories.tsx @@ -133,7 +133,7 @@ export const Rollover: AppStory = { await expect(canvas.queryByText(WARNING)).not.toBeInTheDocument(); const warning = await canvas.findByRole("button", { name: /Context budget warning/ }); await userEvent.click(warning); - await expect(canvas.getByText(WARNING)).toBeVisible(); + await waitFor(() => expect(canvas.getByText(WARNING)).toBeVisible()); await userEvent.click(warning); const tool = await canvas.findByText("session_history", { exact: true }); await userEvent.click(tool); From 369b1a6f52056cf3f8bcf1f6de10e8ab57e54828 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 14:36:23 +0200 Subject: [PATCH 2/4] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20capture=20post-c?= =?UTF-8?q?ompaction=20snapshots=20for=20request=20consumption?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bind acknowledgement and context-exceeded discard to the snapshot injected into that request. Serialize local sidecar writes and cleanup so a late consumer cannot delete replacement state. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ic4c8b849c3a49f7269591d140df156251240020b --- .../agentSession.pendingOwnership.test.ts | 219 ++++++++++++++++++ src/node/services/agentSession.ts | 53 +++-- ...compactionHandler.pendingOwnership.test.ts | 182 +++++++++++++++ src/node/services/compactionHandler.ts | 117 ++++++---- 4 files changed, 508 insertions(+), 63 deletions(-) create mode 100644 src/node/services/agentSession.pendingOwnership.test.ts create mode 100644 src/node/services/compactionHandler.pendingOwnership.test.ts diff --git a/src/node/services/agentSession.pendingOwnership.test.ts b/src/node/services/agentSession.pendingOwnership.test.ts new file mode 100644 index 00000000000..c7d1a16ab52 --- /dev/null +++ b/src/node/services/agentSession.pendingOwnership.test.ts @@ -0,0 +1,219 @@ +import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"; +import { EventEmitter } from "events"; +import { readFile } from "fs/promises"; +import * as path from "path"; +import { TURNS_BETWEEN_ATTACHMENTS } from "@/common/constants/attachments"; +import type { PostCompactionAttachment } from "@/common/types/attachment"; +import { createMuxMessage } from "@/common/types/message"; +import { Ok } from "@/common/types/result"; +import type { StreamEndEvent } from "@/common/types/stream"; +import assert from "@/common/utils/assert"; +import type { CompactionHandler } from "./compactionHandler"; +import type { TurnCoordinator } from "./turnCoordinator"; +import type { TurnCompletion } from "./streamManager"; +import { createAgentSessionHarness, createStartedTurnHandle } from "./agentSession.testHarness"; + +const workspaceId = "consumer-pending-owner"; +const model = "openai:gpt-4o"; +interface SessionAccess { + compactionHandler: CompactionHandler; + coordinator: TurnCoordinator; + turnsSinceLastAttachment: number; + getPostCompactionAttachmentsIfNeeded( + includeReadFiles: boolean + ): Promise; + clearStartupAutoRetryAbandon(): Promise; + maybeRetryCompactionOnContextExceeded(): Promise; +} + +afterEach(() => { + mock.restore(); +}); + +describe("pending snapshot consumers", () => { + test.each([ + { outcome: "success", periodic: false, replacement: true }, + { outcome: "context-exceeded", periodic: false, replacement: true }, + { outcome: "success", periodic: true, replacement: true }, + { outcome: "context-exceeded", periodic: true, replacement: true }, + { outcome: "context-exceeded", periodic: true, replacement: false }, + ] as const)( + "$outcome retains its request snapshot (periodic=$periodic, replacement=$replacement)", + async ({ outcome, periodic, replacement }) => { + const completion = Promise.withResolvers(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const emitter = new EventEmitter(); + let calls = 0; + const h = await createAgentSessionHarness({ + workspaceId, + aiEmitter: emitter, + aiServiceOverrides: { + streamMessage: mock(() => { + const messageId = ++calls > 1 ? "retry" : "assistant"; + emitter.emit("stream-start", { + type: "stream-start", + workspaceId, + messageId, + model, + startTime: Date.now(), + }); + return Promise.resolve( + Ok( + calls > 1 + ? createStartedTurnHandle(h.session.closingSignal, messageId) + : { messageId, completion: completion.promise } + ) + ); + }), + }, + }); + const session = h.session as unknown as SessionAccess; + const handler = session.compactionHandler; + const consumer = spyOn(session.coordinator, "consumeCompletion"); + async function publish(id: string) { + const edit = createMuxMessage(id, "assistant", ""); + edit.parts = [ + { + type: "dynamic-tool", + toolCallId: id, + toolName: "file_edit_replace_string", + state: "output-available", + input: { path: `/${id}.ts` }, + output: { success: true, diff: `change ${id}` }, + }, + { + type: "dynamic-tool", + toolCallId: `${id}-read`, + toolName: "file_read", + state: "output-available", + input: { path: `/${id}.ts` }, + output: { success: true }, + }, + ]; + expect( + await handler.withContinuousPendingState( + [edit], + (boundaryMessageId) => + handler.persistContinuousCompaction({ + messages: [edit], + boundaryMessageId, + text: `${id} summary`, + model, + tail: [], + systemMessageTokens: 0, + attachmentTokens: 0, + shouldPersist: () => true, + }), + id + ) + ).toBe(true); + } + let policy: Promise | undefined; + try { + await publish("a"); + if (periodic) { + await session.getPostCompactionAttachmentsIfNeeded(true); + await handler.ackPendingStateConsumed(); + session.turnsSinceLastAttachment = TURNS_BETWEEN_ATTACHMENTS; + // Periodic injection reads current-epoch edits even when RLM read-path injection is off. + const recent = createMuxMessage("recent", "assistant", ""); + recent.parts = [ + { + type: "dynamic-tool", + toolCallId: "recent-edit", + toolName: "file_edit_replace_string", + state: "output-available", + input: { path: "/recent.ts" }, + output: { success: true, diff: "recent change" }, + }, + ]; + await h.historyService.appendToHistory(workspaceId, recent); + } + const consume = spyOn( + handler, + outcome === "success" ? "ackPendingStateConsumed" : "discardPendingState" + ); + await h.historyService.appendToHistory( + workspaceId, + createMuxMessage("user", "user", "continue") + ); + expect((await h.session.resumeStream({ model, agentId: "exec" })).success).toBe(true); + if (outcome === "success") { + spyOn(session, "clearStartupAutoRetryAbandon").mockImplementationOnce(async () => { + entered.resolve(); + await release.promise; + }); + } else { + spyOn(session, "maybeRetryCompactionOnContextExceeded").mockImplementationOnce( + async () => { + entered.resolve(); + await release.promise; + return false; + } + ); + } + const end: StreamEndEvent = { + type: "stream-end", + workspaceId, + messageId: "assistant", + metadata: { model }, + parts: [{ type: "text", text: "done" }], + }; + completion.resolve( + outcome === "success" + ? { status: "completed", streamEnd: end } + : { + status: "failed", + streamError: { + messageId: "assistant", + error: "too large", + errorType: "context_exceeded", + }, + } + ); + await entered.promise; + const result = consumer.mock.results.at(-1); + assert(result?.type === "return", "Expected an active completion consumer"); + policy = result.value; + const pendingPath = path.join(h.config.sessionsDir, workspaceId, "post-compaction.json"); + let bytes: string | undefined; + if (replacement) { + await publish("b"); + // A successor request captures B while A's policy is suspended. A retains its entry value. + const attachments = await session.getPostCompactionAttachmentsIfNeeded(false); + expect( + attachments + ?.find((attachment) => attachment.type === "edited_files_reference") + ?.files.some((file) => file.path === "/b.ts") + ).toBe(true); + bytes = await readFile(pendingPath, "utf8"); + } + release.resolve(); + await policy; + expect(consume).toHaveBeenCalledTimes(1); + if (replacement) { + expect( + (await handler.peekPendingState())?.diffs.some((diff) => diff.path === "/b.ts") + ).toBe(true); + assert(bytes !== undefined, "Expected replacement file"); + expect(await readFile(pendingPath, "utf8")).toBe(bytes); + } else { + // The failed periodic injection still owns A's retained read-path carryover after ack. + await publish("c"); + expect((await handler.peekPendingState())?.readFiles).toEqual(["/c.ts"]); + } + if (outcome === "context-exceeded") { + expect(calls).toBe(2); + expect(h.session.isBusy()).toBe(true); + } + } finally { + release.resolve(); + completion.resolve({ status: "aborted", abortReason: "system" }); + await policy; + await h.session.dispose(); + await h.cleanup(); + } + } + ); +}); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index d9cf86b59f3..b4ee45e6eb2 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -1016,11 +1016,15 @@ export class AgentSession { private postCompactionReadFilePaths: string[] = []; /** - * When true, clear any persisted post-compaction state after the next successful non-compaction stream. + * Retain the exact injected snapshot so a late completion cannot consume a replacement. * * This is intentionally delayed until stream-end so a crash mid-stream doesn't lose the diffs. */ - private ackPendingPostCompactionStateOnStreamEnd = false; + private pendingPostCompactionStateToAcknowledge: Awaited< + ReturnType + > = null; + /** Periodic reinjection retains the consumed owner of the cached skill/read carryover. */ + private postCompactionState: typeof this.pendingPostCompactionStateToAcknowledge = null; /** * Cached memory session context (memory experiment): index snapshot for @@ -6723,7 +6727,7 @@ export class AgentSession { // Reset per-stream flags (used for retries / crash-safe bookkeeping). this.compactionMonitor.resetForNewStream(); this.clearLiveUsageState(); - this.ackPendingPostCompactionStateOnStreamEnd = false; + this.pendingPostCompactionStateToAcknowledge = null; this.activeStreamHadAnyDelta = false; this.activeStreamHadPostCompactionInjection = false; const providersConfig = this.getProvidersConfigSafe(); @@ -7366,10 +7370,10 @@ export class AgentSession { return true; } - private async maybeRetryWithoutPostCompactionOnContextExceeded(data: { - messageId: string; - errorType?: string; - }): Promise { + private async maybeRetryWithoutPostCompactionOnContextExceeded( + data: { messageId: string; errorType?: string }, + pendingState = this.pendingPostCompactionStateToAcknowledge + ): Promise { const expectedTurnId = this.coordinator.turnId; const expectedOperationId = this.coordinator.operationId; if (data.errorType !== "context_exceeded") { @@ -7405,10 +7409,13 @@ export class AgentSession { }); // The post-compaction context is likely the culprit; discard it so we don't loop. - this.postCompactionLoadedSkills = []; - this.postCompactionReadFilePaths = []; + if (this.postCompactionState === pendingState) { + this.postCompactionLoadedSkills = []; + this.postCompactionReadFilePaths = []; + this.postCompactionState = null; + } try { - await this.compactionHandler.discardPendingState("context_exceeded"); + await this.compactionHandler.discardPendingState("context_exceeded", pendingState); this.onPostCompactionStateChange?.(); } catch (error) { log.warn("Failed to discard pending post-compaction state", { @@ -7608,7 +7615,7 @@ export class AgentSession { this.activeStreamStartedAtMs = undefined; this.activeStreamHadPostCompactionInjection = false; this.activeStreamHadAnyDelta = false; - this.ackPendingPostCompactionStateOnStreamEnd = false; + this.pendingPostCompactionStateToAcknowledge = null; } private async handleStreamError( @@ -7616,6 +7623,7 @@ export class AgentSession { operation = this.coordinator.operationId ): Promise { const turn = this.coordinator.turnId; + const pendingStateToDiscard = this.pendingPostCompactionStateToAcknowledge; this.coordinator.beginPolicy(turn); if (!this.coordinator.isCurrentTurn(turn) || !this.coordinator.isCurrentOperation(operation)) return; @@ -7712,10 +7720,10 @@ export class AgentSession { return; if ( - await this.maybeRetryWithoutPostCompactionOnContextExceeded({ - messageId: data.messageId, - errorType: data.errorType, - }) + await this.maybeRetryWithoutPostCompactionOnContextExceeded( + { messageId: data.messageId, errorType: data.errorType }, + pendingStateToDiscard + ) ) { return; // retry set PREPARING } @@ -7944,6 +7952,7 @@ export class AgentSession { operation = this.coordinator.operationId ): Promise { const turn = this.coordinator.turnId; + const pendingStateToAcknowledge = this.pendingPostCompactionStateToAcknowledge; this.coordinator.beginPolicy(turn); if (!this.coordinator.isCurrentTurn(turn) || !this.coordinator.isCurrentOperation(operation)) return; @@ -8004,10 +8013,11 @@ export class AgentSession { this.emitChatEvent(payload); emittedStreamEnd = true; - if (this.ackPendingPostCompactionStateOnStreamEnd) { - this.ackPendingPostCompactionStateOnStreamEnd = false; + if (pendingStateToAcknowledge) { + if (this.pendingPostCompactionStateToAcknowledge === pendingStateToAcknowledge) + this.pendingPostCompactionStateToAcknowledge = null; try { - await this.compactionHandler.ackPendingStateConsumed(); + await this.compactionHandler.ackPendingStateConsumed(pendingStateToAcknowledge); if ( !this.coordinator.isCurrentTurn(turn) || !this.coordinator.isCurrentOperation(operation) @@ -9726,6 +9736,7 @@ export class AgentSession { * (compactionOccurred + the in-session mirrors). */ async clearPostCompactionState(): Promise { + this.postCompactionState = null; this.memoryContextByModelString.clear(); // In-memory clears stay unconditional: they stop THIS session from // injecting carryover even when the durable discard below fails. @@ -9733,7 +9744,7 @@ export class AgentSession { this.turnsSinceLastAttachment = TURNS_BETWEEN_ATTACHMENTS; this.postCompactionLoadedSkills = []; this.postCompactionReadFilePaths = []; - this.ackPendingPostCompactionStateOnStreamEnd = false; + this.pendingPostCompactionStateToAcknowledge = null; // Durable-or-throw: a swallowed unlink failure would leave the stale // post-compaction.json to re-inject pre-boundary carryover after a // restart while the boundary caller reports success — the same @@ -9812,7 +9823,8 @@ export class AgentSession { // Check if compaction just occurred (immediate injection with cached post-compaction state) const pendingState = await this.compactionHandler.peekPendingState(); if (pendingState !== null) { - this.ackPendingPostCompactionStateOnStreamEnd = true; + this.postCompactionState = pendingState; + this.pendingPostCompactionStateToAcknowledge = pendingState; this.compactionOccurred = true; this.turnsSinceLastAttachment = 0; this.postCompactionLoadedSkills = pendingState.loadedSkills; @@ -9840,6 +9852,7 @@ export class AgentSession { // Check cooldown for subsequent injections (re-read from current history) if (this.compactionOccurred && this.turnsSinceLastAttachment >= TURNS_BETWEEN_ATTACHMENTS) { + this.pendingPostCompactionStateToAcknowledge = this.postCompactionState; this.turnsSinceLastAttachment = 0; return this.generatePostCompactionAttachments(includeReadFiles); } diff --git a/src/node/services/compactionHandler.pendingOwnership.test.ts b/src/node/services/compactionHandler.pendingOwnership.test.ts new file mode 100644 index 00000000000..c58d990df49 --- /dev/null +++ b/src/node/services/compactionHandler.pendingOwnership.test.ts @@ -0,0 +1,182 @@ +import { afterEach, beforeEach, describe, expect, it, mock, spyOn } from "bun:test"; +import { EventEmitter } from "events"; +import * as fs from "fs/promises"; +import * as path from "path"; +import { createMuxMessage } from "@/common/types/message"; +import assert from "@/common/utils/assert"; +import { CompactionHandler } from "./compactionHandler"; +import { createTestHistoryService } from "./testHistoryService"; + +const workspaceId = "pending-consumers"; +const followUp = { text: "wake", model: "openai:gpt-4o", agentId: "exec" }; + +describe("exact pending snapshot consumption", () => { + let store: Awaited>; + let handler: CompactionHandler; + let sessionDir: string; + let pendingPath: string; + + function restart() { + return new CompactionHandler({ + workspaceId, + historyService: store.historyService, + sessionDir, + emitter: new EventEmitter(), + }); + } + + beforeEach(async () => { + store = await createTestHistoryService(); + sessionDir = path.join(store.tempDir, "pending"); + pendingPath = path.join(sessionDir, "post-compaction.json"); + handler = restart(); + }); + + afterEach(async () => { + mock.restore(); + await store.cleanup(); + }); + + async function publish(id?: string) { + if (id) { + const message = createMuxMessage(id, "assistant", ""); + message.parts = [ + { + type: "dynamic-tool", + toolCallId: id, + toolName: "file_read", + state: "output-available", + input: { path: `/${id}.ts` }, + output: { success: true }, + }, + ]; + await store.historyService.appendToHistory(workspaceId, message); + } + expect( + ( + await handler.appendHeartbeatContextResetBoundary({ + boundaryText: "reset", + pendingFollowUp: followUp, + }) + ).success + ).toBe(true); + const state = await handler.peekPendingState(); + assert(state, "Expected published state"); + return state; + } + + it.each( + (["ack", "discard"] as const).flatMap((action) => + [false, true].map((reload) => ({ action, reload })) + ) + )( + "late $action preserves an identical-byte successor (reload=$reload)", + async ({ action, reload }) => { + spyOn(Date, "now").mockReturnValue(1234); + await publish(); + if (reload) handler = restart(); + const consumed = await handler.peekPendingState(); + const previous = await fs.readFile(pendingPath, "utf8"); + await publish(); + expect(await fs.readFile(pendingPath, "utf8")).toBe(previous); + if (action === "ack") await handler.ackPendingStateConsumed(consumed); + else await handler.discardPendingState("context_exceeded", consumed); + expect(await handler.peekPendingState()).not.toBeNull(); + expect(await fs.readFile(pendingPath, "utf8")).toBe(previous); + } + ); + + it.each(["ack", "discard"] as const)( + "%s unlink finishes before a successor write", + async (action) => { + const consumed = await publish("a"); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const unlink = fs.unlink; + spyOn(fs, "unlink").mockImplementationOnce(async (target) => { + entered.resolve(); + await release.promise; + return unlink(target); + }); + const cleanup = + action === "ack" + ? handler.ackPendingStateConsumed(consumed) + : handler.discardPendingState("context_exceeded", consumed); + let replacement: ReturnType | undefined; + try { + await entered.promise; + // Observe the real persistence call after B's cache update; keep A's physical unlink held. + const persistence = handler as unknown as { + persistPendingStateBestEffort(...args: unknown[]): Promise; + }; + const persist = persistence.persistPendingStateBestEffort.bind(handler); + const writeRequested = Promise.withResolvers(); + spyOn(persistence, "persistPendingStateBestEffort").mockImplementation((...args) => { + const result = persist(...args); + writeRequested.resolve(); + return result; + }); + const mkdir = spyOn(fs, "mkdir"); + replacement = publish("b"); + await writeRequested.promise; + expect(mkdir.mock.calls.some(([dir]) => String(dir) === sessionDir)).toBe(false); + release.resolve(); + await cleanup; + await replacement; + expect((await handler.peekPendingState())?.readFiles).toContain("/b.ts"); + expect((await restart().peekPendingState())?.readFiles).toContain("/b.ts"); + } finally { + release.resolve(); + await cleanup; + await replacement; + } + } + ); + + it.each(["ack", "discard"] as const)( + "%s retires older bytes after a failed pending write", + async (action) => { + await publish("a"); + const previous = await fs.readFile(pendingPath, "utf8"); + const mkdir = fs.mkdir; + const failure = spyOn(fs, "mkdir").mockImplementation((async ( + ...args: Parameters + ) => { + if (String(args[0]) === sessionDir) throw new Error("pending mkdir failed"); + return mkdir(...args); + }) as typeof fs.mkdir); + let consumed: Awaited>; + try { + consumed = await publish("b"); + expect(failure.mock.calls.some(([dir]) => String(dir) === sessionDir)).toBe(true); + } finally { + failure.mockRestore(); + } + expect(await fs.readFile(pendingPath, "utf8")).toBe(previous); + if (action === "ack") await handler.ackPendingStateConsumed(consumed); + else await handler.discardPendingState("context_exceeded", consumed); + expect(await handler.peekPendingState()).toBeNull(); + expect(await restart().peekPendingState()).toBeNull(); + } + ); + + it("repeated peeks share consumption authority and a later discard clears retained carryover", async () => { + const first = await publish("a"); + const second = await handler.peekPendingState(); + await handler.ackPendingStateConsumed(first); + await handler.discardPendingState("context_exceeded", second); + const next = await publish("b"); + expect(next.readFiles).toEqual(["/b.ts"]); + }); + + it("a failed unlink cannot let a retried old acknowledgement remove its successor", async () => { + const consumed = await publish("a"); + spyOn(fs, "unlink").mockRejectedValueOnce(new Error("pending unlink failed")); + await handler.ackPendingStateConsumed(consumed); + expect(await fs.readFile(pendingPath, "utf8")).toContain("/a.ts"); + await publish("b"); + await handler.ackPendingStateConsumed(consumed); + expect((await handler.peekPendingState())?.readFiles).toContain("/b.ts"); + expect((await restart().peekPendingState())?.readFiles).toContain("/b.ts"); + }); +}); diff --git a/src/node/services/compactionHandler.ts b/src/node/services/compactionHandler.ts index d4b2c5b5eb5..31eb124777e 100644 --- a/src/node/services/compactionHandler.ts +++ b/src/node/services/compactionHandler.ts @@ -420,6 +420,11 @@ export class CompactionHandler { private cachedFileDiffs: FileEditDiff[] = []; /** Rollback snapshot for synthetic heartbeat reset boundaries that get skipped before dispatch. */ private heartbeatResetRollbackState: HeartbeatResetRollbackState | null = null; + // Request consumers retain this identity rather than claiming whichever snapshot is current later. + private pendingStateOwner = Symbol(); + private pendingStateFileOwner?: symbol; + private pendingStateWrites: Promise = Promise.resolve(); + private readonly pendingStateOwners = new WeakMap(); /** Cached loaded skill snapshots extracted from history before appending compaction summary */ private cachedLoadedSkills: LoadedSkillSnapshot[] = []; /** Cumulative file paths read in summarized epochs (paths only, newest-first, capped). */ @@ -441,16 +446,28 @@ export class CompactionHandler { this.onIdleCompactionOutcome = options.onIdleCompactionOutcome; } + private enqueuePendingStateWrite(operation: () => Promise): Promise { + const result = this.pendingStateWrites.then(operation); + this.pendingStateWrites = result.catch(() => undefined); + return result; + } + private async loadPersistedPendingStateIfNeeded(): Promise { if (this.persistedPendingStateLoaded || this.postCompactionAttachmentsPending) { return; } this.persistedPendingStateLoaded = true; + const owner = Symbol(); + this.pendingStateOwner = owner; let raw: string; try { - raw = await fsPromises.readFile(this.postCompactionStatePath, "utf-8"); + raw = await this.enqueuePendingStateWrite(async () => { + const contents = await fsPromises.readFile(this.postCompactionStatePath, "utf-8"); + this.pendingStateFileOwner = owner; + return contents; + }); } catch { return; } @@ -460,21 +477,21 @@ export class CompactionHandler { parsed = JSON.parse(raw); } catch { log.warn("Invalid post-compaction state JSON; ignoring", { workspaceId: this.workspaceId }); - await this.deletePersistedPendingStateBestEffort(); + await this.deletePersistedPendingStateBestEffort(owner); return; } let state = coercePersistedPostCompactionState(parsed); if (!state) { log.warn("Invalid post-compaction state schema; ignoring", { workspaceId: this.workspaceId }); - await this.deletePersistedPendingStateBestEffort(); + await this.deletePersistedPendingStateBestEffort(owner); return; } if (state.boundaryMessageId) { const history = await this.historyService.getHistoryFromLatestBoundary(this.workspaceId); if (!history.success) { - this.persistedPendingStateLoaded = false; + if (this.pendingStateOwner === owner) this.persistedPendingStateLoaded = false; return; } const boundaryId = history.data.findLast(isDurableContextBoundaryMarker)?.id; @@ -483,11 +500,13 @@ export class CompactionHandler { // nor lose an older, still-pending attachment snapshot. state = state.previousState ?? null; if (!state || (state.boundaryMessageId && state.boundaryMessageId !== boundaryId)) { - await this.deletePersistedPendingStateBestEffort(); + await this.deletePersistedPendingStateBestEffort(owner); return; } } } + // A load must not relabel a snapshot written while its history check was awaiting I/O. + if (this.pendingStateOwner !== owner) return; this.pendingStateBoundaryMessageId = state.boundaryMessageId; this.cachedFileDiffs = state.diffs; this.cachedLoadedSkills = state.loadedSkills; @@ -508,11 +527,13 @@ export class CompactionHandler { return null; } - return { + const state = { diffs: this.cachedFileDiffs, loadedSkills: this.cachedLoadedSkills, readFiles: this.cachedReadFilePaths, }; + this.pendingStateOwners.set(state, this.pendingStateOwner); + return state; } /** @@ -536,47 +557,44 @@ export class CompactionHandler { * seen" memory, so the next compaction must merge them even when the pending * state was consumed in between. */ - async ackPendingStateConsumed(): Promise { - this.pendingStateBoundaryMessageId = undefined; - // If we never loaded persisted state but it exists, clear it anyway. - if (!this.postCompactionAttachmentsPending && !this.persistedPendingStateLoaded) { - await this.loadPersistedPendingStateIfNeeded(); - } - - this.postCompactionAttachmentsPending = false; - this.cachedFileDiffs = []; - await this.deletePersistedPendingStateBestEffort(); + async ackPendingStateConsumed(expected?: PendingPostCompactionState | null): Promise { + await this.consumePendingState(expected, false); } /** * Drop pending post-compaction state (e.g., because it caused context_exceeded). */ - async discardPendingState(reason: string): Promise { - this.pendingStateBoundaryMessageId = undefined; - await this.loadPersistedPendingStateIfNeeded(); - - const hadPendingState = this.postCompactionAttachmentsPending; - if ( - !hadPendingState && - this.cachedLoadedSkills.length === 0 && - this.cachedReadFilePaths.length === 0 - ) { - return; - } - - log.warn("Discarding pending post-compaction state", { + async discardPendingState( + reason: string, + expected?: PendingPostCompactionState | null + ): Promise { + log.debug("Discarding pending post-compaction state", { workspaceId: this.workspaceId, reason, - trackedFiles: this.cachedFileDiffs.length, - loadedSkills: this.cachedLoadedSkills.length, - readFiles: this.cachedReadFilePaths.length, }); + await this.consumePendingState(expected, true); + } - if (hadPendingState) { - await this.ackPendingStateConsumed(); + private async consumePendingState( + expected: PendingPostCompactionState | null | undefined, + discard: boolean + ): Promise { + if (expected === undefined) await this.loadPersistedPendingStateIfNeeded(); + const owner = + expected === undefined + ? this.pendingStateOwner + : expected && this.pendingStateOwners.get(expected); + if (!owner || this.pendingStateOwner !== owner) return; + + // Clear this request's cache before yielding. A queued unlink must not later clear B's cache. + this.pendingStateBoundaryMessageId = undefined; + this.postCompactionAttachmentsPending = false; + this.cachedFileDiffs = []; + if (discard) { + this.cachedLoadedSkills = []; + this.cachedReadFilePaths = []; } - this.cachedLoadedSkills = []; - this.cachedReadFilePaths = []; + await this.deletePersistedPendingStateBestEffort(owner); } /** @@ -591,7 +609,10 @@ export class CompactionHandler { async discardPendingStateDurably(reason: string): Promise { await this.discardPendingState(reason); try { - await fsPromises.unlink(this.postCompactionStatePath); + await this.enqueuePendingStateWrite(async () => { + await fsPromises.unlink(this.postCompactionStatePath); + this.pendingStateFileOwner = undefined; + }); } catch (error) { if (error && typeof error === "object" && "code" in error && error.code === "ENOENT") { return; @@ -600,9 +621,13 @@ export class CompactionHandler { } } - private async deletePersistedPendingStateBestEffort(): Promise { + private async deletePersistedPendingStateBestEffort(owner?: symbol): Promise { try { - await fsPromises.unlink(this.postCompactionStatePath); + await this.enqueuePendingStateWrite(async () => { + if (owner && this.pendingStateFileOwner !== owner) return; + await fsPromises.unlink(this.postCompactionStatePath); + this.pendingStateFileOwner = undefined; + }); } catch { // ignore } @@ -650,9 +675,9 @@ export class CompactionHandler { boundaryMessageId?: string, previousState?: PersistedPostCompactionStateV1 ): Promise { + const owner = Symbol(); + this.pendingStateOwner = owner; try { - await fsPromises.mkdir(this.sessionDir, { recursive: true }); - for (const snapshot of loadedSkills) { assert(snapshot.name.trim().length > 0, "loaded skill snapshot name must not be empty"); } @@ -666,7 +691,13 @@ export class CompactionHandler { ...(boundaryMessageId && { boundaryMessageId, previousState }), }; - await fsPromises.writeFile(this.postCompactionStatePath, JSON.stringify(persisted)); + await this.enqueuePendingStateWrite(async () => { + // This write owns retirement even if it fails and leaves earlier bytes on disk. + // Queueing writes with consumption prevents a held unlink from deleting a later write. + this.pendingStateFileOwner = owner; + await fsPromises.mkdir(this.sessionDir, { recursive: true }); + await fsPromises.writeFile(this.postCompactionStatePath, JSON.stringify(persisted)); + }); } catch (error) { log.warn("Failed to persist post-compaction state", { workspaceId: this.workspaceId, From 95530cff7c6dce8f0eb6023cb62293ca4dcac70d Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 18:32:17 +0200 Subject: [PATCH 3/4] =?UTF-8?q?=F0=9F=A4=96=20fix:=20preserve=20pending=20?= =?UTF-8?q?ownership=20across=20compaction=20refusal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Retire provisional pending state when manual boundary publication fails, including read paths. Preserve the original captured owner when a continuous preparation rolls back, so acknowledgment and discard remain valid even if restoration persistence fails. Five regressions reproduce both findings before the fix. All 214 targeted tests and full static checks pass; independent review approved the 17-line production fix. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: I64df09b02161295313177bf6c3acf0b66873392f --- ...compactionHandler.pendingOwnership.test.ts | 97 ++++++++++++++++--- src/node/services/compactionHandler.ts | 17 +++- 2 files changed, 99 insertions(+), 15 deletions(-) diff --git a/src/node/services/compactionHandler.pendingOwnership.test.ts b/src/node/services/compactionHandler.pendingOwnership.test.ts index c58d990df49..7d4e46bed77 100644 --- a/src/node/services/compactionHandler.pendingOwnership.test.ts +++ b/src/node/services/compactionHandler.pendingOwnership.test.ts @@ -3,6 +3,7 @@ import { EventEmitter } from "events"; import * as fs from "fs/promises"; import * as path from "path"; import { createMuxMessage } from "@/common/types/message"; +import { Err } from "@/common/types/result"; import assert from "@/common/utils/assert"; import { CompactionHandler } from "./compactionHandler"; import { createTestHistoryService } from "./testHistoryService"; @@ -10,6 +11,21 @@ import { createTestHistoryService } from "./testHistoryService"; const workspaceId = "pending-consumers"; const followUp = { text: "wake", model: "openai:gpt-4o", agentId: "exec" }; +function readMessage(id: string) { + const message = createMuxMessage(id, "assistant", ""); + message.parts = [ + { + type: "dynamic-tool", + toolCallId: id, + toolName: "file_read", + state: "output-available", + input: { path: `/${id}.ts` }, + output: { success: true }, + }, + ]; + return message; +} + describe("exact pending snapshot consumption", () => { let store: Awaited>; let handler: CompactionHandler; @@ -39,18 +55,7 @@ describe("exact pending snapshot consumption", () => { async function publish(id?: string) { if (id) { - const message = createMuxMessage(id, "assistant", ""); - message.parts = [ - { - type: "dynamic-tool", - toolCallId: id, - toolName: "file_read", - state: "output-available", - input: { path: `/${id}.ts` }, - output: { success: true }, - }, - ]; - await store.historyService.appendToHistory(workspaceId, message); + await store.historyService.appendToHistory(workspaceId, readMessage(id)); } expect( ( @@ -179,4 +184,72 @@ describe("exact pending snapshot consumption", () => { expect((await handler.peekPendingState())?.readFiles).toContain("/b.ts"); expect((await restart().peekPendingState())?.readFiles).toContain("/b.ts"); }); + + it("a failed manual boundary cannot expose uncommitted successor state", async () => { + const consumed = await publish("a"); + await store.historyService.appendToHistory(workspaceId, readMessage("b")); + await store.historyService.appendToHistory( + workspaceId, + createMuxMessage("compact-b", "user", "compact", { + muxMetadata: { type: "compaction-request", rawCommand: "/compact", parsed: {} }, + }) + ); + spyOn(store.historyService, "appendToHistory").mockResolvedValueOnce(Err("B boundary failed")); + expect( + await handler.handleCompletion({ + type: "stream-end", + workspaceId, + messageId: "b-summary", + metadata: { model: followUp.model }, + parts: [{ type: "text", text: "B summary" }], + }) + ).toBe(false); + + const history = await store.historyService.getHistoryFromLatestBoundary(workspaceId); + assert(history.success, "Expected readable history after failed boundary"); + expect(history.data.some((message) => message.metadata?.compacted === "user")).toBe(false); + expect(await handler.peekPendingState()).toBeNull(); + await handler.ackPendingStateConsumed(consumed); + expect(await handler.peekPendingState()).toBeNull(); + expect(await restart().peekPendingState()).toBeNull(); + }); + + it.each( + (["ack", "discard"] as const).flatMap((action) => + [false, true].map((failedRestore) => ({ action, failedRestore })) + ) + )( + "request A can $action after continuous rollback (failed rewrite=$failedRestore)", + async ({ action, failedRestore }) => { + const consumed = await publish("a"); + let restore: ReturnType> | undefined; + const mkdir = fs.mkdir; + try { + expect( + await handler.withContinuousPendingState( + [readMessage("b")], + () => { + if (failedRestore) { + restore = spyOn(fs, "mkdir").mockImplementation((async ( + ...args: Parameters + ) => { + if (String(args[0]) === sessionDir) throw new Error("restore mkdir failed"); + return mkdir(...args); + }) as typeof fs.mkdir); + } + return Promise.resolve(false); + }, + "b" + ) + ).toBe(false); + } finally { + restore?.mockRestore(); + } + expect((await handler.peekPendingState())?.readFiles).toEqual(["/a.ts"]); + if (action === "ack") await handler.ackPendingStateConsumed(consumed); + else await handler.discardPendingState("context_exceeded", consumed); + expect(await handler.peekPendingState()).toBeNull(); + expect(await restart().peekPendingState()).toBeNull(); + } + ); }); diff --git a/src/node/services/compactionHandler.ts b/src/node/services/compactionHandler.ts index 31eb124777e..f323bcc9e8d 100644 --- a/src/node/services/compactionHandler.ts +++ b/src/node/services/compactionHandler.ts @@ -673,9 +673,9 @@ export class CompactionHandler { loadedSkills: LoadedSkillSnapshot[], readFiles: string[], boundaryMessageId?: string, - previousState?: PersistedPostCompactionStateV1 + previousState?: PersistedPostCompactionStateV1, + owner = Symbol() ): Promise { - const owner = Symbol(); this.pendingStateOwner = owner; try { for (const snapshot of loadedSkills) { @@ -749,6 +749,7 @@ export class CompactionHandler { ): Promise { await this.loadPersistedPendingStateIfNeeded(); const previous = { + owner: this.pendingStateOwner, pending: this.postCompactionAttachmentsPending, diffs: this.cachedFileDiffs, loadedSkills: this.cachedLoadedSkills, @@ -778,6 +779,8 @@ export class CompactionHandler { } finally { // Never roll an older apply back over a newer preparation/consumption. if (!applied && this.pendingStateBoundaryMessageId === boundaryMessageId) { + // Restoring A must preserve the authority already captured by A's request. + this.pendingStateOwner = previous.owner; this.postCompactionAttachmentsPending = previous.pending; this.cachedFileDiffs = previous.diffs; this.cachedLoadedSkills = previous.loadedSkills; @@ -788,7 +791,9 @@ export class CompactionHandler { previous.diffs, previous.loadedSkills, previous.readFiles, - previous.boundaryMessageId + previous.boundaryMessageId, + undefined, + previous.owner ); } else { await this.deletePersistedPendingStateBestEffort(); @@ -1508,8 +1513,14 @@ export class CompactionHandler { ? await this.historyService.updateHistory(this.workspaceId, summaryMessage) : await this.historyService.appendToHistory(this.workspaceId, summaryMessage); if (!persistenceResult.success) { + // No boundary committed: retire this provisional snapshot independently of an older + // request's acknowledgement, which correctly cannot consume its replacement owner. + this.pendingStateOwner = Symbol(); + this.pendingStateBoundaryMessageId = undefined; + this.postCompactionAttachmentsPending = false; this.cachedFileDiffs = []; this.cachedLoadedSkills = []; + this.cachedReadFilePaths = []; await this.deletePersistedPendingStateBestEffort(); const operation = preservedTailCopies.length > 0 From bfe44ab336152822b018aa6696327dcdd6fbe64d Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 19:03:08 +0200 Subject: [PATCH 4/4] =?UTF-8?q?=F0=9F=A4=96=20fix:=20retain=20heartbeat=20?= =?UTF-8?q?rollback=20snapshot=20ownership?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Capture the predecessor owner for heartbeat preparation and preserve it through failed append and contention rollback, including restoration write failure. Existing request acknowledgment/discard authority then survives restoration just as it does for continuous compaction. All restoration call sites audited. Eight regressions reproduced before fixing; 222 targeted tests, 17 dispatch tests and full static checks pass. Independent review approved the nine-line production fix. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Iaf0e58405faee72c8073bd0dad40cc5a3d50df7e --- ...compactionHandler.pendingOwnership.test.ts | 68 +++++++++++++++++++ src/node/services/compactionHandler.ts | 9 ++- 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/src/node/services/compactionHandler.pendingOwnership.test.ts b/src/node/services/compactionHandler.pendingOwnership.test.ts index 7d4e46bed77..117d9cc620a 100644 --- a/src/node/services/compactionHandler.pendingOwnership.test.ts +++ b/src/node/services/compactionHandler.pendingOwnership.test.ts @@ -252,4 +252,72 @@ describe("exact pending snapshot consumption", () => { expect(await restart().peekPendingState()).toBeNull(); } ); + + it.each( + (["failed append", "contention rollback"] as const).flatMap((outcome) => + (["ack", "discard"] as const).flatMap((action) => + [false, true].map((failedRestore) => ({ outcome, action, failedRestore })) + ) + ) + )( + "request A can $action after heartbeat $outcome (failed rewrite=$failedRestore)", + async ({ outcome, action, failedRestore }) => { + const consumed = await publish("a"); + await store.historyService.appendToHistory(workspaceId, readMessage("b")); + let restore: ReturnType> | undefined; + const mkdir = fs.mkdir; + function failRestoreIfRequested() { + if (!failedRestore) return; + restore = spyOn(fs, "mkdir").mockImplementation((async ( + ...args: Parameters + ) => { + if (String(args[0]) === sessionDir) throw new Error("heartbeat restore mkdir failed"); + return mkdir(...args); + }) as typeof fs.mkdir); + } + + try { + if (outcome === "failed append") { + spyOn(store.historyService, "appendToHistory").mockImplementationOnce(() => { + // B's provisional file already exists; only its restoration write should fail. + failRestoreIfRequested(); + return Promise.resolve(Err("heartbeat B append failed")); + }); + expect( + ( + await handler.appendHeartbeatContextResetBoundary({ + boundaryText: "B reset", + pendingFollowUp: followUp, + }) + ).success + ).toBe(false); + } else { + const boundary = await handler.appendHeartbeatContextResetBoundary({ + boundaryText: "B reset", + pendingFollowUp: followUp, + }); + assert(boundary.success, "Expected durable B before contention rollback"); + const rows = await store.historyService.getLastMessages(workspaceId, 1); + assert(rows.success, "Expected readable heartbeat boundary"); + const message = rows.data[0]; + assert(message?.id === boundary.data.summaryMessageId, "Expected B's durable row"); + failRestoreIfRequested(); + expect((await handler.rollbackHeartbeatContextResetBoundary(message)).success).toBe(true); + const history = await store.historyService.getLastMessages(workspaceId, 10); + assert(history.success, "Expected readable history after rollback"); + expect(history.data.some((row) => row.id === message.id)).toBe(false); + } + if (failedRestore) { + expect(restore?.mock.calls.some(([dir]) => String(dir) === sessionDir)).toBe(true); + } + } finally { + restore?.mockRestore(); + } + expect((await handler.peekPendingState())?.readFiles).toEqual(["/a.ts"]); + if (action === "ack") await handler.ackPendingStateConsumed(consumed); + else await handler.discardPendingState("context_exceeded", consumed); + expect(await handler.peekPendingState()).toBeNull(); + expect(await restart().peekPendingState()).toBeNull(); + } + ); }); diff --git a/src/node/services/compactionHandler.ts b/src/node/services/compactionHandler.ts index f323bcc9e8d..9dc0f3aa2ed 100644 --- a/src/node/services/compactionHandler.ts +++ b/src/node/services/compactionHandler.ts @@ -98,6 +98,7 @@ interface PersistedPostCompactionStateV1 { } interface HeartbeatResetRollbackState { + owner: symbol; postCompactionAttachmentsPending: boolean; cachedFileDiffs: FileEditDiff[]; cachedLoadedSkills: LoadedSkillSnapshot[]; @@ -635,6 +636,7 @@ export class CompactionHandler { private captureHeartbeatResetRollbackState(): void { this.heartbeatResetRollbackState = { + owner: this.pendingStateOwner, postCompactionAttachmentsPending: this.postCompactionAttachmentsPending, cachedFileDiffs: [...this.cachedFileDiffs], cachedLoadedSkills: [...this.cachedLoadedSkills], @@ -649,6 +651,8 @@ export class CompactionHandler { return; } + // Keep the original request's authority even if the best-effort restoration write fails. + this.pendingStateOwner = rollbackState.owner; this.postCompactionAttachmentsPending = rollbackState.postCompactionAttachmentsPending; this.cachedFileDiffs = [...rollbackState.cachedFileDiffs]; this.cachedLoadedSkills = [...rollbackState.cachedLoadedSkills]; @@ -659,7 +663,10 @@ export class CompactionHandler { await this.persistPendingStateBestEffort( this.cachedFileDiffs, this.cachedLoadedSkills, - this.cachedReadFilePaths + this.cachedReadFilePaths, + undefined, + undefined, + rollbackState.owner ); } else { await this.deletePersistedPendingStateBestEffort();