From 29d35b61c8325513c64a5760d767717c082e8b1b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Elberte=20Pl=C3=ADnio?= Date: Sun, 19 Jul 2026 00:14:29 -0300 Subject: [PATCH] fix: reaper meta key for android, evidence-walk cost, atomic writes (#46) P0: - Android session create/destroy now sets/clears REAPER_CLEANUP_PENDING_META_KEY using the same protocol as desktop-linux and browser: set it when a stop fails so a leaked emulator stays reaper-trackable, clear it once cleanup is confirmed complete. Added a test proving a leaked android session is picked up and reaped by reapDeadRunningSessions. P1 performance/reliability: - appendAction no longer does a full recursive readdir+lstat walk of the run directory on every evidence action. The artifact byte total is cached per run directory and only rescanned when a cheap entry-name snapshot (readdir only, filtered the same way the walk is) shows the counted tree actually changed; the journal's own live size is read off the already-open handle instead of being folded into the walk, so uncounted control-file churn (the journal lock, the truncation sentinel) never forces a rescan. Orphaned artifacts written directly to disk are still caught, matching prior behavior. - MCP run-log resource reads now cap the amount pulled into memory, mirroring the screenshot guard: an oversized log is read via tail semantics (last N bytes off an open handle) instead of loading the whole file before redacting secrets from it. - The browser supervisor child-shim's stderr buffer is now bounded to the last 64KB when no newline arrives, instead of growing without limit. P1 simplify: - Hoisted writeFileAtomic (tmp + rename + mode-preserving, cleans up the tmp file on any failure) from agent-installers/atomicFile.ts into core (packages/core/src/paths.ts, exported from index.ts). Used it in core/session.ts writeSession, run.ts writeManifest, config.ts writeConfigFile (all three previously leaked the tmp file on rename failure), agent-installers' own call sites, and the two already-safe evidence.ts call sites (active-pointer publish, truncation-sentinel commit) without touching evidence.ts's concurrency-critical claim/lock code paths otherwise. Validated with bun/vitest test runs and tsc --noEmit across every touched package. --- .../agent-installers/src/agents/custom.ts | 8 +- packages/agent-installers/src/atomicFile.ts | 36 --- packages/agent-installers/src/index.ts | 2 - packages/agent-installers/src/jsonConfig.ts | 2 +- packages/agent-installers/src/state.ts | 3 +- packages/agent-installers/src/tomlConfig.ts | 2 +- packages/android/src/session.ts | 56 ++++- packages/android/test/reaper.test.ts | 107 +++++++++ packages/browser/src/supervisor.ts | 6 + packages/browser/test/supervisor.test.ts | 30 +++ packages/core/src/config.ts | 17 +- packages/core/src/evidence.ts | 215 +++++++++++++----- packages/core/src/index.ts | 1 + packages/core/src/paths.ts | 41 ++++ packages/core/src/run.ts | 16 +- packages/core/src/session.ts | 11 +- packages/core/test/evidence.test.ts | 47 ++++ packages/mcp-server/src/resources.ts | 47 +++- packages/mcp-server/test/resources.test.ts | 30 +++ 19 files changed, 532 insertions(+), 145 deletions(-) delete mode 100644 packages/agent-installers/src/atomicFile.ts create mode 100644 packages/android/test/reaper.test.ts diff --git a/packages/agent-installers/src/agents/custom.ts b/packages/agent-installers/src/agents/custom.ts index f5d7e95..7e248f9 100644 --- a/packages/agent-installers/src/agents/custom.ts +++ b/packages/agent-installers/src/agents/custom.ts @@ -1,7 +1,11 @@ import fs from "node:fs"; import path from "node:path"; -import { agentsDir, ensureDir, type EnvLike } from "@pickforge/picklab-core"; -import { writeFileAtomic } from "../atomicFile.js"; +import { + agentsDir, + ensureDir, + writeFileAtomic, + type EnvLike, +} from "@pickforge/picklab-core"; import { MCP_SERVER_NAME, renderJsonSnippet, diff --git a/packages/agent-installers/src/atomicFile.ts b/packages/agent-installers/src/atomicFile.ts deleted file mode 100644 index c81cba1..0000000 --- a/packages/agent-installers/src/atomicFile.ts +++ /dev/null @@ -1,36 +0,0 @@ -import fs from "node:fs"; -import path from "node:path"; - -let tmpCounter = 0; - -export async function writeFileAtomic( - filePath: string, - content: string, -): Promise { - const dir = path.dirname(filePath); - await fs.promises.mkdir(dir, { recursive: true }); - tmpCounter += 1; - const tmp = path.join( - dir, - `.${path.basename(filePath)}.tmp-${process.pid}-${tmpCounter}`, - ); - let mode: number | undefined; - try { - mode = (await fs.promises.stat(filePath)).mode & 0o777; - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code !== "ENOENT" && code !== "ENOTDIR") { - throw error; - } - } - try { - await fs.promises.writeFile(tmp, content, { encoding: "utf8", mode }); - if (mode !== undefined) { - await fs.promises.chmod(tmp, mode); - } - await fs.promises.rename(tmp, filePath); - } catch (error) { - await fs.promises.rm(tmp, { force: true }); - throw error; - } -} diff --git a/packages/agent-installers/src/index.ts b/packages/agent-installers/src/index.ts index a57b3e3..c527e09 100644 --- a/packages/agent-installers/src/index.ts +++ b/packages/agent-installers/src/index.ts @@ -9,8 +9,6 @@ export { type RegistrationState, } from "./types.js"; -export { writeFileAtomic } from "./atomicFile.js"; - export { agentsStatePath, readAgentsState, diff --git a/packages/agent-installers/src/jsonConfig.ts b/packages/agent-installers/src/jsonConfig.ts index 4e40896..80b00a2 100644 --- a/packages/agent-installers/src/jsonConfig.ts +++ b/packages/agent-installers/src/jsonConfig.ts @@ -1,5 +1,5 @@ import fs from "node:fs"; -import { writeFileAtomic } from "./atomicFile.js"; +import { writeFileAtomic } from "@pickforge/picklab-core"; import { backupFile } from "./backup.js"; import { BROWSER_MCP_SERVER_NAME, diff --git a/packages/agent-installers/src/state.ts b/packages/agent-installers/src/state.ts index ae6d5ed..3270e18 100644 --- a/packages/agent-installers/src/state.ts +++ b/packages/agent-installers/src/state.ts @@ -1,7 +1,6 @@ import fs from "node:fs"; import path from "node:path"; -import { agentsDir, type EnvLike } from "@pickforge/picklab-core"; -import { writeFileAtomic } from "./atomicFile.js"; +import { agentsDir, writeFileAtomic, type EnvLike } from "@pickforge/picklab-core"; export interface AgentStateEntry { registered: boolean; diff --git a/packages/agent-installers/src/tomlConfig.ts b/packages/agent-installers/src/tomlConfig.ts index 09ecaac..f557539 100644 --- a/packages/agent-installers/src/tomlConfig.ts +++ b/packages/agent-installers/src/tomlConfig.ts @@ -1,5 +1,5 @@ import fs from "node:fs"; -import { writeFileAtomic } from "./atomicFile.js"; +import { writeFileAtomic } from "@pickforge/picklab-core"; import { backupFile } from "./backup.js"; import { renderTomlSnippet } from "./snippet.js"; import type { ChangeResult, McpServerEntry } from "./types.js"; diff --git a/packages/android/src/session.ts b/packages/android/src/session.ts index a3f5e4d..d3ad01c 100644 --- a/packages/android/src/session.ts +++ b/packages/android/src/session.ts @@ -5,6 +5,7 @@ import { getSession, isPidAlive, reapDeadRunningSessions, + REAPER_CLEANUP_PENDING_META_KEY, sessionsDir, updateSession, type AndroidSessionInfo, @@ -104,18 +105,41 @@ export async function createAndroidSession( logDir, }; } catch (error) { + let emulatorGone = true; if (emulator !== undefined) { - await stopEmulator({ - serial: emulator.serial, - pid: emulator.pid, - sdk: opts.sdk, - env: opts.env, - registryEnv, - }).catch(() => {}); + try { + emulatorGone = await stopEmulator({ + serial: emulator.serial, + pid: emulator.pid, + sdk: opts.sdk, + env: opts.env, + registryEnv, + }); + } catch { + emulatorGone = false; + } } - await updateSession(record.id, { status: "error" }, registryEnv).catch( - () => {}, - ); + const clearedMeta = { ...record.meta }; + delete clearedMeta[REAPER_CLEANUP_PENDING_META_KEY]; + await updateSession( + record.id, + emulatorGone + ? { status: "error", meta: clearedMeta } + : { + status: "error", + meta: { + ...record.meta, + [REAPER_CLEANUP_PENDING_META_KEY]: true, + }, + android: { + avdName, + serial: emulator?.serial, + emulatorPid: emulator?.pid, + consolePort: emulator?.consolePort, + }, + }, + registryEnv, + ).catch(() => {}); throw error; } } @@ -147,7 +171,17 @@ export async function destroyAndroidSession( failure = error instanceof Error ? error : new Error(String(error)); } if (!stopped) { - await updateSession(id, { status: "error" }, registryEnv).catch(() => {}); + await updateSession( + id, + { + status: "error", + meta: { + ...record.meta, + [REAPER_CLEANUP_PENDING_META_KEY]: true, + }, + }, + registryEnv, + ).catch(() => {}); throw new Error( `Failed to stop emulator of android session ${id} ` + `(serial ${android.serial ?? "unknown"}, pid ${android.emulatorPid ?? "unknown"})` + diff --git a/packages/android/test/reaper.test.ts b/packages/android/test/reaper.test.ts new file mode 100644 index 0000000..7ec2638 --- /dev/null +++ b/packages/android/test/reaper.test.ts @@ -0,0 +1,107 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, describe, expect, it, vi } from "vitest"; + +let forceStopFailure = false; + +vi.mock("../src/emulator.js", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + stopEmulator: vi.fn(async (opts: Parameters[0]) => { + if (forceStopFailure) { + return false; + } + return actual.stopEmulator(opts); + }), + }; +}); + +import { + REAPER_CLEANUP_PENDING_META_KEY, + getSession, + isPidAlive, + reapDeadRunningSessions, + type EnvLike, +} from "@pickforge/picklab-core"; +import { createAndroidSession, destroyAndroidSession } from "../src/index.js"; + +const tmpRoot = fs.mkdtempSync( + path.join(os.tmpdir(), "picklab-android-reaper-"), +); +const home = path.join(tmpRoot, "home"); +const projectDir = path.join(tmpRoot, "project"); +fs.mkdirSync(home, { recursive: true }); +fs.mkdirSync(projectDir, { recursive: true }); +const registryEnv: EnvLike = { ...process.env, PICKLAB_HOME: home }; + +afterAll(() => { + fs.rmSync(tmpRoot, { recursive: true, force: true }); +}); + +function writeExecutable(filePath: string, content: string): void { + fs.mkdirSync(path.dirname(filePath), { recursive: true }); + fs.writeFileSync(filePath, content, { mode: 0o755 }); +} + +function makeFakeSdk(): string { + const sdk = path.join(tmpRoot, "sdk"); + writeExecutable( + path.join(sdk, "emulator", "emulator"), + "#!/bin/sh\nPATH=/usr/bin:/bin\nexec sleep 60\n", + ); + writeExecutable( + path.join(sdk, "platform-tools", "adb"), + [ + "#!/bin/sh", + 'case "$*" in', + " *getprop*) echo 1 ;;", + ' devices) printf "List of devices attached\\nemulator-5554\\tdevice\\n" ;;', + ' *"emu kill"*) exit 0 ;;', + "esac", + "exit 0", + ].join("\n"), + ); + return sdk; +} + +describe("android reaper tracking", () => { + it("keeps a leaked android session reaper-trackable after a failed destroy", async () => { + const sdk = makeFakeSdk(); + const session = await createAndroidSession({ + projectDir, + registryEnv, + sdk, + port: 5554, + env: { PATH: "" }, + bootPollIntervalMs: 20, + bootTimeoutMs: 5_000, + }); + + forceStopFailure = true; + try { + await expect( + destroyAndroidSession(session.id, registryEnv, { + sdk, + env: { PATH: "" }, + timeoutMs: 300, + }), + ).rejects.toThrow(/Failed to stop emulator/); + + const leaked = await getSession(session.id, registryEnv); + expect(leaked?.status).toBe("error"); + expect(leaked?.meta?.[REAPER_CLEANUP_PENDING_META_KEY]).toBe(true); + expect(leaked?.android?.emulatorPid).toBe(session.emulatorPid); + expect(isPidAlive(session.emulatorPid)).toBe(true); + } finally { + forceStopFailure = false; + } + + const reaped = await reapDeadRunningSessions(registryEnv); + expect(reaped.map((record) => record.id)).toContain(session.id); + expect(await getSession(session.id, registryEnv)).toBeUndefined(); + expect(isPidAlive(session.emulatorPid)).toBe(false); + }, 20_000); +}); diff --git a/packages/browser/src/supervisor.ts b/packages/browser/src/supervisor.ts index 107bfff..2719389 100644 --- a/packages/browser/src/supervisor.ts +++ b/packages/browser/src/supervisor.ts @@ -39,6 +39,7 @@ const child = spawn(binary, args, { stdio: ["inherit", "inherit", "pipe"], }); let stderrBuffer = ""; +const MAX_STDERR_BUFFER_BYTES = 64 * 1024; function forwardBrowserStderr(line) { process.stderr.write( line.replace( @@ -64,6 +65,11 @@ child.stderr.setEncoding("utf8"); child.stderr.on("data", (chunk) => { stderrBuffer += chunk; flushBrowserStderr(); + // No newline arrived to flush a complete line; bound the buffer so an + // unbounded stream of newline-free stderr can never grow it forever. + if (stderrBuffer.length > MAX_STDERR_BUFFER_BYTES) { + stderrBuffer = stderrBuffer.slice(-MAX_STDERR_BUFFER_BYTES); + } }); child.stderr.once("end", () => flushBrowserStderr(true)); child.once("error", (error) => { diff --git a/packages/browser/test/supervisor.test.ts b/packages/browser/test/supervisor.test.ts index ff1c5fc..df75a1a 100644 --- a/packages/browser/test/supervisor.test.ts +++ b/packages/browser/test/supervisor.test.ts @@ -60,6 +60,36 @@ describe("buildSupervisedBrowserCommand", () => { } }); + it("bounds unbounded newline-free stderr instead of buffering it all", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "picklab-supervisor-")); + const browser = path.join(tmp, "browser.cjs"); + fs.writeFileSync( + browser, + [ + 'const fs = require("fs");', + 'fs.writeSync(2, "HEAD_MARKER_ABC");', + 'fs.writeSync(2, "a".repeat(5 * 1024 * 1024));', + 'fs.writeSync(2, "TAIL_MARKER_XYZ");', + "process.exit(0);", + ].join("\n"), + ); + try { + const command = buildSupervisedBrowserCommand( + process.execPath, + process.execPath, + [browser], + ); + const result = await runCommand(command.command, command.args); + + expect(result.ok).toBe(true); + expect(result.stderr).toContain("TAIL_MARKER_XYZ"); + expect(result.stderr).not.toContain("HEAD_MARKER_ABC"); + expect(result.stderr.length).toBeLessThan(200 * 1024); + } finally { + fs.rmSync(tmp, { recursive: true, force: true }); + } + }); + it("rejects missing executable paths", () => { expect(() => buildSupervisedBrowserCommand("", "/usr/bin/chromium", []), diff --git a/packages/core/src/config.ts b/packages/core/src/config.ts index 0b5aac1..1e2751d 100644 --- a/packages/core/src/config.ts +++ b/packages/core/src/config.ts @@ -4,6 +4,7 @@ import { ensureDir, globalConfigPath, projectConfigPath, + writeFileAtomic, type EnvLike, } from "./paths.js"; @@ -102,24 +103,12 @@ export async function loadConfig( ) as PicklabConfig; } -let tmpCounter = 0; - async function writeConfigFile( filePath: string, config: PicklabConfig, ): Promise { - const dir = await ensureDir(path.dirname(filePath)); - tmpCounter += 1; - const tmp = path.join( - dir, - `.${path.basename(filePath)}.tmp-${process.pid}-${tmpCounter}`, - ); - await fs.promises.writeFile( - tmp, - `${JSON.stringify(config, null, 2)}\n`, - "utf8", - ); - await fs.promises.rename(tmp, filePath); + await ensureDir(path.dirname(filePath)); + await writeFileAtomic(filePath, `${JSON.stringify(config, null, 2)}\n`); } export async function saveProjectConfig( diff --git a/packages/core/src/evidence.ts b/packages/core/src/evidence.ts index 03392bc..73a3dd4 100644 --- a/packages/core/src/evidence.ts +++ b/packages/core/src/evidence.ts @@ -1,7 +1,7 @@ import fs from "node:fs"; import path from "node:path"; import { setTimeout as delay } from "node:timers/promises"; -import { ensureDir, runsDir } from "./paths.js"; +import { ensureDir, runsDir, writeFileAtomic } from "./paths.js"; import { isPidAlive, processIdentityMatches, @@ -63,8 +63,6 @@ const EMPTY_CLAIM_GRACE_ATTEMPTS = 4; const MAX_CLAIM_ATTEMPTS = 10_000; const SAFE_NAME_PATTERN = /^[a-z0-9][a-z0-9._-]*$/i; -let pointerTmpCounter = 0; - export type EvidenceActionStatus = "ok" | "error" | "cancelled" | "timeout"; /** @@ -741,18 +739,7 @@ export async function beginEvidenceRun( // Publish atomically (temp + rename) so a concurrent reader never sees a // torn pointer — only the intact claim or the intact full pointer. - pointerTmpCounter += 1; - const tmp = path.join( - parent, - `.active-${sessionId}.json.tmp-${ownerPid}-${pointerTmpCounter}`, - ); - await fs.promises.writeFile(tmp, pointerContent, "utf8"); - try { - await fs.promises.rename(tmp, pointerPath); - } catch (renameError) { - await fs.promises.unlink(tmp).catch(() => {}); - throw renameError; - } + await writeFileAtomic(pointerPath, pointerContent); // Final confirmation that the published pointer is ours. const publishedRaw = await readTextIfPresent(pointerPath); @@ -922,7 +909,96 @@ function buildTruncationMarker( * same run observes the same total, so the cap cannot be evaded by spreading * writes over many calls or many processes. */ -async function measureRunEvidenceBytes(runDir: string): Promise { +interface RunByteCacheEntry { + /** Cumulative bytes of every counted file except the journal itself. */ + artifactBytes: number; + /** + * The exact same *counted* entry names measurement itself would see (every + * non-symlink file except the manifest, the journal, and dot-prefixed + * control/temp files, plus every non-symlink subdirectory), keyed by path + * relative to runDir. Deliberately excludes uncounted control files (the + * journal lock, the truncation sentinel, the manifest) so their churn on + * every append never forces a rescan. + */ + entries: Map; +} + +/** In-process cache of the last-measured artifact footprint per run directory. */ +const runByteCache = new Map(); + +/** Drop a run's cached byte-accounting state (e.g. once it is pruned). */ +function forgetRunByteCache(runDir: string): void { + runByteCache.delete(runDir); +} + +/** + * Snapshot the set of *counted* entry names in every directory of the run tree + * (never following symlinks, never touching a file's contents) — the same + * filter `walkRunArtifactBytes` applies. Comparing this against a prior + * snapshot is a cheap, correct way to tell whether anything that contributes + * to the artifact total was added, removed, or renamed, without lstat-ing any + * file and without being fooled by uncounted control-file churn (the journal + * lock and truncation sentinel are created and removed on every append). + * Returns `undefined` if the run directory itself is gone. + */ +async function snapshotRunDirEntries( + runDir: string, +): Promise | undefined> { + const result = new Map(); + const walk = async (dir: string, rel: string): Promise => { + let entries: fs.Dirent[]; + try { + entries = await fs.promises.readdir(dir, { withFileTypes: true }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } + const files: string[] = []; + const subdirs: string[] = []; + for (const entry of entries) { + if (entry.isSymbolicLink()) continue; + if (entry.isDirectory()) { + subdirs.push(entry.name); + continue; + } + if (!entry.isFile()) continue; + const rel2 = rel === "" ? entry.name : `${rel}/${entry.name}`; + if (rel2 === "manifest.json") continue; + if (rel2 === EVIDENCE_ACTION_LOG) continue; + if (entry.name.startsWith(".")) continue; + files.push(entry.name); + } + files.sort(); + subdirs.sort(); + result.set(rel, files.join("\0")); + for (const name of subdirs) { + const childRel = rel === "" ? name : `${rel}/${name}`; + if (!(await walk(path.join(dir, name), childRel))) return false; + } + return true; + }; + return (await walk(runDir, "")) ? result : undefined; +} + +function dirEntriesEqual( + a: Map, + b: Map, +): boolean { + if (a.size !== b.size) return false; + for (const [rel, signature] of a) { + if (b.get(rel) !== signature) return false; + } + return true; +} + +/** + * Full recursive scan of every counted file under the run directory (every + * file except the action journal, the manifest, and dot-prefixed control/temp + * files; symlinks are never followed). This is the ground truth the cache in + * `measureRunArtifactBytes` falls back to whenever the directory-mtime + * snapshot shows the tree may have changed. + */ +async function walkRunArtifactBytes(runDir: string): Promise { let total = 0; const walk = async (dir: string): Promise => { let entries: fs.Dirent[]; @@ -941,9 +1017,12 @@ async function measureRunEvidenceBytes(runDir: string): Promise { } if (!entry.isFile()) continue; const rel = path.relative(runDir, full); - // Exclude the manifest and any dot-prefixed control/temp file (the - // truncation sentinel, `.manifest.json.tmp-*`, pointer temps). + // Exclude the manifest, the journal (measured live, separately, since it + // is appended in place rather than replaced), and any dot-prefixed + // control/temp file (the truncation sentinel, `.manifest.json.tmp-*`, + // pointer temps). if (rel === "manifest.json") continue; + if (rel === EVIDENCE_ACTION_LOG) continue; if (entry.name.startsWith(".")) continue; try { const stat = await fs.promises.lstat(full); @@ -958,6 +1037,62 @@ async function measureRunEvidenceBytes(runDir: string): Promise { return total; } +/** + * Measure the cumulative on-disk bytes of every counted artifact under the run + * directory, excluding the action journal (whose live size the caller adds + * separately, since it is appended to in place). Because this reads real bytes + * on disk it stays consistent across processes: the cap cannot be evaded by + * spreading writes over many calls or many processes, or by writing artifact + * files directly without ever referencing them in a journaled action. + * + * A full recursive lstat scan is only performed the first time a run is seen, + * or when an entry-name snapshot (readdir only, no lstat) shows the counted + * tree actually changed since the last scan (a file was added, removed, or + * renamed somewhere in the tree). Otherwise the previously measured total is + * reused as-is, turning the common case — many actions in a row that write no + * new artifact — into a handful of `readdir` calls instead of a full recursive + * `readdir`+`lstat` of every file on every append. + */ +async function measureRunArtifactBytes(runDir: string): Promise { + const cached = runByteCache.get(runDir); + const currentEntries = await snapshotRunDirEntries(runDir); + if ( + cached !== undefined && + currentEntries !== undefined && + dirEntriesEqual(cached.entries, currentEntries) + ) { + return cached.artifactBytes; + } + const artifactBytes = await walkRunArtifactBytes(runDir); + if (currentEntries === undefined) { + forgetRunByteCache(runDir); + } else { + runByteCache.set(runDir, { artifactBytes, entries: currentEntries }); + } + return artifactBytes; +} + +/** + * Measure the cumulative on-disk evidence footprint of a run: the action + * journal plus every artifact file written under the run directory + * (screenshots, logs, and any other run-relative artifact). Control and summary + * files that are not auto-generated evidence — the manifest, the truncation + * sentinel, and transient dot-prefixed temp files — are excluded, as are + * symlinks (never followed, so a planted link cannot skew the count or escape + * the tree). + * + * Because this reads real bytes on disk, the count is inherently cumulative + * across appends and consistent across processes: any process appending to the + * same run observes the same total, so the cap cannot be evaded by spreading + * writes over many calls or many processes. + */ +async function measureRunEvidenceBytes( + runDir: string, + journalBytes: number, +): Promise { + return (await measureRunArtifactBytes(runDir)) + journalBytes; +} + async function repairTornJournalTail( handle: fs.promises.FileHandle, ): Promise { @@ -1065,8 +1200,12 @@ export async function appendAction( await repairTornJournalTail(handle); // Derive current usage from the run directory's real on-disk bytes (journal // + artifacts), so the cap is cumulative across every prior append and every - // process. `externalBytes` only adds bytes not yet under the run dir. - const used = (await measureRunEvidenceBytes(runDir)) + externalBytes; + // process. `externalBytes` only adds bytes not yet under the run dir. The + // journal's own live size comes straight off the open handle (already + // repaired above), since it is appended to in place rather than replaced. + const journalBytes = (await handle.stat()).size; + const used = + (await measureRunEvidenceBytes(runDir, journalBytes)) + externalBytes; if (used >= maxBytes) { // Already at or beyond the cap. Ensure the one-time marker exists, then @@ -1230,7 +1369,6 @@ async function journalHasTruncationMarker(runDir: string): Promise { */ async function commitTruncationSentinel( sentinelPath: string, - runDir: string, ownerPid: number, ownerStartTicks: number | undefined, ): Promise { @@ -1241,18 +1379,7 @@ async function commitTruncationSentinel( committedAt: new Date().toISOString(), }; if (ownerStartTicks !== undefined) commit.ownerStartTicks = ownerStartTicks; - pointerTmpCounter += 1; - const tmp = path.join( - runDir, - `${TRUNCATION_SENTINEL}.tmp-${ownerPid}-${pointerTmpCounter}`, - ); - await fs.promises.writeFile(tmp, `${JSON.stringify(commit)}\n`, "utf8"); - try { - await fs.promises.rename(tmp, sentinelPath); - } catch (error) { - await fs.promises.unlink(tmp).catch(() => {}); - throw error; - } + await writeFileAtomic(sentinelPath, `${JSON.stringify(commit)}\n`); } /** @@ -1311,12 +1438,7 @@ async function writeTruncationMarkerOnce( state.claim.ownerPid === ownerPid && state.claim.ownerStartTicks === ownerStartTicks; if (ownedByCaller && (await journalHasTruncationMarker(runDir))) { - await commitTruncationSentinel( - sentinelPath, - runDir, - ownerPid, - ownerStartTicks, - ); + await commitTruncationSentinel(sentinelPath, ownerPid, ownerStartTicks); } return false; } @@ -1368,12 +1490,7 @@ async function writeTruncationMarkerOnce( // commit its sentinel (crash between append and commit). Never write a // second one: recover by committing the sentinel over the existing marker. if (await journalHasTruncationMarker(runDir)) { - await commitTruncationSentinel( - sentinelPath, - runDir, - ownerPid, - ownerStartTicks, - ); + await commitTruncationSentinel(sentinelPath, ownerPid, ownerStartTicks); return false; } @@ -1386,12 +1503,7 @@ async function writeTruncationMarkerOnce( ); } appended = true; - await commitTruncationSentinel( - sentinelPath, - runDir, - ownerPid, - ownerStartTicks, - ); + await commitTruncationSentinel(sentinelPath, ownerPid, ownerStartTicks); return true; } catch (error) { // The append failed, so no marker was written: clear our claim so the @@ -1644,6 +1756,7 @@ export async function pruneFinalizedEvidenceRuns( const activeNow = await collectActiveRunIds(parent); if (activeNow.has(dirName)) continue; await fs.promises.rm(dir, { recursive: true, force: true }); + forgetRunByteCache(dir); removed.push(dirName); } return removed; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index f63725e..dee19ef 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -9,6 +9,7 @@ export { projectConfigPath, runsDir, sessionsDir, + writeFileAtomic, type EnvLike, } from "./paths.js"; diff --git a/packages/core/src/paths.ts b/packages/core/src/paths.ts index 59c4ab5..6cc2331 100644 --- a/packages/core/src/paths.ts +++ b/packages/core/src/paths.ts @@ -37,6 +37,47 @@ export async function ensureDir(dir: string): Promise { return dir; } +let atomicTmpCounter = 0; + +/** + * Write a file atomically: write to a sibling temp file, preserve the target's + * existing permission mode (if any), then rename over the destination. The + * rename is atomic on the same filesystem, so a reader never observes a + * partially written file. On any failure the temp file is removed rather than + * left behind. + */ +export async function writeFileAtomic( + filePath: string, + content: string, +): Promise { + const dir = path.dirname(filePath); + await fs.promises.mkdir(dir, { recursive: true }); + atomicTmpCounter += 1; + const tmp = path.join( + dir, + `.${path.basename(filePath)}.tmp-${process.pid}-${atomicTmpCounter}`, + ); + let mode: number | undefined; + try { + mode = (await fs.promises.stat(filePath)).mode & 0o777; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code !== "ENOENT" && code !== "ENOTDIR") { + throw error; + } + } + try { + await fs.promises.writeFile(tmp, content, { encoding: "utf8", mode }); + if (mode !== undefined) { + await fs.promises.chmod(tmp, mode); + } + await fs.promises.rename(tmp, filePath); + } catch (error) { + await fs.promises.rm(tmp, { force: true }); + throw error; + } +} + /** * Confinement guard for an ephemeral browser profile. In addition to lexical * containment, every existing path from the sessions directory through the diff --git a/packages/core/src/run.ts b/packages/core/src/run.ts index 4dcb2e0..5c261e1 100644 --- a/packages/core/src/run.ts +++ b/packages/core/src/run.ts @@ -1,6 +1,6 @@ import fs from "node:fs"; import path from "node:path"; -import { ensureDir, runsDir } from "./paths.js"; +import { ensureDir, runsDir, writeFileAtomic } from "./paths.js"; export type RunStatus = "running" | "completed" | "failed"; export type ArtifactType = "screenshot" | "log" | "report" | "other"; @@ -58,8 +58,6 @@ export interface CreateRunOptions { const SLUG_PATTERN = /^[a-z0-9][a-z0-9._-]*$/i; -let tmpCounter = 0; - function assertValidSlug(slug: string): void { if (!SLUG_PATTERN.test(slug) || slug.includes("..")) { throw new Error( @@ -79,17 +77,7 @@ function formatTimestamp(date: Date): string { async function writeManifest(runDir: string, manifest: RunManifest): Promise { const target = path.join(runDir, "manifest.json"); - tmpCounter += 1; - const tmp = path.join( - runDir, - `.manifest.json.tmp-${process.pid}-${tmpCounter}`, - ); - await fs.promises.writeFile( - tmp, - `${JSON.stringify(manifest, null, 2)}\n`, - "utf8", - ); - await fs.promises.rename(tmp, target); + await writeFileAtomic(target, `${JSON.stringify(manifest, null, 2)}\n`); } export class RunHandle { diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 9f9bb60..b23ca29 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -8,6 +8,7 @@ import { isProfileConfined, sessionsDir, runsDir, + writeFileAtomic, type EnvLike, } from "./paths.js"; import { @@ -84,8 +85,6 @@ const ID_PREFIXES: Record = { const SESSION_ID_PATTERN = /^(desk|andr|duo|brow)-[0-9a-f]{6,}$/; const MAX_ID_ATTEMPTS = 5; -let tmpCounter = 0; - function isValidSessionId(id: string): boolean { return SESSION_ID_PATTERN.test(id); } @@ -114,13 +113,7 @@ async function writeSession( ): Promise { const dir = await ensureDir(sessionsDir(env)); const target = path.join(dir, `${record.id}.json`); - tmpCounter += 1; - const tmp = path.join( - dir, - `.${record.id}.json.tmp-${process.pid}-${tmpCounter}`, - ); - await fs.promises.writeFile(tmp, serialize(record), "utf8"); - await fs.promises.rename(tmp, target); + await writeFileAtomic(target, serialize(record)); } export async function createSession( diff --git a/packages/core/test/evidence.test.ts b/packages/core/test/evidence.test.ts index d871abe..c55e472 100644 --- a/packages/core/test/evidence.test.ts +++ b/packages/core/test/evidence.test.ts @@ -731,6 +731,53 @@ describe("evidence cap and truncation", () => { expect(result.outcome).toBe("appended"); expect(await isEvidenceTruncated(run.dir)).toBe(false); }); + + it("reuses the cached artifact total across metadata-only appends instead of rescanning every file", async () => { + const { run } = await beginEvidenceRun(project, "desk-cache00"); + const screenshots = path.join(run.dir, "screenshots"); + // Seed a large number of on-disk artifacts up front. + await Promise.all( + Array.from({ length: 40 }, (_, i) => + fs.promises.writeFile( + path.join(screenshots, `seed-${i}.bin`), + Buffer.alloc(1024), + ), + ), + ); + // First append after the artifacts land always does a full scan, seeding + // the cache with the correct total. + const seeded = await appendAction(run.dir, action({ actionId: "seed" }), { + maxBytes: 1_000_000, + }); + expect(seeded.usedBytes).toBeGreaterThanOrEqual(40 * 1024); + + const lstatSpy = vi.spyOn(fs.promises, "lstat"); + lstatSpy.mockClear(); + for (let i = 0; i < 10; i += 1) { + await appendAction(run.dir, action({ actionId: `meta-${i}` }), { + maxBytes: 1_000_000, + }); + } + const lstatCalls = lstatSpy.mock.calls.length; + lstatSpy.mockRestore(); + // A full rescan touches every one of the 40 seeded files on every call + // (400+ lstats across 10 calls). The cache should keep this bounded to + // roughly the number of directories checked per call, independent of the + // artifact count. + expect(lstatCalls).toBeLessThan(40); + + // A new artifact written directly to disk (no appendAction call in + // between) must still be picked up on the next append: the cache must + // never permanently undercount. + await fs.promises.writeFile( + path.join(screenshots, "late.bin"), + Buffer.alloc(2048), + ); + const after = await appendAction(run.dir, action({ actionId: "after" }), { + maxBytes: 1_000_000, + }); + expect(after.usedBytes).toBeGreaterThanOrEqual(seeded.usedBytes + 2048); + }); }); describe("truncation marker durability", () => { diff --git a/packages/mcp-server/src/resources.ts b/packages/mcp-server/src/resources.ts index 331918a..7c13dc1 100644 --- a/packages/mcp-server/src/resources.ts +++ b/packages/mcp-server/src/resources.ts @@ -23,6 +23,8 @@ import { sessionStatusEntry } from "./tools/session.js"; const SAFE_NAME_PATTERN = /^[A-Za-z0-9._-]+$/; const MAX_BLOB_BYTES = 8 * 1024 * 1024; +/** Upper bound on how many trailing bytes of an oversized run log are read into memory. */ +const MAX_LOG_TAIL_BYTES = 1 * 1024 * 1024; function decodeVariable(variables: Variables, label: string): string { const raw = variables[label]; @@ -217,6 +219,32 @@ async function readTextFileNoFollow( } } +// Read only the trailing `maxBytes` of a file (tail semantics), so an +// oversized log never gets pulled fully into memory just to redact secrets +// from it. Drops a partial leading line so the returned text starts cleanly. +async function readTailUtf8( + filePath: string, + fileSize: number, + maxBytes: number, +): Promise { + const handle = await fs.promises.open( + filePath, + fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW, + ); + try { + const length = Math.min(maxBytes, fileSize); + const position = fileSize - length; + const buffer = Buffer.alloc(length); + const { bytesRead } = await handle.read(buffer, 0, length, position); + const text = buffer.subarray(0, bytesRead).toString("utf8"); + if (position === 0) return text; + const firstNewline = text.indexOf("\n"); + return firstNewline === -1 ? text : text.slice(firstNewline + 1); + } finally { + await handle.close(); + } +} + async function isRootFileSafe( ctx: ServerContext, runId: string, @@ -584,18 +612,33 @@ export function registerResources(server: McpServer, ctx: ServerContext): void { filePath, () => new Error(`Log not found: ${runId}/${name}`), ); + let stat: fs.Stats; + try { + stat = await fs.promises.stat(filePath); + } catch { + throw new Error(`Log not found: ${runId}/${name}`); + } let raw: string; + let truncated = false; try { - raw = await fs.promises.readFile(filePath, "utf8"); + if (stat.size > MAX_LOG_TAIL_BYTES) { + truncated = true; + raw = await readTailUtf8(filePath, stat.size, MAX_LOG_TAIL_BYTES); + } else { + raw = await fs.promises.readFile(filePath, "utf8"); + } } catch { throw new Error(`Log not found: ${runId}/${name}`); } + const text = redactSecrets(raw); return { contents: [ { uri: uri.href, mimeType: "text/plain", - text: redactSecrets(raw), + text: truncated + ? `[truncated: showing last ${MAX_LOG_TAIL_BYTES} of ${stat.size} bytes]\n${text}` + : text, }, ], }; diff --git a/packages/mcp-server/test/resources.test.ts b/packages/mcp-server/test/resources.test.ts index d4a16c5..10fb1b7 100644 --- a/packages/mcp-server/test/resources.test.ts +++ b/packages/mcp-server/test/resources.test.ts @@ -245,6 +245,36 @@ describe("resource reads", () => { expect(text).not.toContain(PLANTED_TOKEN); }); + it("caps an oversized log to its trailing bytes instead of loading it all", async () => { + const bigPath = path.join( + dirs.projectDir, + ".picklab", + "runs", + RUN_ID, + "logs", + "big.log", + ); + const filler = "x".repeat(1024); + const lines: string[] = []; + for (let i = 0; i < 1200; i += 1) { + lines.push(`line-${i}-${filler}`); + } + lines.push(`tail-marker Authorization: Bearer ${PLANTED_TOKEN}`); + fs.writeFileSync(bigPath, `${lines.join("\n")}\n`); + expect(fs.statSync(bigPath).size).toBeGreaterThan(1024 * 1024); + + const { contents } = await lab.client.readResource({ + uri: `picklab://runs/${RUN_ID}/logs/big.log`, + }); + const text = first(contents).text as string; + expect(text).toContain("[truncated:"); + expect(text).toContain("tail-marker"); + expect(text).toContain("[REDACTED]"); + expect(text).not.toContain(PLANTED_TOKEN); + expect(text).not.toContain("line-0-"); + expect(text.length).toBeLessThan(1024 * 1024 + 1024); + }); + it("reads a session status as JSON", async () => { const { contents } = await lab.client.readResource({ uri: `picklab://sessions/${sessionId}/status`,