From 6c6c05dccbb3a6580498e83b5d303ac8bf1a3d85 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 30 Sep 2026 03:59:20 -0500 Subject: [PATCH] Retain unchanged agent files with their warm native owner Probe complete materializations in bounded children, atomically transfer current-run collection authority, and preserve the exact agent home across warm turns. Retire before collecting changed or uncertain copies, fence stale claims and cleanup, and retain unresolved retirement ownership. Co-Authored-By: Paperclip --- .../__tests__/agent-directory-probe.test.ts | 79 +++++++++ .../agent-directory-working-copies.test.ts | 156 ++++++++++++++++++ server/src/services/agent-directory-probe.ts | 115 +++++++++++++ .../agent-directory-working-copies.ts | 91 ++++++++-- .../agent-instruction-working-copies.ts | 20 ++- server/src/services/heartbeat.ts | 36 +++- .../native-session-executor.test.ts | 140 ++++++++++++++++ .../native-runtime/native-session-executor.ts | 149 ++++++++++++++--- 8 files changed, 736 insertions(+), 50 deletions(-) create mode 100644 server/src/__tests__/agent-directory-probe.test.ts create mode 100644 server/src/services/agent-directory-probe.ts diff --git a/server/src/__tests__/agent-directory-probe.test.ts b/server/src/__tests__/agent-directory-probe.test.ts new file mode 100644 index 00000000000..59ee14b1608 --- /dev/null +++ b/server/src/__tests__/agent-directory-probe.test.ts @@ -0,0 +1,79 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import fs from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import childProcess, { execFileSync } from "node:child_process"; +import { captureDirectorySnapshot } from "@paperclipai/adapter-utils/workspace-restore-merge"; +import { agentDirectoryBaselineDigest, agentDirectoryProbeProgram, observeLocalAgentDirectory, probeAgentDirectory } from "../services/agent-directory-probe.js"; + +describe("stable live agent directory observation", () => { + let root: string; + beforeEach(() => { + root = fs.realpathSync(fs.mkdtempSync(join(tmpdir(), "agent-directory-probe-"))); + fs.mkdirSync(join(root, "notes")); + fs.writeFileSync(join(root, "AGENTS.md"), "Instructions\n"); + fs.writeFileSync(join(root, "notes", "memory.bin"), Buffer.from([0, 1, 255])); + }); + afterEach(() => { vi.restoreAllMocks(); fs.rmSync(root, { recursive: true, force: true }); }); + it("matches the complete materialized baseline and the actual remote Node program", async () => { + const baseline = await captureDirectorySnapshot(root); + const observed = probeAgentDirectory(root); + expect(observed.digest).toBe(agentDirectoryBaselineDigest(baseline)); + const remote = JSON.parse(execFileSync(process.execPath, ["-e", agentDirectoryProbeProgram, root], { encoding: "utf8", env: {}, timeout: 10_000 })); + expect(remote).toEqual(observed); + }); + it("runs the local observation in its bounded credential-free child", async () => { + expect(await observeLocalAgentDirectory(root)).toEqual(probeAgentDirectory(root)); + }); + it.each(["timeout", "oversized", "malformed"])("rejects %s local-child observations", async kind => { + const stub = vi.spyOn(childProcess, "execFile").mockImplementation(((file: string, args: string[], options: object, callback: (error: Error | null, stdout: string, stderr: string) => void) => { + expect(file).toBe(process.execPath); + expect(args).toEqual(["-e", agentDirectoryProbeProgram, root]); + expect(options).toMatchObject({ cwd: root, env: {}, timeout: 10_000, maxBuffer: 1024, killSignal: "SIGKILL" }); + callback(kind === "timeout" ? Object.assign(new Error("timed out"), { killed: true, signal: "SIGKILL" }) : null, + kind === "oversized" ? "x".repeat(1025) : "{}", ""); + return {} as childProcess.ChildProcess; + }) as typeof childProcess.execFile); + await expect(observeLocalAgentDirectory(root)).rejects.toThrow(); + expect(stub).toHaveBeenCalledOnce(); + }); + it.each(["edit", "add", "remove", "mode"])("detects a legitimate %s", change => { + const before = probeAgentDirectory(root); + if (change === "edit") fs.writeFileSync(join(root, "AGENTS.md"), "New instructions\n"); + if (change === "add") fs.writeFileSync(join(root, "new.txt"), "New"); + if (change === "remove") fs.unlinkSync(join(root, "notes", "memory.bin")); + if (change === "mode") fs.chmodSync(join(root, "AGENTS.md"), 0o700); + expect(probeAgentDirectory(root).digest).not.toBe(before.digest); + }); + it.each(["symlink", "hardlink", "reserved"])("rejects %s entries", kind => { + if (kind === "symlink") fs.symlinkSync(join(root, "AGENTS.md"), join(root, "unsafe")); + if (kind === "hardlink") fs.linkSync(join(root, "AGENTS.md"), join(root, "unsafe")); + if (kind === "reserved") fs.mkdirSync(join(root, ".paperclip-runtime")); + expect(() => probeAgentDirectory(root)).toThrow(); + }); + it("detects file mutation during a bounded read", () => { + const read = fs.readSync; + let changed = false; + vi.spyOn(fs, "readSync").mockImplementation(((...args: Parameters) => { + const result = Reflect.apply(read, fs, args); + if (!changed) { changed = true; fs.appendFileSync(join(root, "AGENTS.md"), "Concurrent change"); } + return result; + }) as typeof read); + expect(() => probeAgentDirectory(root)).toThrow(/changed|grew/); + }); + it("fails on unreadable observations instead of reporting unchanged", () => { + vi.spyOn(fs, "openSync").mockImplementation(() => { throw Object.assign(new Error("denied"), { code: "EACCES" }); }); + expect(() => probeAgentDirectory(root)).toThrow("denied"); + }); + it("does not let matching replacement bytes preserve the pinned root identity", () => { + const before = probeAgentDirectory(root); + const retired = `${root}-retired`; + fs.renameSync(root, retired); + try { fs.cpSync(retired, root, { recursive: true }); expect(probeAgentDirectory(root).identity).not.toBe(before.identity); } + finally { fs.rmSync(retired, { recursive: true, force: true }); } + }); + it("rejects excluded/incomplete baselines", async () => { + const baseline = await captureDirectorySnapshot(root); + expect(() => agentDirectoryBaselineDigest({ ...baseline, exclude: ["notes"] })).toThrow("Incomplete"); + }); +}); diff --git a/server/src/__tests__/agent-directory-working-copies.test.ts b/server/src/__tests__/agent-directory-working-copies.test.ts index 17d7acdb882..8c113acf928 100644 --- a/server/src/__tests__/agent-directory-working-copies.test.ts +++ b/server/src/__tests__/agent-directory-working-copies.test.ts @@ -12,6 +12,7 @@ import { eq } from "drizzle-orm"; import { agentFileStore, fileHash, inspectAgentFile, snapshotAgentFiles, MAX_AGENT_FILE_BYTES, MAX_AGENT_DIRECTORY_BYTES, MAX_AGENT_DIRECTORY_ENTRIES } from "../services/agent-file-store.js"; import { agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, environmentLeases, environments, agentInstructionWorkingCopies, agentInstructionRevisions, agentInstructionHeads, createDb } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { agentDirectoryWorkingCopyService } from "../services/agent-directory-working-copies.js"; import { agentInstructionRevisionService } from "../services/agent-instruction-revisions.js"; import { agentInstructionWorkingCopyService, instructionWorkingCopyGuidance } from "../services/agent-instruction-working-copies.js"; import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js"; @@ -71,6 +72,161 @@ describe("persistent agent directories", () => { await fs.writeFile(path.join(root, entryFile), initial); }); + describe("warm directory ownership", () => { + async function nextRun() { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", responsibleUserId: userId }); + return runId; + } + it("adopts one physical root and fences stale cleanup through final save", async () => { + const first = await run(); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false); + const secondId = await nextRun(); + const second = await copies.adopt({ companyId, agentId, runId: secondId, previousRunId: first.runId, cwd: home }); + expect(second).toMatchObject({ runId: secondId, localRoot: first.localRoot, executionRoot: first.executionRoot, + responsibleUserId: userId, receipt: { materializationRunId: first.runId } }); + expect((await copies.get(companyId, first.runId))?.receipt?.retainedByRunId).toBe(secondId); + await copies.release(companyId, first.runId); + expect((await copies.collectStopped({ companyId, runId: first.runId }))?.processStoppedAt).toBeNull(); + await copies.reportUnavailable(companyId, first.runId); + expect(await fs.readFile(path.join(first.localRoot, entryFile), "utf8")).toBe(initial); + await copies.release(companyId, secondId); + expect(await copies.hasChanges({ companyId, runId: secondId })).toBe(false); + const thirdId = await nextRun(); + const third = await copies.adopt({ companyId, agentId, runId: thirdId, previousRunId: secondId, cwd: home }); + expect(third?.executionRoot).toBe(first.executionRoot); + await fs.writeFile(path.join(first.localRoot, "warm-memory.txt"), "third turn edit"); + expect(await copies.hasChanges({ companyId, runId: thirdId })).toBe(true); + expect(await copies.collectStopped({ companyId, runId: thirdId })).toMatchObject({ state: "saved", receipt: { cleanupPending: false } }); + expect(await fs.readFile(path.join(root, "warm-memory.txt"), "utf8")).toBe("third turn edit"); + await expect(fs.stat(first.localRoot)).rejects.toMatchObject({ code: "ENOENT" }); + await copies.release(companyId, first.runId); + }); + it.each(["company", "agent", "workspace", "environment"])("rejects foreign %s adoption", async mismatch => { + const first = await run(); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false); + const secondId = await nextRun(); + expect(await copies.adopt({ companyId: mismatch === "company" ? randomUUID() : companyId, + agentId: mismatch === "agent" ? randomUUID() : agentId, runId: secondId, previousRunId: first.runId, + cwd: mismatch === "workspace" ? `${home}-other` : home, + ...(mismatch === "environment" ? { target: { kind: "remote" as const, transport: "sandbox" as const, environmentId: randomUUID(), remoteCwd: "/workspace" } } : {}), + })).toBeNull(); + expect((await copies.get(companyId, first.runId))?.receipt?.retainedByRunId).toBeUndefined(); + await copies.collectStopped({ companyId, runId: first.runId }); + }); + it("allows only one stale concurrent adoption with identical updatedAt timestamps", async () => { + const fixed = new Date("2030-01-01T00:00:00.000Z"); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(fixed); + try { + const first = await run(); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false); + const next = await Promise.all([nextRun(), nextRun()]); + let reads = 0; + let releaseReads!: () => void; + const bothRead = new Promise(resolve => { releaseReads = resolve; }); + // Both claimants see the same unclaimed row before either can enter + // the agent-row transaction. Date precision cannot distinguish them. + const staleReader = async () => { + const captured = await copies.get(companyId, first.runId); + reads++; + if (reads >= 3) { if (reads === 4) releaseReads(); await bothRead; } + return captured; + }; + const service = agentDirectoryWorkingCopyService(db, staleReader, async (row, values) => { + const [updated] = await db.update(agentInstructionWorkingCopies).set({ ...values, updatedAt: fixed }) + .where(eq(agentInstructionWorkingCopies.runId, row.runId)).returning(); + return updated; + }); + const settled = await Promise.allSettled(next.map(runId => service.adopt({ companyId, agentId, runId, previousRunId: first.runId, cwd: home }))); + const winners = settled.filter(result => result.status === "fulfilled" && result.value); + expect(winners).toHaveLength(1); + expect(settled.filter(result => result.status === "rejected")).toHaveLength(1); + const prior = await copies.get(companyId, first.runId); + expect(prior?.updatedAt).toEqual(fixed); + const winnerId = prior?.receipt?.retainedByRunId as string; + expect(next).toContain(winnerId); + expect(await copies.get(companyId, next.find(id => id !== winnerId)!)).toBeNull(); + await copies.collectStopped({ companyId, runId: winnerId }); + } finally { vi.useRealTimers(); } + }); + it("fences a stale collector when adoption commits immediately before its stop CAS", async () => { + const first = await run(); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false); + const secondId = await nextRun(); + const update = db.update.bind(db); + let injected = false; + const spy = vi.spyOn(db, "update").mockImplementation(table => { + const builder = update(table); + if (table !== agentInstructionWorkingCopies) return builder; + const set = builder.set.bind(builder); + builder.set = (values => { + const query = set(values); + if (!("processStoppedAt" in values) || !values.processStoppedAt || injected) return query; + const returning = query.returning.bind(query); + query.returning = (async () => { + injected = true; + expect(await copies.adopt({ companyId, agentId, runId: secondId, previousRunId: first.runId, cwd: home })).not.toBeNull(); + return returning(); + }) as typeof query.returning; + return query; + }) as typeof builder.set; + return builder; + }); + try { + expect(await copies.collectStopped({ companyId, runId: first.runId })).toMatchObject({ + receipt: { retainedByRunId: secondId }, processStoppedAt: null, attempts: 0, + }); + expect(injected).toBe(true); + expect(await fs.readFile(path.join(first.localRoot, entryFile), "utf8")).toBe(initial); + expect((await copies.get(companyId, secondId))?.processStoppedAt).toBeNull(); + } finally { spy.mockRestore(); } + await copies.collectStopped({ companyId, runId: secondId }); + }); + it("requires current-run authorization before transferring collection authority", async () => { + const first = await run(); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false); + const secondId = await nextRun(); + await db.update(heartbeatRuns).set({ responsibleUserId: null }).where(eq(heartbeatRuns.id, secondId)); + await expect(copies.adopt({ companyId, agentId, runId: secondId, previousRunId: first.runId, cwd: home })).rejects.toMatchObject({ status: 403 }); + expect((await copies.get(companyId, first.runId))?.receipt?.retainedByRunId).toBeUndefined(); + await copies.collectStopped({ companyId, runId: first.runId }); + }); + it("durably preserves an adopted directory after retirement failure", async () => { + const first = await run(); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false); + const secondId = await nextRun(); + await copies.adopt({ companyId, agentId, runId: secondId, previousRunId: first.runId, cwd: home }); + await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native" }).where(eq(heartbeatRuns.id, secondId)); + expect(await copies.reportRetirementUnconfirmed(companyId, secondId)).toMatchObject({ + state: "pending_collection", errorCode: "INSTRUCTION_STOP_UNCONFIRMED", processStoppedAt: null, + }); + await copies.reportRetirementUnconfirmed(companyId, first.runId); + await copies.release(companyId, first.runId); + await copies.release(companyId, secondId); + await copies.recoverStopped(); // A terminal run with no stop receipt grants no collection authority. + expect((await copies.get(companyId, secondId))?.processStoppedAt).toBeNull(); + expect(await fs.readFile(path.join(first.localRoot, entryFile), "utf8")).toBe(initial); + await copies.collectStopped({ companyId, runId: secondId }); // Explicit test owner retirement. + }); + it("declines reuse after canonical instruction changes", async () => { + const first = await run(); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false); + await fs.writeFile(path.join(root, entryFile), "Changed canonical instructions"); + const secondId = await nextRun(); + expect(await copies.adopt({ companyId, agentId, runId: secondId, previousRunId: first.runId, cwd: home })).toBeNull(); + await copies.collectStopped({ companyId, runId: first.runId }); + }); + it("requires stop for matching bytes at a replaced materialized root", async () => { + const first = await run(); + const replaced = `${first.localRoot}-old`; + await fs.rename(first.localRoot, replaced); + await fs.cp(replaced, first.localRoot, { recursive: true }); + expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(true); + await copies.collectStopped({ companyId, runId: first.runId }); + }); + }); + it.each([".paperclip-runtime/state", "notes/.paperclip-runtime/state", "promptTemplate.legacy.md"])("rejects reserved board path %s before mutation", async (reserved) => { await expect(agentFileStore(db).write({ ...target(), path: reserved, bytes: Buffer.from("reserved"), baseHash: null }, board())).rejects.toMatchObject({ status: 422 }); await expect(fs.stat(path.join(root, reserved))).rejects.toMatchObject({ code: "ENOENT" }); diff --git a/server/src/services/agent-directory-probe.ts b/server/src/services/agent-directory-probe.ts new file mode 100644 index 00000000000..f9ee172827e --- /dev/null +++ b/server/src/services/agent-directory-probe.ts @@ -0,0 +1,115 @@ +import { createHash } from "node:crypto"; +import fs from "node:fs"; +import path from "node:path"; +import childProcess from "node:child_process"; +import type { DirectorySnapshot } from "@paperclipai/adapter-utils/workspace-restore-merge"; + +/** Only for a live unchanged-turn observation. Never a save or stop receipt. */ +export function agentDirectoryBaselineDigest(snapshot: DirectorySnapshot): string { + if (snapshot.exclude.length || snapshot.ignoredPaths) throw new Error("Incomplete agent directory baseline"); + const entries = [...snapshot.entries].sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0); + return createHash("sha256").update(JSON.stringify(entries)).digest("hex"); +} + +// This self-contained function also runs inside the original remote target. +// Keep all dependencies explicit so remote observation never reads a local mirror. +export function probeAgentDirectory( + root: string, + modules = { fs, path, createHash }, +): { digest: string; identity: string } { + const { fs, path, createHash } = modules; + const deadline = Date.now() + 5_000; + const checkTime = () => { if (Date.now() > deadline) throw new Error("Agent directory probe deadline"); }; + const stamp = (s: fs.BigIntStats) => [s.dev, s.ino, s.mode, s.nlink, s.size, s.mtimeNs, s.ctimeNs].join(":"); + const identity = (s: fs.BigIntStats) => [s.dev, s.ino, s.birthtimeNs].join(":"); + if (!path.isAbsolute(root) || path.resolve(root) !== root || root.includes("\0")) throw new Error("Invalid agent directory root"); + // Fixed macOS system aliases are resolved by the caller, not followed here. + const ancestors = new Map(); + let ancestor = path.parse(root).root; + for (const segment of root.slice(ancestor.length).split(path.sep)) { + ancestor = path.join(ancestor, segment); + const s = fs.lstatSync(ancestor, { bigint: true }); + if (!s.isDirectory() || s.isSymbolicLink()) throw new Error("Unsafe agent directory ancestor"); + ancestors.set(ancestor, identity(s)); + } + const metadata = new Map(); + const entries: Array<[string, { kind: "dir" } | { kind: "file"; mode: number; hash: string }]> = []; + let total = 0; + function walk(relative: string, verify: boolean) { + checkTime(); + const absolute = path.join(root, relative); + const before = fs.lstatSync(absolute, { bigint: true }); + if (!before.isDirectory() || before.isSymbolicLink()) throw new Error("Unsafe agent directory"); + const names = fs.readdirSync(absolute).sort(); + if (verify) { + if (metadata.get(relative) !== stamp(before)) throw new Error("Agent directory changed during probe"); + } else metadata.set(relative, stamp(before)); + for (const name of names) { + checkTime(); + const next = relative ? `${relative}/${name}` : name; + if (next.length > 512 || name === "." || name === ".." || name.includes("\\") || name.includes("\0") + || name === ".paperclip-runtime" || next === "promptTemplate.legacy.md") throw new Error("Unsafe agent file path"); + const filename = path.join(root, next); + const s = fs.lstatSync(filename, { bigint: true }); + if (s.isDirectory()) { + if (!verify) entries.push([next, { kind: "dir" }]); + walk(next, verify); + } else { + if (!s.isFile() || s.nlink !== 1n || s.size > 256n * 1024n * 1024n) throw new Error("Unsafe or oversized agent file"); + if (verify) { + if (metadata.get(next) !== stamp(s)) throw new Error("Agent file changed during probe"); + } else { + total += Number(s.size); + if (total > 2 * 1024 * 1024 * 1024) throw new Error("Agent directory probe byte limit"); + const fd = fs.openSync(filename, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW); + try { + if (stamp(fs.fstatSync(fd, { bigint: true })) !== stamp(s)) throw new Error("Agent file replaced during open"); + const hash = createHash("sha256"), buffer = Buffer.alloc(64 * 1024); + let size = 0; + for (;;) { + checkTime(); + const count = fs.readSync(fd, buffer, 0, buffer.length, null); + if (!count) break; + size += count; + if (size > Number(s.size)) throw new Error("Agent file grew during probe"); + hash.update(buffer.subarray(0, count)); + } + if (size !== Number(s.size) || stamp(fs.fstatSync(fd, { bigint: true })) !== stamp(s) + || stamp(fs.lstatSync(filename, { bigint: true })) !== stamp(s)) throw new Error("Agent file changed during read"); + metadata.set(next, stamp(s)); + entries.push([next, { kind: "file", mode: Number(s.mode), hash: hash.digest("hex") }]); + } finally { fs.closeSync(fd); } + } + } + if (entries.length > 100_000) throw new Error("Agent directory probe entry limit"); + } + if (stamp(fs.lstatSync(absolute, { bigint: true })) !== stamp(before) + || JSON.stringify(fs.readdirSync(absolute).sort()) !== JSON.stringify(names)) throw new Error("Agent directory membership changed"); + } + const initial = fs.lstatSync(root, { bigint: true }); + walk("", false); + walk("", true); + if (stamp(fs.lstatSync(root, { bigint: true })) !== stamp(initial)) throw new Error("Agent directory root changed"); + for (const [ancestor, expected] of ancestors) { + const current = fs.lstatSync(ancestor, { bigint: true }); + if (!current.isDirectory() || current.isSymbolicLink() || identity(current) !== expected) throw new Error("Agent directory ancestor changed"); + } + entries.sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0); + return { digest: createHash("sha256").update(JSON.stringify(entries)).digest("hex"), identity: identity(initial) }; +} + +export const agentDirectoryProbeProgram = `const probe=${probeAgentDirectory.toString()}; process.stdout.write(JSON.stringify(probe(process.argv[1], {fs:require('node:fs'),path:require('node:path'),createHash:require('node:crypto').createHash})))`; + +/** Bound child: no shell, inherited credentials, or event-loop hashing. */ +export async function observeLocalAgentDirectory(root: string) { + const result = await new Promise<{ stdout: string; stderr: string }>((resolve, reject) => { + childProcess.execFile(process.execPath, ["-e", agentDirectoryProbeProgram, root], { + cwd: root, env: {}, timeout: 10_000, maxBuffer: 1024, killSignal: "SIGKILL", encoding: "utf8", + }, (error, stdout, stderr) => { if (error) reject(error); else resolve({ stdout, stderr }); }); + }); + if (result.stdout.length > 1024) throw new Error("Agent directory observation exceeds bound"); + const value = JSON.parse(result.stdout); + if (!value || typeof value.digest !== "string" || !/^[a-f0-9]{64}$/.test(value.digest) + || typeof value.identity !== "string" || !/^[0-9]+:[0-9]+:[0-9]+$/.test(value.identity)) throw new Error("Invalid agent directory observation"); + return value as { digest: string; identity: string }; +} diff --git a/server/src/services/agent-directory-working-copies.ts b/server/src/services/agent-directory-working-copies.ts index ab3dde2e112..007ee0ba576 100644 --- a/server/src/services/agent-directory-working-copies.ts +++ b/server/src/services/agent-directory-working-copies.ts @@ -1,6 +1,6 @@ import fs from "node:fs/promises"; import path from "node:path"; -import { and, eq } from "drizzle-orm"; +import { and, eq, isNull, sql } from "drizzle-orm"; import { agents, environmentLeases, environments, agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db"; import { syncDirectoryToSsh, restoreWorkspaceFromSshExecution } from "@paperclipai/adapter-utils/ssh"; import { prepareAdapterExecutionTargetRuntime, runAdapterExecutionTargetShellCommand, type AdapterExecutionTarget, type PreparedAdapterExecutionTargetRuntime } from "@paperclipai/adapter-utils/execution-target"; @@ -13,6 +13,7 @@ import { HttpError, conflict, notFound } from "../errors.js"; import type { AuthorizationActor } from "./authorization.js"; import type { EnvironmentRuntimeService } from "./environment-runtime.js"; import type { Environment, EnvironmentLease } from "@paperclipai/shared"; +import { agentDirectoryBaselineDigest, agentDirectoryProbeProgram, observeLocalAgentDirectory } from "./agent-directory-probe.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; type Copy = typeof copies.$inferSelect; @@ -50,21 +51,43 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string workspaceBaseline: baseline(row), workspaceGitSnapshot: null, workspaceFileMode: "all", workspaceExclude: [".paperclip-runtime", ".paperclip-runtime/**"] }); } + async function observe(row: Copy, target?: AdapterExecutionTarget | null) { + if (row.location === "local") { + const real = await fs.realpath(row.localRoot); + if (real !== row.localRoot && !(process.platform === "darwin" && real === `/private${row.localRoot}` && /^\/(tmp|var|etc)\//.test(row.localRoot))) throw new Error("Agent directory root changed"); + // A bounded child keeps hashing off the controller event loop and inherits + // no provider credentials. execFile terminates it on timeout. + return observeLocalAgentDirectory(real); + } + if (!target || target.kind !== "remote" || row.location !== `remote:${target.environmentId ?? ""}`) throw new Error("Agent directory environment changed"); + const origin = typeof row.receipt?.materializationRunId === "string" ? row.receipt.materializationRunId : row.runId; + if (row.executionRoot !== path.posix.join(target.remoteCwd, ".paperclip-runtime", "agent-files", row.agentId, origin)) throw new Error("Agent directory root changed"); + const quote = (value: string) => `'${value.replaceAll("'", `'"'"'`)}'`; + const result = await runAdapterExecutionTargetShellCommand(row.runId, target, + `node -e ${quote(agentDirectoryProbeProgram)} ${quote(row.executionRoot)}`, { cwd: target.remoteCwd, env: {}, timeoutSec: 10 }); + if (result.exitCode !== 0 || result.timedOut || result.stdout.length > 1024) throw new Error("Agent directory observation unavailable"); + const value = JSON.parse(result.stdout); + if (!value || typeof value.digest !== "string" || !/^[a-f0-9]{64}$/.test(value.digest) || typeof value.identity !== "string") throw new Error("Invalid agent directory observation"); + return value as { digest: string; identity: string }; + } async function prepare(input: { companyId: string; agentId: string; runId: string; target?: AdapterExecutionTarget | null; cwd: string }) { const [agent] = await db.select().from(agents).where(and(eq(agents.id, input.agentId), eq(agents.companyId, input.companyId))); if (!agent) throw notFound("Agent not found"); if (agentInstructionsBundleMode(agent) !== "managed") return null; const bound = await resolveInstructionActor(db, { type: "agent", companyId: input.companyId, agentId: input.agentId, runId: input.runId }); const root = resolveManagedInstructionsRoot(agent); - const localRoot = path.join(path.dirname(root), "file-sync", "runs", input.runId, "live"); + let row = await get(input.companyId, input.runId); + if (row?.receipt?.retainedByRunId) throw conflict("Agent directory ownership passed to another run"); + const origin = typeof row?.receipt?.materializationRunId === "string" ? row.receipt.materializationRunId : input.runId; + if (!/^[a-zA-Z0-9_-]{1,128}$/.test(origin)) throw conflict("Invalid agent directory owner"); + const localRoot = path.join(path.dirname(root), "file-sync", "runs", origin, "live"); // Local copies live outside the task cwd. Remote copies use the reserved, // excluded runtime area inside the provider's confined workspace. They are // synchronized independently; provider HOME is never reinterpreted. const executionRoot = input.target?.kind === "remote" - ? path.posix.join(input.target.remoteCwd, ".paperclip-runtime", "agent-files", input.agentId, input.runId) + ? path.posix.join(input.target.remoteCwd, ".paperclip-runtime", "agent-files", input.agentId, origin) : localRoot; const location = input.target?.kind === "remote" ? `remote:${input.target.environmentId ?? ""}` : "local"; - let row = await get(input.companyId, input.runId); if (row && (row.localRoot !== localRoot || row.executionRoot !== executionRoot || row.agentId !== input.agentId || row.location !== location)) throw conflict("Agent directory belongs to a different execution environment"); if (row && !completed.has(row.state) && row.state !== "preparing") { if (input.target?.kind === "remote" && !transports.has(key(row))) transports.set(key(row), await transport(row, input.target, true)); @@ -74,7 +97,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string // leaves a recoverable preparing receipt instead of an orphan directory. const preparing = { entryFile: deriveBundleState(agent).entryFile, baseRevisionId: null, baseHash: "preparing", localRoot, executionRoot, location, state: "preparing", candidateBase64: null, candidateHash: null, - receipt: { schema: AGENT_FILES_CONTRACT, ...(input.target?.kind === "remote" + receipt: { schema: AGENT_FILES_CONTRACT, workspaceCwd: input.cwd, materializationRunId: origin, ...(input.target?.kind === "remote" ? { cleanup: { leaseId: input.target.leaseId ?? null, remoteCwd: input.target.remoteCwd } } : {}) }, processStoppedAt: null, attempts: 0, nextAttemptAt: null, errorCode: null, errorMessage: null }; if (row) row = await patch(row, preparing); @@ -121,7 +144,9 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string if (excluded.exitCode !== 0 || excluded.timedOut) throw new Error("Could not exclude agent files from task Git staging"); transports.set(key(row), await transport(row, input.target, false)); } - return await patch(row, { state: "prepared" }); + const observed = await observe(row, input.target).catch(() => null); + return await patch(row, { state: "prepared", receipt: { ...row.receipt, + ...(observed?.digest === agentDirectoryBaselineDigest(snapshot) ? { materializationIdentity: observed.identity } : {}) } }); } catch (error) { // Preparation failed before a provider could start using this copy. row = await patch(row, { state: "unavailable", processStoppedAt: new Date(), errorCode: "AGENT_FILES_PREPARE_FAILED", @@ -143,16 +168,51 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string } return inspectAgentDirectory(row.localRoot); } - async function hasChanges(_row: Copy, _target?: AdapterExecutionTarget | null) { - // A whole-directory collector needs a quiescent provider, including its - // child processes. Checkpoint and close at the turn boundary before reading - // either local or remote files. The provider conversation remains resumable; - // ordinary file edits do not change the session configuration digest. - return true; + async function hasChanges(row: Copy, target?: AdapterExecutionTarget | null) { + try { + if (row.receipt?.retainedByRunId || typeof row.receipt?.materializationIdentity !== "string") return true; + const observed = await observe(row, target); + if (observed.identity !== row.receipt.materializationIdentity || observed.digest !== agentDirectoryBaselineDigest(baseline(row))) return true; + const unchanged = await patch(row, { state: "unchanged_turn", nextAttemptAt: null }); + return unchanged.state !== "unchanged_turn" || Boolean(unchanged.receipt?.retainedByRunId) || Boolean(unchanged.processStoppedAt); + } catch { return true; } + } + /** Rebind one physical materialization; earlier runs lose collection authority. */ + async function adopt(input: { companyId: string; agentId: string; runId: string; previousRunId: string; target?: AdapterExecutionTarget | null; cwd: string }) { + let prior = await get(input.companyId, input.previousRunId); + if (!prior || prior.agentId !== input.agentId || prior.state !== "unchanged_turn" || prior.receipt?.retainedByRunId || prior.processStoppedAt + || prior.receipt?.workspaceCwd !== input.cwd || prior.location !== (input.target?.kind === "remote" ? `remote:${input.target.environmentId ?? ""}` : "local")) return null; + if (await hasChanges(prior, input.target)) return null; + prior = (await get(input.companyId, input.previousRunId))!; + const bound = await resolveInstructionActor(db, { type: "agent", companyId: input.companyId, agentId: input.agentId, runId: input.runId }); + const adopted = await store.locked(input.companyId, input.agentId, bound, false, async (tx, _agent, canonical) => { + const current = await inspectAgentDirectory(canonical); + if (agentDirectoryBaselineDigest(current.snapshot) !== agentDirectoryBaselineDigest(baseline(prior))) return null; + const [claimed] = await tx.update(copies).set({ receipt: { ...prior.receipt, retainedByRunId: input.runId }, updatedAt: new Date() }) + .where(and(eq(copies.companyId, input.companyId), eq(copies.runId, prior.runId), eq(copies.agentId, input.agentId), eq(copies.state, "unchanged_turn"), + eq(copies.updatedAt, prior.updatedAt), eq(copies.baseHash, prior.baseHash), eq(copies.localRoot, prior.localRoot), + isNull(copies.processStoppedAt), sql`${copies.receipt}->>'retainedByRunId' IS NULL`)).returning(); + if (!claimed) throw conflict("Agent directory ownership changed"); + const { createdAt: _createdAt, updatedAt: _updatedAt, ...values } = prior; + const [next] = await tx.insert(copies).values({ ...values, runId: input.runId, responsibleUserId: bound.onBehalfOfUserId!, state: "prepared", + receipt: { ...prior.receipt, materializationRunId: prior.receipt?.materializationRunId ?? prior.runId }, updatedAt: new Date(), + }).returning(); + return next; + }); + // Do not move the live transport until the database ownership commit succeeds. + if (adopted) { + const runtime = transports.get(key(prior)); + if (runtime) { transports.delete(key(prior)); transports.set(key(adopted), runtime); } + } + return adopted; } async function collectStopped(row: Copy, target?: AdapterExecutionTarget | null) { + if (row.receipt?.retainedByRunId) return row; if (completed.has(row.state)) { await release(row); return (await get(row.companyId, row.runId))!; } row = await patch(row, { processStoppedAt: row.processStoppedAt ?? new Date() }); + // Adoption may win between the initial read and the stop CAS. The returned + // current row is authoritative; a transferred alias cannot inspect or save. + if (row.receipt?.retainedByRunId || !row.processStoppedAt) return row; // The stopped working copy is the only temporary tree. Retry transient I/O // at this boundary, then clean it up; there are no preserved run snapshots. let inspected: Awaited> | undefined; @@ -188,6 +248,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string return (await get(row.companyId, row.runId))!; } async function release(row: Copy, target?: AdapterExecutionTarget | null) { + if (row.receipt?.retainedByRunId || !row.processStoppedAt) return; // Collection and environment teardown can both release the same copy. // A compact successful cleanup receipt is final, even after restart. if (completed.has(row.state) && row.processStoppedAt && row.receipt?.cleanupPending === false) return; @@ -197,7 +258,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string await runtime?.cleanupWorkspaceSnapshot?.().catch(() => { cleanupPending = true; }); const cleanupTarget = runtime?.target ?? target; if (row.processStoppedAt && completed.has(row.state) && cleanupTarget?.kind === "remote") { - const expected = path.posix.join(cleanupTarget.remoteCwd, ".paperclip-runtime", "agent-files", row.agentId, row.runId); + const expected = path.posix.join(cleanupTarget.remoteCwd, ".paperclip-runtime", "agent-files", row.agentId, typeof row.receipt?.materializationRunId === "string" ? row.receipt.materializationRunId : row.runId); if (row.executionRoot !== expected) throw new Error("Agent directory cleanup path changed"); const quoted = `'${expected.replaceAll("'", `'"'"'`)}'`; const remoteCleanupFailed = await runAdapterExecutionTargetShellCommand(row.runId, cleanupTarget, `rm -rf -- ${quoted}`, @@ -237,7 +298,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string const [environment] = await db.select().from(environments).where(eq(environments.id, lease.environmentId)); if (!environment) return false; const remoteCwd = cleanup?.remoteCwd ?? lease.metadata?.remoteCwd; - if (typeof remoteCwd !== "string" || row.executionRoot !== path.posix.join(remoteCwd, ".paperclip-runtime", "agent-files", row.agentId, row.runId)) return false; + if (typeof remoteCwd !== "string" || row.executionRoot !== path.posix.join(remoteCwd, ".paperclip-runtime", "agent-files", row.agentId, typeof row.receipt?.materializationRunId === "string" ? row.receipt.materializationRunId : row.runId)) return false; const result = await environmentRuntime.execute({ environment: environment as Environment, lease: lease as EnvironmentLease, command: "rm", args: ["-rf", "--", row.executionRoot], cwd: remoteCwd, env: {}, timeoutMs: 15_000, bypassSession: true }); return result.exitCode === 0 && !result.timedOut; @@ -251,7 +312,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string return fn(current); }); } - return { prepare, hasChanges, + return { prepare, hasChanges, adopt, collectStopped: (row: Copy, target?: AdapterExecutionTarget | null) => serial(row, current => collectStopped(current, target)), release: (row: Copy) => serial(row, release), }; diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts index 7611dcb614f..6bb2576f85d 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -62,6 +62,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen async function patch(row: Copy, values: Partial) { const [updated] = await db.update(copies).set({ ...values, updatedAt: new Date() }).where(and( scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash), + sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`, )).returning(); // A late cleanup must not overwrite an explicit resolution or a newer // baseline. Canonical content has its own independent head CAS. @@ -294,8 +295,9 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen async function recoverStopped() { const pending = await db.select({ copy: copies, runtimeMode: heartbeatRuns.runtimeMode }).from(copies) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, copies.companyId), eq(heartbeatRuns.id, copies.runId))) - .where(and(or(inArray(copies.state, ["prepared", "pending_collection"]), + .where(and(or(inArray(copies.state, ["prepared", "pending_collection", "unchanged_turn"]), and(eq(copies.state, "preparing"), sql`${copies.receipt}->>'schema' = 'paperclip.agent-files.v1'`)), + sql`NOT (coalesce(${copies.receipt}, '{}'::jsonb) ? 'retainedByRunId')`, inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]), lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1))).limit(20); for (const { copy: row, runtimeMode } of pending) { @@ -352,5 +354,19 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen } async function release(companyId: string, runId: string) { liveTargets.delete(targetKey(companyId, runId)); const row = await get(companyId, runId); if (row && isAgentDirectoryCopy(row)) await directories.release(row); } - return { prepare, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release }; + async function reportRetirementUnconfirmed(companyId: string, runId: string) { + const row = await get(companyId, runId); + if (!row || !isAgentDirectoryCopy(row) || row.receipt?.retainedByRunId || row.processStoppedAt) return row; + return patch(row, { state: "pending_collection", errorCode: "INSTRUCTION_STOP_UNCONFIRMED", + errorMessage: "The retained provider did not confirm retirement. Its registered agent directory is preserved; no save is claimed.", nextAttemptAt: null }); + } + async function adopt(input: Parameters[0]) { + const row = await directories.adopt(input); + if (row) { + liveTargets.delete(targetKey(input.companyId, input.previousRunId)); + if (input.target) liveTargets.set(targetKey(input.companyId, input.runId), input.target); + } + return row; + } + return { prepare, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release }; } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index fd341bae281..60d3eb9c309 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -220,6 +220,7 @@ import { buildNativeRuntimeContext, cancelNativeSession, claimNativeRestartRecoveries, + claimWarmNativeInstructionCopy, closeWarmNativeSessionsForEnvironment, closeIdleWarmNativeSessionsForRestart, currentNativeControllerIdentity, @@ -20247,6 +20248,7 @@ export function heartbeatService( Parameters[0] | null = null; let nativeSessionResumeScheduled = false; let nativeOwnershipHeld = false; + let releaseWarmInstructionPreparation: (() => Promise) | null = null; let nativeDispatchStarted = false; let nativeWorkspaceFinalizeScheduled = false; let nativeWorkspaceSync: Awaited< @@ -22419,6 +22421,10 @@ export function heartbeatService( const executionTarget = realizationResult.executionTarget; let instructionCopy: Awaited> = null; let instructionSave: Record | null = null; + const instructionPreparationKey = createHash("sha256").update(JSON.stringify({ + adapterType: agent.adapterType, adapterConfig: agent.adapterConfig, runtimeConfig: agent.runtimeConfig, sessionConfigMetadata, + workspaceId: persistedExecutionWorkspace?.id ?? run.id, cwd: executionWorkspace.cwd, + })).digest("hex"); const collectStoppedInstructions = async () => { if (!instructionCopy) return; let saved = await instructionCopies.collectStopped({ companyId: agent.companyId, runId: run.id, target: executionTarget }); @@ -22440,6 +22446,12 @@ export function heartbeatService( : saved.errorMessage ?? "Instruction edits were not saved. Review the preserved candidate in the agent instruction editor."), payload: instructionSave }); }; + const nativeInstructionWorkingCopy = () => instructionCopy ? { + ...(isAgentDirectoryCopy(instructionCopy) ? { runId: run.id, preparationKey: instructionPreparationKey } : {}), + hasChanges: () => instructionCopies.hasChanges({ companyId: agent.companyId, runId: run.id, target: executionTarget }), + collectStopped: collectStoppedInstructions, + retirementFailed: async () => { await instructionCopies.reportRetirementUnconfirmed(agent.companyId, run.id); }, + } : undefined; if (managedAiRuntime && aiBinding) { try { await assertManagedAiProjectAuth({ ...resolvedConfig, cwd: executionWorkspace.cwd }, aiBinding.provider, executionTarget); } catch { throw new ConfigurationIncompleteFailure("Project authentication conflicts with this agent’s managed AI connection", { configurationIncomplete: { reason: "ai_connection_incompatible", actionUrl: `/agents/${agent.id}/runtime` } }); } @@ -23256,14 +23268,26 @@ export function heartbeatService( try { // Missing contract fields on a restored session mean the deployed // legacy format. New sessions opt into whole-directory persistence. - const priorFileRun = taskSessionForRun?.lastRunId + const priorFileRun = taskSession?.lastRunId ? await db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.id, taskSessionForRun.lastRunId), eq(heartbeatRuns.companyId, agent.companyId), eq(heartbeatRuns.agentId, agent.id))).then(rows => rows[0]) + eq(heartbeatRuns.id, taskSession.lastRunId), eq(heartbeatRuns.companyId, agent.companyId), eq(heartbeatRuns.agentId, agent.id))).then(rows => rows[0]) : null; const savedFileInput = parseObject(parseObject(run.runnerProfileJson).nativeExecutionInput); const priorFileInput = Object.keys(savedFileInput).length ? savedFileInput : parseObject(parseObject(priorFileRun?.profile).nativeExecutionInput); const priorWorkingCopy = parseObject(parseObject(parseObject(priorFileInput.runtimeContext).instructions).workingCopy); - instructionCopy = await instructionCopies.prepare({ + if (nativeRuntimeResolution.kind === "native" && priorWorkingCopy.kind === "agent_files" && taskSession) { + releaseWarmInstructionPreparation = await claimWarmNativeInstructionCopy({ + priorExecution: parseNativeExecutionInput(priorFileInput), companyId: agent.companyId, agentId: agent.id, + executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id, cwd: executionWorkspace.cwd, + environmentId: executionTarget?.environmentId ?? null, runId: run.id, preparationKey: taskSessionForRun ? instructionPreparationKey : `reset:${run.id}`, + adopt: async previousRunId => { + instructionCopy = await instructionCopies.adopt({ companyId: agent.companyId, agentId: agent.id, + runId: run.id, previousRunId, target: executionTarget, cwd: executionWorkspace.cwd }); + return nativeInstructionWorkingCopy() ?? null; + }, + }); + } + instructionCopy ??= await instructionCopies.prepare({ companyId: agent.companyId, agentId: agent.id, runId: run.id, target: executionTarget, cwd: executionWorkspace.cwd, legacy: Object.keys(priorFileInput).length > 0 && priorWorkingCopy.kind !== "agent_files", @@ -24449,10 +24473,7 @@ export function heartbeatService( }, onLog, onEvent: onAdapterEvent, - instructionWorkingCopy: instructionCopy ? { - hasChanges: () => instructionCopies.hasChanges({ companyId: agent.companyId, runId: run.id, target: executionTarget }), - collectStopped: collectStoppedInstructions, - } : undefined, + instructionWorkingCopy: nativeInstructionWorkingCopy(), preparationSpans: nativeRunnerPreparationSpans, // Bootstrap with executable/home discovery while keeping // configured provider values and the server-selected @@ -26331,6 +26352,7 @@ export function heartbeatService( } // A retained or unverified process stays above this release boundary. // If no stopped-copy capture occurred, preserve an explicit loss report. + await releaseWarmInstructionPreparation?.(); const uncapturedInstructions = await instructionCopies.reportUnavailable(run.companyId, run.id); if (uncapturedInstructions?.state === "unavailable") { await appendRunEvent(run, { eventType: "instruction_save", stream: "system", level: "warn", diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index b6916345f3d..d306612e2f8 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -41,6 +41,11 @@ import { import { createHash } from "node:crypto"; import { DatabaseSync } from "node:sqlite"; import { nativeSha256 } from "./canonical.js"; +import { agentDirectoryWorkingCopyService } from "../agent-directory-working-copies.js"; +import { probeAgentDirectory } from "../agent-directory-probe.js"; +import { AGENT_FILES_CONTRACT } from "../agent-file-store.js"; +import { captureDirectorySnapshot, directorySnapshotSha256, serializeDirectorySnapshot } from "@paperclipai/adapter-utils/workspace-restore-merge"; +import { agentInstructionWorkingCopies } from "@paperclipai/db"; import * as noLaunchProofModule from "./native-maintenance-no-launch.js"; import { NativeSessionCleanupQuarantinedError, @@ -276,6 +281,7 @@ import { buildNativeHarnessBackupManifest, cancelNativeSession, closeWarmNativeSessionsForEnvironment, + claimWarmNativeInstructionCopy, closeIdleWarmNativeSessionsForRestart, createGovernedWaitEventObservation, createRemoteRunnerProcessLauncher, @@ -5971,6 +5977,140 @@ describe("native session same-turn steering", () => { }); describe("native warm session supervision", () => { + it.each(["idle", "edit", "add", "remove"])("retains the actual agent_files copy across warm turns, then retires for %s", async ending => { + const root = await realpath(await mkdtemp(join(tmpdir(), "warm-agent-files-"))); + const localRoot = join(root, "run", "live"); + await mkdir(localRoot, { recursive: true }); + await writeFile(join(localRoot, "AGENTS.md"), "Keep the registered directory.\n"); + const snapshot = await captureDirectorySnapshot(localRoot); + let row = { companyId: execution.binding.companyId, agentId: execution.binding.agentId, + runId: "warm-real-agent-files", localRoot, executionRoot: localRoot, location: "local", + entryFile: "AGENTS.md", state: "prepared", attempts: 0, processStoppedAt: null, + baseRevisionId: null, candidateBase64: null, candidateHash: null, errorCode: null, errorMessage: null, nextAttemptAt: null, + responsibleUserId: "test-user", createdAt: new Date(), updatedAt: new Date(), + baseHash: directorySnapshotSha256(snapshot), receipt: { schema: AGENT_FILES_CONTRACT, baseline: serializeDirectorySnapshot(snapshot), materializationIdentity: probeAgentDirectory(localRoot).identity }, + } as typeof agentInstructionWorkingCopies.$inferSelect; + const copies = agentDirectoryWorkingCopyService({} as Db, async () => row, + async (_prior, values) => (row = { ...row, ...values } as typeof row)); + const order: string[] = []; + const close = vi.fn(async () => { order.push("closed"); }); + const collectStopped = async () => { + order.push("collected"); + expect(close).toHaveBeenCalledOnce(); + // The real DB service suite verifies changed-file persistence. Here the + // fake backend proves stop precedes granting its collection callback. + if (ending === "idle") await copies.collectStopped(row); + }; + const identity = "warm-real-agent-files"; + const warmExecution = { ...execution, binding: { ...execution.binding, runId: identity, executionWorkspaceId: `${identity}-workspace` }, + session: { ...execution.session, normalizedSessionId: identity, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } }, + } as NativeExecutionInputV1; + const session = { close }; + state.execute.mockReset().mockImplementation(async (options) => { + await options.onSession?.(session); + return { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" }, + turnId: identity, normalizedSessionId: identity, providerSessionId: identity, + driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + try { + await executePaperclipNativeSession({ db: leaseDb(warmExecution), execution: warmExecution, + runnerInstanceId: identity, runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId: identity, remoteCwd: `/tmp/${identity}` }, + instructionWorkingCopy: { runId: identity, preparationKey: "same-preparation", hasChanges: () => copies.hasChanges(row), collectStopped } }); + expect(close).not.toHaveBeenCalled(); + expect(order).toEqual([]); + expect(await readFile(join(localRoot, "AGENTS.md"), "utf8")).toContain("registered"); + const second = { ...warmExecution, binding: { ...warmExecution.binding, runId: `${identity}-second` } }; + const currentCopy = { runId: second.binding.runId, preparationKey: "same-preparation", hasChanges: () => copies.hasChanges(row), collectStopped }; + const release = await claimWarmNativeInstructionCopy({ priorExecution: warmExecution, + companyId: warmExecution.binding.companyId, agentId: warmExecution.binding.agentId, + executionWorkspaceId: warmExecution.binding.executionWorkspaceId, cwd: warmExecution.workspace.cwd, + environmentId: identity, runId: second.binding.runId, preparationKey: "same-preparation", + adopt: async () => currentCopy }); + expect(release).toBeTypeOf("function"); + await executePaperclipNativeSession({ db: leaseDb(second), execution: second, runnerInstanceId: identity, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId: identity, remoteCwd: `/tmp/${identity}` }, + instructionWorkingCopy: currentCopy }); + expect(state.execute.mock.calls.at(-1)?.[0].existingSession).toBe(session); + expect(close).not.toHaveBeenCalled(); + expect(row.executionRoot).toBe(localRoot); + await release?.(); // Consumed preparation cannot retire the retained owner. + expect(close).not.toHaveBeenCalled(); + if (ending === "idle") { + await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "idle retirement" }); + await expect(access(localRoot)).rejects.toMatchObject({ code: "ENOENT" }); + } else { + if (ending === "edit") await writeFile(join(localRoot, "AGENTS.md"), "Changed instructions"); + if (ending === "add") await writeFile(join(localRoot, "memory.txt"), "New memory"); + if (ending === "remove") await rm(join(localRoot, "AGENTS.md")); + const adopt = vi.fn(async () => currentCopy); + expect(await claimWarmNativeInstructionCopy({ priorExecution: second, + companyId: second.binding.companyId, agentId: second.binding.agentId, + executionWorkspaceId: second.binding.executionWorkspaceId, cwd: second.workspace.cwd, + environmentId: identity, runId: `${identity}-third`, preparationKey: "same-preparation", adopt })).toBeNull(); + expect(adopt).not.toHaveBeenCalled(); + } + expect(order).toEqual(["closed", "collected"]); + } finally { + await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" }); + await rm(root, { recursive: true, force: true }); + } + }); + + it.each(["final-config", "late-edit", "close-failure", "abandon", "concurrent", "foreign-company", "foreign-workspace"])("fences owned instruction preparation: %s", async scenario => { + const identity = `warm-copy-${scenario}`; + const order: string[] = []; + let allowClose = scenario !== "close-failure"; + const session = { close: vi.fn(async () => { order.push("close"); if (!allowClose) throw new Error("retirement unconfirmed"); }) }; + const retirementFailed = vi.fn(async () => {}); + let lateEdit = false; + const copy = { runId: identity, preparationKey: "key", hasChanges: async () => lateEdit, + collectStopped: vi.fn(async () => { order.push("collect"); }), retirementFailed }; + const first = { ...execution, binding: { ...execution.binding, runId: identity, executionWorkspaceId: `${identity}-workspace` }, + session: { ...execution.session, normalizedSessionId: identity, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } }, + } as NativeExecutionInputV1; + const target = { kind: "remote" as const, transport: "sandbox" as const, environmentId: identity, remoteCwd: `/tmp/${identity}` }; + state.execute.mockReset().mockImplementation(async options => { + await options.onSession?.(session); + return { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" }, + turnId: identity, normalizedSessionId: identity, providerSessionId: identity, + driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + const second = { ...first, binding: { ...first.binding, runId: `${identity}-second` } }; + const current = { ...copy, runId: second.binding.runId }; + const adopt = vi.fn(async () => current); + const claim = { priorExecution: first, companyId: first.binding.companyId, agentId: first.binding.agentId, + executionWorkspaceId: first.binding.executionWorkspaceId, cwd: first.workspace.cwd, environmentId: identity, + runId: second.binding.runId, preparationKey: "key", adopt }; + try { + await executePaperclipNativeSession({ db: leaseDb(first), execution: first, runnerInstanceId: identity, runnerExecutionTarget: target, instructionWorkingCopy: copy }); + if (scenario.startsWith("foreign")) { + expect(await claimWarmNativeInstructionCopy({ ...claim, + ...(scenario === "foreign-company" ? { companyId: "other" } : { executionWorkspaceId: "other" }) })).toBeNull(); + expect(adopt).not.toHaveBeenCalled(); expect(session.close).not.toHaveBeenCalled(); + } else { + const release = await claimWarmNativeInstructionCopy(claim); + if (scenario === "concurrent") await expect(claimWarmNativeInstructionCopy({ ...claim, runId: "competitor" })).rejects.toThrow("busy"); + if (scenario === "final-config" || scenario === "late-edit") { + lateEdit = scenario === "late-edit"; + const changed = scenario === "late-edit" ? second : { ...second, session: { ...second.session, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 99_000 } } }; + await expect(executePaperclipNativeSession({ db: leaseDb(changed), execution: changed, runnerInstanceId: identity, + runnerExecutionTarget: target, instructionWorkingCopy: current })).rejects.toThrow(scenario === "late-edit" + ? "native_instruction_preparation_copy_changed" : "native_instruction_preparation_configuration_changed"); + expect(state.execute).toHaveBeenCalledOnce(); + } + if (scenario === "close-failure") { + await expect(release!()).rejects.toThrow("retirement unconfirmed"); + expect(retirementFailed).toHaveBeenCalledOnce(); + expect(copy.collectStopped).not.toHaveBeenCalled(); + expect(order).toEqual(["close"]); + } else { + await release?.(); + expect(order).toEqual(["close", "collect"]); + } + } + } finally { allowClose = true; await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" }); } + }); + it.each([ { changed: false, closeFails: false }, { changed: true, closeFails: false }, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e8ffeb681bc..475275d89a6 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -391,7 +391,18 @@ function clearNativeRuntimeRequestResolutions(runId: string): void { } } +export type NativeInstructionWorkingCopy = { + /** Present only for a complete registered agent_files materialization. */ + runId?: string; + preparationKey?: string; + hasChanges: () => Promise; + collectStopped: () => Promise; + retirementFailed?: () => Promise; +}; + type WarmNativeSession = { + instructionWorkingCopy?: NativeInstructionWorkingCopy; + instructionPreparationRunId?: string; managedAiCredentialIdentity?: string; credentialRunId?: string; githubAccess?: NativeGitHubAccess; @@ -412,7 +423,15 @@ type WarmNativeSession = { async function closeWarmNativeSession(entry: WarmNativeSession, reason: string) { // Revoke before awaiting process retirement/checkpoint IO. const stopping = entry.githubAccess?.stop(); - try { await entry.session.close({ reason }); } + try { + try { await entry.session.close({ reason }); } + catch (error) { + try { await entry.instructionWorkingCopy?.retirementFailed?.(); } + catch (receiptError) { throw new AggregateError([error, receiptError], "Warm instruction retirement and receipt both failed"); } + throw error; + } + await entry.instructionWorkingCopy?.collectStopped(); + } finally { await stopping; } } @@ -421,6 +440,59 @@ const warmNativeSessions = new Map(); // not inspect or quarantine that owner's state until the save has finished. const closingWarmNativeSessions = new Map>(); +/** Reserve the existing physical copy before heartbeat composes AGENT_HOME and + * task guidance. The final execution/configuration fence still owns admission. */ +export async function claimWarmNativeInstructionCopy(input: { + priorExecution: NativeExecutionInput; + companyId: string; + agentId: string; + executionWorkspaceId: string; + cwd: string; + environmentId: string | null; + runId: string; + preparationKey: string; + adopt: (previousRunId: string) => Promise; +}): Promise<(() => Promise) | null> { + const prior = input.priorExecution; + if (prior.binding.companyId !== input.companyId || prior.binding.agentId !== input.agentId + || prior.binding.executionWorkspaceId !== input.executionWorkspaceId || prior.workspace.cwd !== input.cwd) return null; + const scope = nativeSessionScopeKey(prior); + const entry = warmNativeSessions.get(scope); + if (!entry || !entry.instructionWorkingCopy?.runId || entry.environmentId !== input.environmentId) return null; + if (entry.busy) throw new Error("native_session_supervisor_busy"); + entry.busy = true; + if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); + entry.idleTimer = null; + const retire = async (reason: string) => { + if (warmNativeSessions.get(scope) !== entry) return; + entry.closeOnReleaseReason = reason; + try { + await closeWarmNativeSession(entry, reason); + if (warmNativeSessions.get(scope) === entry) warmNativeSessions.delete(scope); + } finally { entry.busy = false; } + }; + try { + if (entry.closeOnReleaseReason !== undefined || entry.instructionWorkingCopy.preparationKey !== input.preparationKey || await entry.instructionWorkingCopy.hasChanges()) { + await retire("warm instruction materialization changed before preparation"); + return null; + } + const adopted = await input.adopt(entry.instructionWorkingCopy.runId); + if (!adopted) { + await retire("warm instruction materialization cannot be adopted"); + return null; + } + entry.instructionWorkingCopy = adopted; + entry.instructionPreparationRunId = input.runId; + return async () => { + if (entry.instructionPreparationRunId === input.runId) await retire("warm instruction preparation abandoned"); + }; + } catch (error) { + if (entry.closeOnReleaseReason === undefined) await retire("warm instruction preparation failed"); + throw error; + } +} + + /** * Close idle native sessions before an operator destroys their remote * environment. A warm runner owns a long-lived sandbox command stream, so the @@ -467,19 +539,23 @@ async function closeIdleWarmNativeSessions(input: { continue; } if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); - // Remove ownership before awaiting close so a racing continuation cannot - // adopt a session whose transport is already shutting down. - warmNativeSessions.delete(sessionId); + // Keep a fenced owner until both retirement and stopped collection succeed. + // Failure must not make a live directory eligible for a replacement owner. + entry.busy = true; + entry.closeOnReleaseReason = input.reason; const closing = Promise.resolve().then(() => closeWarmNativeSession(entry, input.reason), ); closingWarmNativeSessions.set(sessionId, closing); try { await closing; + if (warmNativeSessions.get(sessionId) === entry) warmNativeSessions.delete(sessionId); closed += 1; } catch { + if (!entry.instructionWorkingCopy?.runId) warmNativeSessions.delete(sessionId); failed += 1; } finally { + entry.busy = false; if (closingWarmNativeSessions.get(sessionId) === closing) { closingWarmNativeSessions.delete(sessionId); } @@ -5223,7 +5299,7 @@ function hasIdleWarmNativeSessionOwner(input: { }): boolean { if (input.execution.session.lifecyclePolicy.mode !== "warm") return false; const entry = warmNativeSessions.get(nativeSessionScopeKey(input.execution)); - if (!entry || entry.busy) return false; + if (!entry || (entry.busy && entry.instructionPreparationRunId !== input.execution.binding.runId)) return false; // A verified idle owner proves the checkpoint belongs to this session even // when its process must later rotate to a new run-scoped broker capability. // Local environments also have an id. Compare the same environment binding @@ -5935,23 +6011,27 @@ async function releaseWarmNativeSession( entry.lastActivityAt = new Date().toISOString(); if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); if (failed || entry.closeOnReleaseReason !== undefined) { - warmNativeSessions.delete(sessionId); - const closing = closeWarmNativeSession(entry, - entry.closeOnReleaseReason ?? "warm native session failed"); - // Restart checkpointing is required to restore this successful session. - // Surface failure instead of reporting a clean release without authority. - if (entry.closeOnReleaseReason !== undefined) await closing; - else await closing.catch(() => undefined); + const requireCleanClose = entry.closeOnReleaseReason !== undefined; + entry.closeOnReleaseReason ??= "warm native session failed"; + entry.busy = true; + try { + await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + if (warmNativeSessions.get(sessionId) === entry) warmNativeSessions.delete(sessionId); + } catch (error) { + if (!entry.instructionWorkingCopy?.runId) warmNativeSessions.delete(sessionId); + if (requireCleanClose) throw error; + } finally { entry.busy = false; } return; } entry.idleTimer = setTimeout(() => { const current = warmNativeSessions.get(sessionId); - // clearTimeout cannot revoke an already-queued callback. The entry object - // is the idle timer's ownership fence across a later warm acquisition. if (current !== entry || current.busy) return; - warmNativeSessions.delete(sessionId); - void closeWarmNativeSession(current, "warm native session idle timeout") - .catch(() => undefined); + current.busy = true; + current.closeOnReleaseReason = "warm native session idle timeout"; + void closeWarmNativeSession(current, current.closeOnReleaseReason) + .then(() => { if (warmNativeSessions.get(sessionId) === current) warmNativeSessions.delete(sessionId); }) + .catch(() => { if (!current.instructionWorkingCopy?.runId && warmNativeSessions.get(sessionId) === current) warmNativeSessions.delete(sessionId); }) + .finally(() => { current.busy = false; }); }, idleTimeoutMs); entry.idleTimer.unref(); } @@ -7205,10 +7285,7 @@ export async function executePaperclipNativeSession(input: { onEvent?: (event: AdapterRuntimeEvent) => Promise; /** Only this run's registered private instruction entry. * Probe at terminal; persist only after owned shutdown. */ - instructionWorkingCopy?: { - hasChanges: () => Promise; - collectStopped: () => Promise; - }; + instructionWorkingCopy?: NativeInstructionWorkingCopy; /** Persist task-level continuity before a durable goal can outlive this run. */ onGoalCheckpoint?: (snapshot: PersistedNativeSession) => Promise; sessionGoalControl?: NativeSessionGoalControl | null; @@ -8099,17 +8176,36 @@ async function executePaperclipNativeSessionWithinScope( (input.runnerEnvironment?.PAPERCLIP_RUNNER_NETWORK_ACCESS === "enabled") ) { - if (entry.busy) throw new Error("native_session_supervisor_busy"); + const preparedCopy = entry.instructionPreparationRunId === input.execution.binding.runId; + if (entry.busy && !preparedCopy) throw new Error("native_session_supervisor_busy"); if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); - warmNativeSessions.delete(warmSessionId); - await closeWarmNativeSession(entry, "warm native session configuration changed"); + entry.busy = true; + entry.closeOnReleaseReason = "warm native session configuration changed"; + try { + await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + warmNativeSessions.delete(warmSessionId); + } finally { entry.busy = false; } + if (preparedCopy) throw new Error("native_instruction_preparation_configuration_changed"); persistedWarmSession = loadWarmNativeCheckpoint( input.execution, warmConfigDigest, input.runnerExecutionTarget?.kind ?? "local", ); } else { - if (entry.busy) throw new Error("native_session_supervisor_busy"); + if (entry.busy && entry.instructionPreparationRunId !== input.execution.binding.runId) throw new Error("native_session_supervisor_busy"); + if (entry.instructionWorkingCopy?.runId && entry.instructionPreparationRunId !== input.execution.binding.runId) { + throw new Error("native_instruction_materialization_not_claimed"); + } + if (entry.instructionPreparationRunId !== undefined && await entry.instructionWorkingCopy!.hasChanges()) { + entry.closeOnReleaseReason = "warm instruction copy changed during preparation"; + try { + await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + warmNativeSessions.delete(warmSessionId); + } finally { entry.busy = false; } + throw new Error("native_instruction_preparation_copy_changed"); + } + entry.instructionPreparationRunId = undefined; + entry.instructionWorkingCopy = input.instructionWorkingCopy; entry.busy = true; entry.ownerToken = warmSessionOwnerToken; entry.environmentId = @@ -8443,6 +8539,7 @@ async function executePaperclipNativeSessionWithinScope( existing.session = session; } else warmNativeSessions.set(warmSessionId, { + instructionWorkingCopy: input.instructionWorkingCopy, managedAiCredentialIdentity: input.managedAiCredentialIdentity, githubAuthenticationMode: input.runnerEnvironment?.PAPERCLIP_GITHUB_AUTH_MODE, @@ -9128,7 +9225,7 @@ async function executePaperclipNativeSessionWithinScope( lifecyclePolicy.idleTimeoutMs, false, ); - if (collectInstructions) await instructionCopy!.collectStopped(); + // releaseWarmNativeSession closes and collects through the same owner. } const adapterResult: AdapterExecutionResult = { exitCode: native.terminal.runTerminalState === "succeeded" ? 0 : 1,