diff --git a/packages/coding-agent/.changes/res-1267-atomic-persistence.md b/packages/coding-agent/.changes/res-1267-atomic-persistence.md new file mode 100644 index 0000000000..7f978f9a85 --- /dev/null +++ b/packages/coding-agent/.changes/res-1267-atomic-persistence.md @@ -0,0 +1 @@ +- Made every durable JSON/JSONL state write crash-safe through one shared atomic-write owner (temp file + rename, Windows rename retry): auth.json is no longer written in place (an interrupted write can no longer log you out everywhere), the auth migration writes its destination before destroying its sources, racing first-time settings writers no longer silently discard each other, and the kernel bootstrap lock can no longer be stolen mid-reclaim. Session files now repair crash damage (torn tails, zero-filled records) at open instead of silently losing the next message, and a session lease whose owner file is momentarily unreadable is no longer treated as stale and destroyed. diff --git a/packages/coding-agent/src/core/auth-storage.ts b/packages/coding-agent/src/core/auth-storage.ts index 389bc64f85..981e37661c 100644 --- a/packages/coding-agent/src/core/auth-storage.ts +++ b/packages/coding-agent/src/core/auth-storage.ts @@ -15,10 +15,11 @@ import { type OAuthProviderId, } from "@earendil-works/pi-ai"; import { getOAuthApiKey, getOAuthProvider, getOAuthProviders } from "@earendil-works/pi-ai/oauth"; -import { chmodSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "fs"; +import { closeSync, existsSync, fchmodSync, mkdirSync, openSync, readFileSync, writeSync } from "fs"; import { dirname, join } from "path"; import lockfile from "proper-lockfile"; import { getAgentDir } from "../config.js"; +import { realpathIfPresentSync, writeFileAtomicSync } from "../utils/atomic-file.js"; import { clearPrimeCliCredentials, getPrimeCliConfigPath, @@ -116,9 +117,21 @@ export class FileAuthStorageBackend implements AuthStorageBackend { } private ensureFileExists(): void { - if (!existsSync(this.authPath)) { - writeFileSync(this.authPath, "{}", "utf-8"); - chmodSync(this.authPath, 0o600); + let descriptor: number; + try { + // Exclusive create: a racing initializer must never replace saved credentials. + descriptor = openSync(this.authPath, "wx", 0o600); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") { + throw error; + } + return; + } + try { + writeSync(descriptor, "{}"); + fchmodSync(descriptor, 0o600); // Exact bits despite the umask. + } finally { + closeSync(descriptor); } } @@ -171,8 +184,7 @@ export class FileAuthStorageBackend implements AuthStorageBackend { const current = existsSync(this.authPath) ? readFileSync(this.authPath, "utf-8") : undefined; const { result, next } = fn(current); if (next !== undefined) { - writeFileSync(this.authPath, next, "utf-8"); - chmodSync(this.authPath, 0o600); + writeFileAtomicSync(realpathIfPresentSync(this.authPath), next, { mode: 0o600 }); } return result; } finally { @@ -216,8 +228,7 @@ export class FileAuthStorageBackend implements AuthStorageBackend { const { result, next } = await fn(current); throwIfCompromised(); if (next !== undefined) { - writeFileSync(this.authPath, next, "utf-8"); - chmodSync(this.authPath, 0o600); + writeFileAtomicSync(realpathIfPresentSync(this.authPath), next, { mode: 0o600 }); } throwIfCompromised(); return result; diff --git a/packages/coding-agent/src/core/cron-jobs.ts b/packages/coding-agent/src/core/cron-jobs.ts index c9dede0476..28fa49b139 100644 --- a/packages/coding-agent/src/core/cron-jobs.ts +++ b/packages/coding-agent/src/core/cron-jobs.ts @@ -1,16 +1,8 @@ import { randomUUID } from "node:crypto"; -import { - closeSync, - existsSync, - fsyncSync, - mkdirSync, - openSync, - readFileSync, - renameSync, - writeFileSync, -} from "node:fs"; +import { existsSync, mkdirSync, readFileSync, renameSync } from "node:fs"; import { dirname, join, resolve } from "node:path"; import { lockSync } from "proper-lockfile"; +import { writeFileAtomicSync } from "../utils/atomic-file.js"; import { getSessionArtifactPathForFile } from "./session-manager.js"; export type AgentCronJobStatus = "active" | "paused" | "completed" | "cancelled"; @@ -1558,27 +1550,8 @@ function writeJobsFile(path: string, jobs: readonly AgentCronJob[], mergeCurrent } function writeJobsState(path: string, state: CronJobsState): void { - const directory = dirname(path); - mkdirSync(directory, { recursive: true, mode: 0o700 }); - const tempPath = `${path}.${process.pid}.${randomUUID()}.tmp`; - const descriptor = openSync(tempPath, "w", 0o600); - try { - writeFileSync(descriptor, `${JSON.stringify(state, null, 2)}\n`, "utf-8"); - fsyncSync(descriptor); - } finally { - closeSync(descriptor); - } - renameSync(tempPath, path); - try { - const directoryDescriptor = openSync(directory, "r"); - try { - fsyncSync(directoryDescriptor); - } finally { - closeSync(directoryDescriptor); - } - } catch { - // Directory fsync is unavailable on some platforms; the atomic rename still protects readers. - } + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + writeFileAtomicSync(path, `${JSON.stringify(state, null, 2)}\n`, { mode: 0o600, fsync: true, fsyncDir: true }); } function claimDueInState(state: CronJobsState, dueAt: Date, claimedAt: Date): AgentCronDispatch[] { diff --git a/packages/coding-agent/src/core/kernel/bootstrap.ts b/packages/coding-agent/src/core/kernel/bootstrap.ts index 2390749308..e432a5591e 100644 --- a/packages/coding-agent/src/core/kernel/bootstrap.ts +++ b/packages/coding-agent/src/core/kernel/bootstrap.ts @@ -9,6 +9,7 @@ import { createInterface } from "node:readline/promises"; import { setTimeout as sleep } from "node:timers/promises"; import { fileURLToPath } from "node:url"; import { getPackageDir } from "../../config.js"; +import { tryAcquireDirLock } from "../../utils/dir-lock.js"; import type { PythonSkillRuntimeInfo } from "../skills.js"; const BOOTSTRAP_SCHEMA = 9; @@ -449,16 +450,6 @@ function processIsRunning(pid: number): boolean { } } -async function readLockPid(lockDir: string): Promise { - try { - const raw = await readFile(path.join(lockDir, "pid"), "utf8"); - const pid = Number.parseInt(raw.trim(), 10); - return Number.isInteger(pid) && pid > 0 ? pid : null; - } catch { - return null; - } -} - async function lockMissingPidIsStale(lockDir: string): Promise { try { const lockStat = await stat(lockDir); @@ -473,19 +464,13 @@ async function acquireBootstrapLock(venv: string): Promise<() => Promise> await mkdir(path.dirname(lockDir), { recursive: true }); for (;;) { - try { - await mkdir(lockDir); - await writeFile(path.join(lockDir, "pid"), `${process.pid}\n`, "utf8"); + const attempt = await tryAcquireDirLock(lockDir, async (ownerPid) => + ownerPid === undefined ? !(await lockMissingPidIsStale(lockDir)) : processIsRunning(ownerPid), + ); + if (attempt === "acquired") { return () => rm(lockDir, { recursive: true, force: true }); - } catch (error) { - if (!isNodeError(error, "EEXIST")) throw error; - - const pid = await readLockPid(lockDir); - if (pid === null ? await lockMissingPidIsStale(lockDir) : !processIsRunning(pid)) { - await rm(lockDir, { recursive: true, force: true }); - continue; - } - + } + if (attempt === "held") { await sleep(BOOTSTRAP_LOCK_RETRY_MS); } } diff --git a/packages/coding-agent/src/core/model-registry.ts b/packages/coding-agent/src/core/model-registry.ts index 4ebf6c3b8d..6a3d2be43e 100644 --- a/packages/coding-agent/src/core/model-registry.ts +++ b/packages/coding-agent/src/core/model-registry.ts @@ -22,12 +22,13 @@ import { } from "@earendil-works/pi-ai"; import { registerBuiltinMcpOAuthProviders } from "@earendil-works/pi-ai/mcp"; import { registerOAuthProvider, resetOAuthProviders } from "@earendil-works/pi-ai/oauth"; -import { existsSync, readFileSync, renameSync, writeFileSync } from "fs"; +import { existsSync, readFileSync } from "fs"; import { dirname, join } from "path"; import { type Static, type TProperties, Type } from "typebox"; import type { Validator } from "typebox/compile"; import type { TLocalizedValidationError } from "typebox/error"; import { getAgentDir } from "../config.js"; +import { writeFileAtomicSync } from "../utils/atomic-file.js"; import type { AuthSourceToken, AuthStatus, AuthStorage } from "./auth-storage.js"; import { PRIME_INFERENCE_PROVIDER_ID } from "./prime-inference-auth.js"; import { @@ -926,9 +927,7 @@ export class ModelRegistry { return; } try { - const tmpPath = `${cachePath}.${process.pid}.tmp`; - writeFileSync(tmpPath, JSON.stringify({ ...cache, modelIds: [...cache.modelIds] }), { mode: 0o600 }); - renameSync(tmpPath, cachePath); + writeFileAtomicSync(cachePath, JSON.stringify({ ...cache, modelIds: [...cache.modelIds] }), { mode: 0o600 }); } catch { // A failed cache write only requires a later refetch. } diff --git a/packages/coding-agent/src/core/refinement/refinement.ts b/packages/coding-agent/src/core/refinement/refinement.ts index 76cc701556..a8a6b18042 100644 --- a/packages/coding-agent/src/core/refinement/refinement.ts +++ b/packages/coding-agent/src/core/refinement/refinement.ts @@ -1,19 +1,10 @@ -import { randomUUID } from "node:crypto"; -import { - appendFileSync, - existsSync, - mkdirSync, - readFileSync, - renameSync, - statSync, - unlinkSync, - writeFileSync, -} from "node:fs"; +import { appendFileSync, existsSync, mkdirSync, readFileSync, statSync } from "node:fs"; import { join } from "node:path"; import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core"; import type { Model } from "@earendil-works/pi-ai"; import { completeSimple } from "@earendil-works/pi-ai"; import { getAgentDir } from "../../config.js"; +import { realpathIfPresentSync, writeFileAtomicSync } from "../../utils/atomic-file.js"; import { serializeConversation } from "../compaction/utils.js"; import { convertToLlm } from "../messages.js"; import type { CustomEntry } from "../session-manager.js"; @@ -344,17 +335,10 @@ export function mergeHarnessStates(globalState: HarnessState, localState?: Harne export function saveHarnessState(harnessStateDir: string, state: HarnessState): string { const statePath = getHarnessStatePath(harnessStateDir); - const tempPath = `${statePath}.${process.pid}.${randomUUID()}.tmp`; mkdirSync(harnessStateDir, { recursive: true }); - try { - const mode = existsSync(statePath) ? statSync(statePath).mode & 0o777 : 0o600; - writeFileSync(tempPath, `${JSON.stringify(state, null, 2)}\n`, { encoding: "utf8", mode }); - renameSync(tempPath, statePath); - } finally { - if (existsSync(tempPath)) { - unlinkSync(tempPath); - } - } + const targetPath = realpathIfPresentSync(statePath); + const mode = existsSync(targetPath) ? statSync(targetPath).mode & 0o777 : 0o600; + writeFileAtomicSync(targetPath, `${JSON.stringify(state, null, 2)}\n`, { mode }); return statePath; } diff --git a/packages/coding-agent/src/core/session-lease.ts b/packages/coding-agent/src/core/session-lease.ts index af0f205dbd..277e34f634 100644 --- a/packages/coding-agent/src/core/session-lease.ts +++ b/packages/coding-agent/src/core/session-lease.ts @@ -50,7 +50,7 @@ export class SessionLease { try { withLeaseGuard(this.directory, () => { const owner = readLeaseOwner(this.directory); - if (owner?.token === this.token) { + if (typeof owner === "object" && owner.token === this.token) { rmSync(this.directory, { recursive: true, force: true }); } }); @@ -83,9 +83,16 @@ export function canonicalSessionPath(sessionPath: string): string { } } -function readLeaseOwner(directory: string): SessionLeaseOwner | undefined { +// "absent" (missing/garbage) is safely stale; "unreadable" may be a LIVE lease and must never be reclaimed. +function readLeaseOwner(directory: string): SessionLeaseOwner | "absent" | "unreadable" { + let raw: string; try { - const parsed = JSON.parse(readFileSync(join(directory, "owner.json"), "utf8")) as Partial; + raw = readFileSync(join(directory, "owner.json"), "utf8"); + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ENOENT" ? "absent" : "unreadable"; + } + try { + const parsed = JSON.parse(raw) as Partial; if ( parsed.version !== 1 || typeof parsed.token !== "string" || @@ -93,11 +100,11 @@ function readLeaseOwner(directory: string): SessionLeaseOwner | undefined { typeof parsed.sessionPath !== "string" || typeof parsed.createdAt !== "string" ) { - return undefined; + return "absent"; } return parsed as SessionLeaseOwner; } catch { - return undefined; + return "absent"; } } @@ -298,11 +305,15 @@ export function acquireSessionLease( } catch (error) { rmSync(candidateDirectory, { recursive: true, force: true }); const code = (error as NodeJS.ErrnoException).code; - if (code !== "EEXIST" && code !== "ENOTEMPTY") { + // win32 reports rename-onto-existing-directory as EPERM/EACCES, not EEXIST. + if (code !== "EEXIST" && code !== "ENOTEMPTY" && code !== "EPERM" && code !== "EACCES") { throw error; } const existingOwner = readLeaseOwner(directory); - if (existingOwner && isLeaseOwnerAlive(existingOwner)) { + if (existingOwner === "unreadable") { + continue; + } + if (existingOwner !== "absent" && isLeaseOwnerAlive(existingOwner)) { throw new SessionAlreadyActiveError(canonicalPath, existingOwner.activeSessionId); } reclaimStaleLease(directory); @@ -310,7 +321,7 @@ export function acquireSessionLease( } const owner = existsSync(directory) ? readLeaseOwner(directory) : undefined; - if (owner && isLeaseOwnerAlive(owner)) { + if (typeof owner === "object" && isLeaseOwnerAlive(owner)) { throw new SessionAlreadyActiveError(canonicalPath, owner.activeSessionId); } throw new Error(`Could not acquire session lease: ${canonicalPath}`); diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index fc85697878..1a44e69263 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -3,22 +3,22 @@ import type { AssistantMessage, ImageContent, Message, ServiceTier, TextContent, import { randomUUID } from "crypto"; import { appendFileSync, - chmodSync, chownSync, + closeSync, existsSync, + fstatSync, mkdirSync, + openSync, readdirSync, readFileSync, - realpathSync, - renameSync, - rmSync, + readSync, statSync, - writeFileSync, } from "fs"; import { readdir, readFile, stat } from "fs/promises"; import { basename, dirname, join, resolve } from "path"; import { v7 as uuidv7 } from "uuid"; import { getAgentDir as getDefaultAgentDir, getSessionsDir } from "../config.js"; +import { realpathIfPresentSync, writeFileAtomicSync } from "../utils/atomic-file.js"; import { readFirstLineSync, readLinesAsBuffers } from "../utils/file-lines.js"; import { captureGitContext, type GitContext, gitContextsEqual } from "../utils/git.js"; import { @@ -60,15 +60,6 @@ const CONTENT_ENTRY_TYPES = new Set([ "branch_summary", ]); -function realpathIfPresent(path: string): string { - try { - return realpathSync(path); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") return path; - throw error; - } -} - function statMetadataIfPresent(path: string): { mode: number; uid: number; gid: number } | undefined { try { const { mode, uid, gid } = statSync(path); @@ -588,6 +579,125 @@ async function parseEntriesFromBufferAsync(buffer: Buffer): Promise return entries; } +// Crash damage (torn tail, zero-filled append) poisons the NEXT append into the +// same physical line, so the file is repaired once at open, not tolerated in memory. +const REPAIR_SUSPICION_WINDOW_BYTES = 1024 * 1024; + +// A bounded tail read gates the full repair scan: clean opens stay O(window). +function tailLooksDamaged(targetPath: string): boolean { + let descriptor: number; + try { + descriptor = openSync(targetPath, "r"); + } catch { + return false; + } + try { + const size = fstatSync(descriptor).size; + if (size === 0) return false; + const windowBytes = Math.min(size, REPAIR_SUSPICION_WINDOW_BYTES); + const window = Buffer.allocUnsafe(windowBytes); + readSync(descriptor, window, 0, windowBytes, size - windowBytes); + if (window.includes(0)) return true; + if (window[windowBytes - 1] !== 0x0a) return true; + const previousNewline = window.lastIndexOf(0x0a, windowBytes - 2); + // No boundary inside the window: the final line exceeds it; scan to be sure. + if (previousNewline === -1 && windowBytes < size) return true; + const lastLine = window.subarray(previousNewline + 1, windowBytes - 1); + // A blank final line is benign (the loader skips it) and appends stay safe. + return lastLine.length > 0 && !parsesAsJson(lastLine); + } catch { + return true; + } finally { + closeSync(descriptor); + } +} + +function repairJsonlDamage(filePath: string): void { + const targetPath = realpathIfPresentSync(filePath); + if (!tailLooksDamaged(targetPath)) return; + let buffer: Buffer; + let snapshot: { size: number; mtimeMs: number }; + try { + buffer = readFileSync(targetPath); + const measured = statSync(targetPath); + snapshot = { size: measured.size, mtimeMs: measured.mtimeMs }; + } catch { + return; + } + if (buffer.length === 0 || snapshot.size !== buffer.length) return; + const keptLines: Buffer[] = []; + let recoveredNulLines = 0; + let droppedLines = 0; + let repairedTail = false; + let dirty = false; + let start = 0; + while (start < buffer.length) { + let end = buffer.indexOf(0x0a, start); + const terminated = end !== -1; + if (!terminated) end = buffer.length; + let lineStart = start; + while (lineStart < end && buffer[lineStart] === 0) lineStart++; + const line = buffer.subarray(lineStart, end); + if (lineStart > start) { + dirty = true; + if (line.length > 0 && parsesAsJson(line)) { + keptLines.push(line); + recoveredNulLines++; + } else { + droppedLines++; + } + } else if (!terminated) { + // An unterminated tail merges with the next append: re-terminate or truncate. + dirty = true; + if (line.length > 0 && parsesAsJson(line)) { + keptLines.push(line); + repairedTail = true; + } else { + droppedLines++; + } + } else if (end + 1 >= buffer.length && line.length > 0 && !parsesAsJson(line)) { + dirty = true; + droppedLines++; + } else { + keptLines.push(line); + } + start = end + 1; + } + if (!dirty) return; + const metadata = statMetadataIfPresent(targetPath); + const content = keptLines.length > 0 ? `${keptLines.map((kept) => kept.toString("utf8")).join("\n")}\n` : ""; + try { + writeFileAtomicSync(targetPath, content, { + ...(metadata === undefined ? {} : { mode: metadata.mode }), + beforeRename: (tempPath) => { + if (metadata !== undefined) chownSync(tempPath, metadata.uid, metadata.gid); + // A concurrent appender wins; skipping the repair is safe (next open retries). + const current = statSync(targetPath); + if (current.size !== snapshot.size || current.mtimeMs !== snapshot.mtimeMs) { + throw new RepairSupersededError(); + } + }, + }); + } catch (error) { + if (error instanceof RepairSupersededError) return; + throw error; + } + console.error( + `Repaired crash damage in ${targetPath}: recovered ${recoveredNulLines} zero-filled line(s), dropped ${droppedLines} unrecoverable line(s)${repairedTail ? ", restored the trailing newline" : ""}`, + ); +} + +class RepairSupersededError extends Error {} + +function parsesAsJson(line: Buffer): boolean { + try { + JSON.parse(line.toString("utf8")); + return true; + } catch { + return false; + } +} + function finalizeLoadedEntries(entries: FileEntry[]): FileEntry[] { if (entries.length === 0) return entries; const header = entries[0]; @@ -1180,6 +1290,7 @@ export class SessionManager { setSessionFile(sessionFile: string, preloadedEntries?: FileEntry[]): void { this.sessionFile = resolve(sessionFile); if (existsSync(this.sessionFile)) { + if (this.persist && preloadedEntries === undefined) repairJsonlDamage(this.sessionFile); this.fileEntries = preloadedEntries ?? loadEntriesFromFile(this.sessionFile); // If file was empty or corrupted (no valid header), truncate and start fresh @@ -1294,21 +1405,16 @@ export class SessionManager { private _rewriteFile(): void { if (!this.persist || !this.sessionFile) return; const content = `${this.fileEntries.map((e) => JSON.stringify(e)).join("\n")}\n`; - const targetPath = realpathIfPresent(this.sessionFile); + const targetPath = realpathIfPresentSync(this.sessionFile); const directory = dirname(targetPath); mkdirSync(directory, { recursive: true }); - const tempPath = join(directory, `.${basename(targetPath)}.${process.pid}.${randomUUID()}.tmp`); - try { - const metadata = statMetadataIfPresent(targetPath); - writeFileSync(tempPath, content, metadata === undefined ? undefined : { mode: metadata.mode }); - if (metadata !== undefined) { - chownSync(tempPath, metadata.uid, metadata.gid); - chmodSync(tempPath, metadata.mode); - } - renameSync(tempPath, targetPath); - } finally { - rmSync(tempPath, { force: true }); - } + const metadata = statMetadataIfPresent(targetPath); + writeFileAtomicSync(targetPath, content, { + ...(metadata === undefined ? {} : { mode: metadata.mode }), + beforeRename: (tempPath) => { + if (metadata !== undefined) chownSync(tempPath, metadata.uid, metadata.gid); + }, + }); this._notifyPersistListeners(); } @@ -2032,6 +2138,7 @@ export class SessionManager { if (!existsSync(path)) { return SessionManager.open(path, sessionDir, cwdOverride); } + repairJsonlDamage(path); const entries = await loadEntriesFromFileAsync(path); if (entries.length === 0) { return SessionManager.open(path, sessionDir, cwdOverride); diff --git a/packages/coding-agent/src/core/settings-manager.ts b/packages/coding-agent/src/core/settings-manager.ts index e2b7ee8e77..7abf0f563a 100644 --- a/packages/coding-agent/src/core/settings-manager.ts +++ b/packages/coding-agent/src/core/settings-manager.ts @@ -1,9 +1,10 @@ import type { ServiceTier, Transport } from "@earendil-works/pi-ai"; -import { existsSync, mkdirSync, readFileSync, renameSync, unlinkSync, writeFileSync } from "fs"; +import { existsSync, mkdirSync, readFileSync } from "fs"; import { homedir } from "os"; import { dirname, join } from "path"; import lockfile from "proper-lockfile"; import { CONFIG_DIR_NAME, getAgentDir } from "../config.js"; +import { writeFileAtomicSync } from "../utils/atomic-file.js"; const RECENT_MODELS_LIMIT = 20; export const DEFAULT_IDLE_EVICTION_MINUTES = 90; @@ -282,20 +283,20 @@ export class FileSettingsStorage implements SettingsStorage { release = this.acquireLockSyncWithRetry(path); } const current = fileExists ? readFileSync(path, "utf-8") : undefined; - const next = fn(current); + let next = fn(current); if (next !== undefined) { if (!existsSync(dir)) { mkdirSync(dir, { recursive: true }); } if (!release) { release = this.acquireLockSyncWithRetry(path); + // The first-write read ran unlocked; a racing first writer may have landed since. + if (existsSync(path)) { + next = fn(readFileSync(path, "utf-8")); + } } - const temporaryPath = `${path}.${process.pid}.${Date.now()}.tmp`; - try { - writeFileSync(temporaryPath, next, { encoding: "utf-8", mode: 0o600 }); - renameSync(temporaryPath, path); - } finally { - if (existsSync(temporaryPath)) unlinkSync(temporaryPath); + if (next !== undefined) { + writeFileAtomicSync(path, next, { mode: 0o600 }); } } } finally { diff --git a/packages/coding-agent/src/core/telemetry.ts b/packages/coding-agent/src/core/telemetry.ts index bb32bce5fd..15dbcb5fd1 100644 --- a/packages/coding-agent/src/core/telemetry.ts +++ b/packages/coding-agent/src/core/telemetry.ts @@ -1,9 +1,10 @@ import { randomUUID } from "node:crypto"; -import { lstatSync, mkdirSync, readFileSync, renameSync, unlinkSync, writeFileSync } from "node:fs"; +import { lstatSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { arch, platform } from "node:os"; import { join } from "node:path"; import type { AssistantMessage, Usage } from "@earendil-works/pi-ai"; import { detectInstallMethod, VERSION } from "../config.js"; +import { writeFileAtomicSync } from "../utils/atomic-file.js"; import type { AgentSession, AgentSessionEvent } from "./agent-session.js"; import type { AgentExecutionMode } from "./agent-session-config.js"; import type { AuthCredential, AuthStatus } from "./auth-storage.js"; @@ -234,21 +235,7 @@ function readInstallationId(path: string): string | undefined { } function writeTelemetryStateAtomically(path: string, state: TelemetryState): void { - const temporaryPath = `${path}.${process.pid}.${randomUUID()}.tmp`; - try { - writeFileSync(temporaryPath, JSON.stringify(state, null, 2), { - encoding: "utf8", - flag: "wx", - mode: 0o600, - }); - renameSync(temporaryPath, path); - } finally { - try { - unlinkSync(temporaryPath); - } catch { - // The rename succeeded or the temporary file was never created. - } - } + writeFileAtomicSync(path, JSON.stringify(state, null, 2), { mode: 0o600 }); } export function getOrCreateTelemetryInstallationId(agentDir: string, randomId: () => string = randomUUID): string { diff --git a/packages/coding-agent/src/migrations.ts b/packages/coding-agent/src/migrations.ts index 0085d7b48f..5e9f45d915 100644 --- a/packages/coding-agent/src/migrations.ts +++ b/packages/coding-agent/src/migrations.ts @@ -18,6 +18,7 @@ import { import { basename, dirname, join } from "path"; import { CONFIG_DIR_NAME, getAgentDir, getBinDir, getSessionsDir } from "./config.js"; import { migrateKeybindingsConfig } from "./core/keybindings.js"; +import { realpathIfPresentSync, writeFileAtomicSync } from "./utils/atomic-file.js"; import { readFirstLineSync } from "./utils/file-lines.js"; const MIGRATION_GUIDE_URL = @@ -42,7 +43,7 @@ export function migrateAuthToAuthJson(): string[] { const migrated: Record = {}; const providers: string[] = []; - // Migrate oauth.json + let oauthReadable = false; if (existsSync(oauthPath)) { try { const oauth = JSON.parse(readFileSync(oauthPath, "utf-8")); @@ -50,17 +51,18 @@ export function migrateAuthToAuthJson(): string[] { migrated[provider] = { type: "oauth", ...(cred as object) }; providers.push(provider); } - renameSync(oauthPath, `${oauthPath}.migrated`); + oauthReadable = true; } catch { // Skip on error } } - // Migrate settings.json apiKeys + let settingsWithoutApiKeys: string | undefined; + let settingsMode: number | undefined; if (existsSync(settingsPath)) { try { - const content = readFileSync(settingsPath, "utf-8"); - const settings = JSON.parse(content); + settingsMode = statSync(settingsPath).mode & 0o777; + const settings = JSON.parse(readFileSync(settingsPath, "utf-8")); if (settings.apiKeys && typeof settings.apiKeys === "object") { for (const [provider, key] of Object.entries(settings.apiKeys)) { if (!migrated[provider] && typeof key === "string") { @@ -69,16 +71,40 @@ export function migrateAuthToAuthJson(): string[] { } } delete settings.apiKeys; - writeFileSync(settingsPath, JSON.stringify(settings, null, 2)); + settingsWithoutApiKeys = JSON.stringify(settings, null, 2); } } catch { // Skip on error } } + // The destination must be durable before any source is destroyed. if (Object.keys(migrated).length > 0) { mkdirSync(dirname(authPath), { recursive: true }); - writeFileSync(authPath, JSON.stringify(migrated, null, 2), { mode: 0o600 }); + writeFileAtomicSync(realpathIfPresentSync(authPath), JSON.stringify(migrated, null, 2), { + mode: 0o600, + fsync: true, + fsyncDir: true, + }); + } + // Source cleanup is best-effort: with auth.json durable, leftovers are inert. + try { + if (oauthReadable) { + renameSync(oauthPath, `${oauthPath}.migrated`); + } + } catch { + // Skip on error + } + try { + if (settingsWithoutApiKeys !== undefined) { + writeFileAtomicSync( + realpathIfPresentSync(settingsPath), + settingsWithoutApiKeys, + settingsMode === undefined ? {} : { mode: settingsMode }, + ); + } + } catch { + // Skip on error } return providers; diff --git a/packages/coding-agent/src/modes/daemon/command-recovery-journal.ts b/packages/coding-agent/src/modes/daemon/command-recovery-journal.ts index 05fcec0644..d78a8b7dfc 100644 --- a/packages/coding-agent/src/modes/daemon/command-recovery-journal.ts +++ b/packages/coding-agent/src/modes/daemon/command-recovery-journal.ts @@ -1,5 +1,6 @@ -import { chmodSync, closeSync, fsyncSync, mkdirSync, openSync, readFileSync, renameSync, writeSync } from "node:fs"; +import { chmodSync, closeSync, fsyncSync, mkdirSync, openSync, readFileSync, writeSync } from "node:fs"; import { dirname } from "node:path"; +import { writeFileAtomicSync } from "../../utils/atomic-file.js"; import type { DaemonClientId, DaemonCommandId, DaemonResponse } from "./daemon-protocol.js"; interface ReceivedRecord { @@ -184,7 +185,6 @@ export class CommandRecoveryJournal { } private compact(): void { - const tempPath = `${this.path}.${process.pid}.tmp`; const records: JournalRecord[] = []; for (const [key, entry] of this.entries) { records.push(entry.received); @@ -198,20 +198,11 @@ export class CommandRecoveryJournal { }); } } - const descriptor = openSync(tempPath, "w", 0o600); - try { - writeSync(descriptor, `${records.map((record) => JSON.stringify(record)).join("\n")}\n`); - fsyncSync(descriptor); - } finally { - closeSync(descriptor); - } - renameSync(tempPath, this.path); - const directoryDescriptor = openSync(dirname(this.path), "r"); - try { - fsyncSync(directoryDescriptor); - } finally { - closeSync(directoryDescriptor); - } + writeFileAtomicSync(this.path, `${records.map((record) => JSON.stringify(record)).join("\n")}\n`, { + mode: 0o600, + fsync: true, + fsyncDir: true, + }); this.recordCount = records.length; } } diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index a052ee863f..33a7189f52 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -8,7 +8,7 @@ import { spawn } from "node:child_process"; import { createHash, randomUUID, timingSafeEqual } from "node:crypto"; -import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; import { stat } from "node:fs/promises"; import { createConnection, createServer, type Server, type Socket } from "node:net"; import { basename, dirname, isAbsolute, join, resolve } from "node:path"; @@ -107,6 +107,7 @@ import { import { resolveSessionPath } from "../../core/session-resolver.js"; import type { SessionStats } from "../../core/session-stats.js"; import { type SideQuestionRun, startSideQuestion } from "../../core/side-question.js"; +import { tryAcquireDirLock } from "../../utils/dir-lock.js"; import { killTrackedDetachedChildren } from "../../utils/shell.js"; import { createAgentConnectionCommands, @@ -845,41 +846,13 @@ export class AgentDaemon { let ownsLock = false; try { for (let attempt = 0; attempt < 3 && !ownsLock; attempt++) { - const token = randomUUID(); - const candidateDirectory = `${lockDirectory}.candidate-${process.pid}-${token}`; - mkdirSync(candidateDirectory, { mode: 0o700 }); - writeFileSync(join(candidateDirectory, "pid"), `${process.pid}\n`, { - mode: 0o600, - }); - try { - renameSync(candidateDirectory, lockDirectory); - ownsLock = true; - break; - } catch (error) { - rmSync(candidateDirectory, { recursive: true, force: true }); - const code = (error as NodeJS.ErrnoException).code; - if (code !== "EEXIST" && code !== "ENOTEMPTY") { - throw error; - } - let ownerPid: number | undefined; - try { - ownerPid = Number(readFileSync(join(lockDirectory, "pid"), "utf8").trim()); - } catch { - // An invalid owner is reclaimed atomically below. - } - if (ownerPid && this.isProcessAlive(ownerPid)) { - return; - } - const staleDirectory = `${lockDirectory}.stale-${process.pid}-${token}`; - try { - renameSync(lockDirectory, staleDirectory); - rmSync(staleDirectory, { recursive: true, force: true }); - } catch (reclaimError) { - if ((reclaimError as NodeJS.ErrnoException).code !== "ENOENT") { - throw reclaimError; - } - } + const result = await tryAcquireDirLock(lockDirectory, (ownerPid) => + ownerPid !== undefined ? this.isProcessAlive(ownerPid) : false, + ); + if (result === "held") { + return; } + ownsLock = result === "acquired"; } if (!ownsLock) { return; diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts index 599fcd75db..47d9736c9c 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts @@ -1,19 +1,10 @@ import { createHash, randomUUID } from "node:crypto"; -import { - existsSync, - mkdirSync, - readdirSync, - readFileSync, - realpathSync, - renameSync, - rmSync, - statSync, - writeFileSync, -} from "node:fs"; +import { existsSync, mkdirSync, readdirSync, readFileSync, realpathSync, renameSync, rmSync, statSync } from "node:fs"; import { homedir } from "node:os"; import { basename, dirname, join, resolve } from "node:path"; import lockfile from "proper-lockfile"; import { getProcessStartId } from "../../core/session-lease.js"; +import { writeFileAtomicSync } from "../../utils/atomic-file.js"; import { defaultDaemonSocketDir, normalizeSocketPath } from "./daemon-socket.js"; const DAEMON_SUPERVISOR_REGISTRY_DIR_ENV = "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_REGISTRY_DIR"; @@ -933,14 +924,7 @@ function readShutdownAdmission(path: string): DaemonShutdownAdmissionRecord | un } function writeJsonAtomically(path: string, value: unknown): void { - const tempPath = `${path}.${process.pid}.${randomUUID()}.tmp`; - try { - writeFileSync(tempPath, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 }); - renameSync(tempPath, path); - } catch (error) { - rmSync(tempPath, { force: true }); - throw error; - } + writeFileAtomicSync(path, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 }); } function startupFencePath(directory: string, socketPath: string): string { diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 55ade4be1e..b6cdccba07 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -1,15 +1,6 @@ import { type ChildProcess, spawn } from "node:child_process"; import { createHash, randomBytes, randomUUID } from "node:crypto"; -import { - chmodSync, - existsSync, - mkdirSync, - readdirSync, - readFileSync, - renameSync, - rmSync, - writeFileSync, -} from "node:fs"; +import { chmodSync, existsSync, mkdirSync, readdirSync, readFileSync, rmSync } from "node:fs"; import { createServer, type Server, type Socket } from "node:net"; import { basename, dirname, join, resolve } from "node:path"; import { Writable } from "node:stream"; @@ -61,6 +52,7 @@ import { canonicalSessionPath, getProcessStartId, SessionAlreadyActiveError } fr import { getSessionArtifactPathForFile, readSessionInfo, type SessionInfo } from "../../core/session-manager.js"; import { looksLikeSessionPath } from "../../core/session-resolver.js"; import { SettingsManager } from "../../core/settings-manager.js"; +import { writeFileAtomicSync } from "../../utils/atomic-file.js"; import { isProcessAlive, processIdExists, signalProcessGroupOrProcess } from "../../utils/child-process.js"; import type { AgentConnectionHeartbeat } from "../agent-connection/types.js"; import { attachJsonlLineReader, serializeJsonLine } from "../rpc/jsonl.js"; @@ -1393,10 +1385,7 @@ export class DaemonSupervisor { socketPath: this.socketPath, defaultSessionConfig: durableAgentSessionRuntimeConfig(this.defaultSessionConfig), }; - const tempPath = `${this.supervisorConfigPath}.${process.pid}.tmp`; - writeFileSync(tempPath, `${JSON.stringify(persisted, null, 2)}\n`, { mode: 0o600 }); - chmodSync(tempPath, 0o600); - renameSync(tempPath, this.supervisorConfigPath); + writeFileAtomicSync(this.supervisorConfigPath, `${JSON.stringify(persisted, null, 2)}\n`, { mode: 0o600 }); } private hasPersistedWorkerDescriptors(): boolean { @@ -1408,10 +1397,7 @@ export class DaemonSupervisor { private persistWorker(worker: ResidentWorker): void { worker.descriptor.updatedAt = new Date().toISOString(); const persisted = durableDaemonWorkerDescriptor(worker.descriptor); - const tempPath = `${worker.descriptorPath}.${process.pid}.tmp`; - writeFileSync(tempPath, `${JSON.stringify(persisted, null, 2)}\n`, { mode: 0o600 }); - chmodSync(tempPath, 0o600); - renameSync(tempPath, worker.descriptorPath); + writeFileAtomicSync(worker.descriptorPath, `${JSON.stringify(persisted, null, 2)}\n`, { mode: 0o600 }); } private deleteWorkerDescriptor(worker: { descriptorPath: string; descriptor: DaemonWorkerDescriptor }): void { @@ -6212,14 +6198,15 @@ export class DaemonSupervisor { } const path = getDaemonUpdateRestartManifestPath(this.socketPath, agentDir); mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); - const tempPath = `${path}.${process.pid}.tmp`; - writeFileSync(tempPath, `${JSON.stringify(manifest)}\n`, { mode: 0o600 }); - chmodSync(tempPath, 0o600); - const validated = JSON.parse(readFileSync(tempPath, "utf8")) as DaemonUpdateRestartManifest; - if (!Array.isArray(validated.sessions) || validated.sessions.length !== manifest.sessions.length) { - throw new Error("Could not validate aggregate update manifest"); - } - renameSync(tempPath, path); + writeFileAtomicSync(path, `${JSON.stringify(manifest)}\n`, { + mode: 0o600, + beforeRename: (tempPath) => { + const validated = JSON.parse(readFileSync(tempPath, "utf8")) as DaemonUpdateRestartManifest; + if (!Array.isArray(validated.sessions) || validated.sessions.length !== manifest.sessions.length) { + throw new Error("Could not validate aggregate update manifest"); + } + }, + }); } /** diff --git a/packages/coding-agent/src/modes/daemon/rlm-subagent-display.ts b/packages/coding-agent/src/modes/daemon/rlm-subagent-display.ts index 7f03cb5c87..9aec66783e 100644 --- a/packages/coding-agent/src/modes/daemon/rlm-subagent-display.ts +++ b/packages/coding-agent/src/modes/daemon/rlm-subagent-display.ts @@ -1,6 +1,7 @@ -import { closeSync, fsyncSync, mkdirSync, openSync, renameSync, rmSync, writeSync } from "node:fs"; +import { mkdirSync } from "node:fs"; import { readFile } from "node:fs/promises"; import { join } from "node:path"; +import { writeFileAtomicSync } from "../../utils/atomic-file.js"; /** * Per-child RLM subagent hydration/display metadata. @@ -54,21 +55,7 @@ function isRlmSubagentDisplayEntry(value: unknown): value is RlmSubagentDisplayE export function writeRlmSubagentDisplayEntry(entry: RlmSubagentDisplayEntry): void { const path = rlmSubagentDisplayPath(entry.sessionDir); mkdirSync(entry.sessionDir, { recursive: true }); - const tempPath = `${path}.tmp-${process.pid}-${Date.now()}`; - const handle = openSync(tempPath, "wx", 0o600); - try { - try { - writeSync(handle, `${JSON.stringify(entry)}\n`); - fsyncSync(handle); - } finally { - closeSync(handle); - } - renameSync(tempPath, path); - } catch (error) { - // A failed write, fsync, or rename must not leak the temp file. - rmSync(tempPath, { force: true }); - throw error; - } + writeFileAtomicSync(path, `${JSON.stringify(entry)}\n`, { mode: 0o600, fsync: true }); } export async function readRlmSubagentDisplayEntry(sessionDir: string): Promise { diff --git a/packages/coding-agent/src/utils/atomic-file.ts b/packages/coding-agent/src/utils/atomic-file.ts new file mode 100644 index 0000000000..549c80eac2 --- /dev/null +++ b/packages/coding-agent/src/utils/atomic-file.ts @@ -0,0 +1,119 @@ +import { randomUUID } from "node:crypto"; +import { + chmodSync, + closeSync, + fsyncSync, + openSync, + readlinkSync, + realpathSync, + renameSync, + rmSync, + writeSync, +} from "node:fs"; +import { dirname, resolve } from "node:path"; + +const WIN32_RENAME_ATTEMPTS = 5; + +function sleepSync(ms: number): void { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); +} + +// Windows raises transient EPERM/EACCES when the destination is held open (antivirus, indexer). +function renameOntoSync(from: string, to: string): void { + for (let attempt = 1; ; attempt++) { + try { + renameSync(from, to); + return; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if ( + process.platform !== "win32" || + (code !== "EPERM" && code !== "EACCES") || + attempt >= WIN32_RENAME_ATTEMPTS + ) { + throw error; + } + sleepSync(10 * attempt); + } + } +} + +export interface WriteFileAtomicOptions { + mode?: number; + /** fsync the temp file before the rename. */ + fsync?: boolean; + /** Best-effort directory fsync after the rename. */ + fsyncDir?: boolean; + /** Runs on the written temp file before it replaces the destination (validation, ownership). */ + beforeRename?: (tempPath: string) => void; +} + +/** Durable-write owner: temp file beside the destination, then an atomic rename. */ +export function writeFileAtomicSync(path: string, data: string, options: WriteFileAtomicOptions = {}): void { + const tempPath = `${path}.${process.pid}.${randomUUID()}.tmp`; + try { + const descriptor = options.mode === undefined ? openSync(tempPath, "wx") : openSync(tempPath, "wx", options.mode); + try { + // writeSync may return a short count without throwing; a partial temp must never be renamed in. + const bytes = Buffer.from(data, "utf8"); + let offset = 0; + while (offset < bytes.length) { + const written = writeSync(descriptor, bytes, offset, bytes.length - offset); + if (written <= 0) throw new Error(`Short write persisting ${path}`); + offset += written; + } + if (options.fsync) fsyncSync(descriptor); + } finally { + closeSync(descriptor); + } + // openSync's mode is masked by the umask; enforce the requested bits exactly. + if (options.mode !== undefined) chmodSync(tempPath, options.mode); + options.beforeRename?.(tempPath); + renameOntoSync(tempPath, path); + } finally { + rmSync(tempPath, { force: true }); + } + if (options.fsyncDir) { + try { + const directoryDescriptor = openSync(dirname(path), "r"); + try { + fsyncSync(directoryDescriptor); + } finally { + closeSync(directoryDescriptor); + } + } catch { + // Unavailable on some platforms; the atomic rename still protects readers. + } + } +} + +/** Resolve symlink aliases so a replace lands on the real file (in-place-write parity). */ +export function realpathIfPresentSync(path: string): string { + try { + return realpathSync(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + throw error; + } + } + // ENOENT also means a DANGLING symlink chain: follow it like in-place writes did. + let current = path; + for (let hop = 0; hop < 32; hop++) { + let target: string; + try { + target = readlinkSync(current); + } catch { + return current; + } + // A relative target resolves against the link's PHYSICAL parent directory. + let parent = dirname(current); + try { + parent = realpathSync(parent); + } catch { + // Fall back to the alias parent. + } + current = resolve(parent, target); + } + // A loud failure beats silently replacing an intermediate link (or looping on a cycle). + throw new Error(`Too many symlink hops resolving ${path}`); +} diff --git a/packages/coding-agent/src/utils/dir-lock.ts b/packages/coding-agent/src/utils/dir-lock.ts new file mode 100644 index 0000000000..9311cb44bd --- /dev/null +++ b/packages/coding-agent/src/utils/dir-lock.ts @@ -0,0 +1,207 @@ +import { randomUUID } from "node:crypto"; +import { + closeSync, + fstatSync, + linkSync, + openSync, + readdirSync, + readFileSync, + renameSync, + rmSync, + statSync, + writeFileSync, +} from "node:fs"; +import { basename, dirname, join } from "node:path"; + +export type DirLockAttempt = "acquired" | "held" | "reclaimed"; + +/** + * link(2)-published lock file: born with its owner content, EEXIST the only + * collision signal; stale locks are renamed aside, verified, then deleted or + * restored. A directory at the lock path is a legacy lock from the old protocol. + */ +const CANDIDATE_SWEEP_AGE_MS = 60 * 60 * 1000; + +// The candidate prefix can never match the lock; the age gate spares mid-publish rivals. +function sweepAbandonedCandidates(lockPath: string): void { + try { + const directory = dirname(lockPath); + const prefix = `${basename(lockPath)}.candidate-`; + const cutoff = Date.now() - CANDIDATE_SWEEP_AGE_MS; + for (const name of readdirSync(directory)) { + if (!name.startsWith(prefix)) continue; + try { + const abandoned = join(directory, name); + if (statSync(abandoned).mtimeMs < cutoff) { + rmSync(abandoned, { force: true }); + } + } catch { + // Litter collection only. + } + } + } catch { + // Litter collection only: the next acquire retries. + } +} + +export async function tryAcquireDirLock( + lockPath: string, + ownerAlive: (ownerPid: number | undefined) => Promise | boolean, +): Promise { + sweepAbandonedCandidates(lockPath); + return acquireAttempt(lockPath, ownerAlive, true); +} + +async function acquireAttempt( + lockPath: string, + ownerAlive: (ownerPid: number | undefined) => Promise | boolean, + retryOnSweptCandidate: boolean, +): Promise { + const token = `${process.pid}-${randomUUID()}`; + const tempPath = `${lockPath}.candidate-${token}`; + writeFileSync(tempPath, `${process.pid}\n`, { mode: 0o600 }); + try { + try { + linkSync(tempPath, lockPath); + return "acquired"; + } catch (error) { + // NFS can report failure for a link that landed: nlink 2 means it published. + let candidateSwept = false; + let recheckedNlink: number | undefined; + try { + recheckedNlink = statSync(tempPath).nlink; + } catch (statError) { + // Only a definite ENOENT means the candidate was swept. + if ((statError as NodeJS.ErrnoException).code !== "ENOENT") { + throw error; + } + candidateSwept = true; + } + if (recheckedNlink === 2) { + return "acquired"; + } + if (candidateSwept && retryOnSweptCandidate) { + return acquireAttempt(lockPath, ownerAlive, false); + } + if ((error as NodeJS.ErrnoException).code !== "EEXIST") { + throw error; + } + } + // One immutable dev+ino capture keys every later decision about the judged lock. + let captured: { dev: bigint; ino: bigint; isDir: boolean }; + try { + const measured = statSync(lockPath, { bigint: true }); + captured = { dev: measured.dev, ino: measured.ino, isDir: measured.isDirectory() }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return "reclaimed"; + } + // An unjudgeable lock may be live: fail safe. + return "held"; + } + if (captured.ino === 0n) { + // Some Windows filesystems report no stable file index: identity unavailable. + return "held"; + } + // An open descriptor pins the inode number against Linux's immediate reuse. + let pinned: number | undefined; + try { + try { + pinned = openSync(lockPath, "r"); + const pinnedIdentity = fstatSync(pinned, { bigint: true }); + if (pinnedIdentity.dev !== captured.dev || pinnedIdentity.ino !== captured.ino) { + // The lock changed hands between the capture and the pin: treat as live. + return "held"; + } + } catch { + // Unpinnable (Windows directories): stat identity without the reuse guarantee. + } + return await judgeAndReclaim(lockPath, ownerAlive, captured, token); + } finally { + if (pinned !== undefined) closeSync(pinned); + } + } finally { + try { + rmSync(tempPath, { force: true }); + } catch { + // Cleanup only: a leaked candidate must never mask a settled acquisition. + } + } +} + +async function judgeAndReclaim( + lockPath: string, + ownerAlive: (ownerPid: number | undefined) => Promise | boolean, + captured: { dev: bigint; ino: bigint; isDir: boolean }, + token: string, +): Promise { + { + const judged = readOwnerRaw(lockPath, captured.isDir); + if (judged === "unreadable") { + // A transient read failure may hide a LIVE lock: never judge it stale. + return "held"; + } + if (await ownerAlive(strictPid(judged === "absent" ? undefined : judged))) { + return "held"; + } + const asidePath = `${lockPath}.stale-${token}`; + try { + renameSync(lockPath, asidePath); + } catch (reclaimError) { + // ENOENT: a racing reclaimer moved it first. + if ((reclaimError as NodeJS.ErrnoException).code === "ENOENT") { + return "reclaimed"; + } + // A lost-reply rename: if the lock path is gone, something moved - fall to verify. + if (statIdentity(lockPath) !== undefined) { + throw reclaimError; + } + } + const aside = statIdentity(asidePath); + if (aside !== undefined && aside.dev === captured.dev && aside.ino === captured.ino) { + rmSync(asidePath, { recursive: true, force: true }); + return "reclaimed"; + } + // Not the judged lock: restore, never delete. Known dirs rename back; everything + // else links back (link can never replace a rival). Any failure leaves it aside. + try { + if (aside?.isDir === true) { + renameSync(asidePath, lockPath); + } else { + linkSync(asidePath, lockPath); + rmSync(asidePath, { force: true }); + } + } catch { + // Preserved aside. + } + return "held"; + } +} + +function statIdentity(path: string): { dev: bigint; ino: bigint; isDir: boolean } | undefined { + try { + const measured = statSync(path, { bigint: true }); + return { dev: measured.dev, ino: measured.ino, isDir: measured.isDirectory() }; + } catch { + return undefined; + } +} + +// "absent" is safely stale territory; "unreadable" may be a LIVE lock (transient EPERM/EBUSY). +function readOwnerRaw(path: string, legacyDir: boolean): string | "absent" | "unreadable" { + try { + return readFileSync(legacyDir ? join(path, "pid") : path, "utf8"); + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ENOENT" ? "absent" : "unreadable"; + } +} + +// kill(0)/kill(-n) probe our own process group: only an exact positive integer owns. +function strictPid(raw: string | undefined): number | undefined { + const trimmed = raw?.trim(); + if (trimmed === undefined || !/^\d+$/.test(trimmed)) { + return undefined; + } + const parsed = Number.parseInt(trimmed, 10); + return Number.isInteger(parsed) && parsed > 0 ? parsed : undefined; +} diff --git a/packages/coding-agent/test/atomic-persistence.test.ts b/packages/coding-agent/test/atomic-persistence.test.ts new file mode 100644 index 0000000000..e8630656bf --- /dev/null +++ b/packages/coding-agent/test/atomic-persistence.test.ts @@ -0,0 +1,319 @@ +import { + mkdirSync, + mkdtempSync, + readdirSync, + readFileSync, + rmSync, + utimesSync, + writeFileSync, + type writeSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +type WriteSync = typeof writeSync; +const shortWrites = vi.hoisted(() => ({ remaining: 0 })); +const rmFault = vi.hoisted(() => ({ error: undefined as Error | undefined })); +const linkSweep = vi.hoisted(() => ({ remaining: 0 })); +const asideStatFault = vi.hoisted(() => ({ remaining: 0, plantRivalAt: undefined as string | undefined })); +const renamePerformThenThrow = vi.hoisted(() => ({ remaining: 0 })); +vi.mock("node:fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + renameSync: ((from: Parameters[0], to: Parameters[1]) => { + if (renamePerformThenThrow.remaining > 0 && String(to).includes(".stale-")) { + renamePerformThenThrow.remaining--; + actual.renameSync(from, to); + throw Object.assign(new Error("EIO: reply lost"), { code: "EIO" }); + } + return actual.renameSync(from, to); + }) as typeof actual.renameSync, + statSync: ((path: Parameters[0], options?: never) => { + if (asideStatFault.remaining > 0 && String(path).includes(".stale-")) { + asideStatFault.remaining--; + if (asideStatFault.plantRivalAt !== undefined) { + actual.writeFileSync(asideStatFault.plantRivalAt, "31337\n"); + } + throw Object.assign(new Error("EPERM: probe blocked"), { code: "EPERM" }); + } + return actual.statSync(path, options); + }) as typeof actual.statSync, + linkSync: ((existing: Parameters[0], created: Parameters[1]) => { + if (linkSweep.remaining > 0) { + linkSweep.remaining--; + // A rival's sweep claims the candidate between write and publish. + actual.rmSync(existing, { force: true }); + } + return actual.linkSync(existing, created); + }) as typeof actual.linkSync, + rmSync: ((path: Parameters[0], options?: Parameters[1]) => { + if (rmFault.error && String(path).includes(".candidate-")) throw rmFault.error; + return actual.rmSync(path, options); + }) as typeof actual.rmSync, + writeSync: ((fd: number, data: NodeJS.ArrayBufferView | string, offset?: number, length?: number) => { + if (shortWrites.remaining > 0 && typeof data === "string" && data.length > 1) { + shortWrites.remaining--; + return (actual.writeSync as WriteSync)(fd, data.slice(0, 1) as never); + } + if (shortWrites.remaining > 0 && typeof length === "number" && length > 1) { + shortWrites.remaining--; + return (actual.writeSync as WriteSync)(fd, data as NodeJS.ArrayBufferView, offset, 1); + } + return (actual.writeSync as WriteSync)(fd, data as never, offset as never, length as never); + }) as WriteSync, + }; +}); + +import { writeFileAtomicSync } from "../src/utils/atomic-file.js"; +import { tryAcquireDirLock } from "../src/utils/dir-lock.js"; + +const tempDirs: string[] = []; + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "prime-atomic-persistence-")); + tempDirs.push(dir); + return dir; +} + +describe("writeFileAtomicSync", () => { + it("completes short kernel writes and preserves the destination when a write fails", () => { + const dir = createTempDir(); + const path = join(dir, "state.json"); + shortWrites.remaining = 3; + try { + writeFileAtomicSync(path, JSON.stringify({ key: "value".repeat(10) })); + } finally { + shortWrites.remaining = 0; + } + expect(JSON.parse(readFileSync(path, "utf8"))).toEqual({ key: "value".repeat(10) }); + + expect(() => + writeFileAtomicSync(path, "next", { + beforeRename: () => { + throw new Error("validation failed"); + }, + }), + ).toThrow("validation failed"); + expect(JSON.parse(readFileSync(path, "utf8"))).toEqual({ key: "value".repeat(10) }); + expect(readdirSync(dir).filter((name) => name.endsWith(".tmp"))).toEqual([]); + }); +}); + +function seedDirLock(dir: string, name: string, pid: string): string { + const lockDir = join(dir, name); + mkdirSync(lockDir); + writeFileSync(join(lockDir, "pid"), pid); + return lockDir; +} + +describe("tryAcquireDirLock", () => { + it("recovers a lost-reply rename: reclaims a stale lock, restores a swapped rival", async () => { + const dir = createTempDir(); + const lockPath = join(dir, "eio.lock"); + writeFileSync(lockPath, "999999999\n"); + renamePerformThenThrow.remaining = 1; + try { + expect(await tryAcquireDirLock(lockPath, () => false)).toBe("reclaimed"); + } finally { + renamePerformThenThrow.remaining = 0; + } + expect(readdirSync(dir)).toEqual([]); + expect(await tryAcquireDirLock(lockPath, () => false)).toBe("acquired"); + + rmSync(lockPath, { force: true }); + writeFileSync(lockPath, "999999999\n"); + renamePerformThenThrow.remaining = 1; + try { + const swapped = await tryAcquireDirLock(lockPath, () => { + // A rival replaces the judged-stale lock before the rename fires. + rmSync(lockPath, { force: true }); + writeFileSync(lockPath, "777777\n"); + return false; + }); + expect(swapped).toBe("held"); + } finally { + renamePerformThenThrow.remaining = 0; + } + expect(readFileSync(lockPath, "utf8").trim()).toBe("777777"); + }); + + it("parks a moved lock whose shape cannot be probed instead of deleting or clobbering", async () => { + const dir = createTempDir(); + const lockPath = join(dir, "shape.lock"); + writeFileSync(lockPath, "999999999\n"); + // A rival publishes at the path while the aside probe is failing: only a + // link-back (which cannot replace it) is a safe restore attempt. + asideStatFault.remaining = 1; + asideStatFault.plantRivalAt = lockPath; + + try { + expect(await tryAcquireDirLock(lockPath, () => false)).toBe("held"); + } finally { + asideStatFault.remaining = 0; + asideStatFault.plantRivalAt = undefined; + } + expect(readFileSync(lockPath, "utf8").trim()).toBe("31337"); + const parked = readdirSync(dir) + .filter((name) => name.includes(".stale-")) + .map((name) => readFileSync(join(dir, name), "utf8").trim()); + expect(parked).toEqual(["999999999"]); + }); + + it("retries with a fresh candidate when a rival's sweep claims the first mid-publish", async () => { + const dir = createTempDir(); + const lockPath = join(dir, "suspended.lock"); + linkSweep.remaining = 1; + + try { + expect(await tryAcquireDirLock(lockPath, () => false)).toBe("acquired"); + } finally { + linkSweep.remaining = 0; + } + expect(readFileSync(lockPath, "utf8").trim()).toBe(String(process.pid)); + }); + + it("sweeps abandoned candidates on acquire while sparing fresh ones and the lock", async () => { + const dir = createTempDir(); + const lockPath = join(dir, "swept.lock"); + const abandoned = join(dir, "swept.lock.candidate-1234-dead"); + const fresh = join(dir, "swept.lock.candidate-5678-mid-publish"); + writeFileSync(abandoned, "1234\n"); + const twoHoursAgo = new Date(Date.now() - 2 * 60 * 60 * 1000); + utimesSync(abandoned, twoHoursAgo, twoHoursAgo); + writeFileSync(fresh, "5678\n"); + + expect(await tryAcquireDirLock(lockPath, () => false)).toBe("acquired"); + + const names = readdirSync(dir).sort(); + expect(names).not.toContain("swept.lock.candidate-1234-dead"); + expect(names).toContain("swept.lock.candidate-5678-mid-publish"); + expect(readFileSync(lockPath, "utf8").trim()).toBe(String(process.pid)); + }); + + it("reports held instead of reclaiming when the owner cannot be read", async () => { + const dir = createTempDir(); + const lockDir = join(dir, "opaque.lock"); + // Legacy lock whose pid entry is a directory: every read fails non-ENOENT. + mkdirSync(join(lockDir, "pid"), { recursive: true }); + + expect(await tryAcquireDirLock(lockDir, () => false)).toBe("held"); + expect(readdirSync(dir)).toEqual(["opaque.lock"]); + }); + + it("keeps a settled acquisition when candidate cleanup fails", async () => { + const dir = createTempDir(); + const lockPath = join(dir, "cleanup.lock"); + rmFault.error = new Error("EBUSY: held by scanner"); + + try { + expect(await tryAcquireDirLock(lockPath, () => false)).toBe("acquired"); + } finally { + rmFault.error = undefined; + } + expect(readFileSync(lockPath, "utf8").trim()).toBe(String(process.pid)); + }); + + it("publishes the lock as a file born with its owner and puts back a swapped file lock", async () => { + const dir = createTempDir(); + const lockPath = join(dir, "file.lock"); + const alive = (ownerPid: number | undefined) => ownerPid === process.pid || ownerPid === 424242; + + expect(await tryAcquireDirLock(lockPath, alive)).toBe("acquired"); + expect(readFileSync(lockPath, "utf8").trim()).toBe(String(process.pid)); + expect(await tryAcquireDirLock(lockPath, alive)).toBe("held"); + + // Stale file lock: dead owner content is reclaimed. + writeFileSync(lockPath, "999999999\n"); + const swapped = await tryAcquireDirLock(lockPath, () => { + // A rival replaces the lock while the staleness judgment runs. + rmSync(lockPath, { force: true }); + writeFileSync(lockPath, "424242\n"); + return false; + }); + expect(swapped).toBe("held"); + expect(readFileSync(lockPath, "utf8").trim()).toBe("424242"); + }); + + // kill(0) probes our own process group; parseInt would trust "123garbage" as 123. + it.each([ + [ + "0\n", + (ownerPid: number | undefined) => (ownerPid === undefined ? false : process.kill(ownerPid, 0) !== undefined), + ], + ["123garbage\n", (ownerPid: number | undefined) => ownerPid === 123], + ])("treats a legacy lock with pid content %j as stale", async (pidContent, alive) => { + const dir = createTempDir(); + const lockDir = join(dir, "hygiene.lock"); + mkdirSync(lockDir); + writeFileSync(join(lockDir, "pid"), pidContent); + + expect(await tryAcquireDirLock(lockDir, alive)).toBe("reclaimed"); + expect(await tryAcquireDirLock(lockDir, alive)).toBe("acquired"); + }); + + it("puts back a lock that changed owners between the staleness judgment and the reclaim", async () => { + const dir = createTempDir(); + const lockDir = seedDirLock(dir, "raced.lock", "2147483647\n"); + + const result = await tryAcquireDirLock(lockDir, () => { + // The stale owner releases and a rival acquires while this judgment runs. + rmSync(lockDir, { recursive: true, force: true }); + mkdirSync(lockDir); + writeFileSync(join(lockDir, "pid"), "424242\n"); + return false; + }); + + expect(result).toBe("held"); + expect(readFileSync(join(lockDir, "pid"), "utf8").trim()).toBe("424242"); + + // Pid reuse: a NEW lock with the SAME pid content is a different inode and + // must be restored, not judged identical and deleted. + const reused = await tryAcquireDirLock(lockDir, () => { + rmSync(lockDir, { recursive: true, force: true }); + mkdirSync(lockDir); + writeFileSync(join(lockDir, "pid"), "424242\n"); + return false; + }); + expect(reused).toBe("held"); + expect(readFileSync(join(lockDir, "pid"), "utf8").trim()).toBe("424242"); + + // Cross-shape swap: a judged FILE lock replaced by a rival's legacy DIR is + // restored at the path by the moved entry's own shape (rename-back). + rmSync(lockDir, { recursive: true, force: true }); + writeFileSync(lockDir, "999999999\n"); + const crossType = await tryAcquireDirLock(lockDir, () => { + rmSync(lockDir, { recursive: true, force: true }); + mkdirSync(lockDir); + writeFileSync(join(lockDir, "pid"), "555555\n"); + return false; + }); + expect(crossType).toBe("held"); + expect(readFileSync(join(lockDir, "pid"), "utf8").trim()).toBe("555555"); + }); + + it("acquires over a stale lock without deleting a lock that changed owners", async () => { + const dir = createTempDir(); + const lockDir = join(dir, "work.lock"); + // A stale lock: dead owner. + mkdirSync(lockDir); + writeFileSync(join(lockDir, "pid"), "2147483647\n"); + + const alive = (ownerPid: number | undefined) => ownerPid === process.pid; + expect(await tryAcquireDirLock(lockDir, alive)).toBe("reclaimed"); + expect(await tryAcquireDirLock(lockDir, alive)).toBe("acquired"); + expect(readFileSync(lockDir, "utf8").trim()).toBe(String(process.pid)); + + // Every rival attempt against the live owner reports "held" and leaves the lock alone. + const rivals = await Promise.all(Array.from({ length: 8 }, () => tryAcquireDirLock(lockDir, alive))); + expect(rivals).toEqual(Array.from({ length: 8 }, () => "held")); + expect(readFileSync(lockDir, "utf8").trim()).toBe(String(process.pid)); + }); +}); diff --git a/packages/coding-agent/test/auth-storage.test.ts b/packages/coding-agent/test/auth-storage.test.ts index 049d254206..c5fe8005af 100644 --- a/packages/coding-agent/test/auth-storage.test.ts +++ b/packages/coding-agent/test/auth-storage.test.ts @@ -1,4 +1,4 @@ -import { existsSync, mkdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { existsSync, lstatSync, mkdirSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { registerOAuthProvider } from "@earendil-works/pi-ai/oauth"; @@ -6,6 +6,23 @@ import lockfile from "proper-lockfile"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { AuthStorage } from "../src/core/auth-storage.js"; +const renameFault = vi.hoisted(() => ({ error: undefined as Error | undefined })); +const absenceIllusion = vi.hoisted(() => ({ paths: new Set() })); +vi.mock("node:fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + renameSync: (from: Parameters[0], to: Parameters[1]) => { + if (renameFault.error && String(to).endsWith("auth.json")) throw renameFault.error; + return actual.renameSync(from, to); + }, + existsSync: (path: Parameters[0]) => { + if (absenceIllusion.paths.has(String(path))) return false; + return actual.existsSync(path); + }, + }; +}); + describe("AuthStorage", () => { let tempDir: string; let authJsonPath: string; @@ -951,6 +968,97 @@ describe("AuthStorage", () => { }); describe("persistence semantics", () => { + test("first-run initialization survives a restrictive umask", () => { + const previousUmask = process.umask(0o700); + try { + authStorage = AuthStorage.create(authJsonPath); + authStorage.set("openai", { type: "api_key", key: "masked-key" }); + } finally { + process.umask(previousUmask); + } + + expect(statSync(authJsonPath).mode & 0o777).toBe(0o600); + const onDisk = JSON.parse(readFileSync(authJsonPath, "utf-8")) as Record; + expect(onDisk.openai.key).toBe("masked-key"); + }); + + test.each([ + [ + "an existing target", + (): { alias: string; target: string } => { + const target = join(tempDir, "real-auth.json"); + writeFileSync(target, "{}"); + rmSync(authJsonPath, { force: true }); + symlinkSync(target, authJsonPath); + return { alias: authJsonPath, target }; + }, + ], + [ + "a dangling absolute target", + (): { alias: string; target: string } => { + const target = join(tempDir, "vault", "auth.json"); + mkdirSync(join(tempDir, "vault"), { recursive: true }); + symlinkSync(target, authJsonPath); + return { alias: authJsonPath, target }; + }, + ], + [ + "a dangling relative target under a symlinked directory", + (): { alias: string; target: string } => { + const realDir = join(tempDir, "real-dir"); + mkdirSync(realDir, { recursive: true }); + const aliasDir = join(tempDir, "alias-dir"); + symlinkSync(realDir, aliasDir); + symlinkSync("./credentials.json", join(aliasDir, "auth.json")); + return { alias: join(aliasDir, "auth.json"), target: join(realDir, "credentials.json") }; + }, + ], + ])("writes through a symlinked auth.json (%s) with the alias intact", (_name, setup) => { + const { alias, target } = setup(); + authStorage = AuthStorage.create(alias); + + authStorage.set("openai", { type: "api_key", key: "through-alias" }); + + expect(lstatSync(alias).isSymbolicLink()).toBe(true); + const real = JSON.parse(readFileSync(target, "utf-8")) as Record; + expect(real.openai.key).toBe("through-alias"); + }); + + test("initialization never replaces credentials another process already saved", () => { + authStorage = AuthStorage.create(authJsonPath); + // A rival process persists credentials between the absence check and the write. + writeAuthJson({ anthropic: { type: "api_key", key: "already-saved" } }); + absenceIllusion.paths.add(authJsonPath); + + try { + const backend = (authStorage as unknown as { storage: { ensureFileExists(): void } }).storage; + backend.ensureFileExists(); + } finally { + absenceIllusion.paths.delete(authJsonPath); + } + + const onDisk = JSON.parse(readFileSync(authJsonPath, "utf-8")) as Record; + expect(onDisk.anthropic.key).toBe("already-saved"); + }); + + test("a write failing at the replace boundary leaves the previous credentials intact", () => { + writeAuthJson({ anthropic: { type: "api_key", key: "old-key" } }); + authStorage = AuthStorage.create(authJsonPath); + renameFault.error = new Error("disk full"); + + try { + authStorage.set("anthropic", { type: "api_key", key: "new-key" }); + } finally { + renameFault.error = undefined; + } + + expect(authStorage.drainErrors().map((error) => String(error))).toEqual([ + expect.stringContaining("disk full"), + ]); + const onDisk = JSON.parse(readFileSync(authJsonPath, "utf-8")) as Record; + expect(onDisk.anthropic.key).toBe("old-key"); + }); + test("set preserves unrelated external edits", () => { writeAuthJson({ anthropic: { type: "api_key", key: "old-anthropic" }, diff --git a/packages/coding-agent/test/migrations.test.ts b/packages/coding-agent/test/migrations.test.ts index 01f39a3f89..ebc5ab2d92 100644 --- a/packages/coding-agent/test/migrations.test.ts +++ b/packages/coding-agent/test/migrations.test.ts @@ -1,9 +1,36 @@ -import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { + chmodSync, + existsSync, + lstatSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + statSync, + symlinkSync, + writeFileSync, +} from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { ENV_AGENT_DIR } from "../src/config.js"; -import { migrateLegacySessionDirsToSessionRoot, migrateSessionsFromAgentRoot } from "../src/migrations.js"; +import { + migrateAuthToAuthJson, + migrateLegacySessionDirsToSessionRoot, + migrateSessionsFromAgentRoot, +} from "../src/migrations.js"; + +const atomicWriteMock = vi.hoisted(() => ({ error: undefined as Error | undefined })); +vi.mock("../src/utils/atomic-file.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + writeFileAtomicSync: (path: string, data: string, options?: object) => { + if (atomicWriteMock.error && path.endsWith("auth.json")) throw atomicWriteMock.error; + return actual.writeFileAtomicSync(path, data, options); + }, + }; +}); describe("session migrations", () => { const tempDirs: string[] = []; @@ -105,3 +132,82 @@ describe("session migrations", () => { expect(existsSync(join(sessionsDir, "session-2.jsonl"))).toBe(false); }); }); + +describe("auth migration ordering", () => { + const tempDirs: string[] = []; + const previousAgentDir = process.env[ENV_AGENT_DIR]; + + afterEach(() => { + vi.restoreAllMocks(); + atomicWriteMock.error = undefined; + if (previousAgentDir === undefined) { + delete process.env[ENV_AGENT_DIR]; + } else { + process.env[ENV_AGENT_DIR] = previousAgentDir; + } + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } + }); + + function makeAgentDir(): string { + const agentDir = mkdtempSync(join(tmpdir(), "prime-agent-auth-migration-")); + tempDirs.push(agentDir); + process.env[ENV_AGENT_DIR] = agentDir; + return agentDir; + } + + it("preserves the settings file's own mode when stripping apiKeys", () => { + const agentDir = makeAgentDir(); + const settingsPath = join(agentDir, "settings.json"); + writeFileSync(settingsPath, JSON.stringify({ theme: "dark", apiKeys: { openai: "sk-key" } })); + chmodSync(settingsPath, 0o600); + + migrateAuthToAuthJson(); + + expect(statSync(settingsPath).mode & 0o777).toBe(0o600); + expect(JSON.parse(readFileSync(settingsPath, "utf-8")).apiKeys).toBeUndefined(); + }); + + it("strips apiKeys through a symlinked settings.json without replacing the alias", () => { + const agentDir = makeAgentDir(); + const realSettings = join(agentDir, "dotfiles-settings.json"); + const settingsPath = join(agentDir, "settings.json"); + writeFileSync(realSettings, JSON.stringify({ theme: "dark", apiKeys: { openai: "sk-key" } })); + symlinkSync(realSettings, settingsPath); + + migrateAuthToAuthJson(); + + expect(lstatSync(settingsPath).isSymbolicLink()).toBe(true); + expect(JSON.parse(readFileSync(realSettings, "utf-8")).apiKeys).toBeUndefined(); + expect(JSON.parse(readFileSync(join(agentDir, "auth.json"), "utf-8")).openai.key).toBe("sk-key"); + }); + + it("migrates credentials through a dangling auth.json symlink to its target", () => { + const agentDir = makeAgentDir(); + writeFileSync(join(agentDir, "oauth.json"), JSON.stringify({ anthropic: { access: "token" } })); + const target = join(agentDir, "vault-auth.json"); + symlinkSync(target, join(agentDir, "auth.json")); + + migrateAuthToAuthJson(); + + expect(lstatSync(join(agentDir, "auth.json")).isSymbolicLink()).toBe(true); + expect(JSON.parse(readFileSync(target, "utf-8")).anthropic.type).toBe("oauth"); + }); + + it("keeps every credential source when the auth.json write fails", () => { + const agentDir = makeAgentDir(); + const oauthPath = join(agentDir, "oauth.json"); + const settingsPath = join(agentDir, "settings.json"); + writeFileSync(oauthPath, JSON.stringify({ anthropic: { access: "token" } })); + writeFileSync(settingsPath, JSON.stringify({ theme: "dark", apiKeys: { openai: "sk-key" } })); + atomicWriteMock.error = new Error("disk full"); + + expect(() => migrateAuthToAuthJson()).toThrow("disk full"); + + // A crash at the destination write must leave both sources recoverable. + expect(existsSync(join(agentDir, "auth.json"))).toBe(false); + expect(existsSync(oauthPath)).toBe(true); + expect(JSON.parse(readFileSync(settingsPath, "utf-8")).apiKeys).toEqual({ openai: "sk-key" }); + }); +}); diff --git a/packages/coding-agent/test/session-lease.test.ts b/packages/coding-agent/test/session-lease.test.ts index 87338b6866..1a2a214199 100644 --- a/packages/coding-agent/test/session-lease.test.ts +++ b/packages/coding-agent/test/session-lease.test.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import { mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { lockSync } from "proper-lockfile"; @@ -116,6 +116,22 @@ describe("session leases", () => { lease?.release(); }); + it("never reclaims a lease whose owner file cannot be read", () => { + const agentDir = createTempDir(); + const sessionPath = canonicalSessionPath(resolve(agentDir, "unreadable.jsonl")); + const key = createHash("sha256").update(sessionPath).digest("hex"); + const lockDirectory = join(agentDir, "session-leases", `${key}.lock`); + // owner.json as a directory: every read fails with a non-ENOENT error, the + // same shape as a transient EPERM/EBUSY on Windows. That may be a LIVE + // lease, so acquisition must fail instead of destroying it. + mkdirSync(join(lockDirectory, "owner.json"), { recursive: true }); + + expect(() => acquireSessionLease(sessionPath, agentDir, enabledEnvironment("intruder"))).toThrow( + "Could not acquire session lease", + ); + expect(existsSync(join(lockDirectory, "owner.json"))).toBe(true); + }); + it("reports guard contention as a coordination failure", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(join(agentDir, "session.jsonl")); diff --git a/packages/coding-agent/test/session-manager-flush.test.ts b/packages/coding-agent/test/session-manager-flush.test.ts index 50ddafd555..1b9df9cd4e 100644 --- a/packages/coding-agent/test/session-manager-flush.test.ts +++ b/packages/coding-agent/test/session-manager-flush.test.ts @@ -13,6 +13,7 @@ import { statSync, symlinkSync, type writeFileSync, + type writeSync as writeSyncFs, } from "node:fs"; import { tmpdir } from "node:os"; import { basename, dirname, join } from "node:path"; @@ -22,27 +23,33 @@ type ChmodSync = typeof chmodSync; type ChownSync = typeof chownSync; type RenameSync = typeof renameSync; type WriteFileSync = typeof writeFileSync; +type WriteSync = typeof writeSyncFs; const fsMocks = vi.hoisted(() => ({ actualWriteFileSync: undefined as WriteFileSync | undefined, + actualWriteSync: undefined as WriteSync | undefined, chmodSync: vi.fn(), chownSync: vi.fn(), renameSync: vi.fn(), writeFileSync: vi.fn(), + writeSync: vi.fn(), })); vi.mock("node:fs", async (importOriginal) => { const actual = await importOriginal(); fsMocks.actualWriteFileSync = actual.writeFileSync; + fsMocks.actualWriteSync = actual.writeSync; fsMocks.chmodSync.mockImplementation(actual.chmodSync); fsMocks.chownSync.mockImplementation(actual.chownSync); fsMocks.renameSync.mockImplementation(actual.renameSync); fsMocks.writeFileSync.mockImplementation(actual.writeFileSync); + fsMocks.writeSync.mockImplementation(actual.writeSync); return { ...actual, chmodSync: fsMocks.chmodSync, chownSync: fsMocks.chownSync, renameSync: fsMocks.renameSync, writeFileSync: fsMocks.writeFileSync, + writeSync: fsMocks.writeSync, }; }); @@ -103,10 +110,10 @@ describe("SessionManager.flushNow", () => { const tempPrefix = `.${basename(file)}.`; mgr.appendMessage({ role: "user", content: "pending", timestamp: Date.now() }); - fsMocks.writeFileSync.mockImplementationOnce((path, data, options) => { - fsMocks.actualWriteFileSync!(path, Buffer.from(String(data)).subarray(0, 12), options); + fsMocks.writeSync.mockImplementationOnce(((fd: number, data: string) => { + fsMocks.actualWriteSync!(fd, Buffer.from(String(data)).subarray(0, 12)); throw new Error("disk full"); - }); + }) as unknown as WriteSync); expect(() => mgr.flushNow()).toThrow("disk full"); expect(readFileSync(file)).toEqual(before); @@ -137,7 +144,7 @@ describe("SessionManager.flushNow", () => { expect(fsMocks.chmodSync).toHaveBeenCalledWith(tempPath, before.mode & 0o777); expect(fsMocks.renameSync).toHaveBeenCalledWith(tempPath, join(dirname(tempPath as string), basename(file))); expect(fsMocks.chownSync.mock.invocationCallOrder[0]!).toBeLessThan( - fsMocks.chmodSync.mock.invocationCallOrder[0]!, + fsMocks.renameSync.mock.invocationCallOrder[0]!, ); expect(fsMocks.chmodSync.mock.invocationCallOrder[0]!).toBeLessThan( fsMocks.renameSync.mock.invocationCallOrder[0]!, diff --git a/packages/coding-agent/test/session-manager/file-operations.test.ts b/packages/coding-agent/test/session-manager/file-operations.test.ts index 2b91908eee..021a60a87b 100644 --- a/packages/coding-agent/test/session-manager/file-operations.test.ts +++ b/packages/coding-agent/test/session-manager/file-operations.test.ts @@ -1,7 +1,23 @@ -import { mkdirSync, readFileSync, rmSync, writeFileSync } from "fs"; +import { chmodSync, lstatSync, mkdirSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from "fs"; import { tmpdir } from "os"; import { join } from "path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const fullReadCounter = vi.hoisted(() => ({ suffix: undefined as string | undefined, count: 0 })); +vi.mock("node:fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + readFileSync: ((path: Parameters[0], options?: never) => { + // Suffix match: repair resolves the realpath (/private/var vs /var on macOS). + if (fullReadCounter.suffix !== undefined && String(path).endsWith(fullReadCounter.suffix)) { + fullReadCounter.count++; + } + return actual.readFileSync(path, options); + }) as typeof actual.readFileSync, + }; +}); + import { computeOwnAndTotalUsage } from "../../src/core/context-tree.js"; import { findMostRecentSession, @@ -529,6 +545,127 @@ describe("SessionManager.setSessionFile with corrupted files", () => { rmSync(tempDir, { recursive: true, force: true }); }); + // The suspicion gate must keep clean opens at ONE full read (the loader's own). + it.each([ + [ + "a clean large session", + (): string[] => { + const filler = "x".repeat(2048); + const lines: string[] = []; + for (let index = 0; index < 2000; index++) { + lines.push( + JSON.stringify({ + type: "message", + id: `m${index}`, + parentId: index === 0 ? null : `m${index - 1}`, + message: { role: "user", content: filler, timestamp: index }, + }), + ); + } + return lines; + }, + ], + [ + "a benign trailing blank line", + (): string[] => [ + JSON.stringify({ + type: "message", + id: "m1", + parentId: null, + message: { role: "user", content: "hi", timestamp: 1 }, + }), + "", + ], + ], + ])("opens %s with exactly one full read", (_name, buildLines) => { + const file = join(tempDir, "gate.jsonl"); + const header = { + type: "session", + version: 3, + id: "gate-session", + timestamp: "2026-01-01T00:00:00Z", + cwd: "/tmp", + }; + writeFileSync(file, `${[JSON.stringify(header), ...buildLines()].join("\n")}\n`); + fullReadCounter.suffix = "gate.jsonl"; + fullReadCounter.count = 0; + + try { + SessionManager.open(file, tempDir); + expect(fullReadCounter.count).toBe(1); + } finally { + fullReadCounter.suffix = undefined; + } + }); + + it("repairs crash damage at open: torn tail truncated, zero-filled record recovered, appends stay separate lines", () => { + const file = join(tempDir, "crashed.jsonl"); + const header = { + type: "session", + version: 3, + id: "crashed-session", + timestamp: "2026-01-01T00:00:00Z", + cwd: "/tmp", + }; + const kept = { + type: "message", + id: "m1", + parentId: null, + message: { role: "user", content: "kept", timestamp: 1 }, + }; + const zeroFilled = { + type: "message", + id: "m2", + parentId: "m1", + message: { role: "user", content: "recovered", timestamp: 2 }, + }; + const damaged = `${JSON.stringify(header)}\n${JSON.stringify(kept)}\n\u0000\u0000\u0000\u0000${JSON.stringify(zeroFilled)}\n{"type":"message","id":"torn`; + writeFileSync(file, damaged); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); + + try { + const sm = SessionManager.open(file, tempDir); + expect(sm.getHeader()?.id).toBe("crashed-session"); + expect(sm.getEntries().map((entry) => entry.id)).toEqual(["m1", "m2"]); + sm.appendMessage({ role: "user", content: "after crash", timestamp: 3 }); + sm.flushNow(); + + const lines = readFileSync(file, "utf-8").split("\n").filter(Boolean); + const parsed = lines.map((line) => JSON.parse(line)); + expect(parsed.map((entry) => entry.id ?? entry.type)).toEqual([ + "crashed-session", + "m1", + "m2", + expect.any(String), + ]); + expect(parsed.at(-1)?.message?.content).toBe("after crash"); + expect(errorSpy).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + } + }); + + it("repairs a damaged transcript through its symlink alias at the real file", () => { + const realFile = join(tempDir, "real.jsonl"); + const alias = join(tempDir, "alias.jsonl"); + const header = { type: "session", version: 3, id: "sym-session", timestamp: "2026-01-01T00:00:00Z", cwd: "/tmp" }; + writeFileSync(realFile, `${JSON.stringify(header)}\n{"type":"message","id":"torn`); + chmodSync(realFile, 0o600); + symlinkSync(realFile, alias); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); + + try { + SessionManager.open(alias, tempDir); + expect(lstatSync(alias).isSymbolicLink()).toBe(true); + const repaired = readFileSync(realFile, "utf-8"); + expect(repaired.endsWith("\n")).toBe(true); + expect(repaired).not.toContain("torn"); + expect(statSync(realFile).mode & 0o777).toBe(0o600); + } finally { + errorSpy.mockRestore(); + } + }); + it("truncates and rewrites empty file with valid header", () => { const emptyFile = join(tempDir, "empty.jsonl"); writeFileSync(emptyFile, ""); diff --git a/packages/coding-agent/test/settings-manager-bug.test.ts b/packages/coding-agent/test/settings-manager-bug.test.ts index 7fafa2458d..8c916b279a 100644 --- a/packages/coding-agent/test/settings-manager-bug.test.ts +++ b/packages/coding-agent/test/settings-manager-bug.test.ts @@ -1,7 +1,7 @@ import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "fs"; import { join } from "path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { SettingsManager } from "../src/core/settings-manager.js"; +import { FileSettingsStorage, SettingsManager } from "../src/core/settings-manager.js"; describe("SettingsManager - External Edit Preservation", () => { const testDir = join(process.cwd(), "test-settings-bug-tmp"); @@ -22,6 +22,26 @@ describe("SettingsManager - External Edit Preservation", () => { } }); + it("re-reads under the late first-write lock so a racing first writer is not discarded", () => { + const settingsPath = join(agentDir, "settings.json"); + const storage = new FileSettingsStorage(projectDir, agentDir); + expect(existsSync(settingsPath)).toBe(false); + + const seen: Array = []; + storage.withLock("global", (current) => { + seen.push(current); + if (current === undefined) { + // A rival first writer lands between the unlocked read and the lock. + writeFileSync(settingsPath, JSON.stringify({ theme: "rival" })); + return JSON.stringify({ mine: true }); + } + return JSON.stringify({ ...JSON.parse(current), mine: true }); + }); + + expect(seen).toEqual([undefined, JSON.stringify({ theme: "rival" })]); + expect(JSON.parse(readFileSync(settingsPath, "utf-8"))).toEqual({ theme: "rival", mine: true }); + }); + it("should preserve file changes to packages array when changing unrelated setting", async () => { const settingsPath = join(agentDir, "settings.json"); diff --git a/prime-agent-runtime/src/rlm/harness.py b/prime-agent-runtime/src/rlm/harness.py index ed14e31d1a..b529c09524 100644 --- a/prime-agent-runtime/src/rlm/harness.py +++ b/prime-agent-runtime/src/rlm/harness.py @@ -11,9 +11,11 @@ import json import os +import stat from dataclasses import asdict, dataclass, field, fields from datetime import datetime, timezone from pathlib import Path +from uuid import uuid4 from typing import Any, Literal HarnessKind = Literal["prompt", "memory", "skill", "subagent"] @@ -295,8 +297,21 @@ def save(self) -> "HarnessState": }, "refinements": [asdict(event) for event in self.refinements], } - with self.file_path.open("w", encoding="utf-8") as f: - json.dump(data, f, indent=2, ensure_ascii=False) + # Atomic replace on the real file: aliases survive, readers never see a torn file. + target_path = Path(os.path.realpath(self.file_path)) + temp_path = target_path.with_name(f"{target_path.name}.{os.getpid()}.{uuid4().hex}.tmp") + try: + mode = stat.S_IMODE(os.stat(target_path).st_mode) + except FileNotFoundError: + mode = 0o600 + try: + # The temp carries its final mode from creation: no umask-open window. + descriptor = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode) + with os.fdopen(descriptor, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) + os.replace(temp_path, target_path) + finally: + temp_path.unlink(missing_ok=True) self._loaded_mtime = self._disk_mtime() return self diff --git a/prime-agent-runtime/test/test_harness.py b/prime-agent-runtime/test/test_harness.py index 7409f32a87..8ed89af89c 100644 --- a/prime-agent-runtime/test/test_harness.py +++ b/prime-agent-runtime/test/test_harness.py @@ -134,6 +134,77 @@ def test_persists_entries_and_refinements(self) -> None: self.assertIn("receiver_role='child'", overview) self.assertIn("refinements: 1", reloaded.overview()) + def test_save_failure_preserves_previous_state_on_disk(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + state = HarnessState(Path(temp_dir) / "harness_state.json") + state.create_memory("Durable fact", "Written before the crash.") + + crashing = HarnessState(state.file_path) + original_dump = json.dump + + def torn_dump(data: object, fh: object, **kwargs: object) -> None: + fh.write('{"schema": 1, "entr') # type: ignore[attr-defined] + raise OSError("disk full") + + json.dump = torn_dump # type: ignore[assignment] + try: + with self.assertRaises(OSError): + crashing.create_memory("Doomed fact", "Interrupted mid-write.") + finally: + json.dump = original_dump + + # The interrupted save must not have truncated the durable state. + reloaded = HarnessState(state.file_path) + titles = [entry.title for entry in reloaded.entries["memory"].values()] + self.assertEqual(titles, ["Durable fact"]) + + def test_save_preserves_restrictive_file_mode(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + state = HarnessState(Path(temp_dir) / "harness_state.json") + state.create_memory("First", "Creates the file.") + os.chmod(state.file_path, 0o600) + + state.create_memory("Second", "Replaces the file.") + + self.assertEqual(os.stat(state.file_path).st_mode & 0o777, 0o600) + + def test_save_temp_file_is_never_looser_than_the_destination(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + state = HarnessState(Path(temp_dir) / "harness_state.json") + state.create_memory("First", "Creates the file.") + os.chmod(state.file_path, 0o600) + + observed_modes: list[int] = [] + original_open = os.open + + def observing_open(path: object, flags: int, mode: int = 0o777, **kwargs: object) -> int: + if str(path).endswith(".tmp"): + observed_modes.append(mode) + return original_open(path, flags, mode, **kwargs) + + os.open = observing_open # type: ignore[assignment] + try: + state.create_memory("Second", "Replaces the file.") + finally: + os.open = original_open + + self.assertEqual(observed_modes, [0o600]) + self.assertEqual(os.stat(state.file_path).st_mode & 0o777, 0o600) + + def test_save_writes_through_a_symlinked_state_file(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + real_path = Path(temp_dir) / "real_state.json" + alias = Path(temp_dir) / "harness_state.json" + HarnessState(real_path).create_memory("Seed", "Creates the real file.") + alias.symlink_to(real_path) + + state = HarnessState(alias) + state.create_memory("Through alias", "Must land in the real file.") + + self.assertTrue(alias.is_symlink()) + titles = [entry.title for entry in HarnessState(real_path).entries["memory"].values()] + self.assertEqual(titles, ["Seed", "Through alias"]) + def test_load_ignores_unknown_json_keys(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: state_path = Path(temp_dir) / "harness_state.json"