mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 11:01:16 +01:00
fix(recovery): preserve the bound remote writer before continuation (#18803)
## Thinking Path > - Paperclip manages agent work and preserves task files. > - A remote workspace records the exact run that owns its unsaved files. > - A later setup failure can become the latest task run without owning those files. > - Retry only inspected that latest run, so the original writer had no file-recovery owner. > - A released reusable source also failed the legacy retention rule. > - This change registers the bound writer, saves its files, and then admits the saved user request once. > - Existing stop, ownership, approval and execution-evidence checks remain effective. ## Linked Issues or Issue Description **What happened?** A native run prepared a remote workspace and failed before finalization. Its sandbox stopped with a dedicated stop receipt. The workspace stayed dirty and retained that run as its writer. A later setup failure became the latest task run. Retry inspected the newer run and left the original file obligation unregistered. New attempts failed at the writer fence. Registration alone was insufficient. After file recovery, continuation still required the Retry source to be one of the older recovery sources. A newer pre-provider failure did not satisfy that check. Workspace selection could also choose a fresh directory instead of the recovered files. **Expected behavior** Find the writer through the task's current workspace binding. Register only its exact prepared source. Preserve the released lease and its stop receipts. Save the files before continuation. Independently prove that a newer setup failure never dispatched work. Select the verified saved workspace and admit the Retry once. **Steps to reproduce** 1. Fail a native run with a prepared remote workspace and a released reusable sandbox. 2. Keep matching dedicated stop receipts and the workspace's dirty writer authority. 3. Record a newer pre-provider setup failure for the same task. 4. Request Retry. Before the fix, no file-maintenance record owns the actual writer. **Paperclip version or commit** Reproduced before production edits on master `f7ad1dc7ca`. The branch was rebased on master `adb49a44346849e751548b2fb640693dbfad4b6f`. **Deployment mode** Authenticated hosted execution with Paperclip Runner and remote workspace storage. Companion: #18798 (now merged upstream) fixes a pre-existing inbox test synchronization race. Its test-only commit `110aa88489cad926a451c972ad731ede469347c1` is cherry-picked here with attribution so this PR can run the complete CI matrix. Related: #17800 registers missing preservation for the selected source run. #18678 holds unfinished remote exports. #18718 repairs a prior warm-controller projection failure. #17916 addresses a separate legacy warm-copyback policy case. ## What Changed - Discover the failed writer through the company-scoped task/workspace binding during Retry and authenticated message admission. - Reuse the exact prepared-source validator for released remote workspaces. Registration changes neither the lease policy nor its stop receipts. Physical recovery retains its full allocation checks. - Check a newer failed preparation separately before allowing an older, repaired source to release continuation. Provider events, leases, results and execution ownership remain disqualifying evidence. - Select the bound saved workspace through the continuation's verified repair receipt. - Add PostgreSQL safety cases and an HTTP Retry workflow with real file capture and saved-copy reads. Document the execution contract. - Reuse #18798's completion barrier for the existing inbox cancellation test; preserve its eight-slot limit, rejection and coalescing assertions. ## Verification - RED before production edits: the exact stopped writer has no `workspaceRestoreRecovery` after Retry. Six unsafe-source controls pass on the baseline. - GREEN: the expanded Retry/message/preview selection passes 17 cases. All 627 tests in the five affected recovery suites pass on recovery commit `b572304f772d236b8a8bdf095e13e651d808f41c`, including Stop, ownership, saved messages and duplicate Retry. - The HTTP workflow registers maintenance, keeps execution held before save, reads back the saved file bytes, and admits exactly one continuation with `resume_preserved_workspace`. A registration-only repair first failed after save because it did not reconcile the newer Retry source. The incident-shaped pair passes with an independent active stranded-task action present. - Greptile identified the path where the newer setup failure already owns a reconciliation hold. Its new regression failed before the correction (saved files, zero queued continuations); the contradictory-dispatch control passed. All four HTTP variants pass after the correction, covering both hold shapes and both dispatch-evidence controls. - Repository typecheck and build passed. Server typecheck and build also passed after the recovery changes. Module-boundary and whitespace checks pass. - The full local Vitest command failed in the GitHub fast-response acknowledgement test (`delivery_unknown`: one response write instead of zero). That isolated case passes on this branch and unchanged master; the original full-run cause remains unestablished. A sibling run passed 199 test cases but timed out in workspace-suite database cleanup. Unchanged master also passed all 126 workspace cases and timed out in the same cleanup hook; all five acknowledgement outcomes passed in that baseline comparison. These local failures remain recorded rather than being reported as suite passes. - The first hosted matrix failed the pre-existing inbox cancellation assertion (expected nine computes, saw eight). The same assertion failed in [master CI](https://github.com/paperclipai/paperclip/actions/runs/41664339150/job/126015526840) and reproduced locally before applying #18798's existing repair. All six inbox HTTP/database cases pass after that repair. Final head `7f2fafcbace74cb1ad8ea68c8375a69b42ed30d9` adds only that attributed test change to the verified recovery code. - [Final hosted CI](https://github.com/paperclipai/paperclip/actions/runs/41677310377) passes on head `7f2fafcbace74cb1ad8ea68c8375a69b42ed30d9`, including `ci / verify`, `ci / e2e`, typecheck, build, every server/chat/workspace group, Runner checks and all eight browser shards. There are 54 successful check results, two informational neutral results and three skipped results, with no pending or failing results. - Greptile and the Paperclip reviewer both report 5/5 for that final head; no review threads remain open. The Paperclip review check is informational (`neutral`) and records its local-testing limitations. - The file workflow uses real PostgreSQL, subprocess file capture, encrypted saved objects and production recovery code. Provider resume/stop calls are simulated at the environment driver boundary. This does not prove a live Daytona or Codex recovery. No deployment or original-task retry is claimed. ## Risks - File-maintenance discovery must not authorize another executor. Existing writer, coordinator and physical allocation checks still own that decision. - A false bootstrap summary cannot prove that a newer run never executed. The new path also requires its immutable native-adapter claim, unresolved preparation state, absent coordinator/process/session/lease/result evidence, and no dispatch events. - Additional database reads run only during explicit task admission or Retry strategy selection. They are company/task scoped and add no polling loop. The saved-writer query returns one matching snapshot and omits prompt bodies. - No schema, credential, provider profile or deployment setting changes. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository inspection, code execution and test tools. The runtime does not expose the exact serving 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:
DottaandPaperclip authored and GitHub committed
1 parent
1c9ca5829f
commit
367f29478d
12 files changed
+342
-84
No files matched your search
@@ -1683,6 +1683,19 @@ 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.
|
||||
|
||||
Retry and authenticated new messages also inspect the task's bound remote
|
||||
workspace for its exact failed writer. A newer setup failure cannot hide that
|
||||
writer's file obligation. A released reusable source can register preservation
|
||||
when its prepared descriptor, current workspace authority, and dedicated stop
|
||||
receipts agree. Registration keeps the lease's policy, release state, and stop
|
||||
receipts. The existing worker verifies allocation ownership before file-only
|
||||
resume. It saves the files before ordinary continuation admission; repeated
|
||||
requests do not create another recovery generation or replay provider work.
|
||||
If Retry names a newer failed preparation, continuation separately verifies that
|
||||
it never reached provider dispatch. Contradictory execution evidence keeps the
|
||||
hold. After preservation, the accepted continuation selects the bound saved
|
||||
workspace through its verified repair receipt instead of discarding its files.
|
||||
|
||||
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
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js";
|
||||
import { and, eq, inArray, isNotNull, or } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { claimedAdapterType } from "./conversation-continuation.js";
|
||||
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
@@ -15,6 +15,25 @@ type Run = Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"controllerBootId" | "controllerLeaseExpiresAt">;
|
||||
type Coordinator = typeof nativeRunFinalizations.$inferSelect;
|
||||
|
||||
export async function isFailedNativePreparation(db: Db, run: Run, coordinator: Coordinator | undefined) {
|
||||
const recovery = run.resultJson?.executionRecovery as Record<string, unknown> | undefined;
|
||||
if (run.status !== "failed" || !run.finishedAt || run.runtimeMode !== "legacy" || run.runtimeModeResolvedAt
|
||||
|| run.errorCode !== "setup_failed" || run.executionStage !== "preparing"
|
||||
|| claimedAdapterType(run) !== "paperclip_runner" || coordinator
|
||||
|| recovery?.kind !== "bootstrap" || recovery.providerWorkStarted !== false
|
||||
|| run.processPid || run.processGroupId || run.processStartedAt || run.sessionIdAfter
|
||||
|| run.nativeIssueId || run.nativeSessionId || run.runnerProfileJson?.nativeExecutionInput != null
|
||||
|| run.runnerProfileJson?.nativeWorkspaceSync != null || run.runnerProfileJson?.remoteWorkspace != null
|
||||
|| run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return false;
|
||||
const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
)).limit(1);
|
||||
const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and(
|
||||
eq(nativeRunResults.companyId, run.companyId), eq(nativeRunResults.runId, run.id),
|
||||
)).limit(1);
|
||||
return !lease && !result && !await nativeStartupHasExecutionEvidence(db, run);
|
||||
}
|
||||
|
||||
/** Candidate for explicit user Retry only, never automatic replay after Stop. */
|
||||
export async function canRetryStoppedRun(db: Db, run: Run): Promise<boolean> {
|
||||
if (run.errorCode === "native_session_cleanup_quarantined") return false;
|
||||
@@ -66,6 +85,10 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co
|
||||
: !hasRemoteTerminationReceipt(lease))) return false;
|
||||
// Reject contradictory retained evidence, including a crash after a launch
|
||||
// request but before the PID callback. Provider events never certify a stop.
|
||||
return !await nativeStartupHasExecutionEvidence(db, run);
|
||||
}
|
||||
|
||||
async function nativeStartupHasExecutionEvidence(db: Db, run: Run) {
|
||||
const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
or(isNotNull(heartbeatRunEvents.sourceEventId),
|
||||
@@ -73,5 +96,5 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co
|
||||
"harness.ready", "session.started", "session.resumed", "session.updated", "turn.started",
|
||||
"provider.event", "provider.rpc_result", "tool.execution.started"])),
|
||||
)).limit(1);
|
||||
return !execution;
|
||||
return Boolean(execution);
|
||||
}
|
||||
@@ -6,12 +6,12 @@ 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 { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } 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";
|
||||
import { readQueuedInteractionResponse } from "./queued-interaction-response.js";
|
||||
import { isCancelledNativeStartup } from "./cancelled-native-startup.js";
|
||||
import { isCancelledNativeStartup, isFailedNativePreparation } from "./cancelled-native-startup.js";
|
||||
import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped } from "./native-local-process-stop.js";
|
||||
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
|
||||
import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js";
|
||||
@@ -214,6 +214,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
return blocked("send_intent_missing", "Send the saved message again to continue.");
|
||||
const authorizedAt = nativeQueuedSend ? interruptRequestedAt!
|
||||
: response?.comment.createdAt ?? comment?.createdAt ?? retryWake?.requestedAt ?? new Date();
|
||||
if (!input.dryRun) await preserveTaskRemoteWorkspaceWriter(db, companyId, issueId);
|
||||
const blocker = await getExecutionBlocker(db, companyId, issueId);
|
||||
const [boundary] = await db.select({ createdAt: issueComments.createdAt }).from(issues).innerJoin(issueComments, and(
|
||||
eq(issueComments.companyId, issues.companyId), eq(issueComments.issueId, issues.id),
|
||||
@@ -280,6 +281,16 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
!run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved.");
|
||||
if (!queuedInterrupt && !queuedRequest && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
|
||||
if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically.");
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
|
||||
)).for("update");
|
||||
const [lockedRun] = await db.select(continuationRunFields).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, run.id),
|
||||
)).for("update");
|
||||
if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId ||
|
||||
lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null;
|
||||
run = lockedRun;
|
||||
const failedPreparation = await isFailedNativePreparation(db, run, coordinator);
|
||||
const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
|
||||
run.errorCode === "execution_reconciliation_required" &&
|
||||
!run.processPid && !run.processGroupId && !run.nativeSessionId;
|
||||
@@ -295,7 +306,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
const freshIntent = await canSelectFreshWorkspace({ db, companyId, issueId, agentId, actorId,
|
||||
sourceRunId: run.id, evidence: strategyEvidence, taskRetryRequestId: input.taskRetryRequestId,
|
||||
queuedRequestId: input.queuedCommentRequestId ?? savedChatRequestId });
|
||||
const legacyUserTurn = run.runtimeMode === "legacy" &&
|
||||
const legacyUserTurn = failedPreparation || run.runtimeMode === "legacy" &&
|
||||
action.cause === "legacy_execution_requires_reconciliation" &&
|
||||
(isConversationAdapter(agent.adapterType) ||
|
||||
(agent.adapterType === "paperclip_runner" && (freshIntent || await hasVerifiedWorkspaceRepair(db, run))));
|
||||
@@ -303,27 +314,13 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
// operator Stop or redeliver a message already consumed by this run.
|
||||
if (queuedRequest && !queuedInterrupt && (run.contextSnapshot?.wakeCommentId === commentId ||
|
||||
(Array.isArray(run.contextSnapshot?.wakeCommentIds) && run.contextSnapshot.wakeCommentIds.includes(commentId)))) return null;
|
||||
if (legacyUserTurn) {
|
||||
if (legacyUserTurn && !failedPreparation) {
|
||||
const historicalAdapter = await historicalAdapterType(db, run);
|
||||
// A settings change never converts a known process/webhook execution into
|
||||
// a conversation. Those adapters retain their reconciliation contract.
|
||||
if (historicalAdapter && !isConversationAdapter(historicalAdapter)) return null;
|
||||
if (!historicalAdapter && agent.adapterType === "paperclip_runner") return null;
|
||||
}
|
||||
// For pre-upgrade rows without adapter evidence, only a new explicit user
|
||||
// turn is allowed, after the termination proofs below. This does not infer
|
||||
// an old adapter type, certify old outcomes, or authorize automatic replay.
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
|
||||
)).for("update");
|
||||
// Same lock order as the native claim. Re-read the run while holding both
|
||||
// locks before accepting the never-claimed startup proof.
|
||||
const [lockedRun] = await db.select(continuationRunFields).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, run.id),
|
||||
)).for("update");
|
||||
if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId ||
|
||||
lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null;
|
||||
run = lockedRun;
|
||||
const interruptedByThisClick = response?.source.requiresFreshSession === true &&
|
||||
interruptIntent?.targetRunId === run.id && run.resultJson?.operatorInterrupted === true &&
|
||||
run.resultJson?.interruptionSource === "issue_comment_interrupt" && run.resultJson?.interruptedIssueId === issueId &&
|
||||
@@ -351,7 +348,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
return blocked("workspace_repair_required", "Paperclip is recovering saved work before continuing. Your message is saved.");
|
||||
}
|
||||
const cancelledStartup = await isCancelledNativeStartup(db, run, coordinator);
|
||||
if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission && !cancelledStartup) return null;
|
||||
if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission && !cancelledStartup && !failedPreparation) return null;
|
||||
if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission &&
|
||||
!(queuedRequest && run.runtimeMode === "native" &&
|
||||
(run.status !== "cancelled" || authorizedAt > run.finishedAt!)) &&
|
||||
@@ -406,7 +403,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
}))) return null;
|
||||
} else {
|
||||
if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return blocked("local_cleanup", "Waiting for the previous environment to finish cleanup. Your message will start automatically.");
|
||||
if (!unusedAdmission && !cancelledStartup) {
|
||||
if (!unusedAdmission && !cancelledStartup && !failedPreparation) {
|
||||
if (freshIntent && (run.processPid || run.processGroupId) && !await recordedLocalExecutionStopped(db, run))
|
||||
return blocked("process_identity_missing", "Paperclip is verifying that the previous session stopped.");
|
||||
// A missing process identity is not evidence that a provider exited.
|
||||
@@ -435,6 +432,17 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
}
|
||||
sources.push(run);
|
||||
}
|
||||
if (taskRetry && !sources.some(run => run.id === taskRetry.sourceRunId)) {
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, taskRetry.sourceRunId),
|
||||
)).for("update");
|
||||
const [source] = await db.select(continuationRunFields).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, taskRetry.sourceRunId), eq(heartbeatRuns.agentId, agentId),
|
||||
)).for("update");
|
||||
if (!source || (source.nativeIssueId ?? source.issueId ?? source.contextSnapshot?.issueId) !== issueId
|
||||
|| adapterExecutionControls.has(source.id) || !await isFailedNativePreparation(db, source, coordinator)) return null;
|
||||
sources.push(source);
|
||||
}
|
||||
const nativeSources = sources.filter(run => run.runtimeMode === "native");
|
||||
const executedSources = sources.filter(run => run.runtimeMode === "native" || run.errorCode !== "execution_reconciliation_required" || run.startedAt);
|
||||
if (!executedSources.length || (retry && !sources.some(run => run.id === input.failedRunId)) ||
|
||||
|
||||
@@ -9,6 +9,7 @@ import { adapterExecutionControls } from "./adapter-execution-control.js";
|
||||
import { hasPendingRunCleanup } from "./execution-blocker.js";
|
||||
import { toExecutionWorkspace } from "./execution-workspaces.js";
|
||||
import { isBoundIsolatedTaskDirectory } from "./isolated-task-directory.js";
|
||||
import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
|
||||
export type ContinuationReadiness =
|
||||
| { mode: "resume_preserved_workspace"; preservationReceiptId: string }
|
||||
@@ -167,6 +168,33 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick<typeof heart
|
||||
if (intent.strategy !== "fresh_isolated_workspace") {
|
||||
const preservationReceiptId = await verifiedWorkspaceRepairReceipt(db, source);
|
||||
if (preservationReceiptId) return { mode: "resume_preserved_workspace", preservationReceiptId };
|
||||
const [bound] = await db.select({ id: executionWorkspaces.id, metadata: executionWorkspaces.metadata }).from(issues)
|
||||
.innerJoin(executionWorkspaces, and(eq(executionWorkspaces.companyId, issues.companyId),
|
||||
eq(executionWorkspaces.id, issues.executionWorkspaceId)))
|
||||
.where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId)));
|
||||
const authority = readRemoteWorkspaceAuthority(bound?.metadata);
|
||||
if (bound && authority && !authority.dirty && !authority.writerRunId && authority.snapshotId) {
|
||||
const recovered = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId,
|
||||
contextSnapshot: sql<Record<string, unknown>>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId')`,
|
||||
runnerProfileJson: sql<Record<string, unknown>>`jsonb_build_object('remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace',
|
||||
'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync')`,
|
||||
}).from(issueRecoveryActions)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, issueRecoveryActions.companyId),
|
||||
sql`${heartbeatRuns.id}::text = ${issueRecoveryActions.evidence}->>'runId'`))
|
||||
.where(and(eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
eq(heartbeatRuns.agentId, run.agentId),
|
||||
sql`${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,runId}' = ${run.id}`,
|
||||
sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,workspaceId}' = ${bound.id}`,
|
||||
sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,snapshotId}' = ${authority.snapshotId}`)).limit(1);
|
||||
for (const writer of recovered) {
|
||||
const remote = remoteWorkspaceRunReferenceSchema.safeParse(writer.runnerProfileJson?.remoteWorkspace);
|
||||
if (!remote.success || remote.data.state !== "saved" || remote.data.workspaceId !== bound.id
|
||||
|| remote.data.epoch !== authority.epoch || remote.data.snapshotId !== authority.snapshotId) continue;
|
||||
const receipt = await verifiedWorkspaceRepairReceipt(db, writer);
|
||||
if (receipt) return { mode: "resume_preserved_workspace", preservationReceiptId: receipt };
|
||||
}
|
||||
}
|
||||
const workspace = run.contextSnapshot?.previousRunId === source.id
|
||||
? await reusableDirectoryAfterPreparationFailure(db, source, issueId) : null;
|
||||
if (workspace?.cwd) {
|
||||
|
||||
@@ -62,6 +62,7 @@ import {
|
||||
} from "../native-runtime/native-review-participant.js";
|
||||
import { claimQueuedNativeReviewRun } from "../native-runtime/native-review-dispatch.js";
|
||||
import { adapterExecutionControls } from "../adapter-execution-control.js";
|
||||
import { preserveTaskRemoteWorkspaceWriter } from "../native-runtime/native-failed-workspace-recovery.js";
|
||||
import { executionFailureRetryCount } from "../execution-recovery-attempt.js";
|
||||
import {
|
||||
slackMentionAllowsRunStart,
|
||||
@@ -3486,6 +3487,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
)).limit(1)
|
||||
: [];
|
||||
const verifiedConversationReset = Boolean(resetComment && isConversationReset(resetComment.body));
|
||||
if (wakeCommentId && opts.requestedByActorType === "user")
|
||||
await preserveTaskRemoteWorkspaceWriter(tx as unknown as Db, issue.companyId, issue.id);
|
||||
const executionBlocker = await getExecutionBlocker(
|
||||
tx as unknown as Db, issue.companyId, issue.id,
|
||||
{ conversationResetCommentId: verifiedConversationReset ? wakeCommentId : null },
|
||||
|
||||
@@ -3,6 +3,7 @@ import { and, eq, inArray, isNotNull, ne, or, sql } from "drizzle-orm";
|
||||
import { environmentLeases, environments, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { hasWorkspaceRestoreFailure, WORKSPACE_RESTORE_FAILURE_CODES } from "@paperclipai/shared";
|
||||
import { hasConfirmedSandboxStopAndRetain, prepareSandboxStopAndRetain } from "./sandbox-stop-and-retain.js";
|
||||
import { preparedRemoteWorkspaceSource } from "./remote-workspace-lifecycle.js";
|
||||
|
||||
import { hasRequiredWorkspaceRecovery, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js";
|
||||
|
||||
@@ -77,6 +78,11 @@ export async function preserveLegacyWorkspaceRestoreSources(db: Db, run: Run, so
|
||||
if (recovery?.schema === LEGACY_WORKSPACE_RECOVERY_SCHEMA && recovery.runId === run.id) {
|
||||
retained.push(lease.id); continue;
|
||||
}
|
||||
if (source && run.runtimeMode === "native" && hasConfirmedSandboxStopAndRetain(lease)
|
||||
&& ["reuse_by_environment", "reuse_by_execution_workspace"].includes(lease.leasePolicy)
|
||||
&& await preparedRemoteWorkspaceSource(db, lease)) {
|
||||
retained.push(lease.id); continue;
|
||||
}
|
||||
// Cancellation can finish the dedicated stop before copy-back reports its
|
||||
// failure. Only the host's original source and two matching receipts can
|
||||
// turn that released row into a file-repair hold.
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import type { CancelledWorkspacePreservationScope } from "../cancelled-workspace-preservation-notifications.js";
|
||||
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
|
||||
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
||||
import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { environmentLeases, executionWorkspaces, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js";
|
||||
import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
@@ -22,6 +22,32 @@ function workspaceRecoveryReference(profile: Record<string, unknown> | null | un
|
||||
return readNativeWorkspaceSyncReference(profile?.nativeWorkspaceSync) ?? (remote.success ? remote.data : null);
|
||||
}
|
||||
|
||||
export async function preserveTaskRemoteWorkspaceWriter(db: Db, companyId: string, issueId: string) {
|
||||
const [bound] = await db.select({ id: executionWorkspaces.id, metadata: executionWorkspaces.metadata }).from(issues)
|
||||
.innerJoin(executionWorkspaces, and(eq(executionWorkspaces.id, issues.executionWorkspaceId),
|
||||
eq(executionWorkspaces.companyId, issues.companyId)))
|
||||
.where(and(eq(issues.companyId, companyId), eq(issues.id, issueId)));
|
||||
if (!bound) return;
|
||||
const authority = readRemoteWorkspaceAuthority(bound.metadata);
|
||||
if (!authority?.dirty || !authority.writerRunId) return;
|
||||
const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId,
|
||||
contextSnapshot: sql<Record<string, unknown>>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId')`,
|
||||
runnerProfileJson: sql<Record<string, unknown>>`jsonb_build_object('remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace',
|
||||
'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync')`,
|
||||
}).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, authority.writerRunId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "failed"),
|
||||
sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text,
|
||||
${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`,
|
||||
));
|
||||
const reference = remoteWorkspaceRunReferenceSchema.safeParse(run?.runnerProfileJson.remoteWorkspace);
|
||||
if (!run || run.runnerProfileJson.nativeWorkspaceSync != null || !reference.success || reference.data.state !== "prepared"
|
||||
|| reference.data.workspaceId !== bound.id || reference.data.epoch !== authority.epoch) return;
|
||||
try { return await preserveNativeFailedWorkspace(db, run); }
|
||||
catch (error) { if (!(error instanceof NativeWorkspaceFinalizationBusyError)) throw error; }
|
||||
}
|
||||
|
||||
export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "nativeIssueId" | "issueId" | "contextSnapshot" | "runnerProfileJson">,
|
||||
options: { allowPreterminal?: boolean; requireCancelledStop?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) {
|
||||
|
||||
@@ -1,49 +1,10 @@
|
||||
import { environmentLeases, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { and, eq, exists, sql } from "drizzle-orm";
|
||||
import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js";
|
||||
import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
|
||||
|
||||
type WorkspaceSourceLease = Pick<typeof environmentLeases.$inferSelect,
|
||||
"id" | "companyId" | "heartbeatRunId" | "provider" | "providerLeaseId">;
|
||||
|
||||
/** Preparation publishes file authority before native execution is selected.
|
||||
* Neither runtimeMode nor a later execution-input projection owns these files. */
|
||||
async function preparedRemoteWorkspaceSource(db: Db, lease: WorkspaceSourceLease) {
|
||||
if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return null;
|
||||
const [bound] = await db.select({ lease: environmentLeases, run: {
|
||||
id: heartbeatRuns.id, runtimeMode: heartbeatRuns.runtimeMode,
|
||||
controllerBootId: heartbeatRuns.controllerBootId,
|
||||
issueId: sql<string | null>`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text,
|
||||
${heartbeatRuns.contextSnapshot}->>'issueId')`,
|
||||
reference: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`,
|
||||
legacyReference: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`,
|
||||
} }).from(environmentLeases)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId),
|
||||
eq(heartbeatRuns.companyId, environmentLeases.companyId)))
|
||||
.where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId),
|
||||
eq(environmentLeases.heartbeatRunId, lease.heartbeatRunId), eq(environmentLeases.provider, lease.provider),
|
||||
eq(environmentLeases.providerLeaseId, lease.providerLeaseId),
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'->>'state' = 'prepared'`)).limit(1);
|
||||
if (!bound) return null;
|
||||
const parsed = remoteWorkspaceRunReferenceSchema.safeParse(bound.run.reference);
|
||||
if (!parsed.success || parsed.data.state !== "prepared" || bound.run.legacyReference != null) return null;
|
||||
const reference = parsed.data;
|
||||
const issueId = bound.run.issueId;
|
||||
if (typeof issueId !== "string" || bound.lease.issueId !== issueId || reference.leaseId !== bound.lease.id
|
||||
|| reference.providerLeaseId !== bound.lease.providerLeaseId || reference.remoteCwd !== bound.lease.metadata?.remoteCwd
|
||||
|| bound.lease.executionWorkspaceId && bound.lease.executionWorkspaceId !== reference.workspaceId) return null;
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(and(
|
||||
eq(executionWorkspaces.companyId, lease.companyId), eq(executionWorkspaces.id, reference.workspaceId))).limit(1);
|
||||
const authority = readRemoteWorkspaceAuthority(workspace?.metadata);
|
||||
if (!workspace || workspace.sourceIssueId !== issueId && workspace.mode !== "shared_workspace"
|
||||
|| !authority || !authority.dirty || authority.writerRunId !== bound.run.id || authority.epoch !== reference.epoch
|
||||
|| authority.snapshotId !== reference.snapshotId || !authority.location
|
||||
|| authority.location.environmentId !== bound.lease.environmentId || authority.location.leaseId !== bound.lease.id
|
||||
|| authority.location.providerLeaseId !== bound.lease.providerLeaseId || authority.location.remoteCwd !== reference.remoteCwd) return null;
|
||||
return { ...bound, workspace: workspace!, reference, issueId };
|
||||
}
|
||||
import { preparedRemoteWorkspaceSource, type WorkspaceSourceLease } from "../remote-workspace-lifecycle.js";
|
||||
|
||||
/** File-only recovery for setup that never reached native admission. The live
|
||||
* preparer must still own its controller; orphan repair accepts terminal runs
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { executionWorkspaces, executionWorkspaceSnapshots, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority, type EnvironmentLease } from "@paperclipai/shared";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, executionWorkspaces, executionWorkspaceSnapshots, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema, type EnvironmentLease } from "@paperclipai/shared";
|
||||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
|
||||
type Workspace = Pick<typeof executionWorkspaces.$inferSelect, "id" | "companyId" | "metadata">;
|
||||
@@ -9,6 +9,44 @@ export type DestroyRemoteWorkspaceLease = (input: {
|
||||
companyId: string; workspaceId: string; leaseId: string; environmentId: string; providerLeaseId: string;
|
||||
}) => Promise<EnvironmentLease | null>;
|
||||
|
||||
export type WorkspaceSourceLease = Pick<typeof environmentLeases.$inferSelect,
|
||||
"id" | "companyId" | "heartbeatRunId" | "provider" | "providerLeaseId">;
|
||||
|
||||
export async function preparedRemoteWorkspaceSource(db: Db, lease: WorkspaceSourceLease) {
|
||||
if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return null;
|
||||
const [bound] = await db.select({ lease: environmentLeases, run: {
|
||||
id: heartbeatRuns.id, runtimeMode: heartbeatRuns.runtimeMode,
|
||||
controllerBootId: heartbeatRuns.controllerBootId,
|
||||
issueId: sql<string | null>`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text,
|
||||
${heartbeatRuns.contextSnapshot}->>'issueId')`,
|
||||
reference: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`,
|
||||
legacyReference: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`,
|
||||
} }).from(environmentLeases)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId),
|
||||
eq(heartbeatRuns.companyId, environmentLeases.companyId)))
|
||||
.where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId),
|
||||
eq(environmentLeases.heartbeatRunId, lease.heartbeatRunId), eq(environmentLeases.provider, lease.provider),
|
||||
eq(environmentLeases.providerLeaseId, lease.providerLeaseId),
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'->>'state' = 'prepared'`)).limit(1);
|
||||
if (!bound) return null;
|
||||
const parsed = remoteWorkspaceRunReferenceSchema.safeParse(bound.run.reference);
|
||||
if (!parsed.success || parsed.data.state !== "prepared" || bound.run.legacyReference != null) return null;
|
||||
const reference = parsed.data;
|
||||
const issueId = bound.run.issueId;
|
||||
if (typeof issueId !== "string" || bound.lease.issueId !== issueId || reference.leaseId !== bound.lease.id
|
||||
|| reference.providerLeaseId !== bound.lease.providerLeaseId || reference.remoteCwd !== bound.lease.metadata?.remoteCwd
|
||||
|| bound.lease.executionWorkspaceId && bound.lease.executionWorkspaceId !== reference.workspaceId) return null;
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(and(
|
||||
eq(executionWorkspaces.companyId, lease.companyId), eq(executionWorkspaces.id, reference.workspaceId))).limit(1);
|
||||
const authority = readRemoteWorkspaceAuthority(workspace?.metadata);
|
||||
if (!workspace || workspace.sourceIssueId !== issueId && workspace.mode !== "shared_workspace"
|
||||
|| !authority || !authority.dirty || authority.writerRunId !== bound.run.id || authority.epoch !== reference.epoch
|
||||
|| authority.snapshotId !== reference.snapshotId || !authority.location
|
||||
|| authority.location.environmentId !== bound.lease.environmentId || authority.location.leaseId !== bound.lease.id
|
||||
|| authority.location.providerLeaseId !== bound.lease.providerLeaseId || authority.location.remoteCwd !== reference.remoteCwd) return null;
|
||||
return { ...bound, workspace: workspace!, reference, issueId };
|
||||
}
|
||||
|
||||
export async function inspectRemoteWorkspacePersistence(db: Db | Transaction, workspace: Workspace) {
|
||||
const authority = readRemoteWorkspaceAuthority(workspace.metadata);
|
||||
if (!authority || authority.dirty || authority.writerRunId || !authority.snapshotId) {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agents, agentWakeupRequests, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db";
|
||||
import { agents, agentWakeupRequests, companies, createDb, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { priorRecoveryApiReceipt, unresolvedRecoveryToolInvocation } from "./recovery-effect-receipts.js";
|
||||
import { requestTaskRetry } from "./task-retry.js";
|
||||
@@ -11,6 +11,7 @@ import { createSystemRecovery, failSystemRecovery, readSystemRecovery, SystemRec
|
||||
import { instanceSettingsService } from "./instance-settings.js";
|
||||
import { heartbeatService } from "./heartbeat.js";
|
||||
import { taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./chat-channels/task-retry-authority.js";
|
||||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
|
||||
describe("durable task Retry admission", () => {
|
||||
let postgres: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -57,6 +58,90 @@ describe("durable task Retry admission", () => {
|
||||
expect(await requestTaskRetry(input)).toMatchObject({ accepted: false, stale: true });
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(0);
|
||||
});
|
||||
it.each(["stopped", "running", "owned", "changed_workspace", "changed_source", "missing_stop", "destroyed", "foreign_task"].flatMap(state =>
|
||||
["retry", "message"].map(entry => ({ state, entry }))).concat([{ state: "stopped", entry: "preview" }]))(
|
||||
"preserves the actual remote writer behind a newer setup failure only with exact stopped authority: $entry $state", async ({ state, entry }) => {
|
||||
const input = await seed();
|
||||
const [latest] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.observedRunId));
|
||||
const writerId = randomUUID(), workspaceId = randomUUID(), pluginId = randomUUID();
|
||||
const sourceIssueId = state === "foreign_task" ? randomUUID() : input.issueId;
|
||||
if (sourceIssueId !== input.issueId) await db.insert(issues).values({ id: sourceIssueId,
|
||||
companyId: input.companyId, title: "Another task", status: "blocked", assigneeAgentId: latest.agentId });
|
||||
await db.update(heartbeatRuns).set({ errorCode: "setup_failed", resultJson: {
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false },
|
||||
} }).where(eq(heartbeatRuns.id, latest.id));
|
||||
await db.insert(heartbeatRuns).values({ id: writerId, companyId: input.companyId, agentId: latest.agentId,
|
||||
issueId: sourceIssueId, nativeIssueId: sourceIssueId, runtimeMode: "native",
|
||||
status: state === "running" ? "running" : "failed", finishedAt: state === "running" ? null : new Date(),
|
||||
createdAt: new Date(latest.createdAt.getTime() - 60_000),
|
||||
contextSnapshot: { issueId: sourceIssueId, executionWorkspaceId: workspaceId },
|
||||
resultJson: { workspaceRestoreFailure: "restore_failed", originalDiagnostic: true },
|
||||
});
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: input.companyId, issueId: sourceIssueId,
|
||||
runId: writerId, phase: "terminal_failure", leaseOwner: state === "owned" ? "live-controller" : null });
|
||||
const [environment] = await db.insert(environments).values({ name: `Stopped source ${writerId}`, driver: "sandbox" }).returning();
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId: input.companyId, issueId: sourceIssueId,
|
||||
heartbeatRunId: writerId, environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(),
|
||||
status: "released", cleanupStatus: "success", leasePolicy: "reuse_by_environment", acquiredAt: new Date(Date.now() - 60_000),
|
||||
releasedAt: new Date(),
|
||||
}).returning();
|
||||
const confirmedAt = lease.releasedAt!.toISOString();
|
||||
const identity = { companyId: input.companyId, runId: writerId, leaseId: lease.id,
|
||||
provider: lease.provider, providerLeaseId: lease.providerLeaseId, confirmedAt };
|
||||
const metadata = { driver: "sandbox", pluginId, sandboxProviderPlugin: true, remoteCwd: "/work",
|
||||
...(state === "missing_stop" ? {} : {
|
||||
remoteExecutionTermination: { ...identity, schema: "paperclip.remote-termination.v1", state: state === "destroyed" ? "destroyed" : "stopped" },
|
||||
sandboxStopAndRetainReceipt: { ...identity, schema: "paperclip.sandbox-stop-and-retain-receipt.v1",
|
||||
pluginId, method: "environmentStopLease", requestId: randomUUID() },
|
||||
}),
|
||||
};
|
||||
await db.update(environmentLeases).set({ metadata }).where(eq(environmentLeases.id, lease.id));
|
||||
const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, epoch: 3,
|
||||
snapshotId: randomUUID(), leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/work", state: "prepared" };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { remoteWorkspace: {
|
||||
...reference, ...(state === "changed_source" ? { providerLeaseId: randomUUID() } : {}),
|
||||
} } }).where(eq(heartbeatRuns.id, writerId));
|
||||
const authority = { schema: "paperclip.remote-workspace.v1", epoch: state === "changed_workspace" ? 4 : 3,
|
||||
snapshotId: reference.snapshotId, source: { kind: "empty" }, writerRunId: writerId, dirty: true,
|
||||
location: { environmentId: environment.id, leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/work" } };
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: input.companyId, sourceIssueId,
|
||||
name: "Unfinished remote writer", mode: "isolated_workspace", strategyType: "project_primary", cwd: "/work",
|
||||
providerType: "cloud_sandbox", metadata: { remoteWorkspace: authority } });
|
||||
await db.update(environmentLeases).set({ executionWorkspaceId: workspaceId }).where(eq(environmentLeases.id, lease.id));
|
||||
await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, input.issueId));
|
||||
|
||||
if (entry === "retry") {
|
||||
const receipt = await requestTaskRetry(input);
|
||||
expect((await requestTaskRetry(input)).requestId).toBe(receipt.requestId);
|
||||
} else {
|
||||
const [comment] = await db.insert(issueComments).values({ companyId: input.companyId, issueId: input.issueId,
|
||||
authorType: "user", authorUserId: input.actorId, body: "Continue from saved work.",
|
||||
createdAt: new Date() }).returning();
|
||||
for (let attempt = 0; attempt < 2; attempt++) await db.transaction(tx => admitExplicitNativeContinuation({
|
||||
db: tx as unknown as typeof db, companyId: input.companyId, issueId: input.issueId, agentId: latest.agentId,
|
||||
actorType: "user", actorId: input.actorId, commentId: comment.id, successorRunId: randomUUID(),
|
||||
reason: "issue_commented", dryRun: entry === "preview",
|
||||
}));
|
||||
}
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, writerId));
|
||||
const repairs = (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, input.issueId)))
|
||||
.filter(action => action.evidence.workspaceRestoreRecovery);
|
||||
if (state === "stopped" && entry !== "preview") {
|
||||
expect(source.resultJson).toMatchObject({ originalDiagnostic: true,
|
||||
workspaceRestoreRecovery: { leaseIds: [lease.id] } });
|
||||
expect(repairs).toHaveLength(1);
|
||||
expect(repairs[0].evidence.runId).toBe(writerId);
|
||||
} else {
|
||||
expect(source.resultJson).not.toHaveProperty("workspaceRestoreRecovery");
|
||||
expect(repairs).toHaveLength(0);
|
||||
}
|
||||
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0])
|
||||
.toMatchObject({ status: "released", leasePolicy: "reuse_by_environment", metadata });
|
||||
expect((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)))[0].metadata?.remoteWorkspace)
|
||||
.toEqual(authority);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, input.companyId))).toHaveLength(2);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(entry === "retry" ? 1 : 0);
|
||||
});
|
||||
it.each(["prepared", "mismatched", "owned"] as const)("reconstructs missing native preservation on deliberate Retry only for an exact unowned source: %s", async state => {
|
||||
const input = await seed();
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, input.issueId));
|
||||
|
||||
@@ -13,7 +13,7 @@ import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js"
|
||||
import { instanceSettingsService } from "./instance-settings.js";
|
||||
import { notifyChatActionWork } from "./chat-work-notifications.js";
|
||||
import { authorizationService } from "./authorization.js";
|
||||
import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js";
|
||||
import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js";
|
||||
import { NativeWorkspaceFinalizationBusyError } from "./native-runtime/native-workspace-finalization-ownership.js";
|
||||
import { adapterExecutionControls } from "./adapter-execution-control.js";
|
||||
import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js";
|
||||
@@ -256,6 +256,7 @@ export async function requestTaskRetry(input: {
|
||||
}
|
||||
const busy = ["queued", "running", "scheduled_retry"].includes(latest.status);
|
||||
if (!busy) {
|
||||
await preserveTaskRemoteWorkspaceWriter(db, issue.companyId, issue.id);
|
||||
// Historical deadline failures can predate their preservation record.
|
||||
// Register the exact prepared source; this proves neither a stopped
|
||||
// executor nor permission to replay its work.
|
||||
|
||||
@@ -338,12 +338,17 @@ describe("automatic retained workspace recovery", () => {
|
||||
}, 30_000);
|
||||
|
||||
it.each([
|
||||
{ status: "failed", runtimeMode: "native", uploadFailure: "none" },
|
||||
{ status: "cancelled", runtimeMode: "native", uploadFailure: "none" },
|
||||
{ status: "failed", runtimeMode: "legacy", uploadFailure: "final" },
|
||||
{ status: "failed", runtimeMode: "legacy", uploadFailure: "initial" },
|
||||
] as const)("saves the authoritative remote workspace after $runtimeMode $status without replaying a turn or certifying external effects ($uploadFailure upload failure)", async ({ status, runtimeMode, uploadFailure }) => {
|
||||
{ status: "failed", runtimeMode: "native", scenario: "none" },
|
||||
{ status: "failed", runtimeMode: "native", scenario: "registration" },
|
||||
{ status: "failed", runtimeMode: "native", scenario: "registration_after_dispatch" },
|
||||
{ status: "failed", runtimeMode: "native", scenario: "registration_existing_hold" },
|
||||
{ status: "failed", runtimeMode: "native", scenario: "registration_existing_hold_after_dispatch" },
|
||||
{ status: "cancelled", runtimeMode: "native", scenario: "none" },
|
||||
{ status: "failed", runtimeMode: "legacy", scenario: "final" },
|
||||
{ status: "failed", runtimeMode: "legacy", scenario: "initial" },
|
||||
] as const)("saves the authoritative remote workspace after $runtimeMode $status without replaying a turn or certifying external effects ($scenario)", async ({ status, runtimeMode, scenario }) => {
|
||||
const f = await seed(runtimeMode, false);
|
||||
let retryRequestId: string | undefined;
|
||||
const remoteCwd = await realpath(f.remote);
|
||||
targets.get(f.lease.id)!.remoteCwd = remoteCwd;
|
||||
f.runtime.resumeRunLease.mockImplementation(async () => ({ providerLeaseId: f.providerLeaseId, metadata: { remoteCwd } }));
|
||||
@@ -361,7 +366,7 @@ describe("automatic retained workspace recovery", () => {
|
||||
metadata: { ...boundLease.metadata, remoteCwd, agentId: (await run(f.runId)).agentId } })
|
||||
.where(eq(environmentLeases.id, f.lease.id));
|
||||
const before = await run(f.runId);
|
||||
if (uploadFailure === "final") {
|
||||
if (scenario === "final") {
|
||||
// A reused allocation can fail during preparation, before native input or
|
||||
// its finalization coordinator exists. Keep a real saved predecessor and
|
||||
// the ordinary exact lease handoff; neither grants a new provider turn.
|
||||
@@ -415,7 +420,7 @@ describe("automatic retained workspace recovery", () => {
|
||||
expect((await recoverySources.inspectWorkspaceRecoverySource(db, f.input)).sources).toHaveLength(1);
|
||||
const failedUpload = vi.spyOn(getStorageService(), "putFile").mockRejectedValueOnce(new Error("Synthetic saved-copy outage"));
|
||||
try {
|
||||
if (uploadFailure === "initial") {
|
||||
if (scenario === "initial") {
|
||||
// Exercise the original setup boundary, not a failed final save that
|
||||
// retries the same operation. The initial ledger reservation survives.
|
||||
await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId));
|
||||
@@ -453,7 +458,48 @@ describe("automatic retained workspace recovery", () => {
|
||||
epoch: authority.epoch, snapshotId: authority.snapshotId });
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false);
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
if (uploadFailure === "final") await makeRetryDue(f.actionId);
|
||||
if (scenario === "final") await makeRetryDue(f.actionId);
|
||||
}
|
||||
if (scenario.startsWith("registration")) {
|
||||
await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId));
|
||||
await db.update(heartbeatRuns).set({ resultJson: { workspaceRestoreFailure: "restore_failed", originalDiagnostic: true },
|
||||
runnerProfileJson: { ...(await run(f.runId)).runnerProfileJson, nativeResourceDisposition: "keep_running" },
|
||||
}).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [stopped] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
|
||||
const termination = stopped.metadata!.remoteExecutionTermination as Record<string, unknown>;
|
||||
await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment", releasedAt: new Date(String(termination.confirmedAt)), metadata: { ...stopped.metadata,
|
||||
sandboxStopAndRetainReceipt: { ...termination, schema: "paperclip.sandbox-stop-and-retain-receipt.v1",
|
||||
pluginId: stopped.metadata!.pluginId, method: "environmentStopLease", requestId: randomUUID() },
|
||||
} }).where(eq(environmentLeases.id, f.lease.id));
|
||||
const [later] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: before.agentId,
|
||||
issueId: f.issueId, runtimeMode: "legacy", status: "failed", errorCode: "setup_failed", finishedAt: new Date(),
|
||||
executionStage: "preparing", runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
||||
createdAt: new Date(before.createdAt.getTime() + 1), contextSnapshot: { issueId: f.issueId },
|
||||
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } },
|
||||
}).returning();
|
||||
if (scenario.endsWith("after_dispatch")) await appendHeartbeatRunEvent(db, {
|
||||
companyId: f.companyId, runId: later.id, agentId: before.agentId, eventType: "adapter.invoke", stream: "system",
|
||||
level: "info", message: "Execution started before the contradictory bootstrap result.", payload: null,
|
||||
});
|
||||
const existingHold = scenario.includes("existing_hold");
|
||||
await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
kind: existingHold ? "active_run_watchdog" : "stranded_assigned_issue",
|
||||
cause: existingHold ? "legacy_execution_requires_reconciliation" : "stranded_assigned_issue", fingerprint: `latest-setup:${later.id}`,
|
||||
ownerType: "board", nextAction: "Inspect the failed setup before continuing.",
|
||||
evidence: existingHold ? { runId: later.id, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } }
|
||||
: { latestRunId: later.id, latestRunStatus: "failed", latestRunErrorCode: "setup_failed" } });
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: before.agentId, status: "running" });
|
||||
const app = boardRecoveryApp(), retryIntent = { observedRunId: later.id, idempotencyKey: randomUUID() };
|
||||
const retry = (await request(app).post(`/api/issues/${f.issueId}/retry`).send(retryIntent).expect(202)).body;
|
||||
retryRequestId = retry.requestId;
|
||||
expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(retryIntent).expect(202)).body.requestId).toBe(retryRequestId);
|
||||
await heartbeatService(db).resumeTaskRetry(f.companyId, retryRequestId!);
|
||||
const [registered] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId),
|
||||
sql`${issueRecoveryActions.evidence}->>'runId' = ${f.runId}`));
|
||||
expect(registered).toBeDefined();
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(3);
|
||||
f.actionId = f.input.actionId = registered.id;
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async () => {}) ?? []);
|
||||
}
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
const recoveredAction = await action(f.actionId);
|
||||
@@ -474,24 +520,44 @@ describe("automatic retained workspace recovery", () => {
|
||||
}
|
||||
finally { await extracted.cleanup(); }
|
||||
expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n");
|
||||
expect(await run(f.runId)).toMatchObject({ status, errorCode: uploadFailure === "initial" ? "setup_failed" : "execution_interrupted",
|
||||
expect(await run(f.runId)).toMatchObject({ status, errorCode: scenario === "initial" ? "setup_failed" : "execution_interrupted",
|
||||
runnerProfileJson: { remoteWorkspace: { state: "saved", snapshotId: saved!.snapshotId } } });
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)))
|
||||
.toHaveLength(retryRequestId ? 1 : 0);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toHaveLength(0);
|
||||
expect(await db.select().from(costEvents).where(eq(costEvents.companyId, f.companyId))).toHaveLength(0);
|
||||
expect(await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, f.runId),
|
||||
eq(workspaceOperations.phase, "workspace_recovery_snapshot")))).toHaveLength(0);
|
||||
expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(uploadFailure === "final" ? 2 : 1);
|
||||
expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(scenario === "final" ? 2 : 1);
|
||||
expect(f.runtime.retryPendingSandboxTeardown).toHaveBeenCalledTimes(runtimeMode === "legacy" ? 2 : 1);
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true);
|
||||
await scheduleRecoveredWorkspaceCleanup(db, await run(f.runId));
|
||||
const [settledLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
|
||||
if (retryRequestId) {
|
||||
expect(settledLease).toMatchObject({ status: "released", leasePolicy: "reuse_by_environment" });
|
||||
await heartbeatService(db).resumeTaskRetry(f.companyId, retryRequestId);
|
||||
await heartbeatService(db).resumeTaskRetry(f.companyId, retryRequestId);
|
||||
const queued = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
|
||||
if (scenario.endsWith("after_dispatch")) {
|
||||
expect(queued).toHaveLength(0);
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retryRequestId)))[0])
|
||||
.toMatchObject({ status: "deferred_issue_execution", runId: null });
|
||||
return;
|
||||
}
|
||||
expect(queued).toHaveLength(1);
|
||||
expect(await readFreshWorkspaceReadiness(db, queued[0]))
|
||||
.toMatchObject({ mode: "resume_preserved_workspace", preservationReceiptId: expect.any(String) });
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ executionWorkspaceId: workspaceId });
|
||||
expect((await run(f.runId)).resultJson).toMatchObject({ originalDiagnostic: true });
|
||||
expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(1);
|
||||
return;
|
||||
}
|
||||
if (runtimeMode === "legacy") {
|
||||
expect(settledLease).toMatchObject({ status: "released",
|
||||
leasePolicy: uploadFailure === "initial" ? "retain_on_failure" : "reuse_by_execution_workspace" });
|
||||
leasePolicy: scenario === "initial" ? "retain_on_failure" : "reuse_by_execution_workspace" });
|
||||
expect((await run(f.runId)).runtimeMode).toBe("legacy");
|
||||
expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual([]);
|
||||
if (uploadFailure === "final") expect(snapshots.find(snapshot => snapshot.id === authority.snapshotId)).toMatchObject({ status: "superseded" });
|
||||
if (scenario === "final") expect(snapshots.find(snapshot => snapshot.id === authority.snapshotId)).toMatchObject({ status: "superseded" });
|
||||
else expect(snapshots.find(snapshot => snapshot.generation === 1)).toMatchObject({ status: "pending" });
|
||||
// Explicit Retry may admit the preserved files. Saturate agent capacity
|
||||
// so this tests admission without dispatching a new agent turn.
|
||||
@@ -504,7 +570,7 @@ describe("automatic retained workspace recovery", () => {
|
||||
expect(await readFreshWorkspaceReadiness(db, successor))
|
||||
.toMatchObject({ mode: "resume_preserved_workspace", preservationReceiptId: expect.any(String) });
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, successor.id))).toEqual([]);
|
||||
expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(uploadFailure === "final" ? 2 : 1);
|
||||
expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(scenario === "final" ? 2 : 1);
|
||||
return;
|
||||
}
|
||||
if (status === "failed") {
|
||||
|
||||
Reference in new issue
Block a user