diff --git a/.oxlintrc.json b/.oxlintrc.json index 705f1d69357..75a6da8cf08 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -1298,8 +1298,7 @@ "src/node/services/tools/todo.ts": ["@/node/utils/writeFileAtomic", "fs/promises"], "src/node/services/tools/xum_agents_path.ts": ["fs/promises"], "src/node/services/tools/xum_agents_read.ts": ["fs/promises"], - "src/node/services/tools/xum_agents_write.ts": ["fs/promises"], - "src/node/services/compactionCancellation.ts": ["fs"] + "src/node/services/tools/xum_agents_write.ts": ["fs/promises"] } } ] diff --git a/eslint.config.mjs b/eslint.config.mjs index e612f6da4e2..532f0f20741 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -176,8 +176,6 @@ const HOST_IMPORT_ALLOW = { "src/node/services/tools/xum_agents_path.ts": ["fs/promises"], "src/node/services/tools/xum_agents_read.ts": ["fs/promises"], "src/node/services/tools/xum_agents_write.ts": ["fs/promises"], - // Phase 2: each move PR removes its own pairs. - "src/node/services/compactionCancellation.ts": ["fs"], }; const SHARED_RESTRICTED_PATHS = [AI_TOKENIZER_ENCODING_RESTRICTION]; diff --git a/scripts/noHostImports.test.ts b/scripts/noHostImports.test.ts index 2eb066a1678..82f6cded28f 100644 --- a/scripts/noHostImports.test.ts +++ b/scripts/noHostImports.test.ts @@ -149,8 +149,6 @@ const FROZEN_ALLOW_PAIRS = [ "src/node/services/tools/xum_agents_path.ts fs/promises", "src/node/services/tools/xum_agents_read.ts fs/promises", "src/node/services/tools/xum_agents_write.ts fs/promises", - // Phase 2 (session files): the 15 pairs of the history layer. - "src/node/services/compactionCancellation.ts fs", ]; function allowMap(): Record { diff --git a/src/node/services/agentSession.compactionCancellation.test.ts b/src/node/services/agentSession.compactionCancellation.test.ts index 21c8303b991..bec234c3261 100644 --- a/src/node/services/agentSession.compactionCancellation.test.ts +++ b/src/node/services/agentSession.compactionCancellation.test.ts @@ -1930,8 +1930,13 @@ describe("compaction cancellation runtime", () => { spyOn(fresh.historyService, "cleanupCompactionFollowUp").mockImplementationOnce( async (...args) => { const result = await cleanup(...args); + const foreignHistory = new HistoryService(h.config); const foreign = new CompactionCancellation( - new FileCompactionCancellationStorage(new HistoryService(h.config), workspaceId) + new FileCompactionCancellationStorage( + foreignHistory, + workspaceId, + foreignHistory.files + ) ); await foreign.cancel(); successorNonce = (await foreign.read())?.nonce; diff --git a/src/node/services/agentSession.turnCompletion.test.ts b/src/node/services/agentSession.turnCompletion.test.ts index 1a4f8918a81..3560aafb5cb 100644 --- a/src/node/services/agentSession.turnCompletion.test.ts +++ b/src/node/services/agentSession.turnCompletion.test.ts @@ -86,9 +86,9 @@ describe("AgentSession turn completion", () => { }, }) ); - const open = fileIO.open; + const open = fs.open; let failed = false; - const reading = spyOn(fileIO, "open").mockImplementation( + const reading = spyOn(fs, "open").mockImplementation( async (...args: Parameters) => { if (args[0] === storage.path && !failed) { failed = true; @@ -139,23 +139,21 @@ describe("AgentSession turn completion", () => { const stream = spyOn(h.aiService, "streamMessage"); const storage = h.historyService.getCompactionCancellationStorage(workspaceId); const failure = Object.assign(new Error(`${code}: cancellation read unavailable`), { code }); - const open = fileIO.open; - let reading: ReturnType> | undefined; + const open = fs.open; + let reading: ReturnType> | undefined; const warning = spyOn(log, "warn"); try { expect(await h.session.sendMessage("original", sendOptions)).toEqual(Ok(undefined)); expect(await h.session.cancelCompaction(true)).toEqual(Ok(undefined)); const cancellationBytes = await fileIO.readFile(storage.path); - reading = spyOn(fileIO, "open").mockImplementation( - async (...args: Parameters) => { - if (args[0] === storage.path) { - entered.resolve(); - await release.promise; - throw failure; - } - return open(...args); + reading = spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { + if (args[0] === storage.path) { + entered.resolve(); + await release.promise; + throw failure; } - ); + return open(...args); + }); const streamError = { messageId: "assistant-1", error: "provider failed", diff --git a/src/node/services/compactionCancellation.storage.test.ts b/src/node/services/compactionCancellation.storage.test.ts index 37b0b627af3..164c0d661d1 100644 --- a/src/node/services/compactionCancellation.storage.test.ts +++ b/src/node/services/compactionCancellation.storage.test.ts @@ -91,7 +91,11 @@ describe("inactive real cancellation storage", () => { mutationCommitted.mockClear(); h = await createTestHistoryService(); foreign = new HistoryService(h.config); - storage = new FileCompactionCancellationStorage(h.historyService, workspaceId); + storage = new FileCompactionCancellationStorage( + h.historyService, + workspaceId, + h.historyService.files + ); state = new CompactionCancellation(storage); sessionDir = path.dirname(storage.path); expect( @@ -509,14 +513,18 @@ describe("inactive real cancellation storage", () => { expect(initial.version).toBe(1); if (kind === "scoped") { const other = new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ); await other.read(); await other.narrow(initial.nonce, summary); } settled.resolve(kind !== "failed"); expect(await stopping).toBe("applied"); - const persisted = await new FileCompactionCancellationStorage(foreign, workspaceId).read(); + const persisted = await new FileCompactionCancellationStorage( + foreign, + workspaceId, + foreign.files + ).read(); expect(persisted?.nonce).toBe(initial.nonce); expect(persisted?.version).toBe(kind === "joined" ? 2 : 1); if (kind === "joined") { @@ -525,7 +533,7 @@ describe("inactive real cancellation storage", () => { ); expect( await new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ).read() ).toEqual(persisted); } else expect(persisted?.settledGeneration).toBeUndefined(); @@ -656,7 +664,7 @@ describe("inactive real cancellation storage", () => { // These real writes acquire the same locks: waiting for physical settlement must release them. if (successorKind === "nonce") { await new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ).cancel({ retainUntilReplacement: true }); } else { await foreign.withCompactionStorageLock(workspaceId, async (_dir, checkLock) => { @@ -755,7 +763,7 @@ describe("inactive real cancellation storage", () => { expect(await storage.read()).toBeNull(); failure.mockRestore(); const successor = new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ); expect(await successor.cancel({ retainUntilReplacement: true })).toBe("applied"); const successorBytes = await fs.readFile(storage.path); @@ -914,20 +922,18 @@ describe("inactive real cancellation storage", () => { nodeFs.writeFileSync(generationPath, "foreign-generation"); }; if (phase === "retirement") { - const open = nodeFs.promises.open; - spyOn(nodeFs.promises, "open").mockImplementation( - async (...args: Parameters) => { - const handle = await open(...args); - if (args[0] === storage.path) { - const close = handle.close.bind(handle); - spyOn(handle, "close").mockImplementation(async () => { - await close(); - if (!displaced) displace(); - }); - } - return handle; + const open = fs.open; + spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { + const handle = await open(...args); + if (args[0] === storage.path) { + const close = handle.close.bind(handle); + spyOn(handle, "close").mockImplementation(async () => { + await close(); + if (!displaced) displace(); + }); } - ); + return handle; + }); } else { const target = phase === "generation" ? generationPath : storage.path; afterCompactionStaging(target, () => { @@ -1134,40 +1140,38 @@ describe("inactive real cancellation storage", () => { let totalRead = 0; let largestBuffer = 0; let closed = false; - const open = nodeFs.promises.open; - spyOn(nodeFs.promises, "open").mockImplementation( - async (...args: Parameters) => { - const handle = await open(...args); - if (args[0] === storage.path) { - const close = handle.close.bind(handle); - spyOn(handle, "close").mockImplementation(async () => { - await close(); - closed = true; - }); - const read = handle.read.bind(handle); - spyOn(handle, "read").mockImplementation((async ( - buffer: Buffer, - offset: number, - length: number, - position: number - ) => { - reads++; - largestBuffer = Math.max(largestBuffer, buffer.byteLength); - if (kind === "read error") throw new Error("descriptor read failed"); - // Exercise legal short reads; a single read must not mistake them for EOF. - const result = await read( - buffer, - offset, - Math.min(length, SESSION_HISTORY_MAX_LINE_BYTES / 16), - position - ); - totalRead += result.bytesRead; - return result; - }) as typeof handle.read); - } - return handle; + const open = fs.open; + spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { + const handle = await open(...args); + if (args[0] === storage.path) { + const close = handle.close.bind(handle); + spyOn(handle, "close").mockImplementation(async () => { + await close(); + closed = true; + }); + const read = handle.read.bind(handle); + spyOn(handle, "read").mockImplementation((async ( + buffer: Buffer, + offset: number, + length: number, + position: number + ) => { + reads++; + largestBuffer = Math.max(largestBuffer, buffer.byteLength); + if (kind === "read error") throw new Error("descriptor read failed"); + // Exercise legal short reads; a single read must not mistake them for EOF. + const result = await read( + buffer, + offset, + Math.min(length, SESSION_HISTORY_MAX_LINE_BYTES / 16), + position + ); + totalRead += result.bytesRead; + return result; + }) as typeof handle.read); } - ); + return handle; + }); if (kind === "limit") expect(await storage.read()).toEqual(record("bounded")); else await assert.rejects( @@ -1190,7 +1194,7 @@ describe("inactive real cancellation storage", () => { expect(await storage.read()).toBeNull(); await state.cancel(); const first = (await state.read())!; - const other = new FileCompactionCancellationStorage(foreign, workspaceId); + const other = new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files); expect(await other.read()).toEqual(first); const loaded = (await other.read())!; loaded.nonce = "changed locally"; @@ -1268,7 +1272,7 @@ describe("inactive real cancellation storage", () => { await state.cancel(); const old = (await state.read())!; const successor = new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ); await successor.cancel({ retainUntilReplacement: true }); const expected = await successor.read(); @@ -1410,8 +1414,11 @@ describe("inactive real cancellation storage", () => { storage.mutate(mutation, () => true, mutationCommitted), /not configured/ ); - const refusing = new FileCompactionCancellationStorage(foreign, workspaceId, () => - Promise.resolve(() => Promise.resolve(false)) + const refusing = new FileCompactionCancellationStorage( + foreign, + workspaceId, + foreign.files, + () => Promise.resolve(() => Promise.resolve(false)) ); await assert.rejects( refusing.mutate(mutation, () => true, mutationCommitted), @@ -1419,17 +1426,22 @@ describe("inactive real cancellation storage", () => { ); expect(await storage.read()).toEqual(retained); let current = true; - const verifying = new FileCompactionCancellationStorage(foreign, workspaceId, (witness) => { - expect(witness.nonce).toBe(retained.nonce); - expect(nodeFs.existsSync(historyWriteLockPath(h.config.rootDir, workspaceId))).toBe(false); - return Promise.resolve(async () => { - expect( - await fs.readFile(historyWriteLockPath(h.config.rootDir, workspaceId), "utf8") - ).not.toBe(""); - current = false; - return true; - }); - }); + const verifying = new FileCompactionCancellationStorage( + foreign, + workspaceId, + foreign.files, + (witness) => { + expect(witness.nonce).toBe(retained.nonce); + expect(nodeFs.existsSync(historyWriteLockPath(h.config.rootDir, workspaceId))).toBe(false); + return Promise.resolve(async () => { + expect( + await fs.readFile(historyWriteLockPath(h.config.rootDir, workspaceId), "utf8") + ).not.toBe(""); + current = false; + return true; + }); + } + ); expect(await verifying.mutate(mutation, () => current, mutationCommitted)).toBe("superseded"); expect(await storage.read()).toEqual(retained); // This injected authority exercises the seam only; real accepted-row proof belongs to H2b. @@ -1545,7 +1557,7 @@ describe("inactive real cancellation storage", () => { expect(state.blocksRecovery).toBe(true); expect(await storage.read()).toEqual(old); const successor = new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ); await successor.cancel(); expect(await state.retry()).toBe("superseded"); @@ -1601,7 +1613,7 @@ describe("inactive real cancellation storage", () => { failingRename.mockRestore(); if (change === "Stop") { await new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ).cancel(); } else { await foreign.getContinuousCompactionJournal(workspaceId).advanceGeneration(); @@ -1876,7 +1888,7 @@ describe("inactive real cancellation storage", () => { await fs.writeFile(storage.path, "{bad"); await assert.rejects(storage.read(), MalformedCompactionCancellationError); const successor = new CompactionCancellation( - new FileCompactionCancellationStorage(foreign, workspaceId) + new FileCompactionCancellationStorage(foreign, workspaceId, foreign.files) ); await successor.cancel({ retainUntilReplacement: true }); const committed = mock(() => undefined); diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index e1a30336eef..b6adceeb589 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -1,6 +1,5 @@ import { randomUUID } from "node:crypto"; import { isDeepStrictEqual } from "node:util"; -import { promises as fs, rmSync } from "node:fs"; import * as path from "node:path"; import { z } from "zod"; import { COMPACTION_CANCELLATION_FILE } from "@/constants/continuousCompaction"; @@ -8,7 +7,7 @@ import { SESSION_HISTORY_MAX_LINE_BYTES } from "@/common/constants/contextBudget import { isPlainObject } from "@/common/utils/isPlainObject"; import type { HistoryService } from "./historyService"; import { publishCompactionFile } from "./continuousCompactionJournal"; -import { nodeSessionFiles } from "./sessionFiles/nodeSessionFiles"; +import type { SessionFiles } from "./sessionFiles/sessionFiles"; import { hasAmbiguousResetKeys } from "./historyScanner"; import type { MuxMessage } from "@/common/types/message"; import { log } from "./log"; @@ -179,6 +178,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation constructor( private readonly history: HistoryService, private readonly workspaceId: string, + private readonly files: SessionFiles, // Prepare expensive history evidence outside both locks; the returned verifier only // revalidates it under the lock. A caller-provided nonce alone grants no authority. private readonly prepareVerification?: ( @@ -195,7 +195,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation async read(): Promise { let contents: string; try { - const handle = await fs.open(this.path, "r"); + const handle = await this.files.promises.open(this.path, "r"); try { // Bound allocation even if the file grows; short reads do not imply EOF. const buffer = Buffer.alloc(SESSION_HISTORY_MAX_LINE_BYTES + 1); @@ -343,7 +343,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation contents, isCurrent, // Host files until this storage takes the history's SessionFiles. - nodeSessionFiles, + this.files, () => { frontier.nonce = committed.nonce; // Install inherited retention before cleanup can admit a newer read. @@ -398,7 +398,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation this.path, contents, isCurrent, - nodeSessionFiles, + this.files, () => onCommitted(committed), checkLock )) @@ -430,7 +430,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation : undefined; await checkLock(); if (!isCurrent()) return "superseded"; - rmSync(this.path, { force: true }); + this.files.rmSync(this.path, { force: true }); onCommitted(null, retiredPredecessor); return "applied"; }); @@ -476,7 +476,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation this.path, contents, isCurrent, - nodeSessionFiles, + this.files, () => onCommitted(record), checkLock )) @@ -541,7 +541,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation this.path, serialized.contents, isCurrent, - nodeSessionFiles, + this.files, onCommitted, checkLock )) @@ -549,7 +549,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation throw new Error("Cancellation repair was superseded"); return serialized.record; } - rmSync(this.path, { force: true }); + this.files.rmSync(this.path, { force: true }); onCommitted(); return null; } diff --git a/src/node/services/historyService.ts b/src/node/services/historyService.ts index 01609d1cdcf..9cc5745a20b 100644 --- a/src/node/services/historyService.ts +++ b/src/node/services/historyService.ts @@ -4268,15 +4268,20 @@ export class HistoryService { /** Shared history-lock authority for Stop, replacement acceptance, and recovery. */ getCompactionCancellationStorage(workspaceId: string): FileCompactionCancellationStorage { - return new FileCompactionCancellationStorage(this, workspaceId, async (witness, signal) => { - const evidence = await this.prepareCompactionReplacementWitness( - workspaceId, - witness.nonce, - signal - ); - if (!evidence.success) throw new Error(evidence.error); - return async () => (await evidence.data()) !== null; - }); + return new FileCompactionCancellationStorage( + this, + workspaceId, + this.files, + async (witness, signal) => { + const evidence = await this.prepareCompactionReplacementWitness( + workspaceId, + witness.nonce, + signal + ); + if (!evidence.success) throw new Error(evidence.error); + return async () => (await evidence.data()) !== null; + } + ); } private async captureCompactionReplacementUnderHistoryLock( diff --git a/src/node/services/workspaceGoals.repro.test.ts b/src/node/services/workspaceGoals.repro.test.ts index ec5097b1532..1698949de50 100644 --- a/src/node/services/workspaceGoals.repro.test.ts +++ b/src/node/services/workspaceGoals.repro.test.ts @@ -1,7 +1,7 @@ // Deterministic repros of races in goal continuation and heartbeat delivery. All are fixed: each // test fails at its "Target assertion" with its fix reverted. Each paired control runs the same // harness without the racing step. -import { promises as fsPromises } from "node:fs"; +import * as fsPromises from "node:fs/promises"; import * as path from "path"; import assert from "@/common/utils/assert"; import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"; diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 17f777aac99..e8e15180a30 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -1813,7 +1813,8 @@ async function collectReferencedStagedAttachmentPaths(input: { try { // Streamed: every snapshot archive scans the full append-only history. for (const stagedPath of await extractStagedAttachmentPathsFromFile( - path.join(input.sessionDir, fileName) + path.join(input.sessionDir, fileName), + input.historyService.files )) { paths.add(stagedPath); } @@ -14777,7 +14778,9 @@ export class WorkspaceService // Forking from a prior assistant response intentionally discards any later in-flight // state so the new workspace resumes cleanly at the chosen branch point. - await fsPromises.rm(path.join(newSessionDir, "partial.json"), { force: true }); + await this.historyService.files.promises.rm(path.join(newSessionDir, "partial.json"), { + force: true, + }); if (this.sessionTimingService) { await this.sessionTimingService.clearTimingFile(newWorkspaceId); } else { diff --git a/src/node/utils/attachments/stageWorkspaceAttachment.ts b/src/node/utils/attachments/stageWorkspaceAttachment.ts index 51919206721..006987f17cc 100644 --- a/src/node/utils/attachments/stageWorkspaceAttachment.ts +++ b/src/node/utils/attachments/stageWorkspaceAttachment.ts @@ -19,6 +19,8 @@ import { shellQuote } from "@/common/utils/shell"; import type { Runtime } from "@/node/runtime/Runtime"; import { execBuffered } from "@/node/utils/runtime/helpers"; import { log } from "@/node/services/log"; +import { nodeSessionFiles } from "@/node/services/sessionFiles/nodeSessionFiles"; +import type { SessionFiles } from "@/node/services/sessionFiles/sessionFiles"; import { ensurePrivateDir, isErrnoWithCode } from "@/node/utils/fs"; import { ensureGitInfoExclude } from "@/node/utils/git/ensureGitInfoExclude"; @@ -508,14 +510,19 @@ const HISTORY_SCAN_CHUNK_BYTES = 1024 * 1024; * append-only history never has to be held in memory whole. Paths cannot span lines: they come * from JSON strings, which escape newlines. */ -export async function extractStagedAttachmentPathsFromFile(filePath: string): Promise { +export async function extractStagedAttachmentPathsFromFile( + filePath: string, + // Session history files go through the HistoryService's SessionFiles. Other callers read + // host files. + files: SessionFiles = nodeSessionFiles +): Promise { const paths = new Set(); const scan = (text: string) => { for (const stagedPath of extractStagedAttachmentPathsFromText(text)) { paths.add(stagedPath); } }; - const handle = await fsPromises.open(filePath, "r"); + const handle = await files.promises.open(filePath, "r"); try { const decoder = new StringDecoder("utf8"); const chunk = Buffer.alloc(HISTORY_SCAN_CHUNK_BYTES);