mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 02:26:12 +02:00
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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
3d21d375de
commit
6c6c05dccb
8 files changed
+736
-50
No files matched your search
@@ -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<typeof read>) => {
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -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<void>(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" });
|
||||
|
||||
@@ -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<string, string>();
|
||||
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<string, string>();
|
||||
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 };
|
||||
}
|
||||
@@ -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<ReturnType<typeof retrieve>> | 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),
|
||||
};
|
||||
|
||||
@@ -62,6 +62,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
async function patch(row: Copy, values: Partial<typeof copies.$inferInsert>) {
|
||||
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<typeof directories.adopt>[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 };
|
||||
}
|
||||
@@ -220,6 +220,7 @@ import {
|
||||
buildNativeRuntimeContext,
|
||||
cancelNativeSession,
|
||||
claimNativeRestartRecoveries,
|
||||
claimWarmNativeInstructionCopy,
|
||||
closeWarmNativeSessionsForEnvironment,
|
||||
closeIdleWarmNativeSessionsForRestart,
|
||||
currentNativeControllerIdentity,
|
||||
@@ -20247,6 +20248,7 @@ export function heartbeatService(
|
||||
Parameters<typeof cleanupGitHubOperationLaunchers>[0] | null = null;
|
||||
let nativeSessionResumeScheduled = false;
|
||||
let nativeOwnershipHeld = false;
|
||||
let releaseWarmInstructionPreparation: (() => Promise<void>) | 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<ReturnType<typeof instructionCopies.prepare>> = null;
|
||||
let instructionSave: Record<string, unknown> | 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",
|
||||
|
||||
@@ -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 },
|
||||
|
||||
@@ -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<boolean>;
|
||||
collectStopped: () => Promise<void>;
|
||||
retirementFailed?: () => Promise<void>;
|
||||
};
|
||||
|
||||
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<string, WarmNativeSession>();
|
||||
// not inspect or quarantine that owner's state until the save has finished.
|
||||
const closingWarmNativeSessions = new Map<string, Promise<void>>();
|
||||
|
||||
/** 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<NativeInstructionWorkingCopy | null>;
|
||||
}): Promise<(() => Promise<void>) | 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<void>;
|
||||
/** Only this run's registered private instruction entry.
|
||||
* Probe at terminal; persist only after owned shutdown. */
|
||||
instructionWorkingCopy?: {
|
||||
hasChanges: () => Promise<boolean>;
|
||||
collectStopped: () => Promise<void>;
|
||||
};
|
||||
instructionWorkingCopy?: NativeInstructionWorkingCopy;
|
||||
/** Persist task-level continuity before a durable goal can outlive this run. */
|
||||
onGoalCheckpoint?: (snapshot: PersistedNativeSession) => Promise<void>;
|
||||
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,
|
||||
|
||||
Reference in new issue
Block a user