mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 20:21:50 +01:00
fix(recovery): reconstruct missing native workspace obligations (#17877)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A failed native run can still own prepared files in a remote sandbox. > - Cleanup must preserve those files before another executor can start. > - Older failure producers can omit the preservation record. > - A saved message then waits for cleanup without a recovery owner. > - This change reconstructs that record during user continuation and orphan cleanup. > - The existing worker saves the files, verifies stop, and returns input to normal admission. ## Linked Issues or Issue Description **What happened?** A failed native run retained a prepared workspace and an active sandbox lease. It had no file-preservation obligation. Cleanup protected the files. The next user message waited for a stop receipt that no worker owned. **Expected behavior** A new message registers the missing maintenance work at once. Existing orphan cleanup recovers historical saved input. Ordinary Retry remains available. Saved input can start once, after verified preservation and containment. **Steps to reproduce** 1. Persist a failed native run with an exact prepared remote workspace and its active lease. 2. Leave its finalizer terminal and unowned. Omit the workspace recovery record. 3. Send a user comment or run orphan cleanup. 4. Observe that no preservation worker starts and the comment remains deferred. **Paperclip version or commit** Reproduced on `bd382eee536f4ada680072b1cc19100163c3d04a`. **Deployment mode** Native Runner with a remote sandbox. Local verification uses disposable PostgreSQL 18 and real temporary workspace files. Refs #17788 and #17800. Those fixes preserve future deadline failures and register historical preservation on Retry. This change adds automatic recovery for new and saved messages. Refs #17839 for the separate restart-failure producer repair. ## What Changed - Register missing preservation before an authenticated continuation waits for remote cleanup. - Reconstruct the same exact-source obligation during the existing orphan lease sweep. - Notify the existing preservation worker in the registration transaction. Do not add a timer or history scan. - Test ephemeral and reusable sources, live owners, active executors, mismatched sources, rollback, lock contention, Retry and saved-input delivery. - Document the recovery contract and feature-map behavior. ## Verification - Red: both new orphan-recovery cases fail on unchanged production code. Four ownership and source negative controls pass. - Red: a new comment starts no preservation work without the admission repair. - All 461 test cases across five affected suites pass across the local runs. One existing 30-second timing case timed out under concurrent verification; its isolated rerun passed with the same deadline and assertions. - The three final-head entry journeys pass again: HTTP Retry, HTTP comment creation, and historical saved-input recovery. - `pnpm -r typecheck` and `pnpm build` pass on final head `8fab8b2ba9049f640fd884094a2007a69465e6da`. The affected server typecheck and build also pass. - The integration journey holds the provider stop reply, checks that no successor starts, preserves real files and newer host edits, and checks one saved-input successor. - Module-boundary checks, map checks and `git diff --check` pass. - Native macOS PostgreSQL cannot start because host semaphore capacity is exhausted. Database verification uses separate disposable PostgreSQL databases for each suite. An earlier shared-database attempt caused fixture interference and is not counted as a pass. - Greptile gives 5/5 with no findings or unresolved threads on `8fab8b2ba9049f640fd884094a2007a69465e6da`. All final-head GitHub checks are settled: 55 passed, two Storybook checks skipped, and one neutral Paperclip review. Full CI test, build, typecheck, Runner and browser lanes pass. No full local `pnpm test:run` pass is claimed. The fix is not deployed. This PR makes no claim of a live hosted model turn or recovery of the original incident. ## Risks - Registration grants maintenance authority only. Existing coordinator, descriptor, allocation, stop, file-receipt and execution gates remain effective. - A locked source remains held. Its one-second lock timeout does not prevent an unrelated source from registering recovery. - New preservation writes wake existing work. Settled sources produce no new recurring work. Historical discovery uses the existing orphan cleanup lifecycle and its backoff. - Provider lifecycle is simulated in local integration. PostgreSQL, HTTP routing, file export/merge, the worker and queue admission are real. - No schema, credential, model, trust, retry-budget or assignment change. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository inspection, tool use and code execution. The runtime does not expose the exact model ID or context-window size. ## 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>
This commit is contained in:
1 parent
b8f6d9b23f
commit
663fe667d6
9 files changed
+213
-25
No files matched your search
@@ -1402,6 +1402,15 @@ replay: the provider stop receipt, saved-file receipt, current execution authori
|
||||
and effect fences still control successor admission. Repeated clicks retain the
|
||||
same recovery generation and request.
|
||||
|
||||
An authenticated new message also registers that missing obligation before it
|
||||
waits for cleanup. Orphan cleanup reconstructs it for a failed native run that
|
||||
still owns an exact prepared remote workspace, including historical saved input.
|
||||
Both paths wake the existing preservation worker at commit instead of waiting
|
||||
indefinitely for an export that has no owner. A live executor or finalizer, accepted result, or mismatched source
|
||||
remains protected. Cleanup never interprets this registration as a stop receipt
|
||||
or permission to replay provider work. Repeated cleanup visits register the source
|
||||
once; settled sources add no continuing discovery or new maintenance timer.
|
||||
|
||||
Every continuation carries the triggering request, ordered user direction, interaction outcomes, completed work, and explicit history coverage. A delivered message remains part of the task's request after its connection or approval resolves. The original title is background; a completed Notion read does not satisfy a later Gmail request. Author and source-trust boundaries survive rendering into both native and legacy prompts. Missing required history must be fetched before dispatch rather than described as complete.
|
||||
|
||||
### Remote runner process evidence
|
||||
|
||||
@@ -40,6 +40,9 @@ Agent access: [companion](../agent-access-map/recovery.md).
|
||||
for the same stop-and-save workflow before Retry admits another executor.
|
||||
Retry reconstructs missing preservation for historical failures only from an
|
||||
exact prepared source without a live finalization owner.
|
||||
New messages and orphan cleanup can register missing preservation automatically
|
||||
and wake the existing worker. Saved input still waits for verified stop and file
|
||||
persistence, then enters ordinary admission once.
|
||||
- `unknown-outcome`: `recovery/unknown-outcome` — uncertain external side effects require inspection and
|
||||
an explicit decision, not automatic replay.
|
||||
- `user-continuation`: `recovery/user-continuation` — saved user messages after a stop are reconsidered once
|
||||
|
||||
@@ -16,11 +16,12 @@ import { join } from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { beforeAll, afterAll, describe, it, expect, vi } from "vitest";
|
||||
import {
|
||||
approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections,
|
||||
applyPendingMigrations, approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections,
|
||||
agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions,
|
||||
issues, nativeRunFinalizations, nativeRunResults, completionContracts, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers,
|
||||
} from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
|
||||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
import { adapterExecutionControls, createAdapterExecutionControl } from "./adapter-execution-control.js";
|
||||
import { CONVERSATION_CONTINUATION_POLICY } from "./conversation-continuation.js";
|
||||
@@ -29,12 +30,19 @@ import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnv
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { createDurableChatWakeupRequest } from "./durable-chat-wakeup.js";
|
||||
import { recordNativeFinalizationFailure } from "./native-runtime/native-run-finalizer.js";
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_CONTINUATION_TEST_DATABASE_URL");
|
||||
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("explicit native conversation continuation", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("explicit-native-message-"); db = createDb(database.connectionString); }, 90_000);
|
||||
afterAll(async () => { await database?.cleanup(); });
|
||||
beforeAll(async () => {
|
||||
if (externalUrl) {
|
||||
await applyPendingMigrations(externalUrl);
|
||||
database = { connectionString: externalUrl, cleanup: async () => {} };
|
||||
} else database = await startEmbeddedPostgresTestDatabase("explicit-native-message-");
|
||||
db = createDb(database.connectionString);
|
||||
}, 90_000);
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
|
||||
async function seed() {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
|
||||
const sourceRunId = randomUUID(), successorRunId = randomUUID();
|
||||
|
||||
@@ -4,6 +4,7 @@ import { readTaskRetryIntent } from "./task-retry.js";
|
||||
import { readSystemRecovery } from "./system-recovery-state.js";
|
||||
import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js";
|
||||
import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js";
|
||||
import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
|
||||
import { canContinueCancelledRun } from "./run-cancellation.js";
|
||||
@@ -298,6 +299,10 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
run.status === "cancelled" && !coordinator.leaseOwner
|
||||
? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details."
|
||||
: "Waiting for the previous run to finish recovery. Your message will start automatically.");
|
||||
const preserved = hasRequiredWorkspaceRecovery(run.resultJson) ? null
|
||||
: await preserveNativeFailedWorkspace(db, run);
|
||||
if (preserved && hasRequiredWorkspaceRecovery(preserved.resultJson)) return blocked("workspace_repair_required",
|
||||
"Paperclip is recovering saved work before continuing. Your message is saved.");
|
||||
const leases = await db.select()
|
||||
.from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environments, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { activityLog, agents, applyPendingMigrations, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environments, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
@@ -15,6 +15,9 @@ import { claimNativeRestartRecoveries } from "../native-runtime/native-restart-r
|
||||
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
|
||||
import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
|
||||
import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
|
||||
@@ -169,7 +172,8 @@ describe("heartbeat recovery module callbacks", () => {
|
||||
});
|
||||
});
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_TEST_DATABASE_URL");
|
||||
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const describePostgres = support.supported ? describe : describe.skip;
|
||||
if (!support.supported) console.warn(`Skipping recovery module database tests: ${support.reason}`);
|
||||
|
||||
@@ -177,10 +181,13 @@ describePostgres("heartbeat recovery module ownership", () => {
|
||||
let db: Db;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-");
|
||||
if (externalUrl) {
|
||||
await applyPendingMigrations(externalUrl);
|
||||
database = { connectionString: externalUrl, cleanup: async () => {} };
|
||||
} else database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-");
|
||||
db = createDb(database.connectionString);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
afterAll(async () => { await database?.cleanup(); });
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
|
||||
|
||||
async function seedRun() {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), issueId = randomUUID();
|
||||
@@ -190,6 +197,68 @@ describePostgres("heartbeat recovery module ownership", () => {
|
||||
return (await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, invocationSource: "automation", status: "running", runtimeMode: "legacy" }).returning())[0];
|
||||
}
|
||||
|
||||
it.each(["ephemeral", "reuse_by_environment", "live_owner", "live_executor", "mismatched_source", "finalizing"] as const)(
|
||||
"reconstructs missing native preservation during orphan cleanup without replay: %s", async state => {
|
||||
const run = await seedRun();
|
||||
const [environment] = await db.insert(environments).values({ name: run.id, driver: "sandbox" }).returning();
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, issueId: run.issueId,
|
||||
heartbeatRunId: run.id, environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(),
|
||||
status: "active", leasePolicy: state === "reuse_by_environment" ? state : "ephemeral",
|
||||
metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: randomUUID(), remoteCwd: "/work" },
|
||||
}).returning();
|
||||
await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", nativeIssueId: run.issueId,
|
||||
finishedAt: new Date(), errorCode: "execution_finalization_deadline_exceeded", contextSnapshot: { issueId: run.issueId },
|
||||
runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared",
|
||||
descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null,
|
||||
workspaceId: run.id, leaseId: lease.id, providerLeaseId: lease.providerLeaseId,
|
||||
remoteCwd: state === "mismatched_source" ? "/other" : "/work", resourceDisposition: "keep_running" } },
|
||||
}).where(eq(heartbeatRuns.id, run.id));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: run.companyId, issueId: run.issueId!, runId: run.id,
|
||||
phase: state === "finalizing" ? "workspace_finalizing" : "terminal_failure",
|
||||
leaseOwner: state === "live_owner" ? "another-controller" : null });
|
||||
const deps = callbacks(db);
|
||||
if (state === "live_executor") deps.activeRunExecutions.add(run.id);
|
||||
const recovery = createHeartbeatRecovery(db, deps);
|
||||
const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.workspaceRecovery, wake);
|
||||
const shouldRecover = state === "ephemeral" || state === "reuse_by_environment";
|
||||
try {
|
||||
expect(await recovery.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: shouldRecover ? 1 : 0 });
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
const [retained] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id));
|
||||
const records = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!));
|
||||
expect(source.status).toBe("failed");
|
||||
if (shouldRecover) {
|
||||
expect(source.resultJson).toMatchObject({ workspaceRestoreRecovery: { leaseIds: [lease.id] } });
|
||||
expect(retained).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure",
|
||||
metadata: { sandboxStopAndRetain: { leaseId: lease.id, runId: run.id } } });
|
||||
expect(records).toHaveLength(1);
|
||||
expect(records[0].evidence).toMatchObject({ automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } });
|
||||
expect(wake).toHaveBeenCalledOnce();
|
||||
const registered = source.resultJson;
|
||||
expect(await recovery.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: 0 });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].resultJson).toEqual(registered);
|
||||
expect(wake).toHaveBeenCalledOnce();
|
||||
} else {
|
||||
expect(source.resultJson).toBeNull();
|
||||
expect(retained.status).toBe("active");
|
||||
expect(records).toHaveLength(0);
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
}
|
||||
expect(hasRemoteTerminationReceipt(retained)).toBe(false);
|
||||
expect(retained.metadata).not.toHaveProperty("remoteExecutionTermination");
|
||||
expect(deps.executeRun).not.toHaveBeenCalled();
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, run.companyId))).toHaveLength(1);
|
||||
} finally {
|
||||
unsubscribe();
|
||||
await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id));
|
||||
await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!));
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
await db.delete(activityLog).where(eq(activityLog.runId, run.id));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["restored", "restore_failed"])("keeps the sandbox available until adapter workspace settlement: %s", async outcome => {
|
||||
const run = await seedRun();
|
||||
const deps = callbacks(db);
|
||||
|
||||
@@ -3,7 +3,7 @@ import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js
|
||||
import type { WorkspaceWorkers } from "../plugin-workspaces.js";
|
||||
import { createHeartbeatAgentStatus } from "./agent-status.js";
|
||||
import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js";
|
||||
import { preserveCancelledNativeWorkspaces } from "../native-runtime/native-failed-workspace-recovery.js";
|
||||
import { preserveCancelledNativeWorkspaces, preserveOrphanedNativeWorkspace } from "../native-runtime/native-failed-workspace-recovery.js";
|
||||
import { nativeWorkspaceExportPendingForLease, pendingNativeWorkspaceFinalizationCondition } from "../native-runtime/native-workspace-finalization-state.js";
|
||||
import { CHAT_COMPLETION_WAKE_REASON } from "../chat-completion-delivery.js";
|
||||
import {
|
||||
@@ -1486,6 +1486,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
let recovered = 0;
|
||||
for (const { lease } of rows) {
|
||||
if (await nativeWorkspaceExportPendingForLease(db, lease)) {
|
||||
if (lease.heartbeatRunId && !activeRunExecutions.has(lease.heartbeatRunId)
|
||||
&& !adapterExecutionControls.has(lease.heartbeatRunId)
|
||||
&& await preserveOrphanedNativeWorkspace(db, lease)) recovered += 1;
|
||||
await deferOrphanedActiveLease(lease.id);
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro
|
||||
import { getExecutionBlocker } from "../execution-blocker.js";
|
||||
import { markExecutionReconciliation, settleUnrecoverableExecutions, validateExecutionReconciliation } from "../execution-recovery-resolution.js";
|
||||
import { prepareSandboxStopAndRetain, settleStopOnlyCleanup } from "../sandbox-stop-and-retain.js";
|
||||
import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js";
|
||||
import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, preserveOrphanedNativeWorkspace } from "./native-failed-workspace-recovery.js";
|
||||
import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js";
|
||||
import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js";
|
||||
import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js";
|
||||
@@ -51,6 +51,40 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
const readLease = async (id: string) => (await db.select().from(environmentLeases).where(eq(environmentLeases.id, id)))[0]!;
|
||||
const actions = async (issueId: string) => db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
|
||||
it("keeps another orphan recoverable while a task lock delays preservation", async () => {
|
||||
const busy = await seed(), free = await seed();
|
||||
let unlock!: () => void, ready!: () => void;
|
||||
const released = new Promise<void>(resolve => { unlock = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { ready = resolve; });
|
||||
const holding = db.transaction(async tx => {
|
||||
await tx.select().from(issues).where(eq(issues.id, busy.issueId)).for("update");
|
||||
ready(); await released;
|
||||
});
|
||||
await acquired;
|
||||
try {
|
||||
expect(await preserveOrphanedNativeWorkspace(db, busy.lease)).toBe(false);
|
||||
expect(await preserveOrphanedNativeWorkspace(db, free.lease)).toBe(true);
|
||||
expect((await readLease(busy.lease.id)).status).toBe("active");
|
||||
expect(await readLease(free.lease.id)).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" });
|
||||
expect(await actions(busy.issueId)).toHaveLength(0);
|
||||
} finally { unlock(); await holding; }
|
||||
expect(await preserveOrphanedNativeWorkspace(db, busy.lease)).toBe(true);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, busy.companyId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("rolls back orphan registration with its enclosing transaction", async () => {
|
||||
const f = await seed();
|
||||
await expect(db.transaction(async tx => {
|
||||
expect(await preserveOrphanedNativeWorkspace(tx as unknown as typeof db, f.lease)).toBe(true);
|
||||
throw new Error("rollback preservation");
|
||||
})).rejects.toThrow("rollback preservation");
|
||||
expect(await readRun(f.runId)).toEqual(f.run);
|
||||
expect(await readLease(f.lease.id)).toEqual(f.lease);
|
||||
expect(await actions(f.issueId)).toHaveLength(0);
|
||||
expect(await preserveOrphanedNativeWorkspace(db, f.lease)).toBe(true);
|
||||
expect(await actions(f.issueId)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(["settled", "historical", "live_owner", "retry", "nonterminal", "wrong_wake", "wrong_hold_run", "no_hold"])("retires only an exact superseded native resume obligation: %s", async scenario => {
|
||||
const f = await seed(), actionId = randomUUID();
|
||||
await db.insert(issueRecoveryActions).values({ id: actionId, companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
|
||||
@@ -9,6 +9,7 @@ import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js";
|
||||
import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js";
|
||||
import { hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
||||
|
||||
export async function preserveNativeFailedWorkspace(db: Db, selected: typeof heartbeatRuns.$inferSelect, options: { allowPreterminal?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) {
|
||||
if (readNativeWorkspaceSyncReference(selected.runnerProfileJson?.nativeWorkspaceSync)?.state !== "prepared") return;
|
||||
@@ -53,6 +54,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea
|
||||
// Their Runner descriptor is not evidence that user files need exporting.
|
||||
const retainFiles = !nativeRunHasNoWorkspaceAccess(run);
|
||||
const alreadyRetained = hasRequiredWorkspaceRecovery(run.resultJson);
|
||||
if (retainFiles && !alreadyRetained) await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
||||
let retained = !retainFiles || alreadyRetained ? run : await recordRequiredWorkspaceRestoreFailure(tx as unknown as Db, run, { workspaceRestoreFailure: "restore_failed" }, source);
|
||||
if (!retained || retainFiles && !hasRequiredWorkspaceRecovery(retained.resultJson)) return;
|
||||
if (options.allowPreterminal && coordinator.phase === "observed") {
|
||||
@@ -81,6 +83,36 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea
|
||||
});
|
||||
}
|
||||
|
||||
export async function preserveOrphanedNativeWorkspace(db: Db, lease: Pick<typeof environmentLeases.$inferSelect,
|
||||
"companyId" | "heartbeatRunId">): Promise<boolean> {
|
||||
if (!lease.heartbeatRunId) return false;
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, lease.companyId), eq(heartbeatRuns.id, lease.heartbeatRunId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "failed"),
|
||||
)).limit(1);
|
||||
if (!run || hasRequiredWorkspaceRecovery(run.resultJson)) return false;
|
||||
try {
|
||||
const retained = await preserveNativeFailedWorkspace(db, run);
|
||||
return Boolean(retained && hasRequiredWorkspaceRecovery(retained.resultJson));
|
||||
} catch (error) {
|
||||
if (error instanceof NativeWorkspaceFinalizationBusyError) return false;
|
||||
const code = retentionContentionCode(error);
|
||||
if (!code) throw error;
|
||||
logger.warn({ runId: run.id, code }, "Native workspace retention is waiting for its current owner");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function retentionContentionCode(error: unknown): string | null {
|
||||
let current = error;
|
||||
for (let depth = 0; depth < 4 && current && typeof current === "object"; depth++) {
|
||||
const value = current as { code?: unknown; cause?: unknown };
|
||||
if (value.code === "55P03" || value.code === "57014") return value.code;
|
||||
current = value.cause;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
const cancelledWorkspaceCursors = new WeakMap<Db, string>();
|
||||
|
||||
/** Older controllers could acknowledge Stop without pinning prepared files.
|
||||
@@ -113,13 +145,7 @@ export async function preserveCancelledNativeWorkspaces(db: Db) {
|
||||
try { await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true }); }
|
||||
catch (error) {
|
||||
if (error instanceof NativeWorkspaceFinalizationBusyError) continue;
|
||||
let current: unknown = error;
|
||||
let code: string | null = null;
|
||||
for (let depth = 0; depth < 4 && current && typeof current === "object"; depth++) {
|
||||
const value = current as { code?: unknown; cause?: unknown };
|
||||
if (value.code === "55P03" || value.code === "57014") { code = value.code; break; }
|
||||
current = value.cause;
|
||||
}
|
||||
const code = retentionContentionCode(error);
|
||||
if (!code) throw error;
|
||||
logger.warn({ runId: run.id, code }, "Cancelled native workspace retention is waiting for its current owner");
|
||||
}
|
||||
|
||||
@@ -137,7 +137,7 @@ describe("automatic retained workspace recovery", () => {
|
||||
const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!;
|
||||
const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!;
|
||||
|
||||
it("repairs an unregistered historical deadline failure through HTTP Retry without admitting work before verified stop", async () => {
|
||||
it.each(["HTTP Retry", "new comment", "automatic cleanup"] as const)("repairs an unregistered historical deadline failure through %s without admitting work before verified stop", async entry => {
|
||||
const f = await seed("native", true, "keep_running");
|
||||
const source = await run(f.runId);
|
||||
await db.update(heartbeatRuns).set({ resultJson: null, errorCode: "execution_finalization_deadline_exceeded" })
|
||||
@@ -155,7 +155,7 @@ describe("automatic retained workspace recovery", () => {
|
||||
await stopped;
|
||||
return { providerLeaseId: f.providerLeaseId, state: "stopped" };
|
||||
});
|
||||
const errors = vi.fn(), heartbeat = heartbeatService(db);
|
||||
const errors = vi.fn(), heartbeat = heartbeatService(db, { environmentRuntime: f.environmentRuntime });
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors });
|
||||
const worker = registerWorkspaceRecoveryWork(db, coordinator, f.environmentRuntime, heartbeat.resumeRemoteStopComments);
|
||||
const successors = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
|
||||
@@ -166,11 +166,39 @@ describe("automatic retained workspace recovery", () => {
|
||||
app.use((req, _res, next) => { req.actor = { type: "board", source: "local_implicit", userId: "board" }; next(); });
|
||||
app.use("/api", issueRoutes(db, {} as Parameters<typeof issueRoutes>[1]));
|
||||
app.use(errorHandler);
|
||||
const body = { idempotencyKey: randomUUID(), observedRunId: f.runId };
|
||||
const retry = await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202);
|
||||
expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202)).body.requestId).toBe(retry.body.requestId);
|
||||
let requestId: string | undefined;
|
||||
let commentId: string | undefined;
|
||||
if (entry === "HTTP Retry") {
|
||||
const body = { idempotencyKey: randomUUID(), observedRunId: f.runId };
|
||||
const retry = await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202);
|
||||
requestId = retry.body.requestId;
|
||||
expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202)).body.requestId).toBe(requestId);
|
||||
} else if (entry === "new comment") {
|
||||
const response = await request(app).post(`/api/issues/${f.issueId}/comments`)
|
||||
.send({ body: "Is this fixed?" }).expect(201);
|
||||
commentId = response.body.id;
|
||||
expect(await successors()).toHaveLength(0);
|
||||
} else {
|
||||
const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
authorType: "user", authorUserId: "board", body: "Is this fixed?" }).returning();
|
||||
commentId = comment.id;
|
||||
const wake = { source: "automation" as const, triggerDetail: "system" as const,
|
||||
reason: "issue_reopened_via_comment", requestedByActorType: "user" as const, requestedByActorId: "board",
|
||||
payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" },
|
||||
contextSnapshot: { source: "issue.comment.reopen", issueId: f.issueId, wakeCommentId: comment.id },
|
||||
};
|
||||
await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: source.agentId,
|
||||
source: wake.source, triggerDetail: wake.triggerDetail, reason: wake.reason,
|
||||
requestedByActorType: "user", requestedByActorId: "board", status: "deferred_issue_execution",
|
||||
payload: { ...wake.payload, _paperclipWakeContext: { ...wake.contextSnapshot, wakeReason: wake.reason,
|
||||
wakeSource: wake.source, wakeTriggerDetail: wake.triggerDetail, wakeCommentIds: [comment.id] } },
|
||||
});
|
||||
expect(await successors()).toHaveLength(0);
|
||||
if (entry === "automatic cleanup") await heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0 });
|
||||
}
|
||||
await expect.poll(() => f.runtime.retryPendingSandboxTeardown.mock.calls.length, { timeout: 20_000 }).toBe(1);
|
||||
await heartbeat.resumeTaskRetry(f.companyId, retry.body.requestId);
|
||||
if (requestId) await heartbeat.resumeTaskRetry(f.companyId, requestId);
|
||||
else await heartbeat.resumeQueuedRuns();
|
||||
expect(await successors()).toHaveLength(0);
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false);
|
||||
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].metadata)
|
||||
@@ -179,11 +207,14 @@ describe("automatic retained workspace recovery", () => {
|
||||
releaseStop();
|
||||
await expect.poll(async () => (await successors()).length, { timeout: 20_000 }).toBe(1);
|
||||
const [successor] = await successors();
|
||||
if (commentId) expect(successor.contextSnapshot?.wakeCommentIds).toContain(commentId);
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true);
|
||||
expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n");
|
||||
expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ mode: "resume_preserved_workspace" });
|
||||
await Promise.all([heartbeat.resumeTaskRetry(f.companyId, retry.body.requestId), heartbeat.resumeRemoteStopComments(await run(f.runId))]);
|
||||
if (requestId) expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ mode: "resume_preserved_workspace" });
|
||||
else expect(successor.contextSnapshot?.previousRunId).toBe(f.runId);
|
||||
await Promise.all([requestId ? heartbeat.resumeTaskRetry(f.companyId, requestId) : heartbeat.resumeQueuedRuns(),
|
||||
heartbeat.resumeRemoteStopComments(await run(f.runId))]);
|
||||
expect(await successors()).toHaveLength(1);
|
||||
expect(f.runtime.resumeRunLease).toHaveBeenCalledOnce();
|
||||
expect((await run(f.runId)).status).toBe("failed");
|
||||
|
||||
Reference in new issue
Block a user