diff --git a/doc/agent-files.md b/doc/agent-files.md index 4afb88bca78..e3a6698fcfe 100644 --- a/doc/agent-files.md +++ b/doc/agent-files.md @@ -109,7 +109,29 @@ probe must verify the full baseline and root identity without unsafe paths or concurrent changes. Local hashing runs in an isolated child; remote observation runs at the registered remote root. A failed or uncertain probe requires stopped collection. The next authorized run claims that same materialization within the -company, agent, workspace, environment and configuration scope. The prior run +company, agent, workspace, environment and configuration scope. Projectless runs +use the stable workspace descriptor (cwd, repository URL/ref and branch), rather +than the per-run workspace placeholder. A remote handoff verifies both DB leases +belong to the same company, environment and provider allocation, then binds the +successor run's active lease and collector without re-uploading the host mirror. +Heartbeat explicitly permits a collection-only successor handoff before checking +warm reuse. A changed or uncertain directory records durable `retirementRequired`; +that receipt cannot authorize reuse. The successor capability is installed only +after the ownership transaction commits. Configuration changes and abandoned +preparation then stop the provider and collect through that current capability, +never an expired prior lease. Ordinary adoption remains unchanged-only. Failed +authorization or lease validation retires the prior owner without borrowing an +unverified successor; if its lease is unusable, the bytes remain pending. +Before initial admission, the remote transfer's two empty scratch directories are +removed with exact-path, identity-checked `rmdir` operations. Any content, link or +uncertain identity fails preparation; the complete probe never excludes them. +An exact recorded lease that is missing or no longer matches the run/environment +blocks retrieval as well as deletion. Legacy receipts without a lease ID require +exactly one company/run/environment lease, including when a transport is still +cached. Missing or ambiguous ownership preserves pending files without remote +commands. No SSH exception bypasses this fence, and no-ID receipts grant no warm +adoption authority. +The original materialization root remains unchanged. The prior run loses collection authority; stale cleanup cannot remove the current owner's root. Edits, additions, removals or changed configuration retire the owner before @@ -175,9 +197,14 @@ If retirement fails, the controller records unresolved ownership and preserves the materialized root. A crash after ownership transfer can leave the new run without independently recorded stop proof; terminal run status or the old run's alias is not enough to collect that root. Recovery leaves it preserved until the -required proof is available. For a remote run with an exact termination receipt -but unretrievable bytes, recovery records an explicit unavailable/no-save outcome -and completes owned cleanup without restarting a provider. +required proof is available. Exact destruction of the current owner's remote +allocation permits an explicit unavailable/no-save outcome and owned local +cleanup. A stopped-but-retained allocation instead stays pending with its files +preserved. The current remote execute API may restart a stopped sandbox even +when it bypasses a persistent session, so recovery cannot use it for retrieval +or deletion. Safe retrieval from a stopped allocation remains a transport gap; +live owner retirement still collects before the environment stops. No save is +claimed for the pending case, and stale prior-run leases cannot authorize cleanup. ## Verification diff --git a/doc/architecture/runner-rich-acp-capabilities.md b/doc/architecture/runner-rich-acp-capabilities.md index f35ab72379b..804196b55f0 100644 --- a/doc/architecture/runner-rich-acp-capabilities.md +++ b/doc/architecture/runner-rich-acp-capabilities.md @@ -1,5 +1,77 @@ # Rich ACP integration and qualification report +## Qualification checkpoint — September 30, 2026, after 64 attempts + +All three providers remain unqualified. The retained paid baseline is Cursor v7, +Copilot v7 and Pi v9, with native runtime source `5b8e4454ef0bf12d0bb068c2e41d8c9df9356a1c`. +Controller and Product harness revisions are recorded independently below. +New usage-projection source is under development and has no paid qualification. + +| New retained evidence | Outcome and boundary | +| --- | --- | +| Cursor v7 Daytona hello, controller3d21 | Six Product checks passed. One owned sandbox was deleted and remained absent through the late-create observation; local processes and temporary roots retired. Post-run authority passed. This qualifies only this case. | +| Copilot v7 Daytona denial, controller40064, attempt02 | Failed remote observer startup before permission proof. Cleanup confirmed removal of the owned sandbox and local processes. Original end source-checkout audit failed because the checkout moved during its final audit; an additive audit after exact restoration passed, without regrading the failed attempt. | +| Cursor v7 protocol get-task-context | All four semantic checks passed, but the unchanged eval rejected missing usage/cost coverage. No usage receipt or priced estimate was available. Preserve accounting_failure. | +| Copilot v7 seven-case protocol suite | Six cases passed: task context, document creation, task completion, human confirmation, context/document/progress workflow and governed waiting. Context-before-action failed: progress was committed before the context read. Both calls succeeded and six of seven checks in that case passed. State history confirms ordering; no grader or projection defect was established. No automatic retries. | + +Protocol cases use the seeded mock control plane; they do not qualify browser, +Daytona or production mutations. Copilot estimates for all seven measured cases +have complete token-price coverage, while native billed dollars remain unknown. +The failed orientation case is a model-behavior result, not an infrastructure +retry candidate. The five final independent cases passed 35 checks in total. + +The remote observer fix in [#14696](https://github.com/paperclipai/paperclip/pull/14696) +sends a small nonce-bound snapshot request instead of forwarding its source code +over an 8 KiB control socket. The exact latest source `a1145db4d8dbde17e624f8a14ef2726b33d09083` +passes Apex 5/5 and CI. Its 41 deterministic tests include generated install, +snapshot and close traffic; negative tests require actual directory creation and +socket listening. It still needs live proof in the combined controller candidate. + +The warm-directory review identified stable projectless scope, Darwin path +aliases, current lease ownership, stopped-versus-destroyed recovery, and composed +remote coverage gaps. New composed tests also exposed transfer scratch inside +AGENT_HOME, stale first-lease callbacks during handoff and compact receipts losing +the original materialization root. The corrected source passed 541 pure tests, direct server typecheck and all 63 +selected database-backed cases (49 directory-service and 14 composed remote +scenarios), with owned process/IPC/temporary-root cleanup verified. These repairs +are awaiting a fresh commit review in [#14695](https://github.com/paperclipai/paperclip/pull/14695). Earlier failing DB and +paid attempts remain retained; neither source tests nor session-ID reuse alone +prove paid warm process continuity. + +Cursor's pinned native `TurnEndedUpdate` has optional input, output, cache-read, +cache-write and reasoning counters. Its vendor ACP prompt response does not +project them. No evidence shows Paperclip dropped a supplied ACP usage receipt in +the failed attempt. The new candidate records bounded per-invocation observations +as explicitly partial diagnostics. It must not sum parent/child totals, invent +missing zeros or claim billing until the backend's aggregation/cache/reasoning +semantics and exact selected model pricing are verified. JSON-RPC errors do not +return this diagnostic envelope; reused child runs lack unambiguous callback +attribution. Those are explicit gaps, not complete accounting. The shared patch +changes identity declarations, requiring new Cursor/Copilot/Pi candidate identities +and new qualification; prior receipts remain bound to their original runtime. + +The private ledger has 64 closed records and no active paid invocation. Latest +Copilot account observation: Pro active, 23/1,500 included credits used, additional +usage disabled at $0. Latest Cursor observation: Pro+, fixed account-wide $25 cap, +$0 on-demand observed. These account totals may lag and are not per-run invoices. +Known Pi API cost is $0.040507624; the conservative retained infrastructure upper +bound is $0.635265. Cursor's entire $25 cap remains reserved globally. Unknown +native or infrastructure invoice totals remain unknown; no combined actual-spend +total is inferred. The $100 campaign ceiling remains in force. Pi v9 still needs +an enforceable provider spending bound before further paid calls. + +Evidence references (private attempts remain inspectable without publishing raw +provider transcripts): Cursor Daytona hello result `3ec06ed1aee8a5523262d4855dca3483de24139480da76f37d75a38570ecb843`; +Copilot denial02 result `048ec74bedf508080a099ea6be39f5d4ec159f31985dbe4c52f286a8f8813021`; +Cursor accounting reconciliation `4999b3c0abec441a4de2c6f45c6cb683cbfe4d94413c30f0333da3b6e53dbc03`; +Copilot ordering reconciliation `cf14df413fc9a98f6047842be21585e6c4d3fa72d4199ea9331885665193c152`; +five-case account reconciliation `0063361b04c33d724c9649be3790857a640bbe04357ffafd8c078e8cea40eb22`. + +## Historical checkpoint before the latest 11 attempts + +The following source and cost observations are retained historical checkpoints; +the dated qualification checkpoint above states the current result. + Updated: 2026-09-30 UTC. **Cursor v7, Copilot v7 and Pi v9 remain unqualified.** Paid controller and Product harness source is recorded per case: `40064d28522de25fea85c1297f35b41bb8a8897a` or `3d21d375de2b6249d9f5322bf001ce0ce0e052aa`. Native runtime source remains `5b8e4454ef0bf12d0bb068c2e41d8c9df9356a1c`. Runtime builds for macOS ARM64, macOS x64 and Linux x64 are complete. The corrected 3d21 controller uses unchanged compiled UI/package outputs built on 40064; it executes the controller TypeScript from 3d21. These distinct source roles do not relabel the runtime or older failures. Exact-head CI and Greptile now pass for Cursor `c58a8881f2cb771b35e19e1bf56e08cb89397e83` (53 successful checks, two skips, Greptile 5/5) and Copilot `3d21d375de2b6249d9f5322bf001ce0ce0e052aa` (54 successes, four skips, Greptile 5/5). Copilot's first server-shard run timed out in an unchanged legacy Cursor test; an isolated reproduction and one failed-job rerun passed. Superseded cancelled checks are not failures. The new warm-directory fix needs its own full checks and live proof. Foundation [#14430](https://github.com/paperclipai/paperclip/pull/14430) and Runner Eval definitions [#33](https://github.com/paperclipai/paperclip-evals/pull/33) are merged. The v7/v7/v9 definition update [#34](https://github.com/paperclipai/paperclip-evals/pull/34) merged as `52f6e897c08d236776a21723257e12c909d71137`, with 136 deterministic tests and 21 validated cells. These definitions are not paid proof. Provider PRs remain drafts. diff --git a/server/src/__tests__/agent-directory-probe.test.ts b/server/src/__tests__/agent-directory-probe.test.ts index 59ee14b1608..21aaf6f7f08 100644 --- a/server/src/__tests__/agent-directory-probe.test.ts +++ b/server/src/__tests__/agent-directory-probe.test.ts @@ -3,8 +3,9 @@ import fs from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import childProcess, { execFileSync } from "node:child_process"; +import { prepareAdapterExecutionTargetRuntime } from "@paperclipai/adapter-utils/execution-target"; import { captureDirectorySnapshot } from "@paperclipai/adapter-utils/workspace-restore-merge"; -import { agentDirectoryBaselineDigest, agentDirectoryProbeProgram, observeLocalAgentDirectory, probeAgentDirectory } from "../services/agent-directory-probe.js"; +import { agentDirectoryBaselineDigest, agentDirectoryProbeProgram, observeLocalAgentDirectory, probeAgentDirectory, retireAgentDirectoryTransferScratch, agentDirectoryTransferCleanupProgram } from "../services/agent-directory-probe.js"; describe("stable live agent directory observation", () => { let root: string; @@ -22,6 +23,61 @@ describe("stable live agent directory observation", () => { const remote = JSON.parse(execFileSync(process.execPath, ["-e", agentDirectoryProbeProgram, root], { encoding: "utf8", env: {}, timeout: 10_000 })); expect(remote).toEqual(observed); }); + it.runIf(process.platform === "darwin")("accepts only the observing Darwin host's fixed system aliases", () => { + const alias = root.replace(/^\/private(?=\/(tmp|var|etc)\/)/, ""); + expect(alias).not.toBe(root); + const observed = probeAgentDirectory(root); + expect(probeAgentDirectory(alias)).toEqual(observed); + const remote = JSON.parse(execFileSync(process.execPath, ["-e", agentDirectoryProbeProgram, alias], { encoding: "utf8", env: {}, timeout: 10_000 })); + expect(remote).toEqual(observed); + // A Linux remote cannot inherit the controller's Darwin exception. + expect(() => execFileSync(process.execPath, ["-e", agentDirectoryProbeProgram.replace("platform:process.platform", "platform:'linux'"), alias], + { encoding: "utf8", env: {}, timeout: 10_000, stdio: "pipe" })).toThrow(); + fs.symlinkSync(root, join(root, "redirect")); + expect(() => probeAgentDirectory(`${alias}/redirect`)).toThrow(); + }); + it("observes the complete agent directory after the actual remote transfer", async () => { + const remote = `${root}-remote`; + fs.mkdirSync(remote); + const target = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: remote, + runner: { execute: async (input: Parameters[0]) => ({ + stdout: execFileSync(input.command, input.args ?? [], { cwd: input.cwd, input: input.stdin, encoding: "utf8", + env: { PATH: process.env.PATH, ...input.env }, timeout: input.timeoutMs, maxBuffer: 32 * 1024 * 1024 }), + stderr: "", exitCode: 0, signal: null, timedOut: false, pid: null, startedAt: new Date().toISOString(), + }) } }; + let runtime: Awaited> | undefined; + try { + const baseline = await captureDirectorySnapshot(root); + runtime = await prepareAdapterExecutionTargetRuntime({ target, runId: "probe-transfer", adapterKey: "agent-files", + workspaceLocalDir: root, workspaceRemoteDir: remote, syncWorkspace: true, workspaceBaseline: baseline, + workspaceGitSnapshot: null, workspaceFileMode: "all", workspaceExclude: [".paperclip-runtime", ".paperclip-runtime/**"] }); + expect(fs.readdirSync(`${remote}/.paperclip-runtime`)).toEqual(["agent-files"]); + expect(fs.readdirSync(`${remote}/.paperclip-runtime/agent-files`)).toEqual([]); + execFileSync(process.execPath, ["-e", agentDirectoryTransferCleanupProgram, remote], { env: {}, timeout: 10_000 }); + expect(() => probeAgentDirectory(remote)).not.toThrow(); + expect(probeAgentDirectory(remote).digest).toBe(agentDirectoryBaselineDigest(baseline)); + } finally { await runtime?.cleanupWorkspaceSnapshot?.(); fs.rmSync(remote, { recursive: true, force: true }); } + }); + it.each(["file", "extra-directory", "symlink", "race"])("fails closed on %s in transfer-owned scratch without deleting contents", kind => { + const parent = join(root, ".paperclip-runtime"), scratch = join(parent, "agent-files"); + fs.mkdirSync(scratch, { recursive: true }); + const retained = join(parent, "retained.txt"); + if (kind === "file") fs.writeFileSync(join(scratch, "retained.txt"), "keep"); + if (kind === "extra-directory") fs.mkdirSync(join(parent, "unknown")); + if (kind === "symlink") { fs.rmdirSync(scratch); fs.symlinkSync(join(root, "notes"), scratch); } + if (kind === "race") { + const rmdir = fs.rmdirSync; + vi.spyOn(fs, "rmdirSync").mockImplementation(((directory: fs.PathLike) => { + if (directory === scratch) fs.writeFileSync(retained, "concurrent bytes"); + return rmdir(directory); + }) as typeof rmdir); + } + expect(() => retireAgentDirectoryTransferScratch(root)).toThrow(); + expect(fs.existsSync(parent)).toBe(true); + if (kind === "file") expect(fs.readFileSync(join(scratch, "retained.txt"), "utf8")).toBe("keep"); + if (kind === "race") expect(fs.readFileSync(retained, "utf8")).toBe("concurrent bytes"); + expect(fs.existsSync(join(root, "notes", "memory.bin"))).toBe(true); + }); it("runs the local observation in its bounded credential-free child", async () => { expect(await observeLocalAgentDirectory(root)).toEqual(probeAgentDirectory(root)); }); diff --git a/server/src/__tests__/agent-directory-working-copies.test.ts b/server/src/__tests__/agent-directory-working-copies.test.ts index 3c20b8750fc..a55e134b3d6 100644 --- a/server/src/__tests__/agent-directory-working-copies.test.ts +++ b/server/src/__tests__/agent-directory-working-copies.test.ts @@ -540,13 +540,13 @@ describe("persistent agent directories", () => { expect(execute).toHaveBeenCalledTimes(2); }); - it.each([false, true])("recovers a retained remote unchanged turn only with exact stop proof: %s", async stopped => { + it.each(["missing", "stopped", "destroyed"])("recovers a retained remote unchanged turn only with exact stop proof: %s", async stopped => { const copy = await run(); const environmentId = randomUUID(), leaseId = randomUUID(), remoteCwd = "/fixture/task"; const lease = { id: leaseId, companyId, environmentId, heartbeatRunId: copy.runId, provider: "daytona", providerLeaseId: "retained-sandbox" }; await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "sandbox" }); await db.insert(environmentLeases).values({ ...lease, status: "released", releasedAt: new Date(), cleanupStatus: "success", - metadata: stopped ? { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) } : {} }); + metadata: stopped !== "missing" ? { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: stopped }) } : {} }); await db.update(heartbeatRuns).set({ status: "succeeded", runtimeMode: "native" }).where(eq(heartbeatRuns.id, copy.runId)); await db.update(agentInstructionWorkingCopies).set({ state: "unchanged_turn", location: `remote:${environmentId}`, executionRoot: path.posix.join(remoteCwd, ".paperclip-runtime", "agent-files", agentId, copy.runId), @@ -556,19 +556,44 @@ describe("persistent agent directories", () => { expect((await copies.reportUnavailable(companyId, copy.runId))?.state).toBe("unchanged_turn"); await copies.recoverStopped(); const recovered = (await copies.get(companyId, copy.runId))!; - if (stopped) { + if (stopped === "destroyed") { expect(recovered).toMatchObject({ state: "unavailable", errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE", receipt: { cleanupPending: false } }); expect(recovered.processStoppedAt).toBeInstanceOf(Date); await expect(fs.stat(copy.localRoot)).rejects.toMatchObject({ code: "ENOENT" }); await copies.recoverStopped(); expect((await copies.get(companyId, copy.runId))?.updatedAt).toEqual(recovered.updatedAt); } else { - expect(recovered).toMatchObject({ state: "unchanged_turn", processStoppedAt: null }); + expect(recovered).toMatchObject({ state: stopped === "stopped" ? "pending_collection" : "unchanged_turn", processStoppedAt: null }); + if (stopped === "stopped") { + expect(recovered.errorCode).toBe("INSTRUCTION_STOPPED_REMOTE_COLLECTION_PENDING"); + expect(recovered.receipt?.baseline).toBeDefined(); + await copies.release(companyId, copy.runId); + await copies.recoverCaptured(); + } expect(await fs.readFile(path.join(copy.localRoot, entryFile), "utf8")).toBe(initial); } expect(execute).not.toHaveBeenCalled(); // Never restart a remote provider to recover bytes. }); + it.each([0, 2])("preserves a legacy no-ID remote copy with %s matching leases", async count => { + const copy = await run(); + const environmentId = randomUUID(), remoteCwd = "/fixture/task"; + await db.insert(environments).values({ id: environmentId, name: `Legacy remote ${environmentId}`, driver: "sandbox" }); + for (let index = 0; index < count; index++) await db.insert(environmentLeases).values({ + companyId, environmentId, heartbeatRunId: copy.runId, provider: "daytona", providerLeaseId: `allocation-${index}` }); + await db.update(agentInstructionWorkingCopies).set({ location: `remote:${environmentId}`, + executionRoot: path.posix.join(remoteCwd, ".paperclip-runtime", "agent-files", agentId, copy.runId), + receipt: { ...copy.receipt, cleanup: { remoteCwd } } }).where(eq(agentInstructionWorkingCopies.runId, copy.runId)); + const execute = vi.fn(); + copies = agentInstructionWorkingCopyService(db, { environmentRuntime: { execute } as unknown as EnvironmentRuntimeService }); + const pending = await copies.collectStopped({ companyId, runId: copy.runId }); + expect(pending).toMatchObject({ state: "pending_collection", attempts: 0, processStoppedAt: null, errorCode: "INSTRUCTION_REMOTE_LEASE_UNVERIFIED" }); + expect(pending!.receipt?.baseline).toBeDefined(); + await copies.release(companyId, copy.runId); + expect(await fs.readFile(path.join(copy.localRoot, entryFile), "utf8")).toBe(initial); + expect(execute).not.toHaveBeenCalled(); + }); + it("finishes remote cleanup from a destruction receipt after the environment is deleted", async () => { const copy = await run(); const environmentId = randomUUID(), leaseId = randomUUID(), remoteCwd = "/fixture/task"; @@ -650,10 +675,14 @@ describe("persistent agent directories", () => { } }, }; - const executionTarget = { kind: "remote" as const, transport: "sandbox" as const, environmentId: randomUUID(), remoteCwd, runner }; + const executionTarget = { kind: "remote" as const, transport: "sandbox" as const, environmentId: randomUUID(), leaseId: "", remoteCwd, runner }; + await db.insert(environments).values({ id: executionTarget.environmentId, name: "Fixture sandbox", driver: "sandbox" }); const prepare = async () => { const runId = randomUUID(); await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", responsibleUserId: userId }); + executionTarget.leaseId = randomUUID(); + await db.insert(environmentLeases).values({ id: executionTarget.leaseId, companyId, environmentId: executionTarget.environmentId, + heartbeatRunId: runId, provider: "daytona", providerLeaseId: "fixture-allocation" }); return (await copies.prepare({ ...target(), runId, cwd: home, target: executionTarget }))!; }; await fs.mkdir(path.join(root, "build")); @@ -696,6 +725,9 @@ describe("persistent agent directories", () => { await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", responsibleUserId: userId }); const target = { kind: "remote" as const, transport: "ssh" as const, environmentId: randomUUID(), remoteCwd, spec: { host: "unused.invalid", port: 22, username: "test", remoteCwd } }; + await db.insert(environments).values({ id: target.environmentId, name: "Fixture SSH", driver: "ssh" }); + // Legacy no-ID receipt: only the unique run/environment lease authorizes retrieval. + await db.insert(environmentLeases).values({ companyId, environmentId: target.environmentId, heartbeatRunId: runId, provider: "ssh", providerLeaseId: "fixture-host" }); const copy = (await copies.prepare({ companyId, agentId, runId, cwd: home, target }))!; expect(stage).toHaveBeenCalledWith(expect.objectContaining({ remoteDir: copy.executionRoot })); expect(await fs.readFile(path.join(copy.executionRoot, entryFile), "utf8")).toBe(initial); diff --git a/server/src/services/agent-directory-probe.ts b/server/src/services/agent-directory-probe.ts index f9ee172827e..ce895d336aa 100644 --- a/server/src/services/agent-directory-probe.ts +++ b/server/src/services/agent-directory-probe.ts @@ -15,7 +15,7 @@ export function agentDirectoryBaselineDigest(snapshot: DirectorySnapshot): strin // Keep all dependencies explicit so remote observation never reads a local mirror. export function probeAgentDirectory( root: string, - modules = { fs, path, createHash }, + modules = { fs, path, createHash, platform: process.platform }, ): { digest: string; identity: string } { const { fs, path, createHash } = modules; const deadline = Date.now() + 5_000; @@ -23,7 +23,14 @@ export function probeAgentDirectory( 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. + // Resolve only the observing host's fixed Darwin aliases. The exact full + // counterpart excludes nested/user symlinks; canonical ancestors remain strict. + const requestedRoot = root; + if (modules.platform === "darwin" && /^\/(tmp|var|etc)\//.test(root)) { + const resolved = fs.realpathSync(root); + if (resolved !== `/private${root}`) throw new Error("Unsafe agent directory alias"); + root = resolved; + } const ancestors = new Map(); let ancestor = path.parse(root).root; for (const segment of root.slice(ancestor.length).split(path.sep)) { @@ -94,11 +101,53 @@ export function probeAgentDirectory( const current = fs.lstatSync(ancestor, { bigint: true }); if (!current.isDirectory() || current.isSymbolicLink() || identity(current) !== expected) throw new Error("Agent directory ancestor changed"); } + if (requestedRoot !== root && fs.realpathSync(requestedRoot) !== root) throw new Error("Agent directory alias 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})))`; +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,platform:process.platform})))`; + +/** Retire only the two empty directories created by initial sandbox transfer. + * Run before provider admission; this is never an exclusion from live probing. */ +export function retireAgentDirectoryTransferScratch(root: string, modules = { fs, path, platform: process.platform }): void { + const { fs, path } = modules; + if (!path.isAbsolute(root) || path.resolve(root) !== root || root.includes("\0")) throw new Error("Invalid agent directory root"); + const requestedRoot = root; + if (modules.platform === "darwin" && /^\/(tmp|var|etc)\//.test(root)) { + const resolved = fs.realpathSync(root); + if (resolved !== `/private${root}`) throw new Error("Unsafe agent directory alias"); + root = resolved; + } + const identities = new Map(); + const identify = (directory: string) => { + const stat = fs.lstatSync(directory, { bigint: true }); + if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("Unsafe transfer scratch directory"); + return [stat.dev, stat.ino, stat.birthtimeNs].join(":"); + }; + let ancestor = path.parse(root).root; + for (const segment of root.slice(ancestor.length).split(path.sep)) { + ancestor = path.join(ancestor, segment); identities.set(ancestor, identify(ancestor)); + } + const parent = path.join(root, ".paperclip-runtime"), scratch = path.join(parent, "agent-files"); + identities.set(parent, identify(parent)); identities.set(scratch, identify(scratch)); + const verify = () => { + for (const [directory, expected] of identities) if (identify(directory) !== expected) throw new Error("Transfer scratch identity changed"); + if (requestedRoot !== root && fs.realpathSync(requestedRoot) !== root) throw new Error("Agent directory alias changed"); + }; + const entries = fs.readdirSync(parent); + if (entries.length !== 1 || entries[0] !== "agent-files" || fs.readdirSync(scratch).length) throw new Error("Unexpected transfer scratch contents"); + verify(); + // rmdir atomically refuses nonempty directories. Never recursively delete or + // follow a link; any change stops preparation before the provider can start. + fs.rmdirSync(scratch); identities.delete(scratch); + verify(); + if (fs.readdirSync(parent).length) throw new Error("Transfer scratch changed during retirement"); + fs.rmdirSync(parent); identities.delete(parent); + verify(); +} + +export const agentDirectoryTransferCleanupProgram = `const retire=${retireAgentDirectoryTransferScratch.toString()}; retire(process.argv[1], {fs:require('node:fs'),path:require('node:path'),platform:process.platform})`; /** Bound child: no shell, inherited credentials, or event-loop hashing. */ export async function observeLocalAgentDirectory(root: string) { diff --git a/server/src/services/agent-directory-working-copies.ts b/server/src/services/agent-directory-working-copies.ts index 007ee0ba576..919477391bc 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, isNull, sql } from "drizzle-orm"; +import { and, eq, inArray, 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,7 +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 { agentDirectoryBaselineDigest, agentDirectoryProbeProgram, agentDirectoryTransferCleanupProgram, observeLocalAgentDirectory } from "./agent-directory-probe.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; type Copy = typeof copies.$inferSelect; @@ -35,11 +35,10 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string const store = agentFileStore(db); async function transport(row: Copy, target: AdapterExecutionTarget, recovering: boolean) { if (target.kind !== "remote" || row.location !== `remote:${target.environmentId ?? ""}`) throw new Error("Agent directory environment changed"); - if (recovering && target.transport === "ssh") throw new Error("The original SSH collector is unavailable"); if (target.transport === "ssh") { // Reuse SSH's plain directory transfer without its task-workspace suffix // or Git-history discovery. The registered root is the exact writable root. - await syncDirectoryToSsh({ spec: target.spec, localDir: row.localRoot, remoteDir: row.executionRoot, exclude: [".paperclip-runtime"] }); + if (!recovering) await syncDirectoryToSsh({ spec: target.spec, localDir: row.localRoot, remoteDir: row.executionRoot, exclude: [".paperclip-runtime"] }); return { target, workspaceRemoteDir: row.executionRoot, runtimeRootDir: null, assetDirs: {}, additionalSourceDirs: {}, additionalSourceFailures: [], workspaceSyncSnapshot: null, restoreWorkspace: () => restoreWorkspaceFromSshExecution({ spec: target.spec, localDir: row.localRoot, @@ -53,11 +52,8 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string } 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); + // The same bounded program validates aliases on its actual host. + return observeLocalAgentDirectory(row.localRoot); } 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; @@ -142,7 +138,15 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string `node -e ${quote(instructionGitExcludeProgram)} ${quote(input.target.remoteCwd)}`, { cwd: input.target.remoteCwd, env: {}, timeoutSec: 15 }); 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)); + const runtime = await transport(row, input.target, false); + transports.set(key(row), runtime); + if (input.target.transport === "sandbox") { + if (runtime.runtimeRootDir !== path.posix.join(row.executionRoot, ".paperclip-runtime", "agent-files")) throw new Error("Agent directory transfer scratch path changed"); + const retired = await runAdapterExecutionTargetShellCommand(input.runId, input.target, + `node -e ${quote(agentDirectoryTransferCleanupProgram)} ${quote(row.executionRoot)}`, + { cwd: input.target.remoteCwd, env: {}, timeoutSec: 10 }); + if (retired.exitCode !== 0 || retired.timedOut) throw new Error("Agent directory transfer scratch could not be safely retired"); + } } const observed = await observe(row, input.target).catch(() => null); return await patch(row, { state: "prepared", receipt: { ...row.receipt, @@ -170,7 +174,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string } async function hasChanges(row: Copy, target?: AdapterExecutionTarget | null) { try { - if (row.receipt?.retainedByRunId || typeof row.receipt?.materializationIdentity !== "string") return true; + if (row.receipt?.retainedByRunId || row.receipt?.retirementRequired || 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 }); @@ -178,16 +182,47 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string } 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 }) { + async function adopt(input: { companyId: string; agentId: string; runId: string; previousRunId: string; target?: AdapterExecutionTarget | null; cwd: string; allowRetirementHandoff?: boolean }) { 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 oldRuntime = transports.get(key(prior)); + const remote = input.target?.kind === "remote" ? input.target : null; + const priorCleanup = prior.receipt?.cleanup as { leaseId?: string; remoteCwd?: string } | undefined; + const validLeases = (leases: Array) => { + if (!remote) return true; + const previous = leases.find(lease => lease.id === priorCleanup?.leaseId); + const successor = leases.find(lease => lease.id === remote.leaseId); + return Boolean(previous && successor && previous.id !== successor.id + && previous.companyId === input.companyId && successor.companyId === input.companyId + && previous.heartbeatRunId === input.previousRunId && successor.heartbeatRunId === input.runId + && previous.environmentId === remote.environmentId && successor.environmentId === remote.environmentId + && previous.provider && previous.provider === successor.provider + && previous.providerLeaseId && previous.providerLeaseId === successor.providerLeaseId + && successor.status === "active" && !successor.releasedAt + && !hasRemoteTerminationReceipt(previous) && !hasRemoteTerminationReceipt(successor) + && (remote.transport !== "sandbox" || remote.providerKey === successor.provider) + && (successor.metadata?.remoteCwd === undefined || successor.metadata.remoteCwd === remote.remoteCwd)); + }; + const leaseIds = remote && priorCleanup?.leaseId && remote.leaseId ? [priorCleanup.leaseId, remote.leaseId] : []; + if (remote && (!oldRuntime || oldRuntime.target.kind !== "remote" || oldRuntime.target.transport !== remote.transport + || oldRuntime.target.leaseId !== priorCleanup?.leaseId || oldRuntime.target.remoteCwd !== remote.remoteCwd + || priorCleanup?.remoteCwd !== remote.remoteCwd || leaseIds.length !== 2 + || !validLeases(await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), inArray(environmentLeases.id, leaseIds)))))) return null; 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 observedChange = await hasChanges(prior, input.target); + if (observedChange && !input.allowRetirementHandoff) return null; + prior = (await get(input.companyId, input.previousRunId))!; + // adopt_remote constructs a collector on the successor target without + // uploading the stale host mirror over the retained remote directory. + const replacement = remote ? await transport({ ...prior, runId: input.runId }, remote, true) : undefined; + let adopted: Copy | null; + try { adopted = await store.locked(input.companyId, input.agentId, bound, false, async (tx, _agent, canonical) => { + if (remote && !validLeases(await tx.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, input.companyId), inArray(environmentLeases.id, leaseIds))).for("update"))) return null; const current = await inspectAgentDirectory(canonical); - if (agentDirectoryBaselineDigest(current.snapshot) !== agentDirectoryBaselineDigest(baseline(prior))) return null; + const canonicalChanged = agentDirectoryBaselineDigest(current.snapshot) !== agentDirectoryBaselineDigest(baseline(prior)); + if (canonicalChanged && !input.allowRetirementHandoff) 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), @@ -195,20 +230,32 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string 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(), + receipt: { ...prior.receipt, materializationRunId: prior.receipt?.materializationRunId ?? prior.runId, + ...(remote ? { cleanup: { leaseId: remote.leaseId, remoteCwd: remote.remoteCwd } } : {}), + ...(observedChange || canonicalChanged ? { retirementRequired: true } : {}) }, updatedAt: new Date(), }).returning(); return next; - }); + }); } catch (error) { await replacement?.cleanupWorkspaceSnapshot?.(); throw error; } // 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); } - } + if (replacement && oldRuntime) { + const cleanup = replacement.cleanupWorkspaceSnapshot; + replacement.cleanupWorkspaceSnapshot = async () => { + try { await cleanup?.(); } finally { await oldRuntime.cleanupWorkspaceSnapshot?.(); } + }; + transports.delete(key(prior)); transports.set(key(adopted), replacement); + } + } else await replacement?.cleanupWorkspaceSnapshot?.(); 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))!; } + const remoteLease = row.location !== "local" ? await ownedRemoteLease(row) : null; + if (row.location !== "local" && (!remoteLease + || remoteLease.releasedAt || remoteLease.status !== "active" || hasRemoteTerminationReceipt(remoteLease))) { + return recoverStoppedRemote(row); + } 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. @@ -257,16 +304,25 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string let cleanupPending = false; await runtime?.cleanupWorkspaceSnapshot?.().catch(() => { cleanupPending = true; }); const cleanupTarget = runtime?.target ?? target; - if (row.processStoppedAt && completed.has(row.state) && cleanupTarget?.kind === "remote") { + const lease = row.location !== "local" ? await ownedRemoteLease(row) : null; + const terminated = lease && hasRemoteTerminationReceipt(lease); + const destroyed = terminated && (lease.metadata?.remoteExecutionTermination as { state?: string } | undefined)?.state === "destroyed"; + const cleanupLeaseId = (row.receipt?.cleanup as { leaseId?: string } | undefined)?.leaseId; + // Maintenance may run while the collector is still cached. A cached target + // cannot override the current lease's stopped/destroyed/uncertain state. + const remoteUnavailable = row.location !== "local" && !destroyed && (!lease || terminated + || Boolean(lease && (lease.releasedAt || lease.status !== "active")) + || Boolean(cleanupLeaseId && (!lease || cleanupTarget?.kind === "remote" && cleanupTarget.leaseId !== lease.id))); + if (remoteUnavailable) cleanupPending = true; + else if (!destroyed && row.processStoppedAt && completed.has(row.state) && cleanupTarget?.kind === "remote") { 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}`, { cwd: cleanupTarget.remoteCwd, env: {}, timeoutSec: 15 }).then(result => result.exitCode !== 0 || result.timedOut, () => true); cleanupPending ||= remoteCleanupFailed; - } else if (row.processStoppedAt && completed.has(row.state) && row.location !== "local") { - // Rebind only the original host-owned lease. Never acquire a replacement - // sandbox or persist transport credentials in a working-copy receipt. + } else if (!destroyed && row.processStoppedAt && completed.has(row.state) && row.location !== "local") { + // Rebind only the current host-owned lease. Never acquire a replacement. cleanupPending ||= !await cleanupRemoteAfterRestart(row).catch(() => false); } if (completed.has(row.state) && row.processStoppedAt) { @@ -277,23 +333,48 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string return; } await patch(row, { receipt: { schema: AGENT_FILES_CONTRACT, state: row.state, appliedCandidateHash: row.candidateHash, storageWarning: row.receipt?.storageWarning ?? null, cleanupPending, + materializationRunId: row.receipt?.materializationRunId ?? row.runId, ...(cleanupPending ? { cleanup: row.receipt?.cleanup } : {}) }, nextAttemptAt: cleanupPending ? new Date(Date.now() + 30_000) : null }); } } + async function ownedRemoteLease(row: Copy) { + const cleanup = row.receipt?.cleanup as { leaseId?: string } | undefined; + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, row.companyId), eq(environmentLeases.heartbeatRunId, row.runId), + cleanup?.leaseId ? eq(environmentLeases.id, cleanup.leaseId) : eq(environmentLeases.environmentId, row.location.slice("remote:".length)), + )); + if (leases.length !== 1) return null; + const lease = leases[0]!; + const destroyedWithoutEnvironment = lease.environmentId === null && hasRemoteTerminationReceipt(lease) + && (lease.metadata?.remoteExecutionTermination as { state?: string } | undefined)?.state === "destroyed"; + if (row.location !== `remote:${lease.environmentId}` && !destroyedWithoutEnvironment) return null; + return lease; + } + async function recoverStoppedRemote(row: Copy) { + const lease = await ownedRemoteLease(row); + const destroyed = lease && hasRemoteTerminationReceipt(lease) + && (lease.metadata?.remoteExecutionTermination as { state?: string } | undefined)?.state === "destroyed"; + if (!destroyed) { + // execute (including bypassSession) may restart a stopped sandbox. Keep + // the only remaining bytes; a stop receipt is not a destruction receipt. + return patch(row, { state: "pending_collection", errorCode: lease ? "INSTRUCTION_STOPPED_REMOTE_COLLECTION_PENDING" : "INSTRUCTION_REMOTE_LEASE_UNVERIFIED", + errorMessage: lease ? "The remote provider stopped with its agent directory retained. Safe stopped-file retrieval is unavailable; no save or discard is claimed." + : "The current remote lease could not be verified. Its agent directory is preserved; no retrieval, save or discard is claimed.", nextAttemptAt: null }); + } + row = await patch(row, { state: "unavailable", processStoppedAt: row.processStoppedAt ?? new Date(), + errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE", errorMessage: "The exact remote sandbox was destroyed before its agent directory was retrieved. No instruction save is claimed.", nextAttemptAt: null }); + await release(row); + return (await get(row.companyId, row.runId))!; + } async function cleanupRemoteAfterRestart(row: Copy): Promise { const cleanup = row.receipt?.cleanup as { leaseId?: string; remoteCwd?: string } | undefined; - // Older preview receipts did not record the lease ID. Accept only a unique - // lease still bound to this run and environment in that compatibility case. - const leases = await db.select().from(environmentLeases).where(and( - eq(environmentLeases.companyId, row.companyId), - cleanup?.leaseId ? eq(environmentLeases.id, cleanup.leaseId) : and( - eq(environmentLeases.environmentId, row.location.slice("remote:".length)), eq(environmentLeases.heartbeatRunId, row.runId)), - )); - if (leases.length !== 1) return false; - const lease = leases[0]!; + const lease = await ownedRemoteLease(row); + if (!lease) return false; const termination = lease.metadata?.remoteExecutionTermination as { state?: string } | undefined; - if (hasRemoteTerminationReceipt(lease) && termination?.state === "destroyed") return true; + if (hasRemoteTerminationReceipt(lease)) return termination?.state === "destroyed"; + // Never execute a cleanup command against a released/uncertain allocation. + if (lease.releasedAt || lease.status !== "active") return false; if (!environmentRuntime || !lease.environmentId || row.location !== `remote:${lease.environmentId}`) return false; const [environment] = await db.select().from(environments).where(eq(environments.id, lease.environmentId)); if (!environment) return false; @@ -313,6 +394,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string }); } return { prepare, hasChanges, adopt, + recoverStoppedRemote: (row: Copy) => serial(row, current => current.receipt?.retainedByRunId ? Promise.resolve(current) : recoverStoppedRemote(current)), 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 8a27c1f9ddf..1bda36f2029 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -311,16 +311,8 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen const result = await collectStopped({ companyId: row.companyId, runId: row.runId }); if (result && isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result); } else if (row.location !== "local" && await remoteExecutionHasStopped(db, row.companyId, row.runId)) { - // An unchanged-turn receipt protects a live warm owner. Only this - // independently stop-proved branch may turn lost remote bytes into a - // terminal no-save receipt and release the retained materialization. - const unavailable = isAgentDirectoryCopy(row) && row.state === "unchanged_turn" - ? await patch(row, { state: "unavailable", errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE", - errorMessage: "The remote provider stopped before its retained agent directory could be retrieved. No instruction save is claimed.", nextAttemptAt: null }) - : await reportUnavailable(row.companyId, row.runId); - if (unavailable && isAgentDirectoryCopy(unavailable)) { - await directories.release(await patch(unavailable, { processStoppedAt: unavailable.processStoppedAt ?? new Date() })); - } + if (isAgentDirectoryCopy(row)) await directories.recoverStoppedRemote(row); + else await reportUnavailable(row.companyId, row.runId); } else if (row.state === "prepared") { await patch(row, { state: "pending_collection", errorCode: "INSTRUCTION_STOP_UNCONFIRMED", errorMessage: "The provider's stop has not been confirmed. Instruction collection is pending; no save is claimed.", nextAttemptAt: null }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 60d3eb9c309..574ef6a8bf5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -221,6 +221,7 @@ import { cancelNativeSession, claimNativeRestartRecoveries, claimWarmNativeInstructionCopy, + nativeSessionWorkspaceScope, closeWarmNativeSessionsForEnvironment, closeIdleWarmNativeSessionsForRestart, currentNativeControllerIdentity, @@ -22423,7 +22424,10 @@ export function heartbeatService( 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, + workspace: nativeSessionWorkspaceScope({ + binding: { runId: run.id, executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id }, + workspace: executionWorkspace, + }), cwd: executionWorkspace.cwd, })).digest("hex"); const collectStoppedInstructions = async () => { if (!instructionCopy) return; @@ -23278,14 +23282,17 @@ export function heartbeatService( 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}`, + executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id, workspace: executionWorkspace, + environmentId: executionTarget?.environmentId ?? null, runId: run.id, preparationKey: instructionPreparationKey, forceRetirement: !taskSessionForRun, adopt: async previousRunId => { instructionCopy = await instructionCopies.adopt({ companyId: agent.companyId, agentId: agent.id, - runId: run.id, previousRunId, target: executionTarget, cwd: executionWorkspace.cwd }); + runId: run.id, previousRunId, target: executionTarget, cwd: executionWorkspace.cwd, allowRetirementHandoff: true }); return nativeInstructionWorkingCopy() ?? null; }, }); + // A collection-only handoff was retired. Re-enter normal + // preparation instead of composing a root already collected. + if (!releaseWarmInstructionPreparation) instructionCopy = null; } instructionCopy ??= await instructionCopies.prepare({ companyId: agent.companyId, agentId: agent.id, runId: run.id, 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 d306612e2f8..056d061a132 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { readFileSync } from "node:fs"; import { access, + chmod, cp, lstat, mkdir, @@ -35,6 +36,7 @@ import { validatePrpEvent, parseNativeExecutionInput, type NativeExecutionInputV1, + type NativeExecutionInputV3, type NativeExecutionInput, type PrpEvent, } from "@paperclipai/paperclip-runner"; @@ -282,6 +284,7 @@ import { cancelNativeSession, closeWarmNativeSessionsForEnvironment, claimWarmNativeInstructionCopy, + nativeSessionWorkspaceScope, closeIdleWarmNativeSessionsForRestart, createGovernedWaitEventObservation, createRemoteRunnerProcessLauncher, @@ -5977,7 +5980,251 @@ 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 => { + it.each(["collect", "stopped", "destroyed", "missing-lease", "foreign-run", "foreign-environment", "deleted-environment", "claim-edit", "claim-add", "claim-remove", "claim-remove-entry", "claim-canonical", "claim-config", "claim-uncertain"])("composes remote warm ownership with real DB successor leases and current-run collection: %s", async ending => { + const tables = await import("@paperclipai/db"); + const { eq } = await import("drizzle-orm"); + const { randomUUID } = await import("node:crypto"); + const { promisify } = await import("node:util"); + const { execFile } = await import("node:child_process"); + const executeCommand = promisify(execFile); + const { startEmbeddedPostgresTestDatabase } = await import("../../__tests__/helpers/embedded-postgres.js"); + const { agentInstructionWorkingCopyService } = await import("../agent-instruction-working-copies.js"); + const { buildNativeRuntimeContext } = await import("./runtime-context.js"); + const { prepareNativeHeartbeatRun } = await import("./prepare-native-run.js"); + const { buildNativeCompletionContract } = await import("./completion-contracts.js"); + const home = await realpath(await mkdtemp(join(tmpdir(), "warm-remote-db-"))); + const previousHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = home; + const database = await startEmbeddedPostgresTestDatabase("warm-remote-db-"); + const db = tables.createDb(database.connectionString); + const companyId = randomUUID(), agentId = randomUUID(), userId = randomUUID(), environmentId = randomUUID(), issueId = randomUUID(); + const { resolveManagedInstructionsRoot } = await import("../agent-instructions.js"); + const canonical = resolveManagedInstructionsRoot({ id: agentId, companyId, name: "Warm remote", adapterConfig: {} }), remoteCwd = join(home, "remote"); + let currentLease = "", failCleanup = false, failProbe = false, closed = false; + const commands: Array<{ leaseId: string; command: string }> = []; + const commandRunner = (leaseId: string): import("@paperclipai/adapter-utils/command-managed-runtime").CommandManagedRuntimeRunner => ({ execute: async input => { + expect(leaseId).toBe(currentLease); // The replaced collector must never execute. + commands.push({ leaseId, command: `${input.command} ${(input.args ?? []).join(" ")}` }); + if (failProbe) { + failProbe = false; + expect(commands.at(-1)!.command).toContain("Agent directory probe deadline"); + return { exitCode: 1, signal: null, timedOut: false, stdout: "", stderr: "injected uncertain observation", pid: null, startedAt: new Date().toISOString() }; + } + if (failCleanup && commands.at(-1)!.command.includes("rm -rf --")) { + failCleanup = false; + return { exitCode: 1, signal: null, timedOut: false, stdout: "", stderr: "injected cleanup failure", pid: null, startedAt: new Date().toISOString() }; + } + const env = { PATH: process.env.PATH, ...input.env }, args = [...(input.args ?? [])]; + if (input.stdin != null && (args[0] === "-c" || args[0] === "-lc")) { + Object.assign(env, { PAPERCLIP_TEST_STDIN: input.stdin }); + args[1] = `printf '%s' "$PAPERCLIP_TEST_STDIN" | (${args[1]})`; + } + const result = await executeCommand(input.command, args, { cwd: input.cwd, env, timeout: input.timeoutMs, maxBuffer: 32 * 1024 * 1024 }); + return { exitCode: 0, signal: null, timedOut: false, ...result, pid: null, startedAt: new Date().toISOString() }; + } }); + let copies = agentInstructionWorkingCopyService(db); + const agent = { id: agentId, companyId, name: "Warm remote", adapterConfig: { + instructionsBundleMode: "managed", instructionsRootPath: canonical, instructionsEntryFile: "AGENTS.md" } }; + let active: Awaited>; + const session = { close: vi.fn(async () => { closed = true; }) }; + state.execute.mockReset().mockImplementation(async options => { + await options.onSession?.(session); + return { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" }, + turnId: "warm-turn", normalizedSessionId: environmentId, providerSessionId: environmentId, + driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + let bodyFailure: unknown; + try { + await db.insert(tables.companies).values({ id: companyId, name: "Warm tests", issuePrefix: "WARM" }); + await db.insert(tables.authUsers).values({ id: userId, name: "Editor", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(tables.agents).values(agent); + await db.insert(tables.companyMemberships).values([{ companyId, principalType: "user", principalId: userId, membershipRole: "operator" }, { companyId, principalType: "agent", principalId: agentId, membershipRole: "member" }]); + await db.insert(tables.principalPermissionGrants).values({ companyId, principalType: "user", principalId: userId, permissionKey: "agents:configure", scope: { agentIds: [agentId] } }); + await db.insert(tables.environments).values({ id: environmentId, name: "Owned fake remote", driver: "sandbox" }); + const [issue] = await db.insert(tables.issues).values({ id: issueId, companyId, title: "Warm remote", status: "in_progress", assigneeAgentId: agentId }).returning(); + await mkdir(canonical, { recursive: true }); await mkdir(remoteCwd); + await writeFile(join(canonical, "AGENTS.md"), "Retain this exact directory.\n"); + await writeFile(join(canonical, "optional-memory.txt"), "Optional memory\n"); + let prior: NativeExecutionInputV3 | undefined; + let originalRoot = "", materializationRunId = ""; + for (let turn = 0; turn < 3; turn++) { + const runId = randomUUID(); + if (currentLease) { + await db.update(tables.environmentLeases).set({ status: "expired", releasedAt: new Date(), cleanupStatus: "success" }).where(eq(tables.environmentLeases.id, currentLease)); + await db.update(tables.heartbeatRuns).set({ status: "succeeded" }).where(eq(tables.heartbeatRuns.id, prior!.binding.runId)); + } + currentLease = randomUUID(); + const [run] = await db.insert(tables.heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", responsibleUserId: userId, + status: "running", runtimeMode: "native", nativeIssueId: issueId, nativeSessionId: environmentId, runnerInstanceId: environmentId, + contextSnapshot: { issueId } }).returning(); + await db.insert(tables.nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed" }); + await db.update(tables.issues).set({ executionRunId: runId }).where(eq(tables.issues.id, issueId)); + await db.insert(tables.environmentLeases).values({ id: currentLease, companyId, environmentId, heartbeatRunId: runId, + provider: "daytona", providerLeaseId: "same-owned-allocation", leasePolicy: "reuse_by_environment", metadata: { remoteCwd } }); + // Match the production preparation seam and existing native DB tests: + // pin the issue's completion contract and durable run/lease identities. + await prepareNativeHeartbeatRun({ db, run: run!, issue: issue!, environmentLeaseId: currentLease }); + const [contract] = await db.select().from(tables.completionContracts).where(eq(tables.completionContracts.issueId, issueId)); + const target = { kind: "remote" as const, transport: "sandbox" as const, providerKey: "daytona", environmentId, leaseId: currentLease, remoteCwd, runner: commandRunner(currentLease) }; + const current: NativeExecutionInputV3 = { ...execution, schema: "paperclip.native-execution-input.v3", + executionMode: "default", planningContext: null, runtimeContext: nativeRuntimeContextFixture(), + completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, binding: { companyId, agentId, issueId, runId, executionWorkspaceId: runId }, + workspace: { cwd: remoteCwd, repoUrl: null, repoRef: null, branchName: null }, + session: { ...execution.session, normalizedSessionId: environmentId, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } } }; + const preparationKey = turn === 1 && ending === "claim-config" ? "changed-config" : "same-config"; + let retiredReceipt: Record | null = null; + const capability = () => ({ runId, preparationKey, hasChanges: () => copies.hasChanges({ companyId, runId, target }), + collectStopped: async () => { expect(closed).toBe(true); retiredReceipt = (await copies.get(companyId, runId))?.receipt ?? null; await copies.collectStopped({ companyId, runId, target }); } }); + let release: (() => Promise) | null = null; + if (prior) { + // Same environment is insufficient: reject a different physical allocation. + await db.update(tables.environmentLeases).set({ providerLeaseId: "foreign-allocation" }).where(eq(tables.environmentLeases.id, currentLease)); + const before = commands.length; + expect(await copies.adopt({ companyId, agentId, runId, previousRunId: prior.binding.runId, target, cwd: remoteCwd })).toBeNull(); + expect(commands).toHaveLength(before); + await db.update(tables.environmentLeases).set({ providerLeaseId: "same-owned-allocation" }).where(eq(tables.environmentLeases.id, currentLease)); + if (ending === "claim-edit") await writeFile(join(originalRoot, "AGENTS.md"), "Remote edit before next claim\n"); + if (ending === "claim-add") await writeFile(join(originalRoot, "memory.txt"), "Remote addition before next claim\n"); + if (ending === "claim-remove") await rm(join(originalRoot, "optional-memory.txt")); + if (ending === "claim-remove-entry") await rm(join(originalRoot, "AGENTS.md")); + if (ending === "claim-canonical") await writeFile(join(canonical, "AGENTS.md"), "Canonical edit before next claim\n"); + if (["claim-edit", "claim-add", "claim-remove", "claim-remove-entry", "claim-canonical"].includes(ending)) { + // Ordinary adoption remains unchanged-only and cannot alias this owner. + expect(await copies.adopt({ companyId, agentId, runId, previousRunId: prior.binding.runId, target, cwd: remoteCwd })).toBeNull(); + expect((await copies.get(companyId, prior.binding.runId))?.receipt?.retainedByRunId).toBeUndefined(); + } + failProbe = ending === "claim-uncertain"; + release = await claimWarmNativeInstructionCopy({ priorExecution: prior, companyId, agentId, executionWorkspaceId: runId, + workspace: current.workspace, environmentId, runId, preparationKey, adopt: async previousRunId => { + active = await copies.adopt({ companyId, agentId, runId, previousRunId, target, cwd: remoteCwd, allowRetirementHandoff: true }); + return active ? capability() : null; + } }); + if (ending.startsWith("claim-")) { + expect(release).toBeNull(); + expect(session.close).toHaveBeenCalledOnce(); + expect(state.execute).toHaveBeenCalledOnce(); // No reuse was admitted. + if (ending !== "claim-config") expect(retiredReceipt).toMatchObject({ retirementRequired: true }); + expect(retiredReceipt).toMatchObject({ materializationRunId, cleanup: { leaseId: currentLease, remoteCwd } }); + expect((await copies.get(companyId, runId))?.processStoppedAt).toBeInstanceOf(Date); + const compact = (await copies.get(companyId, runId))!.receipt; + expect(compact).toMatchObject({ materializationRunId, cleanupPending: false }); + expect(compact?.baseline).toBeUndefined(); + await expect(access(originalRoot)).rejects.toMatchObject({ code: "ENOENT" }); + if (ending === "claim-edit") expect(await readFile(join(canonical, "AGENTS.md"), "utf8")).toBe("Remote edit before next claim\n"); + if (ending === "claim-add") expect(await readFile(join(canonical, "memory.txt"), "utf8")).toBe("Remote addition before next claim\n"); + if (ending === "claim-remove") await expect(access(join(canonical, "optional-memory.txt"))).rejects.toMatchObject({ code: "ENOENT" }); + if (ending === "claim-remove-entry") { + expect((await copies.get(companyId, runId))!).toMatchObject({ state: "unavailable", errorCode: "AGENT_FILES_SAVE_FAILED" }); + expect(await readFile(join(canonical, "AGENTS.md"), "utf8")).toBe("Retain this exact directory.\n"); + } + if (ending === "claim-canonical") expect(await readFile(join(canonical, "AGENTS.md"), "utf8")).toBe("Canonical edit before next claim\n"); + // Heartbeat rematerializes only after the successor proved retirement. + active = await copies.prepare({ companyId, agentId, runId, target, cwd: remoteCwd }); + expect(active).toMatchObject({ state: "prepared", executionRoot: originalRoot, processStoppedAt: null }); + expect(active!.receipt?.retirementRequired).toBeUndefined(); + expect(await copies.hasChanges({ companyId, runId, target })).toBe(false); + await copies.reportUnavailable(companyId, prior.binding.runId); + await copies.release(companyId, prior.binding.runId); + expect((await lstat(originalRoot)).isDirectory()).toBe(true); + await copies.collectStopped({ companyId, runId, target }); + return; + } + expect(release).toBeTypeOf("function"); + await copies.reportUnavailable(companyId, prior.binding.runId); + await copies.release(companyId, prior.binding.runId); + expect(active!.receipt).toMatchObject({ materializationRunId, cleanup: { leaseId: currentLease, remoteCwd } }); + } else { + active = await copies.prepare({ companyId, agentId, runId, target, cwd: remoteCwd }); + originalRoot = active!.executionRoot; materializationRunId = runId; + } + expect(active!.executionRoot).toBe(originalRoot); + current.runtimeContext = await buildNativeRuntimeContext({ db, agent, runId, runtimeConfig: {}, runtimeSkillEntries: [], + instructionWorkingCopy: { rootPath: active!.executionRoot, entryPath: "AGENTS.md", kind: "agent_files" } }); + expect(current.runtimeContext.instructions.workingCopy?.rootPath).toBe(originalRoot); + await executePaperclipNativeSession({ db, execution: current, runnerInstanceId: environmentId, runnerExecutionTarget: target, instructionWorkingCopy: capability() }); + if (prior) expect(state.execute.mock.calls.at(-1)?.[0].existingSession).toBe(session); + expect(closed).toBe(false); + await release?.(); expect(closed).toBe(false); + prior = current; + } + // A write after the live unchanged probe is still collected on retirement. + await writeFile(join(originalRoot, "late-memory.txt"), "Latest owner's remote bytes"); + if (["missing-lease", "foreign-run", "foreign-environment", "deleted-environment"].includes(ending)) { + if (ending === "missing-lease") await db.delete(tables.environmentLeases).where(eq(tables.environmentLeases.id, currentLease)); + if (ending === "foreign-run") await db.update(tables.environmentLeases).set({ heartbeatRunId: materializationRunId }).where(eq(tables.environmentLeases.id, currentLease)); + if (ending === "foreign-environment") { + const other = randomUUID(); await db.insert(tables.environments).values({ id: other, name: "Other", driver: "sandbox" }); + await db.update(tables.environmentLeases).set({ environmentId: other }).where(eq(tables.environmentLeases.id, currentLease)); + } + if (ending === "deleted-environment") await db.delete(tables.environments).where(eq(tables.environments.id, environmentId)); + const before = commands.length; + await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "lease identity changed" }); + expect(commands).toHaveLength(before); + const pending = (await copies.get(companyId, prior!.binding.runId))!; + expect(pending.state).toBe("pending_collection"); + expect(pending.receipt?.baseline).toBeDefined(); + expect(await readFile(join(originalRoot, "late-memory.txt"), "utf8")).toBe("Latest owner's remote bytes"); + return; + } + if (ending !== "collect") { + const { remoteTerminationReceipt } = await import("../remote-execution-termination.js"); + const [lease] = await db.select().from(tables.environmentLeases).where(eq(tables.environmentLeases.id, currentLease)); + await db.update(tables.environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", + metadata: { ...lease!.metadata, remoteExecutionTermination: remoteTerminationReceipt(lease!, { providerLeaseId: lease!.providerLeaseId, state: ending }) } }).where(eq(tables.environmentLeases.id, currentLease)); + await db.update(tables.heartbeatRuns).set({ status: "succeeded" }).where(eq(tables.heartbeatRuns.id, prior!.binding.runId)); + const before = commands.length; + await copies.recoverStopped(); // The original remote collector is still cached. + expect(commands).toHaveLength(before); + const recovered = (await copies.get(companyId, prior!.binding.runId))!; + expect(recovered.state).toBe(ending === "destroyed" ? "unavailable" : "pending_collection"); + await expect(access(join(canonical, "late-memory.txt"))).rejects.toMatchObject({ code: "ENOENT" }); + if (ending === "stopped") { + expect(await readFile(join(originalRoot, "late-memory.txt"), "utf8")).toBe("Latest owner's remote bytes"); + expect(recovered.receipt?.baseline).toBeDefined(); + await copies.collectStopped({ companyId, runId: prior!.binding.runId }); + await copies.release(companyId, prior!.binding.runId); + expect(commands).toHaveLength(before); + } else await expect(access(active!.localRoot)).rejects.toMatchObject({ code: "ENOENT" }); + await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "already terminated owner" }); + expect(commands).toHaveLength(before); + return; + } + failCleanup = true; + await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "test retirement" }); + expect(session.close).toHaveBeenCalledOnce(); + expect(await readFile(join(canonical, "late-memory.txt"), "utf8")).toBe("Latest owner's remote bytes"); + const saved = (await copies.get(companyId, prior!.binding.runId))!; + expect(saved).toMatchObject({ state: "saved", receipt: { cleanupPending: true, materializationRunId, cleanup: { leaseId: currentLease } } }); + const cleanup = vi.fn(async input => { + expect(input.lease.id).toBe(currentLease); + expect(input.lease.heartbeatRunId).toBe(prior!.binding.runId); + return commandRunner(currentLease).execute(input); + }); + copies = agentInstructionWorkingCopyService(db, { environmentRuntime: { execute: cleanup } as never }); + await db.update(tables.agentInstructionWorkingCopies).set({ nextAttemptAt: null }).where(eq(tables.agentInstructionWorkingCopies.runId, prior!.binding.runId)); + await copies.recoverCaptured(); + expect(cleanup).toHaveBeenCalledOnce(); + await expect(access(originalRoot)).rejects.toMatchObject({ code: "ENOENT" }); + expect((await copies.get(companyId, prior!.binding.runId))!.receipt?.cleanupPending).toBe(false); + } catch (error) { bodyFailure = error; throw error; } + finally { + const cleanupFailures: unknown[] = []; + try { await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "test cleanup" }); } catch (error) { cleanupFailures.push(error); } + try { await database.cleanup(); } catch (error) { cleanupFailures.push(error); } + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + try { + const writable = async (directory: string): Promise => { + await chmod(directory, 0o700); + for (const entry of await readdir(directory, { withFileTypes: true })) if (entry.isDirectory()) await writable(join(directory, entry.name)); + }; + await writable(home); await rm(home, { recursive: true, force: true }); + } catch (error) { cleanupFailures.push(error); } + if (cleanupFailures.length) throw new AggregateError([...(bodyFailure ? [bodyFailure] : []), ...cleanupFailures], "Warm DB test body/cleanup failure"); + } + }, 90_000); + + it.each(["idle", "edit", "add", "remove", "projectless"])("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 }); @@ -5999,10 +6246,10 @@ describe("native warm session supervision", () => { 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); + if (ending === "idle" || ending === "projectless") await copies.collectStopped(row); }; const identity = "warm-real-agent-files"; - const warmExecution = { ...execution, binding: { ...execution.binding, runId: identity, executionWorkspaceId: `${identity}-workspace` }, + const warmExecution = { ...execution, binding: { ...execution.binding, runId: identity, executionWorkspaceId: ending === "projectless" ? identity : `${identity}-workspace` }, session: { ...execution.session, normalizedSessionId: identity, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } }, } as NativeExecutionInputV1; const session = { close }; @@ -6019,11 +6266,26 @@ describe("native warm session supervision", () => { 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 second = { ...warmExecution, binding: { ...warmExecution.binding, runId: `${identity}-second`, + ...(ending === "projectless" ? { executionWorkspaceId: `${identity}-second` } : {}) } }; + expect(nativeSessionWorkspaceScope(second)).toEqual(nativeSessionWorkspaceScope(warmExecution)); const currentCopy = { runId: second.binding.runId, preparationKey: "same-preparation", hasChanges: () => copies.hasChanges(row), collectStopped }; + if (ending === "projectless") { + const adopt = vi.fn(async () => currentCopy); + for (const changed of [ + { ...second.workspace, cwd: "/different" }, { ...second.workspace, repoUrl: "https://example.test/other" }, + { ...second.workspace, repoRef: "other" }, { ...second.workspace, branchName: "other" }, + ]) expect(await claimWarmNativeInstructionCopy({ priorExecution: warmExecution, + companyId: second.binding.companyId, agentId: second.binding.agentId, executionWorkspaceId: second.binding.runId, + workspace: changed, environmentId: identity, runId: second.binding.runId, preparationKey: "same-preparation", adopt })).toBeNull(); + expect(await claimWarmNativeInstructionCopy({ priorExecution: warmExecution, + companyId: second.binding.companyId, agentId: second.binding.agentId, executionWorkspaceId: "managed-workspace", + workspace: second.workspace, environmentId: identity, runId: second.binding.runId, preparationKey: "same-preparation", adopt })).toBeNull(); + expect(adopt).not.toHaveBeenCalled(); expect(close).not.toHaveBeenCalled(); + } const release = await claimWarmNativeInstructionCopy({ priorExecution: warmExecution, companyId: warmExecution.binding.companyId, agentId: warmExecution.binding.agentId, - executionWorkspaceId: warmExecution.binding.executionWorkspaceId, cwd: warmExecution.workspace.cwd, + executionWorkspaceId: second.binding.executionWorkspaceId, workspace: second.workspace, environmentId: identity, runId: second.binding.runId, preparationKey: "same-preparation", adopt: async () => currentCopy }); expect(release).toBeTypeOf("function"); @@ -6035,19 +6297,19 @@ describe("native warm session supervision", () => { expect(row.executionRoot).toBe(localRoot); await release?.(); // Consumed preparation cannot retire the retained owner. expect(close).not.toHaveBeenCalled(); - if (ending === "idle") { + if (ending === "idle" || ending === "projectless") { 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); + const adopt = vi.fn(async () => ({ ...currentCopy, runId: `${identity}-third` })); expect(await claimWarmNativeInstructionCopy({ priorExecution: second, companyId: second.binding.companyId, agentId: second.binding.agentId, - executionWorkspaceId: second.binding.executionWorkspaceId, cwd: second.workspace.cwd, + executionWorkspaceId: second.binding.executionWorkspaceId, workspace: second.workspace, environmentId: identity, runId: `${identity}-third`, preparationKey: "same-preparation", adopt })).toBeNull(); - expect(adopt).not.toHaveBeenCalled(); + expect(adopt).toHaveBeenCalledOnce(); } expect(order).toEqual(["closed", "collected"]); } finally { @@ -6079,7 +6341,7 @@ describe("native warm session supervision", () => { 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, + executionWorkspaceId: first.binding.executionWorkspaceId, workspace: first.workspace, environmentId: identity, runId: second.binding.runId, preparationKey: "key", adopt }; try { await executePaperclipNativeSession({ db: leaseDb(first), execution: first, runnerInstanceId: identity, runnerExecutionTarget: target, instructionWorkingCopy: copy }); @@ -6111,6 +6373,56 @@ describe("native warm session supervision", () => { } finally { allowClose = true; await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" }); } }); + it.each(["config", "reset", "changed", "unavailable", "auth", "wrong-capability", "retirement-failure"])("hands off collection before probing or retiring a warm owner: %s", async scenario => { + const identity = `warm-successor-${scenario}`, order: string[] = []; + let allowClose = scenario !== "retirement-failure"; + const session = { close: vi.fn(async () => { order.push("stop"); if (!allowClose) throw new Error("stop unconfirmed"); }) }; + const priorCopy = { runId: identity, preparationKey: "before", hasChanges: vi.fn(async () => false), + collectStopped: vi.fn(async () => { order.push("prior-pending"); }), retirementFailed: vi.fn(async () => {}) }; + 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 current = { runId: `${identity}-next`, preparationKey: scenario === "config" ? "after" : "before", + hasChanges: vi.fn(async () => { order.push("current-probe"); return true; }), + collectStopped: vi.fn(async () => { order.push("current-collect"); }), retirementFailed: vi.fn(async () => {}) }; + const adopt = vi.fn(async () => { + order.push("handoff"); + if (scenario === "auth") throw new Error("authorization denied"); + if (scenario === "unavailable") return null; + return scenario === "wrong-capability" ? { ...current, runId: "foreign" } : current; + }); + try { + await executePaperclipNativeSession({ db: leaseDb(first), execution: first, runnerInstanceId: identity, runnerExecutionTarget: target, instructionWorkingCopy: priorCopy }); + priorCopy.hasChanges.mockClear(); + const claiming = claimWarmNativeInstructionCopy({ priorExecution: first, companyId: first.binding.companyId, agentId: first.binding.agentId, + executionWorkspaceId: first.binding.executionWorkspaceId, workspace: first.workspace, environmentId: identity, + runId: current.runId, preparationKey: current.preparationKey, forceRetirement: scenario === "reset", adopt }); + if (["unavailable", "auth", "wrong-capability", "retirement-failure"].includes(scenario)) { + await expect(claiming).rejects.toThrow(({ unavailable: "handoff_unavailable", auth: "authorization denied", + "wrong-capability": "handoff_mismatch", "retirement-failure": "stop unconfirmed" } as Record)[scenario]); + } else expect(await claiming).toBeNull(); + expect(priorCopy.hasChanges).not.toHaveBeenCalled(); + expect(state.execute).toHaveBeenCalledOnce(); + if (["unavailable", "auth", "wrong-capability"].includes(scenario)) { + expect(order).toEqual(["handoff", "stop", "prior-pending"]); + expect(current.collectStopped).not.toHaveBeenCalled(); + } else { + expect(priorCopy.collectStopped).not.toHaveBeenCalled(); + expect(order).toEqual(["handoff", ...(["config", "reset"].includes(scenario) ? [] : ["current-probe"]), "stop", ...(scenario === "retirement-failure" ? [] : ["current-collect"])]); + if (scenario === "retirement-failure") { + expect(current.retirementFailed).toHaveBeenCalledOnce(); + expect(current.collectStopped).not.toHaveBeenCalled(); + } + } + } 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 475275d89a6..d68805145c9 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -447,15 +447,18 @@ export async function claimWarmNativeInstructionCopy(input: { companyId: string; agentId: string; executionWorkspaceId: string; - cwd: string; + workspace: NativeExecutionInput["workspace"]; environmentId: string | null; runId: string; preparationKey: string; + forceRetirement?: boolean; 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; + || JSON.stringify(nativeSessionWorkspaceScope(prior)) !== JSON.stringify(nativeSessionWorkspaceScope({ + binding: { ...prior.binding, runId: input.runId, executionWorkspaceId: input.executionWorkspaceId }, workspace: input.workspace, + })) || prior.workspace.cwd !== input.workspace.cwd) return null; const scope = nativeSessionScopeKey(prior); const entry = warmNativeSessions.get(scope); if (!entry || !entry.instructionWorkingCopy?.runId || entry.environmentId !== input.environmentId) return null; @@ -472,17 +475,21 @@ export async function claimWarmNativeInstructionCopy(input: { } finally { entry.busy = false; } }; try { - if (entry.closeOnReleaseReason !== undefined || entry.instructionWorkingCopy.preparationKey !== input.preparationKey || await entry.instructionWorkingCopy.hasChanges()) { + const preparationChanged = input.forceRetirement === true || entry.instructionWorkingCopy.preparationKey !== input.preparationKey; + // The prior capability may close over an expired remote lease. Rebind + // collection authority first; adoption itself grants no reuse permission. + const adopted = await input.adopt(entry.instructionWorkingCopy.runId); + if (!adopted) { + await retire("warm instruction materialization cannot be handed off"); + throw new Error("native_instruction_materialization_handoff_unavailable"); + } + if (adopted.runId !== input.runId || adopted.preparationKey !== input.preparationKey) throw new Error("native_instruction_materialization_handoff_mismatch"); + entry.instructionWorkingCopy = adopted; + entry.instructionPreparationRunId = input.runId; + if (entry.closeOnReleaseReason !== undefined || preparationChanged || await adopted.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"); }; @@ -1556,7 +1563,7 @@ function nativeSessionKey(execution: NativeExecutionInput): string { ); } -function nativeSessionWorkspaceScope(execution: NativeExecutionInput) { +export function nativeSessionWorkspaceScope(execution: { binding: Pick; workspace: Pick }) { // Projectless local runs use the heartbeat run id as a durable placeholder // rather than fabricating an execution_workspaces row. Do not let that // per-run placeholder break continuity for the same provider session; the