mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 16:58:52 +01:00
DottaandPaperclip
f5e5b05a33
Land the thirteen-file native recovery fix independently from #18245. Retain verified Stop evidence and queued user comments. Retire stale preparation ownership atomically and require paired sandbox lease handoff receipts before workspace recovery. Co-Authored-By: Paperclip <noreply@paperclip.ing>
486 lines
36 KiB
TypeScript
486 lines
36 KiB
TypeScript
import { classifyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
|
import { createHash } from "node:crypto";
|
|
import { adapterExecutionControls } from "./adapter-execution-control.js";
|
|
import { isAcknowledgedNativeStop } from "./acknowledged-native-stop.js";
|
|
import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
|
|
import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
|
|
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
|
|
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
|
import { beginIdleTrackedWork } from "./task-admission.js";
|
|
import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js";
|
|
import { resumePreservedResultFinalization } from "./preserved-result-finalization.js";
|
|
import { and, asc, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm";
|
|
import { agentWakeupRequests, environmentLeases, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, workspaceOperations, type Db } from "@paperclipai/db";
|
|
import type { Environment, EnvironmentLease } from "@paperclipai/shared";
|
|
import { conflict } from "../errors.js";
|
|
import { logger } from "../middleware/logger.js";
|
|
import type { EnvironmentRuntimeService } from "./environment-runtime.js";
|
|
import { resolveEnvironmentExecutionTarget } from "./environment-execution-target.js";
|
|
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
|
import { hasStopOnlyCleanup, prepareHistoricalSandboxStopVerification, prepareSandboxStopAndRetain, prepareStoppedSandboxForRecovery, settleAbsentSandbox, settleStopOnlyCleanup } from "./sandbox-stop-and-retain.js";
|
|
import { requestSandboxAbsenceProof } from "./sandbox-absence-proof.js";
|
|
import { persistActivity, publishActivity } from "./activity-log.js";
|
|
import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js";
|
|
import { hasVerifiedWorkspaceRepair, readWorkspaceRecovery, type WorkspaceRecovery } from "./workspace-recovery-receipt.js";
|
|
import { withNativeWorkspaceFinalizationOwnership } from "./native-runtime/native-workspace-finalization-ownership.js";
|
|
import { readNativeWorkspaceSyncReference, readNativeWorkspaceSyncLocalCwd, resumeNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js";
|
|
import { classifyNativeWorkspaceFailure } from "./native-runtime/native-workspace-failure.js";
|
|
|
|
import { adoptSystemRecovery, loadSystemRecovery, updateSystemRecovery } from "./system-recovery.js";
|
|
import { SYSTEM_RECOVERY_MAX_RETRIES, createSystemRecovery, failSystemRecovery, readSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure, type SystemRecoveryState } from "./system-recovery-state.js";
|
|
import { inspectWorkspaceRecoverySource, workspaceRecoverySourceEligible, workspaceSourceBindingFailure } from "./workspace-recovery-source.js";
|
|
import { preserveWorkspaceSnapshot } from "./workspace-recovery-snapshot.js";
|
|
import { adoptHistoricalWorkspaceRecoveries } from "./workspace-recovery-backfill.js";
|
|
|
|
const exportCauses = ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"];
|
|
const terminal = ["failed", "interrupted", "timed_out", "cancelled", "succeeded"];
|
|
type Source = { companyId: string; issueId: string; runId: string; actionId: string };
|
|
|
|
type StopEvidenceSource = Omit<Source, "actionId"> & { actionId?: string };
|
|
|
|
async function recheckStoppedWorkspaceRecovery(db: Db, source: StopEvidenceSource, observe: boolean) {
|
|
return db.transaction(async tx => {
|
|
const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(
|
|
eq(issues.id, source.issueId), eq(issues.companyId, source.companyId))).for("update");
|
|
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.runId),
|
|
eq(heartbeatRuns.companyId, source.companyId))).for("update");
|
|
if (!issue || !run || run.runtimeMode !== "native" || run.nativeIssueId !== issue.id ||
|
|
!isAcknowledgedNativeStop(run) || adapterExecutionControls.has(run.id)) return 0;
|
|
const cancellation = run.resultJson?.executionCancellation as { state?: unknown; proof?: unknown; acknowledgedAt?: unknown } | undefined;
|
|
if (cancellation?.state !== "acknowledged" || cancellation.proof !== "provider_termination_receipt") return 0;
|
|
const [coordinator] = await tx.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner })
|
|
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, source.companyId),
|
|
eq(nativeRunFinalizations.runId, source.runId), eq(nativeRunFinalizations.issueId, source.issueId)));
|
|
if (!coordinator || coordinator.leaseOwner || !["terminal_failure", "applied"].includes(coordinator.phase)) return 0;
|
|
const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, source.companyId),
|
|
eq(environmentLeases.heartbeatRunId, source.runId)));
|
|
if (!leases.length || leases.some(lease => !lease.provider || lease.provider === "local" || !hasRemoteTerminationReceipt(lease))) return 0;
|
|
const intent = run.resultJson!.nativeCancellation as Record<string, unknown>;
|
|
const fingerprint = createHash("sha256").update(JSON.stringify({ companyId: source.companyId, issueId: source.issueId,
|
|
runId: source.runId, intentAuditId: intent.intentAuditId, acknowledgementAuditId: intent.acknowledgementAuditId,
|
|
acknowledgedAt: cancellation.acknowledgedAt, leases: leases.sort((a, b) => a.id.localeCompare(b.id)).map(lease => {
|
|
const receipt = lease.metadata!.remoteExecutionTermination as Record<string, unknown>;
|
|
return { leaseId: lease.id, provider: lease.provider, providerLeaseId: lease.providerLeaseId,
|
|
state: receipt.state, confirmedAt: receipt.confirmedAt };
|
|
}) })).digest("hex");
|
|
const actions = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, source.companyId),
|
|
eq(issueRecoveryActions.sourceIssueId, source.issueId), or(inArray(issueRecoveryActions.status, ["active", "escalated"]), and(
|
|
eq(issueRecoveryActions.status, "resolved"), sql`${issueRecoveryActions.evidence}#>>'{automaticRecovery,policy}' = 'preserve_without_replay_v1'`,
|
|
sql`${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'blocked'`,
|
|
sql`${issueRecoveryActions.evidence}#>>'{automaticRecovery,runId}' = ${source.runId}`)),
|
|
sql`${issueRecoveryActions.evidence}->>'runId' = ${source.runId}`,
|
|
source.actionId ? eq(issueRecoveryActions.id, source.actionId) : undefined)).for("update");
|
|
let changed = 0;
|
|
for (const action of actions) {
|
|
const current = readSystemRecovery(action.evidence), workspace = readWorkspaceRecovery(action.evidence), now = Date.now();
|
|
if (!current || current.sourceRunId !== source.runId || current.phase !== "verify_execution" ||
|
|
!workspace || workspace.runId !== source.runId || ["complete", "exhausted"].includes(current.state) || current.retriesUsed >= SYSTEM_RECOVERY_MAX_RETRIES ||
|
|
(current.deadlineAt !== null && now >= current.deadlineAt)) continue;
|
|
const stopWait = current.failure?.retryClass === "recheck_evidence" &&
|
|
["agent_stop_unconfirmed", "copyback_stop_unconfirmed"].includes(current.failure.code);
|
|
if (current.state !== "running" && current.failure && !stopWait) continue;
|
|
const prior = action.evidence.workspaceStopEvidence as Record<string, unknown> | undefined;
|
|
const same = prior?.schema === "paperclip.workspace-stop-evidence.v1" && prior.sourceRunId === source.runId &&
|
|
prior.generationId === current.generationId && prior.fingerprint === fingerprint;
|
|
if ((same && prior!.consumed === true) || (!observe && (!same || prior!.consumed !== false))) continue;
|
|
const due = Math.max(now, current.providerCooldownAt ?? 0);
|
|
const advance = current.state === "awaiting_evidence" && stopWait && current.attempts > 0 &&
|
|
current.deadlineAt !== null && due < current.deadlineAt && current.nextAttemptAt !== null && current.nextAttemptAt > due;
|
|
const consumed = current.state !== "running";
|
|
if (!advance && same && prior!.consumed === consumed) continue;
|
|
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
|
if (advance && !await updateSystemRecovery(tx as unknown as Db, { ...source, actionId: action.id }, current,
|
|
{ ...current, revision: current.revision + 1, nextAttemptAt: due })) continue;
|
|
const marker = { schema: "paperclip.workspace-stop-evidence.v1", sourceRunId: source.runId,
|
|
generationId: current.generationId, fingerprint, consumed };
|
|
const recorded = await tx.update(issueRecoveryActions).set({ evidence: sql`${issueRecoveryActions.evidence} ||
|
|
${JSON.stringify({ workspaceStopEvidence: marker })}::jsonb`, updatedAt: new Date() }).where(and(
|
|
eq(issueRecoveryActions.companyId, source.companyId), eq(issueRecoveryActions.id, action.id),
|
|
sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,generationId}' = ${current.generationId}`,
|
|
sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,revision}' = ${String(current.revision + (advance ? 1 : 0))}`))
|
|
.returning({ id: issueRecoveryActions.id });
|
|
if (!recorded.length) throw new Error("workspace_stop_evidence_changed");
|
|
changed += 1;
|
|
}
|
|
return changed;
|
|
});
|
|
}
|
|
|
|
export function observeStoppedWorkspaceRecovery(db: Db, source: StopEvidenceSource) {
|
|
return recheckStoppedWorkspaceRecovery(db, source, true);
|
|
}
|
|
|
|
/** Admission records intent only. No request handler runs physical recovery. */
|
|
export async function requestWorkspaceRecovery(input: Source & { db: Db; idempotencyKey: string; actorId: string }) {
|
|
return input.db.transaction(async tx => {
|
|
const [issue] = await tx.select().from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))).for("update");
|
|
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))).for("update");
|
|
const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.id, input.actionId),
|
|
eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId))).for("update");
|
|
if (!issue || !run || !action || action.evidence.runId !== run.id
|
|
|| (run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId) !== issue.id
|
|
|| !terminal.includes(run.status) || (!hasRequiredWorkspaceRecovery(run.resultJson) && !exportCauses.includes(action.cause))) {
|
|
throw conflict("The workspace recovery changed. Refresh the task before retrying.");
|
|
}
|
|
const prior = readWorkspaceRecovery(action.evidence);
|
|
if (prior?.requestIds.includes(input.idempotencyKey) || prior?.status === "recovered") return { actionId: action.id, status: prior.status };
|
|
const pending = prior && ["queued", "recovering"].includes(prior.status);
|
|
const state: WorkspaceRecovery = pending ? { ...prior!, requestIds: [...prior!.requestIds, input.idempotencyKey] }
|
|
: { schema: "paperclip.workspace-recovery.v1", runId: run.id, status: "queued", stage: "verifying_stop",
|
|
attempt: prior?.attempt ?? 0, attemptLimit: (prior?.attempt ?? 0) + 1,
|
|
requestIds: [...(prior?.requestIds ?? []), input.idempotencyKey] };
|
|
const previous = readSystemRecovery(action.evidence);
|
|
const active = previous && ["scheduled", "running", "awaiting_evidence"].includes(previous.state);
|
|
const system = previous?.state === "running" ? previous : active ? { ...previous, revision: previous.revision + 1,
|
|
nextAttemptAt: Math.max(Date.now(), previous.providerCooldownAt ?? 0) } : { ...createSystemRecovery(run.id),
|
|
revision: (previous?.revision ?? -1) + 1,
|
|
preservationOperationIds: previous?.preservationOperationIds ?? [],
|
|
providerCooldownAt: previous?.providerCooldownAt ?? null,
|
|
nextAttemptAt: previous?.providerCooldownAt ?? null };
|
|
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
|
await tx.update(issueRecoveryActions).set({ ownerType: "system", ownerAgentId: null, ownerUserId: null,
|
|
evidence: { ...action.evidence, workspaceRecovery: state, systemRecovery: system }, updatedAt: new Date() })
|
|
.where(eq(issueRecoveryActions.id, action.id));
|
|
await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "user", actorId: input.actorId,
|
|
action: "issue.workspace_recovery_requested", entityType: "issue", entityId: issue.id, runId: run.id,
|
|
details: { recoveryActionId: action.id, idempotencyKey: input.idempotencyKey,
|
|
...(previous && previous.generationId !== system.generationId ? { previousGeneration: previous } : {}) } });
|
|
return { actionId: action.id, status: state.status };
|
|
});
|
|
}
|
|
|
|
/** Physical repair remains serialized by the existing source ownership lock.
|
|
* Neither preservation nor cleanup creates an agent turn. */
|
|
export async function recoverRetainedWorkspace(input: Source & { db: Db; environmentRuntime: EnvironmentRuntimeService }) {
|
|
const { db } = input;
|
|
await adoptSystemRecovery(db, input);
|
|
const initial = await loadSystemRecovery(db, input);
|
|
if (!initial || initial.state === "complete" || initial.state === "exhausted" ||
|
|
(initial.nextAttemptAt !== null && initial.nextAttemptAt > Date.now())) return;
|
|
const runAttempt = async (ownership: { token: string; assertHeld(): Promise<void> }) => {
|
|
let current = await loadSystemRecovery(db, input);
|
|
if (!current) return;
|
|
const reserved = reserveSystemRecoveryAttempt(current);
|
|
if (!reserved) {
|
|
if (current.state !== "exhausted" && current.deadlineAt !== null && Date.now() >= current.deadlineAt)
|
|
await updateSystemRecovery(db, input, current, { ...current, revision: current.revision + 1,
|
|
state: "exhausted", nextAttemptAt: null });
|
|
return;
|
|
}
|
|
if (!await updateSystemRecovery(db, input, current, reserved)) return;
|
|
current = reserved;
|
|
const progress = async (patch: Partial<SystemRecoveryState>) => {
|
|
await ownership.assertHeld();
|
|
const next = { ...current!, ...patch, revision: current!.revision + 1 };
|
|
if (!await updateSystemRecovery(db, input, current!, next)) throw new Error("workspace_recovery_generation_changed");
|
|
current = next;
|
|
};
|
|
const observeActive = async () => {
|
|
await ownership.assertHeld();
|
|
await db.update(issueRecoveryActions).set({ evidence: sql`jsonb_set(${issueRecoveryActions.evidence},
|
|
'{systemRecoveryActiveAt}', ${JSON.stringify(Date.now())}::jsonb)` }).where(and(
|
|
eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, input.actionId),
|
|
sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,attemptId}' = ${reserved.attemptId}`));
|
|
};
|
|
let observing = Promise.resolve();
|
|
const observer = setInterval(() => { observing = observing.then(observeActive).catch(() => undefined); }, 15_000);
|
|
try {
|
|
await observeActive();
|
|
const inspected = await inspectWorkspaceRecoverySource(db, input);
|
|
let sourceUnavailable = false;
|
|
for (const source of inspected.sources) {
|
|
const { lease, environment, remoteCwd } = source;
|
|
if (hasRemoteTerminationReceipt(lease)
|
|
&& (lease.metadata?.remoteExecutionTermination as Record<string, unknown> | undefined)?.state === "destroyed") {
|
|
sourceUnavailable = true;
|
|
continue;
|
|
}
|
|
if (lease.provider === "daytona" && lease.providerLeaseId) {
|
|
const absence = await requestSandboxAbsenceProof({ companyId: input.companyId, issueId: input.issueId, runId: input.runId, leaseId: lease.id,
|
|
provider: "daytona", providerLeaseId: lease.providerLeaseId, ownerToken: ownership.token });
|
|
if (absence) {
|
|
await ownership.assertHeld();
|
|
if (!await settleAbsentSandbox(db, lease, absence)) throw new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence");
|
|
sourceUnavailable = true;
|
|
continue;
|
|
}
|
|
}
|
|
let reference = source.reference;
|
|
if (reference) {
|
|
try { await readNativeWorkspaceSyncLocalCwd({ runId: input.runId, companyId: input.companyId,
|
|
workspaceId: reference.workspaceId, reference }); }
|
|
catch (error) {
|
|
// Missing local merge metadata permits a separate snapshot of the
|
|
// exact bound allocation. Corrupt metadata never weakens binding.
|
|
if ((error as NodeJS.ErrnoException)?.code === "ENOENT") reference = null;
|
|
else throw workspaceSourceBindingFailure("local_descriptor_invalid");
|
|
}
|
|
}
|
|
let stoppedLease = lease;
|
|
if (!hasRemoteTerminationReceipt(stoppedLease)) {
|
|
const pending = await db.transaction(async tx => {
|
|
const [fresh] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
|
|
eq(environmentLeases.id, lease.id))).for("update");
|
|
if (!fresh || fresh.metadata?.pendingCleanupInFlight === true) return null;
|
|
const prepared = hasStopOnlyCleanup(fresh) ? fresh : fresh.status === "released"
|
|
? await prepareHistoricalSandboxStopVerification(tx as unknown as Db, fresh)
|
|
: await prepareSandboxStopAndRetain(tx as unknown as Db, fresh);
|
|
if (!prepared) return null;
|
|
const [claimed] = await tx.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} ||
|
|
${JSON.stringify({ pendingCleanupAttemptId: ownership.token, pendingCleanupInFlight: true,
|
|
pendingCleanupLeaseExpiresAtMs: Date.now() + 15 * 60_000 })}::jsonb` })
|
|
.where(eq(environmentLeases.id, lease.id)).returning();
|
|
return claimed;
|
|
});
|
|
if (!pending) throw new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence");
|
|
// The cleanup service owns exact-allocation termination. No process ID
|
|
// or terminal run status is substituted for its provider receipt.
|
|
const receipt = await input.environmentRuntime.retryPendingSandboxTeardown({
|
|
environment: environment as unknown as Environment, lease: pending as EnvironmentLease }).catch(async () => {
|
|
await settleStopOnlyCleanup(db, pending, { attemptId: ownership.token });
|
|
throw new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence");
|
|
});
|
|
const settled = await settleStopOnlyCleanup(db, pending, { attemptId: ownership.token, receipt });
|
|
if (!settled || !hasRemoteTerminationReceipt(settled))
|
|
throw new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence");
|
|
stoppedLease = settled;
|
|
}
|
|
const target = await resolveEnvironmentExecutionTarget({ db, companyId: input.companyId, adapterType: "paperclip_runner",
|
|
environment, leaseId: lease.id, leaseMetadata: stoppedLease.metadata,
|
|
lease: stoppedLease as EnvironmentLease, environmentRuntime: input.environmentRuntime });
|
|
if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.remoteCwd !== remoteCwd)
|
|
throw workspaceSourceBindingFailure("execution_target_invalid");
|
|
await ownership.assertHeld();
|
|
const cleanupLease = await db.transaction(async tx => {
|
|
const [fresh] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
|
|
eq(environmentLeases.id, lease.id))).for("update");
|
|
if (!fresh || !hasRemoteTerminationReceipt(fresh) || fresh.providerLeaseId !== lease.providerLeaseId)
|
|
throw new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence");
|
|
const pending = await prepareStoppedSandboxForRecovery(tx as unknown as Db, fresh);
|
|
if (!pending) throw workspaceSourceBindingFailure("recovery_stop_intent_invalid");
|
|
const [claimed] = await tx.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({
|
|
pendingCleanupAttemptId: ownership.token, pendingCleanupInFlight: true, pendingCleanupLeaseExpiresAtMs: Date.now() + 15 * 60_000,
|
|
})}::jsonb` }).where(eq(environmentLeases.id, fresh.id)).returning();
|
|
return claimed;
|
|
});
|
|
let renewal = Promise.resolve();
|
|
const keepalive = setInterval(() => { renewal = renewal.then(async () => {
|
|
await ownership.assertHeld();
|
|
await db.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, '{pendingCleanupLeaseExpiresAtMs}', ${JSON.stringify(Date.now() + 15 * 60_000)}::jsonb)` })
|
|
.where(and(eq(environmentLeases.id, lease.id), sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${ownership.token}`));
|
|
}).catch(() => undefined); }, 30_000);
|
|
let stopSettled = false;
|
|
try {
|
|
const opened = await input.environmentRuntime.resumeWorkspaceRecoveryLease({ source: input,
|
|
leaseId: cleanupLease.id, environmentId: environment.id, ownership });
|
|
if (opened?.providerLeaseId !== lease.providerLeaseId) throw workspaceSourceBindingFailure("reopened_allocation_mismatch");
|
|
if (opened.metadata?.remoteCwd !== remoteCwd) throw workspaceSourceBindingFailure("reopened_cwd_mismatch");
|
|
await inspectWorkspaceRecoverySource(db, input);
|
|
await progress({ phase: "preserve_files" });
|
|
let operationId: string;
|
|
if (reference) {
|
|
const copied = await resumeNativeWorkspaceSync({ db, runId: input.runId, target, assertOwnership: ownership.assertHeld });
|
|
if (!copied) throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required");
|
|
const after = await inspectWorkspaceRecoverySource(db, input);
|
|
const finalized = after.sources.find(item => item.lease.id === lease.id)?.reference;
|
|
if (finalized?.state !== "finalized" || !finalized.finalHostSha256)
|
|
throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required");
|
|
await ownership.assertHeld();
|
|
const [existing] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
|
eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.heartbeatRunId, input.runId),
|
|
eq(workspaceOperations.phase, "workspace_recovery"), eq(workspaceOperations.status, "succeeded"),
|
|
sql`${workspaceOperations.metadata}->>'leaseId' = ${lease.id}`)).limit(1);
|
|
const copy = existing ?? (await db.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,
|
|
heartbeatRunId: input.runId, executionWorkspaceId: lease.executionWorkspaceId, phase: "workspace_recovery",
|
|
status: "succeeded", finishedAt: new Date(), metadata: { recoveryActionId: input.actionId, leaseId: lease.id,
|
|
providerLeaseId: lease.providerLeaseId, descriptorSha256: finalized.descriptorSha256,
|
|
finalHostSha256: finalized.finalHostSha256 } }).returning({ id: workspaceOperations.id }))[0];
|
|
operationId = copy.id;
|
|
} else {
|
|
operationId = await preserveWorkspaceSnapshot({ ...input, leaseId: lease.id,
|
|
providerLeaseId: lease.providerLeaseId!, target, assertHeld: ownership.assertHeld });
|
|
}
|
|
await progress({ preservationOperationIds: [...new Set([...current!.preservationOperationIds, operationId])] });
|
|
clearInterval(keepalive); await renewal;
|
|
await ownership.assertHeld();
|
|
const receipt = await input.environmentRuntime.retryPendingSandboxTeardown({ environment: environment as unknown as Environment,
|
|
lease: cleanupLease as EnvironmentLease });
|
|
const settled = await settleStopOnlyCleanup(db, cleanupLease, { attemptId: ownership.token, receipt });
|
|
stopSettled = Boolean(settled && hasRemoteTerminationReceipt(settled));
|
|
if (!stopSettled) throw new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence");
|
|
} finally {
|
|
clearInterval(keepalive); await renewal;
|
|
if (!stopSettled) {
|
|
// A lost resume/export reply retains the same stop-only obligation.
|
|
// The maintenance worker must never turn it into destructive cleanup.
|
|
try {
|
|
await ownership.assertHeld();
|
|
const receipt = await input.environmentRuntime.retryPendingSandboxTeardown({ environment: environment as unknown as Environment,
|
|
lease: cleanupLease as EnvironmentLease });
|
|
await settleStopOnlyCleanup(db, cleanupLease, { attemptId: ownership.token, receipt });
|
|
} catch { /* The persisted stop-only obligation remains authoritative. */ }
|
|
}
|
|
}
|
|
}
|
|
await ownership.assertHeld();
|
|
if (sourceUnavailable) throw new SystemRecoveryFailure("source_unavailable", "use_alternative_strategy");
|
|
const next = { ...current!, revision: current!.revision + 1, phase: "reconcile_effects" as const,
|
|
state: "complete" as const, nextAttemptAt: null };
|
|
const activity = await db.transaction(async tx => {
|
|
await notifyNativeResultCommit(tx, input);
|
|
await notifyNativeWorkspaceFinalization(tx, input);
|
|
await notifyNativeStatusReassessment(tx, input);
|
|
await tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,
|
|
heartbeatRunId: input.runId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(),
|
|
metadata: { owningService: "workspace_recovery", preservationOperationIds: next.preservationOperationIds } });
|
|
if (!await updateSystemRecovery(tx as unknown as Db, input, current!, next))
|
|
throw new Error("workspace_recovery_generation_changed");
|
|
return persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "workspace-recovery",
|
|
action: "issue.workspace_recovered", entityType: "issue", entityId: input.issueId, runId: input.runId,
|
|
details: { recoveryActionId: input.actionId, summary: "Saved work recovered." } });
|
|
});
|
|
current = next;
|
|
try { publishActivity(activity.publication); } catch { logger.warn({ runId: input.runId }, "Recovery activity delivery deferred"); }
|
|
} catch (error) {
|
|
await ownership.assertHeld();
|
|
if (current!.state === "complete") {
|
|
logger.warn({ runId: input.runId }, "Preserved workspace activity delivery deferred");
|
|
return;
|
|
}
|
|
const classification = classifyNativeWorkspaceFailure(error);
|
|
const failure = error instanceof SystemRecoveryFailure ? error
|
|
: classifyWorkspaceRestoreFailure(error) === "restore_invalid_metadata"
|
|
? new SystemRecoveryFailure("snapshot_verification_failed", "service_required")
|
|
: classification.code === "workspace_sync_out_unsafe_archive"
|
|
? new SystemRecoveryFailure("archive_unsafe", "service_required")
|
|
: classification.code === "workspace_sync_out_unrecoverable"
|
|
? new SystemRecoveryFailure("source_unavailable", "use_alternative_strategy")
|
|
: new SystemRecoveryFailure(current!.phase === "verify_execution" ? "provider_unavailable" : "export_interrupted", "retry_transient");
|
|
const retryAfterAt = (error as { retryAfterAt?: unknown })?.retryAfterAt;
|
|
await updateSystemRecovery(db, input, current!, failSystemRecovery(current!, failure, {
|
|
...(typeof retryAfterAt === "number" && Number.isFinite(retryAfterAt) ? { retryAfterAt } : {}),
|
|
}));
|
|
if (failure.retryClass === "recheck_evidence" && ["agent_stop_unconfirmed", "copyback_stop_unconfirmed"].includes(failure.code))
|
|
await recheckStoppedWorkspaceRecovery(db, input, false);
|
|
logger.warn({ err: error, runId: input.runId, failureCode: failure.code, phase: current!.phase }, "Workspace preservation deferred; source retained");
|
|
} finally { clearInterval(observer); await observing; }
|
|
};
|
|
const owned = await withNativeWorkspaceFinalizationOwnership({ db, companyId: input.companyId, runId: input.runId,
|
|
retainedRepairActionId: input.actionId }, runAttempt);
|
|
if (!owned.acquired && owned.reason === "owner_unverified") {
|
|
const current = await loadSystemRecovery(db, input);
|
|
const reserved = current && reserveSystemRecoveryAttempt(current);
|
|
if (current && reserved) await updateSystemRecovery(db, input, current,
|
|
failSystemRecovery(reserved, new SystemRecoveryFailure("copyback_stop_unconfirmed", "recheck_evidence")));
|
|
await recheckStoppedWorkspaceRecovery(db, input, false);
|
|
}
|
|
}
|
|
|
|
const cursors = new WeakMap<Db, string>();
|
|
const workInFlight = new WeakMap<Db, Set<string>>();
|
|
const unfinishedCursors = new WeakMap<Db, string>();
|
|
|
|
/** Destruction is a later cleanup obligation, authorized only by a matching
|
|
* saved-file receipt and the original recorded disposition. */
|
|
export async function scheduleRecoveredWorkspaceCleanup(db: Db, run: typeof heartbeatRuns.$inferSelect) {
|
|
if (!await hasVerifiedWorkspaceRepair(db, run)) return;
|
|
const ref = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync);
|
|
if (ref?.resourceDisposition !== "destroy") return;
|
|
const [lease] = await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, ref.leaseId)));
|
|
if (!lease || !hasRemoteTerminationReceipt(lease)) return;
|
|
const [other] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
|
ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider!), eq(environmentLeases.providerLeaseId, lease.providerLeaseId!),
|
|
)).limit(1);
|
|
if (other) return;
|
|
await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "ephemeral",
|
|
updatedAt: new Date(), metadata: sql`(${environmentLeases.metadata} - 'workspaceRestoreRecovery' - 'remoteExecutionTermination') ||
|
|
${JSON.stringify({ workspaceRecoveryCleanup: { runId: run.id, descriptorSha256: ref.descriptorSha256 },
|
|
pendingCleanupInFlight: false, pendingCleanupRetryAfterMs: 0, pendingCleanupRetryAttempts: 0, pendingCleanupLeaseExpiresAtMs: 0 })}::jsonb`,
|
|
}).where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, ref.leaseId),
|
|
eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.providerLeaseId, ref.providerLeaseId),
|
|
eq(environmentLeases.status, "released"), eq(environmentLeases.cleanupStatus, "success"),
|
|
sql`${environmentLeases.metadata} = ${JSON.stringify(lease.metadata)}::jsonb`,
|
|
sql`${environmentLeases.metadata}->>'workspaceRecoveryCleanup' is null`)), { companyId: run.companyId, runId: run.id });
|
|
}
|
|
export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedOnly = false, obligationsOnly = false, inFlightRunIds: string[] = []) {
|
|
const cleanup = sql`(${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,resourceDisposition}' = 'destroy' and exists (
|
|
select 1 from ${environmentLeases} where ${environmentLeases.companyId} = ${heartbeatRuns.companyId}
|
|
and ${environmentLeases.heartbeatRunId} = ${heartbeatRuns.id}
|
|
and ${environmentLeases.id}::text = ${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,leaseId}'
|
|
and ${environmentLeases.status} = 'released' and ${environmentLeases.cleanupStatus} = 'success'
|
|
and ${environmentLeases.metadata}->'workspaceRecoveryCleanup' is null))`;
|
|
const waitingInput = sql`exists (select 1 from ${agentWakeupRequests}
|
|
where ${agentWakeupRequests.companyId} = ${issueRecoveryActions.companyId}
|
|
and ${agentWakeupRequests.status} = 'deferred_issue_execution' and ${agentWakeupRequests.requestedByActorType} = 'user'
|
|
and ${agentWakeupRequests.payload}->>'issueId' = ${issueRecoveryActions.sourceIssueId}::text)`;
|
|
return db.select({ action: issueRecoveryActions, run: heartbeatRuns }).from(issueRecoveryActions)
|
|
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, issueRecoveryActions.companyId), eq(heartbeatRuns.id,
|
|
sql`case when ${issueRecoveryActions.evidence}->>'runId' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
|
|
then (${issueRecoveryActions.evidence}->>'runId')::uuid end`)))
|
|
.where(and(inArray(heartbeatRuns.status, terminal),
|
|
workspaceRecoverySourceEligible(true),
|
|
unfinishedOnly ? sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')` : undefined,
|
|
obligationsOnly ? sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')
|
|
or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete' and ${cleanup}) or ${waitingInput})` : undefined,
|
|
sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence', 'complete', 'exhausted')
|
|
and (${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' = 'null'::jsonb
|
|
or ${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' <= to_jsonb(${Date.now()}::bigint)))`,
|
|
sql`(${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' is distinct from 'recovered' or
|
|
${cleanup} or ${waitingInput})`,
|
|
cursor ? gt(issueRecoveryActions.id, cursor) : undefined,
|
|
inFlightRunIds.length ? notInArray(heartbeatRuns.id, inFlightRunIds) : undefined,
|
|
)).orderBy(asc(issueRecoveryActions.id)).limit(5);
|
|
}
|
|
|
|
/** Bounded discovery shares the periodic recovery lifecycle. A repaired source
|
|
* still owes cleanup even without saved input or after task reconciliation. */
|
|
export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRuntimeService,
|
|
onRecovered: (run: typeof heartbeatRuns.$inferSelect) => Promise<void>, obligationsOnly = false,
|
|
onFinished?: (continueSweep: boolean) => void, passiveSweep?: { cursor?: string }) {
|
|
await adoptHistoricalWorkspaceRecoveries(db);
|
|
let pending = workInFlight.get(db);
|
|
if (!pending) { pending = new Set(); workInFlight.set(db, pending); }
|
|
if (pending.size >= 2) return;
|
|
// Locally running exports already own their slot. Exclude them before
|
|
// pagination so a page containing only held work cannot prevent wrapping.
|
|
const inFlightRunIds = [...pending];
|
|
let rows = await retainedWorkspaceCandidates(db, unfinishedCursors.get(db), true, false, inFlightRunIds);
|
|
if (!rows.length) rows = await retainedWorkspaceCandidates(db, undefined, true, false, inFlightRunIds);
|
|
const unfinished = rows.length > 0;
|
|
if (!rows.length) rows = await retainedWorkspaceCandidates(db, passiveSweep ? passiveSweep.cursor : cursors.get(db), false, obligationsOnly, inFlightRunIds);
|
|
if (!rows.length && !passiveSweep) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds);
|
|
const work: Promise<void>[] = [];
|
|
let hasUnvisited = false;
|
|
for (const { action, run } of rows) {
|
|
if (pending.size >= 2) { hasUnvisited = true; break; }
|
|
if (unfinished) unfinishedCursors.set(db, action.id);
|
|
else if (passiveSweep) passiveSweep.cursor = action.id;
|
|
else cursors.set(db, action.id);
|
|
if (pending.has(run.id)) continue;
|
|
pending.add(run.id);
|
|
const finish = beginIdleTrackedWork();
|
|
const physical = ["scheduled", "running", "awaiting_evidence"].includes(readSystemRecovery(action.evidence)?.state ?? "");
|
|
work.push((async () => { try {
|
|
await recoverRetainedWorkspace({ db, environmentRuntime: runtime, companyId: action.companyId, issueId: action.sourceIssueId, runId: run.id, actionId: action.id });
|
|
const [updated] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id));
|
|
const recovered = readWorkspaceRecovery(updated?.evidence ?? null)?.status === "recovered";
|
|
const exhausted = readSystemRecovery(updated?.evidence)?.state === "exhausted";
|
|
if (recovered || exhausted) {
|
|
const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
|
if (current) {
|
|
if (recovered) {
|
|
await resumePreservedResultFinalization(db, current.companyId, current.id);
|
|
await scheduleRecoveredWorkspaceCleanup(db, current);
|
|
}
|
|
// Only normal dispatch can consume existing authorized input. An
|
|
// exhausted export permits evaluation of an isolated continuation,
|
|
// never another export attempt or a replay of an old assignment.
|
|
await onRecovered(current);
|
|
}
|
|
}
|
|
} catch { logger.warn({ runId: run.id }, "Retained workspace remains pending"); }
|
|
finally { pending.delete(run.id); finish(); onFinished?.(physical || hasUnvisited); } })());
|
|
}
|
|
return work;
|
|
}
|