diff --git a/src/browser/features/RightSidebar/CodeReview/ReviewPanel.tsx b/src/browser/features/RightSidebar/CodeReview/ReviewPanel.tsx index 2d2d13a607..f2f23ccf11 100644 --- a/src/browser/features/RightSidebar/CodeReview/ReviewPanel.tsx +++ b/src/browser/features/RightSidebar/CodeReview/ReviewPanel.tsx @@ -1360,9 +1360,8 @@ export const ReviewPanel: React.FC = ({ if (!api || isCreating) return; let cancelled = false; - const prevRefreshTrigger = lastFileTreeRefreshTriggerRef.current; - lastFileTreeRefreshTriggerRef.current = refreshTrigger; - const isManualRefresh = refreshTrigger !== 0 && prevRefreshTrigger !== refreshTrigger; + const isManualRefresh = + refreshTrigger !== 0 && lastFileTreeRefreshTriggerRef.current !== refreshTrigger; const numstatCommand = buildGitDiffCommand( filters.diffBase, @@ -1486,6 +1485,7 @@ export const ReviewPanel: React.FC = ({ }); if (cancelled) return; + lastFileTreeRefreshTriggerRef.current = refreshTrigger; setFileTree(tree); } catch (err) { console.error("Failed to load file tree:", err); @@ -1519,9 +1519,8 @@ export const ReviewPanel: React.FC = ({ if (!api || isCreating) return; let cancelled = false; - const prevRefreshTrigger = lastDiffRefreshTriggerRef.current; - lastDiffRefreshTriggerRef.current = refreshTrigger; - const isManualRefresh = refreshTrigger !== 0 && prevRefreshTrigger !== refreshTrigger; + const isManualRefresh = + refreshTrigger !== 0 && lastDiffRefreshTriggerRef.current !== refreshTrigger; const effectiveIncludeUncommitted = getEffectiveReviewIncludeUncommitted({ assistedOnly: filters.assistedOnly, @@ -1652,6 +1651,7 @@ export const ReviewPanel: React.FC = ({ if (cancelled) return; + lastDiffRefreshTriggerRef.current = refreshTrigger; setDiagnosticInfo(data.diagnosticInfo); // Preserve object references for unchanged hunks to prevent unnecessary re-renders. diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 9906c1a910..30dd754601 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -2733,6 +2733,7 @@ export class AgentSession { // when the edit target is outside the active context window. const truncateTargetId = await this.getEditTruncateTargetId(editMessageId); + this.clearUsageState(); const truncateResult = await this.historyService.truncateAfterMessage( this.workspaceId, truncateTargetId @@ -3418,6 +3419,11 @@ export class AgentSession { }; } + /** Prevent cached usage from auto-compacting a rewritten context. */ + clearUsageState(): void { + this.lastUsageState = undefined; + } + /** * Persist a manual user message + emit a stream-error chat event when a * pre-stream gate (e.g. the unpriced-model budget gate) rejects a send. @@ -4583,6 +4589,7 @@ export class AgentSession { }); } + this.clearUsageState(); const clearResult = this.clearHistoryForHardRestart ? await this.clearHistoryForHardRestart({ monitorHistoryLockHeld: context.monitorHistoryLockHeld === true, @@ -5196,7 +5203,7 @@ export class AgentSession { // Compaction collapses history to a boundary summary, so prior context-usage snapshots // are stale. Clear them to prevent immediate re-trigger loops on the follow-up turn. - this.lastUsageState = undefined; + this.clearUsageState(); if (completedCompactionRequest?.source === "auto-compaction") { this.emitChatEvent({ @@ -6724,6 +6731,7 @@ export class AgentSession { pendingFollowUp: params.pendingFollowUp, }); if (result.success) { + this.clearUsageState(); this.onPostCompactionStateChange?.(); } return result; diff --git a/src/node/services/historyService.test.ts b/src/node/services/historyService.test.ts index 52d4ec3822..a550efa79a 100644 --- a/src/node/services/historyService.test.ts +++ b/src/node/services/historyService.test.ts @@ -1,12 +1,13 @@ import { describe, it, expect, beforeEach, afterEach } from "bun:test"; import { CONTEXT_BOUNDARY_KINDS } from "@/common/constants/contextBoundary"; import { HistoryService } from "./historyService"; -import { Config } from "@/node/config"; +import type { Config } from "@/node/config"; +import { createTestHistoryService } from "./testHistoryService"; 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 * as path from "path"; -import * as os from "os"; /** Collect all messages via iterateFullHistory (replaces removed getFullHistory). */ async function collectFullHistory(service: HistoryService, workspaceId: string) { @@ -48,25 +49,17 @@ async function appendNumberedMessages( describe("HistoryService", () => { let service: HistoryService; let config: Config; - let tempDir: string; + let cleanup: () => Promise; beforeEach(async () => { - // Create a temporary directory for test files - tempDir = path.join(os.tmpdir(), `mux-test-${Date.now()}-${Math.random()}`); - await fs.mkdir(tempDir, { recursive: true }); - - // Create a Config with the temp directory - config = new Config(tempDir); - service = new HistoryService(config); + const testService = await createTestHistoryService(); + service = testService.historyService; + config = testService.config; + cleanup = testService.cleanup; }); afterEach(async () => { - // Clean up temp directory - try { - await fs.rm(tempDir, { recursive: true, force: true }); - } catch { - // Ignore cleanup errors - } + await cleanup(); }); describe("getHistory", () => { @@ -1287,18 +1280,15 @@ describe("HistoryService", () => { const scanStarted = new Promise((resolve) => { markScanStarted = resolve; }); - const originalIterateFullHistory: HistoryService["iterateFullHistory"] = - service.iterateFullHistory.bind(service); - const blockingIterateFullHistory: HistoryService["iterateFullHistory"] = async ( - workspaceIdArg, - direction, - visitor - ) => { + const internal = service as unknown as { + iterateFullHistoryUnlocked: HistoryService["iterateFullHistory"]; + }; + const originalIterateFullHistory = internal.iterateFullHistoryUnlocked.bind(service); + internal.iterateFullHistoryUnlocked = async (workspaceIdArg, direction, visitor) => { markScanStarted(); await scanReleased; return originalIterateFullHistory(workspaceIdArg, direction, visitor); }; - service.iterateFullHistory = blockingIterateFullHistory; const scan = service.getMessagesForCompactionEpoch(workspaceId, { workspaceId, @@ -2037,6 +2027,200 @@ describe("HistoryService", () => { expect(await fs.readFile(archivePath(wsId), "utf-8")).toBe(archiveBefore); }); + it("does not reseed usage from before a partial prefix truncation", async () => { + await appendNumberedMessages(service, wsId, 8); + await service.appendToHistory( + wsId, + createMuxMessage("assistant-usage", "assistant", "reply", { + contextUsage: { inputTokens: 95_000, outputTokens: 100, totalTokens: 95_100 }, + contextProviderMetadata: { openai: {} }, + model: "openai:gpt-4o", + }) + ); + await service.appendToHistory( + wsId, + createMuxMessage("assistant-provider-metadata", "assistant", "reply", { + contextProviderMetadata: { openai: {} }, + model: "openai:gpt-4o", + }) + ); + + const truncateResult = await service.truncateHistory(wsId, 0.5); + expect(truncateResult.success).toBe(true); + + const restarted = new HistoryService(config); + const remaining = await restarted.getHistoryFromLatestBoundary(wsId); + expect(remaining.success).toBe(true); + if (remaining.success) { + const retainedAssistant = remaining.data.find( + (message) => message.id === "assistant-usage" + ); + expect(retainedAssistant).toBeDefined(); + expect(retainedAssistant?.metadata?.contextUsage).toBeUndefined(); + expect(retainedAssistant?.metadata?.contextProviderMetadata).toBeUndefined(); + const providerMetadataOnly = remaining.data.find( + (message) => message.id === "assistant-provider-metadata" + ); + expect(providerMetadataOnly).toBeDefined(); + expect(providerMetadataOnly?.metadata?.contextProviderMetadata).toBeUndefined(); + } + }); + + async function expectWorkflowDisplayTruncationPreservesUsage(withResetBoundary: boolean) { + if (withResetBoundary) { + await appendNumberedMessages(service, wsId, 12); + await service.appendToHistory( + wsId, + createMuxMessage("reset-boundary", "assistant", "", { + contextBoundaryKind: CONTEXT_BOUNDARY_KINDS.RESET, + }) + ); + } + await service.appendToHistory( + wsId, + createMuxMessage( + "workflow-display", + "user", + `workflow trigger display ${"x".repeat(2_000)}`, + { muxMetadata: { type: "workflow-trigger-display", rawCommand: "/wf", runId: "run-1" } } + ) + ); + await service.appendToHistory(wsId, createMuxMessage("user-active", "user", "prompt")); + await service.appendToHistory( + wsId, + createMuxMessage("assistant-active", "assistant", "active reply", { + contextUsage: { inputTokens: 95_000, outputTokens: 100, totalTokens: 95_100 }, + model: "openai:gpt-4o", + }) + ); + + expect((await service.truncateHistory(wsId, 0.5)).success).toBe(true); + + const active = await service.getHistoryFromLatestBoundary(wsId); + expect(active.success).toBe(true); + if (active.success) { + expect(active.data.find((message) => message.id === "workflow-display")).toBeUndefined(); + const retainedAssistant = active.data.find((message) => message.id === "assistant-active"); + expect(retainedAssistant).toBeDefined(); + expect(retainedAssistant?.metadata?.contextUsage).toMatchObject({ inputTokens: 95_000 }); + } + } + + it("preserves active usage when uncompacted truncation removes only workflow display rows", () => + expectWorkflowDisplayTruncationPreservesUsage(false)); + + it("preserves active usage when truncation removes only workflow display rows", () => + expectWorkflowDisplayTruncationPreservesUsage(true)); + + it("preserves active usage when truncation removes only sealed rows", async () => { + await appendNumberedMessages(service, wsId, 8); + await service.appendToHistory(wsId, boundaryMessage("boundary-1", 1)); + await service.appendToHistory( + wsId, + createMuxMessage("active-usage", "assistant", "reply", { + contextUsage: { inputTokens: 95_000, outputTokens: 100, totalTokens: 95_100 }, + model: "openai:gpt-4o", + }) + ); + + expect((await service.truncateHistory(wsId, 0.2)).success).toBe(true); + + const active = await service.getHistoryFromLatestBoundary(wsId); + expect(active.success).toBe(true); + if (active.success) { + expect( + active.data.find((message) => message.id === "active-usage")?.metadata?.contextUsage + ).toBeDefined(); + } + }); + + it("restores a markerless archive tombstone left by an older truncation", async () => { + await appendNumberedMessages(service, wsId, 3); + await service.appendToHistory(wsId, boundaryMessage("boundary-1", 1)); + await service.appendToHistory(wsId, createMuxMessage("post-0", "user", "after")); + await fs.rename(archivePath(wsId), `${archivePath(wsId)}.truncate`); + + const restarted = new HistoryService(config); + const full = await collectFullHistory(restarted, wsId); + expect(full.map((message) => message.id)).toEqual([ + "msg-0", + "msg-1", + "msg-2", + "boundary-1", + "post-0", + ]); + }); + + it("restores an interrupted archive tombstone when only the final chat matches", async () => { + await appendNumberedMessages(service, wsId, 3); + await service.appendToHistory(wsId, boundaryMessage("boundary-1", 1)); + await service.appendToHistory(wsId, createMuxMessage("post-0", "user", "after")); + const chatContents = await fs.readFile(chatPath(wsId), "utf-8"); + const hash = (contents: string) => createHash("sha256").update(contents).digest("hex"); + await fs.writeFile( + `${archivePath(wsId)}.truncate.json`, + JSON.stringify({ + phase: "prepared", + finalArchiveHash: hash("replacement archive\n"), + finalChatHash: hash(chatContents), + }) + ); + await fs.rename(archivePath(wsId), `${archivePath(wsId)}.truncate`); + + const restarted = new HistoryService(config); + const full = await collectFullHistory(restarted, wsId); + expect(full.map((message) => message.id)).toEqual([ + "msg-0", + "msg-1", + "msg-2", + "boundary-1", + "post-0", + ]); + }); + + it("recovers the source transaction before copying a fork snapshot", async () => { + const targetWorkspaceId = "forked-workspace"; + await appendNumberedMessages(service, wsId, 3); + await service.appendToHistory(wsId, boundaryMessage("boundary-1", 1)); + await service.appendToHistory(wsId, createMuxMessage("post-0", "user", "after")); + const chatContents = await fs.readFile(chatPath(wsId), "utf-8"); + const hash = (contents: string) => createHash("sha256").update(contents).digest("hex"); + await fs.writeFile( + `${archivePath(wsId)}.truncate.json`, + JSON.stringify({ + finalArchiveHash: hash("replacement archive\n"), + finalChatHash: hash(chatContents), + }) + ); + await fs.rename(archivePath(wsId), `${archivePath(wsId)}.truncate`); + + const result = await service.copyHistorySnapshotToNewWorkspace(wsId, targetWorkspaceId); + expect(result.success).toBe(true); + expect( + (await collectFullHistory(service, targetWorkspaceId)).map((message) => message.id) + ).toEqual(["msg-0", "msg-1", "msg-2", "boundary-1", "post-0"]); + }); + + it("does not restore a committed archive tombstone before appending", async () => { + await appendNumberedMessages(service, wsId, 3); + await service.appendToHistory(wsId, boundaryMessage("boundary-1", 1)); + await service.appendToHistory(wsId, createMuxMessage("post-0", "user", "after")); + await fs.writeFile( + `${archivePath(wsId)}.truncate.json`, + JSON.stringify({ finalArchiveHash: null, finalChatHash: null }) + ); + await fs.rename(archivePath(wsId), `${archivePath(wsId)}.truncate`); + await fs.rm(chatPath(wsId)); + + const restarted = new HistoryService(config); + const message = createMuxMessage("new-msg", "user", "fresh"); + expect((await restarted.appendToHistory(wsId, message)).success).toBe(true); + expect(message.metadata?.historySequence).toBe(0); + expect((await collectFullHistory(restarted, wsId)).map((item) => item.id)).toEqual([ + "new-msg", + ]); + }); + it("hasHistory sees archive-only workspaces", async () => { await appendNumberedMessages(service, wsId, 1); await service.appendToHistory(wsId, boundaryMessage("boundary-1", 1)); diff --git a/src/node/services/historyService.ts b/src/node/services/historyService.ts index df71938525..8266b07a2d 100644 --- a/src/node/services/historyService.ts +++ b/src/node/services/historyService.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import * as fs from "fs/promises"; import * as path from "path"; import writeFileAtomic from "write-file-atomic"; @@ -20,9 +21,13 @@ 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, isDurableContextBoundaryMarker, } from "@/common/utils/messages/compactionBoundary"; +import { filterWorkflowDisplayOnlyMessages } from "@/common/utils/workflowRunMessages"; import { CHAT_FILE_NAME, CHAT_ARCHIVE_FILE_NAME } from "@/common/constants/paths"; import { isRefusalFinishReason } from "@/common/utils/messages/refusalFinishReason"; import { getErrorMessage } from "@/common/utils/errors"; @@ -41,6 +46,33 @@ 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 stripContextUsage(message: MuxMessage): MuxMessage { + if (!message.metadata) { + return message; + } + return { + ...message, + metadata: { + ...message.metadata, + contextUsage: undefined, + contextProviderMetadata: undefined, + }, + }; +} + function getCompactionMetadataToPreserve( workspaceId: string, existingMessage: MuxMessage, @@ -171,6 +203,217 @@ export class HistoryService { return path.join(this.config.getSessionDir(workspaceId), this.CHAT_ARCHIVE_FILE); } + private getTruncateTransactionPath(workspaceId: string): string { + return `${this.getChatArchivePath(workspaceId)}.truncate.json`; + } + + private async readExistingFile(filePath: string): Promise { + try { + return await fs.readFile(filePath, "utf-8"); + } catch (error) { + if (error && typeof error === "object" && "code" in error && error.code === "ENOENT") { + return null; + } + throw error; + } + } + + private historyContentsHash(contents: string): string { + return createHash("sha256").update(contents).digest("hex"); + } + + private parseTruncateTransaction(contents: string): { + finalArchiveHash: string | null; + finalChatHash: string | null; + } | null { + try { + const parsed: unknown = JSON.parse(contents); + if (parsed === null || typeof parsed !== "object") { + return null; + } + const marker = parsed as Record; + const finalArchiveHash = marker.finalArchiveHash; + const finalChatHash = marker.finalChatHash; + if ( + (finalArchiveHash !== null && typeof finalArchiveHash !== "string") || + (finalChatHash !== null && typeof finalChatHash !== "string") + ) { + return null; + } + return { finalArchiveHash, finalChatHash }; + } catch { + return null; + } + } + + private historyContentsMatch(contents: string | null, hash: string | null): boolean { + return hash === null + ? contents === null + : contents !== null && this.historyContentsHash(contents) === hash; + } + + private async recoverTruncateTransactionUnlocked(workspaceId: string): Promise { + const archivePath = this.getChatArchivePath(workspaceId); + const archiveTombstonePath = `${archivePath}.truncate`; + const tombstoneExists = await fs.stat(archiveTombstonePath).then( + () => true, + (error: unknown) => { + if (error && typeof error === "object" && "code" in error && error.code === "ENOENT") { + return false; + } + throw error; + } + ); + const markerPath = this.getTruncateTransactionPath(workspaceId); + const markerContents = await this.readExistingFile(markerPath); + if (markerContents === null) { + if (!tombstoneExists) { + return false; + } + const archiveExists = (await this.readExistingFile(archivePath)) !== null; + if (archiveExists) { + await fs.rm(archiveTombstonePath); + } else { + await fs.rename(archiveTombstonePath, archivePath); + } + return false; + } + + const marker = this.parseTruncateTransaction(markerContents); + if (!tombstoneExists) { + await fs.rm(markerPath, { force: true }); + if (marker === null) { + return false; + } + const archiveContents = await this.readExistingFile(archivePath); + const chatContents = await this.readExistingFile(this.getChatHistoryPath(workspaceId)); + return ( + this.historyContentsMatch(archiveContents, marker.finalArchiveHash) && + this.historyContentsMatch(chatContents, marker.finalChatHash) + ); + } + + if (marker !== null) { + const archiveContents = await this.readExistingFile(archivePath); + const chatContents = await this.readExistingFile(this.getChatHistoryPath(workspaceId)); + const committed = + this.historyContentsMatch(archiveContents, marker.finalArchiveHash) && + this.historyContentsMatch(chatContents, marker.finalChatHash); + if (committed) { + await fs.rm(archiveTombstonePath); + await fs.rm(markerPath, { force: true }); + return true; + } + } + + await fs.rm(archivePath, { force: true }); + await fs.rename(archiveTombstonePath, archivePath); + await fs.rm(markerPath, { force: true }); + return false; + } + + private async withRecoveredHistoryLock( + workspaceId: string, + operation: () => Promise + ): Promise { + return this.fileLocks.withLock(workspaceId, async () => { + await this.recoverTruncateTransactionUnlocked(workspaceId); + return operation(); + }); + } + + private async withRecoveredHistoryResultLock( + workspaceId: string, + errorPrefix: string, + operation: () => Promise> + ): Promise> { + try { + return await this.withRecoveredHistoryLock(workspaceId, operation); + } catch (error) { + return Err(`${errorPrefix}: ${getErrorMessage(error)}`); + } + } + + private async rewriteHistoryFilesUnlocked( + workspaceId: string, + finalArchiveContents: string | null, + finalChatContents: string | null + ): Promise { + const archivePath = this.getChatArchivePath(workspaceId); + const archiveTombstonePath = `${archivePath}.truncate`; + const markerPath = this.getTruncateTransactionPath(workspaceId); + const archiveExists = await fs.stat(archivePath).then( + () => true, + (error: unknown) => { + if (error && typeof error === "object" && "code" in error && error.code === "ENOENT") { + return false; + } + throw error; + } + ); + if (!archiveExists) { + assert(finalArchiveContents === null, "cannot replace a missing history archive"); + if (finalChatContents === null) { + await fs.rm(this.getChatHistoryPath(workspaceId), { force: true }); + } else { + await writeFileAtomic(this.getChatHistoryPath(workspaceId), finalChatContents); + } + return; + } + + await writeFileAtomic( + markerPath, + JSON.stringify({ + finalArchiveHash: + finalArchiveContents === null ? null : this.historyContentsHash(finalArchiveContents), + finalChatHash: + finalChatContents === null ? null : this.historyContentsHash(finalChatContents), + }) + ); + try { + await fs.rename(archivePath, archiveTombstonePath); + } catch (error) { + await fs.rm(markerPath, { force: true }); + throw error; + } + + try { + if (finalArchiveContents !== null) { + await writeFileAtomic(archivePath, finalArchiveContents); + } + if (finalChatContents === null) { + await fs.rm(this.getChatHistoryPath(workspaceId), { force: true }); + } else { + await writeFileAtomic(this.getChatHistoryPath(workspaceId), finalChatContents); + } + } catch (error) { + let committed = false; + try { + committed = await this.recoverTruncateTransactionUnlocked(workspaceId); + } catch (recoveryError) { + log.error("Failed to recover history truncation after write failure", { + workspaceId, + error: recoveryError, + }); + } + if (!committed) { + throw error; + } + return; + } + + try { + await fs.rm(archiveTombstonePath); + await fs.rm(markerPath, { force: true }); + } catch (error) { + this.sealedRotationChecked.delete(workspaceId); + log.warn("History truncation cleanup deferred to the next operation", { + workspaceId, + error, + }); + } + } + private getPartialPath(workspaceId: string): string { return path.join(this.config.getSessionDir(workspaceId), this.PARTIAL_FILE); } @@ -695,6 +938,69 @@ export class HistoryService { workspaceId: string, direction: "forward" | "backward", visitor: (messages: MuxMessage[]) => boolean | void | Promise + ): Promise> { + return this.withRecoveredHistoryResultLock(workspaceId, "Failed to iterate history", () => + this.iterateFullHistoryUnlocked(workspaceId, direction, visitor) + ); + } + + /** Call only while holding workspaceFileLocks for this workspace. */ + async iterateFullHistoryUnderLock( + workspaceId: string, + direction: "forward" | "backward", + visitor: (messages: MuxMessage[]) => boolean | void | Promise + ): Promise> { + try { + await this.recoverTruncateTransactionUnlocked(workspaceId); + return await this.iterateFullHistoryUnlocked(workspaceId, direction, visitor); + } catch (error) { + return Err(`Failed to iterate history: ${getErrorMessage(error)}`); + } + } + + async copyHistorySnapshotToNewWorkspace( + sourceWorkspaceId: string, + targetWorkspaceId: string + ): Promise> { + assert( + sourceWorkspaceId !== targetWorkspaceId, + "history snapshot target must be a new workspace" + ); + const snapshot = await this.withRecoveredHistoryResultLock( + sourceWorkspaceId, + "Failed to read history snapshot", + async () => + Ok({ + archive: await this.readExistingFile(this.getChatArchivePath(sourceWorkspaceId)), + chat: await this.readExistingFile(this.getChatHistoryPath(sourceWorkspaceId)), + }) + ); + if (!snapshot.success) { + return snapshot; + } + + try { + await ensurePrivateDir(this.config.getSessionDir(targetWorkspaceId)); + for (const [targetPath, contents] of [ + [this.getChatArchivePath(targetWorkspaceId), snapshot.data.archive], + [this.getChatHistoryPath(targetWorkspaceId), snapshot.data.chat], + ] as const) { + if (contents === null) { + await fs.rm(targetPath, { force: true }); + } else { + await writeFileAtomic(targetPath, contents); + } + } + return Ok(undefined); + } catch (error) { + return Err(`Failed to copy history snapshot: ${getErrorMessage(error)}`); + } + } + + private async iterateFullHistoryUnlocked( + workspaceId: string, + direction: "forward" | "backward", + visitor: (messages: MuxMessage[]) => boolean | void | Promise ): Promise> { const chatPath = this.getChatHistoryPath(workspaceId); const archivePath = this.getChatArchivePath(workspaceId); @@ -801,6 +1107,15 @@ export class HistoryService { "hasHistoryBeforeSequence requires a non-negative integer" ); + return this.withRecoveredHistoryLock(workspaceId, () => + this.hasHistoryBeforeSequenceUnlocked(workspaceId, beforeHistorySequence) + ); + } + + private async hasHistoryBeforeSequenceUnlocked( + workspaceId: string, + beforeHistorySequence: number + ): Promise { let hasOlder = false; const visitor = (messages: MuxMessage[]): boolean | void => { for (const message of messages) { @@ -816,8 +1131,6 @@ export class HistoryService { } }; - // Newest rows live in chat.jsonl; continue into the sealed archive only - // when the active file has no older rows. const completed = await this.iterateBackward(this.getChatHistoryPath(workspaceId), visitor); if (completed && !hasOlder) { await this.iterateBackward(this.getChatArchivePath(workspaceId), visitor); @@ -845,7 +1158,7 @@ export class HistoryService { "getHistoryBoundaryWindow requires beforeHistorySequence to be a non-negative integer" ); - try { + const operation = async (): Promise> => { // Scan boundaries newest→oldest and pick the first window that has rows older // than the cursor. Boundaries newer than the rotation point live in chat.jsonl; // older ones live in the sealed archive. @@ -875,7 +1188,10 @@ export class HistoryService { "window messages filtered by historySequence must include a sequence" ); - const hasOlder = await this.hasHistoryBeforeSequence(workspaceId, oldestWindowSequence); + const hasOlder = await this.hasHistoryBeforeSequenceUnlocked( + workspaceId, + oldestWindowSequence + ); return Ok({ messages: windowMessages, hasOlder }); } } @@ -900,8 +1216,15 @@ export class HistoryService { "pre-boundary messages filtered by historySequence must include a sequence" ); - const hasOlder = await this.hasHistoryBeforeSequence(workspaceId, oldestWindowSequence); + const hasOlder = await this.hasHistoryBeforeSequenceUnlocked( + workspaceId, + oldestWindowSequence + ); return Ok({ messages: preBoundaryMessages, hasOlder }); + }; + + try { + return await this.withRecoveredHistoryLock(workspaceId, operation); } catch (error) { const message = getErrorMessage(error); return Err(`Failed to read history boundary window: ${message}`); @@ -934,8 +1257,8 @@ export class HistoryService { // The just-compacted epoch can straddle chat-archive.jsonl and chat.jsonl after // sealed-history rotation, so scan the full logical history under the workspace // lock; otherwise a concurrent boundary rotation can move rows between files mid-scan. - const iteration = await this.fileLocks.withLock(workspaceId, () => - this.iterateFullHistory(workspaceId, "forward", (chunk) => { + const iteration = await this.withRecoveredHistoryLock(workspaceId, () => + this.iterateFullHistoryUnlocked(workspaceId, "forward", (chunk) => { for (const message of chunk) { const sequence = message.metadata?.historySequence; if (!isNonNegativeInteger(sequence)) continue; @@ -986,10 +1309,10 @@ export class HistoryService { * that only needs the active compaction epoch. */ async getHistoryFromLatestBoundary(workspaceId: string, skip = 0): Promise> { - try { + const operation = async (): Promise> => { // One-time lazy migration: seal any pre-boundary prefix left in chat.jsonl // by older builds so this read (and every later one) stays O(active epoch). - await this.ensureSealedHistoryRotated(workspaceId); + await this.ensureSealedHistoryRotatedUnlocked(workspaceId); const chatPath = this.getChatHistoryPath(workspaceId); const archivePath = this.getChatArchivePath(workspaceId); @@ -1030,6 +1353,10 @@ export class HistoryService { const archived = await this.readArchivedHistory(workspaceId); const active = await this.readChatHistory(workspaceId); return Ok([...archived, ...active]); + }; + + try { + return await this.withRecoveredHistoryLock(workspaceId, operation); } catch (error) { const message = getErrorMessage(error); return Err(`Failed to read history from boundary: ${message}`); @@ -1050,22 +1377,19 @@ export class HistoryService { * lazily migrates files produced before rotation existed (or by crashes * between boundary write and rotation). */ - private async ensureSealedHistoryRotated(workspaceId: string): Promise { + private async ensureSealedHistoryRotatedUnlocked(workspaceId: string): Promise { if (this.sealedRotationChecked.has(workspaceId)) { return; } - this.sealedRotationChecked.add(workspaceId); try { - // Cheap unlocked probe first so the common no-op case takes no lock. const offset = await this.findLastBoundaryByteOffset(this.getChatHistoryPath(workspaceId)); - if (offset === null || offset === 0) { - return; + if (offset !== null && offset !== 0) { + await this.rotateSealedHistoryUnlocked(workspaceId); } - await this.fileLocks.withLock(workspaceId, () => - this.rotateSealedHistoryUnlocked(workspaceId) - ); + this.sealedRotationChecked.add(workspaceId); } catch (error) { + this.sealedRotationChecked.delete(workspaceId); // Rotation is an optimization — reads remain correct on unrotated files. log.warn("Failed to rotate sealed chat history", { workspaceId, @@ -1145,20 +1469,29 @@ export class HistoryService { * Continues into the sealed archive when the active epoch has fewer than N rows. */ async getLastMessages(workspaceId: string, n: number): Promise> { - try { - const messages = await this.readLastMessagesFromFile(this.getChatHistoryPath(workspaceId), n); - if (messages.length < n) { - const archived = await this.readLastMessagesFromFile( - this.getChatArchivePath(workspaceId), - n - messages.length - ); - return Ok([...archived, ...messages]); + return this.withRecoveredHistoryResultLock( + workspaceId, + `Failed to read last ${n} messages`, + async () => { + try { + const messages = await this.readLastMessagesFromFile( + this.getChatHistoryPath(workspaceId), + n + ); + if (messages.length < n) { + const archived = await this.readLastMessagesFromFile( + this.getChatArchivePath(workspaceId), + n - messages.length + ); + return Ok([...archived, ...messages]); + } + return Ok(messages); + } catch (error) { + const message = getErrorMessage(error); + return Err(`Failed to read last ${n} messages: ${message}`); + } } - return Ok(messages); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to read last ${n} messages: ${message}`); - } + ); } /** @@ -1166,20 +1499,22 @@ export class HistoryService { * Much cheaper than iterateFullHistory() when only an emptiness check is needed. */ async hasHistory(workspaceId: string): Promise { - for (const filePath of [ - this.getChatHistoryPath(workspaceId), - this.getChatArchivePath(workspaceId), - ]) { - try { - const stat = await fs.stat(filePath); - if (stat.size > 0) { - return true; + return this.withRecoveredHistoryLock(workspaceId, async () => { + for (const filePath of [ + this.getChatHistoryPath(workspaceId), + this.getChatArchivePath(workspaceId), + ]) { + try { + const stat = await fs.stat(filePath); + if (stat.size > 0) { + return true; + } + } catch { + // Missing file — keep checking. } - } catch { - // Missing file — keep checking. } - } - return false; + return false; + }); } /** @@ -1524,15 +1859,19 @@ export class HistoryService { } async appendToHistory(workspaceId: string, message: MuxMessage): Promise> { - return this.fileLocks.withLock(workspaceId, async () => { - const result = await this._appendToHistoryUnlocked(workspaceId, message); - if (result.success) { - // A new durable boundary seals the previous epoch — rotate it out of - // chat.jsonl so subsequent reads/rewrites stay O(active epoch). - await this.rotateAfterBoundaryWriteUnlocked(workspaceId, message); + return this.withRecoveredHistoryResultLock( + workspaceId, + "Failed to append history", + async () => { + const result = await this._appendToHistoryUnlocked(workspaceId, message); + if (result.success) { + // A new durable boundary seals the previous epoch — rotate it out of + // chat.jsonl so subsequent reads/rewrites stay O(active epoch). + await this.rotateAfterBoundaryWriteUnlocked(workspaceId, message); + } + return result; } - return result; - }); + ); } /** @@ -1544,76 +1883,80 @@ export class HistoryService { * never in the sealed archive. */ async updateHistory(workspaceId: string, message: MuxMessage): Promise> { - return this.fileLocks.withLock(workspaceId, async () => { - try { - const historyPath = this.getChatHistoryPath(workspaceId); - - // Read the active epoch — structural rewrite requires full file content - const messages = await this.readChatHistory(workspaceId); - const targetSequence = message.metadata?.historySequence; + return this.withRecoveredHistoryResultLock( + workspaceId, + "Failed to update history", + async () => { + try { + const historyPath = this.getChatHistoryPath(workspaceId); - if (targetSequence === undefined) { - return Err("Cannot update message without historySequence"); - } + // Read the active epoch — structural rewrite requires full file content + const messages = await this.readChatHistory(workspaceId); + const targetSequence = message.metadata?.historySequence; - assert( - isNonNegativeInteger(targetSequence), - "updateHistory requires historySequence to be a non-negative integer" - ); + if (targetSequence === undefined) { + return Err("Cannot update message without historySequence"); + } - // Find and replace the message with matching historySequence - let found = false; - let persistedMessage: MuxMessage | undefined; - for (let i = 0; i < messages.length; i++) { - if (messages[i].metadata?.historySequence === targetSequence) { - const existingMessage = messages[i]; - assert(existingMessage, "updateHistory matched message must exist"); - - // Preserve compaction boundary metadata during late in-place rewrites. - // Compaction may update an assistant row first, then a late stream rewrite can - // update that same historySequence and accidentally drop compaction markers. - const preservedCompactionMetadata = getCompactionMetadataToPreserve( - workspaceId, - existingMessage, - message - ); + assert( + isNonNegativeInteger(targetSequence), + "updateHistory requires historySequence to be a non-negative integer" + ); - // Preserve the historySequence, update everything else. - messages[i] = { - ...message, - metadata: { - ...message.metadata, - ...(preservedCompactionMetadata ?? {}), - historySequence: targetSequence, - }, - }; - persistedMessage = messages[i]; - found = true; - break; + // Find and replace the message with matching historySequence + let found = false; + let persistedMessage: MuxMessage | undefined; + for (let i = 0; i < messages.length; i++) { + if (messages[i].metadata?.historySequence === targetSequence) { + const existingMessage = messages[i]; + assert(existingMessage, "updateHistory matched message must exist"); + + // Preserve compaction boundary metadata during late in-place rewrites. + // Compaction may update an assistant row first, then a late stream rewrite can + // update that same historySequence and accidentally drop compaction markers. + const preservedCompactionMetadata = getCompactionMetadataToPreserve( + workspaceId, + existingMessage, + message + ); + + // Preserve the historySequence, update everything else. + messages[i] = { + ...message, + metadata: { + ...message.metadata, + ...(preservedCompactionMetadata ?? {}), + historySequence: targetSequence, + }, + }; + persistedMessage = messages[i]; + found = true; + break; + } } - } - if (!found || !persistedMessage) { - return Err(`No message found with historySequence ${targetSequence}`); - } + if (!found || !persistedMessage) { + return Err(`No message found with historySequence ${targetSequence}`); + } - // Rewrite entire file - const historyEntries = this.serializeHistoryEntries(messages, workspaceId); + // Rewrite entire file + const historyEntries = this.serializeHistoryEntries(messages, workspaceId); - // Atomic write prevents corruption if app crashes mid-write - await writeFileAtomic(historyPath, historyEntries); + // Atomic write prevents corruption if app crashes mid-write + await writeFileAtomic(historyPath, historyEntries); - // Compaction updates the streamed summary row in-place with boundary - // metadata — seal the previous epoch once that lands. Check the persisted - // row (not the incoming message) so preserved boundary metadata counts. - await this.rotateAfterBoundaryWriteUnlocked(workspaceId, persistedMessage); + // Compaction updates the streamed summary row in-place with boundary + // metadata — seal the previous epoch once that lands. Check the persisted + // row (not the incoming message) so preserved boundary metadata counts. + await this.rotateAfterBoundaryWriteUnlocked(workspaceId, persistedMessage); - return Ok(undefined); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to update history: ${message}`); + return Ok(undefined); + } catch (error) { + const message = getErrorMessage(error); + return Err(`Failed to update history: ${message}`); + } } - }); + ); } /** @@ -1625,55 +1968,59 @@ export class HistoryService { const ids = new Set(messageIds); assert(ids.size === messageIds.length, "deleteMessages requires unique message IDs"); - return this.fileLocks.withLock(workspaceId, async () => { - try { - const messages = await this.readChatHistory(workspaceId); - const foundIds = new Set( - messages.filter((message) => ids.has(message.id)).map((message) => message.id) - ); - const missingIds = messageIds.filter((messageId) => !foundIds.has(messageId)); - if (missingIds.length > 0) { - return Err(`Messages not found in active history: ${missingIds.join(", ")}`); - } + return this.withRecoveredHistoryResultLock( + workspaceId, + "Failed to delete messages", + async () => { + try { + const messages = await this.readChatHistory(workspaceId); + const foundIds = new Set( + messages.filter((message) => ids.has(message.id)).map((message) => message.id) + ); + const missingIds = messageIds.filter((messageId) => !foundIds.has(messageId)); + if (missingIds.length > 0) { + return Err(`Messages not found in active history: ${missingIds.join(", ")}`); + } - const filteredMessages = messages.filter((message) => !ids.has(message.id)); - await writeFileAtomic( - this.getChatHistoryPath(workspaceId), - this.serializeHistoryEntries(filteredMessages, workspaceId) - ); + const filteredMessages = messages.filter((message) => !ids.has(message.id)); + await writeFileAtomic( + this.getChatHistoryPath(workspaceId), + this.serializeHistoryEntries(filteredMessages, workspaceId) + ); - const maxSeq = filteredMessages.reduce((max, message) => { - const sequence = message.metadata?.historySequence; - if (sequence === undefined) return max; - if (!isNonNegativeInteger(sequence)) { - log.warn( - "Ignoring malformed persisted historySequence while updating sequence counter after batch delete", - { - workspaceId, - messageId: message.id, - historySequence: sequence, - } - ); - return max; + const maxSeq = filteredMessages.reduce((max, message) => { + const sequence = message.metadata?.historySequence; + if (sequence === undefined) return max; + if (!isNonNegativeInteger(sequence)) { + log.warn( + "Ignoring malformed persisted historySequence while updating sequence counter after batch delete", + { + workspaceId, + messageId: message.id, + historySequence: sequence, + } + ); + return max; + } + return sequence > max ? sequence : max; + }, -1); + const archiveMaxSeq = await this.getArchiveTailMaxSequence(workspaceId); + const nextSeq = Math.max(maxSeq, archiveMaxSeq) + 1; + assert( + isNonNegativeInteger(nextSeq), + "next history sequence counter after batch delete must be a non-negative integer" + ); + const currentCounter = this.sequenceCounters.get(workspaceId); + if (currentCounter === undefined || currentCounter < nextSeq) { + this.sequenceCounters.set(workspaceId, nextSeq); } - return sequence > max ? sequence : max; - }, -1); - const archiveMaxSeq = await this.getArchiveTailMaxSequence(workspaceId); - const nextSeq = Math.max(maxSeq, archiveMaxSeq) + 1; - assert( - isNonNegativeInteger(nextSeq), - "next history sequence counter after batch delete must be a non-negative integer" - ); - const currentCounter = this.sequenceCounters.get(workspaceId); - if (currentCounter === undefined || currentCounter < nextSeq) { - this.sequenceCounters.set(workspaceId, nextSeq); - } - return Ok(undefined); - } catch (error) { - return Err(`Failed to delete messages: ${getErrorMessage(error)}`); + return Ok(undefined); + } catch (error) { + return Err(`Failed to delete messages: ${getErrorMessage(error)}`); + } } - }); + ); } /** @@ -1683,79 +2030,83 @@ export class HistoryService { * messages may already have been appended. */ async deleteMessage(workspaceId: string, messageId: string): Promise> { - return this.fileLocks.withLock(workspaceId, async () => { - try { - // Structural rewrite requires full file content - const messages = await this.readChatHistory(workspaceId); - const filteredMessages = messages.filter((msg) => msg.id !== messageId); - - if (filteredMessages.length === messages.length) { - // Not in the active epoch — the row may live in the sealed archive - // (rare: cleanup paths almost always target recent rows). - const archiveMessages = await this.readArchivedHistory(workspaceId); - const filteredArchive = archiveMessages.filter((msg) => msg.id !== messageId); - if (filteredArchive.length === archiveMessages.length) { - return Err(`Message with ID ${messageId} not found in history`); + return this.withRecoveredHistoryResultLock( + workspaceId, + "Failed to delete message", + async () => { + try { + // Structural rewrite requires full file content + const messages = await this.readChatHistory(workspaceId); + const filteredMessages = messages.filter((msg) => msg.id !== messageId); + + if (filteredMessages.length === messages.length) { + // Not in the active epoch — the row may live in the sealed archive + // (rare: cleanup paths almost always target recent rows). + const archiveMessages = await this.readArchivedHistory(workspaceId); + const filteredArchive = archiveMessages.filter((msg) => msg.id !== messageId); + if (filteredArchive.length === archiveMessages.length) { + return Err(`Message with ID ${messageId} not found in history`); + } + + // Archived rows are strictly older than active rows, so deleting one + // can never affect the sequence counter. + await writeFileAtomic( + this.getChatArchivePath(workspaceId), + this.serializeHistoryEntries(filteredArchive, workspaceId) + ); + return Ok(undefined); } - // Archived rows are strictly older than active rows, so deleting one - // can never affect the sequence counter. - await writeFileAtomic( - this.getChatArchivePath(workspaceId), - this.serializeHistoryEntries(filteredArchive, workspaceId) - ); - return Ok(undefined); - } + const historyPath = this.getChatHistoryPath(workspaceId); + const historyEntries = this.serializeHistoryEntries(filteredMessages, workspaceId); - const historyPath = this.getChatHistoryPath(workspaceId); - const historyEntries = this.serializeHistoryEntries(filteredMessages, workspaceId); + // Atomic write prevents corruption if app crashes mid-write + await writeFileAtomic(historyPath, historyEntries); - // Atomic write prevents corruption if app crashes mid-write - await writeFileAtomic(historyPath, historyEntries); + // Keep the in-memory sequence counter monotonic. It's okay to reuse deleted sequence + // numbers on restart, but we must not regress within a running process. + const maxSeq = filteredMessages.reduce((max, msg) => { + const seq = msg.metadata?.historySequence; + if (seq === undefined) { + return max; + } - // Keep the in-memory sequence counter monotonic. It's okay to reuse deleted sequence - // numbers on restart, but we must not regress within a running process. - const maxSeq = filteredMessages.reduce((max, msg) => { - const seq = msg.metadata?.historySequence; - if (seq === undefined) { - return max; - } + if (!isNonNegativeInteger(seq)) { + log.warn( + "Ignoring malformed persisted historySequence while updating sequence counter after delete", + { + workspaceId, + messageId: msg.id, + historySequence: seq, + } + ); + return max; + } - if (!isNonNegativeInteger(seq)) { - log.warn( - "Ignoring malformed persisted historySequence while updating sequence counter after delete", - { - workspaceId, - messageId: msg.id, - historySequence: seq, - } - ); - return max; + return seq > max ? seq : max; + }, -1); + // Sealed archive rows keep their sequences across active-file deletes. + // Without this floor, deleting the last sequenced active row in a fresh + // process would cache a counter below archived rows and reuse their + // historySequence values on the next append. + const archiveMaxSeq = await this.getArchiveTailMaxSequence(workspaceId); + const nextSeq = Math.max(maxSeq, archiveMaxSeq) + 1; + assert( + isNonNegativeInteger(nextSeq), + "next history sequence counter after delete must be a non-negative integer" + ); + const currentCounter = this.sequenceCounters.get(workspaceId); + if (currentCounter === undefined || currentCounter < nextSeq) { + this.sequenceCounters.set(workspaceId, nextSeq); } - return seq > max ? seq : max; - }, -1); - // Sealed archive rows keep their sequences across active-file deletes. - // Without this floor, deleting the last sequenced active row in a fresh - // process would cache a counter below archived rows and reuse their - // historySequence values on the next append. - const archiveMaxSeq = await this.getArchiveTailMaxSequence(workspaceId); - const nextSeq = Math.max(maxSeq, archiveMaxSeq) + 1; - assert( - isNonNegativeInteger(nextSeq), - "next history sequence counter after delete must be a non-negative integer" - ); - const currentCounter = this.sequenceCounters.get(workspaceId); - if (currentCounter === undefined || currentCounter < nextSeq) { - this.sequenceCounters.set(workspaceId, nextSeq); + return Ok(undefined); + } catch (error) { + const message = getErrorMessage(error); + return Err(`Failed to delete message: ${message}`); } - - return Ok(undefined); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to delete message: ${message}`); } - }); + ); } /** @@ -1769,80 +2120,85 @@ export class HistoryService { messageId: string, options?: { keepTargetMessage?: boolean } ): Promise> { - return this.fileLocks.withLock(workspaceId, async () => { - try { - // Structural rewrite requires full file content - const messages = await this.readChatHistory(workspaceId); - const messageIndex = messages.findIndex((msg) => msg.id === messageId); - - const keepTargetMessage = options?.keepTargetMessage === true; - - if (messageIndex === -1) { - // Editing/forking from a pre-boundary message: the target lives in the - // sealed archive. Everything after the cut (the archive tail AND the - // entire active epoch) is discarded, so collapse the remainder back - // into chat.jsonl and drop the archive. - return this.truncateAfterArchivedMessageUnlocked( - workspaceId, - messageId, - keepTargetMessage + return this.withRecoveredHistoryResultLock( + workspaceId, + "Failed to truncate history", + async () => { + try { + // Structural rewrite requires full file content + const messages = await this.readChatHistory(workspaceId); + const messageIndex = messages.findIndex((msg) => msg.id === messageId); + + const keepTargetMessage = options?.keepTargetMessage === true; + + if (messageIndex === -1) { + // Editing/forking from a pre-boundary message: the target lives in the + // sealed archive. Everything after the cut (the archive tail AND the + // entire active epoch) is discarded, so collapse the remainder back + // into chat.jsonl and drop the archive. + return this.truncateAfterArchivedMessageUnlocked( + workspaceId, + messageId, + keepTargetMessage + ); + } + + // Response-level forks branch from the selected assistant turn, so they retain the target + // message while discarding anything that came after it. + const truncatedMessages = messages.slice( + 0, + keepTargetMessage ? messageIndex + 1 : messageIndex ); - } - // Response-level forks branch from the selected assistant turn, so they retain the target - // message while discarding anything that came after it. - const truncatedMessages = messages.slice( - 0, - keepTargetMessage ? messageIndex + 1 : messageIndex - ); + // Rewrite the history file with truncated messages + const historyPath = this.getChatHistoryPath(workspaceId); + const historyEntries = this.serializeHistoryEntries(truncatedMessages, workspaceId); - // Rewrite the history file with truncated messages - const historyPath = this.getChatHistoryPath(workspaceId); - const historyEntries = this.serializeHistoryEntries(truncatedMessages, workspaceId); + const archiveMaxSeq = await this.getArchiveTailMaxSequence(workspaceId); - // Atomic write prevents corruption if app crashes mid-write - await writeFileAtomic(historyPath, historyEntries); + // Atomic write prevents corruption if app crashes mid-write + await writeFileAtomic(historyPath, historyEntries); - // Update sequence counter to continue from where we truncated. - // Self-healing read path: skip malformed persisted historySequence values. - const maxTruncatedSeq = truncatedMessages.reduce((max, msg) => { - const seq = msg.metadata?.historySequence; - if (seq === undefined) { - return max; - } + // Update sequence counter to continue from where we truncated. + // Self-healing read path: skip malformed persisted historySequence values. + const maxTruncatedSeq = truncatedMessages.reduce((max, msg) => { + const seq = msg.metadata?.historySequence; + if (seq === undefined) { + return max; + } - if (!isNonNegativeInteger(seq)) { - log.warn( - "Ignoring malformed persisted historySequence while updating sequence counter after truncation", - { - workspaceId, - messageId: msg.id, - historySequence: seq, - } - ); - return max; - } + if (!isNonNegativeInteger(seq)) { + log.warn( + "Ignoring malformed persisted historySequence while updating sequence counter after truncation", + { + workspaceId, + messageId: msg.id, + historySequence: seq, + } + ); + return max; + } - return seq > max ? seq : max; - }, -1); - // Sealed archive rows keep their sequences across an active-epoch - // truncation. When the truncation empties the active file, floor the - // counter with the archive max so new appends can never reuse archived - // sequence numbers. - const archiveMaxSeq = await this.getArchiveTailMaxSequence(workspaceId); - const nextSeq = Math.max(maxTruncatedSeq, archiveMaxSeq) + 1; - assert( - isNonNegativeInteger(nextSeq), - "next history sequence counter after truncation must be a non-negative integer" - ); - this.sequenceCounters.set(workspaceId, nextSeq); + return seq > max ? seq : max; + }, -1); + // Sealed archive rows keep their sequences across an active-epoch + // truncation. When the truncation empties the active file, floor the + // counter with the archive max so new appends can never reuse archived + // sequence numbers. + const nextSeq = Math.max(maxTruncatedSeq, archiveMaxSeq) + 1; + assert( + isNonNegativeInteger(nextSeq), + "next history sequence counter after truncation must be a non-negative integer" + ); + this.sequenceCounters.set(workspaceId, nextSeq); - return Ok(undefined); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to truncate history: ${message}`); + return Ok(undefined); + } catch (error) { + const message = getErrorMessage(error); + return Err(`Failed to truncate history: ${message}`); + } } - }); + ); } /** @@ -1869,11 +2225,11 @@ export class HistoryService { keepTargetMessage ? messageIndex + 1 : messageIndex ); - await writeFileAtomic( - this.getChatHistoryPath(workspaceId), + await this.rewriteHistoryFilesUnlocked( + workspaceId, + null, this.serializeHistoryEntries(truncatedMessages, workspaceId) ); - await fs.rm(this.getChatArchivePath(workspaceId), { force: true }); // chat.jsonl may contain sealed epochs again — allow the lazy check to re-run. this.sealedRotationChecked.delete(workspaceId); @@ -1923,138 +2279,130 @@ export class HistoryService { workspaceId: string, percentage: number ): Promise> { - return this.fileLocks.withLock(workspaceId, async () => { - try { - const historyPath = this.getChatHistoryPath(workspaceId); - const archivePath = this.getChatArchivePath(workspaceId); - - // Fast path: 100% truncation = delete entire history (active + sealed archive) - if (percentage >= 1.0) { - // Need sequence numbers for return value before deleting - const messages = [ - ...(await this.readArchivedHistory(workspaceId)), - ...(await this.readChatHistory(workspaceId)), - ]; - const deletedSequences = messages + return this.withRecoveredHistoryResultLock( + workspaceId, + "Failed to truncate history", + async () => { + try { + const archivedMessages = await this.readArchivedHistory(workspaceId); + const chatMessages = await this.readChatHistory(workspaceId); + const messages = [...archivedMessages, ...chatMessages]; + const allSequences = messages .map((msg) => msg.metadata?.historySequence) .filter((s): s is number => isNonNegativeInteger(s)); - await fs.rm(historyPath, { force: true }); - await fs.rm(archivePath, { force: true }); + if (percentage >= 1.0) { + await this.rewriteHistoryFilesUnlocked(workspaceId, null, null); + this.sequenceCounters.set(workspaceId, 0); + return Ok(allSequences); + } - // Reset sequence counter when clearing history - this.sequenceCounters.set(workspaceId, 0); - return Ok(deletedSequences); - } + // Structural rewrite requires full history content (oldest rows live in + // the sealed archive). Percentage truncation is a rare recovery path + // (compaction-failure retry), so the O(total-history) read is acceptable. + if (messages.length === 0) { + return Ok([]); // Nothing to truncate + } - // Structural rewrite requires full history content (oldest rows live in - // the sealed archive). Percentage truncation is a rare recovery path - // (compaction-failure retry), so the O(total-history) read is acceptable. - const messages = [ - ...(await this.readArchivedHistory(workspaceId)), - ...(await this.readChatHistory(workspaceId)), - ]; - if (messages.length === 0) { - return Ok([]); // Nothing to truncate - } + // Get tokenizer for counting (use a default model) + const tokenizer = await getTokenizerForModel(KNOWN_MODELS.SONNET.id); - // Get tokenizer for counting (use a default model) - const tokenizer = await getTokenizerForModel(KNOWN_MODELS.SONNET.id); + // Count tokens for each message + // We stringify the entire message for simplicity - only relative weights matter + const messageTokens: Array<{ message: MuxMessage; tokens: number }> = await Promise.all( + messages.map(async (msg) => { + const tokens = await tokenizer.countTokens(safeStringifyForCounting(msg)); + return { message: msg, tokens }; + }) + ); - // Count tokens for each message - // We stringify the entire message for simplicity - only relative weights matter - const messageTokens: Array<{ message: MuxMessage; tokens: number }> = await Promise.all( - messages.map(async (msg) => { - const tokens = await tokenizer.countTokens(safeStringifyForCounting(msg)); - return { message: msg, tokens }; - }) - ); + // Calculate total tokens and target to remove + const totalTokens = messageTokens.reduce((sum, mt) => sum + mt.tokens, 0); + const tokensToRemove = Math.floor(totalTokens * percentage); - // Calculate total tokens and target to remove - const totalTokens = messageTokens.reduce((sum, mt) => sum + mt.tokens, 0); - const tokensToRemove = Math.floor(totalTokens * percentage); + // Remove messages from beginning until we've removed enough tokens + let tokensRemoved = 0; + let removeCount = 0; + for (const mt of messageTokens) { + if (tokensRemoved >= tokensToRemove) { + break; + } + tokensRemoved += mt.tokens; + removeCount++; + } - // Remove messages from beginning until we've removed enough tokens - let tokensRemoved = 0; - let removeCount = 0; - for (const mt of messageTokens) { - if (tokensRemoved >= tokensToRemove) { - break; + // No-op truncation (percentage 0 or rounding to zero tokens) must not + // rewrite anything — collapsing the archive back into chat.jsonl would + // undo rotation and put lifetime history back on the hot path. + if (removeCount === 0) { + return Ok([]); } - tokensRemoved += mt.tokens; - removeCount++; - } - // No-op truncation (percentage 0 or rounding to zero tokens) must not - // rewrite anything — collapsing the archive back into chat.jsonl would - // undo rotation and put lifetime history back on the hot path. - if (removeCount === 0) { - return Ok([]); - } + // If we're removing all messages, use fast path + if (removeCount >= messages.length) { + await this.rewriteHistoryFilesUnlocked(workspaceId, null, null); + this.sequenceCounters.set(workspaceId, 0); + return Ok(allSequences); + } - // If we're removing all messages, use fast path - if (removeCount >= messages.length) { - await fs.rm(historyPath, { force: true }); - await fs.rm(archivePath, { force: true }); - this.sequenceCounters.set(workspaceId, 0); - const deletedSequences = messages + const activeContextChanged = prefixCutChangesActiveContext(messages, removeCount); + const sanitize = activeContextChanged + ? stripContextUsage + : (message: MuxMessage) => message; + const remainingMessages = messages.slice(removeCount).map(sanitize); + const deletedMessages = messages.slice(0, removeCount); + const deletedSequences = deletedMessages .map((msg) => msg.metadata?.historySequence) .filter((s): s is number => isNonNegativeInteger(s)); - return Ok(deletedSequences); - } + const remainingArchiveCount = Math.max(0, archivedMessages.length - removeCount); + const remainingArchive = remainingMessages.slice(0, remainingArchiveCount); + const remainingChat = remainingMessages.slice(remainingArchiveCount); - // Keep messages after removeCount - const remainingMessages = messages.slice(removeCount); - const deletedMessages = messages.slice(0, removeCount); - const deletedSequences = deletedMessages - .map((msg) => msg.metadata?.historySequence) - .filter((s): s is number => isNonNegativeInteger(s)); - - // Collapse the remainder into chat.jsonl and drop the archive (the cut - // may fall anywhere inside it). It may contain old boundaries; a later - // boundary write re-seals it. - const historyEntries = this.serializeHistoryEntries(remainingMessages, workspaceId); - - // Atomic write prevents corruption if app crashes mid-write - await writeFileAtomic(historyPath, historyEntries); - await fs.rm(archivePath, { force: true }); - this.sealedRotationChecked.delete(workspaceId); - - // Update sequence counter to continue from where we are. - // Self-healing read path: skip malformed persisted historySequence values. - const maxRemainingSeq = remainingMessages.reduce((max, msg) => { - const seq = msg.metadata?.historySequence; - if (seq === undefined) { - return max; - } + await this.rewriteHistoryFilesUnlocked( + workspaceId, + remainingArchive.length > 0 + ? this.serializeHistoryEntries(remainingArchive, workspaceId) + : null, + this.serializeHistoryEntries(remainingChat, workspaceId) + ); + this.sealedRotationChecked.delete(workspaceId); + + // Update sequence counter to continue from where we are. + // Self-healing read path: skip malformed persisted historySequence values. + const maxRemainingSeq = remainingMessages.reduce((max, msg) => { + const seq = msg.metadata?.historySequence; + if (seq === undefined) { + return max; + } - if (!isNonNegativeInteger(seq)) { - log.warn( - "Ignoring malformed persisted historySequence while updating sequence counter after truncateHistory", - { - workspaceId, - messageId: msg.id, - historySequence: seq, - } - ); - return max; - } + if (!isNonNegativeInteger(seq)) { + log.warn( + "Ignoring malformed persisted historySequence while updating sequence counter after truncateHistory", + { + workspaceId, + messageId: msg.id, + historySequence: seq, + } + ); + return max; + } - return seq > max ? seq : max; - }, -1); - const nextSeq = maxRemainingSeq + 1; - assert( - isNonNegativeInteger(nextSeq), - "next history sequence counter after truncateHistory must be a non-negative integer" - ); - this.sequenceCounters.set(workspaceId, nextSeq); + return seq > max ? seq : max; + }, -1); + const nextSeq = maxRemainingSeq + 1; + assert( + isNonNegativeInteger(nextSeq), + "next history sequence counter after truncateHistory must be a non-negative integer" + ); + this.sequenceCounters.set(workspaceId, nextSeq); - return Ok(deletedSequences); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to truncate history: ${message}`); + return Ok(deletedSequences); + } catch (error) { + const message = getErrorMessage(error); + return Err(`Failed to truncate history: ${message}`); + } } - }); + ); } async clearHistory(workspaceId: string): Promise> { @@ -2071,54 +2419,58 @@ export class HistoryService { * IMPORTANT: Should be called AFTER the session directory has been renamed */ async migrateWorkspaceId(oldWorkspaceId: string, newWorkspaceId: string): Promise> { - return this.fileLocks.withLock(newWorkspaceId, async () => { - try { - // Migrate the sealed archive first so a crash mid-migration never leaves - // the active file pointing at a stale-ID archive. - const archiveMessages = await this.readArchivedHistory(newWorkspaceId); - if (archiveMessages.length > 0) { - await writeFileAtomic( - this.getChatArchivePath(newWorkspaceId), - this.serializeHistoryEntries(archiveMessages, newWorkspaceId) - ); - } + return this.withRecoveredHistoryResultLock( + newWorkspaceId, + "Failed to migrate workspace history", + async () => { + try { + // Migrate the sealed archive first so a crash mid-migration never leaves + // the active file pointing at a stale-ID archive. + const archiveMessages = await this.readArchivedHistory(newWorkspaceId); + if (archiveMessages.length > 0) { + await writeFileAtomic( + this.getChatArchivePath(newWorkspaceId), + this.serializeHistoryEntries(archiveMessages, newWorkspaceId) + ); + } - // Read messages from the NEW workspace location (directory was already renamed). - // Structural rewrite requires full file content. - const messages = await this.readChatHistory(newWorkspaceId); - if (messages.length === 0) { - // No active messages to migrate, just transfer the sequence counter. - // Floor it with the archive max: an archive-only session (active file - // deleted/truncated) renamed in a fresh process has no cached counter, - // and seeding 0 would reuse archived historySequence values. - const oldCounter = this.sequenceCounters.get(oldWorkspaceId) ?? 0; - const archiveFloor = (await this.getArchiveTailMaxSequence(newWorkspaceId)) + 1; - this.sequenceCounters.set(newWorkspaceId, Math.max(oldCounter, archiveFloor)); - this.sequenceCounters.delete(oldWorkspaceId); - return Ok(undefined); - } + // Read messages from the NEW workspace location (directory was already renamed). + // Structural rewrite requires full file content. + const messages = await this.readChatHistory(newWorkspaceId); + if (messages.length === 0) { + // No active messages to migrate, just transfer the sequence counter. + // Floor it with the archive max: an archive-only session (active file + // deleted/truncated) renamed in a fresh process has no cached counter, + // and seeding 0 would reuse archived historySequence values. + const oldCounter = this.sequenceCounters.get(oldWorkspaceId) ?? 0; + const archiveFloor = (await this.getArchiveTailMaxSequence(newWorkspaceId)) + 1; + this.sequenceCounters.set(newWorkspaceId, Math.max(oldCounter, archiveFloor)); + this.sequenceCounters.delete(oldWorkspaceId); + return Ok(undefined); + } - // Rewrite all messages with new workspace ID - const newHistoryPath = this.getChatHistoryPath(newWorkspaceId); - const historyEntries = this.serializeHistoryEntries(messages, newWorkspaceId); + // Rewrite all messages with new workspace ID + const newHistoryPath = this.getChatHistoryPath(newWorkspaceId); + const historyEntries = this.serializeHistoryEntries(messages, newWorkspaceId); - // Atomic write prevents corruption if app crashes mid-write - await writeFileAtomic(newHistoryPath, historyEntries); + // Atomic write prevents corruption if app crashes mid-write + await writeFileAtomic(newHistoryPath, historyEntries); - // Transfer sequence counter to new workspace ID - const oldCounter = this.sequenceCounters.get(oldWorkspaceId) ?? 0; - this.sequenceCounters.set(newWorkspaceId, oldCounter); - this.sequenceCounters.delete(oldWorkspaceId); + // Transfer sequence counter to new workspace ID + const oldCounter = this.sequenceCounters.get(oldWorkspaceId) ?? 0; + this.sequenceCounters.set(newWorkspaceId, oldCounter); + this.sequenceCounters.delete(oldWorkspaceId); - log.debug( - `Migrated ${messages.length} messages from ${oldWorkspaceId} to ${newWorkspaceId}` - ); + log.debug( + `Migrated ${messages.length} messages from ${oldWorkspaceId} to ${newWorkspaceId}` + ); - return Ok(undefined); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to migrate workspace ID: ${message}`); + return Ok(undefined); + } catch (error) { + const message = getErrorMessage(error); + return Err(`Failed to migrate workspace ID: ${message}`); + } } - }); + ); } } diff --git a/src/node/services/sessionUsageService.ts b/src/node/services/sessionUsageService.ts index 6cccff4f07..3d66eb5ebb 100644 --- a/src/node/services/sessionUsageService.ts +++ b/src/node/services/sessionUsageService.ts @@ -116,15 +116,16 @@ export class SessionUsageService { this.historyService = historyService; this.getProvidersConfig = getProvidersConfig ?? (() => null); } - /** - * Collect all messages from iterateFullHistory into an array. - * Usage rebuild needs every epoch for accurate totals. - */ + /** Usage rebuild needs every epoch for accurate totals. */ private async collectFullHistory(workspaceId: string): Promise { const messages: MuxMessage[] = []; - const result = await this.historyService.iterateFullHistory(workspaceId, "forward", (chunk) => { - messages.push(...chunk); - }); + const result = await this.historyService.iterateFullHistoryUnderLock( + workspaceId, + "forward", + (chunk) => { + messages.push(...chunk); + } + ); if (!result.success) { log.warn(`Failed to iterate history for ${workspaceId}: ${result.error}`); return []; diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 57496dad20..5866411195 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -13710,6 +13710,8 @@ describe("TaskService", () => { parts: childPartial.parts as StreamEndEvent["parts"], }); + await flushTerminalAttentionDrains(taskService); + const updatedChildPartial = await partialService.readPartial(childId); expect(updatedChildPartial).toBeNull(); @@ -15189,6 +15191,8 @@ describe("TaskService", () => { parts: childPartial.parts as StreamEndEvent["parts"], }); + await flushTerminalAttentionDrains(taskService); + const parentMessages = await collectFullHistory(historyService, parentId); // Original task tool call remains immutable ("running"), and a synthetic report message is appended. expect(parentMessages.length).toBeGreaterThanOrEqual(2); diff --git a/src/node/services/workspaceGoalService.test.ts b/src/node/services/workspaceGoalService.test.ts index 25b9a098a4..6a82658777 100644 --- a/src/node/services/workspaceGoalService.test.ts +++ b/src/node/services/workspaceGoalService.test.ts @@ -1600,23 +1600,22 @@ describe("WorkspaceGoalService", () => { expect(maxActiveContinuations).toBe(1); }); - test("does not dispatch stale continuation candidates after the goal changes", async () => { + test("does not build stale continuation payloads after the goal changes", async () => { await setGoalOk(service, { workspaceId, objective: "Original" }); - // Give the replacement write time to commit before the idle dispatch builds - // its payload; this test is about rejecting an already-stale candidate, not racing setGoal. - const dispatcher = new IdleDispatcher({ debounceMs: 250 }); - const execute = mock(() => Promise.resolve(true)); - service.registerGoalContinuationConsumer(dispatcher, continuationBridge(execute)); + const dispatcher = new IdleDispatcher(); + const requestDispatch = spyOn(dispatcher, "requestDispatch").mockResolvedValue(); + service.registerGoalContinuationConsumer(dispatcher, continuationBridge()); - const request = service.requestContinuationAfterStreamEnd({ + await service.requestContinuationAfterStreamEnd({ workspaceId, sendOptions: { model: "openai:gpt-4o", agentId: "exec" }, streamEndedAtMs: 10_000, }); + expect(requestDispatch).toHaveBeenCalled(); + expect(await service.buildGoalContinuationPayload(workspaceId)).not.toBeNull(); await setGoalOk(service, { workspaceId, objective: "Replacement" }); - await request; - expect(execute).not.toHaveBeenCalled(); + expect(await service.buildGoalContinuationPayload(workspaceId)).toBeNull(); }); test("preserves goal id and accounting for same-objective set", async () => { diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index 0738c1fe30..07effd4c71 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -2,6 +2,9 @@ import { describe, expect, test, mock, beforeEach, afterEach, spyOn, type Mock } import { WorkspaceService, generateForkBranchName, generateForkTitle } from "./workspaceService"; import type { IdleCompactionOutcome } from "./idleCompactionService"; import type { AgentSession } from "./agentSession"; +import { createAgentSessionHarness } from "./agentSession.testHarness"; +import type { AutoCompactionUsageState } from "@/common/utils/compaction/autoCompactionCheck"; +import { createDisplayUsage } from "@/common/utils/tokens/displayUsage"; import { askUserQuestionManager } from "./askUserQuestionManager"; import { WorkspaceLifecycleHooks } from "./workspaceLifecycleHooks"; import { EventEmitter } from "events"; @@ -4273,6 +4276,89 @@ describe("WorkspaceService truncateHistory goal acknowledgment", () => { } }); + test("start-here replacement does not auto-compact the next send from stale usage", async () => { + const { config, historyService, workspaceService, cleanup } = await createServices(); + const workspaceId = "start-here-clears-usage-state"; + const streamMessage = mock((..._args: unknown[]) => Promise.resolve(Ok(undefined))); + const harness = await createAgentSessionHarness({ + workspaceId, + config, + historyService, + aiServiceOverrides: { + streamMessage: streamMessage as unknown as AIService["streamMessage"], + }, + }); + try { + await config.addWorkspace("/tmp/start-here-usage-project", { + id: workspaceId, + name: workspaceId, + projectName: "start-here-usage-project", + projectPath: "/tmp/start-here-usage-project", + runtimeConfig: { type: "local" }, + }); + await historyService.appendToHistory( + workspaceId, + createMuxMessage("pre-start-here-user", "user", "long conversation", {}) + ); + await historyService.appendToHistory( + workspaceId, + createMuxMessage("pre-start-here-assistant", "assistant", "long reply", { + model: "openai:gpt-4o", + contextUsage: { inputTokens: 95_000, outputTokens: 200, totalTokens: 95_200 }, + }) + ); + + (workspaceService as unknown as { sessions: Map }).sessions.set( + workspaceId, + harness.session + ); + (harness.session as unknown as { lastUsageState?: AutoCompactionUsageState }).lastUsageState = + { + lastContextUsage: createDisplayUsage( + { inputTokens: 95_000, outputTokens: 200, totalTokens: 95_200 }, + "openai:gpt-4o" + ), + }; + + expect( + ( + await workspaceService.replaceHistory( + workspaceId, + createMuxMessage("start-here-summary", "assistant", "Start Here summary", { + compacted: "user", + }), + { mode: "append-compaction-boundary" } + ) + ).success + ).toBe(true); + expect( + ( + await harness.session.sendMessage("follow-up after start here", { + model: "openai:gpt-4o", + agentId: "exec", + }) + ).success + ).toBe(true); + + const activeWindow = await historyService.getHistoryFromLatestBoundary(workspaceId); + expect(activeWindow.success).toBe(true); + const activeMessages = activeWindow.success ? activeWindow.data : []; + expect( + activeMessages.filter( + (message) => message.metadata?.muxMetadata?.type === "compaction-request" + ) + ).toHaveLength(0); + expect(activeMessages.find((message) => message.role === "user")?.parts[0]).toMatchObject({ + type: "text", + text: "follow-up after start here", + }); + expect(streamMessage).toHaveBeenCalledTimes(1); + } finally { + harness.session.dispose(); + await cleanup(); + } + }); + test("context reset is a no-op when repeated without provider-eligible messages", async () => { const { config, historyService, workspaceService, cleanup } = await createServices(); const workspaceId = "context-reset-noop"; diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 5aa54d0520..4c4c18c4c8 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -8080,13 +8080,15 @@ export class WorkspaceService extends EventEmitter { const newSessionDir = this.config.getSessionDir(newWorkspaceId); try { - await ensurePrivateDir(newSessionDir); + const historyCopyResult = await this.historyService.copyHistorySnapshotToNewWorkspace( + sourceWorkspaceId, + newWorkspaceId + ); + if (!historyCopyResult.success) { + throw new Error(historyCopyResult.error); + } const sessionFiles = [ - CHAT_FILE_NAME, - // Sealed pre-boundary history must travel with chat.jsonl so the fork - // keeps full Load More/paging access to older epochs. - CHAT_ARCHIVE_FILE_NAME, "session-timing.json", ADDITIONAL_SYSTEM_CONTEXT_FILENAME, // Preserve the enabled/disabled toggle when forking so the fork @@ -9666,6 +9668,9 @@ export class WorkspaceService extends EventEmitter { const effectivePercentage = percentage ?? 1.0; const isFullClear = effectivePercentage >= 1.0; + if (effectivePercentage > 0) { + session?.clearUsageState(); + } const truncate = () => this.historyService.truncateHistory(workspaceId, effectivePercentage); const truncateResult = effectivePercentage > 0 @@ -9758,6 +9763,8 @@ export class WorkspaceService extends EventEmitter { return Err(`Failed to append context reset boundary: ${appendResult.error}`); } + session?.clearUsageState(); + const typedBoundaryMessage = { ...boundaryMessage, type: "message" as const }; if (session) { session.emitChatEvent(typedBoundaryMessage); @@ -9860,6 +9867,7 @@ export class WorkspaceService extends EventEmitter { `replaceHistory received unsupported replace mode: ${String(replaceMode)}` ); + this.sessions.get(workspaceId)?.clearUsageState(); const clearResult = await this.clearHistoryWithRetiredBashMonitorWakes( workspaceId, () => this.historyService.clearHistory(workspaceId), @@ -9881,6 +9889,8 @@ export class WorkspaceService extends EventEmitter { return Err(`Failed to append summary message: ${appendResult.error}`); } + this.sessions.get(workspaceId)?.clearUsageState(); + // Emit through the session so ORPC subscriptions receive the events const session = this.sessions.get(workspaceId); if (deletedSequences.length > 0) {