mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 17:45:53 +01:00
Keep finalization failures stable and recover valid older leases
Validate paginated legacy lease candidates before choosing a workspace, preserving Postgres timestamp precision. Recover failed saves through a new authorized run and cover post-save finalization failures without replaying terminal cleanup. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
e60c6099c5
commit
816045d643
4 files changed
+104
-46
No files matched your search
@@ -2,6 +2,11 @@
|
||||
|
||||
Shared folders can contain saved files from several sandboxes. The cached-file inspector keeps earlier run failures visible with a **View failed run** link, separately from the last successful save time and direct file-operation errors.
|
||||
|
||||
Finalization runs once per execution, including its file flush and session-state
|
||||
publication. Repeated error cleanup keeps the original failure visible. A later
|
||||
authorized run recovers unsaved edits from the retained sandbox before loading
|
||||
incoming shared files; it does not rewrite the failed run as successful.
|
||||
|
||||
The deployed acceptance entry point is `pnpm test:e2e:work-folders:deployed`.
|
||||
Set `PAPERCLIP_DEPLOYED_STACK_MANIFEST` to a JSON manifest matching
|
||||
`tests/runner-e2e/deployed-stack.ts`, `PAPERCLIP_DEPLOYED_STACK_AUTH` to a private
|
||||
|
||||
@@ -2,7 +2,7 @@ import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { eq, inArray, sql } from "drizzle-orm";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import os from "node:os";
|
||||
@@ -68,6 +68,16 @@ describe("shared sandbox work-folder lifecycle", () => {
|
||||
executionWorkspaceId: null, executionWorkspacePreference: null,
|
||||
environment: { id: environmentId, driver: "sandbox" as const, config: { reuseLease: true } } };
|
||||
expect(await findUnboundLegacyTaskWorkspace(db, input)).toBe(workspaceId);
|
||||
const malformedIds = Array.from({ length: adapterType === "codex_local" ? 101 : 1 }, () => randomUUID());
|
||||
await db.insert(environmentLeases).values(malformedIds.map((id) => ({ id, companyId, environmentId, issueId: task,
|
||||
executionWorkspaceId: workspaceId, heartbeatRunId: runId, status: "retained", leasePolicy: "reuse_by_environment",
|
||||
provider: "daytona", providerLeaseId: "invalid-newer-sandbox", createdAt: new Date(Date.now() + 1000),
|
||||
metadata: { ...metadata, reusableSandboxLease: { ...metadata.reusableSandboxLease, executionWorkspaceId: randomUUID() } } })));
|
||||
await db.update(environmentLeases).set({ createdAt: sql`'2026-01-01T00:00:00.123455Z'::timestamptz` }).where(eq(environmentLeases.id, leaseId));
|
||||
await db.update(environmentLeases).set({ createdAt: sql`'2026-01-01T00:00:00.123456Z'::timestamptz` }).where(inArray(environmentLeases.id, malformedIds));
|
||||
// Includes a second page: validate identities before selecting a candidate.
|
||||
expect(await findUnboundLegacyTaskWorkspace(db, input)).toBe(workspaceId);
|
||||
await db.delete(environmentLeases).where(inArray(environmentLeases.id, malformedIds));
|
||||
for (const bad of [{ companyId: randomUUID() }, { issueId: randomUUID() }, { projectId: randomUUID() },
|
||||
{ agentId: randomUUID() }, { responsibleUserId: null }, { responsibleUserId: "another-user" },
|
||||
{ adapterType: "other-adapter" }, { executionWorkspaceId: randomUUID() },
|
||||
@@ -185,12 +195,14 @@ describe("shared sandbox work-folder lifecycle", () => {
|
||||
}, 120_000);
|
||||
|
||||
it("retains unaudited edits and retries without misreporting a completed checkpoint", async () => {
|
||||
const run = await prepare(path.join(root, "activity-failure"), randomUUID());
|
||||
const physicalId = randomUUID();
|
||||
const run = await prepare(path.join(root, "activity-failure"), physicalId);
|
||||
let completingRun = run;
|
||||
await run.flush();
|
||||
const [before] = await db.select().from(workFolderRuns).where(eq(workFolderRuns.runId, run.manifest.runId));
|
||||
await fs.writeFile(path.join(run.home, "task/activity-proof.txt"), "saved after logging recovers");
|
||||
const beforeCompletion = vi.fn(async () => {
|
||||
const [state] = await db.select().from(workFolderRuns).where(eq(workFolderRuns.runId, run.manifest.runId));
|
||||
const [state] = await db.select().from(workFolderRuns).where(eq(workFolderRuns.runId, completingRun.manifest.runId));
|
||||
expect(state?.state).toBe("saved");
|
||||
expect(state?.manifest.finalCheckpointAt).toBeUndefined();
|
||||
const folder = await workFolderService(db, storage).ensure({ companyId, scope: "task", ownerId: taskId });
|
||||
@@ -207,9 +219,12 @@ describe("shared sandbox work-folder lifecycle", () => {
|
||||
expect((await workFolderService(db, storage).list(folder)).files.some((file) => file.path === "activity-proof.txt")).toBe(false);
|
||||
expect(await fs.readFile(path.join(run.home, "task/activity-proof.txt"), "utf8")).toBe("saved after logging recovers");
|
||||
} finally { activity.mockRestore(); }
|
||||
await run.stop(beforeCompletion); active.splice(active.indexOf(run), 1);
|
||||
await expect(run.stop(beforeCompletion)).rejects.toThrow("activity unavailable");
|
||||
active.splice(active.indexOf(run), 1);
|
||||
completingRun = await prepare(run.home, randomUUID(), physicalId);
|
||||
await completingRun.stop(beforeCompletion); active.splice(active.indexOf(completingRun), 1);
|
||||
expect(beforeCompletion).toHaveBeenCalledTimes(1);
|
||||
const [saved] = await db.select().from(workFolderRuns).where(eq(workFolderRuns.runId, run.manifest.runId));
|
||||
const [saved] = await db.select().from(workFolderRuns).where(eq(workFolderRuns.runId, completingRun.manifest.runId));
|
||||
expect(saved?.state).toBe("saved");
|
||||
expect(saved?.manifest.finalCheckpointAt).toBeTruthy();
|
||||
expect(saved!.lastSavedAt!.getTime()).toBeGreaterThan(before!.lastSavedAt!.getTime());
|
||||
@@ -217,6 +232,21 @@ describe("shared sandbox work-folder lifecycle", () => {
|
||||
expect((await workFolderService(db, storage).list(folder)).files.some((file) => file.path === "activity-proof.txt")).toBe(true);
|
||||
}, 120_000);
|
||||
|
||||
it("does not repeat failed post-save finalization during teardown", async () => {
|
||||
const run = await prepare(path.join(root, "completion-failure"), randomUUID());
|
||||
const finish = vi.fn().mockRejectedValueOnce(new Error("session persistence unavailable")).mockResolvedValue(undefined);
|
||||
const stopped = run.stop(finish);
|
||||
await expect(stopped).rejects.toThrow("session persistence unavailable");
|
||||
expect(run.stop(finish)).toBe(stopped);
|
||||
await expect(run.stop(finish)).rejects.toThrow("session persistence unavailable");
|
||||
expect(finish).toHaveBeenCalledTimes(1);
|
||||
const [state] = await db.select().from(workFolderRuns).where(eq(workFolderRuns.runId, run.manifest.runId));
|
||||
expect(state?.state).toBe("saved");
|
||||
expect(state?.lastSavedAt).toBeTruthy();
|
||||
expect(state?.manifest.finalCheckpointAt).toBeUndefined();
|
||||
active.splice(active.indexOf(run), 1);
|
||||
}, 120_000);
|
||||
|
||||
it("reuses clones and restores saved unpushed work, staged changes, and task files after losing the sandbox", async () => {
|
||||
const repositoryTaskId = randomUUID();
|
||||
await db.insert(issues).values({ id: repositoryTaskId, companyId, projectId, title: "Repository recovery", assigneeAgentId: agentId });
|
||||
@@ -310,10 +340,14 @@ describe("shared sandbox work-folder lifecycle", () => {
|
||||
expect((await svc.list(folder)).files).toHaveLength(0);
|
||||
expect(await fs.readFile(path.join(run.home, "user/private"), "utf8")).toBe("pending private edit");
|
||||
await db.update(companyMemberships).set({ status: "active" }).where(eq(companyMemberships.id, membership!.id));
|
||||
await run.stop(); active.splice(active.indexOf(run), 1);
|
||||
await expect(run.stop()).rejects.toThrow("no longer authorized");
|
||||
active.splice(active.indexOf(run), 1);
|
||||
const retry = await prepare(run.home, randomUUID(), leaseId, userId);
|
||||
await retry.stop(); active.splice(active.indexOf(retry), 1);
|
||||
expect((await svc.list(folder)).files.map((file) => file.path)).toContain("private");
|
||||
}, 120_000);
|
||||
|
||||
it("does not publish a partial repository checkpoint and retries a failed final save", async () => {
|
||||
it("does not publish a partial repository checkpoint and recovers a failed final save in a new run", async () => {
|
||||
const leaseId = randomUUID();
|
||||
const run = await prepare(path.join(root, "interrupted-checkpoint"), leaseId);
|
||||
await run.flush();
|
||||
@@ -331,7 +365,10 @@ describe("shared sandbox work-folder lifecycle", () => {
|
||||
expect(after!.checkpointKey).toBe(before!.checkpointKey);
|
||||
expect(await retainUnsavedWorkFolderLease(db, { id: leaseId, companyId })).toBe(true);
|
||||
} finally { fail.mockRestore(); }
|
||||
await run.stop(); active.splice(active.indexOf(run), 1);
|
||||
await expect(run.stop()).rejects.toThrow("Injected storage outage");
|
||||
active.splice(active.indexOf(run), 1);
|
||||
const retry = await prepare(run.home, randomUUID(), leaseId);
|
||||
await retry.stop(); active.splice(active.indexOf(retry), 1);
|
||||
await fs.rm(run.home, { recursive: true });
|
||||
const recovered = await prepare(path.join(root, "interrupted-recovered"), randomUUID());
|
||||
expect(await fs.readFile(path.join(recovered.primaryRepo, "new-unsaved-file"), "utf8")).toBe("must survive a failed save");
|
||||
|
||||
@@ -21,35 +21,46 @@ export async function findUnboundLegacyTaskWorkspace(db: Db, input: {
|
||||
}) {
|
||||
if (!input.issueId || !input.projectId || input.executionWorkspaceId || input.executionWorkspacePreference
|
||||
|| input.environment?.driver !== "sandbox" || input.environment.config.reuseLease !== true) return null;
|
||||
const [candidate] = await db.select({ lease: environmentLeases }).from(environmentLeases)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId),
|
||||
eq(heartbeatRuns.companyId, environmentLeases.companyId)))
|
||||
.innerJoin(executionWorkspaces, and(eq(executionWorkspaces.id, environmentLeases.executionWorkspaceId),
|
||||
eq(executionWorkspaces.companyId, environmentLeases.companyId)))
|
||||
.where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.issueId, input.issueId),
|
||||
eq(environmentLeases.environmentId, input.environment.id),
|
||||
eq(environmentLeases.leasePolicy, "reuse_by_environment"),
|
||||
inArray(environmentLeases.status, ["retained", "released"]),
|
||||
isNotNull(environmentLeases.providerLeaseId),
|
||||
sql`${environmentLeases.metadata}->>'driver' = 'sandbox'`,
|
||||
sql`${environmentLeases.metadata}->>'workFolderLayout' is distinct from 'scoped'`,
|
||||
sql`${environmentLeases.metadata}->'reusableSandboxLease'->>'version' = '1'`,
|
||||
eq(heartbeatRuns.agentId, input.agentId),
|
||||
input.responsibleUserId === null ? isNull(heartbeatRuns.responsibleUserId)
|
||||
: eq(heartbeatRuns.responsibleUserId, input.responsibleUserId),
|
||||
eq(executionWorkspaces.projectId, input.projectId), eq(executionWorkspaces.status, "active"),
|
||||
or(isNull(executionWorkspaces.sourceIssueId), eq(executionWorkspaces.sourceIssueId, input.issueId)),
|
||||
notExists(db.select({ id: workFolderRuns.runId }).from(workFolderRuns).where(and(
|
||||
eq(workFolderRuns.companyId, input.companyId), sql`${workFolderRuns.manifest}->>'taskId' = ${input.issueId}`)))))
|
||||
.orderBy(desc(environmentLeases.createdAt), desc(environmentLeases.id)).limit(1);
|
||||
if (!candidate) return null;
|
||||
const lease = await bindLegacySandboxIdentity(db, candidate.lease as EnvironmentLease);
|
||||
const scope = record(lease.metadata?.reusableSandboxLease);
|
||||
if (scope?.version !== 2 || scope.issueId !== input.issueId || scope.responsibleUserId !== input.responsibleUserId
|
||||
|| scope.adapterType !== input.adapterType || scope.environmentId !== input.environment.id
|
||||
|| scope.executionWorkspaceId !== lease.executionWorkspaceId) return null;
|
||||
// Normal workspace freshness and provider sentinel checks still run before use.
|
||||
return lease.executionWorkspaceId;
|
||||
let before: { createdAt: string; id: string } | undefined;
|
||||
while (true) {
|
||||
const candidates = await db.select({ lease: environmentLeases,
|
||||
createdAt: sql<string>`${environmentLeases.createdAt}::text` }).from(environmentLeases)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId),
|
||||
eq(heartbeatRuns.companyId, environmentLeases.companyId)))
|
||||
.innerJoin(executionWorkspaces, and(eq(executionWorkspaces.id, environmentLeases.executionWorkspaceId),
|
||||
eq(executionWorkspaces.companyId, environmentLeases.companyId)))
|
||||
.where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.issueId, input.issueId),
|
||||
before ? sql`(${environmentLeases.createdAt}, ${environmentLeases.id}) < (${before.createdAt}::timestamptz, ${before.id}::uuid)` : undefined,
|
||||
eq(environmentLeases.environmentId, input.environment.id),
|
||||
eq(environmentLeases.leasePolicy, "reuse_by_environment"),
|
||||
inArray(environmentLeases.status, ["retained", "released"]),
|
||||
isNotNull(environmentLeases.providerLeaseId),
|
||||
sql`${environmentLeases.metadata}->>'driver' = 'sandbox'`,
|
||||
sql`${environmentLeases.metadata}->>'workFolderLayout' is distinct from 'scoped'`,
|
||||
sql`${environmentLeases.metadata}->'reusableSandboxLease'->>'version' = '1'`,
|
||||
eq(heartbeatRuns.agentId, input.agentId),
|
||||
input.responsibleUserId === null ? isNull(heartbeatRuns.responsibleUserId)
|
||||
: eq(heartbeatRuns.responsibleUserId, input.responsibleUserId),
|
||||
eq(executionWorkspaces.projectId, input.projectId), eq(executionWorkspaces.status, "active"),
|
||||
or(isNull(executionWorkspaces.sourceIssueId), eq(executionWorkspaces.sourceIssueId, input.issueId)),
|
||||
notExists(db.select({ id: workFolderRuns.runId }).from(workFolderRuns).where(and(
|
||||
eq(workFolderRuns.companyId, input.companyId), sql`${workFolderRuns.manifest}->>'taskId' = ${input.issueId}`)))))
|
||||
.orderBy(desc(environmentLeases.createdAt), desc(environmentLeases.id)).limit(100);
|
||||
if (!candidates.length) return null;
|
||||
for (const candidate of candidates) {
|
||||
const lease = await bindLegacySandboxIdentity(db, candidate.lease as EnvironmentLease);
|
||||
const scope = record(lease.metadata?.reusableSandboxLease);
|
||||
if (scope?.version !== 2 || scope.issueId !== input.issueId || scope.responsibleUserId !== input.responsibleUserId
|
||||
|| scope.adapterType !== input.adapterType || scope.environmentId !== input.environment.id
|
||||
|| scope.executionWorkspaceId !== lease.executionWorkspaceId) continue;
|
||||
// Normal workspace freshness and provider sentinel checks still run before use.
|
||||
return lease.executionWorkspaceId;
|
||||
}
|
||||
// Invalid newer lease records must not hide an older recoverable workspace.
|
||||
// Keep Postgres microseconds intact across page boundaries.
|
||||
const last = candidates[candidates.length - 1]!;
|
||||
before = { createdAt: last.createdAt, id: last.lease.id };
|
||||
}
|
||||
}
|
||||
|
||||
/** Keep the old sync/restore contract even after its provider sandbox expires. */
|
||||
|
||||
@@ -321,25 +321,30 @@ export async function prepareSandboxWorkFolders(input: {
|
||||
},
|
||||
async onError() { await saveState("failed", "Files could not be saved; the sandbox must be retained for recovery"); },
|
||||
});
|
||||
return { manifest, home, identityChanged, primaryRepo: bindings.find(({ binding }) => manifest.repositories.some((repo) => repo.bindingId === binding.id && repo.primary))?.root ?? bindings[0]?.root ?? paths.task!,
|
||||
env: { HOME: home, AGENT_HOME: paths.agent!, PAPERCLIP_PRIMARY_REPO: bindings.find(({ binding }) => manifest.repositories.some((repo) => repo.bindingId === binding.id && repo.primary))?.root ?? bindings[0]?.root ?? paths.task!, PAPERCLIP_TASK_DIR: paths.task!, PAPERCLIP_AGENT_DIR: paths.agent!,
|
||||
PAPERCLIP_USER_DIR: paths.user!, PAPERCLIP_PROJECT_DIR: paths.project!, PAPERCLIP_REPOS_DIR: paths.repos! },
|
||||
flush: checkpointer.flush, stop: async (beforeCompletion?: () => Promise<void>) => {
|
||||
let completion: Promise<void> | null = null;
|
||||
function stop(beforeCompletion?: () => Promise<void>) {
|
||||
// Error teardown must observe the original outcome, including failures
|
||||
// after the data save. A later run owns any recovery of this working copy.
|
||||
completion ??= (async () => {
|
||||
await checkpointer.stop();
|
||||
const [run] = await db.select({ refreshRequested: workFolderRuns.refreshRequested }).from(workFolderRuns)
|
||||
.where(eq(workFolderRuns.runId, input.runId));
|
||||
if (run?.refreshRequested) {
|
||||
// The agent has stopped. The successful final flush above protects its
|
||||
// edits before accepting incoming shared files at this safe boundary.
|
||||
// The successful final flush protects edits before incoming refresh.
|
||||
await assertBindings();
|
||||
for (const scope of WORK_FOLDER_SCOPES) await incoming(scope);
|
||||
await db.update(workFolderRuns).set({ refreshRequested: false, baselines, updatedAt: new Date() })
|
||||
.where(eq(workFolderRuns.runId, input.runId));
|
||||
}
|
||||
// Native resume identity must be published after the data is durable,
|
||||
// but before completion can release a new turn onto this sandbox.
|
||||
// Publish resume identity after data is durable, before releasing a turn.
|
||||
await beforeCompletion?.();
|
||||
manifest.finalCheckpointAt = new Date().toISOString();
|
||||
await saveState("saved");
|
||||
} };
|
||||
})();
|
||||
return completion;
|
||||
}
|
||||
return { manifest, home, identityChanged, primaryRepo: bindings.find(({ binding }) => manifest.repositories.some((repo) => repo.bindingId === binding.id && repo.primary))?.root ?? bindings[0]?.root ?? paths.task!,
|
||||
env: { HOME: home, AGENT_HOME: paths.agent!, PAPERCLIP_PRIMARY_REPO: bindings.find(({ binding }) => manifest.repositories.some((repo) => repo.bindingId === binding.id && repo.primary))?.root ?? bindings[0]?.root ?? paths.task!, PAPERCLIP_TASK_DIR: paths.task!, PAPERCLIP_AGENT_DIR: paths.agent!,
|
||||
PAPERCLIP_USER_DIR: paths.user!, PAPERCLIP_PROJECT_DIR: paths.project!, PAPERCLIP_REPOS_DIR: paths.repos! },
|
||||
flush: checkpointer.flush, stop };
|
||||
}
|
||||
Reference in new issue
Block a user