diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b5b2af1a789..325148106d2 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -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 diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 810c4d5939d..8467918f2ec 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -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; type Coordinator = typeof nativeRunFinalizations.$inferSelect; +export async function isFailedNativePreparation(db: Db, run: Run, coordinator: Coordinator | undefined) { + const recovery = run.resultJson?.executionRecovery as Record | 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 { 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); } diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index c93ec6b8681..b45214afd1c 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -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)) || diff --git a/server/src/services/fresh-workspace-continuation.ts b/server/src/services/fresh-workspace-continuation.ts index 25ac8556975..6789a12b928 100644 --- a/server/src/services/fresh-workspace-continuation.ts +++ b/server/src/services/fresh-workspace-continuation.ts @@ -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>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId')`, + runnerProfileJson: sql>`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) { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index d6902a99374..d11eee87ec9 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -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 }, diff --git a/server/src/services/legacy-workspace-restore-recovery.ts b/server/src/services/legacy-workspace-restore-recovery.ts index 47ce54f6f84..45e451b0cc3 100644 --- a/server/src/services/legacy-workspace-restore-recovery.ts +++ b/server/src/services/legacy-workspace-restore-recovery.ts @@ -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. diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index a3e0216da61..8683cac4dc4 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -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 | 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>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId')`, + runnerProfileJson: sql>`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, options: { allowPreterminal?: boolean; requireCancelledStop?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) { diff --git a/server/src/services/native-runtime/native-workspace-finalization-state.ts b/server/src/services/native-runtime/native-workspace-finalization-state.ts index 3701c1fe0b6..00b783fe21c 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-state.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-state.ts @@ -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; - -/** 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`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, - ${heartbeatRuns.contextSnapshot}->>'issueId')`, - reference: sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`, - legacyReference: sql`${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 diff --git a/server/src/services/remote-workspace-lifecycle.ts b/server/src/services/remote-workspace-lifecycle.ts index 7896d5abada..aa4d4097086 100644 --- a/server/src/services/remote-workspace-lifecycle.ts +++ b/server/src/services/remote-workspace-lifecycle.ts @@ -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; @@ -9,6 +9,44 @@ export type DestroyRemoteWorkspaceLease = (input: { companyId: string; workspaceId: string; leaseId: string; environmentId: string; providerLeaseId: string; }) => Promise; +export type WorkspaceSourceLease = Pick; + +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`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, + ${heartbeatRuns.contextSnapshot}->>'issueId')`, + reference: sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`, + legacyReference: sql`${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) { diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index d760071806b..363663258be 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -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>; @@ -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)); diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 71b6b12c292..d1fb8fd9b6e 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -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. diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index a086715dbc8..3768ae1528b 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -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; + 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") {