fix(cso): bound release proof and maintenance work

Co-Authored-By: OpenAI Codex <noreply@openai.com>
This commit is contained in:
Garry TanandOpenAI Codex committed 2026-09-11 05:58:32 +00:00
1 parent f4821d5974
commit cac65ff8dc
12 files changed
+587 -177

No files matched your search

+204 -74
View File
@@ -9,6 +9,7 @@ export const DEFAULT_PUBLIC_ARCHIVE_CACHE_BYTES = 10 * 1024 * 1024 * 1024;
const METADATA_VERSION = 1;
const METADATA_LIMIT = 4096;
const COPY_BUFFER_BYTES = 64 * 1024;
const MAX_CACHE_DIRECTORY_ENTRIES = 100_000;
const CACHE_LOCK_PROTOCOL = 'immutable-cache-lease-set-v3';
const SHA256 = /^[a-f0-9]{64}$/;
const RELATIVE_STAGE_PATH = /^(?!\/)(?!.*(?:^|\/)\.\.?(?:\/|$))(?!.*\\)[^\0-\x1f\x7f]+$/;
@@ -26,6 +27,17 @@ export interface PublicArchiveCacheOptions {
now?: () => number;
}
/** Optional cooperative bounds for synchronous cache work. */
export interface CacheOperationControl {
/** Absolute Unix timestamp in milliseconds. */
deadline?: number;
/** Checked between bounded filesystem operations. */
signal?: AbortSignal;
}
export type CacheOperationInput = CacheOperationControl | number | undefined;
type NormalizedCacheOperationControl = Readonly<{ deadline?: number; signal?: AbortSignal }>;
export interface PublicArchiveCacheEntry {
sha256: string;
path: string;
@@ -69,6 +81,49 @@ function fail(code: ConstructorParameters<typeof CsoError>[0], message: string):
throw new CsoError(code, message);
}
function operationControl(input?: CacheOperationInput): NormalizedCacheOperationControl {
const value = typeof input === 'number' ? { deadline: input } : input ?? {};
if (!value || typeof value !== 'object' || Array.isArray(value)) fail('INVALID_ARGUMENT', 'Cache operation control must be an object or absolute deadline');
if (value.deadline !== undefined && (!Number.isSafeInteger(value.deadline) || value.deadline <= 0))
fail('INVALID_ARGUMENT', 'Cache deadline must be an absolute millisecond timestamp');
if (value.signal !== undefined && typeof value.signal.aborted !== 'boolean')
fail('INVALID_ARGUMENT', 'Cache cancellation signal is invalid');
const control = Object.freeze({ ...(value.deadline === undefined ? {} : { deadline: value.deadline }),
...(value.signal === undefined ? {} : { signal: value.signal }) });
checkOperation(control);
return control;
}
function checkOperation(control: NormalizedCacheOperationControl): void {
if (control.signal?.aborted) fail('CANCELLED', 'Archive-cache operation was cancelled');
if (control.deadline !== undefined && Date.now() >= control.deadline)
fail('DEADLINE', 'Archive-cache operation reached its deadline');
}
function boundedDirectoryNames(path: string, label: string, control: NormalizedCacheOperationControl): string[] {
checkOperation(control);
const directory = fs.opendirSync(path), names: string[] = [];
try {
for (;;) {
checkOperation(control);
const entry = directory.readSync();
if (!entry) break;
if (names.length >= MAX_CACHE_DIRECTORY_ENTRIES) fail('INSUFFICIENT_CAPACITY', `${label} exceeds the cache entry limit`);
names.push(entry.name);
}
} finally { directory.closeSync(); }
checkOperation(control);
return names;
}
function assertEmptyDirectory(path: string, control: NormalizedCacheOperationControl): void {
checkOperation(control);
const directory = fs.opendirSync(path);
try { if (directory.readSync()) fail('UNSAFE_PATH', 'Archive materialization directory must be empty'); }
finally { directory.closeSync(); }
checkOperation(control);
}
function boundedPositiveInteger(value: number, name: string): number {
if (!Number.isSafeInteger(value) || value <= 0) fail('INVALID_ARGUMENT', `${name} must be a positive safe integer`);
return value;
@@ -130,10 +185,11 @@ function assertExistingDirectory(path: string, label: string): string {
return canonical!;
}
function assertContainedAncestors(root: string, relativePath: string): string {
function assertContainedAncestors(root: string, relativePath: string, control: NormalizedCacheOperationControl): string {
const parts = relativePath.split('/');
let cursor = root;
for (const part of parts.slice(0, -1)) {
checkOperation(control);
cursor = join(cursor, part);
let stat: fs.Stats;
try { stat = fs.lstatSync(cursor); }
@@ -154,15 +210,19 @@ function openNoFollow(path: string, flags: number, mode?: number): number {
catch { fail('UNSAFE_PATH', 'Archive file could not be opened without following links'); }
}
function readMetadata(path: string, digest: string): Metadata {
function readMetadata(path: string, digest: string, control: NormalizedCacheOperationControl): Metadata {
checkOperation(control);
let stat: fs.Stats;
try { stat = fs.lstatSync(path); }
catch { fail('INCOMPATIBLE_INPUT', `Cache metadata is missing for ${digest}`); }
assertOwnedRegular(stat!, 'Cache metadata', METADATA_LIMIT);
if ((stat!.mode & 0o077) !== 0) fail('INCOMPATIBLE_INPUT', 'Cache metadata permissions are not private');
let value: unknown;
try { value = JSON.parse(fs.readFileSync(path, 'utf8')); }
catch { fail('INCOMPATIBLE_INPUT', `Cache metadata is invalid for ${digest}`); }
try { checkOperation(control); value = JSON.parse(fs.readFileSync(path, 'utf8')); checkOperation(control); }
catch (error) {
if (error instanceof CsoError) throw error;
fail('INCOMPATIBLE_INPUT', `Cache metadata is invalid for ${digest}`);
}
const record = value as Partial<Metadata>;
if (!value || typeof value !== 'object' || Array.isArray(value) || Object.keys(value).sort().join(',') !== 'bytes,createdAt,lastAccessedAt,sha256,version' ||
record.version !== METADATA_VERSION || record.sha256 !== digest || !Number.isSafeInteger(record.bytes) || Number(record.bytes) < 0 ||
@@ -226,12 +286,14 @@ export class PublicArchiveCache {
}
/** Promote a verified staging file. The staging file is never deleted. */
promote(stagedPath: string, sha256: string): PublicArchiveCacheEntry {
promote(stagedPath: string, sha256: string, operation?: CacheOperationInput): PublicArchiveCacheEntry {
const control = operationControl(operation);
const digest = expectedDigest(sha256), relativePath = stagedRelativePath(stagedPath);
const source = assertContainedAncestors(this.stagingRoot, relativePath);
const source = assertContainedAncestors(this.stagingRoot, relativePath, control);
return this.withLock(() => {
this.cleanIncoming();
this.recoverInterruptedOperations();
checkOperation(control);
this.cleanIncoming(control);
this.recoverInterruptedOperations(control);
let initial: fs.Stats;
try { initial = fs.lstatSync(source); }
catch { fail('MISSING_INPUT', 'Staged archive is missing'); }
@@ -243,24 +305,25 @@ export class PublicArchiveCache {
if (targetExists !== metadataExists) fail('INCOMPATIBLE_INPUT', `Cache entry is incomplete for ${digest}`);
if (targetExists) {
const staged = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!));
const staged = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!), control);
if (staged.digest !== digest) fail('INCOMPATIBLE_INPUT', 'Staged archive does not match its caller-provided SHA-256');
const metadata = this.verifiedEntry(digest);
return this.touch(metadata);
const metadata = this.verifiedEntry(digest, control);
return this.touch(metadata, control);
}
// Authenticate the complete staged object before it is allowed to
// displace any already-verified cache entry. Copying below hashes it a
// second time so a staging race still fails closed.
const authenticated = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!));
const authenticated = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!), control);
if (authenticated.digest !== digest) fail('INCOMPATIBLE_INPUT', 'Staged archive does not match its caller-provided SHA-256');
this.evictToFit(initial!.size);
this.evictToFit(initial!.size, control);
const incoming = join(this.incomingDir, `.incoming-${process.pid}-${randomBytes(12).toString('hex')}`);
let promoted = false;
try {
const staged = this.copyAndHash(source, incoming, this.maxEntryBytes, stableStat(initial!));
const staged = this.copyAndHash(source, incoming, this.maxEntryBytes, stableStat(initial!), control);
if (staged.digest !== digest) fail('INCOMPATIBLE_INPUT', 'Staged archive does not match its caller-provided SHA-256');
if (staged.bytes !== initial!.size) fail('SNAPSHOT_RACE', 'Staged archive changed during promotion');
checkOperation(control);
fs.chmodSync(incoming, 0o400);
// A hard-link followed by unlink is an atomic no-replace publication on
// the cache filesystem. rename(2) would silently replace a raced target.
@@ -269,7 +332,7 @@ export class PublicArchiveCache {
promoted = true;
const now = this.timestamp();
const metadata: Metadata = { version: 1, sha256: digest, bytes: staged.bytes, createdAt: now, lastAccessedAt: now };
try { writeMetadata(metadataPath, metadata, true); }
try { checkOperation(control); writeMetadata(metadataPath, metadata, true); }
catch (error) {
try { this.discardObject(target, 'entry', digest); } catch {}
throw error;
@@ -281,30 +344,36 @@ export class PublicArchiveCache {
if (stat.isFile() && !stat.isSymbolicLink()) fs.unlinkSync(incoming);
}
}
});
}, control);
}
/** Return a cache hit only after hashing every byte and validating metadata. */
get(sha256: string): PublicArchiveCacheEntry | undefined {
get(sha256: string, operation?: CacheOperationInput): PublicArchiveCacheEntry | undefined {
const control = operationControl(operation);
const digest = expectedDigest(sha256);
return this.withLock(() => {
this.cleanIncoming();
this.recoverInterruptedOperations();
checkOperation(control);
this.cleanIncoming(control);
this.recoverInterruptedOperations(control);
const targetExists = existsNoFollow(this.entryPath(digest)), metadataExists = existsNoFollow(this.metadataPath(digest));
if (!targetExists && !metadataExists) return undefined;
if (targetExists !== metadataExists) fail('INCOMPATIBLE_INPUT', `Cache entry is incomplete for ${digest}`);
return this.touch(this.verifiedEntry(digest));
});
return this.touch(this.verifiedEntry(digest, control), control);
}, control);
}
/** Inspect capacity without treating entries as execution-ready cache hits. */
stats(): PublicArchiveCacheStats {
stats(operation?: CacheOperationInput): PublicArchiveCacheStats {
const control = operationControl(operation);
return this.withLock(() => {
this.cleanIncoming();
this.recoverInterruptedOperations();
const entries = this.inventory();
return { entries: entries.length, bytes: entries.reduce((sum, item) => sum + item.bytes, 0), maxBytes: this.maxBytes };
});
checkOperation(control);
this.cleanIncoming(control);
this.recoverInterruptedOperations(control);
const entries = this.inventory(control);
let bytes = 0;
for (const entry of entries) { checkOperation(control); bytes += entry.bytes; }
return { entries: entries.length, bytes, maxBytes: this.maxBytes };
}, control);
}
/**
@@ -313,35 +382,49 @@ export class PublicArchiveCache {
* cache paths. Every source and every copy is fully hashed in the same
* critical section.
*/
materialize(digests: string[], destinationRoot: string): MaterializedArchive[] {
if (!Array.isArray(digests) || !digests.length || digests.some(value => typeof value !== 'string'))
materialize(digests: string[], destinationRoot: string, operation?: CacheOperationInput): MaterializedArchive[] {
const control = operationControl(operation);
if (!Array.isArray(digests) || !digests.length)
fail('INVALID_ARGUMENT', 'Archive materialization requires at least one SHA-256 digest');
const selected = [...new Set(digests.map(expectedDigest))].sort(), destination = assertExistingDirectory(destinationRoot, 'Archive materialization directory');
if (fs.readdirSync(destination).length) fail('UNSAFE_PATH', 'Archive materialization directory must be empty');
if (digests.length > MAX_CACHE_DIRECTORY_ENTRIES)
fail('INSUFFICIENT_CAPACITY', 'Archive materialization exceeds the cache entry limit');
const selectedSet=new Set<string>();for(const digest of digests){checkOperation(control);if(typeof digest!=='string')fail('INVALID_ARGUMENT', 'Archive materialization requires SHA-256 digest strings');selectedSet.add(expectedDigest(digest));}
checkOperation(control);const selected=[...selectedSet].sort();checkOperation(control);
const destination = assertExistingDirectory(destinationRoot, 'Archive materialization directory');
assertEmptyDirectory(destination, control);
if (destination === this.root || destination.startsWith(`${this.root}${sep}`) || this.root.startsWith(`${destination}${sep}`) ||
destination === this.stagingRoot || destination.startsWith(`${this.stagingRoot}${sep}`) || this.stagingRoot.startsWith(`${destination}${sep}`))
fail('UNSAFE_PATH', 'Archive materialization directory must be separate from cache and staging roots');
return this.withLock(() => {
this.cleanIncoming();
this.recoverInterruptedOperations();
checkOperation(control);
this.cleanIncoming(control);
this.recoverInterruptedOperations(control);
const created: string[] = [], result: MaterializedArchive[] = [];
try {
for (const digest of selected) {
const metadata = this.verifiedEntry(digest), source = this.entryPath(digest), initial = fs.lstatSync(source);
checkOperation(control);
const metadata = this.verifiedEntry(digest, control), source = this.entryPath(digest), initial = fs.lstatSync(source);
assertOwnedRegular(initial, 'Cached archive', this.maxEntryBytes, true);
const target = join(destination, digest), copied = this.copyAndHash(source, target, this.maxEntryBytes, stableStat(initial));
const target = join(destination, digest);
let copied: { digest: string; bytes: number };
try { copied = this.copyAndHash(source, target, this.maxEntryBytes, stableStat(initial), control); }
catch (error) {
if (existsNoFollow(target)) try { removeRegular(target, 'Incomplete run-owned archive copy'); } catch {}
throw error;
}
created.push(target);
if (copied.digest !== digest || copied.bytes !== metadata.bytes) fail('SNAPSHOT_RACE', 'Cached archive changed while its run-owned copy was materialized');
fs.chmodSync(target, 0o400);
const verified = this.hashFile(target, this.maxEntryBytes, true);
const verified = this.hashFile(target, this.maxEntryBytes, true, undefined, control);
if (verified.digest !== digest || verified.bytes !== metadata.bytes) fail('SNAPSHOT_RACE', 'Run-owned archive copy failed verification');
created.push(target); result.push(Object.freeze({ sha256: digest, path: target, bytes: verified.bytes }));
result.push(Object.freeze({ sha256: digest, path: target, bytes: verified.bytes }));
}
return result;
} catch (error) {
for (const path of created.reverse()) { try { removeRegular(path, 'Incomplete run-owned archive copy'); } catch {} }
throw error;
}
});
}, control);
}
private timestamp(): number {
@@ -358,20 +441,25 @@ export class PublicArchiveCache {
createdAt: metadata.createdAt, lastAccessedAt: metadata.lastAccessedAt });
}
private touch(metadata: Metadata): PublicArchiveCacheEntry {
private touch(metadata: Metadata, control: NormalizedCacheOperationControl): PublicArchiveCacheEntry {
checkOperation(control);
const updated: Metadata = { ...metadata, lastAccessedAt: Math.max(metadata.lastAccessedAt, this.timestamp()) };
checkOperation(control);
writeMetadata(this.metadataPath(metadata.sha256), updated);
return this.entry(updated);
}
private verifiedEntry(digest: string): Metadata {
const metadata = readMetadata(this.metadataPath(digest), digest);
const result = this.hashFile(this.entryPath(digest), this.maxEntryBytes, true);
private verifiedEntry(digest: string, control: NormalizedCacheOperationControl): Metadata {
checkOperation(control);
const metadata = readMetadata(this.metadataPath(digest), digest, control);
const result = this.hashFile(this.entryPath(digest), this.maxEntryBytes, true, undefined, control);
if (result.digest !== digest || result.bytes !== metadata.bytes) fail('INCOMPATIBLE_INPUT', `Cached archive failed SHA-256 verification: ${digest}`);
return metadata;
}
private hashFile(path: string, maxBytes: number, immutable: boolean, expected?: StableStat): { digest: string; bytes: number } {
private hashFile(path: string, maxBytes: number, immutable: boolean, expected: StableStat | undefined,
control: NormalizedCacheOperationControl): { digest: string; bytes: number } {
checkOperation(control);
const fd = openNoFollow(path, fs.constants.O_RDONLY);
try {
const beforeStat = fs.fstatSync(fd);
@@ -380,19 +468,24 @@ export class PublicArchiveCache {
if (expected && !sameStat(expected, before)) fail('SNAPSHOT_RACE', 'Archive changed before it could be verified');
let bytes = 0;
for (;;) {
checkOperation(control);
const read = fs.readSync(fd, buffer, 0, buffer.length, null);
if (!read) break;
bytes += read;
if (bytes > maxBytes) fail('INSUFFICIENT_CAPACITY', 'Archive exceeded its byte limit while being read');
hash.update(buffer.subarray(0, read));
checkOperation(control);
}
checkOperation(control);
const after = stableStat(fs.fstatSync(fd));
if (!sameStat(before, after) || bytes !== before.size) fail('SNAPSHOT_RACE', 'Archive changed while it was being verified');
return { digest: hash.digest('hex'), bytes };
} finally { fs.closeSync(fd); }
}
private copyAndHash(source: string, destination: string, maxBytes: number, expected: StableStat): { digest: string; bytes: number } {
private copyAndHash(source: string, destination: string, maxBytes: number, expected: StableStat,
control: NormalizedCacheOperationControl): { digest: string; bytes: number } {
checkOperation(control);
const sourceFd = openNoFollow(source, fs.constants.O_RDONLY);
let destinationFd: number | undefined;
try {
@@ -403,6 +496,7 @@ export class PublicArchiveCache {
destinationFd = openNoFollow(destination, fs.constants.O_WRONLY | fs.constants.O_CREAT | fs.constants.O_EXCL, 0o600);
let bytes = 0;
for (;;) {
checkOperation(control);
const read = fs.readSync(sourceFd, buffer, 0, buffer.length, null);
if (!read) break;
bytes += read;
@@ -410,9 +504,17 @@ export class PublicArchiveCache {
if (bytes > maxBytes) fail('INSUFFICIENT_CAPACITY', 'Archive exceeded its byte limit during promotion');
hash.update(buffer.subarray(0, read));
let offset = 0;
while (offset < read) offset += fs.writeSync(destinationFd, buffer, offset, read - offset);
while (offset < read) {
checkOperation(control);
const written = fs.writeSync(destinationFd, buffer, offset, read - offset);
if (written <= 0) fail('PERSISTENCE_FAILED', 'Archive copy stopped before the current chunk was written');
offset += written;
checkOperation(control);
}
}
checkOperation(control);
fs.fsyncSync(destinationFd);
checkOperation(control);
const after = stableStat(fs.fstatSync(sourceFd));
if (!sameStat(before, after) || bytes !== before.size) fail('SNAPSHOT_RACE', 'Staged archive changed during promotion');
return { digest: hash.digest('hex'), bytes };
@@ -422,28 +524,39 @@ export class PublicArchiveCache {
}
}
private inventory(): Metadata[] {
const entryNames = fs.readdirSync(this.entriesDir), metadataNames = fs.readdirSync(this.metadataDir);
for (const name of entryNames) if (!SHA256.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object');
for (const name of metadataNames) if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object');
const entrySet = new Set(entryNames), metadataSet = new Set(metadataNames.map(name => name.slice(0, -5)));
if (entrySet.size !== metadataSet.size || [...entrySet].some(name => !metadataSet.has(name)))
private inventory(control: NormalizedCacheOperationControl): Metadata[] {
checkOperation(control);
const entryNames = boundedDirectoryNames(this.entriesDir, 'Cache entries directory', control),
metadataNames = boundedDirectoryNames(this.metadataDir, 'Cache metadata directory', control);
const entrySet=new Set<string>(),metadataSet=new Set<string>();
for (const name of entryNames) {checkOperation(control);if (!SHA256.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object');entrySet.add(name);}
for (const name of metadataNames) {checkOperation(control);if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object');metadataSet.add(name.slice(0,-5));}
if (entrySet.size !== metadataSet.size) fail('INCOMPATIBLE_INPUT', 'Cache entries and metadata are inconsistent');
for(const name of entrySet){checkOperation(control);if(!metadataSet.has(name))
fail('INCOMPATIBLE_INPUT', 'Cache entries and metadata are inconsistent');
return entryNames.map(digest => {
}
const inventory = entryNames.map(digest => {
checkOperation(control);
const stat = fs.lstatSync(this.entryPath(digest));
assertOwnedRegular(stat, 'Cached archive', this.maxEntryBytes, true);
const metadata = readMetadata(this.metadataPath(digest), digest);
const metadata = readMetadata(this.metadataPath(digest), digest, control);
if (metadata.bytes !== stat.size) fail('INCOMPATIBLE_INPUT', `Cache size metadata is inconsistent for ${digest}`);
return metadata;
});
checkOperation(control);
return inventory;
}
private evictToFit(incomingBytes: number): void {
private evictToFit(incomingBytes: number, control: NormalizedCacheOperationControl): void {
if (!Number.isSafeInteger(incomingBytes) || incomingBytes < 0 || incomingBytes > this.maxBytes)
fail('INSUFFICIENT_CAPACITY', 'Archive cannot fit within the public-cache limit');
const entries = this.inventory().sort((a, b) => a.lastAccessedAt - b.lastAccessedAt || a.createdAt - b.createdAt || a.sha256.localeCompare(b.sha256));
let total = entries.reduce((sum, item) => sum + item.bytes, 0);
checkOperation(control);
const entries = this.inventory(control);checkOperation(control);entries.sort((a, b) => a.lastAccessedAt - b.lastAccessedAt || a.createdAt - b.createdAt || a.sha256.localeCompare(b.sha256));
checkOperation(control);
let total = 0;
for (const item of entries) { checkOperation(control); total += item.bytes; }
for (const item of entries) {
checkOperation(control);
if (total + incomingBytes <= this.maxBytes) break;
const archive = this.moveToRecovery(this.entryPath(item.sha256), 'entry', item.sha256);
const metadata = this.moveToRecovery(this.metadataPath(item.sha256), 'metadata', item.sha256);
@@ -454,18 +567,26 @@ export class PublicArchiveCache {
if (total + incomingBytes > this.maxBytes) fail('INSUFFICIENT_CAPACITY', 'Archive cache could not free enough verified capacity');
}
private cleanIncoming(): void {
for (const name of fs.readdirSync(this.incomingDir)) {
private cleanIncoming(control: NormalizedCacheOperationControl): void {
const incomingNames = boundedDirectoryNames(this.incomingDir, 'Cache incoming directory', control);
let publishedByInode:Map<string,string[]>|undefined;
for (const name of incomingNames) {
checkOperation(control);
if (!/^\.incoming-\d+-[a-f0-9]{24}$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache incoming directory contains an unexpected object');
const incoming = join(this.incomingDir, name), stat = fs.lstatSync(incoming);
if (!stat.isFile() || stat.isSymbolicLink() || ![1, 2].includes(stat.nlink) || stat.size > this.maxEntryBytes ||
(process.getuid && stat.uid !== process.getuid())) fail('UNSAFE_PATH', 'Incomplete cache archive is not a bounded regular file');
if (stat.nlink === 2) {
const matches = fs.readdirSync(this.entriesDir).filter(entry => {
if (!SHA256.test(entry)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object');
const candidate = fs.lstatSync(this.entryPath(entry));
return candidate.dev === stat.dev && candidate.ino === stat.ino;
});
if(!publishedByInode){
publishedByInode=new Map();
for(const entry of boundedDirectoryNames(this.entriesDir, 'Cache entries directory', control)){
checkOperation(control);
if (!SHA256.test(entry)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object');
const candidate=fs.lstatSync(this.entryPath(entry)),key=`${candidate.dev}:${candidate.ino}`,matches=publishedByInode.get(key)??[];
matches.push(entry);publishedByInode.set(key,matches);
}
}
const matches=publishedByInode.get(`${stat.dev}:${stat.ino}`)??[];
if (matches.length !== 1) fail('UNSAFE_PATH', 'Incoming archive hard link does not match one published cache entry');
const target = fs.lstatSync(this.entryPath(matches[0]));
if (!target.isFile() || target.isSymbolicLink() || target.nlink !== 2 || target.size > this.maxEntryBytes ||
@@ -480,23 +601,29 @@ export class PublicArchiveCache {
* Recover only artifacts whose names and inode types prove they belong to an
* interrupted cache transaction. Unknown objects remain a hard failure.
*/
private recoverInterruptedOperations(): void {
for (const name of fs.readdirSync(this.recoveryDir)) {
private recoverInterruptedOperations(control: NormalizedCacheOperationControl): void {
for (const name of boundedDirectoryNames(this.recoveryDir, 'Cache recovery directory', control)) {
checkOperation(control);
if (!/^\.recovery-(?:entry|metadata)-[a-f0-9]{64}-\d+-[a-f0-9]{24}$/.test(name))
fail('INCOMPATIBLE_INPUT', 'Cache recovery directory contains an unexpected object');
removeRegular(join(this.recoveryDir, name), 'Interrupted cache transaction');
}
const entries = fs.readdirSync(this.entriesDir), metadataObjects = fs.readdirSync(this.metadataDir);
const entries = boundedDirectoryNames(this.entriesDir, 'Cache entries directory', control),
metadataObjects = boundedDirectoryNames(this.metadataDir, 'Cache metadata directory', control);
for (const name of metadataObjects) {
if (/^[a-f0-9]{64}\.json\.tmp\.\d+\.[a-f0-9]{8}$/.test(name)) this.recoverMetadataTemp(name);
checkOperation(control);
if (/^[a-f0-9]{64}\.json\.tmp\.\d+\.[a-f0-9]{8}$/.test(name)) this.recoverMetadataTemp(name, control);
else if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object');
}
const metadata = fs.readdirSync(this.metadataDir);
const metadata = boundedDirectoryNames(this.metadataDir, 'Cache metadata directory', control);
for (const name of entries) if (!SHA256.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object');
for (const name of metadata) if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object');
const entrySet = new Set(entries), metadataSet = new Set(metadata.map(name => name.slice(0, -5)));
for (const digest of new Set([...entrySet, ...metadataSet])) {
const entrySet=new Set<string>(),metadataSet=new Set<string>(),digests=new Set<string>();
for(const name of entries){checkOperation(control);entrySet.add(name);digests.add(name);}
for(const name of metadata){checkOperation(control);const digest=name.slice(0,-5);metadataSet.add(digest);digests.add(digest);}
for (const digest of digests) {
checkOperation(control);
if (entrySet.has(digest) === metadataSet.has(digest)) continue;
if (entrySet.has(digest)) this.discardObject(this.entryPath(digest), 'entry', digest);
else this.discardObject(this.metadataPath(digest), 'metadata', digest);
@@ -526,7 +653,8 @@ export class PublicArchiveCache {
removeRegular(quarantined, 'Interrupted cache transaction');
}
private recoverMetadataTemp(name: string): void {
private recoverMetadataTemp(name: string, control: NormalizedCacheOperationControl): void {
checkOperation(control);
const path = join(this.metadataDir, name), stat = fs.lstatSync(path);
if (!stat.isFile() || stat.isSymbolicLink() || ![1, 2].includes(stat.nlink) || stat.size > METADATA_LIMIT ||
(process.getuid && stat.uid !== process.getuid()) || (stat.mode & 0o077) !== 0)
@@ -541,14 +669,16 @@ export class PublicArchiveCache {
try { fs.unlinkSync(path); } catch { fail('PERSISTENCE_FAILED', 'Interrupted cache metadata write could not be removed'); }
}
private withLock<T>(callback: () => T): T {
private withLock<T>(callback: () => T, control: NormalizedCacheOperationControl): T {
checkOperation(control);
const marker = `${JSON.stringify({ protocol: CACHE_LOCK_PROTOCOL })}\n`;
const options={label:'Cache lock protocol',maxBytes:METADATA_LIMIT,validate:(value:unknown)=>{
if(!value||typeof value!=='object'||Array.isArray(value)||Object.keys(value).join(',')!=='protocol'||(value as any).protocol!==CACHE_LOCK_PROTOCOL)
fail('INCOMPATIBLE_INPUT','Archive-cache lock protocol is invalid');
}};
const tempPattern=/^\.lock\.tmp\.(\d{1,10})\.[a-f0-9]{8}$/;
for(const name of fs.readdirSync(this.root)){
for(const name of boundedDirectoryNames(this.root, 'Cache root directory', control)){
checkOperation(control);
const match=name.match(tempPattern);if(!match)continue;
const temporary=join(this.root,name),publisherPid=Number(match[1]);
if(existsNoFollow(this.lockDir))recoverAtomicNoReplaceJson(this.lockDir,options);
@@ -566,7 +696,7 @@ export class PublicArchiveCache {
let protocol:unknown;try{protocol=JSON.parse(fs.readFileSync(this.lockDir,'utf8')).protocol;}catch{}
if(protocol!==CACHE_LOCK_PROTOCOL)fail('INCOMPATIBLE_INPUT','Archive-cache lock protocol is invalid');
}
const result=withStateLock(this.root,callback);
const result=withStateLock(this.root,()=>{checkOperation(control);return callback();});
if(result&&typeof (result as any).then==='function')fail('PERSISTENCE_FAILED','Archive-cache operation unexpectedly became asynchronous');
return result as T;
}
+14 -8
View File
@@ -26,6 +26,7 @@ import { historyForPath } from './history';
import { catalogImageProvisioningPolicy, inspectCatalogImages, openLocalCatalogImageSession, provisionCatalogImages, qualifiedCatalogImages, type CatalogImageSessionFactory } from './image-provisioning';
const VERSION = '3.0.0';
const RETENTION_MAINTENANCE_MS=1_000,RETENTION_MAX_ENTRIES=100_000,REPLAY_LOOKUP_MAX_ENTRIES=10_000;
const productionCatalogImageSession:CatalogImageSessionFactory=deadline=>openLocalCatalogImageSession(process.env,deadline);
/** Internal qualification seam. The public entrypoint below always supplies committed dependencies. */
export interface CsoCliDependencies {
@@ -145,8 +146,8 @@ function parseStart(args:string[]):{repo:string;policy:RunPolicy;comparisonBase?
const scope=named?`domain:${string(named,'scope',100)}`:scopeFlags[0]?.slice(2)??'default';
return {repo,policy:{mode:comprehensive?'comprehensive':'daily',scope,diff,base,offline,budgetSeconds,maxWorkers:3,maxRepairs:3},...(diff||explicitBase?{comparisonBase:base}:{})};
}
async function start(args:string[],dependencies:CsoCliDependencies,parent?:RunReportV3['parent'],requiredAncestor?:string):Promise<RunReportV3>{
const {repo,policy,comparisonBase}=parseStart(args);assertStateOutside(repo);if(!parent)retention();const createdAt=new Date(),run=newRun(repo);
async function start(args:string[],dependencies:CsoCliDependencies,parent?:RunReportV3['parent'],requiredAncestor?:string,startedAt=new Date()):Promise<RunReportV3>{
const {repo,policy,comparisonBase}=parseStart(args),createdAt=startedAt;assertStateOutside(repo);if(!parent)retention(createdAt.getTime(),{deadlineMs:Math.min(createdAt.getTime()+RETENTION_MAINTENANCE_MS,createdAt.getTime()+policy.budgetSeconds*1000-60_000),maxEntries:RETENTION_MAX_ENTRIES});const run=newRun(repo);
let manifest:Awaited<ReturnType<typeof capture>>;try{manifest=await capture(repo,run.dir,comparisonBase,requiredAncestor,{deadlineMs:createdAt.getTime()+policy.budgetSeconds*1000-60_000});}catch(error){fs.rmSync(run.dir,{recursive:true,force:true});throw error;}
const report:RunReportV3={schemaVersion:3,runId:run.runId,repoId:run.repoId,createdAt:createdAt.toISOString(),deadline:new Date(createdAt.getTime()+policy.budgetSeconds*1000).toISOString(),status:'running',completeness:'not assessed',policy,
source:{root:repo,snapshotHash:manifest.executionHash,originalHash:manifest.originalHash,baseCommit:manifest.baseCommit,
@@ -362,11 +363,11 @@ async function scanner(args:string[],sarif=false){
});
}
function scannerOutcome(args:string[]){const {dir}=run(args);if(args.length!==1||!/^[a-z0-9-]{1,40}-[a-f0-9]{16}-[a-f0-9]{16}$/.test(args[0]))throw new CsoError('INVALID_ARGUMENT','scanner-outcome requires one immutable scanner artifact ID');return readJson(join(dir,'scanner-outcomes',`${args[0]}.json`));}
async function recheck(args:string[],dependencies:CsoCliDependencies){if(!args.length)throw new CsoError('INVALID_ARGUMENT','recheck requires a finding ID');retention();const findingId=args.shift()!,runId=need(args,'--run'),originalDir=runDirectory(runId);return await withLock(originalDir,async()=>{const original=loadReport(originalDir),finding=original.findings.find(f=>f.id===findingId);if(!finding)throw new CsoError('MISSING_INPUT','Original finding does not exist');if(original.status!=='finished')throw new CsoError('INVALID_SCHEMA','Recheck requires a finished original audit');
async function recheck(args:string[],dependencies:CsoCliDependencies){const startedAt=new Date();if(!args.length)throw new CsoError('INVALID_ARGUMENT','recheck requires a finding ID');retention(startedAt.getTime(),{deadlineMs:startedAt.getTime()+RETENTION_MAINTENANCE_MS,maxEntries:RETENTION_MAX_ENTRIES});const findingId=args.shift()!,runId=need(args,'--run'),originalDir=runDirectory(runId);return await withLock(originalDir,async()=>{const original=loadReport(originalDir),finding=original.findings.find(f=>f.id===findingId);if(!finding)throw new CsoError('MISSING_INPUT','Original finding does not exist');if(original.status!=='finished')throw new CsoError('INVALID_SCHEMA','Recheck requires a finished original audit');
// Keep the original immutable while the fresh snapshot is captured and
// until its child lineage report has been durably published.
const repo=callerPath(need(args,'--repo'));if(args.length)throw new CsoError('INVALID_ARGUMENT',`Unknown recheck argument: ${args[0]}`);if(repoId(repo)!==original.repoId)throw new CsoError('INCOMPATIBLE_INPUT','Recheck must use the original audited repository');const oldManifest=readJson(join(originalDir,'snapshot.json'));
const preserveBase=original.policy.diff||Boolean(original.source.baseCommit),report=await start(['--repo',repo,...(original.policy.mode==='comprehensive'?['--comprehensive']:[]),...(original.policy.diff?['--diff']:[]),...(preserveBase?['--base',original.policy.base]:[]),'--budget',String(original.policy.budgetSeconds),...(original.policy.offline?['--offline']:[]),...(original.policy.scope==='default'?[]:original.policy.scope.startsWith('domain:')?['--scope',original.policy.scope.slice(7)]:[`--${original.policy.scope}`])],dependencies,{runId,findingId,kind:'recheck'},oldManifest.headCommit);
const preserveBase=original.policy.diff||Boolean(original.source.baseCommit),report=await start(['--repo',repo,...(original.policy.mode==='comprehensive'?['--comprehensive']:[]),...(original.policy.diff?['--diff']:[]),...(preserveBase?['--base',original.policy.base]:[]),'--budget',String(original.policy.budgetSeconds),...(original.policy.offline?['--offline']:[]),...(original.policy.scope==='default'?[]:original.policy.scope.startsWith('domain:')?['--scope',original.policy.scope.slice(7)]:[`--${original.policy.scope}`])],dependencies,{runId,findingId,kind:'recheck'},oldManifest.headCommit,startedAt);
return {runId:report.runId,parent:report.parent};});}
function recordReview(args:string[]){
@@ -453,11 +454,16 @@ async function verify(args:string[],dependencies:CsoCliDependencies){const {dir}
finding.reproduction=result.manifest.before.security==='intended_failure'&&result.manifest.before.booted&&result.manifest.before.legitimate?'reproduced':result.manifest.before.security==='pass'?'disproved':result.manifest.before.booted?'inconclusive':'blocked';
finding.repair=result.manifest.result==='tested'?'tested':result.manifest.result==='runtime_tested'?'runtime_tested':'failed';if(['tested','runtime_tested'].includes(result.manifest.result)){finding.verificationId=result.manifest.id;finding.verificationAssurance={assertions:result.manifest.assertionAssurance!,testCompletion:result.manifest.testCompletionAssurance,review:result.manifest.reviewAssurance};delete finding.reproductionAttemptId;}event(report,'verification',`${result.manifest.id}: ${result.manifest.result}; assertion assurance ${result.manifest.assertionAssurance}; test completion assurance ${result.manifest.testCompletionAssurance}; review assurance ${result.manifest.reviewAssurance}`);saveReport(dir,report);return{result:result.manifest.result,verification:result.manifest,bundle:`bundles/${result.bundle.id}.json`};});}
function replayBundle(stored:{path:string;dir:string},id:string):any{const bundle=validateRepairBundle(readJson(stored.path),id),report=loadReport(stored.dir),finding=report.findings.find(f=>f.verificationId===id&&['tested','runtime_tested'].includes(f.repair));if(!['tested','runtime_tested'].includes(bundle.verification.result)||!finding)throw new CsoError('INCOMPATIBLE_INPUT','Only a helper-recorded runtime-tested or host-reviewed repair bundle can be replayed');return bundle;}
async function withReplayBundle<T>(id:string,fn:(stored:{path:string;dir:string},bundle:any)=>Promise<T>):Promise<T>{if(!/^[a-f0-9]{32}$/.test(id))throw new CsoError('INVALID_ARGUMENT','Bundle identifier must be the 32-character ID returned by verify');const root=privateRoot();for(const repo of fs.readdirSync(root).filter(x=>/^[a-f0-9]{24}$/.test(x)))for(const run of fs.readdirSync(join(root,repo)).filter(x=>/^\d{13}-[a-f0-9]{16}$/.test(x))){const stored={dir:join(root,repo,run),path:join(root,repo,run,'bundles',`${id}.json`)};if(!fs.existsSync(stored.path))continue;let matched=false,value!:T;await withLock(stored.dir,async()=>{if(!fs.existsSync(stored.path))return;const bundle=replayBundle(stored,id);matched=true;value=await fn(stored,bundle);});if(matched)return value;}throw new CsoError('MISSING_INPUT','Repair bundle was not found or expired');}
async function withReplayBundle<T>(id:string,deadline:number,fn:(stored:{path:string;dir:string},bundle:any)=>Promise<T>):Promise<T>{
if(!/^[a-f0-9]{32}$/.test(id))throw new CsoError('INVALID_ARGUMENT','Bundle identifier must be the 32-character ID returned by verify');
let visited=0,stored:{path:string;dir:string}|undefined;const admit=()=>{if(Date.now()>=deadline)throw new CsoError('DEADLINE','Replay exhausted its five-minute budget while locating the recorded bundle');if(++visited>REPLAY_LOOKUP_MAX_ENTRIES)throw new CsoError('INSUFFICIENT_CAPACITY',`Replay bundle lookup exceeded ${REPLAY_LOOKUP_MAX_ENTRIES} private state entries`);};
const root=privateRoot(),repos=fs.opendirSync(root);try{let repo:fs.Dirent|null;search:while((repo=repos.readSync())!==null){admit();if(!repo.isDirectory()||!/^[a-f0-9]{24}$/.test(repo.name))continue;const repoDir=join(root,repo.name),runs=fs.opendirSync(repoDir);try{let run:fs.Dirent|null;while((run=runs.readSync())!==null){admit();if(!run.isDirectory()||!/^\d{13}-[a-f0-9]{16}$/.test(run.name))continue;const candidate={dir:join(repoDir,run.name),path:join(repoDir,run.name,'bundles',`${id}.json`)};if(fs.existsSync(candidate.path)){stored=candidate;break search;}}}finally{runs.closeSync();}}}finally{repos.closeSync();}
if(!stored)throw new CsoError('MISSING_INPUT','Repair bundle was not found or expired');let matched=false,value!:T;await withLock(stored.dir,async()=>{if(!fs.existsSync(stored!.path))return;const bundle=replayBundle(stored!,id);matched=true;value=await fn(stored!,bundle);});if(matched)return value;throw new CsoError('MISSING_INPUT','Repair bundle was not found or expired');
}
function replayManifestValue(manifest:any):unknown{const {id:_,createdAt:__,witnessHash:___,before,after,...stable}=manifest,observation=(value:any)=>{const{output:_,...rest}=value;return rest;};return{...stable,before:observation(before),after:observation(after)};}
async function replay(args:string[],dependencies:CsoCliDependencies){if(!args.length)throw new CsoError('INVALID_ARGUMENT','replay requires a bundle ID');const id=args.shift()!,source=args.includes('--source')?callerPath(need(args,'--source')):undefined;if(args.length)throw new CsoError('INVALID_ARGUMENT',`Unknown replay argument: ${args[0]}`);return await withReplayBundle(id,async(stored,bundle)=>{
async function replay(args:string[],dependencies:CsoCliDependencies){const replayStarted=Date.now(),replayDeadline=replayStarted+300_000;retention(replayStarted,{deadlineMs:replayStarted+RETENTION_MAINTENANCE_MS,maxEntries:RETENTION_MAX_ENTRIES});if(!args.length)throw new CsoError('INVALID_ARGUMENT','replay requires a bundle ID');const id=args.shift()!,source=args.includes('--source')?callerPath(need(args,'--source')):undefined;if(args.length)throw new CsoError('INVALID_ARGUMENT',`Unknown replay argument: ${args[0]}`);return await withReplayBundle(id,replayDeadline,async(stored,bundle)=>{
if(bundle.requiredInputs.archives?.length&&!bundle.requiredInputs.dependencyClosures)throw new CsoError('MISSING_INPUT','Replay bundle predates retained dependency closures and cannot substitute current dependency state');
let workDir=stored.dir,manifest:any,temporary:string|undefined;const retained=join(stored.dir,'snapshot'),replayDeadline=Date.now()+300_000;
let workDir=stored.dir,manifest:any,temporary:string|undefined;const retained=join(stored.dir,'snapshot');
try{
if(fs.existsSync(retained)){manifest=readJson(join(stored.dir,'snapshot.json'));assertSnapshot(stored.dir,manifest);}else if(source){const temp=newRun(source);temporary=temp.dir;workDir=temp.dir;manifest=await capture(source,workDir,undefined,undefined,{deadlineMs:replayDeadline});if(manifest.executionHash!==bundle.requiredInputs.sourceHash||manifest.originalHash!==bundle.requiredInputs.originalHash)throw new CsoError('INCOMPATIBLE_INPUT','Supplied source does not match the bundle input hashes');}else throw new CsoError('MISSING_INPUT','Retained source expired; supply explicitly matching source');
if(manifest.executionHash!==bundle.requiredInputs.sourceHash||manifest.originalHash!==bundle.requiredInputs.originalHash)throw new CsoError('INCOMPATIBLE_INPUT','Retained source hashes do not match the bundle');validateRepairBundle(bundle,id,join(workDir,'snapshot'),manifest);if(bundle.verification.policyHash!==ISOLATION_POLICY_HASH)throw new CsoError('INCOMPATIBLE_INPUT','Current helper isolation policy does not match the recorded bundle');const targetPlatform=bundle.requiredInputs.platform as 'linux/amd64'|'linux/arm64';let runtime;try{runtime=selectRuntime(bundle.verification.runtime.profile,targetPlatform,dependencies.runtimeCatalog);}catch(error:any){throw new CsoError('PREREQUISITE',error?.message||'Qualified replay runtime is unavailable');}const verifier=runtime;if(runtime.image!==bundle.requiredInputs.runtimeImage)throw new CsoError('INCOMPATIBLE_INPUT','Qualified runtime digest does not match the bundle');
@@ -469,7 +475,7 @@ async function replay(args:string[],dependencies:CsoCliDependencies){if(!args.le
}
export async function dispatchCsoCommand(command:string,args:string[],dependencies:CsoCliDependencies):Promise<unknown>{
args=[...args];if(!['start','recheck','doctor','provision-images'].includes(command))retention();
args=[...args];if(!['start','recheck','replay','doctor','provision-images'].includes(command)){const started=Date.now();retention(started,{deadlineMs:started+RETENTION_MAINTENANCE_MS,maxEntries:RETENTION_MAX_ENTRIES});}
let result:unknown;
switch(command){case'start':result=await start(args,dependencies);break;case'doctor':result=await doctor(args,dependencies);break;case'provision-images':result=await provisionImages(args,dependencies);break;case'resume':result=resume(args);break;case'inspect':await inspect(args);return;case'read':await read(args);return;case'history':await history(args);return;case'submit':result=submit(args);break;case'finish':result=finish(args);break;case'import-v2':result=importV2(args);break;case'scan':result=await scanner(args);break;case'scanner-outcome':result=scannerOutcome(args);break;case'import-sarif':result=await scanner(args,true);break;case'record-review':result=recordReview(args);break;case'test-plan':result=testPlan(args);break;case'runtime-plan':result=runtimePlan(args);break;case'recheck':result=await recheck(args,dependencies);break;
case'verify':result=await verify(args,dependencies);break;case'replay':result=await replay(args,dependencies);break;case'patch-hash':if(args.length!==1)throw new CsoError('INVALID_ARGUMENT','patch-hash requires one request JSON file');result={patchHash:patchHash(validateVerificationRequest(readInput(args[0])))};break;default:throw new CsoError('INVALID_ARGUMENT',`Unknown command: ${command}`);}
+90 -32
View File
@@ -22,6 +22,7 @@ import {
const SHA256 = /^[a-f0-9]{64}$/;
const HOST = /^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$/;
const RELATIVE_ARCHIVE = /^(?!\/)(?!.*(?:^|\/)\.\.?(?:\/|$))(?!.*\\)[A-Za-z0-9@._+\/-]{1,1024}$/;
const MAX_ARCHIVES = 25_000;
const MAX_TREE_FILES = 200_000;
const MAX_SOURCE_BYTES = 64 * 1024 * 1024;
// The smallest automatic execution role is the tests container: 768 MiB at
@@ -46,7 +47,8 @@ function deepFreeze<T>(value: T): T {
}
return value;
}
function checkDeadline(deadline: number): void {
function checkDeadline(deadline: number, signal?: AbortSignal): void {
if (signal?.aborted) fail('CANCELLED', 'Preparation was cancelled before the operation completed');
if (!Number.isSafeInteger(deadline) || deadline <= 0) fail('INVALID_ARGUMENT', 'Preparation deadline must be an absolute millisecond timestamp');
if (Date.now() >= deadline) fail('DEADLINE', 'Preparation deadline was reached before the operation completed');
}
@@ -376,10 +378,13 @@ function relativeArchivePath(value: string, label: string): string {
return value;
}
function stagedFileHashes(root: string, relativePath: string, expectedBytes: number): { sha256: string; sha512: string } {
function stagedFileHashes(root: string, relativePath: string, expectedBytes: number, deadline: number,
signal?: AbortSignal): { sha256: string; sha512: string } {
checkDeadline(deadline, signal);
const parts = relativeArchivePath(relativePath, 'Staged archive').split('/');
let cursor = root;
for (const part of parts.slice(0, -1)) {
checkDeadline(deadline, signal);
cursor = join(cursor, part);
let stat: fs.Stats;
try { stat = fs.lstatSync(cursor); } catch { fail('MISSING_INPUT', 'Acquisition staging directory is missing'); }
@@ -399,12 +404,15 @@ function stagedFileHashes(root: string, relativePath: string, expectedBytes: num
const h256 = createHash('sha256'), h512 = createHash('sha512'), buffer = Buffer.allocUnsafe(64 * 1024);
let bytes = 0;
for (;;) {
checkDeadline(deadline, signal);
const count = fs.readSync(fd!, buffer, 0, buffer.length, null);
if (!count) break;
bytes += count;
if (bytes > expectedBytes) fail('SNAPSHOT_RACE', 'Staged archive grew while it was verified');
h256.update(buffer.subarray(0, count)); h512.update(buffer.subarray(0, count));
checkDeadline(deadline, signal);
}
checkDeadline(deadline, signal);
const after = fs.fstatSync(fd!);
if (bytes !== expectedBytes || before.dev !== after.dev || before.ino !== after.ino || before.size !== after.size ||
before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs || before.mode !== after.mode || before.nlink !== after.nlink)
@@ -423,18 +431,23 @@ function closureIdentity(closure: Omit<DependencyClosure, 'closureHash'>): strin
function logicalInput(input: PreparationInput): string { return `${input.name}\0${input.version}`; }
function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdmission, closure: DependencyClosure,
cache: PublicArchiveCache, materializationBase?: string): { mounts: CachedArchiveMount[]; materializedRoot?: string } {
cache: PublicArchiveCache, deadline: number, signal?: AbortSignal,
materializationBase?: string): { mounts: CachedArchiveMount[]; materializedRoot?: string } {
checkDeadline(deadline, signal);
const runtime = admission.runtime, expectedPlanHash = planHash(plan);
const { closureHash, ...closureBody } = closure ?? ({} as DependencyClosure);
if (!closure || closure.schemaVersion !== 1 || closure.stack !== plan.stack || closure.planHash !== expectedPlanHash ||
closure.catalogRevision !== admission.catalogRevision || closure.runtimeId !== runtime.id || closure.runtimeImage !== runtime.image ||
closure.platform !== runtime.platform || !Array.isArray(closure.archives) || !SHA256.test(closureHash) ||
closure.platform !== runtime.platform || !Array.isArray(closure.archives) || closure.archives.length > MAX_ARCHIVES || !SHA256.test(closureHash) ||
closureHash !== closureIdentity(closureBody))
fail('INCOMPATIBLE_INPUT', 'Dependency closure does not match the admitted plan and runtime');
const publicInputs = plan.inputs.map((input, index) => ({ input, index })).filter(item => item.input.kind === 'public');
if (publicInputs.length > MAX_ARCHIVES) fail('INSUFFICIENT_CAPACITY', 'Dependency closure exceeds the archive-count limit');
const publicInputsByIndex = new Map(publicInputs.map(item => [item.index, item]));
const covered = new Set<string>(), paths = new Set<string>(), pending: Array<{ archive: DependencyArchive; input: PreparationInput }> = [];
for (const archive of closure.archives) {
const selected = publicInputs.find(item => item.index === archive.inputIndex);
checkDeadline(deadline, signal);
const selected = publicInputsByIndex.get(archive.inputIndex);
if (!selected || archive.name !== selected.input.name || archive.version !== selected.input.version ||
archive.declaredIntegrity !== (selected.input.integrity ?? 'registry-on-acquisition') || !SHA256.test(archive.sha256) ||
!Number.isSafeInteger(archive.bytes) || archive.bytes < 0 || !plan.registryHosts.includes(archive.requestedHost))
@@ -446,8 +459,11 @@ function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdm
paths.add(archive.installPath); covered.add(logicalInput(selected.input));
pending.push({ archive, input: selected.input });
}
for (const { input } of publicInputs) if (!covered.has(logicalInput(input)))
fail('MISSING_INPUT', `Dependency closure does not contain a compatible public archive for ${input.name}@${input.version}`);
for (const { input } of publicInputs) {
checkDeadline(deadline, signal);
if (!covered.has(logicalInput(input)))
fail('MISSING_INPUT', `Dependency closure does not contain a compatible public archive for ${input.name}@${input.version}`);
}
if (!pending.length) return { mounts: [] };
let base: string;
if (materializationBase) {
@@ -459,8 +475,11 @@ function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdm
const materializedRoot = fs.mkdtempSync(join(base, 'gstack-cso-archives-'));
fs.chmodSync(materializedRoot, 0o700);
try {
const copies = cache.materialize(pending.map(item => item.archive.sha256), materializedRoot), byDigest = new Map(copies.map(copy => [copy.sha256, copy]));
checkDeadline(deadline, signal);
const copies = cache.materialize(pending.map(item => item.archive.sha256), materializedRoot, { deadline, signal }),
byDigest = new Map(copies.map(copy => [copy.sha256, copy]));
const mounts = pending.map(({ archive }) => {
checkDeadline(deadline, signal);
const copy = byDigest.get(archive.sha256);
if (!copy || copy.bytes !== archive.bytes) fail('MISSING_INPUT', `Verified public archive is missing from the offline cache: ${archive.name}@${archive.version}`);
return { inputIndex: archive.inputIndex, name: archive.name, version: archive.version,
@@ -475,7 +494,26 @@ function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdm
}
type TreeEntry = { path: string; kind: 'file' | 'symlink'; mode: number; bytes: number; sha256: string };
function treeManifest(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false): TreeEntry[] {
function boundedTreeNames(directory: string, deadline: number, signal?: AbortSignal): string[] {
checkDeadline(deadline, signal);
const handle = fs.opendirSync(directory), names: string[] = [];
try {
for (;;) {
checkDeadline(deadline, signal);
const entry = handle.readSync();
if (!entry) break;
if (names.length >= MAX_TREE_FILES) fail('INSUFFICIENT_CAPACITY', 'Preparation directory exceeds its bounded manifest limit');
names.push(entry.name);
}
} finally { handle.closeSync(); }
checkDeadline(deadline, signal);
names.sort();
checkDeadline(deadline, signal);
return names;
}
function treeManifest(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false,
signal?: AbortSignal): TreeEntry[] {
checkDeadline(deadline, signal);
const root = resolve(rootPath);
let rootStat: fs.Stats, canonicalRoot: string;
try { rootStat = fs.lstatSync(root); canonicalRoot = fs.realpathSync(root); }
@@ -486,9 +524,10 @@ function treeManifest(rootPath: string, maxBytes: number, deadline: number, allo
const entries: TreeEntry[] = [];
let total = 0, nodes = 0;
const walk = (directory: string, prefix = ''): void => {
checkDeadline(deadline);
const before = fs.readdirSync(directory).sort();
checkDeadline(deadline, signal);
const before = boundedTreeNames(directory, deadline, signal);
for (const name of before) {
checkDeadline(deadline, signal);
const path = join(directory, name), relativePath = prefix ? `${prefix}/${name}` : name;
nodes++;
if (nodes > MAX_TREE_FILES || Buffer.byteLength(name) > 255 || Buffer.byteLength(relativePath) > 4096)
@@ -525,7 +564,8 @@ function treeManifest(rootPath: string, maxBytes: number, deadline: number, allo
try {
const initial = fs.fstatSync(fd!), hash = createHash('sha256'), buffer = Buffer.allocUnsafe(64 * 1024);
let readBytes = 0;
for (;;) { const count = fs.readSync(fd!, buffer, 0, buffer.length, null); if (!count) break; readBytes += count; hash.update(buffer.subarray(0, count)); checkDeadline(deadline); }
for (;;) { checkDeadline(deadline, signal); const count = fs.readSync(fd!, buffer, 0, buffer.length, null); if (!count) break; readBytes += count; hash.update(buffer.subarray(0, count)); checkDeadline(deadline, signal); }
checkDeadline(deadline, signal);
const final = fs.fstatSync(fd!);
if (readBytes !== initial.size || initial.dev !== final.dev || initial.ino !== final.ino || initial.size !== final.size ||
initial.mode !== final.mode || initial.mtimeMs !== final.mtimeMs || initial.ctimeMs !== final.ctimeMs)
@@ -533,13 +573,15 @@ function treeManifest(rootPath: string, maxBytes: number, deadline: number, allo
entries.push({ path: relativePath, kind: 'file', mode: initial.mode & 0o777, bytes: initial.size, sha256: hash.digest('hex') });
} finally { fs.closeSync(fd!); }
}
if (canonical(before) !== canonical(fs.readdirSync(directory).sort())) fail('SNAPSHOT_RACE', 'Preparation tree membership changed while hashing');
checkDeadline(deadline, signal);
if (canonical(before) !== canonical(boundedTreeNames(directory, deadline, signal))) fail('SNAPSHOT_RACE', 'Preparation tree membership changed while hashing');
};
walk(root);
return entries;
}
function treeHash(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false): string {
return sha256(canonical(treeManifest(rootPath, maxBytes, deadline, allowContainedSymlinks)));
function treeHash(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false,
signal?: AbortSignal): string {
return sha256(canonical(treeManifest(rootPath, maxBytes, deadline, allowContainedSymlinks, signal)));
}
function dependencyOutput(stack: CsoStack, path: string): boolean {
const first = path.split('/')[0];
@@ -567,13 +609,14 @@ function preparedEnvironment(plan: PreparationPlan): Record<string, string> {
return environment;
}
function provePreparedProjection(snapshot: string, preparedRoot: string, stack: CsoStack,
transformations: OfflinePreparationRequest['transformations'], deadline: number): { hash: string; transformations: PreparedApplication['transformations']; manifestHash: string; dependencyHash:string } {
const source = treeManifest(snapshot, MAX_SOURCE_BYTES, deadline), prepared = treeManifest(preparedRoot, MAX_PREPARED_BYTES, deadline, true),
transformations: OfflinePreparationRequest['transformations'], deadline: number, signal?: AbortSignal): { hash: string; transformations: PreparedApplication['transformations']; manifestHash: string; dependencyHash:string } {
const source = treeManifest(snapshot, MAX_SOURCE_BYTES, deadline, false, signal), prepared = treeManifest(preparedRoot, MAX_PREPARED_BYTES, deadline, true, signal),
sourceByPath = new Map(source.map(entry => [entry.path, entry])), preparedByPath = new Map(prepared.map(entry => [entry.path, entry])),
synthetic = new Map(transformations.map(item => [item.path, item]));
if (synthetic.size !== transformations.length) fail('INVALID_SCHEMA', 'Offline preparation transformations contain duplicate paths');
const actualTransformations: PreparedApplication['transformations'] = [];
for (const entry of source) {
checkDeadline(deadline, signal);
const actual = preparedByPath.get(entry.path), transformed = synthetic.get(entry.path);
if (!actual || actual.kind !== 'file') fail('ISOLATION_FAILED', `Offline lifecycle execution removed or replaced captured source: ${entry.path}`);
if (transformed) {
@@ -581,12 +624,14 @@ function provePreparedProjection(snapshot: string, preparedRoot: string, stack:
} else if (canonical(actual) !== canonical(entry)) fail('ISOLATION_FAILED', `Offline lifecycle execution changed captured source bytes or mode: ${entry.path}`);
}
for (const transformed of transformations) {
checkDeadline(deadline, signal);
const actual = preparedByPath.get(transformed.path);
if (!actual || actual.kind !== 'file' || actual.sha256 !== transformed.sha256)
fail('ISOLATION_FAILED', `Offline preparation did not preserve its declared transformation: ${transformed.path}`);
actualTransformations.push({ path: transformed.path, sha256: transformed.sha256, mode: actual.mode, reason: transformed.reason });
}
for (const entry of prepared) {
checkDeadline(deadline, signal);
if (sourceByPath.has(entry.path) || synthetic.has(entry.path) || dependencyOutput(stack, entry.path)) continue;
fail('ISOLATION_FAILED', `Offline lifecycle execution wrote outside its dependency roots: ${entry.path}`);
}
@@ -639,18 +684,23 @@ export class PreparationExecutor {
admission: PreparationRuntimeAdmission;
snapshot: string;
deadline: number;
signal?: AbortSignal;
offline?: boolean;
existingClosure?: DependencyClosure;
}): Promise<DependencyClosure> {
checkDeadline(options.deadline);
checkDeadline(options.deadline, options.signal);
const hash = validatePlan(options.plan, options.snapshot), runtime = runtimeFromAdmission(options.plan, options.admission);
checkDeadline(options.deadline, options.signal);
validateRunner(this.options.runner, options.plan.stack);
if (options.existingClosure) {
const verified = validateClosure(options.plan, options.admission, options.existingClosure, this.options.cache, this.options.materializationRoot);
const verified = validateClosure(options.plan, options.admission, options.existingClosure, this.options.cache,
options.deadline, options.signal, this.options.materializationRoot);
if (verified.materializedRoot) fs.rmSync(verified.materializedRoot, { recursive: true, force: true });
return options.existingClosure;
}
const publicInputs = options.plan.inputs.map((input, index) => ({ input, index })).filter(item => item.input.kind === 'public');
if (publicInputs.length > MAX_ARCHIVES) fail('INSUFFICIENT_CAPACITY', 'Preparation plan exceeds the archive-count limit');
const publicInputsByIndex = new Map(publicInputs.map(item => [item.index, item]));
if (options.offline && publicInputs.length) fail('MISSING_INPUT', 'Offline preparation requires a matching retained dependency closure and verified cache entries');
if (!publicInputs.length) {
const body: Omit<DependencyClosure, 'closureHash'> = { schemaVersion: 1, stack: options.plan.stack, planHash: hash,
@@ -665,14 +715,14 @@ export class PreparationExecutor {
schemaVersion: 1, planHash: hash, stack: options.plan.stack,
runtime: { id: runtime.id, image: runtime.image, platform: runtime.platform }, metadata: structuredClone(options.plan.metadata),
inputs: structuredClone(publicInputs), commands: structuredClone(options.plan.acquisition), stagingRoot: acquisitionStaging,
deadline: options.deadline, limits: { maxArchives: 25_000,
deadline: options.deadline, limits: { maxArchives: MAX_ARCHIVES,
maxArchiveBytes: Math.min(this.options.cache.maxEntryBytes, MAX_ACQUISITION_ARCHIVE_BYTES),
maxTotalArchiveBytes: Math.min(this.options.cache.maxBytes, MAX_ACQUISITION_ARCHIVE_BYTES), maxOutputBytes: MAX_OUTPUT },
network: { mode: 'registry-restricted', allowedHosts: [...options.plan.registryHosts] }, sourceMounted: false,
};
try {
const receipt = await this.options.runner.acquire(deepFreeze(request));
checkDeadline(options.deadline);
checkDeadline(options.deadline, options.signal);
validateAcquisitionReceipt(receipt, request, options.plan);
if (!Array.isArray(receipt.artifacts) || receipt.artifacts.length > request.limits.maxArchives)
fail('TOOL_FAILED', 'Acquisition receipt contains an invalid number of archives');
@@ -680,7 +730,8 @@ export class PreparationExecutor {
uniqueBytes = new Map<string, number>(), validated: Array<{ artifact: AcquisitionArtifactReceipt; selected: typeof publicInputs[number] }> = [];
let totalUniqueBytes = 0;
for (const artifact of receipt.artifacts) {
const selected = publicInputs.find(item => item.index === artifact.inputIndex);
checkDeadline(options.deadline, options.signal);
const selected = publicInputsByIndex.get(artifact.inputIndex);
if (!selected || seenInputs.has(artifact.inputIndex) || !SHA256.test(artifact.sha256) || artifact.registryResponseSha256 !== artifact.sha256 ||
!Number.isSafeInteger(artifact.bytes) || artifact.bytes < 0 || artifact.bytes > this.options.cache.maxEntryBytes ||
!options.plan.registryHosts.includes(artifact.requestedHost) || !receipt.network.contactedHosts.includes(artifact.requestedHost))
@@ -696,7 +747,7 @@ export class PreparationExecutor {
if (!receipt.network.contactedHosts.includes(resolvedHost)) fail('TOOL_FAILED', 'Direct archive response host was not contacted through the registry broker');
}
if (selected.input.url && artifact.requestedUrl !== selected.input.url) fail('TOOL_FAILED', 'Acquired archive request URL does not match its lockfile URL');
const hashes = stagedFileHashes(request.stagingRoot, artifact.stagingPath, artifact.bytes);
const hashes = stagedFileHashes(request.stagingRoot, artifact.stagingPath, artifact.bytes, options.deadline, options.signal);
if (hashes.sha256 !== artifact.sha256 || (selected.input.integritySource === 'lock' && !integrityMatches(selected.input.integrity, hashes)) ||
(selected.input.integritySource !== 'lock' && selected.input.integritySource !== 'registry-on-acquisition'))
fail('INCOMPATIBLE_INPUT', `Acquired archive failed lock/registry integrity verification: ${selected.input.name}@${selected.input.version}`);
@@ -711,9 +762,10 @@ export class PreparationExecutor {
// and the aggregate unique-byte ceiling have been validated.
try {
for (const { artifact, selected } of validated) {
checkDeadline(options.deadline, options.signal);
const absoluteStaged = resolve(request.stagingRoot, ...artifact.stagingPath.split('/'));
const cacheRelative = relative(this.options.cache.stagingRoot, absoluteStaged).split(sep).join('/');
const cached = this.options.cache.promote(cacheRelative, artifact.sha256);
const cached = this.options.cache.promote(cacheRelative, artifact.sha256, { deadline: options.deadline, signal: options.signal });
archives.push({ inputIndex: artifact.inputIndex, name: selected.input.name, version: selected.input.version,
installPath: artifact.installPath, sha256: cached.sha256, bytes: cached.bytes, requestedHost: artifact.requestedHost,
requestedUrl: artifact.requestedUrl, resolvedUrl: artifact.resolvedUrl,
@@ -726,8 +778,11 @@ export class PreparationExecutor {
}
}
const covered = new Set(archives.map(archive => logicalInput(options.plan.inputs[archive.inputIndex])));
for (const { input } of publicInputs) if (!covered.has(logicalInput(input)))
fail('MISSING_INPUT', `Acquisition did not produce a compatible archive for ${input.name}@${input.version}`);
for (const { input } of publicInputs) {
checkDeadline(options.deadline, options.signal);
if (!covered.has(logicalInput(input))) fail('MISSING_INPUT', `Acquisition did not produce a compatible archive for ${input.name}@${input.version}`);
}
checkDeadline(options.deadline, options.signal);
archives.sort((a, b) => a.installPath.localeCompare(b.installPath));
const body: Omit<DependencyClosure, 'closureHash'> = { schemaVersion: 1, stack: options.plan.stack, planHash: hash,
catalogRevision: options.admission.catalogRevision, runtimeId: runtime.id, runtimeImage: runtime.image, platform: runtime.platform,
@@ -748,12 +803,15 @@ export class PreparationExecutor {
snapshot: string;
closure: DependencyClosure;
deadline: number;
signal?: AbortSignal;
database?: RailsDatabaseSelection;
}): Promise<PreparedApplication> {
checkDeadline(options.deadline);
checkDeadline(options.deadline, options.signal);
const hash = validatePlan(options.plan, options.snapshot), runtime = runtimeFromAdmission(options.plan, options.admission);
checkDeadline(options.deadline, options.signal);
validateRunner(this.options.runner, options.plan.stack);
const materialized = validateClosure(options.plan, options.admission, options.closure, this.options.cache, this.options.materializationRoot),
const materialized = validateClosure(options.plan, options.admission, options.closure, this.options.cache,
options.deadline, options.signal, this.options.materializationRoot),
archives = materialized.mounts;
let database: OfflinePreparationRequest['database'];
let synthetic: Array<{ path: string; content: string }> = [];
@@ -772,7 +830,7 @@ export class PreparationExecutor {
} else if (options.database) fail('INVALID_ARGUMENT', 'Database preparation is only valid for Rails');
const transformations = synthetic.map(file => ({ ...file, sha256: sha256(file.content), reason: 'Synthetic isolated Rails test configuration' }));
const configurationHash = sha256(canonical(transformations)), databaseHash = sha256(canonical(database ?? null));
const sourceHash = treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline);
const sourceHash = treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline, false, options.signal);
const request: OfflinePreparationRequest = {
schemaVersion: 1, planHash: hash, stack: options.plan.stack,
runtime: { id: runtime.id, image: runtime.image, platform: runtime.platform }, sourceRoot: resolve(options.snapshot), sourceHash,
@@ -786,16 +844,16 @@ export class PreparationExecutor {
try {
const result = await this.options.runner.prepareOffline(deepFreeze(request));
returnedPreparedRoot = typeof result?.preparedRoot === 'string' ? result.preparedRoot : undefined;
checkDeadline(options.deadline);
checkDeadline(options.deadline, options.signal);
validateOfflineReceipt(result?.receipt, request);
const preparedRoot = resolve(result.preparedRoot);
const sourceRoot = resolve(options.snapshot);
if (preparedRoot === sourceRoot || preparedRoot.startsWith(`${sourceRoot}${sep}`) || sourceRoot.startsWith(`${preparedRoot}${sep}`) ||
preparedRoot === this.options.cache.root || preparedRoot.startsWith(`${this.options.cache.root}${sep}`))
fail('UNSAFE_PATH', 'Prepared application must be a separate disposable copy outside cache and source roots');
const projection = provePreparedProjection(options.snapshot, preparedRoot, options.plan.stack, request.transformations, options.deadline),
const projection = provePreparedProjection(options.snapshot, preparedRoot, options.plan.stack, request.transformations, options.deadline, options.signal),
preparedManifestHash = projection.manifestHash,preparedDependencyHash=projection.dependencyHash;
if (treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline) !== sourceHash)
if (treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline, false, options.signal) !== sourceHash)
fail('SNAPSHOT_RACE', 'Offline preparation changed its read-only input source');
completed = true;
return Object.freeze({ schemaVersion: 1 as const, stack: options.plan.stack, preparedRoot, sourceHash, preparedManifestHash,preparedDependencyHash,
+5 -1
View File
@@ -272,7 +272,11 @@ export function assertSnapshot(runDir: string, manifest: SnapshotManifest): void
const presentPaths=new Set(entries.map(e=>e.path)),presentIds=new Set(entries.map(e=>e.pathId)),deletedPaths=new Set(deleted.map(item=>item.path)),deletedIds=new Set(deleted.map(item=>item.pathId));
if(presentPaths.size!==entries.length||presentIds.size!==entries.length||deletedPaths.size!==deleted.length||deletedIds.size!==deleted.length||deleted.some(item=>presentPaths.has(item.path)||presentIds.has(item.pathId))||canonical(deleted.map(item=>item.path))!==canonical([...deletedPaths].sort())||!/^([a-f0-9]{64})$/.test(manifest.originalHash)||snapshotOriginalIdentity(entries,deleted)!==manifest.originalHash)throw new CsoError('INCOMPATIBLE_INPUT','Snapshot original identity is inconsistent');
if(manifest.changedPaths!==undefined){if(!Array.isArray(manifest.changedPaths))throw new CsoError('INCOMPATIBLE_INPUT','Snapshot changed paths are invalid');const changed=manifest.changedPaths.map(relativePath);if(new Set(changed).size!==changed.length||canonical(changed)!==canonical([...changed].sort()))throw new CsoError('INCOMPATIBLE_INPUT','Snapshot changed paths are invalid');}
const expected=entries.filter(e=>e.executionHash).sort((a,b)=>a.path.localeCompare(b.path));
// Capture already records entries in Git's deterministic code-unit path
// order. Preserve that manifest order here: localeCompare can reorder an
// uppercase path such as README.md after lowercase source files, producing
// a different execution identity from the one written at capture time.
const expected=entries.filter(e=>e.executionHash);
if(!/^([a-f0-9]{64})$/.test(manifest.executionHash)||sha256(canonical(expected.map(e=>[e.path,e.executionHash,e.mode])))!==manifest.executionHash)throw new CsoError('INCOMPATIBLE_INPUT','Snapshot execution identity is inconsistent');
const before=listed();if(canonical(before)!==canonical(expected.map(e=>e.path)))throw new CsoError('INCOMPATIBLE_INPUT','Retained snapshot membership changed');
for (const e of expected) {
+106 -49
View File
@@ -654,32 +654,40 @@ function acquireRunLease(dir:string):HeldRunLease{
function releaseRunLease(lease:HeldRunLease,path=lease.path):void{const directory=dirname(path);releaseKnownLease(path,join(directory,basename(lease.decision)),join(directory,basename(lease.active)),lease.token,lease.identity,lease.decisionIdentity,true);}
export function withLock<T>(dir: string, fn: () => T): T | Promise<Awaited<T>> {
const lease=acquireRunLease(dir),unlock=()=>releaseRunLease(lease);
let value:T;
try {
const value=fn();
if(value&&typeof (value as any).then==='function')return Promise.resolve(value).finally(unlock) as Promise<Awaited<T>>;
unlock();return value as any;
value=fn();
} catch(error){
try{unlock();}catch(releaseError){throw releaseError;}
throw error;
}
if(value&&typeof (value as any).then==='function')return Promise.resolve(value).finally(unlock) as Promise<Awaited<T>>;
// Keep release errors outside the callback catch path. Retrying an exact
// release after it partially succeeds can only obscure which lease phase
// changed and attempts the same fail-closed cleanup twice.
unlock();return value as any;
}
function boundedMarker(path:string):string{
function boundedMarker(path:string,admit:()=>void=()=>{}):string{
admit();
try{const stat=fs.lstatSync(path);if(stat.isSymbolicLink()||!stat.isFile()||stat.size>8192)return'';return fs.readFileSync(path,'utf8');}catch{return'';}
}
/** A detached watchdog owns these paths until it records exact cleanup or an acknowledgement. */
export function hasPendingWatchdogCleanup(dir:string):boolean{
export function hasPendingWatchdogCleanup(dir:string,admit:()=>void=()=>{}):boolean{
let visited=0,pending=false;
const walk=(at:string,depth:number)=>{
if(pending||depth>6||visited++>4000)return;
let entries:fs.Dirent[];try{entries=fs.readdirSync(at,{withFileTypes:true});}catch{return;}
const entries:fs.Dirent[]=[];let directory:fs.Dir;
admit();try{directory=fs.opendirSync(at);}catch{return;}
try{for(;;){admit();const entry=directory.readSync();if(!entry)break;entries.push(entry);}}finally{directory.closeSync();}
const names=new Set(entries.map(entry=>entry.name));
if(names.has('attempt.ready')&&!names.has('attempt.stopped')&&!boundedMarker(join(at,'attempt.event')).includes('execution-copy cleanup complete')){pending=true;return;}
if(names.has('watchdog.ready')&&!names.has('watchdog.stopped')&&!boundedMarker(join(at,'watchdog.event')).includes('cleanup complete')){pending=true;return;}
if(names.has('attempt.ready')&&!names.has('attempt.stopped')&&!boundedMarker(join(at,'attempt.event'),admit).includes('execution-copy cleanup complete')){pending=true;return;}
if(names.has('watchdog.ready')&&!names.has('watchdog.stopped')&&!boundedMarker(join(at,'watchdog.event'),admit).includes('cleanup complete')){pending=true;return;}
for(const entry of entries){if(!/^[A-Za-z0-9._-]{1,120}$/.test(entry.name)||!entry.isDirectory())continue;walk(join(at,entry.name),depth+1);if(pending)return;}
};
for(const name of ['supervision','preparation-execution']){
const root=join(dir,name);if(!fs.existsSync(root))continue;
admit();const root=join(dir,name);if(!fs.existsSync(root))continue;
admit();
const rootStat=fs.lstatSync(root);if(rootStat.isSymbolicLink()||!rootStat.isDirectory())throw new CsoError('UNSAFE_PATH','Watchdog supervision state is not a private directory');
walk(root,0);if(pending)return true;
}
@@ -693,64 +701,109 @@ export function finalizeReplayTemporary(dir:string):void{
fs.rmSync(dir,{recursive:true,force:true});
}
function repairBundleExpiry(dir:string,run:string,now:number,runExpired:boolean):boolean{
const bundles=join(dir,'bundles');if(!fs.existsSync(bundles))return false;
/** Remove one private tree cooperatively without following links or holding directory handles across checks. */
function boundedRemoveTree(root:string,admit:()=>void,preserveRootName?:string):void{
type Frame={path:string;root:boolean;names?:string[];index:number};
const stack:Frame[]=[{path:root,root:true,index:0}];
while(stack.length){
const frame=stack[stack.length-1];
if(!frame.names){
admit();let stat:fs.Stats;try{stat=fs.lstatSync(frame.path);}catch(error:any){if(error?.code==='ENOENT'){stack.pop();continue;}throw error;}
if(stat.isSymbolicLink()||!stat.isDirectory()){admit();fs.unlinkSync(frame.path);stack.pop();continue;}
const names:string[]=[];admit();const directory=fs.opendirSync(frame.path);
try{for(;;){admit();const entry=directory.readSync();if(!entry)break;if(!(frame.root&&entry.name===preserveRootName))names.push(entry.name);}}finally{directory.closeSync();}
frame.names=names;frame.index=0;
}
if(frame.index<frame.names.length){const name=frame.names[frame.index++];stack.push({path:join(frame.path,name),root:false,index:0});continue;}
if(frame.root&&preserveRootName)return;
admit();fs.rmdirSync(frame.path);stack.pop();
}
}
function consumeLeasedTree(root:string,admit:()=>void):void{
boundedRemoveTree(root,admit,'.mutation-lock-leases');
admit();const directory=fs.opendirSync(root);try{for(;;){admit();const entry=directory.readSync();if(!entry)break;if(entry.name!=='.mutation-lock-leases')throw new CsoError('SNAPSHOT_RACE','Private retention tree changed during bounded cleanup');}}finally{directory.closeSync();}
// Only the helper's fixed-size lease protocol remains. Consuming it with the
// directory preserves the exact-release invariant without an unbounded walk.
admit();fs.rmSync(root,{recursive:true,force:true});
}
function repairBundleExpiry(dir:string,run:string,now:number,runExpired:boolean,admit:()=>void):boolean{
admit();const bundles=join(dir,'bundles');if(!fs.existsSync(bundles))return false;
admit();
const stat=fs.lstatSync(bundles);if(stat.isSymbolicLink()||!stat.isDirectory())throw new CsoError('UNSAFE_PATH','Repair bundle archive is not a private directory');
let retained=false;
for(const name of fs.readdirSync(bundles)){
const match=name.match(/^([a-f0-9]{32})\.json$/);if(!match){if(runExpired)fs.rmSync(join(bundles,name),{recursive:true,force:true});continue;}
let retained=false,remaining=0;admit();const directory=fs.opendirSync(bundles);
try{for(;;){
admit();const entry=directory.readSync();if(!entry)break;const name=entry.name;
const match=name.match(/^([a-f0-9]{32})\.json$/);if(!match){if(runExpired)boundedRemoveTree(join(bundles,name),admit);else remaining++;continue;}
admit();
const value=readJson(join(bundles,name)) as Record<string,any>,id=match[1],created=Date.parse(value?.createdAt),expires=Date.parse(value?.expiresAt);
if(value?.schemaVersion!==3||value?.id!==id||value?.runId!==run||value?.verification?.id!==id||value?.verification?.runId!==run||value?.verification?.createdAt!==value.createdAt||
!Number.isFinite(created)||new Date(created).toISOString()!==value.createdAt||!Number.isFinite(expires)||value.expiresAt!==new Date(created+30*86400_000).toISOString())
throw new CsoError('INCOMPATIBLE_INPUT','Repair bundle retention identity is invalid');
if(expires<=now)fs.unlinkSync(join(bundles,name));else retained=true;
}
if(!fs.readdirSync(bundles).length)fs.rmdirSync(bundles);
if(expires<=now){admit();fs.unlinkSync(join(bundles,name));}else{retained=true;remaining++;}
}}finally{directory.closeSync();}
if(!remaining){admit();fs.rmdirSync(bundles);}
return retained;
}
function cleanupRun(dir:string,run:string,now:number,pinned:boolean):void{
function cleanupRun(dir:string,run:string,now:number,pinned:boolean,admit:()=>void):void{
admit();
let lease:HeldRunLease;
try{lease=acquireRunLease(dir);}catch(error){if(error instanceof CsoError&&error.code==='INSUFFICIENT_CAPACITY')return;throw error;}
let releasePath=lease.path,consumed=false;
try{
if(hasPendingWatchdogCleanup(dir))return;
const created=Number(run.split('-')[0]),runExpired=now-created>30*86400_000,retainedBundle=repairBundleExpiry(dir,run,now,runExpired);
const ephemeral=fs.existsSync(join(dir,EPHEMERAL_REPLAY));
if(hasPendingWatchdogCleanup(dir,admit))return;
const created=Number(run.split('-')[0]),runExpired=now-created>30*86400_000,retainedBundle=repairBundleExpiry(dir,run,now,runExpired,admit);
admit();const ephemeral=fs.existsSync(join(dir,EPHEMERAL_REPLAY));
if(ephemeral||(runExpired&&!pinned&&!retainedBundle)){
const before=fs.lstatSync(dir),tomb=join(dirname(dir),`.retired-${run}-${randomBytes(16).toString('hex')}`);fs.renameSync(dir,tomb);releasePath=join(tomb,'.mutation-lock-leases',basename(lease.path));const after=fs.lstatSync(tomb);
admit();const before=fs.lstatSync(dir),tomb=join(dirname(dir),`.retired-${run}-${randomBytes(16).toString('hex')}`);fs.renameSync(dir,tomb);releasePath=join(tomb,'.mutation-lock-leases',basename(lease.path));admit();const after=fs.lstatSync(tomb);
if(before.dev!==after.dev||before.ino!==after.ino)throw new CsoError('SNAPSHOT_RACE','Expired run changed while it was retired');
// The retired name is outside the public run namespace. Consume the
// exclusive lease with the tree so no release/delete gap can admit a
// second helper against the same directory.
fs.rmSync(tomb,{recursive:true,force:true});consumed=true;return;
consumeLeasedTree(tomb,admit);consumed=true;return;
}
if(now-created>7*86400_000)for(const p of ['snapshot','readable'])fs.rmSync(join(dir,p),{recursive:true,force:true});
if(runExpired)for(const p of ['reviews','replays','dependency-closures','scanner-outcomes','verification-attempts'])fs.rmSync(join(dir,p),{recursive:true,force:true});
if(now-created>7*86400_000)for(const p of ['snapshot','readable'])boundedRemoveTree(join(dir,p),admit);
if(runExpired)for(const p of ['reviews','replays','dependency-closures','scanner-outcomes','verification-attempts'])boundedRemoveTree(join(dir,p),admit);
}finally{if(!consumed)releaseRunLease(lease,releasePath);}
}
function cleanupRetiredRun(dir:string):void{
function cleanupRetiredRun(dir:string,admit:()=>void):void{
admit();
let lease:HeldRunLease;try{lease=acquireRunLease(dir);}catch(error){if(error instanceof CsoError&&error.code==='INSUFFICIENT_CAPACITY')return;throw error;}
let consumed=false;try{if(hasPendingWatchdogCleanup(dir))return;fs.rmSync(dir,{recursive:true,force:true});consumed=true;}finally{if(!consumed)releaseRunLease(lease);}
let consumed=false;try{if(hasPendingWatchdogCleanup(dir,admit))return;consumeLeasedTree(dir,admit);consumed=true;}finally{if(!consumed)releaseRunLease(lease);}
}
export function retention(now = Date.now()): void {
export interface RetentionOptions { deadlineMs?:number; maxEntries?:number }
export interface RetentionResult { complete:boolean; visited:number }
class RetentionBudgetExhausted extends Error {}
export function retention(now = Date.now(),options:RetentionOptions={}): RetentionResult {
const deadlineMs=options.deadlineMs??Number.MAX_SAFE_INTEGER,maxEntries=options.maxEntries??Number.MAX_SAFE_INTEGER;
if(!Number.isSafeInteger(deadlineMs)||deadlineMs<0||!Number.isSafeInteger(maxEntries)||maxEntries<1)throw new CsoError('INVALID_ARGUMENT','Invalid retention maintenance budget');
let visited=0;const admit=()=>{if(Date.now()>=deadlineMs||visited>=maxEntries)throw new RetentionBudgetExhausted();visited++;};
const names=(dir:string,pattern:RegExp):string[]=>{const found:string[]=[];admit();const directory=fs.opendirSync(dir);try{for(;;){admit();const entry=directory.readSync();if(!entry)break;if(pattern.test(entry.name))found.push(entry.name);}}finally{directory.closeSync();}return found;};
const root = privateRoot();
const retainedParents=new Set<string>();
const retainedParents=new Set<string>(),repositories:{repo:string;repoDir:string;runs:string[];retired:string[]}[]=[];
const retainedReport=(repoDir:string,repo:string,run:string):RunReportV3|undefined=>{
const file=join(repoDir,run,'report.json');
try{
admit();
const stat=fs.lstatSync(file);
if(!stat.isFile()||stat.isSymbolicLink()||stat.nlink!==1||stat.size<=0||stat.size>MAX_STATE_FILE||
(process.getuid&&stat.uid!==process.getuid())||(process.platform!=='win32'&&(stat.mode&0o077)!==0))return;
admit();
const report=JSON.parse(fs.readFileSync(file,'utf8'));
if(report?.schemaVersion!==3||report.runId!==run||report.repoId!==repo||!Array.isArray(report.coverage)||!Array.isArray(report.findings)||
!['running','finished','interrupted'].includes(report.status))return;
return report as RunReportV3;
}catch{return;}
}catch(error){if(error instanceof RetentionBudgetExhausted)throw error;return;}
};
for(const repo of fs.readdirSync(root).filter(x=>/^[a-f0-9]{24}$/.test(x))){
const repoDir=secureDirectory(join(root,repo));
for(const run of fs.readdirSync(repoDir).filter(x=>/^\d{13}-[a-f0-9]{16}$/.test(x))){
try{
for(const repo of names(root,/^[a-f0-9]{24}$/)){
admit();const repoDir=secureDirectory(join(root,repo)),entries=names(repoDir,/^(?:\d{13}-[a-f0-9]{16}|\.retired-\d{13}-[a-f0-9]{16}-[a-f0-9]{32})$/),runs=entries.filter(x=>/^\d{13}-/.test(x)),retired=entries.filter(x=>x.startsWith('.retired-'));
repositories.push({repo,repoDir,runs,retired});
}
// Discover every live recheck pin before destructive cleanup. An exhausted
// discovery pass returns without deleting a parent that may still be in use.
for(const {repo,repoDir,runs} of repositories)for(const run of runs){
if(now-Number(run.split('-')[0])>30*86400_000)continue;
const report=retainedReport(repoDir,repo,run),parent=report?.parent as Record<string,unknown>|undefined;
if(!report||!['running','interrupted'].includes(report.status)||!Number.isFinite(Date.parse(report.deadline))||Date.parse(report.deadline)<=now||
@@ -759,23 +812,27 @@ export function retention(now = Date.now()): void {
const original=retainedReport(repoDir,repo,parent.runId);
if(original?.status==='finished'&&original.findings.some(f=>f.id===parent.findingId))retainedParents.add(`${repo}/${parent.runId}`);
}
}
for (const repo of fs.readdirSync(root).filter(x => /^[a-f0-9]{24}$/.test(x))) {
const repoDir = secureDirectory(join(root,repo));
for(const retired of fs.readdirSync(repoDir).filter(x=>/^\.retired-\d{13}-[a-f0-9]{16}-[a-f0-9]{32}$/.test(x)))cleanupRetiredRun(join(repoDir,retired));
for (const run of fs.readdirSync(repoDir).filter(x => /^\d{13}-[a-f0-9]{16}$/.test(x))) {
const dir = secureDirectory(join(repoDir,run));
// Every destructive retention decision owns the same exclusive lease as
// writers and replay. Whole runs are atomically retired before release.
cleanupRun(dir,run,now,retainedParents.has(`${repo}/${run}`));
for (const {repo,repoDir,runs,retired} of repositories) {
for(const name of retired)cleanupRetiredRun(join(repoDir,name),admit);
for (const run of runs) {
admit();const dir = secureDirectory(join(repoDir,run));
// Every destructive retention decision owns the same exclusive lease as
// writers and replay. Whole runs are atomically retired before release.
cleanupRun(dir,run,now,retainedParents.has(`${repo}/${run}`),admit);
}
}
}
const legacy=join(root,'legacy-imports');
if(fs.existsSync(legacy)){
const directory=fs.lstatSync(legacy);if(directory.isSymbolicLink()||!directory.isDirectory())throw new CsoError('UNSAFE_PATH','Legacy report archive is not a private directory');
for(const name of fs.readdirSync(legacy).filter(value=>/^[a-f0-9]{64}\.json$/.test(value))){
const file=join(legacy,name),stat=fs.lstatSync(file);if(stat.isSymbolicLink()||!stat.isFile()||(process.getuid&&stat.uid!==process.getuid()))throw new CsoError('UNSAFE_PATH','Legacy report archive contains an unsafe artifact');
if(now-stat.mtimeMs>30*86400_000)fs.unlinkSync(file);
admit();const legacy=join(root,'legacy-imports');
if(fs.existsSync(legacy)){
admit();const directory=fs.lstatSync(legacy);if(directory.isSymbolicLink()||!directory.isDirectory())throw new CsoError('UNSAFE_PATH','Legacy report archive is not a private directory');
for(const name of names(legacy,/^[a-f0-9]{64}\.json$/)){
admit();const file=join(legacy,name),stat=fs.lstatSync(file);if(stat.isSymbolicLink()||!stat.isFile()||(process.getuid&&stat.uid!==process.getuid()))throw new CsoError('UNSAFE_PATH','Legacy report archive contains an unsafe artifact');
admit();
if(now-stat.mtimeMs>30*86400_000)fs.unlinkSync(file);
}
}
return{complete:true,visited};
}catch(error){
if(error instanceof RetentionBudgetExhausted)return{complete:false,visited};
throw error;
}
}
+4 -4
View File
@@ -317,7 +317,7 @@ export async function verifyRepair(params:{runId:string;runDir:string;manifest:S
for(const boundary of [...new Set([...request.boundaryFiles,...request.testFiles,...beforeStartPlan.entrypointFiles])]){const e=params.manifest.entries.find(x=>x.path===boundary);if(!e||e.transformation)throw new CsoError('INCOMPATIBLE_INPUT',`Snapshot transformation changes or withholds a verification input: ${boundary}`);}
for(const change of request.changes){const path=relativePath(change.path),entry=params.manifest.entries.find(x=>x.path===path);containedFile(snapshot,path);if(change.beforeSha256===null){if(entry)throw new CsoError('INCOMPATIBLE_INPUT',`Declared new repair path already exists in the snapshot: ${path}`);}else if(!entry||entry.transformation)throw new CsoError('INCOMPATIBLE_INPUT',`Snapshot transformation changes or withholds a repair input: ${path}`);}
let guardedCleanup:(()=>Promise<void>)|undefined;if(params.watchdogPath){const deadline=Math.min(params.attemptDeadline??Date.now()+300_000,Date.now()+300_000);guardedCleanup=await attemptGuard(params.runDir,work,params.watchdogPath,deadline);}secureDirectory(observations);
let certified:ReturnType<typeof certify>|undefined,beforeObs:VerificationObservation|undefined,afterObs:VerificationObservation|undefined,beforeWitness:AssertionWitnessReceipt|undefined,afterWitness:AssertionWitnessReceipt|undefined,failure:unknown;
let certified:ReturnType<typeof certify>|undefined,beforeObs:VerificationObservation|undefined,afterObs:VerificationObservation|undefined,beforeWitness:AssertionWitnessReceipt|undefined,afterWitness:AssertionWitnessReceipt|undefined,failure:unknown,missingExternalWitness=false;
try{
preparePatchedSource(snapshot,after,request);
const afterTestPlan=canonicalTestPlan(after,stack),afterStartPlan=canonicalStartPlan(after,stack,request.port);if(canonical(afterTestPlan)!==canonical(beforeTestPlan))throw new CsoError('ASSERTION_FAILED','Repair changed the canonical project test suite, runner configuration, or discovered test inputs');
@@ -335,13 +335,13 @@ export async function verifyRepair(params:{runId:string;runDir:string;manifest:S
afterObs=observedAfter;
certified=certify({...params,request,identityRequest,before:beforeObs,after:afterObs,beforeRoot:snapshot,afterRoot:after,startPlanHash:beforeStartPlan.signature,testPlanHash:beforeTestPlan.signature,testToolchain:beforeTestPlan.toolchain,minimumPassingTests:beforeTestPlan.minimumPassingTests,...(beforeWitness&&afterWitness?{witness:{before:beforeWitness,after:afterWitness}}:{})});
}catch(error){failure=error;}
try{if(guardedCleanup)await guardedCleanup();else fs.rmSync(work,{recursive:true,force:true});}catch(error){failure=error;certified=undefined;}
if(!failure&&certified&&!certified.bundle){const manifest=certified.manifest,review=manifest.review,missingWitness=!manifest.assertionAssurance&&manifest.testCompletionAssurance==='self_reported'&&manifest.result==='inconclusive'&&manifest.before.booted&&manifest.before.legitimate&&manifest.before.security==='intended_failure'&&manifest.before.existingTests&&manifest.after.booted&&manifest.after.legitimate&&manifest.after.security==='pass'&&manifest.after.existingTests&&review.independent&&review.rootCauseRepaired&&review.featurePreserved&&!review.boundaryMocks;failure=missingWitness?new CsoError('PREREQUISITE','Repair verification retained self-reported project-test diagnostics but requires a helper-authenticated out-of-process external assertion witness before a runtime-tested bundle can be issued'):new CsoError('ASSERTION_FAILED',manifest.result==='inconclusive'?'Repair verification was inconclusive; no repair bundle was issued':'Repair failed one or more required boot, control, security, existing-test, or review assertions; no repair bundle was issued');}
try{if(guardedCleanup)await guardedCleanup();else fs.rmSync(work,{recursive:true,force:true});}catch(error){failure=error;certified=undefined;missingExternalWitness=false;}
if(!failure&&certified&&!certified.bundle){const manifest=certified.manifest,review=manifest.review;missingExternalWitness=!manifest.assertionAssurance&&manifest.testCompletionAssurance==='self_reported'&&manifest.result==='inconclusive'&&manifest.before.booted&&manifest.before.legitimate&&manifest.before.security==='intended_failure'&&manifest.before.existingTests&&manifest.after.booted&&manifest.after.legitimate&&manifest.after.security==='pass'&&manifest.after.existingTests&&review.independent&&review.rootCauseRepaired&&review.featurePreserved&&!review.boundaryMocks;failure=missingExternalWitness?new CsoError('PREREQUISITE','Repair verification retained self-reported project-test diagnostics but requires a helper-authenticated out-of-process external assertion witness before a runtime-tested bundle can be issued'):new CsoError('ASSERTION_FAILED',manifest.result==='inconclusive'?'Repair verification was inconclusive; no repair bundle was issued':'Repair failed one or more required boot, control, security, existing-test, or review assertions; no repair bundle was issued');}
if(failure){
if(beforeObs){
const cause=failure instanceof CsoError?failure:new CsoError('ASSERTION_FAILED','Repair validation failed after the before-phase observation');
const reproduction=!beforeObs.booted?'blocked':!beforeObs.legitimate?'inconclusive':beforeObs.security==='intended_failure'?'reproduced':beforeObs.security==='pass'?'disproved':'inconclusive',harnessHash=verificationHarnessHash(request,snapshot);
const missingWitness=cause.code==='PREREQUISITE'&&cause.message.includes('out-of-process external assertion witness')&&reproduction==='reproduced'&&afterObs?.booted===true&&afterObs.legitimate===true&&afterObs.security==='pass'&&beforeObs.existingTests&&afterObs.existingTests;
const missingWitness=missingExternalWitness&&cause.code==='PREREQUISITE'&&reproduction==='reproduced'&&afterObs?.booted===true&&afterObs.legitimate===true&&afterObs.security==='pass'&&beforeObs.existingTests&&afterObs.existingTests;
const raw={schemaVersion:3 as const,artifactKind:'repair_candidate' as const,runId:params.runId,findingId:identityRequest.findingId,createdAt:new Date().toISOString(),bundleIssued:false as const,runtime:{image:params.runtime.image,platform:params.runtime.platform,profile:params.runtime.id},policyHash:params.policyHash,requestHash:sha256(canonical(identityRequest)),harnessHash,sourceHash:params.manifest.executionHash,request:identityRequest,patchHash:patchHash(identityRequest),testToolchain:beforeTestPlan.toolchain,testCompletionAssurance:'self_reported' as const,...(params.preparation?{preparationHash:sha256(canonical(params.preparation))}:{}),before:{...beforeObs,inputHash:harnessHash},...(afterObs?{after:{...afterObs,inputHash:harnessHash}}:{}),reproduction,repair:missingWitness?'proposed' as const:'failed' as const,failure:{code:cause.code,message:cause.message}},safe=sanitizeHelperForJson(raw) as Omit<FailedVerificationAttempt,'id'>,id=sha256(canonical(safe)).slice(0,32),attempt:FailedVerificationAttempt={...safe,id};
validateVerificationObservation(attempt.before);if(attempt.after)validateVerificationObservation(attempt.after);if(params.persist!==false)writeJsonExclusive(join(params.runDir,'verification-attempts',`${id}.json`),attempt);
throw new VerificationAttemptError(cause,attempt);
+56 -6
View File
@@ -55,6 +55,17 @@ export interface EvalFinding {
id: string; evidence: 'supported' | 'hypothesis' | 'legacy_review'; claimedTested: boolean;
/** Written only by the independent evaluator after reviewing the reported trace. */
judgment: 'correct' | 'incorrect' | 'unadjudicated'; matchedCaseId?: string;
/**
* Written only by the independent evaluator. A producer's tested claim is
* valid only when this finding, rather than merely another finding in the
* same corpus cell, owns the trusted repair and recheck observations.
*/
trustedVerification?: {
repair: Outcome;
repairEvidenceHash?: string;
recheck: Outcome;
recheckEvidenceHash?: string;
};
}
export interface EvalResult {
cellId: string; sourceHash: string; skillHash: string; model: string; host: string; budgetSeconds: number;
@@ -445,11 +456,33 @@ function validateResult(result: EvalResult, cell: EvalCell, corpus: CorpusManife
if (result.recheckEvidenceHash !== undefined && !HEX.test(result.recheckEvidenceHash)) throw new Error('INVALID_RECHECK_EVIDENCE');
if (result.usage && (result.usage.source !== 'host' || (result.usage.tokens !== undefined && !validNumber(result.usage.tokens)) || (result.usage.costUSD !== undefined && !validNumber(result.usage.costUSD)))) throw new Error('INVALID_EVAL_USAGE');
if (result.producerReceiptHash !== undefined && !HEX.test(result.producerReceiptHash)) throw new Error('INVALID_PRODUCER_RECEIPT_PROVENANCE');
const ids = new Set<string>();
const ids = new Set<string>(), repairEvidence = new Set<string>(), recheckEvidence = new Set<string>();
for (const finding of result.findings) {
if (!finding.id || ids.has(finding.id) || !['supported', 'hypothesis', 'legacy_review'].includes(finding.evidence) || !['correct', 'incorrect', 'unadjudicated'].includes(finding.judgment) || typeof finding.claimedTested !== 'boolean') throw new Error('INVALID_FINDING_JUDGMENT');
ids.add(finding.id);
if (finding.judgment === 'correct' && (cell.variant !== 'vulnerable' || finding.matchedCaseId !== cell.caseId)) throw new Error('INVALID_ORACLE_MATCH');
if (finding.trustedVerification !== undefined) {
const verification = finding.trustedVerification as Record<string, unknown>, allowed = new Set(['repair', 'repairEvidenceHash', 'recheck', 'recheckEvidenceHash']);
if (!verification || typeof verification !== 'object' || Array.isArray(verification) || Object.keys(verification).some(key => !allowed.has(key)) ||
!['passed', 'failed', 'blocked', 'not_attempted'].includes(String(verification.repair)) ||
!['passed', 'failed', 'blocked', 'not_attempted'].includes(String(verification.recheck)) || !finding.claimedTested || cell.version !== 'v3' || cell.mode !== 'comprehensive')
throw new Error('INVALID_TRUSTED_FINDING_VERIFICATION');
const repairHash = verification.repairEvidenceHash, recheckHash = verification.recheckEvidenceHash;
if (!(repairHash === undefined || (typeof repairHash === 'string' && HEX.test(repairHash))) ||
!(recheckHash === undefined || (typeof recheckHash === 'string' && HEX.test(recheckHash))) ||
(verification.repair === 'passed') !== (repairHash !== undefined) ||
(verification.recheck === 'passed') !== (recheckHash !== undefined) ||
(verification.recheck === 'passed' && verification.repair !== 'passed'))
throw new Error('INVALID_TRUSTED_FINDING_VERIFICATION');
if (typeof repairHash === 'string') {
if (repairEvidence.has(repairHash)) throw new Error('DUPLICATE_TRUSTED_REPAIR_BINDING');
repairEvidence.add(repairHash);
}
if (typeof recheckHash === 'string') {
if (recheckEvidence.has(recheckHash)) throw new Error('DUPLICATE_TRUSTED_RECHECK_BINDING');
recheckEvidence.add(recheckHash);
}
}
}
if (cell.mode === 'daily' && [result.setup, result.reproduction, result.repair, result.recheck].some(value => value !== 'not_attempted')) throw new Error('DAILY_EVAL_EXECUTED_APPLICATION');
}
@@ -465,14 +498,29 @@ function measuredRate(cells: EvalCell[], results: Map<string, EvalResult>, field
if (!result.oracleEvidenceHash || result.oracleVersion !== corpus.version || result.setup !== 'passed') return false;
if (field === 'reproduction') return true;
if (result.reproduction !== 'passed' || result.repair !== 'passed' || !result.heldOutAssertionsPassed) return false;
if (cell.version === 'v3' && !result.findings.some(finding => trustedFindingVerification(cell, result, finding, 'repair', corpus))) return false;
if (field === 'repair') return true;
// Correct alternative patches need not match the reference fix byte-for-byte.
// Closure needs a fresh source and a separate trusted current-source observation.
return result.freshRecheck && !!result.currentSourceHash && result.currentSourceHash !== cell.sourceHash
&& !!result.recheckEvidenceHash && result.recheckEvidenceHash !== result.oracleEvidenceHash;
return result.freshRecheck && !!result.currentSourceHash && result.currentSourceHash !== cell.sourceHash &&
!!result.recheckEvidenceHash && result.recheckEvidenceHash !== result.oracleEvidenceHash &&
(cell.version !== 'v3' || result.findings.some(finding => trustedFindingVerification(cell, result, finding, 'recheck', corpus)));
}).length, cells.length);
}
function trustedFindingVerification(cell: EvalCell, result: EvalResult, finding: EvalFinding, stage: 'repair' | 'recheck', corpus: CorpusManifest): boolean {
const verification = finding.trustedVerification;
if (cell.version !== 'v3' || cell.mode !== 'comprehensive' || cell.variant !== 'vulnerable' || finding.evidence !== 'supported' ||
finding.judgment !== 'correct' || finding.matchedCaseId !== cell.caseId || !finding.claimedTested || !verification ||
result.setup !== 'passed' || result.reproduction !== 'passed' || result.repair !== 'passed' || !result.heldOutAssertionsPassed ||
verification.repair !== 'passed' || verification.repairEvidenceHash !== result.oracleEvidenceHash || result.oracleVersion !== corpus.version)
return false;
if (stage === 'repair') return true;
return result.recheck === 'passed' && verification.recheck === 'passed' && verification.recheckEvidenceHash === result.recheckEvidenceHash &&
result.freshRecheck && !!result.currentSourceHash && result.currentSourceHash !== cell.sourceHash &&
!!result.recheckEvidenceHash && result.recheckEvidenceHash !== result.oracleEvidenceHash;
}
export function scoreEval(matrix: EvalMatrix, observations: EvalResult[], qualification: ReleaseQualification = { containment: {} }, corpus: CorpusManifest = loadCorpusManifest()) {
validateMatrix(matrix, corpus);
if (!Array.isArray(observations)) throw new Error('INVALID_EVAL_RESULTS');
@@ -503,8 +551,7 @@ export function scoreEval(matrix: EvalMatrix, observations: EvalResult[], qualif
const tokens = completed.flatMap(result => result.usage?.tokens === undefined ? [] : [result.usage.tokens]);
const falseTested = selected.reduce((count, cell) => {
const result = results.get(cell.id); if (!result) return count;
const valid = cell.version === 'v3' && cell.variant === 'vulnerable' && measuredRate([cell], results, 'repair', corpus).numerator === 1;
return count + result.findings.filter(finding => finding.claimedTested && !(valid && finding.evidence === 'supported' && finding.judgment === 'correct')).length;
return count + result.findings.filter(finding => finding.claimedTested && !trustedFindingVerification(cell, result, finding, 'repair', corpus)).length;
}, 0);
groups.push({ version, mode, cells: selected.length, submitted: completed.length, missing: selected.length - completed.length,
reports: rate(completed.filter(result => result.reportPresent).length, selected.length), precision: rate(correct, reported), recall: rate(expected.filter(found).length, expected.length), highCriticalRecall: rate(highCritical.filter(found).length, highCritical.length),
@@ -526,7 +573,10 @@ export function scoreEval(matrix: EvalMatrix, observations: EvalResult[], qualif
const eligible = matrix.cells.filter(cell => cell.version === 'v3' && cell.mode === 'comprehensive' && cell.variant === 'vulnerable' && cell.stack === stack);
// A repaired oracle case demonstrates the full find-to-repair workflow only
// when the producer also reported the matching supported finding.
const successful = new Set(eligible.filter(cell => found(cell) && measuredRate([cell], results, 'repair', corpus).numerator === 1).map(cell => cell.caseId));
const successful = new Set(eligible.filter(cell => {
const result = results.get(cell.id);
return result?.findings.some(finding => trustedFindingVerification(cell, result, finding, 'repair', corpus));
}).map(cell => cell.caseId));
return [stack, { correctHeldOutRepairs: successful.size, denominator: new Set(eligible.map(cell => cell.caseId)).size }];
}));
const assessedAll = observations.length === matrix.cells.length && groups.every(group => !group.unadjudicated);
+50
View File
@@ -106,6 +106,56 @@ describe('CSO immutable public archive cache', () => {
expect(copies.map(item => createHash('sha256').update(fs.readFileSync(item.path)).digest('hex')).sort()).toEqual([first.digest, second.digest].sort());
});
test('bounds every public operation by an optional absolute deadline', () => {
const { root, staging, cache } = fixture(1024);
const archive = stage(staging, 'deadline', 'bounded');
const expired = Date.now() - 1;
expect(code(() => cache.promote(archive.path, archive.digest, expired))).toBe('DEADLINE');
const entry = cache.promote(archive.path, archive.digest);
expect(code(() => cache.get(entry.sha256, { deadline: expired }))).toBe('DEADLINE');
expect(code(() => cache.stats({ deadline: expired }))).toBe('DEADLINE');
const destination = path.join(root, 'deadline-materialization'); fs.mkdirSync(destination, { mode: 0o700 });
expect(code(() => cache.materialize([entry.sha256], destination, { deadline: expired }))).toBe('DEADLINE');
expect(fs.readdirSync(destination)).toEqual([]);
});
test('cancels between archive-copy chunks and removes the partial incoming object', () => {
const { staging, cacheRoot, cache } = fixture(512 * 1024);
const archive = stage(staging, 'cancel-copy', Buffer.alloc(192 * 1024, 0x61));
const controller = new AbortController(), original = fs.writeSync.bind(fs); let archiveWrites = 0;
const writer = spyOn(fs, 'writeSync').mockImplementation(((fd: number, buffer: string | NodeJS.ArrayBufferView,
offsetOrPosition?: number | null, lengthOrEncoding?: number | BufferEncoding, position?: number | null) => {
const written = (original as any)(fd, buffer, offsetOrPosition, lengthOrEncoding, position);
if (Buffer.isBuffer(buffer) && lengthOrEncoding === 64 * 1024 && written > 0) { archiveWrites++; controller.abort(); }
return written;
}) as typeof fs.writeSync);
try {
expect(code(() => cache.promote(archive.path, archive.digest, { deadline: Date.now() + 60_000, signal: controller.signal }))).toBe('CANCELLED');
} finally { writer.mockRestore(); }
expect(archiveWrites).toBe(1);
expect(fs.readdirSync(path.join(cacheRoot, 'incoming'))).toEqual([]);
expect(fs.readdirSync(path.join(cacheRoot, 'entries'))).toEqual([]);
expect(fs.readdirSync(path.join(cacheRoot, 'metadata'))).toEqual([]);
});
test('cancels between cache-hit read chunks without damaging the immutable entry', () => {
const { staging, cacheRoot, cache } = fixture(512 * 1024);
const archive = stage(staging, 'cancel-read', Buffer.alloc(192 * 1024, 0x62)), entry = cache.promote(archive.path, archive.digest);
const controller = new AbortController(), original = fs.readSync.bind(fs); let archiveReads = 0;
const reader = spyOn(fs, 'readSync').mockImplementation(((fd: number, buffer: NodeJS.ArrayBufferView,
offset: number, length: number, position: number | null) => {
const read = original(fd, buffer, offset, length, position);
if (length === 64 * 1024 && read > 0) { archiveReads++; controller.abort(); }
return read;
}) as typeof fs.readSync);
try {
expect(code(() => cache.get(entry.sha256, { signal: controller.signal }))).toBe('CANCELLED');
} finally { reader.mockRestore(); }
expect(archiveReads).toBe(1);
expect(fs.existsSync(entry.path)).toBe(true);
expect(fs.readdirSync(path.join(cacheRoot, 'metadata'))).toEqual([`${entry.sha256}.json`]);
});
test('never overwrites an immutable entry and recovers interrupted entry publication', () => {
const { staging, cacheRoot, cache } = fixture();
const archive = stage(staging, 'one', 'same bytes');
+15 -2
View File
@@ -43,11 +43,13 @@ function portableSkill(version: 'v2' | 'v3', sectionMarker: string): string {
function syntheticResult(cell: EvalCell): EvalResult {
const runtime = cell.mode === 'comprehensive';
const positive = cell.variant === 'vulnerable';
const repairEvidenceHash = 'c'.repeat(64), recheckEvidenceHash = 'd'.repeat(64);
return { cellId: cell.id, sourceHash: cell.sourceHash, skillHash: cell.skillHash, model: cell.model, host: cell.host, budgetSeconds: cell.budgetSeconds,
reportPresent: true, reportComplete: true,
findings: positive ? [{ id: 'finding-1', evidence: cell.version === 'v2' ? 'legacy_review' : 'supported', claimedTested: runtime && cell.version === 'v3', judgment: 'correct', matchedCaseId: cell.caseId }] : [],
findings: positive ? [{ id: 'finding-1', evidence: cell.version === 'v2' ? 'legacy_review' : 'supported', claimedTested: runtime && cell.version === 'v3', judgment: 'correct', matchedCaseId: cell.caseId,
...(runtime && cell.version === 'v3' ? { trustedVerification: { repair: 'passed' as const, repairEvidenceHash, recheck: 'passed' as const, recheckEvidenceHash } } : {}) }] : [],
setup: runtime ? 'passed' : 'not_attempted', reproduction: runtime && positive ? 'passed' : 'not_attempted', repair: runtime && positive ? 'passed' : 'not_attempted', recheck: runtime && positive ? 'passed' : 'not_attempted',
...(runtime && positive ? { oracleEvidenceHash: 'c'.repeat(64), oracleVersion: CORPUS_VERSION, currentSourceHash: corpus.cases.find(fixture => fixture.id === cell.caseId)!.filesHash.fixed, recheckEvidenceHash: 'd'.repeat(64) } : {}),
...(runtime && positive ? { oracleEvidenceHash: repairEvidenceHash, oracleVersion: CORPUS_VERSION, currentSourceHash: corpus.cases.find(fixture => fixture.id === cell.caseId)!.filesHash.fixed, recheckEvidenceHash } : {}),
heldOutAssertionsPassed: runtime && positive, freshRecheck: runtime && positive, latencyMs: 1500, firstUsefulResultMs: positive ? 500 : null,
};
}
@@ -247,6 +249,14 @@ describe('CSO matched evaluation accounting', () => {
const result = syntheticResult(cell); delete result.oracleEvidenceHash;
const score = scoreEval(matrix, [result]); expect(group(score, 'v3', 'comprehensive').falseTested).toBe(1); expect(group(score, 'v3', 'comprehensive').repair.numerator).toBe(0);
});
test('one trusted cell repair cannot certify another claimed-tested finding', () => {
const results = matrix.cells.map(syntheticResult), cell = matrix.cells.find(item => item.version === 'v3' && item.mode === 'comprehensive' && item.variant === 'vulnerable')!, result = results.find(item => item.cellId === cell.id)!;
result.findings.push({ id: 'unbound-tested-claim', evidence: 'supported', claimedTested: true, judgment: 'correct', matchedCaseId: cell.caseId });
const score = scoreEval(matrix, results, qualification);
expect(group(score, 'v3', 'comprehensive').falseTested).toBe(1); expect(score.gates.zeroFalselyTestedRepairs).toBe('fail'); expect(score.status).toBe('partial');
result.findings[1].trustedVerification = structuredClone(result.findings[0].trustedVerification);
expect(() => scoreEval(matrix, results, qualification)).toThrow('DUPLICATE_TRUSTED_REPAIR_BINDING');
});
test('per-stack held-out repair requires the matching supported discovery', () => {
const results = matrix.cells.map(syntheticResult);
for (let index = 0; index < matrix.cells.length; index++) {
@@ -280,6 +290,9 @@ describe('CSO matched evaluation accounting', () => {
expect(() => scoreEval(matrix, [result, result])).toThrow('UNKNOWN_OR_DUPLICATE');
expect(() => scoreEval(matrix, [{ ...result, model: 'other-model' }])).toThrow('UNMATCHED_EVAL_RESULT');
expect(() => scoreEval(matrix, [{ ...result, setup: 'passed' }])).toThrow('DAILY_EVAL_EXECUTED');
const comprehensive = syntheticResult(matrix.cells.find(cell => cell.version === 'v3' && cell.mode === 'comprehensive' && cell.variant === 'vulnerable')!);
comprehensive.findings[0].trustedVerification = { repair: 'failed', repairEvidenceHash: 'not-a-hash', recheck: 'not_attempted' };
expect(() => scoreEval(matrix, [comprehensive])).toThrow('INVALID_TRUSTED_FINDING_VERIFICATION');
});
test('missing containment tests remain unmeasured and a failed canary fails qualification', () => {
const results = matrix.cells.map(syntheticResult);
+38 -1
View File
@@ -1,4 +1,4 @@
import { afterEach, describe, expect, test } from 'bun:test';
import { afterEach, describe, expect, spyOn, test } from 'bun:test';
import * as fs from 'node:fs';
import * as os from 'node:os';
import * as path from 'node:path';
@@ -218,6 +218,43 @@ describe('CSO constrained dependency preparation executor', () => {
throw new Error('expected poisoned cache reuse to fail');
});
test('threads cancellation into retained-closure cache verification', async () => {
const source = snapshot('node'), plan = inspectPreparation(source, 'node'), admission = admitPreparationRuntime({ plan, platform: 'linux/amd64', catalog: catalog('node') });
const runner = new FakeRunner(), cache = cacheFixture(), executor = new PreparationExecutor({ cache, runner }), deadline = Date.now() + 60_000;
const closure = await executor.acquire({ plan, admission, snapshot: source, deadline });
const controller = new AbortController(), original = fs.readSync.bind(fs); let cacheReads = 0;
const reader = spyOn(fs, 'readSync').mockImplementation(((fd: number, buffer: NodeJS.ArrayBufferView,
offset: number, length: number, position: number | null) => {
const read = original(fd, buffer, offset, length, position);
if (length === 64 * 1024 && read > 0) { cacheReads++; controller.abort(); }
return read;
}) as typeof fs.readSync);
try {
await executor.acquire({ plan, admission, snapshot: source, deadline, signal: controller.signal, offline: true, existingClosure: closure });
throw new Error('expected retained closure verification to be cancelled');
} catch (error) { expect(errorCode(error)).toBe('CANCELLED'); }
finally { reader.mockRestore(); }
expect(cacheReads).toBe(1); expect(runner.acquireCalls).toBe(1);
});
test('cancels acquisition while independently hashing staged archive bytes', async () => {
const source = snapshot('node'), plan = inspectPreparation(source, 'node'), admission = admitPreparationRuntime({ plan, platform: 'linux/amd64', catalog: catalog('node') });
const runner = new FakeRunner(), cache = cacheFixture(), executor = new PreparationExecutor({ cache, runner });
const controller = new AbortController(), original = fs.readSync.bind(fs); let stagedReads = 0;
const reader = spyOn(fs, 'readSync').mockImplementation(((fd: number, buffer: NodeJS.ArrayBufferView,
offset: number, length: number, position: number | null) => {
const read = original(fd, buffer, offset, length, position);
if (length === 64 * 1024 && read > 0) { stagedReads++; controller.abort(); }
return read;
}) as typeof fs.readSync);
try {
await executor.acquire({ plan, admission, snapshot: source, deadline: Date.now() + 60_000, signal: controller.signal });
throw new Error('expected staged archive verification to be cancelled');
} catch (error) { expect(errorCode(error)).toBe('CANCELLED'); }
finally { reader.mockRestore(); }
expect(stagedReads).toBe(1); expect(runner.acquireCalls).toBe(1); expect(cache.stats().entries).toBe(0);
});
test('run-owned archive copies survive a cache eviction between validation and runner start', async () => {
const source = snapshot('node'), plan = inspectPreparation(source, 'node'), admission = admitPreparationRuntime({ plan, platform: 'linux/amd64', catalog: catalog('node') });
const runner = new FakeRunner(), base = root('cso-eviction-race-'), staging = path.join(base, 'staging'); fs.mkdirSync(staging, { mode: 0o700 });
+1
View File
@@ -24,6 +24,7 @@ describe('CSO retained snapshot identity',()=>{
for(const entry of [dependency,venv]){expect(entry).toMatchObject({originalHash:'not-read',transformation:'excluded: host dependencies, metadata, state, or agent configuration'});expect(entry?.executionHash).toBeUndefined();}
expect(agent).toMatchObject({originalHash:sha256('untrusted agent instructions\n'),transformation:'excluded: host dependencies, metadata, state, or agent configuration'});expect(agent?.executionHash).toBeUndefined();expect(fs.readFileSync(path.join(runDir,'readable','packages/web/.agents/skills/x/SKILL.md'),'utf8')).toBe('untrusted agent instructions\n');
});
test('validates the captured execution identity when uppercase and lowercase paths coexist',async()=>{const {repo,runDir,git}=fixture();fs.writeFileSync(path.join(repo,'README.md'),'fixture documentation\n');git('add','README.md');git('commit','-qm','add uppercase path');const manifest=await capture(repo,runDir);expect(manifest.entries.filter(item=>item.executionHash).map(item=>item.path)).toEqual(['README.md','app.js']);expect(()=>assertSnapshot(runDir,manifest)).not.toThrow();});
test('binds original, sanitized, and opaque path identities',async()=>{const {repo,runDir}=fixture(),manifest=await capture(repo,runDir);assertSnapshot(runDir,manifest);const changedEntry=structuredClone(manifest);changedEntry.entries[0].originalHash='a'.repeat(64);expect(()=>assertSnapshot(runDir,changedEntry)).toThrow('original identity');const changedRoot=structuredClone(manifest);changedRoot.originalHash='b'.repeat(64);expect(()=>assertSnapshot(runDir,changedRoot)).toThrow('original identity');const changedPathId=structuredClone(manifest);changedPathId.entries[0].pathId='c'.repeat(32);expect(()=>assertSnapshot(runDir,changedPathId)).toThrow('invalid source entry');});
test.skipIf(process.platform==='win32')('preserves retained source modes under the required restrictive umask',async()=>{const {repo,runDir,git}=fixture(),script=path.join(repo,'tool.sh');fs.writeFileSync(script,'#!/bin/sh\nexit 0\n');fs.chmodSync(path.join(repo,'app.js'),0o644);fs.chmodSync(script,0o755);git('add','app.js','tool.sh');git('commit','-qm','add executable input');const previous=process.umask(0o077);let manifest;try{manifest=await capture(repo,runDir);}finally{process.umask(previous);}expect(manifest.entries.find(item=>item.path==='app.js')?.mode).toBe(0o644);expect(manifest.entries.find(item=>item.path==='tool.sh')?.mode).toBe(0o755);expect(fs.statSync(path.join(runDir,'snapshot','app.js')).mode&0o777).toBe(0o644);expect(fs.statSync(path.join(runDir,'snapshot','tool.sh')).mode&0o777).toBe(0o755);assertSnapshot(runDir,manifest);});
test('rejects a replaced snapshot root before following it',async()=>{const {repo,runDir}=fixture(),manifest=await capture(repo,runDir),snapshot=path.join(runDir,'snapshot'),replacement=path.join(runDir,'snapshot-replacement');fs.renameSync(snapshot,replacement);fs.symlinkSync(replacement,snapshot);expect(()=>assertSnapshot(runDir,manifest)).toThrow('private owned directory');});
+4
View File
@@ -29,6 +29,9 @@ describe('CSO dirty snapshot boundary',()=>{
test('rejects a source mode change between copying and manifest persistence',async()=>{const p=repo(),run=newRun(p),source=path.join(p,'tracked.txt'),target=path.join(run.dir,'snapshot','tracked.txt'),write=fs.writeFileSync,patched=spyOn(fs,'writeFileSync').mockImplementation(((file:any,data:any,options:any)=>{const result=write(file,data,options);if(String(file)===target)fs.chmodSync(source,0o755);return result;}) as typeof fs.writeFileSync);try{await expect(capture(p,run.dir)).rejects.toThrow('Source changed during capture');}finally{patched.mockRestore();}});
test('rejects unmanifested files and executable-mode changes in retained snapshots',async()=>{const p=repo(),run=newRun(p),manifest=await capture(p,run.dir),snapshot=path.join(run.dir,'snapshot');assertSnapshot(run.dir,manifest);fs.writeFileSync(path.join(snapshot,'injected.js'),'malicious\n');expect(()=>assertSnapshot(run.dir,manifest)).toThrow('membership');fs.unlinkSync(path.join(snapshot,'injected.js'));fs.chmodSync(path.join(snapshot,'tracked.txt'),0o755);expect(()=>assertSnapshot(run.dir,manifest)).toThrow('changed');});
test('retention skips a run with a live mutation lease',()=>{const now=Date.now(),repo='d'.repeat(24),run=`${now-31*86400_000}-${'1'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run)),payload=path.join(dir,'report.json');fs.writeFileSync(payload,'live lease payload\n',{mode:0o600});withLock(dir,()=>{retention(now);expect(fs.existsSync(payload)).toBe(true);});retention(now);expect(fs.existsSync(dir)).toBe(false);});
test('bounded retention stops before destructive cleanup when its maintenance budget is exhausted',()=>{const now=Date.now(),repo='e'.repeat(24),run=`${now-31*86400_000}-${'8'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run));fs.writeFileSync(path.join(dir,'report.json'),'expired private state\n',{mode:0o600});const result=retention(now,{deadlineMs:0,maxEntries:1});expect(result).toEqual({complete:false,visited:0});expect(fs.existsSync(dir)).toBe(true);retention(now);expect(fs.existsSync(dir)).toBe(false);});
test('bounded retention completes discovery before deleting any run',()=>{const now=Date.now(),repo='f'.repeat(24),run=`${now-31*86400_000}-${'9'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run));fs.writeFileSync(path.join(dir,'report.json'),'expired private state\n',{mode:0o600});const result=retention(now,{deadlineMs:now+60_000,maxEntries:6});expect(result).toEqual({complete:false,visited:6});expect(fs.existsSync(dir)).toBe(true);retention(now);expect(fs.existsSync(dir)).toBe(false);});
test('bounded retention resumes an interrupted tree deletion without one recursive walk',()=>{const previous=process.env.GSTACK_HOME,isolate=tmp();process.env.GSTACK_HOME=isolate;try{const now=Date.now(),repo='1'.repeat(24),run=`${now-31*86400_000}-${'a'.repeat(16)}`,repoDir=path.join(privateRoot(),repo),dir=secureDirectory(path.join(repoDir,run)),snapshot=secureDirectory(path.join(dir,'snapshot'));for(let index=0;index<20;index++)fs.writeFileSync(path.join(snapshot,`${index}.txt`),'private state\n',{mode:0o600});const result=retention(now,{deadlineMs:now+60_000,maxEntries:30});expect(result).toEqual({complete:false,visited:30});expect(fs.existsSync(dir)).toBe(false);expect(fs.readdirSync(repoDir).some(name=>name.startsWith(`.retired-${run}-`))).toBe(true);retention(now);expect(fs.readdirSync(repoDir).some(name=>name.startsWith(`.retired-${run}-`))).toBe(false);}finally{if(previous===undefined)delete process.env.GSTACK_HOME;else process.env.GSTACK_HOME=previous;}});
test('repair bundles expire from their authenticated bundle time, independent of run age',()=>{const now=Date.now(),repo='c'.repeat(24),run=`${now-31*86400_000}-${'2'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run)),bundles=secureDirectory(path.join(dir,'bundles')),expiredId='3'.repeat(32),retainedId='4'.repeat(32),record=(id:string,created:number)=>{const createdAt=new Date(created).toISOString();return{schemaVersion:3,id,runId:run,createdAt,expiresAt:new Date(created+30*86400_000).toISOString(),verification:{id,runId:run,createdAt}};};writeHelperJson(path.join(bundles,`${expiredId}.json`),record(expiredId,now-31*86400_000));writeHelperJson(path.join(bundles,`${retainedId}.json`),record(retainedId,now-86400_000));retention(now);expect(fs.existsSync(path.join(bundles,`${expiredId}.json`))).toBe(false);expect(fs.existsSync(path.join(bundles,`${retainedId}.json`))).toBe(true);expect(fs.existsSync(dir)).toBe(true);retention(now+30*86400_000);expect(fs.existsSync(dir)).toBe(false);});
test('replay temporary state survives cleanup failure until every detached watchdog acknowledges cleanup',()=>{const now=Date.now(),repo='b'.repeat(24),run=`${now}-${'5'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run)),control=secureDirectory(path.join(dir,'supervision','repair-attempt')),docker=secureDirectory(path.join(dir,'preparation-execution','offline-attempt'));fs.writeFileSync(path.join(control,'attempt.ready'),'ready\n',{mode:0o600});fs.writeFileSync(path.join(control,'attempt.terminal'),'normal cleanup complete\n',{mode:0o600});fs.writeFileSync(path.join(docker,'watchdog.ready'),'ready\n',{mode:0o600});fs.writeFileSync(path.join(docker,'watchdog.event'),'cleanup incomplete; retrying exact journaled resources\n',{mode:0o600});finalizeReplayTemporary(dir);expect(fs.existsSync(path.join(dir,'.ephemeral-replay.json'))).toBe(true);retention(now);expect(fs.existsSync(control)).toBe(true);expect(fs.existsSync(docker)).toBe(true);fs.writeFileSync(path.join(control,'attempt.stopped'),'normal cleanup acknowledged\n',{mode:0o600});retention(now);expect(fs.existsSync(dir)).toBe(true);fs.writeFileSync(path.join(docker,'watchdog.stopped'),'normal cleanup acknowledged\n',{mode:0o600});retention(now);expect(fs.existsSync(dir)).toBe(false);});
test('retention recovers a hidden run retirement left by a dead helper',()=>{const now=Date.now(),repo='a'.repeat(24),run=`${now-31*86400_000}-${'6'.repeat(16)}`,retired=secureDirectory(path.join(privateRoot(),repo,`.retired-${run}-${'7'.repeat(32)}`));fs.writeFileSync(path.join(retired,'payload'),'private state\n',{mode:0o600});retention(now);expect(fs.existsSync(retired)).toBe(false);});
@@ -161,6 +164,7 @@ describe('private state and process output',()=>{
test('an active lease binds a separate decision to its candidate and active inode',()=>{const dir=tmp();withLock(dir,()=>{const leases=path.join(dir,'.mutation-lock-leases'),names=fs.readdirSync(leases);expect(names.length).toBe(3);const candidate=names.find(name=>name.endsWith('.json'))!,decision=names.find(name=>name.endsWith('.decision'))!,active=names.find(name=>name.includes('.active.'))!,candidateStat=fs.statSync(path.join(leases,candidate)),decisionStat=fs.statSync(path.join(leases,decision)),activeStat=fs.statSync(path.join(leases,active)),record=JSON.parse(fs.readFileSync(path.join(leases,decision),'utf8'));expect(`${candidateStat.dev}:${candidateStat.ino}`).toBe(`${activeStat.dev}:${activeStat.ino}`);expect(candidateStat.nlink).toBe(2);expect(activeStat.nlink).toBe(2);expect(`${decisionStat.dev}:${decisionStat.ino}`).not.toBe(`${candidateStat.dev}:${candidateStat.ino}`);expect(decisionStat.nlink).toBe(1);expect(record.kind).toBe('ticket');expect(record.candidateDev).toBe(String(candidateStat.dev));expect(record.candidateIno).toBe(String(candidateStat.ino));});expect(fs.readdirSync(path.join(dir,'.mutation-lock-leases'))).toEqual([]);});
test('post-link verification failure cleans every recorded lease phase',()=>{const dir=tmp();expect(withLock(dir,()=>1)).toBe(1);const leases=path.join(dir,'.mutation-lock-leases'),fakeToken='f'.repeat(32),fakeTemp=path.join(leases,`${fakeToken}.json.tmp.${process.pid}.deadbeef`),link=fs.linkSync,injectedOwner=JSON.stringify({pid:process.pid,token:fakeToken,createdAt:Date.now()})+'\n';let injected=false,caught:unknown;const publisher=spyOn(fs,'linkSync').mockImplementation(((source:any,target:any)=>{const result=link(source,target);if(!injected&&String(target).includes('.active.')){injected=true;fs.writeFileSync(fakeTemp,injectedOwner,{mode:0o600,flag:'wx'});}return result;}) as typeof fs.linkSync);try{withLock(dir,()=>1);}catch(error){caught=error;}finally{publisher.mockRestore();}expect(injected).toBe(true);expect(caught).toMatchObject({code:'INSUFFICIENT_CAPACITY'});expect(fs.readdirSync(leases)).toEqual([path.basename(fakeTemp)]);fs.unlinkSync(fakeTemp);expect(withLock(dir,()=>2)).toBe(2);expect(fs.readdirSync(leases)).toEqual([]);});
test('an async callback retains its lease until settlement without blocking its own event loop',async()=>{const dir=tmp();let release!:()=>void;const pending=new Promise<void>(resolve=>{release=resolve;}),outer=withLock(dir,async()=>{await pending;return 7;}),started=Date.now();expect(()=>withLock(dir,()=>99)).toThrow('Another operation in this helper');expect(Date.now()-started).toBeLessThan(200);release();expect(await outer).toBe(7);expect(withLock(dir,()=>9)).toBe(9);});
test('a synchronous exact-release failure is attempted only once',()=>{const dir=tmp(),lstat=fs.lstatSync;let observed=0,reader:any;try{expect(()=>withLock(dir,()=>{const leases=path.join(dir,'.mutation-lock-leases'),names=fs.readdirSync(leases),candidate=path.join(leases,names.find(name=>name.endsWith('.json'))!),active=path.join(leases,names.find(name=>name.includes('.active.'))!),value=fs.readFileSync(active);fs.unlinkSync(active);fs.writeFileSync(active,value,{mode:0o600,flag:'wx'});reader=spyOn(fs,'lstatSync').mockImplementation(((file:any,options?:any)=>{if(String(file)===candidate)observed++;return options===undefined?lstat(file):lstat(file,options);}) as typeof fs.lstatSync);})).toThrow('active phase changed before cleanup');expect(observed).toBe(1);}finally{reader?.mockRestore();}});
test('exact release rejects active or decision inode substitution and overrides callback success or failure',()=>{for(const phase of ['active','decision'] as const){const dir=tmp();let replacement='';expect(()=>withLock(dir,()=>{const leases=path.join(dir,'.mutation-lock-leases'),name=fs.readdirSync(leases).find(value=>phase==='active'?value.includes('.active.'):value.endsWith('.decision'))!,lease=path.join(leases,name),value=fs.readFileSync(lease);fs.unlinkSync(lease);fs.writeFileSync(lease,value,{mode:0o600,flag:'wx'});replacement=lease;if(phase==='active')throw new Error('callback failed');return 1;})).toThrow(/changed before (cleanup|exact release)/);expect(fs.existsSync(replacement)).toBe(true);fs.rmSync(dir,{recursive:true,force:true});}});
test('concurrent stale-lock recovery never admits overlapping report writers',async()=>{const dir=tmp(),lock=path.join(dir,'.mutation-lock'),script=path.join(dir,'racer.ts');fs.mkdirSync(lock);fs.writeFileSync(path.join(lock,'owner.json'),JSON.stringify({pid:2147483647,token:'stale',createdAt:0,expiresAt:0}));fs.writeFileSync(script,`import fs from 'node:fs';import path from 'node:path';import {withLock} from ${JSON.stringify(path.resolve(import.meta.dir,'../lib/cso/state.ts'))};const dir=process.env.RACE_DIR!;try{await withLock(dir,async()=>{const active=path.join(dir,'active');try{fs.writeFileSync(active,String(process.pid),{flag:'wx'});}catch{fs.appendFileSync(path.join(dir,'overlap'),'yes\\n');}await Bun.sleep(50);try{if(fs.readFileSync(active,'utf8')===String(process.pid))fs.unlinkSync(active);}catch{}});}catch{}`);const children=Array.from({length:8},()=>spawn(process.execPath,[script],{env:{...process.env,RACE_DIR:dir},stdio:'ignore'}));await Promise.all(children.map(child=>new Promise<void>(resolve=>child.on('close',()=>resolve()))));expect(fs.existsSync(path.join(dir,'overlap'))).toBe(false);});
test('concurrent recovery of one dead hard-link publication has a winner and no raw race errors',async()=>{