Integrate persistent computer owners with native and legacy environment execution

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-10 08:31:36 -05:00
1 parent 72bf0c61c6
commit 1dcf713e21
12 files changed
+340 -59

No files matched your search

@@ -0,0 +1,129 @@
import { and, eq } from "drizzle-orm";
import { environmentLeases } from "@paperclipai/db";
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
import { instanceSettingsService } from "./instance-settings.js";
import { resolvePaperclipRunnerIdleTimeoutMs } from "@paperclipai/adapter-utils";
import { createHash } from "node:crypto";
import type { Db } from "@paperclipai/db";
import type { EnvironmentLease } from "@paperclipai/shared";
import type { AdapterComputerExecutionTarget, EffectiveExecutionCapabilities } from "@paperclipai/adapter-utils/execution-target";
import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-connectivity";
import { computerService } from "../modules/computers/index.js";
import { environmentService } from "./environments.js";
import type { EnvironmentRuntimeDriver } from "./environment-runtime.js";
export const COMPUTER_CAPABILITIES: EffectiveExecutionCapabilities = {
reusableLeases: true, nativeSyncIn: false, nativeSyncOut: false,
persistentProcessSessions: true, independentControlCommands: true,
incrementalSessionOutput: false, concurrentSyncOperations: false,
duplexCommandStream: false, runnerWebSocketIngress: true,
};
export function computerOwnerFromLease(lease: EnvironmentLease) {
const value = lease.metadata?.computerOwner as Record<string, unknown> | undefined;
if (!value || typeof value.computerId !== "string" || typeof value.ownerId !== "string" ||
!Number.isSafeInteger(value.generation) || Number(value.generation) < 1 ||
lease.providerLeaseId !== value.ownerId) throw new Error("computer_owner_binding_invalid");
return { computerId: value.computerId, ownerId: value.ownerId, generation: Number(value.generation) };
}
export function createComputerEnvironmentDriver(db: Db): EnvironmentRuntimeDriver {
const computers = computerService(db);
const leases = environmentService(db);
const scope = (lease: EnvironmentLease) => {
if (!lease.environmentId) throw new Error("computer_environment_missing");
return { companyId: lease.companyId, environmentId: lease.environmentId };
};
const recover = (lease: EnvironmentLease) => computers.recover({ ...scope(lease), owner: computerOwnerFromLease(lease) });
const retire = async (lease: EnvironmentLease, status: "released" | "expired" | "failed") => {
const result = await computers.retire({ ...scope(lease), owner: computerOwnerFromLease(lease) });
if (!result.retired) throw new Error("computer_owner_retirement_superseded");
return leases.releaseLease(lease.id, status, { cleanupStatus: "success",
remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) });
};
return {
driver: "computer",
async acquireRunLease(input) {
if (!(await instanceSettingsService(db).getExperimental()).enableBoatEnvironments) throw new Error("boat_environments_disabled");
if (!input.agentId || !input.heartbeatRunId) throw new Error("computer_agent_run_required");
const idleTimeoutMs = resolvePaperclipRunnerIdleTimeoutMs(input.environment.config.runnerIdleTimeoutMs);
const binding = await computers.admit({ companyId: input.companyId, environmentId: input.environment.id,
agentId: input.agentId, runId: input.heartbeatRunId, idleTimeoutMs,
sessionKey: createHash("sha256").update(JSON.stringify([input.agentId, input.executionWorkspaceId ?? input.issueId, input.adapterType])).digest("hex") });
try {
// The durable owner advanced generation; prior retained bookkeeping no
// longer owns the process and must not be reaped as an allocation.
await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", updatedAt: new Date() })
.where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, input.environment.id),
eq(environmentLeases.providerLeaseId, binding.owner.ownerId), eq(environmentLeases.status, "retained")));
return await leases.acquireLease({ companyId: input.companyId, environmentId: input.environment.id,
executionWorkspaceId: input.executionWorkspaceId, issueId: input.issueId, heartbeatRunId: input.heartbeatRunId,
leasePolicy: "reuse_by_environment", provider: "boat", providerLeaseId: binding.owner.ownerId,
metadata: { driver: "computer", agentId: input.agentId, executionWorkspaceMode: input.executionWorkspaceMode,
computerOwner: binding.owner, listenerPort: binding.listenerPort, remoteCwd: binding.remoteCwd,
fileAuthority: { kind: "remote-persistent", placementId: binding.placementId, root: binding.remoteCwd, agentHome: binding.agentHome } } });
} catch (error) {
await computers.retire({ companyId: input.companyId, environmentId: input.environment.id, owner: binding.owner });
throw error;
}
},
releaseRunLease: ({ lease, status }) => retire(lease, status),
destroyRunLease: ({ lease }) => retire(lease, "released"),
async resumeRunLease({ lease }) { await recover(lease); return lease; },
async retryPendingSandboxTeardown({ lease }) {
const result = await computers.retire({ ...scope(lease), owner: computerOwnerFromLease(lease) });
if (!result.retired) throw new Error("computer_owner_retirement_superseded");
return { providerLeaseId: lease.providerLeaseId, state: "stopped" };
},
async realizeWorkspace({ lease, workspace }) {
const request = workspace.metadata?.workspaceRealizationRequest as { source?: { projectId?: string; repoUrl?: string; branchName?: string; strategy?: string }; issueId?: string } | undefined;
const result = await computers.realizeWorkspace({ ...scope(lease), owner: computerOwnerFromLease(lease),
projectId: request?.source?.projectId ?? undefined, taskId: request?.issueId ?? undefined,
repositoryUrl: request?.source?.repoUrl ?? undefined, branch: request?.source?.branchName ?? undefined,
mode: request?.source?.strategy === "git_worktree" ? "worktree" : "shared" });
await leases.updateLeaseMetadata(lease.id, { ...lease.metadata, remoteCwd: result.remoteCwd,
fileAuthority: { kind: "remote-persistent", placementId: result.placementId, root: result.remoteCwd, agentHome: result.agentHome } });
return { cwd: result.remoteCwd, metadata: { workspaceRealization: { mode: "in_place", authoritativeRoot: result.remoteCwd,
pathAliases: [], outboundRestorePaths: [] } } };
},
async execute({ lease, command, args, cwd, env, stdin, timeoutMs, onLog }) {
const binding = await recover(lease);
return binding.runner.execute({ command, args, cwd: cwd ?? binding.remoteCwd, env, stdin, timeoutMs, onLog });
},
async resolveCapabilities() { return COMPUTER_CAPABILITIES; },
};
}
export async function computerExecutionTarget(db: Db, lease: EnvironmentLease, idleTimeoutMs = resolvePaperclipRunnerIdleTimeoutMs(undefined)): Promise<AdapterComputerExecutionTarget> {
if (!lease.environmentId) throw new Error("computer_environment_missing");
const computers = computerService(db);
const scope = { companyId: lease.companyId, environmentId: lease.environmentId };
const owner = computerOwnerFromLease(lease);
const binding = await computers.recover({ ...scope, owner });
const ingress = async (path: string): Promise<RunnerIngressEndpoint> => {
const endpoint = await binding.ingress({ path });
return { kind: "authenticated_websocket", websocketUrl: endpoint.url,
secretHeaders: Object.entries(endpoint.secretHeaders).map(([name, value]) => ({ name, value })),
generation: `${owner.ownerId}:${owner.generation}`, refresh: () => ingress(path), close: async () => {} };
};
return { kind: "remote", transport: "computer", providerKey: "boat", environmentId: lease.environmentId, leaseId: lease.id,
remoteCwd: binding.remoteCwd, listenerPort: binding.listenerPort, runner: binding.runner, shellCommand: "bash",
effectiveCapabilities: COMPUTER_CAPABILITIES, reusableLeaseConfigured: true,
runnerLifecyclePolicy: { mode: "warm", idleTimeoutMs },
resourceAuthority: { kind: "computer-owner", ...owner },
fileAuthority: { kind: "remote-persistent", placementId: binding.placementId, root: binding.remoteCwd, agentHome: binding.agentHome },
workspaceRealization: { mode: "in_place", authoritativeRoot: binding.remoteCwd, pathAliases: [], outboundRestorePaths: [] },
launch: binding.launch, inspectProcess: binding.inspectProcess,
retainWarm: (idleTimeoutMs) => computers.retainWarm({ ...scope, owner, idleTimeoutMs }),
retire: async () => {
const result = await computers.retire({ ...scope, owner });
if (result.retired) await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success",
remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) });
},
getRunnerIngressEndpoint: ({ port, path }) => {
if (port !== binding.listenerPort) throw new Error("computer_runner_port_mismatch");
return ingress(path);
},
computerTool: { command: "/opt/ascii/cua-driver/cua-driver", args: ["mcp", "--socket", "/run/ascii-cua/driver.sock"] },
};
}
@@ -75,6 +75,7 @@ export const ENVIRONMENT_DRIVER_CAPABILITY_SUPPORT: Record<
EnvironmentDriver,
EnvironmentDriverCapabilitySupport
> = {
computer: { driver: "computer", supportedCapabilities: ALL_CAPABILITY_SUPPORT },
local: { driver: "local", supportedCapabilities: NO_CAPABILITY_SUPPORT },
ssh: { driver: "ssh", supportedCapabilities: NO_CAPABILITY_SUPPORT },
sandbox: { driver: "sandbox", supportedCapabilities: ALL_CAPABILITY_SUPPORT },
@@ -131,6 +132,7 @@ export interface EnvironmentDriverTraits {
* it.
*/
export const ENVIRONMENT_DRIVER_TRAITS: Record<EnvironmentDriver, EnvironmentDriverTraits> = {
computer: { driver: "computer", realizesWorkspace: true, runsWorkspaceOffHost: true, confinesStagedProjects: false, hasLeaseCapabilityModel: true },
local: {
driver: "local",
realizesWorkspace: true,
@@ -1,3 +1,5 @@
import { resolvePaperclipRunnerIdleTimeoutMs } from "@paperclipai/adapter-utils";
import { computerExecutionTarget } from "./computer-environment-driver.js";
import type { Db } from "@paperclipai/db";
import type { Environment, EnvironmentLease } from "@paperclipai/shared";
import { adapterSupportsRemoteManagedEnvironments } from "@paperclipai/shared";
@@ -246,6 +248,10 @@ export async function resolveEnvironmentExecutionTarget(input: {
// bridge, so the surface stays inert until the host injects a real recorder.
duplexObservabilityRecorder?: DuplexObservabilityRecorder | null;
}): Promise<AdapterExecutionTarget | null> {
if (input.environment.driver === "computer") {
if (!input.lease || input.lease.companyId !== input.companyId) throw new Error("computer_lease_required");
return computerExecutionTarget(input.db, input.lease, resolvePaperclipRunnerIdleTimeoutMs(input.environment.config?.runnerIdleTimeoutMs));
}
if (input.environment.driver === "local") {
return {
kind: "local",
@@ -305,14 +305,14 @@ export function environmentRunOrchestrator(
// when the admitted per-turn policy deliberately disables reusable leases.
const expected = input.reattachRemoteLease;
const lease = await environmentsSvc.getLeaseById(expected.leaseId);
if (!lease || environment.driver !== "sandbox" ||
if (!lease || (environment.driver !== "sandbox" && environment.driver !== "computer") ||
lease.companyId !== input.companyId || lease.environmentId !== environment.id ||
lease.heartbeatRunId !== input.heartbeatRunId || lease.issueId !== input.issueId ||
lease.executionWorkspaceId !== (input.persistedExecutionWorkspace?.id ?? null) ||
lease.metadata?.agentId !== input.agentId || lease.status !== "active" ||
lease.releasedAt !== null || lease.cleanupStatus !== null ||
(lease.expiresAt !== null && new Date(lease.expiresAt).getTime() <= Date.now()) ||
lease.provider !== environment.config.provider || lease.providerLeaseId !== expected.providerLeaseId ||
lease.provider !== (environment.driver === "computer" ? "boat" : environment.config.provider) || lease.providerLeaseId !== expected.providerLeaseId ||
lease.metadata?.remoteCwd !== expected.remoteCwd) {
throw new Error("native_remote_recovery_lease_mismatch");
}
@@ -517,8 +517,9 @@ export function environmentRunOrchestrator(
// Step 3: Persist realization metadata on lease and execution workspace
if (Object.keys(workspaceRealization).length > 0) {
const refreshedLease = environment.driver === "computer" ? await environmentsSvc.getLeaseById(lease.id) : null;
const nextLeaseMetadata = {
...(lease.metadata ?? {}),
...(refreshedLease?.metadata ?? lease.metadata ?? {}),
workspaceRealization,
};
const updatedLease = await environmentsSvc.updateLeaseMetadata(lease.id, nextLeaseMetadata);
@@ -594,6 +595,14 @@ export function environmentRunOrchestrator(
);
}
if (executionTarget?.kind === "remote" && executionTarget.transport === "computer" && persistedExecutionWorkspace) {
persistedExecutionWorkspace = await executionWorkspacesSvc.update(persistedExecutionWorkspace.id, {
cwd: executionTarget.remoteCwd,
providerRef: executionTarget.remoteCwd,
metadata: { ...(persistedExecutionWorkspace.metadata ?? {}),
fileAuthority: { ...executionTarget.fileAuthority, environmentId: environment.id } },
}) ?? persistedExecutionWorkspace;
}
return {
lease,
workspaceRealization,
@@ -1,3 +1,4 @@
import { createComputerEnvironmentDriver } from "./computer-environment-driver.js";
import { beginIdleTrackedWork } from "./task-admission.js";
import { hasStopOnlyCleanup, prepareSandboxStopAndRetain, readStopOnlyCleanup, settleStopOnlyCleanup, stopOnlyCleanupKey } from "./sandbox-stop-and-retain.js";
import { JsonRpcCallError, readEnvironmentAcquisitionDiagnostic, readEnvironmentCreationCleanupError } from "@paperclipai/plugin-sdk";
@@ -3839,6 +3840,7 @@ export function environmentRuntimeService(
const defaultDrivers = [
createLocalEnvironmentDriver(db),
createSshEnvironmentDriver(db),
createComputerEnvironmentDriver(db),
createSandboxEnvironmentDriver(db, {
pluginWorkerManager: options.pluginWorkerManager,
pluginWorkerReadyTimeoutMs: options.pluginWorkerReadyTimeoutMs,
+38 -10
View File
@@ -1,4 +1,5 @@
import { selectHeartbeatRuntime } from "./heartbeat/runtime-selection.js";
import { readNativeComputerWorkspaceReference } from "./native-runtime/native-workspace-sync.js";
import { createHeartbeatRunCompletion } from "./heartbeat/run-completion.js";
export {
MAX_TURN_CONTINUATION_WAKE_REASON,
@@ -3850,7 +3851,7 @@ export function heartbeatService(
{
useProjectWorkspace:
requestedExecutionWorkspaceMode !== "agent_default",
anchorWorkspace: requestedShouldReuseExisting && reusableExistingExecutionWorkspace?.strategyType === "git_worktree"
anchorWorkspace: selectedEnvironmentForConfig?.driver !== "computer" && requestedShouldReuseExisting && reusableExistingExecutionWorkspace?.strategyType === "git_worktree"
? await resolveReusedGitWorkspaceAnchor({
agent,
workspace: reusableExistingExecutionWorkspace,
@@ -3900,7 +3901,7 @@ export function heartbeatService(
repoRef: resolvedWorkspace.repoRef,
additionalWorkspaces: resolvedWorkspace.additionalWorkspaces,
} satisfies ExecutionWorkspaceInput;
await assertGitWorktreeBaseWorkspaceReady({
if (selectedEnvironmentForConfig?.driver !== "computer") await assertGitWorktreeBaseWorkspaceReady({
requestedExecutionWorkspaceMode,
config: hostExecutionWorkspaceConfig,
issue: issueRef,
@@ -4035,7 +4036,18 @@ export function heartbeatService(
executionWorkspace,
reusedExecutionWorkspace,
policy: resolvedWorkspaceReusePolicy,
} = isDotRun ? { executionWorkspace: { ...executionWorkspaceBase, strategy: "project_primary" as const, cwd: resolvedWorkspace.cwd, branchName: null, worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false } as RealizedExecutionWorkspace, reusedExecutionWorkspace: false, policy: workspaceReuseProvisioningPolicy } : await provisionExecutionWorkspaceForFreshnessDecision<RealizedExecutionWorkspace>(
} = (isDotRun || selectedEnvironmentForConfig?.driver === "computer") ? {
executionWorkspace: {
...executionWorkspaceBase,
strategy: selectedEnvironmentForConfig?.driver === "computer" ? latestWorkspaceStrategyType : "project_primary",
cwd: resolvedWorkspace.cwd,
branchName: selectedEnvironmentForConfig?.driver === "computer" && latestWorkspaceStrategyType === "git_worktree" && issueId
? reusableExistingExecutionWorkspace?.branchName ?? `paperclip/task-${issueId}` : null,
worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false,
} as RealizedExecutionWorkspace,
reusedExecutionWorkspace: false,
policy: workspaceReuseProvisioningPolicy,
} : await provisionExecutionWorkspaceForFreshnessDecision<RealizedExecutionWorkspace>(
{
requestedShouldReuseExisting,
existingExecutionWorkspaceId:
@@ -4150,7 +4162,7 @@ export function heartbeatService(
issueRef?.executionWorkspacePreference ?? null;
let issueExecutionWorkspaceModeForRun =
issueExecutionWorkspaceSettings?.mode ?? null;
const warmReusableExecutionWorkspace =
const warmReusableExecutionWorkspace = selectedEnvironmentForConfig?.driver === "computer" ||
selectedEnvironmentForConfig?.driver === "sandbox" &&
selectedEnvironmentConfigForFingerprint.reuseLease === true &&
selectedEnvironmentConfigForFingerprint.runnerLifecycleMode === "warm";
@@ -4468,7 +4480,7 @@ export function heartbeatService(
const remoteRecovery = runOptions.nativeRestartRecovery?.kind === "reattach_remote_runner"
? runOptions.nativeRestartRecovery : null;
const recoveryWorkspace = remoteRecovery
? readNativeWorkspaceSyncReference(parseObject(run.runnerProfileJson).nativeWorkspaceSync) : null;
? (readNativeWorkspaceSyncReference(parseObject(run.runnerProfileJson).nativeWorkspaceSync) ?? readNativeComputerWorkspaceReference(parseObject(run.runnerProfileJson).nativeComputerWorkspace)) : null;
if (remoteRecovery && (!recoveryWorkspace || remoteRecovery.runId !== run.id ||
recoveryWorkspace.providerLeaseId !== remoteRecovery.remote.providerLeaseId ||
recoveryWorkspace.remoteCwd !== remoteRecovery.remote.remoteCwd)) {
@@ -4614,6 +4626,10 @@ export function heartbeatService(
// Preserve the host-owned source before adapter context can share lease
// metadata. A later copy-back failure must not adopt a rebound source.
const workspaceRestoreSource = structuredClone(realizationResult.lease);
if (executionTarget?.kind === "remote" && executionTarget.transport === "computer") {
executionWorkspace.cwd = executionTarget.remoteCwd;
if (executionWorkspace.strategy === "git_worktree") executionWorkspace.worktreePath = executionTarget.remoteCwd;
}
let instructionCopy: Awaited<ReturnType<typeof instructionCopies.prepare>> = null;
let instructionSave: Record<string, unknown> | null = null;
const instructionPreparationKey = createHash("sha256").update(JSON.stringify({
@@ -4927,7 +4943,8 @@ export function heartbeatService(
branchName: executionWorkspace.branchName,
worktreePath: executionWorkspace.worktreePath,
realization: workspaceRealization,
agentHome: await (async () => {
agentHome: executionTarget?.kind === "remote" && executionTarget.transport === "computer"
? executionTarget.fileAuthority.agentHome : await (async () => {
const home = resolveDefaultAgentWorkspaceDir(agent.id);
await fs.mkdir(home, { recursive: true });
return home;
@@ -5373,7 +5390,7 @@ export function heartbeatService(
typeof entry[0] === "string" && typeof entry[1] === "string",
),
);
const runtimeServices = await ensureRuntimeServicesForRun({
const runtimeServices = selectedEnvironmentForConfig?.driver === "computer" ? [] : await ensureRuntimeServicesForRun({
db,
runId: run.id,
agent: {
@@ -5512,9 +5529,9 @@ export function heartbeatService(
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);
const warmFiles = nativeRuntimeResolution.kind === "native" && (nativeRuntimeResolution.profile.backend === "codex_app_server" ||
const warmFiles = nativeRuntimeResolution.kind === "native" && ((executionTarget?.kind === "remote" && executionTarget.transport === "computer") || nativeRuntimeResolution.profile.backend === "codex_app_server" ||
(nativeRuntimeResolution.profile.backend === "acpx_runtime" && parseObject(agent.adapterConfig).acpxAgent === "cursor")) &&
(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox"
(executionTarget?.kind === "remote" && (executionTarget.transport === "sandbox" || executionTarget.transport === "computer")
? executionTarget.runnerLifecyclePolicy?.mode === "warm"
: parseObject(agent.adapterConfig).lifecycleMode === "warm");
if (warmFiles && taskSessionForRun?.lastRunId) {
@@ -5689,6 +5706,17 @@ export function heartbeatService(
}
let adapterFinalizeOutcome: "succeeded" | "failed" | null = null;
const inspectFinalizeWorkspaceBranch = async () => {
// A remote pathname must never be inspected as a controller Git tree.
if (executionTarget?.kind === "remote" && executionTarget.transport === "computer") {
if (executionWorkspace.strategy === "git_worktree" && executionWorkspace.branchName) {
const branch = await executionTarget.runner!.execute({ command: "git", args: ["symbolic-ref", "--quiet", "--short", "HEAD"],
cwd: executionTarget.remoteCwd, timeoutMs: 10_000, bypassSession: true });
if (branch.exitCode !== 0 || branch.timedOut || branch.stdout.trim() !== executionWorkspace.branchName) {
throw new Error("computer_workspace_branch_mismatch");
}
}
return null;
}
const workspaceRecord = persistedExecutionWorkspace?.id
? await executionWorkspacesSvc.getById(
persistedExecutionWorkspace.id,
@@ -6252,7 +6280,7 @@ export function heartbeatService(
onProviderStopped: collectStoppedInstructions,
onDispatch: markDispatchStarted,
signal: executionControl.controller.signal,
...(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox" ? {
...(executionTarget?.kind === "remote" && (executionTarget.transport === "sandbox" || executionTarget.transport === "computer") ? {
stopRemoteStartup: async () => {
// Scope comes from the running host invocation, never agent
// config. Keep adapter ownership until setup has unwound.
+2 -31
View File
@@ -91,7 +91,7 @@ import {
} from "../issue-queued-comment-queue.js";
import {
cancelNativeSession,
closeWarmNativeSessionsForEnvironment,
closeWarmNativeSessionsForRun,
} from "../native-runtime/index.js";
import { runningProcesses } from "../../adapters/index.js";
import { parseObject } from "../../adapters/utils.js";
@@ -292,36 +292,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
if (providerResourceDisposition === "destroy") {
const closeResult = await (
options.closeWarmNativeSessionsForRun ??
(async ({ runId, reason }) => {
const environmentIds = await db
.selectDistinct({ environmentId: environmentLeases.environmentId })
.from(environmentLeases)
.where(
and(
eq(environmentLeases.heartbeatRunId, runId),
eq(environmentLeases.status, "active"),
),
)
.then((rows) =>
rows.flatMap((row) =>
typeof row.environmentId === "string" &&
row.environmentId.length > 0
? [row.environmentId]
: [],
),
);
const aggregate = { closed: 0, busy: 0, failed: 0 };
for (const environmentId of environmentIds) {
const result = await closeWarmNativeSessionsForEnvironment({
environmentId,
reason,
});
aggregate.closed += result.closed;
aggregate.busy += result.busy;
aggregate.failed += result.failed;
}
return aggregate;
})
closeWarmNativeSessionsForRun
)({
runId: input.runId,
reason: "terminal heartbeat run destroyed its environment lease",
+14 -1
View File
@@ -3455,7 +3455,7 @@ export function createHeartbeatWorkspaceResolver(db: Db) {
agent: typeof agents.$inferSelect,
context: Record<string, unknown>,
previousSessionParams: Record<string, unknown> | null,
opts?: { useProjectWorkspace?: boolean | null },
opts?: { useProjectWorkspace?: boolean | null; executionEnvironmentDriver?: string | null },
): Promise<ResolvedAnchorWorkspaceForRun> {
const issueId =
readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId);
@@ -3507,6 +3507,19 @@ export function createHeartbeatWorkspaceResolver(db: Db) {
preferredProjectWorkspaceId,
);
if (opts?.executionEnvironmentDriver === "computer") {
const selected = projectWorkspaceRows[0];
// This is a stable intent path. Computer realization supplies its actual
// remote path without cloning, statting, or provisioning controller files.
return {
cwd: `/paperclip/remote/${agent.companyId}/${selected?.projectId ?? agent.id}`,
source: selected ? "project_primary" : "agent_home",
projectId: workspaceProjectId, workspaceId: selected?.id ?? null,
repoUrl: selected?.repoUrl ?? null, repoRef: selected?.repoRef ?? null,
workspaceHints: [], warnings: [], baseCwdFallback: false, materializationFailures: [],
};
}
const workspaceHints = projectWorkspaceRows.map((workspace) => ({
workspaceId: workspace.id,
cwd: readNonEmptyString(workspace.cwd),
@@ -1,3 +1,4 @@
import { readNativeComputerWorkspaceReference } from "./native-workspace-sync.js";
import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
import { recordNativeLocalProcessStop } from "../native-local-process-stop.js";
import { randomUUID } from "node:crypto";
@@ -49,7 +50,8 @@ export type NativeRestartRecoveryClaim =
providerAttempt: number;
restartKind: NativeRestartKind;
recoveryRequestId: string | null;
remote: { providerLeaseId: string; remoteCwd: string };
remote: { providerLeaseId: string; remoteCwd: string;
computerOwner?: { computerId: string; ownerId: string; generation: number; listenerPort: number } };
}
| {
kind: "resume_dead_runner";
@@ -615,8 +617,8 @@ export async function claimNativeRestartRecoveries(input: {
const profile = row.run.runnerProfileJson ?? {};
const remoteWorkspace = readNativeWorkspaceSyncReference(
profile.nativeWorkspaceSync,
);
const remoteSandbox = profile.nativeWorkspaceSync != null;
) ?? readNativeComputerWorkspaceReference(profile.nativeComputerWorkspace);
const remoteSandbox = profile.nativeWorkspaceSync != null || profile.nativeComputerWorkspace != null;
const runnerPidAlive =
!remoteSandbox && processIsAlive(row.run.processPid);
const runnerGroupAlive =
@@ -1001,6 +1003,7 @@ export async function claimNativeRestartRecoveries(input: {
remote: {
providerLeaseId: remoteWorkspace!.providerLeaseId,
remoteCwd: remoteWorkspace!.remoteCwd,
...("computerOwner" in remoteWorkspace! ? { computerOwner: remoteWorkspace.computerOwner } : {}),
},
} satisfies NativeRestartRecoveryClaim;
}
@@ -309,6 +309,7 @@ import {
buildNativeHarnessBackupManifest,
cancelNativeSession,
closeWarmNativeSessionsForEnvironment,
closeWarmNativeSessionsForRun,
claimWarmNativeInstructionCopy,
nativeSessionWorkspaceScope,
reserveWarmNativeInstructionDirectory,
@@ -475,6 +476,28 @@ describe("remote controller restart adoption", () => {
};
return { execute, target };
}
it("requires the exact durable computer owner generation and port on restart", async () => {
const { execute, target: sandbox } = fixture();
const { sandboxLeaseAcquisition, ...common } = sandbox;
const inspectProcess = vi.fn(async () => ({ running: true, claim: { nonce: "process-claim" } }));
const target = { ...common, transport: "computer", listenerPort: 45101,
resourceAuthority: { kind: "computer-owner", computerId: "computer", ownerId: "owner", generation: 7 }, inspectProcess };
const computerClaim = { ...claim, remote: { ...claim.remote, providerLeaseId: "owner",
computerOwner: { computerId: "computer", ownerId: "owner", generation: 7, listenerPort: 45101 } } };
const adopted = await verifyRemoteRunnerReattachment({ claim: computerClaim, target: target as never,
identity, runId: "run", normalizedSessionId: "session" });
expect(adopted.pid).toBe(123);
expect(inspectProcess).toHaveBeenCalledOnce();
const readCount = execute.mock.calls.length;
await expect(verifyRemoteRunnerReattachment({ claim: { ...computerClaim, remote: {
...computerClaim.remote, computerOwner: { ...computerClaim.remote.computerOwner, generation: 6 } } }, target: target as never,
identity, runId: "run", normalizedSessionId: "session" })).rejects.toThrow("computer_process_recovery_mismatch");
expect(execute).toHaveBeenCalledTimes(readCount);
await expect(verifyRemoteRunnerReattachment({ claim: { ...computerClaim, remote: {
...computerClaim.remote, computerOwner: { ...computerClaim.remote.computerOwner, listenerPort: 45102 } } }, target: target as never,
identity, runId: "run", normalizedSessionId: "session" })).rejects.toThrow("computer_process_recovery_mismatch");
});
it("adopts the exact live sandbox process without spawning or copying state", async () => {
const { execute, target } = fixture();
const adopted = await verifyRemoteRunnerReattachment({
@@ -7353,6 +7376,40 @@ describe("native warm session supervision", () => {
expect(canonicalNote).toBe("before shutdown\nafter shutdown");
});
it("retains durable computer owners before ack and closes only the requested run", async () => {
const owners: Array<{ runId: string; close: ReturnType<typeof vi.fn>; retainWarm: ReturnType<typeof vi.fn>; retire: ReturnType<typeof vi.fn> }> = [];
for (const suffix of ["a", "b"]) {
const name = `computer-owner-${suffix}`;
const current = { ...execution, binding: { ...execution.binding, runId: name, agentId: name, executionWorkspaceId: name },
session: { ...execution.session, normalizedSessionId: name, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } } as NativeExecutionInputV1;
const close = vi.fn(async () => undefined), retainWarm = vi.fn(async () => undefined), retire = vi.fn(async () => undefined);
const target = { kind: "remote" as const, transport: "computer" as const, environmentId: "shared-computer",
remoteCwd: `/home/user/${name}`, listenerPort: suffix === "a" ? 45001 : 45002,
resourceAuthority: { kind: "computer-owner", computerId: "computer", ownerId: name, generation: 1 },
fileAuthority: { kind: "remote-persistent", placementId: name, root: `/home/user/${name}`, agentHome: `/home/user/${name}/home` },
retainWarm, retire, inspectProcess: async () => ({ running: true, claim: {} }), launch: async () => ({}),
computerTool: { command: "cua-driver", args: ["mcp"] } };
state.execute.mockReset().mockImplementationOnce(async options => {
await options.onSession?.({ close });
return { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" }, turnId: name,
normalizedSessionId: name, providerSessionId: name, driverKind: "test", driverVersion: "1",
nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
});
await executePaperclipNativeSession({ db: leaseDb(current), execution: current, runnerInstanceId: name, runnerExecutionTarget: target as never });
expect(retainWarm).toHaveBeenCalledWith(60_000);
expect(retire).not.toHaveBeenCalled();
owners.push({ runId: name, close, retainWarm, retire });
}
const result = await closeWarmNativeSessionsForRun({ runId: owners[0]!.runId, reason: "cancel exact task" });
expect(result.closed).toBe(1);
expect(owners[0]!.close).toHaveBeenCalledOnce();
expect(owners[0]!.retire).toHaveBeenCalledOnce();
expect(owners[0]!.close.mock.invocationCallOrder[0]).toBeLessThan(owners[0]!.retire.mock.invocationCallOrder[0]!);
expect(owners[1]!.close).not.toHaveBeenCalled();
expect(owners[1]!.retire).not.toHaveBeenCalled();
await closeWarmNativeSessionsForRun({ runId: owners[1]!.runId, reason: "fixture cleanup" });
});
describe("managed directory warm checkpoints", () => {
const result = { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" },
turnId: "turn", normalizedSessionId: "managed", providerSessionId: "provider", driverKind: "test", driverVersion: "1",
@@ -1,3 +1,4 @@
import { adapterExecutionTargetIsCommandBacked, type AdapterComputerExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
import { configuredEnvironment } from "../../vendor/paperclip-runner/index.js";
import { agents } from "@paperclipai/db";
import { isCloudManagedInstance } from "../cloud-instance.js";
@@ -441,6 +442,8 @@ export type NativeInstructionWorkingCopy = {
};
type WarmNativeSession = {
computerTarget?: AdapterComputerExecutionTarget;
runId: string;
instructionWorkingCopy?: NativeInstructionWorkingCopy;
// Keep the admitted physical root independent of the per-run collection
// capability, which instruction preparation adopts before final admission.
@@ -477,6 +480,7 @@ async function closeWarmNativeSession(entry: WarmNativeSession, reason: string,
catch (receiptError) { throw new AggregateError([error, receiptError], "Warm instruction retirement and receipt both failed"); }
throw error;
}
await entry.computerTarget?.retire();
await entry.instructionWorkingCopy?.collectStopped();
if (entry.instructionCopy?.runId !== preserveInstructionsForRunId) await entry.instructionCopy?.collectStopped();
}
@@ -550,7 +554,7 @@ export async function claimWarmNativeInstructionCopy(input: {
function instructionTargetIdentity(target?: AdapterExecutionTarget | null): string {
return JSON.stringify(target?.kind === "remote" ? {
environmentId: target.environmentId, cwd: target.remoteCwd,
providerLeaseId: target.transport === "sandbox" ? target.sandboxLeaseAcquisition?.providerLeaseId : target.spec,
providerLeaseId: target.transport === "sandbox" ? target.sandboxLeaseAcquisition?.providerLeaseId : target.transport === "computer" ? target.resourceAuthority.computerId : target.spec,
} : { kind: "local", environmentId: target?.environmentId });
}
@@ -611,6 +615,11 @@ export async function closeWarmNativeSessionsForEnvironment(input: {
return closeIdleWarmNativeSessions(input);
}
/** Cancel one attempt without closing other agents on its computer. */
export async function closeWarmNativeSessionsForRun(input: { runId: string; reason: string }) {
return closeIdleWarmNativeSessions(input);
}
/** Suspend idle owners and persist their remote backup before a controller
* exits. Active turns keep their separate authenticated restart handoff. */
export async function closeIdleWarmNativeSessionsForRestart(): Promise<{
@@ -624,6 +633,7 @@ export async function closeIdleWarmNativeSessionsForRestart(): Promise<{
async function closeIdleWarmNativeSessions(input: {
environmentId?: string;
runId?: string;
reason: string;
closeBusyOnRelease?: boolean;
}): Promise<{ closed: number; busy: number; failed: number }> {
@@ -631,6 +641,7 @@ async function closeIdleWarmNativeSessions(input: {
let busy = 0;
let failed = 0;
for (const [sessionId, entry] of [...warmNativeSessions]) {
if (input.runId !== undefined && entry.runId !== input.runId) continue;
if (input.environmentId !== undefined && entry.environmentId !== input.environmentId) {
continue;
}
@@ -6165,6 +6176,8 @@ async function releaseWarmNativeSession(
} finally { entry.busy = false; }
return;
}
// Durable shared-computer ownership must be retained before completion is acknowledged.
await entry.computerTarget?.retainWarm(idleTimeoutMs);
entry.idleTimer = setTimeout(() => {
const current = warmNativeSessions.get(sessionId);
if (current !== entry || current.busy) return;
@@ -8495,6 +8508,8 @@ async function executePaperclipNativeSessionWithinScope(
entry.instructionPreparationRunId = undefined;
entry.instructionWorkingCopy = input.instructionWorkingCopy?.checkpointWarm ? undefined : input.instructionWorkingCopy;
entry.busy = true;
entry.runId = input.execution.binding.runId;
entry.computerTarget = input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "computer" ? input.runnerExecutionTarget : undefined;
entry.ownerToken = warmSessionOwnerToken;
entry.environmentId =
input.runnerExecutionTarget?.environmentId ?? null;
@@ -8861,6 +8876,8 @@ async function executePaperclipNativeSessionWithinScope(
? input.execution.binding.runId
: undefined,
session,
runId: input.execution.binding.runId,
computerTarget: input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "computer" ? input.runnerExecutionTarget : undefined,
ownerToken: warmSessionOwnerToken,
configDigest: warmConfigDigest,
configuredEnvironmentDigest,
@@ -8875,6 +8892,10 @@ async function executePaperclipNativeSessionWithinScope(
lastActivityAt: new Date().toISOString(),
});
const owner = warmNativeSessions.get(warmSessionId);
if (owner) {
owner.runId = input.execution.binding.runId;
owner.computerTarget = input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "computer" ? input.runnerExecutionTarget : undefined;
}
if (owner && input.instructionWorkingCopy?.runId && input.instructionWorkingCopy.checkpointWarm) owner.instructionCopy = {
runId: input.instructionWorkingCopy.runId, root: input.instructionWorkingCopy.root, targetIdentity: instructionTargetIdentity(input.runnerExecutionTarget),
collectStopped: input.instructionWorkingCopy.collectStopped,
@@ -10681,19 +10702,23 @@ export async function verifyRemoteRunnerReattachment(input: {
}) {
const { claim, target, identity } = input;
if (
target?.kind !== "remote" ||
target.transport !== "sandbox" ||
!adapterExecutionTargetIsCommandBacked(target) ||
!target.runner ||
claim.runId !== input.runId ||
identity.runId !== input.runId ||
identity.normalizedSessionId !== input.normalizedSessionId ||
claim.remote.providerLeaseId !==
target.sandboxLeaseAcquisition?.providerLeaseId ||
target.sandboxLeaseAcquisition.outcome === "replacement" ||
claim.remote.providerLeaseId !== (target.transport === "computer" ? target.resourceAuthority.ownerId : target.sandboxLeaseAcquisition?.providerLeaseId) ||
(target.transport === "sandbox" && target.sandboxLeaseAcquisition?.outcome === "replacement") ||
claim.remote.remoteCwd !== target.remoteCwd
) {
throw new Error("native_remote_recovery_lease_mismatch");
}
if (target.transport === "computer") {
const owner = claim.remote.computerOwner;
if (!owner || owner.computerId !== target.resourceAuthority.computerId || owner.ownerId !== target.resourceAuthority.ownerId ||
owner.generation !== target.resourceAuthority.generation || owner.listenerPort !== target.listenerPort ||
!(await target.inspectProcess()).running) throw new Error("computer_process_recovery_mismatch");
}
const runner = target.runner;
const stateDirectory = posix.join(
target.remoteCwd,
@@ -10943,7 +10968,13 @@ export function createRemoteRunnerProcessLauncher(input: {
// into its own session instead; its own bounded diagnostics directory and
// durable PRP state remain the authorities, and the controller monitors
// the exact persisted process identity below.
const launchResult = await runner.execute({
const launchResult = input.target.transport === "computer"
? await input.target.launch({ command: "sh", args: ["-c", 'mkdir -p -- "$(dirname -- "$1")"; ' + REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT,
"paperclip-runner-child", input.processIdentityPath, identityNonce, input.runnerInstanceId,
input.diagnosticsDirectory, input.remoteBinary, ...remoteArgs],
cwd: input.target.remoteCwd, env: processEnvironment(spec.environment) })
.then(() => ({ exitCode: 0, timedOut: false }))
: await runner.execute({
command: "sh",
args: [
"-c",
@@ -11104,7 +11135,7 @@ export function resolveRemoteRunnerTransportMode(input: {
throw new Error("runner_transport_ineligible: remote target is required");
}
const requiredMode =
input.target.transport === "sandbox" &&
adapterExecutionTargetIsCommandBacked(input.target) &&
input.target.effectiveCapabilities?.runnerWebSocketIngress === true
? "listen_ws"
: "dial_wss";
@@ -13056,7 +13087,7 @@ async function createRunnerdBackendWithinSessionClaim(
HOME: posix.join(remoteRunnerFilesystemRoot!, "codex-home"),
CODEX_HOME: posix.join(remoteRunnerFilesystemRoot!, "codex-home"),
PAPERCLIP_WORKSPACE_CWD: remoteTarget!.remoteCwd,
...(remoteTarget!.transport === "sandbox"
...(adapterExecutionTargetIsCommandBacked(remoteTarget)
? { PAPERCLIP_RUNNER_EXTERNAL_SANDBOX: "1" }
: {}),
}
@@ -13305,7 +13336,8 @@ async function createRunnerdBackendWithinSessionClaim(
baseInstructions: recoveryContext?.baseInstructions,
runnerFilesystemRoot: remoteRunnerFilesystemRoot ?? undefined,
resumeWorkingDirectory: runnerExecution.workspace.cwd,
externallySandboxed: remoteTarget?.transport === "sandbox",
externallySandboxed: adapterExecutionTargetIsCommandBacked(remoteTarget),
computerTool: remoteTarget?.transport === "computer" ? remoteTarget.computerTool : undefined,
opencodeRuntimeDirectory: remoteRunnerFilesystemRoot
? posix.join(remoteRunnerFilesystemRoot, "opencode")
: undefined,
@@ -2,7 +2,7 @@ import fs from "node:fs/promises";
import { createReadStream } from "node:fs";
import path from "node:path";
import { createHash, randomUUID } from "node:crypto";
import { eq } from "drizzle-orm";
import { eq, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { environmentLeases, heartbeatRuns } from "@paperclipai/db";
import type { EnvironmentLease } from "@paperclipai/shared";
@@ -825,6 +825,21 @@ async function finalizePreparedRuntime(input: {
return reference;
}
/** Persistent remote bytes have no controller export/restore descriptor. */
export interface NativeComputerWorkspaceReference {
leaseId: string; providerLeaseId: string; remoteCwd: string;
computerOwner: { computerId: string; ownerId: string; generation: number; listenerPort: number };
}
export function readNativeComputerWorkspaceReference(value: unknown): NativeComputerWorkspaceReference | null {
const parsed = parseObject(value);
if (parsed.kind !== "remote-persistent" || typeof parsed.leaseId !== "string" ||
typeof parsed.providerLeaseId !== "string" || typeof parsed.remoteCwd !== "string" ||
typeof parsed.computerId !== "string" || !Number.isSafeInteger(parsed.ownerGeneration) || Number(parsed.ownerGeneration) < 1 ||
!Number.isInteger(parsed.listenerPort) || Number(parsed.listenerPort) < 1 || Number(parsed.listenerPort) > 65535) return null;
return { leaseId: parsed.leaseId, providerLeaseId: parsed.providerLeaseId, remoteCwd: parsed.remoteCwd,
computerOwner: { computerId: parsed.computerId, ownerId: parsed.providerLeaseId, generation: Number(parsed.ownerGeneration), listenerPort: Number(parsed.listenerPort) } };
}
export async function prepareNativeWorkspaceSync(input: {
db: Db;
runId: string;
@@ -837,6 +852,20 @@ export async function prepareNativeWorkspaceSync(input: {
sameRunRecovery?: boolean;
resourceDisposition?: NativeWorkspaceResourceDisposition;
}): Promise<PreparedNativeWorkspaceSync | null> {
if (input.target?.kind === "remote" && input.target.transport === "computer") {
const target = input.target;
if (input.restartRecovery?.kind === "reattach_remote_runner" &&
(input.restartRecovery.remote.providerLeaseId !== target.resourceAuthority.ownerId ||
input.restartRecovery.remote.remoteCwd !== target.remoteCwd)) throw new Error("native_remote_recovery_lease_mismatch");
const descriptor = { kind: "remote-persistent", leaseId: input.lease.id,
providerLeaseId: target.resourceAuthority.ownerId, computerId: target.resourceAuthority.computerId,
ownerGeneration: target.resourceAuthority.generation, listenerPort: target.listenerPort, remoteCwd: target.remoteCwd,
placementId: target.fileAuthority.placementId };
await input.db.update(heartbeatRuns).set({
runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ nativeComputerWorkspace: descriptor })}::jsonb`,
}).where(eq(heartbeatRuns.id, input.runId));
return null;
}
if (input.target?.kind !== "remote" || input.target.transport !== "sandbox") {
return null;
}