From c474cb0feb4b1dd77c43f379bf60b8f001bf9b40 Mon Sep 17 00:00:00 2001 From: snowykr Date: Thu, 8 Oct 2026 04:46:55 +0000 Subject: [PATCH] fix(storage): admit GC captures from retained file descriptors Scalar preallocation checks missed original link identity and complete named metadata fences. Keep admission, bounded reads and cleanup on the retained descriptor so rejected captures cannot expand allocation or replace errors. Lore-id: 706deaf2 Constraint: stored quotas do not establish invocation budgets or quiescence Constraint: preserve original descriptor and scope generation fences Confidence: high Scope-risk: bounded Tested: 420 tests across 13 GC and storage suites, 3765 assertions; agent checks Not-tested: Windows and Darwin native execution --- docs/session.md | 2 + .../descriptor-bounded-gc-receipt-capture.md | 4 + .../session/internal/managed-session-scope.ts | 5 +- .../internal/managed-session-storage.ts | 71 +- .../managed-gc-retirement-journal.test.ts | 1006 +++++++++++++---- .../coding-agent/test/session-storage.test.ts | 185 +++ 6 files changed, 1006 insertions(+), 267 deletions(-) create mode 100644 packages/coding-agent/changelog.d/descriptor-bounded-gc-receipt-capture.md diff --git a/docs/session.md b/docs/session.md index 784d4e6d4d2..d162fa1c33c 100644 --- a/docs/session.md +++ b/docs/session.md @@ -61,6 +61,8 @@ Existing cleanup-pending history consumers replay incrementally with fresh per-r Each transcript's committed GC retirement history admits at most 50,000 distinct persisted state files and 512 MiB of unique stored receipt bytes, with the same 64 MiB individual-receipt limit. Separately, a scope's shared receipt directory admits at most 50,000 retirement-receipt entries across all transcript targets. Readers, publishers, prepared-target discovery, and protocol inspectors use streamed bounded inventories and no-follow bounded reads; an over-limit target or shared directory fails the whole inspection rather than disappearing from discovery. Replay keeps incremental continuation evidence, checks canonical numeric contiguous attempts, and rejects immutable-field changes or authority expansion, including shrink then reintroduction. Publishers hold a scope-wide lease while admitting the projected new receipt and per-transcript bytes before candidate codec serialization, buffer allocation, or publication, preventing concurrent appends for distinct transcripts from overshooting the shared directory-entry limit. An idempotent nonappend does not consume another stored slot. These stored-history limits are distinct from the fresh per-replay inventory/read/work limits above; no aggregate operation-wide cumulative read, transcript/hash, nested-work, or append-reservation budget is installed. None of these limits establish writer quiescence or physical reclamation. +Leased and read-only GC receipt readers admit stored-history bytes from the original no-follow descriptor's size and link count before allocating receipt buffers. Reads cannot expand beyond that admitted size, and final descriptor and named-file generation checks reject growth or replacement. Reader cleanup preserves the original capture failure when closing also fails. These checks do not establish operation-wide traversal admission or writer quiescence. + Discovery retains each processed scope's original binding and receipt inventory, including authenticated absence, until its terminal consistency check. A new populated scope or journal inserted into an already-processed scope causes refusal instead of silent target omission or adoption of newly observed authority. Original readers are closed on success and refusal. Exchange-placeholder capture and cleanup verification use a bounded native empty-directory proof. It retains no-follow root/parent handles, stops at the first child instead of traversing or hashing foreign payload, and rechecks the original native identity and metadata. An unexpected child or replacement refuses cleanup without changing it. This proof does not establish physical reclamation, revoke live writers, or bound other GC traversal paths. diff --git a/packages/coding-agent/changelog.d/descriptor-bounded-gc-receipt-capture.md b/packages/coding-agent/changelog.d/descriptor-bounded-gc-receipt-capture.md new file mode 100644 index 00000000000..9121c576714 --- /dev/null +++ b/packages/coding-agent/changelog.d/descriptor-bounded-gc-receipt-capture.md @@ -0,0 +1,4 @@ +### Fixed + +- Admit GC receipt-history bytes from the original descriptor before allocation, bound reads to its captured size, and reject named-file generation changes in leased and read-only receipt readers. +- Preserve a managed-file capture failure when reader cleanup also fails. diff --git a/packages/coding-agent/src/session/internal/managed-session-scope.ts b/packages/coding-agent/src/session/internal/managed-session-scope.ts index ff7e7cffb20..832db661f7a 100644 --- a/packages/coding-agent/src/session/internal/managed-session-scope.ts +++ b/packages/coding-agent/src/session/internal/managed-session-scope.ts @@ -2972,8 +2972,9 @@ function readManagedGcReceiptSnapshot( history?: MutableManagedGcReceiptHistory, ): ManagedFileSnapshot | null { try { - return store.readExpectedBounded(relativePath, MANAGED_GC_RECEIPT_MAX_BYTES, size => { - if (history) managedGcAdmitHistoryBytes(history, size, relativePath); + return store.readExpectedBounded(relativePath, MANAGED_GC_RECEIPT_MAX_BYTES, (size, descriptor) => { + if (descriptor.nlink !== 1n || descriptor.size !== size) throw new Error("source_changed"); + if (history) managedGcAdmitHistoryBytes(history, descriptor.size, relativePath); }); } catch (error) { if (error instanceof Error && error.message === "artifact_capacity_exceeded") managedGcJournalCapacity(); diff --git a/packages/coding-agent/src/session/internal/managed-session-storage.ts b/packages/coding-agent/src/session/internal/managed-session-storage.ts index 093a0386c15..ca8adcfcdcb 100644 --- a/packages/coding-agent/src/session/internal/managed-session-storage.ts +++ b/packages/coding-agent/src/session/internal/managed-session-storage.ts @@ -2805,7 +2805,10 @@ export class ManagedSessionDescendantStore { named.isSymbolicLink() || named.dev !== before.dev || named.ino !== before.ino || - named.nlink !== before.nlink + named.nlink !== before.nlink || + named.size !== before.size || + named.mtimeNs !== before.mtimeNs || + named.ctimeNs !== before.ctimeNs ) throw new Error("source_changed"); const rootAfter = fs.lstatSync(this.#baseDir, { bigint: true }); @@ -2851,6 +2854,7 @@ export class ManagedSessionDescendantStore { return { readRange: (start, length) => { if (closed) throw new Error("closed"); + if (start + length > expectedDescriptor.size) throw new Error("range_not_present"); return this.readRangeExpectedSync(relativePath, start, length, expectedDescriptor).bytes; }, close: () => { @@ -2992,7 +2996,7 @@ export class ManagedSessionDescendantStore { readExpectedBounded( relativePath: string, maxBytes: number, - admitSize?: (size: number) => void, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, ): ManagedFileSnapshot | null { if (!Number.isSafeInteger(maxBytes) || maxBytes < 0) throw new Error("invalid_capture_limit"); this.#assertBound(); @@ -3019,17 +3023,37 @@ export class ManagedSessionDescendantStore { if (stat.nlink !== 1n) throw new Error("source_changed"); if (!Number.isSafeInteger(stat.size) || stat.size < 0) throw new Error("source_changed"); if (stat.size > maxBytes) throw new Error("artifact_capacity_exceeded"); - admitSize?.(stat.size); - const bytes = Buffer.alloc(stat.size); const lease = this.openReadLease(this.#relative(resolved), stat); let failed = false; let failure: unknown; + let snapshot: ManagedFileSnapshot | undefined; try { + admitSize?.(stat.size, { + dev: stat.dev, + ino: stat.ino, + nlink: stat.nlink, + size: stat.size, + mtimeNs: stat.mtimeNs, + ctimeNs: stat.ctimeNs, + }); + const bytes = Buffer.alloc(stat.size); for (let offset = 0; offset < bytes.byteLength; ) { const length = Math.min(1024 * 1024, bytes.byteLength - offset); bytes.set(lease.readRange(offset, length), offset); offset += length; } + snapshot = { + bytes, + identity: { + dev: stat.dev, + ino: stat.ino, + nlink: stat.nlink, + size: stat.size, + mtimeNs: stat.mtimeNs, + ctimeNs: stat.ctimeNs, + sha256: createHash("sha256").update(bytes).digest("hex"), + }, + }; } catch (error) { failed = true; failure = error; @@ -3043,18 +3067,8 @@ export class ManagedSessionDescendantStore { } } if (failed) throw failure; - return { - bytes, - identity: { - dev: stat.dev, - ino: stat.ino, - nlink: stat.nlink, - size: stat.size, - mtimeNs: stat.mtimeNs, - ctimeNs: stat.ctimeNs, - sha256: createHash("sha256").update(bytes).digest("hex"), - }, - }; + if (!snapshot) throw new Error("Managed descendant snapshot is unavailable"); + return snapshot; } /** Remove an exact captured file without reopening its pathname as authority. */ @@ -3706,7 +3720,7 @@ export function captureManagedFilePrefixNoFollow(pathname: string, maxBytes: num export function captureManagedFileNoFollowBounded( pathname: string, maxBytes: number, - admitSize?: (size: number) => void, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, ): ManagedFileSnapshot { if (!Number.isSafeInteger(maxBytes) || maxBytes < 0) throw new Error("invalid_capture_limit"); return captureManagedFileNoFollowLimit(pathname, maxBytes, true, admitSize); @@ -3721,9 +3735,12 @@ function captureManagedFileNoFollowLimit( pathname: string, maxBytes?: number, rejectOversized = false, - admitSize?: (size: number) => void, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, ): ManagedFileSnapshot { const fd = fs.openSync(pathname, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK | (fs.constants.O_NOFOLLOW ?? 0)); + let failed = false; + let failure: unknown; + let snapshot: ManagedFileSnapshot | undefined; try { const before = fs.fstatSync(fd, { bigint: true }); if (!before.isFile() || (rejectOversized ? before.nlink !== 1n : before.nlink > 1n)) @@ -3732,7 +3749,7 @@ function captureManagedFileNoFollowLimit( if (!Number.isSafeInteger(fileSize) || fileSize < 0) throw new Error("source_changed"); if (rejectOversized && maxBytes !== undefined && fileSize > maxBytes) throw new Error("artifact_capacity_exceeded"); - admitSize?.(fileSize); + admitSize?.(fileSize, identity(before)); const captureSize = maxBytes === undefined ? fileSize : Math.min(fileSize, maxBytes); const bytes = Buffer.alloc(captureSize); let offset = 0; @@ -3746,10 +3763,22 @@ function captureManagedFileNoFollowLimit( const named = fs.lstatSync(pathname, { bigint: true }); if (!named.isFile() || named.isSymbolicLink() || !sameIdentity(identity(before), identity(named))) throw new Error("source_changed"); - return { bytes, identity: identity(before, createHash("sha256").update(bytes).digest("hex")) }; - } finally { + snapshot = { bytes, identity: identity(before, createHash("sha256").update(bytes).digest("hex")) }; + } catch (error) { + failed = true; + failure = error; + } + try { fs.closeSync(fd); + } catch (error) { + if (!failed) { + failed = true; + failure = error; + } } + if (failed) throw failure; + if (!snapshot) throw new Error("Managed file capture is unavailable"); + return snapshot; } /** Streams a managed file once while retaining only a bounded header prefix and the full descriptor-bound digest. */ diff --git a/packages/coding-agent/test/managed-gc-retirement-journal.test.ts b/packages/coding-agent/test/managed-gc-retirement-journal.test.ts index e73d7ebf9fd..206d4d938ca 100644 --- a/packages/coding-agent/test/managed-gc-retirement-journal.test.ts +++ b/packages/coding-agent/test/managed-gc-retirement-journal.test.ts @@ -37,7 +37,9 @@ import { captureManagedFileNoFollow, captureManagedFileNoFollowBounded, MANAGED_ARTIFACT_MAX_FILES, + MANAGED_ARTIFACT_MAX_TOTAL_BYTES, MANAGED_SESSION_READ_RANGE_MAX_BYTES, + type ManagedFileIdentity, ManagedSessionDescendantStore, } from "../src/session/internal/managed-session-storage"; import { @@ -949,6 +951,50 @@ function readReceiptSuffix(fixture: Fixture, suffix: string): unknown { } } +function managedGcReceiptPath(fixture: Fixture, suffix: string): string { + const key = crypto.createHash("sha256").update(path.resolve(fixture.transcriptPath), "utf8").digest("hex"); + return path.join( + fixture.scope.directoryPath, + ".gjc-managed-session-internal", + "receipts", + `gc-retirement-${key}-${suffix}.json`, + ); +} + +function extendManagedGcReceiptWithWhitespace(pathname: string, targetSize: number): void { + const descriptor = fs.openSync(pathname, "r+"); + try { + const initialSize = fs.fstatSync(descriptor).size; + if (!Number.isSafeInteger(targetSize) || initialSize > targetSize) + throw new Error("managed_gc_receipt_padding_size_invalid"); + const spaces = Buffer.alloc(1024 * 1024, 0x20); + for (let position = initialSize; position < targetSize; ) { + const length = Math.min(spaces.byteLength, targetSize - position); + const written = fs.writeSync(descriptor, spaces, 0, length, position); + if (written !== length) throw new Error("managed_gc_receipt_padding_short_write"); + position += written; + } + } finally { + fs.closeSync(descriptor); + } +} + +function padManagedGcReceiptHistoryTo512MiB(fixture: Fixture): string[] { + const paths = [ + managedGcReceiptPath(fixture, "prepared"), + managedGcReceiptPath(fixture, "artifacts_removed"), + ...Array.from({ length: 6 }, (_, index) => + managedGcReceiptPath(fixture, `owner_pending-${String(index + 1).padStart(8, "0")}`), + ), + ]; + if (paths.length * MANAGED_SESSION_READ_RANGE_MAX_BYTES !== MANAGED_ARTIFACT_MAX_TOTAL_BYTES) + throw new Error("fixture_gc_receipt_history_limit_mismatch"); + for (const pathname of paths) extendManagedGcReceiptWithWhitespace(pathname, MANAGED_SESSION_READ_RANGE_MAX_BYTES); + const storedBytes = paths.reduce((total, pathname) => total + fs.statSync(pathname).size, 0); + if (storedBytes !== MANAGED_ARTIFACT_MAX_TOTAL_BYTES) throw new Error("fixture_gc_receipt_history_not_at_limit"); + return paths; +} + async function cleanupRetryCandidate(fixture: Fixture) { await Bun.write( fixture.transcriptPath, @@ -1185,7 +1231,7 @@ describe("bounded cleanup receipt replay", () => { expect(allocate.mock.calls.some(([size]) => size === 64 * 1024 * 1024 + 1)).toBe(false); }); - it("refuses simulated aggregate descriptor bytes before allocating the ninth receipt", async () => { + it("refuses the ninth actual cleanup receipt after eight real 64 MiB admissions", async () => { const { fixture, tombstonePath } = await injectedCleanupHistory(9); const directory = path.dirname(tombstonePath); const names = fs @@ -1195,29 +1241,32 @@ describe("bounded cleanup receipt replay", () => { (left, right) => Number(/-([0-9]+)\.json$/u.exec(left)?.[1]) - Number(/-([0-9]+)\.json$/u.exec(right)?.[1]), ); const receiptPaths = new Set(names.map(name => path.join(directory, name))); - const actualSizes = new Map([...receiptPaths].map(pathname => [pathname, fs.statSync(pathname).size])); - const admitted = new Set(); - let allocatingPath: string | undefined; + for (const name of names.slice(0, 8)) + extendManagedGcReceiptWithWhitespace(path.join(directory, name), MANAGED_SESSION_READ_RANGE_MAX_BYTES); + const actualSizes = new Map( + [...receiptPaths].map(pathname => { + const stat = fs.statSync(pathname, { bigint: true }); + return [ + pathname, + { dev: stat.dev, ino: stat.ino, size: stat.size, mtimeNs: stat.mtimeNs, ctimeNs: stat.ctimeNs }, + ] as const; + }), + ); + const before = [...actualSizes.entries()]; const readers: Array<{ matchingEntries: number; allocations: number; inventoryClosed: boolean }> = []; let activeReader: (typeof readers)[number] | undefined; - const receiptSize = 64 * 1024 * 1024; - const originalFstat = fs.fstatSync.bind(fs); - const originalLstat = fs.lstatSync.bind(fs); + const receiptSize = MANAGED_SESSION_READ_RANGE_MAX_BYTES; const originalOpendir = fs.opendirSync.bind(fs); const originalOpen = fs.openSync.bind(fs); const originalClose = fs.closeSync.bind(fs); + const originalRead = fs.readSync.bind(fs); const originalAlloc = Buffer.alloc.bind(Buffer); const descriptorPaths = new Map(); - const withReceiptSize = (stat: T): T => - new Proxy(stat, { - get: (value, property) => - property === "size" - ? typeof value.size === "bigint" - ? BigInt(receiptSize) - : receiptSize - : Reflect.get(value, property, value), - }) as T; - vi.spyOn(fs, "opendirSync").mockImplementation(pathname => { + const allocated = new Map(); + const bytesRequested = new Map(); + const bytesRead = new Map(); + const closes = new Map(); + const opendir = vi.spyOn(fs, "opendirSync").mockImplementation(pathname => { const handle = originalOpendir(pathname); if (pathname !== directory) return handle; const reader = { matchingEntries: 0, allocations: 0, inventoryClosed: false }; @@ -1235,51 +1284,100 @@ describe("bounded cleanup receipt replay", () => { }, } as fs.Dir; }); - vi.spyOn(fs, "openSync").mockImplementation(((pathname: fs.PathLike, flags: string | number, mode?: number) => { + const open = vi.spyOn(fs, "openSync").mockImplementation((( + pathname: fs.PathLike, + flags: string | number, + mode?: number, + ) => { const fd = originalOpen(pathname, flags, mode); - if (typeof pathname === "string") descriptorPaths.set(fd, path.resolve(pathname)); + if (typeof pathname === "string" && receiptPaths.has(path.resolve(pathname))) + descriptorPaths.set(fd, path.resolve(pathname)); return fd; }) as typeof fs.openSync); - vi.spyOn(fs, "closeSync").mockImplementation(fd => { + const close = vi.spyOn(fs, "closeSync").mockImplementation(fd => { + const pathname = descriptorPaths.get(fd); try { originalClose(fd); } finally { descriptorPaths.delete(fd); + if (pathname) closes.set(pathname, (closes.get(pathname) ?? 0) + 1); } }); - vi.spyOn(fs, "fstatSync").mockImplementation(((fd: number, options?: { bigint?: boolean }) => { + const read = vi.spyOn(fs, "readSync").mockImplementation((( + fd: number, + buffer: NodeJS.ArrayBufferView, + offset: number, + length: number, + position: number | null, + ) => { const pathname = descriptorPaths.get(fd); - const stat = originalFstat(fd, options as never); - if (!pathname || !receiptPaths.has(pathname)) return stat; - allocatingPath = pathname; - return withReceiptSize(stat); - }) as typeof fs.fstatSync); - vi.spyOn(fs, "lstatSync").mockImplementation(((pathname: fs.PathLike, options?: { bigint?: boolean }) => { - const stat = originalLstat(pathname, options as never); - return typeof pathname === "string" && receiptPaths.has(pathname) ? withReceiptSize(stat) : stat; - }) as typeof fs.lstatSync); - vi.spyOn(Buffer, "alloc").mockImplementation((( + if (pathname) bytesRequested.set(pathname, (bytesRequested.get(pathname) ?? 0) + length); + const count = originalRead(fd, buffer, offset, length, position); + if (pathname) bytesRead.set(pathname, (bytesRead.get(pathname) ?? 0) + count); + return count; + }) as typeof fs.readSync); + const allocate = vi.spyOn(Buffer, "alloc").mockImplementation((( size: number, fill?: string | Uint8Array | number, encoding?: BufferEncoding, ) => { if (size !== receiptSize) return originalAlloc(size, fill as number, encoding); - if (!allocatingPath) throw new Error("cleanup_receipt_allocation_path_missing"); - const pathname = allocatingPath; - allocatingPath = undefined; - admitted.add(pathname); + const pathname = [...descriptorPaths.values()].find(candidate => receiptPaths.has(candidate)); + if (!pathname) throw new Error("cleanup_receipt_allocation_descriptor_missing"); + allocated.set(pathname, (allocated.get(pathname) ?? 0) + 1); if (activeReader?.inventoryClosed) activeReader.allocations++; - return originalAlloc(actualSizes.get(pathname) ?? size); + return originalAlloc(size, fill as number, encoding); }) as typeof Buffer.alloc); const inspect = managedGcProtocolScopeInspectorForScope(fixture.scope); - await expect(inspect([protocolInputFor(fixture.scope)])).rejects.toThrow("managed_gc_journal_capacity_exceeded"); + try { + await expect(inspect([protocolInputFor(fixture.scope)])).rejects.toThrow( + "managed_gc_journal_capacity_exceeded", + ); + } finally { + allocate.mockRestore(); + read.mockRestore(); + close.mockRestore(); + open.mockRestore(); + opendir.mockRestore(); + } expect(names.length).toBeGreaterThan(8); expect(readers).toEqual([ - { matchingEntries: names.length, inventoryClosed: true, allocations: 0 }, - { matchingEntries: names.length, inventoryClosed: true, allocations: 0 }, - { matchingEntries: names.length, inventoryClosed: true, allocations: 8 }, + { matchingEntries: names.length, allocations: 0, inventoryClosed: true }, + { matchingEntries: names.length, allocations: 0, inventoryClosed: true }, + { matchingEntries: names.length, allocations: 8, inventoryClosed: true }, ]); - expect(admitted.size).toBe(8); + expect([...allocated.values()].reduce((total, count) => total + count, 0)).toBe(8); + const openedReceiptPaths = new Set(names.slice(0, 9).map(name => path.join(directory, name))); + for (const pathname of receiptPaths) { + const initialSize = actualSizes.get(pathname)?.size; + if (!openedReceiptPaths.has(pathname)) { + expect(closes.has(pathname)).toBe(false); + expect(allocated.has(pathname)).toBe(false); + expect(bytesRequested.has(pathname)).toBe(false); + expect(bytesRead.has(pathname)).toBe(false); + } else { + expect(closes.get(pathname)).toBe(1); + if (initialSize === BigInt(receiptSize)) { + expect(allocated.get(pathname)).toBe(1); + expect(bytesRequested.get(pathname)).toBe(receiptSize); + expect(bytesRead.get(pathname)).toBe(receiptSize); + } else { + expect(initialSize).toBeLessThan(BigInt(receiptSize)); + expect(allocated.has(pathname)).toBe(false); + expect(bytesRequested.has(pathname)).toBe(false); + expect(bytesRead.has(pathname)).toBe(false); + } + } + } + expect( + [...receiptPaths].map(pathname => { + const stat = fs.statSync(pathname, { bigint: true }); + return [ + pathname, + { dev: stat.dev, ino: stat.ino, size: stat.size, mtimeNs: stat.mtimeNs, ctimeNs: stat.ctimeNs }, + ] as const; + }), + ).toEqual(before); }, 60_000); it("rejects noncanonical suffixes, filename-record mismatches, and gaps", async () => { @@ -1308,6 +1406,188 @@ describe("bounded cleanup receipt replay", () => { }); }); +describe("managed GC receipt descriptor admission", () => { + it("rejects an actual oversized receipt from descriptor size before allocating its contents", async () => { + const fixture = makeFixture(); + await publishManagedGcSessionRetirementReceipt(fixture.scope, preparedReceipt(fixture)); + const receiptPath = managedGcReceiptPath(fixture, "prepared"); + const receiptBytes = fs.readFileSync(receiptPath); + const descriptor = fs.openSync(receiptPath, "r+"); + try { + fs.ftruncateSync(descriptor, MANAGED_SESSION_READ_RANGE_MAX_BYTES + 1); + } finally { + fs.closeSync(descriptor); + } + const before = fs.statSync(receiptPath, { bigint: true }); + const allocations = spyOn(Buffer, "alloc"); + await expect( + readManagedGcSessionRetirementReceiptReadOnly(fixture.scope, fixture.transcriptPath), + ).rejects.toThrow("managed_gc_journal_capacity_exceeded"); + const after = fs.statSync(receiptPath, { bigint: true }); + expect(after.size).toBe(BigInt(MANAGED_SESSION_READ_RANGE_MAX_BYTES + 1)); + expect(after.dev).toBe(before.dev); + expect(after.ino).toBe(before.ino); + expect(after.nlink).toBe(1n); + expect(allocations.mock.calls.some(([size]) => size >= MANAGED_SESSION_READ_RANGE_MAX_BYTES)).toBe(false); + const current = Buffer.alloc(receiptBytes.byteLength); + const readDescriptor = fs.openSync(receiptPath, "r"); + try { + expect(fs.readSync(readDescriptor, current, 0, current.byteLength, 0)).toBe(current.byteLength); + } finally { + fs.closeSync(readDescriptor); + } + expect(current).toEqual(receiptBytes); + }); + + it("rejects a linked actual receipt before allocation under the producer's real scope lock", async () => { + const fixture = makeFixture(); + await publishManagedGcSessionRetirementReceipt(fixture.scope, preparedReceipt(fixture)); + const receiptPath = managedGcReceiptPath(fixture, "prepared"); + const linkedPath = path.join(fixture.temporaryRoot, "linked-prepared-receipt.json"); + const lockPath = path.join( + fixture.scope.directoryPath, + ".gjc-managed-session-internal", + "locks", + `${MANAGED_GC_SCOPE_LOCK_NAME}.lock`, + ); + const before = fs.readFileSync(receiptPath); + const beforeStat = fs.statSync(receiptPath, { bigint: true }); + fs.linkSync(receiptPath, linkedPath); + let openedScopeLock = false; + let receiptOpenCount = 0; + const originalOpen = fs.openSync.bind(fs); + const open = vi.spyOn(fs, "openSync").mockImplementation((( + pathname: fs.PathLike, + flags: string | number, + mode?: number, + ) => { + const descriptor = originalOpen(pathname, flags, mode); + if (typeof pathname === "string" && path.resolve(pathname) === receiptPath) receiptOpenCount++; + if (typeof pathname === "string" && path.resolve(pathname) === lockPath) openedScopeLock = true; + return descriptor; + }) as typeof fs.openSync); + const allocations = spyOn(Buffer, "alloc"); + try { + const next: ManagedGcSessionRetirementReceipt = { + ...preparedReceipt(fixture), + state: "artifacts_removed", + artifactsRemoved: true, + }; + await expect(publishManagedGcSessionRetirementReceipt(fixture.scope, next)).rejects.toThrow("hard_link"); + expect(openedScopeLock).toBe(true); + expect(receiptOpenCount).toBe(0); + expect(allocations.mock.calls.some(([size]) => size === Number(beforeStat.size))).toBe(false); + expect(fs.readFileSync(receiptPath)).toEqual(before); + expect(fs.readFileSync(linkedPath)).toEqual(before); + const afterStat = fs.statSync(receiptPath, { bigint: true }); + expect(afterStat.dev).toBe(beforeStat.dev); + expect(afterStat.ino).toBe(beforeStat.ino); + expect(afterStat.size).toBe(beforeStat.size); + expect(afterStat.nlink).toBe(2n); + } finally { + allocations.mockRestore(); + open.mockRestore(); + } + }); + + it("closes a real receipt inventory after refusing its 50,001st physical entry", async () => { + const fixture = makeFixture(); + await publishManagedGcSessionRetirementReceipt(fixture.scope, preparedReceipt(fixture)); + const receiptDirectory = path.dirname(managedGcReceiptPath(fixture, "prepared")); + for (let index = 0; index < MANAGED_ARTIFACT_MAX_FILES; index++) { + const name = `gc-retirement-untrusted-${String(index).padStart(8, "0")}.entry`; + fs.writeFileSync(path.join(receiptDirectory, name), "", { flag: "wx", mode: 0o600 }); + } + expect(fs.readdirSync(receiptDirectory)).toHaveLength(MANAGED_ARTIFACT_MAX_FILES + 1); + const originalOpendir = fs.opendirSync.bind(fs); + let entriesRead = 0; + let inventoryCloseCount = 0; + const opendir = vi.spyOn(fs, "opendirSync").mockImplementation(pathname => { + const directory = originalOpendir(pathname); + if (path.resolve(String(pathname)) !== receiptDirectory) return directory; + return { + readSync: () => { + const entry = directory.readSync(); + if (entry) entriesRead++; + return entry; + }, + closeSync: () => { + directory.closeSync(); + inventoryCloseCount++; + }, + } as fs.Dir; + }); + const allocations = spyOn(Buffer, "alloc"); + try { + await expect( + readManagedGcSessionRetirementReceiptReadOnly(fixture.scope, fixture.transcriptPath), + ).rejects.toThrow("managed_gc_journal_capacity_exceeded"); + expect(entriesRead).toBe(MANAGED_ARTIFACT_MAX_FILES + 1); + expect(inventoryCloseCount).toBe(1); + expect(allocations.mock.calls.some(([size]) => size >= MANAGED_SESSION_READ_RANGE_MAX_BYTES)).toBe(false); + } finally { + allocations.mockRestore(); + opendir.mockRestore(); + } + }); + + it("admits actual receipt descriptors through 512 MiB and refuses the overflowing read before allocation", async () => { + const fixture = makeFixture(); + await publishManagedGcSessionRetirementReceipt(fixture.scope, preparedReceipt(fixture)); + await publishManagedGcSessionRetirementReceipt(fixture.scope, { + ...preparedReceipt(fixture), + state: "artifacts_removed", + artifactsRemoved: true, + }); + for (let attempt = 1; attempt <= 8; attempt++) { + await publishManagedGcSessionRetirementReceipt( + fixture.scope, + pendingReceipt(fixture, `actual-byte-budget-${attempt}`), + ); + expect(readReceiptSuffix(fixture, `owner_pending-${String(attempt).padStart(8, "0")}`)).toMatchObject({ + state: "owner_pending", + ownerRetirementAttempt: attempt, + }); + } + const pendingPaths = Array.from({ length: 8 }, (_, index) => + managedGcReceiptPath(fixture, `owner_pending-${String(index + 1).padStart(8, "0")}`), + ); + for (const receiptPath of pendingPaths.slice(0, 7)) + extendManagedGcReceiptWithWhitespace(receiptPath, MANAGED_SESSION_READ_RANGE_MAX_BYTES); + const finalDescriptor = fs.openSync(pendingPaths[7]!, "r+"); + try { + fs.ftruncateSync(finalDescriptor, MANAGED_SESSION_READ_RANGE_MAX_BYTES); + } finally { + fs.closeSync(finalDescriptor); + } + const preparedBytes = fs.statSync(managedGcReceiptPath(fixture, "prepared")).size; + const artifactsBytes = fs.statSync(managedGcReceiptPath(fixture, "artifacts_removed")).size; + expect(7 * MANAGED_SESSION_READ_RANGE_MAX_BYTES + preparedBytes + artifactsBytes).toBeLessThanOrEqual( + MANAGED_ARTIFACT_MAX_TOTAL_BYTES, + ); + expect(8 * MANAGED_SESSION_READ_RANGE_MAX_BYTES + preparedBytes + artifactsBytes).toBeGreaterThan( + MANAGED_ARTIFACT_MAX_TOTAL_BYTES, + ); + for (const receiptPath of pendingPaths) { + const stat = fs.statSync(receiptPath, { bigint: true }); + expect(stat.size).toBe(BigInt(MANAGED_SESSION_READ_RANGE_MAX_BYTES)); + expect(stat.nlink).toBe(1n); + } + const allocations = spyOn(Buffer, "alloc"); + try { + await expect( + readManagedGcSessionRetirementReceiptReadOnly(fixture.scope, fixture.transcriptPath), + ).rejects.toThrow("managed_gc_journal_capacity_exceeded"); + const fullReceiptAllocations = allocations.mock.calls + .filter(([size]) => size === MANAGED_SESSION_READ_RANGE_MAX_BYTES) + .map(([size]) => size); + expect(fullReceiptAllocations).toEqual(Array(7).fill(MANAGED_SESSION_READ_RANGE_MAX_BYTES)); + } finally { + allocations.mockRestore(); + } + }); +}); + describe("managed GC retirement journal", () => { it("publishes a canonical receipt exactly at the UTF-8 byte limit and rejects limit plus one before encoding", async () => { const seed = makeFixture(); @@ -1993,7 +2273,7 @@ describe("managed GC retirement journal", () => { } }); - it("budgets discovery and inspector GC receipt rereads per target across genuine sibling scopes", async () => { + it("admits actual 512 MiB receipt histories for sibling discovery", async () => { const fixture = makeFixture(); const sibling = makeSiblingFixture(fixture, "managed-gc-budget-sibling"); for (const current of [fixture, sibling]) { @@ -2006,29 +2286,42 @@ describe("managed GC retirement journal", () => { for (let attempt = 1; attempt <= 6; attempt++) await publishManagedGcSessionRetirementReceipt( current.scope, - pendingReceipt(current, `sibling-pressure-${attempt}`), + pendingReceipt(current, `sibling-actual-boundary-${attempt}`), ); } + const paddedPaths = [ + ...padManagedGcReceiptHistoryTo512MiB(fixture), + ...padManagedGcReceiptHistoryTo512MiB(sibling), + ]; + for (const pathname of paddedPaths) { + const stat = fs.statSync(pathname, { bigint: true }); + expect(stat.size).toBe(BigInt(MANAGED_SESSION_READ_RANGE_MAX_BYTES)); + expect(stat.nlink).toBe(1n); + } const targetKeys = [fixture, sibling].map(current => crypto.createHash("sha256").update(path.resolve(current.transcriptPath), "utf8").digest("hex"), ); - const projectedReads = new Map(); + const actualReads = new Map(); + const actualBytesByTarget = new Map(); const originalReadExpectedBounded = ManagedSessionDescendantStore.prototype.readExpectedBounded; - const pressure = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation( + const boundedRead = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation( function ( this: ManagedSessionDescendantStore, relativePath: string, maxBytes: number, - admitSize?: (size: number) => void, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, ) { - return originalReadExpectedBounded.call(this, relativePath, maxBytes, size => { + return originalReadExpectedBounded.call(this, relativePath, maxBytes, (size, descriptor) => { const key = /^\.gjc-managed-session-internal\/receipts\/gc-retirement-([a-f0-9]{64})-/u.exec( relativePath, )?.[1]; if (key && targetKeys.includes(key)) { - projectedReads.set(key, (projectedReads.get(key) ?? 0) + 1); - admitSize?.(64 * 1024 * 1024); - } else admitSize?.(size); + actualReads.set(key, (actualReads.get(key) ?? 0) + 1); + expect(descriptor.size).toBe(size); + expect(size).toBe(MANAGED_SESSION_READ_RANGE_MAX_BYTES); + actualBytesByTarget.set(key, (actualBytesByTarget.get(key) ?? 0) + size); + } + admitSize?.(size, descriptor); }); }, ); @@ -2039,26 +2332,21 @@ describe("managed GC retirement journal", () => { sessionsRoot: fixture.sessionsRoot, }); expect(discovered).toHaveLength(2); - expect([...projectedReads.values()].sort((left, right) => left - right)).toEqual([8, 8]); - expect([...projectedReads.values()].reduce((sum, count) => sum + count, 0) * 64 * 1024 * 1024).toBe( - 1024 * 1024 * 1024, - ); - expect(discoveryAllocations.mock.calls.some(([size]) => size >= 64 * 1024 * 1024)).toBe(false); - - projectedReads.clear(); - const inspect = managedGcProtocolScopeInspectorForScope(fixture.scope); - const inspected = await inspect([protocolInputFor(fixture.scope), protocolInputFor(sibling.scope)]); - expect(inspected).toHaveLength(2); - expect([...projectedReads.values()].length).toBe(2); - expect([...projectedReads.values()].every(count => count === 32)).toBe(true); - expect(discoveryAllocations.mock.calls.some(([size]) => size >= 64 * 1024 * 1024)).toBe(false); + expect([...actualReads.values()].sort((left, right) => left - right)).toEqual([8, 8]); + expect([...actualBytesByTarget.values()].sort((left, right) => left - right)).toEqual([ + MANAGED_ARTIFACT_MAX_TOTAL_BYTES, + MANAGED_ARTIFACT_MAX_TOTAL_BYTES, + ]); + expect( + discoveryAllocations.mock.calls.filter(([size]) => size === MANAGED_SESSION_READ_RANGE_MAX_BYTES), + ).toHaveLength(16); } finally { - pressure.mockRestore(); + boundedRead.mockRestore(); discoveryAllocations.mockRestore(); } - }); + }, 60_000); - it("rejects the ninth individually admissible state per target before discovery or inspector allocation", async () => { + it("admits eight actual 64 MiB sibling states and rejects the ninth actual state before allocation", async () => { const fixture = makeFixture(); const sibling = makeSiblingFixture(fixture, "managed-gc-ninth-sibling"); for (const current of [fixture, sibling]) { @@ -2076,32 +2364,72 @@ describe("managed GC retirement journal", () => { } const target = fixture.scope.directoryName.localeCompare(sibling.scope.directoryName) > 0 ? fixture : sibling; const unaffected = target === fixture ? sibling : fixture; + const targetKey = crypto.createHash("sha256").update(path.resolve(target.transcriptPath), "utf8").digest("hex"); + const unaffectedKey = crypto + .createHash("sha256") + .update(path.resolve(unaffected.transcriptPath), "utf8") + .digest("hex"); await publishManagedGcSessionRetirementReceipt(target.scope, pendingReceipt(target, "target_ninth_state")); + const paddedHistoryPaths = [ + ...padManagedGcReceiptHistoryTo512MiB(target), + ...padManagedGcReceiptHistoryTo512MiB(unaffected), + ]; + for (const pathname of paddedHistoryPaths) + expect(fs.statSync(pathname, { bigint: true }).size).toBe(BigInt(MANAGED_SESSION_READ_RANGE_MAX_BYTES)); const before = [managedReceiptInventory(fixture.scope), managedReceiptInventory(sibling.scope)]; const keys = [target, unaffected].map(current => crypto.createHash("sha256").update(path.resolve(current.transcriptPath), "utf8").digest("hex"), ); const attempts = new Map(); + const actualAdmissions = new Map(); + const allocatedReceipts = new Map(); + let activeReceiptPath: string | undefined; const originalReadExpectedBounded = ManagedSessionDescendantStore.prototype.readExpectedBounded; + const originalAlloc = Buffer.alloc.bind(Buffer); const boundedRead = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation( function ( this: ManagedSessionDescendantStore, relativePath: string, maxBytes: number, - admitSize?: (size: number) => void, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, ) { - return originalReadExpectedBounded.call(this, relativePath, maxBytes, size => { - const key = /^\.gjc-managed-session-internal\/receipts\/gc-retirement-([a-f0-9]{64})-/u.exec( - relativePath, - )?.[1]; - if (key && keys.includes(key)) { - attempts.set(key, (attempts.get(key) ?? 0) + 1); - admitSize?.(64 * 1024 * 1024); - } else admitSize?.(size); - }); + activeReceiptPath = relativePath; + try { + return originalReadExpectedBounded.call(this, relativePath, maxBytes, (size, descriptor) => { + const key = /^\.gjc-managed-session-internal\/receipts\/gc-retirement-([a-f0-9]{64})-/u.exec( + relativePath, + )?.[1]; + if (key && keys.includes(key)) { + attempts.set(key, (attempts.get(key) ?? 0) + 1); + expect(descriptor.size).toBe(size); + expect(descriptor.nlink).toBe(1n); + const entryKey = `${key}:${relativePath}`; + const entry = actualAdmissions.get(entryKey) ?? { count: 0, bytes: 0 }; + entry.count++; + entry.bytes += size; + actualAdmissions.set(entryKey, entry); + } + admitSize?.(size, descriptor); + }); + } finally { + activeReceiptPath = undefined; + } }, ); - const allocations = spyOn(Buffer, "alloc"); + const allocations = spyOn(Buffer, "alloc").mockImplementation((( + size: number, + fill?: string | Uint8Array | number, + encoding?: BufferEncoding, + ) => { + if (activeReceiptPath && size === MANAGED_SESSION_READ_RANGE_MAX_BYTES) { + const key = /^\.gjc-managed-session-internal\/receipts\/gc-retirement-([a-f0-9]{64})-/u.exec( + activeReceiptPath, + )?.[1]; + const allocationKey = `${key}:${activeReceiptPath}`; + allocatedReceipts.set(allocationKey, (allocatedReceipts.get(allocationKey) ?? 0) + 1); + } + return originalAlloc(size, fill, encoding); + }) as typeof Buffer.alloc); try { await expect( discoverManagedGcSessionRetirementReceipts({ @@ -2110,60 +2438,32 @@ describe("managed GC retirement journal", () => { }), ).rejects.toThrow("managed_gc_journal_capacity_exceeded"); expect([...attempts.values()].sort((left, right) => left - right)).toEqual([8, 9]); - expect(allocations.mock.calls.some(([size]) => size >= 64 * 1024 * 1024)).toBe(false); + expect([...actualAdmissions.entries()].filter(([entry]) => entry.startsWith(`${targetKey}:`))).toHaveLength(9); + expect( + [...actualAdmissions.entries()].filter(([entry]) => entry.startsWith(`${unaffectedKey}:`)), + ).toHaveLength(8); + for (const [entryKey, admission] of actualAdmissions) { + const refused = entryKey.endsWith("owner_pending-00000007.json"); + expect(admission.count).toBe(1); + if (refused) expect(admission.bytes).toBeLessThan(MANAGED_SESSION_READ_RANGE_MAX_BYTES); + else expect(admission.bytes).toBe(MANAGED_SESSION_READ_RANGE_MAX_BYTES); + } + expect([...allocatedReceipts.values()].reduce((total, count) => total + count, 0)).toBe(16); + expect( + [...allocatedReceipts.keys()].some( + entryKey => + entryKey === + `${targetKey}:.gjc-managed-session-internal/receipts/gc-retirement-${targetKey}-owner_pending-00000007.json`, + ), + ).toBe(false); expect([managedReceiptInventory(fixture.scope), managedReceiptInventory(sibling.scope)]).toEqual(before); } finally { boundedRead.mockRestore(); allocations.mockRestore(); } + }, 60_000); - attempts.clear(); - const protocolPath = path.join(target.scope.directoryPath, ".gjc-managed-session-internal"); - const originalOpendirSync = fs.opendirSync.bind(fs); - let inspectorStarted = false; - const opendir = vi.spyOn(fs, "opendirSync").mockImplementation(pathname => { - if (path.resolve(String(pathname)) === protocolPath) inspectorStarted = true; - return originalOpendirSync(pathname); - }); - const inspectorRead = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation( - function ( - this: ManagedSessionDescendantStore, - relativePath: string, - maxBytes: number, - admitSize?: (size: number) => void, - ) { - return originalReadExpectedBounded.call(this, relativePath, maxBytes, size => { - const key = /^\.gjc-managed-session-internal\/receipts\/gc-retirement-([a-f0-9]{64})-/u.exec( - relativePath, - )?.[1]; - if (inspectorStarted && key && keys.includes(key)) { - attempts.set(key, (attempts.get(key) ?? 0) + 1); - admitSize?.(64 * 1024 * 1024); - } else admitSize?.(size); - }); - }, - ); - const inspectorAllocations = spyOn(Buffer, "alloc"); - try { - await expect( - managedGcProtocolScopeInspectorForScope(fixture.scope)([ - protocolInputFor(target.scope), - protocolInputFor(unaffected.scope), - ]), - ).rejects.toThrow("managed_gc_journal_capacity_exceeded"); - } finally { - opendir.mockRestore(); - inspectorRead.mockRestore(); - inspectorAllocations.mockRestore(); - } - expect(inspectorStarted).toBe(true); - expect(attempts.get(keys[0]!)).toBe(9); - expect(attempts.get(keys[1]!)).toBeUndefined(); - expect(inspectorAllocations.mock.calls.some(([size]) => size >= 64 * 1024 * 1024)).toBe(false); - expect([managedReceiptInventory(fixture.scope), managedReceiptInventory(sibling.scope)]).toEqual(before); - }); - - it("refuses inspector history overflow independently in initial comparison and final phases before real receipt allocation", async () => { + it("keeps genuine receipt rereads healthy under the default descriptor-size cap", async () => { const fixture = makeFixture(); const sibling = makeSiblingFixture(fixture, "managed-gc-independent-phase-sibling"); const target = fixture.scope.directoryName.localeCompare(sibling.scope.directoryName) > 0 ? fixture : sibling; @@ -2175,81 +2475,301 @@ describe("managed GC retirement journal", () => { state: "artifacts_removed", artifactsRemoved: true, }); - for (let attempt = 1; attempt <= (current === target ? 7 : 6); attempt++) + for (let attempt = 1; attempt <= 6; attempt++) await publishManagedGcSessionRetirementReceipt( current.scope, pendingReceipt(current, `phase-seed-${attempt}`), ); } const targetKey = crypto.createHash("sha256").update(path.resolve(target.transcriptPath), "utf8").digest("hex"); - const targetPrefix = `.gjc-managed-session-internal/receipts/gc-retirement-${targetKey}-`; + const unaffectedKey = crypto + .createHash("sha256") + .update(path.resolve(unaffected.transcriptPath), "utf8") + .digest("hex"); const before = [managedReceiptInventory(target.scope), managedReceiptInventory(unaffected.scope)]; + const expectedPaths = new Map>([ + [targetKey, new Set(before[0]!.map(entry => `.gjc-managed-session-internal/receipts/${entry.name}`))], + [unaffectedKey, new Set(before[1]!.map(entry => `.gjc-managed-session-internal/receipts/${entry.name}`))], + ]); + const readsByTarget = new Map(); + const capsByTarget = new Map(); const originalRead = ManagedSessionDescendantStore.prototype.readExpectedBounded; - const originalAllocate = Buffer.alloc; + const read = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation(function ( + this: ManagedSessionDescendantStore, + relativePath: string, + maxBytes: number, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, + ) { + const key = /^\.gjc-managed-session-internal\/receipts\/gc-retirement-([a-f0-9]{64})-/u.exec( + relativePath, + )?.[1]; + if (key === targetKey || key === unaffectedKey) { + const caps = capsByTarget.get(key) ?? []; + caps.push(maxBytes); + capsByTarget.set(key, caps); + } + return originalRead.call(this, relativePath, maxBytes, (size, descriptor) => { + if (key === targetKey || key === unaffectedKey) { + const scope = key === targetKey ? target.scope : unaffected.scope; + const actualSize = Number( + fs.statSync(path.join(scope.directoryPath, relativePath), { bigint: true }).size, + ); + expect(size).toBe(actualSize); + expect(descriptor.size).toBe(size); + expect(descriptor.nlink).toBe(1n); + expect(expectedPaths.get(key)?.has(relativePath)).toBe(true); + const reads = readsByTarget.get(key) ?? []; + reads.push(relativePath); + readsByTarget.set(key, reads); + } + admitSize?.(size, descriptor); + }); + }); + try { + const snapshots = await managedGcProtocolScopeInspectorForScope(fixture.scope)([ + protocolInputFor(target.scope), + protocolInputFor(unaffected.scope), + ]); + expect(snapshots).toHaveLength(2); + } finally { + read.mockRestore(); + } + for (const [key, paths] of expectedPaths) { + const reads = readsByTarget.get(key) ?? []; + const caps = capsByTarget.get(key) ?? []; + const passSize = paths.size; + expect(reads).toHaveLength(passSize * 4); + expect(caps).toEqual(Array(passSize * 4).fill(MANAGED_SESSION_READ_RANGE_MAX_BYTES)); + for (let pass = 0; pass < 4; pass++) { + expect(new Set(reads.slice(pass * passSize, (pass + 1) * passSize))).toEqual(paths); + } + } + expect([managedReceiptInventory(target.scope), managedReceiptInventory(unaffected.scope)]).toEqual(before); + }); + + it("rejects each inspector-phase receipt under a smaller real read cap before allocation", async () => { for (const phase of [ - { name: "initial", earlierReads: 9 }, - { name: "comparison", earlierReads: 18 }, - { name: "final", earlierReads: 27 }, - ]) { - let reads = 0; - const admissions: Array<{ relativePath: string; actualSize: number; pressured: boolean }> = []; - const allocated: Array<{ relativePath: string; actualSize: number; pressured: boolean; size: number }> = []; - let activeRead: { relativePath: string; actualSize: number; pressured: boolean } | undefined; - // Negative budget pressure only: the original reader opens genuine files and allocates their actual size. + { name: "initial", earlierReads: 9, samePathRead: 2 }, + { name: "comparison", earlierReads: 18, samePathRead: 3 }, + { name: "final", earlierReads: 27, samePathRead: 4 }, + ] as const) { + const fixture = makeFixture(`managed-gc-phase-${phase.name}`); + const sibling = makeSiblingFixture(fixture, `managed-gc-phase-sibling-${phase.name}`); + for (const current of [fixture, sibling]) { + await publishManagedGcSessionRetirementReceipt(current.scope, preparedReceipt(current)); + await publishManagedGcSessionRetirementReceipt(current.scope, { + ...preparedReceipt(current), + state: "artifacts_removed", + artifactsRemoved: true, + }); + const pendingCount = current === fixture ? 7 : 6; + for (let attempt = 1; attempt <= pendingCount; attempt++) + await publishManagedGcSessionRetirementReceipt( + current.scope, + pendingReceipt(current, `phase-${phase.name}-${attempt}`), + ); + } + + const targetKey = crypto + .createHash("sha256") + .update(path.resolve(fixture.transcriptPath), "utf8") + .digest("hex"); + const targetPrefix = `.gjc-managed-session-internal/receipts/gc-retirement-${targetKey}-`; + const beforeTree = snapshotTree(fixture.sessionsRoot); + const beforeReceipts = managedReceiptInventory(fixture.scope); + const beforeSiblingReceipts = managedReceiptInventory(sibling.scope); + const reads = new Map(); + const descriptorPaths = new Map(); + const triggerFstatIdentities: Array<{ + dev: bigint; + ino: bigint; + nlink: bigint; + size: bigint; + mtimeNs: bigint; + ctimeNs: bigint; + isFile: boolean; + }> = []; + const targetCloseCounts = new Map(); + const targetReadBytes = new Map(); + const targetReadRequests = new Map(); + const targetAllocationCounts = new Map(); + let triggerPath: string | undefined; + let triggerRelativePath: string | undefined; + let triggerDescriptor: number | undefined; + let triggerOpenCount = 0; + let triggerStat: fs.BigIntStats | undefined; + let triggerRequestCap: number | undefined; + let allocationWhileTargetOpen = 0; + const originalRead = ManagedSessionDescendantStore.prototype.readExpectedBounded; + const originalOpen = fs.openSync.bind(fs); + const originalClose = fs.closeSync.bind(fs); + const originalReadSync = fs.readSync.bind(fs); + const originalFstat = fs.fstatSync.bind(fs); + const originalAlloc = Buffer.alloc.bind(Buffer); const read = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation( function ( this: ManagedSessionDescendantStore, relativePath: string, maxBytes: number, - admitSize?: (size: number) => void, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, ) { if (!relativePath.startsWith(targetPrefix)) return originalRead.call(this, relativePath, maxBytes, admitSize); - reads++; - const actualSize = Number( - fs.statSync(path.join(target.scope.directoryPath, relativePath), { bigint: true }).size, - ); - const current = { relativePath, actualSize, pressured: reads > phase.earlierReads }; - const previous = activeRead; - activeRead = current; - try { - return originalRead.call(this, relativePath, maxBytes, size => { - expect(size).toBe(actualSize); - admissions.push(current); - if (!admitSize) throw new Error(`fixture_${phase.name}_admission_callback_missing`); - admitSize(current.pressured ? 64 * 1024 * 1024 : size); - }); - } finally { - activeRead = previous; + const count = (reads.get(relativePath) ?? 0) + 1; + reads.set(relativePath, count); + const totalReads = [...reads.values()].reduce((sum, current) => sum + current, 0); + if (totalReads === phase.earlierReads + 1) { + triggerRelativePath = relativePath; + triggerPath = path.join(fixture.scope.directoryPath, relativePath); + triggerStat = fs.lstatSync(triggerPath, { bigint: true }); + triggerRequestCap = maxBytes; + if (!triggerStat.isFile() || triggerStat.nlink !== 1n || triggerStat.size <= 0n) + throw new Error(`fixture_${phase.name}_receipt_descriptor_unavailable`); + if (typeof admitSize !== "function") + throw new Error(`fixture_${phase.name}_admission_callback_missing`); + if (!Number.isSafeInteger(maxBytes) || maxBytes <= 0) + throw new Error(`fixture_${phase.name}_default_read_cap_invalid`); + return originalRead.call(this, relativePath, 0, admitSize); } + return originalRead.call(this, relativePath, maxBytes, admitSize); }, ); - const allocations = spyOn(Buffer, "alloc").mockImplementation((size, fill, encoding) => { - if (activeRead) allocated.push({ ...activeRead, size }); - return originalAllocate(size, fill, encoding); + const open = spyOn(fs, "openSync").mockImplementation((( + pathname: fs.PathLike, + flags: string | number, + mode?: number, + ) => { + const fd = originalOpen(pathname, flags, mode); + if (typeof pathname === "string" && triggerPath && path.resolve(pathname) === triggerPath) { + descriptorPaths.set(fd, triggerPath); + triggerDescriptor = fd; + triggerOpenCount++; + } + return fd; + }) as typeof fs.openSync); + const close = spyOn(fs, "closeSync").mockImplementation(fd => { + const pathname = descriptorPaths.get(fd); + try { + originalClose(fd); + } finally { + if (pathname) { + descriptorPaths.delete(fd); + targetCloseCounts.set(pathname, (targetCloseCounts.get(pathname) ?? 0) + 1); + } + } }); + const readSync = spyOn(fs, "readSync").mockImplementation((( + fd: number, + buffer: NodeJS.ArrayBufferView, + offset: number, + length: number, + position: number | null, + ) => { + const pathname = descriptorPaths.get(fd); + if (pathname) targetReadRequests.set(pathname, (targetReadRequests.get(pathname) ?? 0) + length); + const bytes = originalReadSync(fd, buffer, offset, length, position); + if (pathname) targetReadBytes.set(pathname, (targetReadBytes.get(pathname) ?? 0) + bytes); + return bytes; + }) as typeof fs.readSync); + const fstat = spyOn(fs, "fstatSync").mockImplementation(((fd: number, options?: { bigint?: boolean }) => { + if (options?.bigint) { + const stat = originalFstat(fd, { ...options, bigint: true }); + const pathname = descriptorPaths.get(fd); + if (pathname) { + if (typeof stat.mtimeNs !== "bigint" || typeof stat.ctimeNs !== "bigint") + throw new Error(`fixture_${phase.name}_opened_descriptor_stat_invalid`); + triggerFstatIdentities.push({ + dev: stat.dev, + ino: stat.ino, + nlink: stat.nlink, + size: stat.size, + mtimeNs: stat.mtimeNs, + ctimeNs: stat.ctimeNs, + isFile: stat.isFile(), + }); + } + return stat; + } + return originalFstat(fd, options ? { ...options, bigint: false } : undefined); + }) as typeof fs.fstatSync); + const allocate = spyOn(Buffer, "alloc").mockImplementation((( + size: number, + fill?: string | Uint8Array | number, + encoding?: BufferEncoding, + ) => { + const pathname = [...descriptorPaths.values()].find(value => value === triggerPath); + if (pathname) { + targetAllocationCounts.set(pathname, (targetAllocationCounts.get(pathname) ?? 0) + 1); + allocationWhileTargetOpen++; + } + return originalAlloc(size, fill, encoding); + }) as typeof Buffer.alloc); + let caught: unknown; try { - await expect( - managedGcProtocolScopeInspectorForScope(fixture.scope)([ - protocolInputFor(target.scope), - protocolInputFor(unaffected.scope), - ]), - ).rejects.toThrow("managed_gc_journal_capacity_exceeded"); + await managedGcProtocolScopeInspectorForScope(fixture.scope)([ + protocolInputFor(fixture.scope), + protocolInputFor(sibling.scope), + ]); + } catch (error) { + caught = error; } finally { + allocate.mockRestore(); + fstat.mockRestore(); + readSync.mockRestore(); + close.mockRestore(); + open.mockRestore(); read.mockRestore(); - allocations.mockRestore(); } - const pressured = admissions.filter(entry => entry.pressured); - const pressuredAllocations = allocated.filter(entry => entry.pressured && entry.size === entry.actualSize); - expect(reads).toBe(phase.earlierReads + 9); - expect(admissions.filter(entry => !entry.pressured)).toHaveLength(phase.earlierReads); - expect(pressured).toHaveLength(9); - expect(pressuredAllocations).toHaveLength(8); - const refused = pressured[8]; - if (!refused) throw new Error(`fixture_${phase.name}_refused_descriptor_missing`); - expect(allocated.some(entry => entry.pressured && entry.relativePath === refused.relativePath)).toBe(false); - expect(pressuredAllocations.every(entry => entry.size > 0 && entry.size < 64 * 1024 * 1024)).toBe(true); - expect([managedReceiptInventory(target.scope), managedReceiptInventory(unaffected.scope)]).toEqual(before); + + expect(caught).toMatchObject({ message: "managed_gc_journal_capacity_exceeded" }); + expect(triggerPath).toBeTypeOf("string"); + expect(triggerStat).toBeDefined(); + expect(triggerRequestCap).toBe(MANAGED_SESSION_READ_RANGE_MAX_BYTES); + expect(triggerStat!.size).toBeGreaterThan(0n); + expect([...reads.values()].reduce((sum, count) => sum + count, 0)).toBe(phase.earlierReads + 1); + expect(triggerRelativePath).toBeTypeOf("string"); + expect(reads.get(triggerRelativePath!)).toBe(phase.samePathRead); + expect(triggerOpenCount).toBe(1); + expect(targetCloseCounts.get(triggerPath!)).toBe(1); + if (triggerDescriptor === undefined) throw new Error(`fixture_${phase.name}_descriptor_not_opened`); + expect(triggerFstatIdentities).toEqual([ + { + dev: triggerStat!.dev, + ino: triggerStat!.ino, + nlink: triggerStat!.nlink, + size: triggerStat!.size, + mtimeNs: triggerStat!.mtimeNs, + ctimeNs: triggerStat!.ctimeNs, + isFile: true, + }, + ]); + expect(() => originalFstat(triggerDescriptor!, { bigint: true })).toThrow(); + expect(targetReadRequests.has(triggerPath!)).toBe(false); + expect(targetReadBytes.has(triggerPath!)).toBe(false); + expect(targetAllocationCounts.has(triggerPath!)).toBe(false); + expect(allocationWhileTargetOpen).toBe(0); + if (!triggerStat) throw new Error(`fixture_${phase.name}_trigger_stat_missing`); + const afterStat = fs.lstatSync(triggerPath!, { bigint: true }); + expect({ + dev: afterStat.dev, + ino: afterStat.ino, + nlink: afterStat.nlink, + size: afterStat.size, + mtimeNs: afterStat.mtimeNs, + ctimeNs: afterStat.ctimeNs, + }).toEqual({ + dev: triggerStat.dev, + ino: triggerStat.ino, + nlink: triggerStat.nlink, + size: triggerStat.size, + mtimeNs: triggerStat.mtimeNs, + ctimeNs: triggerStat.ctimeNs, + }); + expect([managedReceiptInventory(fixture.scope), managedReceiptInventory(sibling.scope)]).toEqual([ + beforeReceipts, + beforeSiblingReceipts, + ]); + expect(snapshotTree(fixture.sessionsRoot)).toEqual(beforeTree); } }); @@ -2343,7 +2863,7 @@ describe("managed GC retirement journal", () => { expect(managedReceiptInventory(fixture.scope)).toHaveLength(1); }); - it("admits aggregate history bytes before candidate encoding and rejects over-budget read allocation", async () => { + it("rejects a candidate after actual 512 MiB history admission before candidate encoding", async () => { const fixture = makeFixture(); await publishManagedGcSessionRetirementReceipt(fixture.scope, preparedReceipt(fixture)); await publishManagedGcSessionRetirementReceipt(fixture.scope, { @@ -2353,7 +2873,19 @@ describe("managed GC retirement journal", () => { }); for (let attempt = 1; attempt <= 6; attempt++) await publishManagedGcSessionRetirementReceipt(fixture.scope, pendingReceipt(fixture, `seed-${attempt}`)); - const candidateReason = "projected_history_capacity_candidate"; + const paddedPaths = padManagedGcReceiptHistoryTo512MiB(fixture); + for (const pathname of paddedPaths) { + const stat = fs.statSync(pathname, { bigint: true }); + expect(stat.size).toBe(BigInt(MANAGED_SESSION_READ_RANGE_MAX_BYTES)); + expect(stat.nlink).toBe(1n); + } + expect(paddedPaths.reduce((total, pathname) => total + fs.statSync(pathname).size, 0)).toBe( + MANAGED_ARTIFACT_MAX_TOTAL_BYTES, + ); + expect((await readManagedGcSessionRetirementReceiptReadOnly(fixture.scope, fixture.transcriptPath))?.state).toBe( + "owner_pending", + ); + const candidateReason = "actual_history_capacity_candidate"; const candidate = pendingReceipt(fixture, candidateReason); const candidateOutcome = candidate.taskArtifactOwnerRetirementOutcome; if (candidateOutcome?.kind !== "uncertain") throw new Error("fixture_candidate_outcome_missing"); @@ -2365,21 +2897,20 @@ describe("managed GC retirement journal", () => { }; const before = managedReceiptInventory(fixture.scope); const originalReadExpectedBounded = ManagedSessionDescendantStore.prototype.readExpectedBounded; - const simulatedAdmissions: number[] = []; - // Resource-pressure simulation only: the real bounded reader returns genuine file bytes and identities; - // only the admission callback receives a projected size before its allocation. + const actualAdmissions: Array<{ relativePath: string; size: number; descriptor: ManagedFileIdentity }> = []; const boundedRead = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation( function ( this: ManagedSessionDescendantStore, relativePath: string, maxBytes: number, - admitSize?: (size: number) => void, + admitSize?: (size: number, descriptor: ManagedFileIdentity) => void, ) { - return originalReadExpectedBounded.call(this, relativePath, maxBytes, size => { + return originalReadExpectedBounded.call(this, relativePath, maxBytes, (size, descriptor) => { if (relativePath.startsWith(".gjc-managed-session-internal/receipts/gc-retirement-")) { - simulatedAdmissions.push(64 * 1024 * 1024); - admitSize?.(64 * 1024 * 1024); - } else admitSize?.(size); + expect(descriptor.size).toBe(size); + actualAdmissions.push({ relativePath, size, descriptor }); + } + admitSize?.(size, descriptor); }); }, ); @@ -2390,28 +2921,12 @@ describe("managed GC retirement journal", () => { let encodedCandidate = false; let candidateBuffer = false; let publishedCandidate = false; + let receiptAllocationCount = 0; + const candidateRelativePath = path + .relative(fixture.scope.directoryPath, managedGcReceiptPath(fixture, "owner_pending-00000007")) + .split(path.sep) + .join("/"); try { - expect( - (await readManagedGcSessionRetirementReceiptReadOnly(fixture.scope, fixture.transcriptPath))?.state, - ).toBe("owner_pending"); - expect( - (await readManagedGcSessionRetirementReceiptReadOnly(fixture.scope, fixture.transcriptPath))?.state, - ).toBe("owner_pending"); - expect((await publishManagedGcSessionRetirementReceipt(fixture.scope, preparedReceipt(fixture))).state).toBe( - "owner_pending", - ); - expect( - ( - await publishManagedGcSessionRetirementReceipt(fixture.scope, { - ...preparedReceipt(fixture), - state: "artifacts_removed", - artifactsRemoved: true, - }) - ).state, - ).toBe("owner_pending"); - expect(simulatedAdmissions).toHaveLength(32); - expect(managedReceiptInventory(fixture.scope)).toEqual(before); - simulatedAdmissions.length = 0; await expect(publishManagedGcSessionRetirementReceipt(fixture.scope, invalidCandidate)).rejects.toThrow( "managed_gc_journal_capacity_exceeded", ); @@ -2424,9 +2939,12 @@ describe("managed GC retirement journal", () => { candidateBuffer = from.mock.calls.some( ([value]) => typeof value === "string" && value.includes(candidateReason), ); - publishedCandidate = publishes.mock.calls.some(([relativePath]) => - String(relativePath).startsWith(".gjc-managed-session-internal/receipts/gc-retirement-"), + publishedCandidate = publishes.mock.calls.some( + ([relativePath]) => String(relativePath) === candidateRelativePath, ); + receiptAllocationCount = allocations.mock.calls.filter( + ([size]) => size === MANAGED_SESSION_READ_RANGE_MAX_BYTES, + ).length; } finally { boundedRead.mockRestore(); allocations.mockRestore(); @@ -2434,48 +2952,48 @@ describe("managed GC retirement journal", () => { from.mockRestore(); publishes.mockRestore(); } - expect(simulatedAdmissions).toHaveLength(8); + expect(actualAdmissions).toHaveLength(8); + expect(actualAdmissions.every(entry => entry.size === MANAGED_SESSION_READ_RANGE_MAX_BYTES)).toBe(true); + expect(actualAdmissions.every(entry => entry.descriptor.nlink === 1n)).toBe(true); + expect(receiptAllocationCount).toBe(8); expect(encodedCandidate).toBe(false); expect(candidateBuffer).toBe(false); expect(publishedCandidate).toBe(false); expect(managedReceiptInventory(fixture.scope)).toEqual(before); - await expect(publishManagedGcSessionRetirementReceipt(fixture.scope, invalidCandidate)).rejects.toThrow( - "task_artifact_owner_retirement_outcome_invalid", - ); - expect(managedReceiptInventory(fixture.scope)).toEqual(before); - await publishManagedGcSessionRetirementReceipt(fixture.scope, candidate); - const afterAppend = managedReceiptInventory(fixture.scope); - const overflowRead = spyOn(ManagedSessionDescendantStore.prototype, "readExpectedBounded").mockImplementation( - function ( - this: ManagedSessionDescendantStore, - relativePath: string, - maxBytes: number, - admitSize?: (size: number) => void, - ) { - return originalReadExpectedBounded.call(this, relativePath, maxBytes, size => { - admitSize?.( - relativePath.startsWith(".gjc-managed-session-internal/receipts/gc-retirement-") - ? 64 * 1024 * 1024 - : size, - ); - }); - }, + const controlFixture = makeFixture("managed-gc-append-candidate-control"); + await publishManagedGcSessionRetirementReceipt(controlFixture.scope, preparedReceipt(controlFixture)); + await publishManagedGcSessionRetirementReceipt(controlFixture.scope, { + ...preparedReceipt(controlFixture), + state: "artifacts_removed", + artifactsRemoved: true, + }); + for (let attempt = 1; attempt <= 6; attempt++) + await publishManagedGcSessionRetirementReceipt( + controlFixture.scope, + pendingReceipt(controlFixture, `control-seed-${attempt}`), + ); + const controlBefore = managedReceiptInventory(controlFixture.scope); + const controlCandidate = pendingReceipt(controlFixture, "valid-history-control"); + const controlOutcome = controlCandidate.taskArtifactOwnerRetirementOutcome; + if (controlOutcome?.kind !== "uncertain") throw new Error("fixture_control_candidate_outcome_missing"); + const invalidControlCandidate: ManagedGcSessionRetirementReceipt = { + ...controlCandidate, + taskArtifactOwnerRetirementOutcome: Object.assign({}, controlOutcome, { + unexpectedBigint: 18_446_744_073_709_551_616n, + }) as typeof controlOutcome, + }; + await expect( + publishManagedGcSessionRetirementReceipt(controlFixture.scope, invalidControlCandidate), + ).rejects.toThrow("task_artifact_owner_retirement_outcome_invalid"); + expect(managedReceiptInventory(controlFixture.scope)).toEqual(controlBefore); + const validControlReceipt = await publishManagedGcSessionRetirementReceipt( + controlFixture.scope, + controlCandidate, ); - const readAllocations = spyOn(Buffer, "alloc"); - let readAllocatedOverLimit = false; - try { - await expect( - readManagedGcSessionRetirementReceiptReadOnly(fixture.scope, fixture.transcriptPath), - ).rejects.toThrow("managed_gc_journal_capacity_exceeded"); - readAllocatedOverLimit = readAllocations.mock.calls.some(([size]) => size >= 64 * 1024 * 1024); - } finally { - overflowRead.mockRestore(); - readAllocations.mockRestore(); - } - expect(readAllocatedOverLimit).toBe(false); - expect(managedReceiptInventory(fixture.scope)).toEqual(afterAppend); - }); + expect(validControlReceipt).toMatchObject({ state: "owner_pending", ownerRetirementAttempt: 7 }); + expect(managedReceiptInventory(controlFixture.scope)).toHaveLength(controlBefore.length + 1); + }, 60_000); it("uses bounded GC rereads when a receipt grows between discovery and inspector verification", async () => { const fixture = makeFixture(); diff --git a/packages/coding-agent/test/session-storage.test.ts b/packages/coding-agent/test/session-storage.test.ts index 3132b932e18..c0f114fef75 100644 --- a/packages/coding-agent/test/session-storage.test.ts +++ b/packages/coding-agent/test/session-storage.test.ts @@ -8,9 +8,11 @@ import * as native from "@gajae-code/natives"; import { logger } from "@gajae-code/utils"; import { captureManagedFileNoFollow, + captureManagedFileNoFollowBounded, ensureManagedDirectory, MANAGED_ARTIFACT_MAX_FILE_BYTES, ManagedCommittedMutationError, + type ManagedFileIdentity, ManagedReplaceError, ManagedSessionDescendantStore, managedDirectoryRoot, @@ -848,6 +850,189 @@ describe("managed descriptor reads", () => { fs.rmSync(root, { recursive: true, force: true }); } }); + + it("admits the opened descriptor before allocation, closes it, and preserves the admission failure", () => { + const root = fs.realpathSync.native(fs.mkdtempSync(path.join(os.tmpdir(), "gjc-managed-admission-"))); + const store = new ManagedSessionDescendantStore(managedDirectoryRoot(root), root); + const transcript = path.join(root, "session.jsonl"); + const bytes = Buffer.from("admitted descriptor bytes\n"); + store.publishNoReplaceSync("session.jsonl", bytes); + const expectedStat = fs.statSync(transcript, { bigint: true }); + const originalOpenReadLease = store.openReadLease.bind(store); + let actualLease: ReturnType | undefined; + let admissionObserved = false; + let forwardedClose = false; + let closeInjectionCount = 0; + const injectedCloseFailure = new Error("injected terminal close failure"); + let observedAdmission: { size: number; descriptor: ManagedFileIdentity } | undefined; + const primaryFailure = new Error("admission rejected before allocation"); + const openLease = vi.spyOn(store, "openReadLease").mockImplementation((relativePath, expectedDescriptor) => { + const lease = originalOpenReadLease(relativePath, expectedDescriptor); + actualLease = lease; + return { + readRange: (start, length) => lease.readRange(start, length), + close: () => { + lease.close(); + if (admissionObserved) { + forwardedClose = true; + closeInjectionCount++; + throw injectedCloseFailure; + } + }, + }; + }); + const read = vi.spyOn(fs, "readSync"); + const allocate = vi.spyOn(Buffer, "alloc"); + let caught: unknown; + try { + try { + store.readExpectedBounded("session.jsonl", bytes.byteLength, (size, descriptor) => { + admissionObserved = true; + observedAdmission = { size, descriptor }; + throw primaryFailure; + }); + } catch (error) { + caught = error; + } + expect(caught).toBe(primaryFailure); + expect(observedAdmission?.size).toBe(bytes.byteLength); + expect(observedAdmission?.descriptor).toMatchObject({ + dev: expectedStat.dev, + ino: expectedStat.ino, + nlink: 1n, + size: bytes.byteLength, + mtimeNs: expectedStat.mtimeNs, + ctimeNs: expectedStat.ctimeNs, + }); + if (!actualLease) throw new Error("managed capture read lease was not opened"); + expect(forwardedClose).toBe(true); + expect(admissionObserved).toBe(true); + expect(closeInjectionCount).toBe(1); + expect(() => actualLease!.readRange(0, 0)).toThrow("closed"); + expect(read).not.toHaveBeenCalled(); + expect(allocate.mock.calls.some(([size]) => size === bytes.byteLength)).toBe(false); + } finally { + allocate.mockRestore(); + read.mockRestore(); + openLease.mockRestore(); + store.close(); + fs.rmSync(root, { recursive: true, force: true }); + } + }); + + it("rejects growth during bounded capture without reading beyond the admitted size", () => { + const root = fs.realpathSync.native(fs.mkdtempSync(path.join(os.tmpdir(), "gjc-managed-growth-"))); + const pathname = path.join(root, "session.jsonl"); + const initial = Buffer.from("initial-sized transcript\n"); + fs.writeFileSync(pathname, initial, { mode: 0o600 }); + const originalRead = fs.readSync.bind(fs); + let bytesRead = 0; + let bytesRequested = 0; + let grew = false; + const read = vi.spyOn(fs, "readSync").mockImplementation((( + fd: number, + buffer: NodeJS.ArrayBufferView, + offset: number, + length: number, + position: number | null, + ) => { + bytesRequested += length; + const count = originalRead(fd, buffer, offset, length, position); + bytesRead += count; + if (!grew) { + grew = true; + fs.truncateSync(pathname, initial.byteLength + 9); + } + return count; + }) as typeof fs.readSync); + const allocate = vi.spyOn(Buffer, "alloc"); + try { + expect(() => captureManagedFileNoFollowBounded(pathname, initial.byteLength)).toThrow("source_changed"); + expect(bytesRead).toBe(initial.byteLength); + expect(bytesRequested).toBe(initial.byteLength); + expect(allocate.mock.calls.some(([size]) => size === initial.byteLength)).toBe(true); + expect(fs.statSync(pathname).size).toBe(initial.byteLength + 9); + } finally { + allocate.mockRestore(); + read.mockRestore(); + fs.rmSync(root, { recursive: true, force: true }); + } + }); + + it("rejects pathname replacement during bounded capture after reading only the original generation", () => { + const root = fs.realpathSync.native(fs.mkdtempSync(path.join(os.tmpdir(), "gjc-managed-replacement-"))); + const pathname = path.join(root, "session.jsonl"); + const detached = `${pathname}.detached`; + const initial = Buffer.from("original generation bytes\n"); + const replacement = Buffer.from("replacement generation must not be read\n"); + fs.writeFileSync(pathname, initial, { mode: 0o600 }); + const originalRead = fs.readSync.bind(fs); + let bytesRead = 0; + let bytesRequested = 0; + let replaced = false; + const read = vi.spyOn(fs, "readSync").mockImplementation((( + fd: number, + buffer: NodeJS.ArrayBufferView, + offset: number, + length: number, + position: number | null, + ) => { + bytesRequested += length; + const count = originalRead(fd, buffer, offset, length, position); + bytesRead += count; + if (!replaced) { + replaced = true; + fs.renameSync(pathname, detached); + fs.writeFileSync(pathname, replacement, { mode: 0o600 }); + } + return count; + }) as typeof fs.readSync); + const allocate = vi.spyOn(Buffer, "alloc"); + try { + expect(() => captureManagedFileNoFollowBounded(pathname, initial.byteLength)).toThrow("source_changed"); + expect(bytesRead).toBe(initial.byteLength); + expect(bytesRequested).toBe(initial.byteLength); + expect(allocate.mock.calls.some(([size]) => size === initial.byteLength)).toBe(true); + expect(fs.readFileSync(detached)).toEqual(initial); + expect(fs.readFileSync(pathname)).toEqual(replacement); + } finally { + allocate.mockRestore(); + read.mockRestore(); + fs.rmSync(root, { recursive: true, force: true }); + } + }); + + it("fences lease ranges to their initial size and returns fresh exact bytes and digest", () => { + const root = fs.realpathSync.native(fs.mkdtempSync(path.join(os.tmpdir(), "gjc-managed-lease-fence-"))); + const store = new ManagedSessionDescendantStore(managedDirectoryRoot(root), root); + try { + const firstBytes = Buffer.from("first descriptor generation\n"); + store.publishNoReplaceSync("session.jsonl", firstBytes); + const firstDescriptor = store.descriptorExpected("session.jsonl"); + if (!firstDescriptor) throw new Error("first managed descriptor missing"); + const firstLease = store.openReadLease("session.jsonl", firstDescriptor); + expect(() => firstLease.readRange(firstBytes.byteLength - 1, 2)).toThrow("range_not_present"); + expect(Buffer.from(firstLease.readRange(0, firstBytes.byteLength))).toEqual(firstBytes); + firstLease.close(); + + const freshBytes = Buffer.from("fresh positive exact transcript bytes\n"); + store.replaceSync("session.jsonl", freshBytes); + const freshDescriptor = store.descriptorExpected("session.jsonl"); + if (!freshDescriptor) throw new Error("fresh managed descriptor missing"); + const freshLease = store.openReadLease("session.jsonl", freshDescriptor); + const exactBytes = Buffer.from(freshLease.readRange(0, freshBytes.byteLength)); + freshLease.close(); + const snapshot = store.readExpectedBounded("session.jsonl", freshBytes.byteLength); + if (!snapshot) throw new Error("fresh bounded snapshot missing"); + expect(exactBytes).toEqual(freshBytes); + expect(snapshot.bytes).toEqual(freshBytes); + expect(snapshot.identity.sha256).toBe(createHash("sha256").update(freshBytes).digest("hex")); + expect(snapshot.identity.size).toBe(freshBytes.byteLength); + } finally { + store.close(); + fs.rmSync(root, { recursive: true, force: true }); + } + }); }); describe.skipIf(process.platform !== "darwin")("authority-absent managed replacement", () => {