diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index daf4bba6ad9..6c6a28cfc13 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1223,6 +1223,12 @@ An explicit recovery action is a typed liveness repair path for a source issue. A terminal native failure can retain a result accepted before checkpoint or cleanup failed. That result is historical evidence, not a live controller. New user input may start a fresh turn after the controller and execution environment have stopped and ordinary admission checks pass. Preserve the failed run, its result, and its recovery budget. Do not commit the old result, infer action outcomes, or replay the failed turn. A message saved while cleanup is pending must be reconsidered after verified cleanup and delivered once. External chat messages retain their original inbound action, wake receipt and actor. Cleanup delegates reconsideration to the live inbound owner, which revalidates the endpoint, conversation generation, principal and exact saved source before normal admission. The existing stranded-queue sweep uses the same owner after a lost callback or restart; it cannot substitute a generic assignment wake. A rejected source must not prevent other saved messages from being reconsidered. Admission must consume its deferred receipt in the same transaction that creates the new run, including in agent chat, where later messages keep their separate turns. Completing that run must not promote the consumed receipt again. Workspace-export repair retains its separate saved-result recovery path. +A host-verified workspace copy-back receipt also permits new user input after +a native coordinator reaches terminal failure with a retained result. The receipt +must bind the company, task, run, allocation, descriptor and final file digest. +The original accepted result stays unchanged. A coordinator that is still +committing, an active owner, or an unverified stop keeps its existing hold. + When a legacy turn admitted by an explicit user message fails or times out, a bounded transient retry needs its own durable continuation authorization. Initial admission binds the message's exact body digest and revision. Scheduling requires that original binding and revalidates the human author, source task, assignee, and execution ownership, then records the successor's receipt in the same transaction as its run. Historical receipts without a message binding cannot authorize automatic retries and are not backfilled. The parent receipt remains intact. Dispatch checks the exact successor, retry parent, and unchanged message binding again. Deleted or edited input, changed actors or ownership, superseded work, and duplicate successors cannot renew authority. One-run Retry and Interrupt intents retain their separate admission contracts; they are not renewable user-message authority. Other retry policies cannot borrow this receipt: their scheduling fails closed. Successful explicit turns do not create a prose-only missing-comment follow-up with spent authority. Explicit retry admission checks the local parent adapter controller and verified environment cleanup. The existing queue-first release policy may retain only its own terminal execution claim while that cleanup is pending; finalization and the periodic stale-lock sweep reconsider the same policy after cleanup. Generic terminal-lock cleanup and stale checkout adoption leave these claims for that settlement. New user input, reassignment, and a successor's claim take precedence. Exhausted retries and revoked or missing authorization release the terminal claim without requesting another retry. Pending or failed environment cleanup cannot mint a successor receipt. diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 95e7bc52985..e86297de002 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -18,7 +18,7 @@ import { beforeAll, afterAll, describe, it, expect, vi } from "vitest"; import { applyPendingMigrations, approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections, agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, - issues, nativeRunFinalizations, nativeRunResults, completionContracts, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers, type Db, + issues, nativeRunFinalizations, nativeRunResults, completionContracts, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers, workspaceOperations, type Db, } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; @@ -26,6 +26,7 @@ import { admitExplicitNativeContinuation } from "./explicit-native-continuation. import { adapterExecutionControls, createAdapterExecutionControl } from "./adapter-execution-control.js"; import { CONVERSATION_CONTINUATION_POLICY } from "./conversation-continuation.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; +import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnvironmentRuntime } from "./heartbeat.js"; import { getExecutionBlocker } from "./execution-blocker.js"; import { createDurableChatWakeupRequest } from "./durable-chat-wakeup.js"; @@ -2260,6 +2261,141 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes await settings.updateExperimental({ enableAgentChat: experimental.enableAgentChat }); } }); + async function recoveredCopybackFixture(origin: "message" | "queued_message" | "task_retry") { + const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f); + const requestId = randomUUID(), operationId = randomUUID(); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId, + provider: "daytona", providerLeaseId: randomUUID() }; + const reference = { schema: "paperclip.native-workspace-sync/v3", state: "finalized", workspaceId: randomUUID(), + leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/workspaces/task", + descriptorSha256: "12".repeat(32), baselineSha256: "34".repeat(32), finalHostSha256: "56".repeat(32), + resourceDisposition: "destroy" }; + const recovery = { ...createSystemRecovery(f.sourceRunId, origin === "task_retry" ? requestId : null), + state: "complete" as const, preservationOperationIds: [operationId] }; + await db.update(issueRecoveryActions).set({ status: "cancelled", cause: "native_workspace_sync_out_retry_exhausted", + evidence: { runId: f.sourceRunId, workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", + runId: f.sourceRunId, status: "recovered", stage: "saving_receipts", attempt: 1, requestIds: [] }, + systemRecovery: recovery, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } }, + }).where(eq(issueRecoveryActions.id, action.id)); + await db.update(heartbeatRuns).set({ processPid: null, createdAt: new Date("2020-01-01T09:00:00Z"), + finishedAt: new Date("2020-01-01T10:00:00Z"), runnerProfileJson: { nativeWorkspaceSync: reference } }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueComments).set({ createdAt: new Date("2020-01-01T13:00:00Z") }).where(eq(issueComments.id, f.commentId)); + await db.insert(environmentLeases).values({ ...lease, status: "expired", leasePolicy: "ephemeral", + releasedAt: new Date(), cleanupStatus: "success", metadata: { + remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }), + }, + }); + await db.insert(workspaceOperations).values({ id: operationId, companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.sourceRunId, phase: "workspace_recovery", status: "succeeded", metadata: { + recoveryActionId: action.id, leaseId: lease.id, providerLeaseId: lease.providerLeaseId, + descriptorSha256: reference.descriptorSha256, finalHostSha256: reference.finalHostSha256, + }, + }); + if (origin !== "message") await db.insert(agentWakeupRequests).values({ id: requestId, companyId: f.companyId, agentId: f.agentId, + source: "on_demand", triggerDetail: "manual", status: "deferred_issue_execution", + reason: origin === "task_retry" ? "task_retry" : "issue_commented", requestedByActorType: "user", requestedByActorId: f.actorId, + payload: { issueId: f.issueId, ...(origin === "task_retry" ? { taskRetry: { schema: "paperclip.task-retry.v1", + generationId: requestId, sourceRunId: f.sourceRunId, actorId: f.actorId, actorSource: "session", strategy: "evaluate" } } + : { commentId: f.commentId, executionWait: { reason: "finalizing_saved_result" }, + _paperclipWakeContext: { source: "issue.comment", issueId: f.issueId, wakeReason: "issue_commented", + wakeCommentId: f.commentId, wakeCommentIds: [f.commentId] } }) }, + }); + return { ...f, resultId, action, lease, reference, operationId, requestId, recovery, + ...(origin === "task_retry" ? { taskRetryRequestId: requestId } + : origin === "queued_message" ? { queuedCommentRequestId: requestId } : {}) }; + } + it.each(["message", "queued_message", "task_retry"] as const)("continues a terminal retained result after verified copy-back: %s", async origin => { + const f = await recoveredCopybackFixture(origin); + const [before] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await verifiedWorkspaceRepairReceipt(db, source)).toBe(f.operationId); + const result = await admit(f); + expect(result).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId }); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(before); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.action.id)); + expect(action.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown", replay: "explicit_user_continuation" }); + expect(action.evidence.systemRecovery).toEqual(f.recovery); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0]) + .toMatchObject({ phase: "terminal_failure", resultId: f.resultId }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + }); + + it.each(["missing", "descriptor", "files", "lease", "provider", "source", "issue", "unfinished", "unrecovered", + "committing", "controller", "remote_stop", "active_run"] as const)("keeps retained-result continuation held without exact settled copy-back: %s", async fault => { + const f = await recoveredCopybackFixture("queued_message"); + const [before] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)); + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, f.operationId)); + if (fault === "missing") await db.delete(workspaceOperations).where(eq(workspaceOperations.id, f.operationId)); + const changed = fault === "descriptor" ? { descriptorSha256: "78".repeat(32) } + : fault === "files" ? { finalHostSha256: "78".repeat(32) } + : fault === "lease" ? { leaseId: randomUUID() } + : fault === "provider" ? { providerLeaseId: randomUUID() } : null; + if (changed) await db.update(workspaceOperations).set({ metadata: { ...operation.metadata, ...changed } }) + .where(eq(workspaceOperations.id, f.operationId)); + if (fault === "source") { + const [other] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "failed" }).returning(); + await db.update(workspaceOperations).set({ heartbeatRunId: other.id }).where(eq(workspaceOperations.id, f.operationId)); + } + if (fault === "issue") { + const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task" }).returning(); + await db.update(workspaceOperations).set({ issueId: other.id }).where(eq(workspaceOperations.id, f.operationId)); + } + if (fault === "unfinished") await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.id, f.operationId)); + if (fault === "unrecovered") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, + runId: f.sourceRunId, systemRecovery: f.recovery, workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", + runId: f.sourceRunId, status: "blocked", stage: "restoring_files", attempt: 1, requestIds: [] }, + } }).where(eq(issueRecoveryActions.id, f.action.id)); + if (fault === "committing") await db.update(nativeRunFinalizations).set({ phase: "result_accepted" }) + .where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + if (fault === "controller") await db.update(nativeRunFinalizations).set({ leaseOwner: "live-controller" }) + .where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + if (fault === "remote_stop") await db.update(environmentLeases).set({ metadata: {} }).where(eq(environmentLeases.id, f.lease.id)); + if (fault === "active_run") await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "running" }); + expect(await admit(f)).toBeNull(); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(before); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.action.id)))[0].evidence) + .not.toHaveProperty("explicitUserContinuation"); + }); + + it("delivers the saved message once after terminal result copy-back, without replaying its accepted result", async () => { + const f = await recoveredCopybackFixture("queued_message"); + const failedRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: failedRunId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "failed", errorCode: "native_provider_terminal_failed", + createdAt: new Date("2020-01-01T11:00:00Z"), finishedAt: new Date("2020-01-01T12:00:00Z"), contextSnapshot: { issueId: f.issueId } }); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId: failedRunId, + phase: "terminal_failure" }); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", fingerprint: failedRunId, + status: "resolved", outcome: "blocked", nextAction: "Continue from recorded work without replay.", + evidence: { runId: failedRunId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }); + const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: failedRunId, + provider: "daytona", providerLeaseId: randomUUID() }; + await db.insert(environmentLeases).values({ ...lease, status: "expired", leasePolicy: "ephemeral", + releasedAt: new Date(), cleanupStatus: "success", metadata: { + remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }), + } }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, f.requestId)); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeExecutionWaitComments(), heartbeat.resumeExecutionWaitComments()]); + const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.requestId)); + expect(delivered).toMatchObject({ status: "coalesced", runId: expect.any(String) }); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ id: delivered.runId, contextSnapshot: { + previousRunId: failedRunId, forceFreshSession: true, wakeCommentIds: [f.commentId], + } }); + await heartbeat.resumeExecutionWaitComments(); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))) + .toHaveLength(1); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0]) + .toMatchObject({ phase: "terminal_failure", resultId: f.resultId }); + }); + it("defers an ordinary follow-up while a recovered export's successful result is still finalizing", async () => { const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f); await db.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, failureDetail: null }) diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index ddf3ae1c46a..bedd90e3c3c 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -370,7 +370,9 @@ export async function admitExplicitNativeContinuation(input: { const snapshotSettled = completedResult && freshIntent && action.id && recovery?.state === "complete" && await verifiedWorkspaceSnapshotReceipt(db, { companyId, issueId, runId: run.id, actionId: action.id, operationIds: recovery.preservationOperationIds }); - if (completedResult && !(freshIntent && snapshotSettled)) + const repairedTerminal = completedResult && coordinator.phase === "terminal_failure" + && await hasVerifiedWorkspaceRepair(db, run); + if (completedResult && !repairedTerminal && !(freshIntent && snapshotSettled)) return blocked("finalizing_saved_result", "Paperclip is finishing the saved result."); } // A repaired export no longer blocks execution. Its machine receipt still