Files
PaperClipAI/server/src/services/native-runtime/native-workspace-sync.ts
T
DottaandPaperclip 71c04c0bfb fix: stop idle environment cleanup polling (#17145)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Agent environments can leave pending teardown or unverified provider
resources.
> - Recovery must preserve those resources until ownership and workspace
safety allow cleanup.
> - Empty maintenance scans keep an idle instance and its database
active.
> - The existing rows, transaction signals, and scheduler can represent
the actual recovery obligation.
> - This pull request schedules pending cleanup and historical
verification only while work remains.
> - An empty or fully verified instance stops these scans without losing
cleanup recovery.

## Linked Issues or Issue Description

**What happened?**

The server scanned pending environment cleanup on its maintenance tick.
The run reaper also scanned it even when it recovered no runs or active
leases. Historical resource verification scanned every five minutes,
including when all history already had valid cleanup proof.

**What did you expect?**

Writes should wake cleanup. Pending rows, deferred orphan buffers, and
unverified history should retain recovery deadlines. When none remain,
these workers should make no periodic database or provider calls.

**Steps to reproduce**

Start a configured instance with no pending leases and fully verified
history. Leave it idle. Observe repeated pending-cleanup and
historical-verification reads from the maintenance paths.

Related: #17106 supplies the preceding login-cleanup migration. #13523
improves lease-reaper query indexes. #16809 addresses reusable-sandbox
recovery authority. This PR keeps those authority checks in their
existing services.

## What Changed

- Register two workers with the existing delivery coordinator and shared
scheduler. Remove their unconditional maintenance-tick calls. Run the
reaper's inline pending cleanup only after it recovers actual work.
- Add commit-aware signals to the audited lease and custom-image
writers. Existing transactions notify directly; other writers use one
small transaction helper. No tables or migrations are added.
- Derive pending cleanup deadlines from the existing row backoff and
retry or claim metadata. Retain a historical retry only while an
unverified candidate exists.
- Register unwritten orphan buffers across runtimes that share the
database handle. Buffer registration wakes recovery without a database
connection. Load the existing spool at startup and retry failed or
capacity-limited loads while needed.
- Isolate spool failures per runtime. Attempt all buffers and persisted
pending rows, then retain a retry for any failed spool.
- Document producer coverage, idle behavior, and remaining blockers.
Update the sleep inventory.
- Repair the inherited CSS-recovery E2E test: the current build no
longer emits task-page CSS. Exercise the actual lazy environment
stylesheet, preserving real network failure, recovery, and bounded
reload assertions.

## Verification

October 24 integration update: merged current master and resolved the
runtime/spool conflicts. Preserve serialized ingestion of records
appended after startup. Plugin replacement failures now register
deferred recovery with an idle hold; concurrent callers share one
persistence attempt. The producer audit includes this new writer.

- Focused cleanup selection: 270 tests now pass across five suites. The
initial run lacked a built plugin SDK and executed no tests; building
the SDK fixed setup. The new concurrent-flush assertion then reproduced
duplicate rows; a single-flight guard fixed it and all 18 plugin
integration tests passed on rerun.
- Server typecheck passed. After review fixes, all 211 tests across the
cleanup worker, real orphan spool, environment runtime, and plugin
integration suites passed. These overlap the earlier focused selection.
The rate-limit test initially expected a write after twenty idle minutes
to wait; corrected it to assert that subsequent writes cannot accelerate
the five-minute cadence.
- Review fixes preserve retries for unreadable spool
directories/records, keep historical provider checks off the startup
wait while joining them at shutdown, and restore their independent
five-minute minimum spacing. The complete [CI
matrix](https://github.com/paperclipai/paperclip/actions/runs/40108905033)
passed on final commit `79ae1589c3c7b924db4bedb337d0c7ed3d95020b`,
including all E2E shards, server/workspace tests, runner checks,
typecheck, build/public-install verification, and security checks.

- CI found a stale E2E asset matcher: `issue-detail-*.css` is no longer
emitted. Both corrected browser tests passed against a fresh production
UI build and disposable local instance. Fresh Greptile review is 5/5 on
that same final commit, with all review threads resolved.

Previous implementation verification:

- `pnpm -r typecheck` and `pnpm build` passed.
- New PostgreSQL worker suite: 12 tests passed. Covers quiet configured
state, verified history, writes, outer commit and rollback, retry
deadlines, restart, drain, deferred orphan buffers, spool failure,
pagination, provider uncertainty, shutdown, and the idle run-reaper
path.
- Worker and environment runtime suites after review fix: 183 tests
passed. A new assertion reproduced buffer starvation before the fix.
Updated transaction fault injection and isolated simulated-instance
database handles. Server typecheck also passed after the fix.
- Run-recovery suites: 58 tests passed. Pending-cleanup suite: 38 tests
passed after updating the logging fixture to seed real orphan recovery;
an idle reaper no longer invokes cleanup.
- Stop and workspace-recovery integration selection: 268 tests passed;
two live-provider tests skipped by their environment gate. This overlaps
the worker suite and is not an additional unique-test count.
- Started the full local `pnpm test:run`, then stopped that older
duplicate run after CI found the fixture mismatch. The corrected 38-test
suite passed locally. The full sharded [CI
matrix](https://github.com/paperclipai/paperclip/actions/runs/39678174479)
passed on final commit `fc0ff3c4554ca4190a8b0b81ec06ef7065b8ad2c`,
including build, typecheck, runner, server, workspace, and E2E checks.
This is not a claim that the local full run completed.

## Risks

- Missing a future producer signal could delay discovery until restart.
The producer audit is documented in `doc/idle-environment-cleanup.md`.
Independent SQL writers and another server process are outside the
existing process-local notification contract.
- Historical candidates blocked by ownership or provider uncertainty
keep a five-minute retry until they can be verified. Provider
destruction, claim fencing, Stop retention, and workspace safety remain
controlled by their existing services.
- This does not remove orphan-active discovery or legacy shutdown
settlement from run recovery. Active custom-image expiry is also still
periodic. External wake coverage does not change.
- Local tests prove these workers become quiet and recover. They do not
establish deployed instance suspension or fleet cost savings.

## Model Used

OpenAI Codex (GPT-6), with code execution and reasoning. The exact
serving model ID and context size were not exposed.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-24 09:04:49 -07:00

1256 lines
49 KiB
TypeScript

import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js";
import { measureRunPhase } from "../execution-preparation-timing.js";
import { emitStartupReuse, type StartupReuseReason } from "@paperclipai/adapter-utils/startup-reuse";
import type { AdapterRuntimeEvent } from "@paperclipai/adapter-utils";
import type { WorkspaceCheckpointMetrics } from "@paperclipai/adapter-utils/workspace-checkpoint";
import type { RuntimeSpanRunner } from "@paperclipai/adapter-utils/acpx-engine/startup-timing";
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 type { Db } from "@paperclipai/db";
import { environmentLeases, heartbeatRuns } from "@paperclipai/db";
import type { EnvironmentLease } from "@paperclipai/shared";
import {
prepareAdapterExecutionTargetRuntime,
type AdapterExecutionTarget,
type PreparedAdapterExecutionTargetRuntime,
} from "@paperclipai/adapter-utils/execution-target";
import { disposeGitWorkspaceSnapshot, type GitWorkspaceSnapshot } from "@paperclipai/adapter-utils/git-workspace-sync";
import {
directorySnapshotSha256,
disposeDirectorySnapshot,
openDirectorySnapshot,
parseDirectorySnapshot,
serializeDirectorySnapshot,
type DirectorySnapshot,
type SerializedDirectorySnapshot,
} from "@paperclipai/adapter-utils/workspace-restore-merge";
import type {
WorkspaceDurableSeedPaths,
WorkspaceInboundMode,
} from "@paperclipai/adapter-utils/sandbox-managed-runtime";
import { assertWorkspaceManifestDiskSpace, isPathManifest, manifestFileSha256, readManifestRecords, type PathManifest, type WorkspacePaths } from "@paperclipai/adapter-utils/workspace-manifest";
import { resolvePaperclipInstanceRoot } from "../../home-paths.js";
import { parseObject } from "../../adapters/utils.js";
import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js";
const LEGACY_DESCRIPTOR_SCHEMA = "paperclip.native-workspace-sync/v1";
const PREVIOUS_DESCRIPTOR_SCHEMA = "paperclip.native-workspace-sync/v2";
const DESCRIPTOR_SCHEMA = "paperclip.native-workspace-sync/v3";
type DescriptorSchema = typeof DESCRIPTOR_SCHEMA | typeof PREVIOUS_DESCRIPTOR_SCHEMA | typeof LEGACY_DESCRIPTOR_SCHEMA;
const STAMP_SCHEMA = "paperclip.native-workspace-stamp/v1";
const STATE_ROOT_NAME = "native-workspace-sync";
const DESCRIPTOR_NAME = "descriptor";
const WORKSPACE_SEED_NAME = "workspace-seed.tar";
const GIT_SEED_NAME = "git-seed.tar";
const REMOTE_STAMP_NAME = "workspace-sync-v1.json";
const MAX_DESCRIPTOR_BYTES = 64 * 1024 * 1024;
const SAFE_SEGMENT_RE = /^[A-Za-z0-9_-]+$/;
type NativeWorkspaceSyncState = "prepared" | "finalized";
export type NativeWorkspaceResourceDisposition =
"keep_running" | "stop_and_retain" | "destroy";
interface NativeWorkspaceSyncDescriptor {
schema: DescriptorSchema;
binding: {
runId: string;
companyId: string;
workspaceId: string;
leaseId: string;
providerLeaseId: string;
localCwd: string;
remoteCwd: string;
};
state: NativeWorkspaceSyncState;
baselineSha256: string;
baseline: SerializedDirectorySnapshot;
gitSnapshot: GitWorkspaceSnapshot | null;
repositories?: NonNullable<GitWorkspaceSnapshot["repositories"]>;
seed: {
workspaceArchiveSha256: string;
gitArchiveSha256: string | null;
} | null;
createdAt: string;
finalizedAt: string | null;
finalHostSha256: string | null;
resourceDisposition: NativeWorkspaceResourceDisposition | null;
}
export interface NativeWorkspaceSyncReference {
schema: DescriptorSchema;
state: NativeWorkspaceSyncState;
descriptorSha256: string;
baselineSha256: string;
finalHostSha256: string | null;
workspaceId: string;
leaseId: string;
providerLeaseId: string;
remoteCwd: string;
resourceDisposition: NativeWorkspaceResourceDisposition | null;
}
export interface PreparedNativeWorkspaceSync {
mode: WorkspaceInboundMode;
reference: NativeWorkspaceSyncReference;
restoreWorkspace(assertOwnership?: () => Promise<void>): Promise<void>;
cleanup(): Promise<void>;
}
export type NativeWorkspaceInboundEvidence =
| {
kind: "existing_run";
restartRecovery: boolean;
sameRunRecovery?: boolean;
sameProviderLease: boolean;
}
| {
kind: "new_run";
acquisition: "created" | "resumed" | "replacement" | null;
hasPriorStamp: boolean;
};
export function classifyNativeWorkspaceInbound(
evidence: NativeWorkspaceInboundEvidence,
): WorkspaceInboundMode {
if (evidence.kind === "existing_run") {
if (!evidence.restartRecovery && !evidence.sameRunRecovery) {
throw new Error("native_workspace_sync_unexpected_existing_descriptor");
}
return evidence.sameProviderLease ? "adopt_remote" : "durable_seed";
}
return evidence.acquisition === "resumed" && evidence.hasPriorStamp
? "adopt_remote"
: "host_current";
}
function canonicalJson(value: unknown): string {
if (Array.isArray(value)) {
return `[${value.map((entry) => canonicalJson(entry)).join(",")}]`;
}
if (value && typeof value === "object") {
const record = value as Record<string, unknown>;
return `{${Object.keys(record)
.sort()
.map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`)
.join(",")}}`;
}
return JSON.stringify(value);
}
function sha256(value: unknown): string {
return createHash("sha256").update(canonicalJson(value)).digest("hex");
}
function requireSafeSegment(value: string, label: string): string {
if (!SAFE_SEGMENT_RE.test(value)) {
throw new Error(`native_workspace_sync_invalid_${label}`);
}
return value;
}
export function nativeWorkspaceSyncStateRoot(): string {
return path.join(resolvePaperclipInstanceRoot(), STATE_ROOT_NAME);
}
function descriptorDirectory(runId: string): string {
return path.join(nativeWorkspaceSyncStateRoot(), requireSafeSegment(runId, "run_id"));
}
function descriptorPath(runId: string, descriptorSha256: string): string {
if (!/^[0-9a-f]{64}$/.test(descriptorSha256)) {
throw new Error("native_workspace_sync_descriptor_digest_invalid");
}
return path.join(
descriptorDirectory(runId),
`${DESCRIPTOR_NAME}.${descriptorSha256}.json`,
);
}
function legacyDescriptorPath(runId: string): string {
return path.join(descriptorDirectory(runId), `${DESCRIPTOR_NAME}.json`);
}
function durableSeedPaths(runId: string): {
workspaceArchivePath: string;
gitArchivePath: string;
} {
const directory = descriptorDirectory(runId);
return {
workspaceArchivePath: path.join(directory, WORKSPACE_SEED_NAME),
gitArchivePath: path.join(directory, GIT_SEED_NAME),
};
}
async function sha256File(filePath: string): Promise<string> {
return measureRunPhase("workspace_sync.archive_hash", () => new Promise((resolveDigest, rejectDigest) => {
const digest = createHash("sha256");
const stream = createReadStream(filePath);
stream.on("data", (chunk) => digest.update(chunk));
stream.on("error", rejectDigest);
stream.on("end", () => resolveDigest(digest.digest("hex")));
}));
}
async function verifiedDurableSeed(input: {
runId: string;
seed: NonNullable<NativeWorkspaceSyncDescriptor["seed"]>;
gitSnapshot: GitWorkspaceSnapshot | null;
repositories?: NonNullable<GitWorkspaceSnapshot["repositories"]>;
}): Promise<WorkspaceDurableSeedPaths> {
try {
const paths = durableSeedPaths(input.runId);
const workspaceStat = await fs.lstat(paths.workspaceArchivePath);
if (workspaceStat.isSymbolicLink() || !workspaceStat.isFile()) {
throw new Error("workspace_sync_out_unrecoverable");
}
if (
(await sha256File(paths.workspaceArchivePath)) !==
input.seed.workspaceArchiveSha256
) {
throw new Error("workspace_sync_out_unrecoverable");
}
if (input.gitSnapshot || input.repositories?.length) {
const gitStat = await fs.lstat(paths.gitArchivePath);
if (
gitStat.isSymbolicLink() ||
!gitStat.isFile() ||
!input.seed.gitArchiveSha256 ||
(await sha256File(paths.gitArchivePath)) !== input.seed.gitArchiveSha256
) {
throw new Error("workspace_sync_out_unrecoverable");
}
} else if (input.seed.gitArchiveSha256 !== null) {
throw new Error("workspace_sync_out_unrecoverable");
}
return {
workspaceArchivePath: paths.workspaceArchivePath,
workspaceArchiveSha256: input.seed.workspaceArchiveSha256,
gitArchivePath: input.gitSnapshot || input.repositories?.length ? paths.gitArchivePath : null,
gitArchiveSha256: input.seed.gitArchiveSha256,
};
} catch (error) {
if (
error instanceof Error &&
error.message === "workspace_sync_out_unrecoverable"
) {
throw error;
}
throw new Error("workspace_sync_out_unrecoverable", { cause: error });
}
}
async function ensurePrivateDirectory(dir: string): Promise<void> {
await fs.mkdir(dir, { recursive: true, mode: 0o700 });
const stat = await fs.lstat(dir);
if (stat.isSymbolicLink() || !stat.isDirectory()) {
throw new Error("native_workspace_sync_state_root_invalid");
}
}
const runDirectoryListeners = new Set<() => void>();
export function onNativeWorkspaceSyncRunDirectory(listener: () => void): () => void {
runDirectoryListeners.add(listener);
return () => {
runDirectoryListeners.delete(listener);
};
}
async function ensureRunDirectory(runId: string): Promise<string> {
const dir = descriptorDirectory(runId);
await ensurePrivateDirectory(dir);
for (const listener of runDirectoryListeners) listener();
return dir;
}
async function writeDescriptor(
descriptor: NativeWorkspaceSyncDescriptor,
): Promise<NativeWorkspaceSyncReference> {
const dir = await ensureRunDirectory(descriptor.binding.runId);
const body = `${canonicalJson(descriptor)}\n`;
const descriptorSha256 = createHash("sha256").update(body).digest("hex");
const finalPath = descriptorPath(descriptor.binding.runId, descriptorSha256);
const existingBody = await fs.readFile(finalPath, "utf8").catch(() => null);
if (existingBody !== null && existingBody !== body) {
throw new Error("native_workspace_sync_descriptor_digest_collision");
}
const tempPath = path.join(dir, `${DESCRIPTOR_NAME}.${randomUUID()}.tmp`);
try {
if (existingBody === null) {
await fs.writeFile(tempPath, body, { encoding: "utf8", mode: 0o600 });
await fs.rename(tempPath, finalPath);
}
} finally {
await fs.rm(tempPath, { force: true }).catch(() => undefined);
}
return {
schema: descriptor.schema,
state: descriptor.state,
descriptorSha256,
baselineSha256: descriptor.baselineSha256,
finalHostSha256: descriptor.finalHostSha256,
workspaceId: descriptor.binding.workspaceId,
leaseId: descriptor.binding.leaseId,
providerLeaseId: descriptor.binding.providerLeaseId,
remoteCwd: descriptor.binding.remoteCwd,
resourceDisposition: descriptor.resourceDisposition,
};
}
export function readNativeWorkspaceSyncReference(
value: unknown,
): NativeWorkspaceSyncReference | null {
const candidate = parseObject(value);
if (
(candidate.schema !== DESCRIPTOR_SCHEMA && candidate.schema !== PREVIOUS_DESCRIPTOR_SCHEMA && candidate.schema !== LEGACY_DESCRIPTOR_SCHEMA) ||
(candidate.state !== "prepared" && candidate.state !== "finalized") ||
typeof candidate.descriptorSha256 !== "string" ||
!/^[0-9a-f]{64}$/.test(candidate.descriptorSha256) ||
typeof candidate.baselineSha256 !== "string" ||
!/^[0-9a-f]{64}$/.test(candidate.baselineSha256) ||
(candidate.finalHostSha256 !== null &&
candidate.finalHostSha256 !== undefined &&
(typeof candidate.finalHostSha256 !== "string" ||
!/^[0-9a-f]{64}$/.test(candidate.finalHostSha256))) ||
typeof candidate.workspaceId !== "string" ||
typeof candidate.leaseId !== "string" ||
typeof candidate.providerLeaseId !== "string" ||
typeof candidate.remoteCwd !== "string" ||
(candidate.resourceDisposition !== null &&
candidate.resourceDisposition !== undefined &&
candidate.resourceDisposition !== "keep_running" &&
candidate.resourceDisposition !== "stop_and_retain" &&
candidate.resourceDisposition !== "destroy")
) {
return null;
}
return {
schema: candidate.schema,
state: candidate.state,
descriptorSha256: candidate.descriptorSha256,
baselineSha256: candidate.baselineSha256,
finalHostSha256:
typeof candidate.finalHostSha256 === "string"
? candidate.finalHostSha256
: null,
workspaceId: candidate.workspaceId,
leaseId: candidate.leaseId,
providerLeaseId: candidate.providerLeaseId,
remoteCwd: candidate.remoteCwd,
resourceDisposition:
candidate.resourceDisposition === "keep_running" ||
candidate.resourceDisposition === "stop_and_retain" ||
candidate.resourceDisposition === "destroy"
? candidate.resourceDisposition
: null,
};
}
function validManifestShape(value: unknown): value is PathManifest {
const record = parseObject(value);
return record.kind === "path_manifest" && record.version === 1 && typeof record.filePath === "string"
&& typeof record.category === "string" && ["overlay", "deleted", "ignored", "baseline"].includes(record.category)
&& typeof record.count === "number" && Number.isSafeInteger(record.count) && record.count >= 0;
}
function* gitManifests(snapshot: GitWorkspaceSnapshot | null): Generator<PathManifest> {
if (!snapshot) return;
for (const paths of [snapshot.overlayPaths, snapshot.deletedPaths, snapshot.ignoredPaths]) if (isPathManifest(paths)) yield paths;
for (const repo of snapshot.repositories ?? []) yield* gitManifests(repo.snapshot);
}
async function verifyManifest(runId: string, manifest: PathManifest): Promise<void> {
const directory = descriptorDirectory(runId);
const name = path.basename(manifest.filePath);
const match = /^manifest\.([0-9a-f]{64})\.sqlite$/.exec(name);
if (!match || manifest.filePath !== path.join(directory, name)) throw new Error("native_workspace_sync_manifest_path_invalid");
for (const dir of [path.dirname(directory), directory]) {
const stat = await fs.lstat(dir);
if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("native_workspace_sync_manifest_root_invalid");
}
if (await manifestFileSha256(manifest.filePath) !== match[1]) throw new Error("native_workspace_sync_manifest_digest_mismatch");
// Verify every record before consumers can mutate a workspace on recovery.
for (const _record of readManifestRecords(manifest)) { /* bounded validation */ }
}
async function persistSnapshotManifests(runId: string, snapshot: NonNullable<PreparedAdapterExecutionTargetRuntime["workspaceSyncSnapshot"]>): Promise<void> {
const baseline = serializeDirectorySnapshot(snapshot.baseline);
const manifests = function* () {
if (baseline.version === 2) {
yield baseline.entries;
if (baseline.ignoredPaths && isPathManifest(baseline.ignoredPaths)) yield baseline.ignoredPaths;
}
yield* gitManifests(snapshot.gitSnapshot);
for (const repo of snapshot.repositories ?? []) yield* gitManifests(repo.snapshot);
};
const copies = new Map<string, string>();
for (const manifest of manifests()) {
const oldPath = manifest.filePath;
const previous = copies.get(oldPath);
if (previous) { manifest.filePath = previous; continue; }
// A shared reference may already have been updated by the baseline.
if (path.dirname(oldPath) === descriptorDirectory(runId)) continue;
const temporary = path.join(descriptorDirectory(runId), `manifest.${randomUUID()}.tmp`);
try {
const sourceStat = await fs.lstat(oldPath);
if (!sourceStat.isFile() || sourceStat.isSymbolicLink()) throw new Error("native_workspace_sync_manifest_invalid");
assertWorkspaceManifestDiskSpace(descriptorDirectory(runId), sourceStat.size);
await fs.copyFile(oldPath, temporary);
await fs.chmod(temporary, 0o600);
const digest = await manifestFileSha256(temporary);
const target = path.join(descriptorDirectory(runId), `manifest.${digest}.sqlite`);
await fs.rename(temporary, target);
copies.set(oldPath, target);
manifest.filePath = target;
} finally { await fs.rm(temporary, { force: true }); }
}
// Ownership tracks original scratch directories, never the new durable path.
await disposeDirectorySnapshot(snapshot.baseline);
await disposeGitWorkspaceSnapshot(snapshot.gitSnapshot);
for (const repo of snapshot.repositories ?? []) await disposeGitWorkspaceSnapshot(repo.snapshot);
}
function parseGitSnapshot(
value: unknown,
nested = false,
): GitWorkspaceSnapshot | null | undefined {
if (value === null) return null;
const candidate = parseObject(value);
const paths = [candidate.overlayPaths, candidate.deletedPaths, candidate.ignoredPaths];
if ((candidate.originUrl !== undefined && candidate.originUrl !== null && typeof candidate.originUrl !== "string") || typeof candidate.headCommit !== "string" || (candidate.branchName !== null && typeof candidate.branchName !== "string")
|| !paths.every((entries) => validManifestShape(entries) || (Array.isArray(entries) && entries.every((entry) =>
typeof entry === "string" && entry.length > 0 && !entry.includes("\0") && !path.posix.isAbsolute(entry) && !path.win32.isAbsolute(entry)
&& !entry.split(/[\\/]/).some((segment) => segment === ".." || segment === "."))))) return undefined;
const repositories: NonNullable<GitWorkspaceSnapshot["repositories"]> = [];
if (candidate.repositories !== undefined) {
if (nested || !Array.isArray(candidate.repositories)) return undefined;
const seen = new Set<string>();
for (const raw of candidate.repositories) {
const repo = parseObject(raw);
if (typeof repo.path !== "string" || !/^\.paperclip-repositories\/[a-zA-Z0-9_-]+$/.test(repo.path) || seen.has(repo.path)) return undefined;
const snapshot = parseGitSnapshot(repo.snapshot, true);
if (!snapshot) return undefined;
seen.add(repo.path);
repositories.push({ path: repo.path, snapshot });
}
}
return {
headCommit: candidate.headCommit,
branchName: candidate.branchName as string | null,
...(candidate.originUrl !== undefined ? { originUrl: candidate.originUrl as string | null } : {}),
overlayPaths: candidate.overlayPaths as WorkspacePaths,
deletedPaths: candidate.deletedPaths as WorkspacePaths,
ignoredPaths: candidate.ignoredPaths as WorkspacePaths,
...(repositories.length ? { repositories } : {}),
};
}
async function readDescriptor(input: {
runId: string;
reference: NativeWorkspaceSyncReference;
}): Promise<{
descriptor: NativeWorkspaceSyncDescriptor;
baseline: DirectorySnapshot;
}> {
const contentAddressedPath = descriptorPath(
input.runId,
input.reference.descriptorSha256,
);
// Early v1 writers used descriptor.json. Prefer the immutable,
// content-addressed name, but keep the digest-verified legacy path readable
// so an interrupted upgrade can still finish its already-proposed result.
const filePath = await fs
.lstat(contentAddressedPath)
.then(() => contentAddressedPath)
.catch(async (error: NodeJS.ErrnoException) => {
if (error.code !== "ENOENT") throw error;
const legacyPath = legacyDescriptorPath(input.runId);
await fs.lstat(legacyPath);
return legacyPath;
});
const stat = await fs.lstat(filePath);
if (
stat.isSymbolicLink() ||
!stat.isFile() ||
stat.size > MAX_DESCRIPTOR_BYTES
) {
throw new Error("native_workspace_sync_descriptor_invalid");
}
const body = await fs.readFile(filePath, "utf8");
if (
createHash("sha256").update(body).digest("hex") !==
input.reference.descriptorSha256
) {
throw new Error("native_workspace_sync_descriptor_digest_mismatch");
}
const parsed = JSON.parse(body) as unknown;
const candidate = parseObject(parsed);
const binding = parseObject(candidate.binding);
const rawBaseline = parseObject(candidate.baseline);
let baseline: DirectorySnapshot | null;
if ((candidate.schema === DESCRIPTOR_SCHEMA || candidate.schema === PREVIOUS_DESCRIPTOR_SCHEMA) && rawBaseline.version === 2
&& Array.isArray(rawBaseline.exclude) && rawBaseline.exclude.every((entry) => typeof entry === "string")
&& validManifestShape(rawBaseline.entries) && rawBaseline.entries.category === "baseline"
&& (rawBaseline.ignoredPaths === undefined || validManifestShape(rawBaseline.ignoredPaths))) {
await verifyManifest(input.runId, rawBaseline.entries);
if (validManifestShape(rawBaseline.ignoredPaths)) await verifyManifest(input.runId, rawBaseline.ignoredPaths);
baseline = openDirectorySnapshot(rawBaseline as unknown as Extract<SerializedDirectorySnapshot, { version: 2 }>);
} else {
baseline = parseDirectorySnapshot(candidate.baseline);
}
const gitSnapshot = parseGitSnapshot(candidate.gitSnapshot);
const repositoryContainer = candidate.repositories === undefined ? null : parseGitSnapshot({
headCommit: "", branchName: null, overlayPaths: [], deletedPaths: [], ignoredPaths: [], repositories: candidate.repositories,
});
if (candidate.repositories !== undefined && (!repositoryContainer || candidate.schema !== DESCRIPTOR_SCHEMA || gitSnapshot)) throw new Error("native_workspace_sync_repositories_invalid");
const repositories = repositoryContainer?.repositories;
for (const repo of repositories ?? []) for (const manifest of gitManifests(repo.snapshot)) await verifyManifest(input.runId, manifest);
for (const manifest of gitManifests(gitSnapshot ?? null)) await verifyManifest(input.runId, manifest);
const rawSeed =
candidate.seed === null || candidate.seed === undefined
? null
: parseObject(candidate.seed);
const seed =
rawSeed === null
? null
: typeof rawSeed.workspaceArchiveSha256 === "string" &&
/^[0-9a-f]{64}$/.test(rawSeed.workspaceArchiveSha256) &&
(rawSeed.gitArchiveSha256 === null ||
(typeof rawSeed.gitArchiveSha256 === "string" &&
/^[0-9a-f]{64}$/.test(rawSeed.gitArchiveSha256)))
? {
workspaceArchiveSha256: rawSeed.workspaceArchiveSha256,
gitArchiveSha256: rawSeed.gitArchiveSha256 as string | null,
}
: undefined;
const finalHostSha256 =
typeof candidate.finalHostSha256 === "string" &&
/^[0-9a-f]{64}$/.test(candidate.finalHostSha256)
? candidate.finalHostSha256
: candidate.finalHostSha256 === null
? null
: undefined;
const resourceDisposition =
candidate.resourceDisposition === "keep_running" ||
candidate.resourceDisposition === "stop_and_retain" ||
candidate.resourceDisposition === "destroy"
? candidate.resourceDisposition
: candidate.resourceDisposition === null
? null
: undefined;
if (
(candidate.schema !== DESCRIPTOR_SCHEMA && candidate.schema !== PREVIOUS_DESCRIPTOR_SCHEMA && candidate.schema !== LEGACY_DESCRIPTOR_SCHEMA) ||
(candidate.state !== "prepared" && candidate.state !== "finalized") ||
!baseline ||
gitSnapshot === undefined ||
seed === undefined ||
binding.runId !== input.runId ||
binding.workspaceId !== input.reference.workspaceId ||
binding.leaseId !== input.reference.leaseId ||
binding.providerLeaseId !== input.reference.providerLeaseId ||
binding.remoteCwd !== input.reference.remoteCwd ||
typeof binding.companyId !== "string" ||
typeof binding.localCwd !== "string" ||
!path.isAbsolute(binding.localCwd) ||
typeof binding.remoteCwd !== "string" ||
!path.posix.isAbsolute(binding.remoteCwd) ||
typeof candidate.baselineSha256 !== "string" ||
candidate.baselineSha256 !== input.reference.baselineSha256 ||
directorySnapshotSha256(baseline) !== candidate.baselineSha256 ||
typeof candidate.createdAt !== "string" ||
!Number.isFinite(Date.parse(candidate.createdAt)) ||
candidate.schema !== input.reference.schema ||
candidate.state !== input.reference.state ||
finalHostSha256 === undefined ||
finalHostSha256 !== input.reference.finalHostSha256 ||
resourceDisposition === undefined ||
resourceDisposition !== input.reference.resourceDisposition ||
(candidate.state === "prepared" &&
(candidate.finalizedAt !== null || finalHostSha256 !== null)) ||
(candidate.state === "finalized" &&
(typeof candidate.finalizedAt !== "string" ||
!Number.isFinite(Date.parse(candidate.finalizedAt)) ||
finalHostSha256 === null))
) {
throw new Error("native_workspace_sync_descriptor_binding_mismatch");
}
return {
descriptor: {
schema: candidate.schema,
binding: {
runId: binding.runId as string,
companyId: binding.companyId,
workspaceId: binding.workspaceId as string,
leaseId: binding.leaseId as string,
providerLeaseId: binding.providerLeaseId as string,
localCwd: binding.localCwd,
remoteCwd: binding.remoteCwd as string,
},
state: candidate.state,
baselineSha256: candidate.baselineSha256,
baseline: candidate.baseline as SerializedDirectorySnapshot,
gitSnapshot,
...(repositories?.length ? { repositories } : {}),
seed,
createdAt: candidate.createdAt,
finalizedAt:
typeof candidate.finalizedAt === "string"
? candidate.finalizedAt
: null,
finalHostSha256,
resourceDisposition,
},
baseline,
};
}
async function persistRunReference(
db: Db,
runId: string,
reference: NativeWorkspaceSyncReference,
): Promise<void> {
await db.transaction(async (tx) => {
const run = await tx
.select({ runnerProfileJson: heartbeatRuns.runnerProfileJson })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.for("update")
.limit(1)
.then((rows) => rows[0] ?? null);
if (!run) throw new Error("native_workspace_sync_run_missing");
await tx
.update(heartbeatRuns)
.set({
runnerProfileJson: {
...parseObject(run.runnerProfileJson),
nativeWorkspaceSync: reference,
},
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, runId));
});
}
async function persistLeaseStamp(input: {
db: Db;
leaseId: string;
stamp: Record<string, unknown>;
}): Promise<void> {
await input.db.transaction(async (tx) => {
const lease = await tx
.select({ metadata: environmentLeases.metadata })
.from(environmentLeases)
.where(eq(environmentLeases.id, input.leaseId))
.for("update")
.limit(1)
.then((rows) => rows[0] ?? null);
if (!lease) throw new Error("native_workspace_sync_lease_missing");
await notifyEnvironmentCleanup(tx);
await tx
.update(environmentLeases)
.set({
metadata: {
...parseObject(lease.metadata),
nativeWorkspaceSync: input.stamp,
},
updatedAt: new Date(),
})
.where(eq(environmentLeases.id, input.leaseId));
});
}
function providerLeaseIdFor(input: {
target: Extract<AdapterExecutionTarget, { transport: "sandbox" }>;
lease: EnvironmentLease;
}): string {
const providerLeaseId =
input.target.sandboxLeaseAcquisition?.providerLeaseId ??
input.lease.providerLeaseId;
if (!providerLeaseId) {
throw new Error("native_workspace_sync_provider_lease_missing");
}
return providerLeaseId;
}
function shellQuote(value: string): string {
return `'${value.replaceAll("'", "'\\''")}'`;
}
async function writeRemoteStamp(input: {
target: Extract<AdapterExecutionTarget, { transport: "sandbox" }>;
stamp: Record<string, unknown>;
}): Promise<void> {
if (!input.target.runner) {
throw new Error("native_workspace_sync_runner_missing");
}
const runtimeDir = path.posix.join(
input.target.remoteCwd,
".paperclip-runtime",
"paperclip-runner",
);
const stampPath = path.posix.join(runtimeDir, REMOTE_STAMP_NAME);
const tempPath = `${stampPath}.tmp`;
const encoded = Buffer.from(
`${canonicalJson(input.stamp)}\n`,
"utf8",
).toString("base64");
const result = await input.target.runner.execute({
command: input.target.shellCommand ?? "sh",
args: [
"-c",
`umask 077; mkdir -p ${shellQuote(runtimeDir)} && printf %s ${shellQuote(encoded)} | base64 -d > ${shellQuote(tempPath)} && mv ${shellQuote(tempPath)} ${shellQuote(stampPath)}`,
],
cwd: "/",
timeoutMs: 15_000,
bypassSession: true,
});
if (result.timedOut || result.exitCode !== 0) {
throw new Error("native_workspace_sync_remote_stamp_failed");
}
}
async function remoteStampMatches(input: {
target: Extract<AdapterExecutionTarget, { transport: "sandbox" }>;
expected: Record<string, unknown>;
gitSnapshot?: GitWorkspaceSnapshot | null;
repositories?: NonNullable<GitWorkspaceSnapshot["repositories"]>;
}): Promise<boolean> {
if (!input.target.runner) return false;
const stampPath = path.posix.join(
input.target.remoteCwd,
".paperclip-runtime",
"paperclip-runner",
REMOTE_STAMP_NAME,
);
// File hashes do not detect empty commits or branch changes. A fresh host
// snapshot may authorize replacement only if the retained sandbox actually
// starts from that same Git identity, including each managed repository.
const gitChecks: string[] = [];
const checkGit = (remoteDir: string, snapshot: GitWorkspaceSnapshot) => {
const git = `git -C ${shellQuote(remoteDir)}`;
gitChecks.push(`test "$(${git} rev-parse HEAD)" = ${shellQuote(snapshot.headCommit)}`);
gitChecks.push(snapshot.branchName === null
? `(${git} symbolic-ref --quiet HEAD >/dev/null 2>&1; test $? -eq 1)`
: `test "$(${git} symbolic-ref --quiet --short HEAD)" = ${shellQuote(snapshot.branchName)}`);
if (snapshot.originUrl !== undefined) gitChecks.push(snapshot.originUrl === null
? `! ${git} config --get remote.origin.url >/dev/null 2>&1`
: `test "$(${git} config --get remote.origin.url)" = ${shellQuote(snapshot.originUrl)}`);
for (const repository of snapshot.repositories ?? []) {
checkGit(path.posix.join(remoteDir, repository.path), repository.snapshot);
}
};
if (input.gitSnapshot) checkGit(input.target.remoteCwd, input.gitSnapshot);
for (const repo of input.repositories ?? []) checkGit(path.posix.join(input.target.remoteCwd, repo.path), repo.snapshot);
const result = await input.target.runner.execute({
command: input.target.shellCommand ?? "sh",
args: [
"-c",
[...gitChecks, `test -f ${shellQuote(stampPath)}`, `cat ${shellQuote(stampPath)}`].join(" && "),
],
cwd: "/",
timeoutMs: 15_000,
bypassSession: true,
});
if (result.timedOut || result.exitCode !== 0) return false;
try {
return (
canonicalJson(JSON.parse(result.stdout)) === canonicalJson(input.expected)
);
} catch {
return false;
}
}
function leaseStamp(input: {
lease: EnvironmentLease;
workspaceId: string;
providerLeaseId: string;
remoteCwd: string;
}): Record<string, unknown> | null {
const stamp = parseObject(
parseObject(input.lease.metadata).nativeWorkspaceSync,
);
if (
stamp.schema !== STAMP_SCHEMA ||
stamp.workspaceId !== input.workspaceId ||
stamp.providerLeaseId !== input.providerLeaseId ||
stamp.remoteCwd !== input.remoteCwd ||
typeof stamp.hostSha256 !== "string" ||
!/^[0-9a-f]{64}$/.test(stamp.hostSha256)
) {
return null;
}
return stamp;
}
function finalizedWorkspaceStamp(input: {
descriptor: NativeWorkspaceSyncDescriptor;
hostSha256: string;
}): Record<string, unknown> {
return {
schema: STAMP_SCHEMA,
workspaceId: input.descriptor.binding.workspaceId,
providerLeaseId: input.descriptor.binding.providerLeaseId,
remoteCwd: input.descriptor.binding.remoteCwd,
hostSha256: input.hostSha256,
finalizedRunId: input.descriptor.binding.runId,
};
}
async function prepareRuntime(input: {
runId: string;
target: Extract<AdapterExecutionTarget, { transport: "sandbox" }>;
workspaceLocalDir: string;
mode: WorkspaceInboundMode;
baseline?: DirectorySnapshot;
gitSnapshot?: GitWorkspaceSnapshot | null;
repositories?: NonNullable<GitWorkspaceSnapshot["repositories"]>;
durableSeed?: WorkspaceDurableSeedPaths;
seedCacheDirectory?: string;
seedCacheCompanyDirectory?: string;
onPhase?: RuntimeSpanRunner;
onCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise<void>;
}): Promise<PreparedAdapterExecutionTargetRuntime> {
const phase = {
host_current: "workspace_sync.prepare_host_current",
durable_seed: "workspace_sync.prepare_durable_seed",
adopt_remote: "workspace_sync.prepare_adopt_remote",
} as const;
return measureRunPhase(phase[input.mode], () => prepareAdapterExecutionTargetRuntime({
runId: input.runId,
target: input.target,
adapterKey: "paperclip-runner",
workspaceLocalDir: input.workspaceLocalDir,
workspaceRemoteDir: input.target.remoteCwd,
workspaceInboundMode: input.mode,
workspaceSnapshotIsolation: "worker",
workspaceDurableSeed: input.durableSeed,
workspaceBaseline: input.baseline,
workspaceGitSnapshot: input.gitSnapshot,
workspaceRepositories: input.repositories ?? (input.baseline ? [] : undefined),
workspaceCheckpoint: true,
onWorkspaceCheckpoint: input.onCheckpoint,
workspaceSeedCacheDirectory: input.seedCacheDirectory,
workspaceSeedCacheCompanyDirectory: input.seedCacheCompanyDirectory,
runtimeSpan: input.onPhase,
}));
}
async function finalizePreparedRuntime(input: {
db: Db;
runId: string;
target: Extract<AdapterExecutionTarget, { transport: "sandbox" }>;
runtime: PreparedAdapterExecutionTargetRuntime;
descriptor: NativeWorkspaceSyncDescriptor;
assertOwnership?: () => Promise<void>;
onPhase?: RuntimeSpanRunner;
onCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise<void>;
}): Promise<NativeWorkspaceSyncReference> {
if (input.onPhase) await input.onPhase("workspace.restore", () => input.runtime.restoreWorkspace());
else await input.runtime.restoreWorkspace();
return measureRunPhase("execution.workspace_commit", async () => {
await input.assertOwnership?.();
const finalSnapshot =
await import("@paperclipai/adapter-utils/workspace-restore-merge").then(
({ captureDirectorySnapshotOffThread: captureDirectorySnapshot }) =>
captureDirectorySnapshot(input.descriptor.binding.localCwd, {
exclude: input.runtime.workspaceSyncSnapshot?.baseline.exclude ?? [],
ignoredPaths: input.runtime.workspaceSyncSnapshot?.baseline.ignoredPaths,
diskBacked: true,
}),
);
const finalHostSha256 = directorySnapshotSha256(finalSnapshot);
await disposeDirectorySnapshot(finalSnapshot);
const stamp = finalizedWorkspaceStamp({
descriptor: input.descriptor,
hostSha256: finalHostSha256,
});
await writeRemoteStamp({ target: input.target, stamp });
const finalizedDescriptor: NativeWorkspaceSyncDescriptor = {
...input.descriptor,
state: "finalized",
finalizedAt: new Date().toISOString(),
finalHostSha256,
};
await input.assertOwnership?.();
const reference = await writeDescriptor(finalizedDescriptor);
await persistRunReference(input.db, input.runId, reference);
await persistLeaseStamp({
db: input.db,
leaseId: input.descriptor.binding.leaseId,
stamp,
});
return reference;
});
}
export async function prepareNativeWorkspaceSync(input: {
db: Db;
runId: string;
companyId: string;
workspaceId: string;
workspaceLocalDir: string;
target: AdapterExecutionTarget | null;
lease: EnvironmentLease;
restartRecovery?: NativeRestartRecoveryClaim;
sameRunRecovery?: boolean;
resourceDisposition?: NativeWorkspaceResourceDisposition;
onPhase?: RuntimeSpanRunner;
onCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise<void>;
onEvent?: (event: AdapterRuntimeEvent) => Promise<void>;
}): Promise<PreparedNativeWorkspaceSync | null> {
if (input.target?.kind !== "remote" || input.target.transport !== "sandbox" || input.target.runnerTask) {
return null;
}
const target = input.target;
const providerLeaseId = providerLeaseIdFor({ target, lease: input.lease });
// Reattachment may never seed a replacement sandbox over a still-live run.
if (input.restartRecovery?.kind === "reattach_remote_runner" &&
(input.restartRecovery.runId !== input.runId ||
input.restartRecovery.remote.providerLeaseId !== providerLeaseId ||
input.restartRecovery.remote.remoteCwd !== target.remoteCwd)) {
throw new Error("native_remote_recovery_lease_mismatch");
}
const run = await measureRunPhase("workspace_sync.run_read", () => input.db
.select({ runnerProfileJson: heartbeatRuns.runnerProfileJson })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.runId))
.limit(1)
.then((rows) => rows[0] ?? null));
if (!run) throw new Error("native_workspace_sync_run_missing");
const existingReference = readNativeWorkspaceSyncReference(
parseObject(run.runnerProfileJson).nativeWorkspaceSync,
);
let runtime: PreparedAdapterExecutionTargetRuntime | undefined;
let descriptor: NativeWorkspaceSyncDescriptor;
let mode: WorkspaceInboundMode = "host_current";
let reuseReason: StartupReuseReason = "created";
const seedPaths = durableSeedPaths(input.runId);
await measureRunPhase("workspace_sync.descriptor_directory", () => ensureRunDirectory(input.runId));
try {
if (existingReference) {
const existing = await measureRunPhase("workspace_sync.descriptor_read", () => readDescriptor({
runId: input.runId,
reference: existingReference,
}));
if (
existing.descriptor.binding.companyId !== input.companyId ||
existing.descriptor.binding.workspaceId !== input.workspaceId ||
path.resolve(existing.descriptor.binding.localCwd) !==
path.resolve(input.workspaceLocalDir) ||
existing.descriptor.binding.remoteCwd !== target.remoteCwd
) {
throw new Error("native_workspace_sync_descriptor_binding_mismatch");
}
const sameProviderLease =
existing.descriptor.binding.providerLeaseId === providerLeaseId;
mode = classifyNativeWorkspaceInbound({
kind: "existing_run",
restartRecovery: Boolean(input.restartRecovery),
sameRunRecovery: input.sameRunRecovery,
sameProviderLease,
});
reuseReason = sameProviderLease ? "verified_reuse" : "replacement";
const durableSeed =
mode === "durable_seed"
? existing.descriptor.seed
? await measureRunPhase("workspace_sync.durable_seed_verify", () => verifiedDurableSeed({
runId: input.runId,
seed: existing.descriptor.seed!,
gitSnapshot: existing.descriptor.gitSnapshot,
repositories: existing.descriptor.repositories,
}))
: (() => {
throw new Error("workspace_sync_out_unrecoverable");
})()
: undefined;
runtime = await prepareRuntime({
runId: input.runId,
target,
onPhase: input.onPhase,
onCheckpoint: input.onCheckpoint,
workspaceLocalDir: input.workspaceLocalDir,
mode,
baseline: existing.baseline,
gitSnapshot: existing.descriptor.gitSnapshot,
repositories: existing.descriptor.repositories,
durableSeed,
});
descriptor = {
...existing.descriptor,
binding: {
...existing.descriptor.binding,
leaseId: target.leaseId ?? input.lease.id,
providerLeaseId,
},
};
} else {
const acquisition = target.sandboxLeaseAcquisition?.outcome ?? null;
reuseReason = acquisition === "resumed" ? "cache_miss" : acquisition === "replacement" ? "replacement" : "created";
const priorStamp = leaseStamp({
lease: input.lease,
workspaceId: input.workspaceId,
providerLeaseId,
remoteCwd: target.remoteCwd,
});
mode = classifyNativeWorkspaceInbound({
kind: "new_run",
acquisition,
hasPriorStamp: priorStamp !== null,
});
const seedCacheCompanyDirectory = path.join(resolvePaperclipInstanceRoot(), STATE_ROOT_NAME, "generations", requireSafeSegment(input.companyId, "company_id"));
const seedCacheDirectory = path.join(seedCacheCompanyDirectory, requireSafeSegment(input.workspaceId, "workspace_id"));
runtime = await prepareRuntime({
runId: input.runId,
target,
onPhase: input.onPhase,
onCheckpoint: input.onCheckpoint,
workspaceLocalDir: input.workspaceLocalDir,
mode,
seedCacheCompanyDirectory,
seedCacheDirectory,
durableSeed: {
workspaceArchivePath: seedPaths.workspaceArchivePath,
gitArchivePath: seedPaths.gitArchivePath,
},
});
const currentSnapshot = runtime.workspaceSyncSnapshot;
if (!currentSnapshot) {
throw new Error("native_workspace_sync_snapshot_missing");
}
let workspaceArchiveSha256 = await sha256File(seedPaths.workspaceArchivePath);
let gitArchiveSha256 = currentSnapshot.gitSnapshot || currentSnapshot.repositories?.length
? await sha256File(seedPaths.gitArchivePath)
: null;
if (acquisition === "resumed" && priorStamp) {
const currentHostSha256 = directorySnapshotSha256(
currentSnapshot.baseline,
);
const verifiedWarmAdoption =
priorStamp.hostSha256 === currentHostSha256 &&
(await measureRunPhase("workspace_sync.warm_stamp", () => remoteStampMatches({ target, expected: priorStamp, gitSnapshot: currentSnapshot.gitSnapshot, repositories: currentSnapshot.repositories })));
if (verifiedWarmAdoption) {
mode = "adopt_remote";
reuseReason = "verified_reuse";
} else {
mode = "host_current";
reuseReason = "workspace_changed";
let preparedSeed: WorkspaceDurableSeedPaths | undefined;
try {
preparedSeed = await measureRunPhase("workspace_sync.durable_seed_verify", () => verifiedDurableSeed({ runId: input.runId,
seed: { workspaceArchiveSha256, gitArchiveSha256 },
gitSnapshot: currentSnapshot.gitSnapshot, repositories: currentSnapshot.repositories }));
} catch (error) {
if (!(error instanceof Error) || error.message !== "workspace_sync_out_unrecoverable") throw error;
}
runtime = await prepareRuntime({
runId: input.runId,
target,
workspaceLocalDir: input.workspaceLocalDir,
mode: preparedSeed ? "durable_seed" : mode,
seedCacheCompanyDirectory,
seedCacheDirectory,
baseline: currentSnapshot.baseline,
gitSnapshot: currentSnapshot.gitSnapshot,
repositories: currentSnapshot.repositories,
onPhase: input.onPhase,
onCheckpoint: input.onCheckpoint,
durableSeed: preparedSeed ?? seedPaths,
});
if (!preparedSeed) {
workspaceArchiveSha256 = await sha256File(seedPaths.workspaceArchivePath);
gitArchiveSha256 = currentSnapshot.gitSnapshot || currentSnapshot.repositories?.length
? await sha256File(seedPaths.gitArchivePath) : null;
}
}
}
const snapshot = runtime.workspaceSyncSnapshot;
if (!snapshot) throw new Error("native_workspace_sync_snapshot_missing");
await measureRunPhase("workspace_sync.manifests_persist", () => persistSnapshotManifests(input.runId, snapshot));
const now = new Date().toISOString();
descriptor = {
schema: DESCRIPTOR_SCHEMA,
binding: {
runId: input.runId,
companyId: input.companyId,
workspaceId: input.workspaceId,
leaseId: target.leaseId ?? input.lease.id,
providerLeaseId,
localCwd: path.resolve(input.workspaceLocalDir),
remoteCwd: target.remoteCwd,
},
state: "prepared",
baselineSha256: directorySnapshotSha256(snapshot.baseline),
baseline: serializeDirectorySnapshot(snapshot.baseline),
gitSnapshot: snapshot.gitSnapshot,
...(snapshot.repositories?.length ? { repositories: snapshot.repositories } : {}),
seed: { workspaceArchiveSha256, gitArchiveSha256 },
createdAt: now,
finalizedAt: null,
finalHostSha256: null,
resourceDisposition: input.resourceDisposition ?? null,
};
}
const preparedRuntime = runtime;
let reference = await measureRunPhase("workspace_sync.descriptor_write", () => writeDescriptor(descriptor));
await measureRunPhase("workspace_sync.reference_write", () => persistRunReference(input.db, input.runId, reference));
await emitStartupReuse(input.onEvent, "workspace", {
state: mode === "adopt_remote" ? "reused" : "fresh",
reason: reuseReason,
});
let restorePromise: Promise<void> | null = null;
return {
mode,
get reference() {
return reference;
},
restoreWorkspace: async (assertOwnership) => {
if (!restorePromise) {
restorePromise = finalizePreparedRuntime({
db: input.db,
runId: input.runId,
target,
runtime: preparedRuntime,
descriptor,
assertOwnership,
onPhase: input.onPhase,
onCheckpoint: input.onCheckpoint,
})
.then((finalizedReference) => {
reference = finalizedReference;
})
.catch((error) => {
restorePromise = null;
throw error;
});
}
await restorePromise;
},
cleanup: async () => { await preparedRuntime.cleanupWorkspaceSnapshot?.(); await cleanupNativeWorkspaceSync(input.runId); },
};
} catch (error) {
await runtime?.cleanupWorkspaceSnapshot?.();
if (!existingReference) await cleanupNativeWorkspaceSync(input.runId);
throw error;
}
}
export async function readNativeWorkspaceSyncLocalCwd(input: {
companyId: string;
runId: string;
workspaceId: string;
reference: NativeWorkspaceSyncReference;
}): Promise<string> {
const existing = await readDescriptor(input);
try {
if (existing.descriptor.binding.companyId !== input.companyId ||
existing.descriptor.binding.workspaceId !== input.workspaceId) {
throw new Error("native_workspace_sync_descriptor_binding_mismatch");
}
return existing.descriptor.binding.localCwd;
} finally {
await disposeDirectorySnapshot(existing.baseline);
await disposeGitWorkspaceSnapshot(existing.descriptor.gitSnapshot);
for (const repository of existing.descriptor.repositories ?? []) {
await disposeGitWorkspaceSnapshot(repository.snapshot);
}
}
}
export async function resumeNativeWorkspaceSync(input: {
db: Db;
runId: string;
target: AdapterExecutionTarget;
assertOwnership?: () => Promise<void>;
onPhase?: RuntimeSpanRunner;
onCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise<void>;
}): Promise<boolean> {
if (input.target.kind !== "remote" || input.target.transport !== "sandbox") {
throw new Error("workspace_sync_out_unrecoverable");
}
const run = await input.db
.select({ runnerProfileJson: heartbeatRuns.runnerProfileJson })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.runId))
.limit(1)
.then((rows) => rows[0] ?? null);
const reference = readNativeWorkspaceSyncReference(
parseObject(run?.runnerProfileJson).nativeWorkspaceSync,
);
if (!reference) return false;
const existing = await readDescriptor({ runId: input.runId, reference });
const providerLeaseId =
input.target.sandboxLeaseAcquisition?.providerLeaseId ??
reference.providerLeaseId;
if (
providerLeaseId !== reference.providerLeaseId ||
input.target.remoteCwd !== reference.remoteCwd
) {
throw new Error("workspace_sync_out_unrecoverable");
}
await input.assertOwnership?.();
if (
existing.descriptor.state === "finalized" &&
existing.descriptor.finalHostSha256
) {
const stamp = finalizedWorkspaceStamp({
descriptor: existing.descriptor,
hostSha256: existing.descriptor.finalHostSha256,
});
if (
!(await remoteStampMatches({ target: input.target, expected: stamp }))
) {
await writeRemoteStamp({ target: input.target, stamp });
}
await persistLeaseStamp({
db: input.db,
leaseId: existing.descriptor.binding.leaseId,
stamp,
});
return true;
}
const runtime = await prepareRuntime({
runId: input.runId,
target: input.target,
onPhase: input.onPhase,
onCheckpoint: input.onCheckpoint,
workspaceLocalDir: existing.descriptor.binding.localCwd,
mode: "adopt_remote",
baseline: existing.baseline,
gitSnapshot: existing.descriptor.gitSnapshot,
repositories: existing.descriptor.repositories,
});
await finalizePreparedRuntime({
db: input.db,
runId: input.runId,
target: input.target,
runtime,
descriptor: existing.descriptor,
assertOwnership: input.assertOwnership,
onPhase: input.onPhase,
onCheckpoint: input.onCheckpoint,
});
return true;
}
export async function cleanupNativeWorkspaceSync(runId: string): Promise<void> {
await fs.rm(descriptorDirectory(runId), { recursive: true, force: true });
}
export const nativeWorkspaceSyncInternals = {
descriptorPath,
legacyDescriptorPath,
durableSeedPaths,
persistSnapshotManifests,
writeDescriptor,
readDescriptor,
readReference: readNativeWorkspaceSyncReference,
sha256,
};