diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 03845bd8fef..5eb3248d65e 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2063,6 +2063,21 @@ ordinary Retry can continue. Finalized workspaces, mismatched sources and accepted results keep their existing settlement paths. Reassignment does not discard the stopped run's files or grant authority over the new assignee. +When the task was reassigned, that file-preservation hold binds the new owner +only until automatic recovery records a completed receipt for that hold. Every +saved operation in the receipt must be a copyback into the task workspace. The +receipt is durable, so later reuse of the workspace by the new owner cannot +apply the hold again. A new owner's turn is a fresh assignment, not a replay of +the stopped run. After the receipt, admission, the final dispatch gate and +stranded-work recovery no longer apply the hold to that owner. Wakes that the +gate cancelled during the copy-back therefore get one bounded continuation. + +The hold row and its no-replay evidence stay unchanged. The hold still applies +when the task returns to the stopped run's agent. It also applies when that agent +is the current review participant. Before the receipt, the hold prevents a second +executor on the workspace. A snapshot-only save does not release the hold, +because the task workspace does not contain the saved files. + The deadline is a backstop, not the normal failure path. When native task preparation fails after a fresh controller claim and before any provider turn, the controller stops renewal and releases only that exact claim. Ordinary diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 53850101e4f..81299d09629 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -4,7 +4,7 @@ import { preDispatchExecutionWaitRecoveryActionPredicate } from "./pre-dispatch- import { and, desc, eq, gt, inArray, not, or, sql } from "drizzle-orm"; import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js"; import { z } from "zod"; -import { agentWakeupRequests, chatConversations, chatEndpoints, environmentLeases, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolConnections, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, chatConversations, chatEndpoints, environmentLeases, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolConnections, workspaceOperations, type Db } from "@paperclipai/db"; import { canContinueCancelledRun, readRunCancellation } from "./run-cancellation.js"; import { canRetryStoppedRun } from "./cancelled-native-startup.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; @@ -13,7 +13,7 @@ import { hasRequiredWorkspaceRecovery, LEGACY_WORKSPACE_RECOVERY_SCHEMA, nativeR import { unreconciledNativeExportPredicate } from "./native-runtime/native-workspace-reconstruction.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; -import { hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; +import { completedWorkspaceRecoveryCondition, hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; import { nativeManualAdmissionHoldsExecution } from "./native-runtime/native-runner-ownership.js"; const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution", @@ -126,6 +126,45 @@ export function nativeExecutionRecoveryActionPredicate() { and ${heartbeatRuns.runtimeMode} = 'native')`; } +/** + * A file-preservation hold binds a newer task owner only until automatic + * recovery saved the stopped run's files. That owner's turn is a fresh + * assignment, not a replay: the stopped run's unknown outcomes stay with its + * own agent, which keeps the hold. The proof is the durable completed-recovery + * receipt bound to this action, so later reuse of the workspace by the new + * owner (new epoch, writer or snapshot) cannot re-arm the hold. + */ +export function restoreHoldReleasedForCurrentOwnerPredicate() { + const sourceRunId = sql`coalesce(${issueRecoveryActions.evidence}->>'runId', ${issueRecoveryActions.evidence}->>'sourceRunId')`; + const operationIds = sql`case when jsonb_typeof(${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}') = 'array' + then ${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}' else '[]'::jsonb end`; + // JSON reads are NULL for other holds. Callers negate this predicate, so it + // must be strictly true or false: NOT NULL would hide unrelated blockers. + return sql`coalesce((${and( + sql`${issueRecoveryActions.evidence}->'workspaceRestoreRecovery'->>'schema' = ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}`, + sql`coalesce(${issueRecoveryActions.evidence}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive'`, + sql`exists (select 1 from ${heartbeatRuns} stopped + join ${issues} task on task.company_id = stopped.company_id and task.id = ${issueRecoveryActions.sourceIssueId} + where stopped.company_id = ${issueRecoveryActions.companyId} + and stopped.id::text = ${sourceRunId} + and stopped.status in ('failed', 'timed_out', 'interrupted', 'cancelled') + and stopped.finished_at is not null + and task.assignee_agent_id is not null + and task.assignee_agent_id <> stopped.agent_id + and not coalesce(task.status = 'in_review' + and task.execution_state->>'status' = 'pending' + and task.execution_state->'currentParticipant'->>'type' = 'agent' + and task.execution_state->'currentParticipant'->>'agentId' = stopped.agent_id::text, false))`, + completedWorkspaceRecoveryCondition(sourceRunId), + // Copyback put the files back into the task workspace that the new owner + // reuses. A snapshot-only save needs an explicit fresh-workspace request + // to attach it, so that hold stays until such a request or a repair. + sql`not exists (select 1 from jsonb_array_elements_text(${operationIds}) saved(id) + join ${workspaceOperations} op on op.id::text = saved.id + where op.company_id = ${issueRecoveryActions.companyId} and op.phase <> 'workspace_recovery')`, + )}), false)`; +} + export async function getExecutionBlocker(db: Db, companyId: string, issueId: string, options?: { conversationResetCommentId?: string | null }): Promise { const [conversation] = await db.select({ agentId: issues.conversationAgentId, boundaryId: issues.conversationBoundaryCommentId, status: issues.status, @@ -139,6 +178,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), executionBlockerPredicate(), + not(restoreHoldReleasedForCurrentOwnerPredicate()), or( unreconciledNativeExportPredicate(), sql`${issueRecoveryActions.evidence}->>'workspaceRestoreFailure' = 'restore_unsafe_archive'`, @@ -166,6 +206,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), executionBlockerPredicate(), + not(restoreHoldReleasedForCurrentOwnerPredicate()), boundary ? or(gt(issueRecoveryActions.createdAt, boundary.createdAt), nativeExecutionRecoveryActionPredicate()) : undefined, )).orderBy(desc(issueRecoveryActions.updatedAt), desc(issueRecoveryActions.id)).limit(1); if (!action) return null; diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 1bcda896860..7593411ffb5 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -19,13 +19,13 @@ import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recorde import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; import { z } from "zod"; -import { and, desc, eq, gt, inArray, isNull, ne, or, sql } from "drizzle-orm"; +import { and, desc, eq, gt, inArray, isNull, ne, not, or, sql } from "drizzle-orm"; import { agents, agentWakeupRequests, approvals, issueApprovals, issueThreadInteractions, environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db, } from "@paperclipai/db"; -import { executionBlockerPredicate, getExecutionBlocker, hasPendingRunCleanup, nativeExecutionRecoveryActionPredicate } from "./execution-blocker.js"; +import { executionBlockerPredicate, getExecutionBlocker, hasPendingRunCleanup, nativeExecutionRecoveryActionPredicate, restoreHoldReleasedForCurrentOwnerPredicate } from "./execution-blocker.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; import { adapterExecutionControls } from "./adapter-execution-control.js"; import { persistActivity } from "./activity-log.js"; @@ -227,6 +227,7 @@ export async function admitExplicitNativeContinuation(input: { eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), boundary ? or(gt(issueRecoveryActions.createdAt, boundary.createdAt), nativeExecutionRecoveryActionPredicate(), blocker?.recoveryActionId ? eq(issueRecoveryActions.id, blocker.recoveryActionId) : undefined) : undefined, + not(restoreHoldReleasedForCurrentOwnerPredicate()), or(executionBlockerPredicate(), and( inArray(issueRecoveryActions.cause, ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]), sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' = 'recovered'`, diff --git a/server/src/services/workspace-recovery-receipt.ts b/server/src/services/workspace-recovery-receipt.ts index be0057ce357..33e44007914 100644 --- a/server/src/services/workspace-recovery-receipt.ts +++ b/server/src/services/workspace-recovery-receipt.ts @@ -1,4 +1,4 @@ -import { and, desc, eq, sql, type AnyColumn } from "drizzle-orm"; +import { and, desc, eq, sql, type AnyColumn, type SQL } from "drizzle-orm"; import { executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema, type ExecutionBlocker } from "@paperclipai/shared"; import { z } from "zod"; @@ -25,7 +25,7 @@ export function readWorkspaceRecovery(evidence: Record | null) return parsed.success ? parsed.data : null; } -export function completedWorkspaceRecoveryCondition(runId: string | AnyColumn) { +export function completedWorkspaceRecoveryCondition(runId: string | AnyColumn | SQL) { const operationIds = sql`case when jsonb_typeof(${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}') = 'array' then ${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}' else '[]'::jsonb end`; return and( diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 921cb2c4355..91d3ad20480 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -2970,6 +2970,91 @@ describe("automatic retained workspace recovery", () => { expect(wake.payload?.taskRetry).toMatchObject({ strategy: "continue" }); }, 30_000); + it("hands a reassigned task to its new assignee once the previous owner's files are recovered", async () => { + // Shape of a mid-run reassignment: the old owner's run was cancelled for + // the reassignment, its copyback was retained, and its no-replay hold was + // recorded after the new assignee's queued wakes had been admitted. + const f = await seed(); + const previousOwnerId = (await run(f.runId)).agentId; + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "issue_reassigned" }).where(eq(heartbeatRuns.id, f.runId)); + const held = await action(f.actionId); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "blocked", resolvedAt: new Date(), + fingerprint: `native-workspace:${f.runId}`, evidence: { ...held.evidence, continuationDelivery: "invalidated", + automaticRecovery: { policy: "preserve_without_replay_v1", runId: f.runId, replay: "blocked", actionOutcome: "unknown" } }, + }).where(eq(issueRecoveryActions.id, f.actionId)); + const newOwnerId = randomUUID(); + await db.insert(agents).values({ id: newOwnerId, companyId: f.companyId, name: "Reviewer", adapterType: "paperclip_runner", + status: "idle", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); + await db.update(issues).set({ assigneeAgentId: newOwnerId, status: "in_progress" }).where(eq(issues.id, f.issueId)); + // The new owner's queued wake reached the final dispatch gate after the hold appeared. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: newOwnerId, runtimeMode: "legacy", + status: "cancelled", finishedAt: new Date(), invocationSource: "automation", errorCode: "execution_reconciliation_required", + contextSnapshot: { issueId: f.issueId }, resultJson: { stopReason: "execution_reconciliation_required", + timeoutSource: "stale_queued_run_gate", executionWait: { issueId: f.issueId, recoveryActionId: f.actionId } } }); + + // No second executor may touch the workspace while its files are unrecovered. + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.runId, workspaceRepairRequired: true }); + + await recoverRetainedWorkspace(f.input); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + // The recovered hold no longer binds the board-chosen new owner... + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + // A different reconciliation hold for the same stopped run, without file + // preservation evidence, still blocks the new owner. + const [other] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", ownerType: "board", cause: "native_continuation_requires_reconciliation", + fingerprint: `other:${f.runId}`, nextAction: "Reconcile the stopped run.", evidence: { runId: f.runId } }).returning(); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ recoveryActionId: other.id }); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, other.id)); + // ...and the stranded-work sweep gives that owner a durable next action. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: newOwnerId, status: "running" }); + const pending = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + inArray(heartbeatRuns.status, ["queued", "scheduled_retry"]))); + await heartbeatService(db).reconcileStrandedAssignedIssues({ companyId: f.companyId, issueId: f.issueId }); + const successors = await pending(); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ agentId: newOwnerId, contextSnapshot: { issueId: f.issueId } }); + // A fresh turn for the new owner, never a replay of the stopped run. + expect(successors[0].retryOfRunId).not.toBe(f.runId); + expect(successors[0].contextSnapshot?.previousRunId).not.toBe(f.runId); + // Recovery stays bounded: a later sweep observes the pending turn instead of adding another. + await heartbeatService(db).reconcileStrandedAssignedIssues({ companyId: f.companyId, issueId: f.issueId }); + expect((await pending()).map(item => item.id)).toEqual([successors[0].id]); + // The release rests on the durable completed-recovery receipt. Later reuse + // of the workspace by the new owner changes live workspace evidence, so the + // live repair check stops verifying; the hold must not re-arm for that owner. + const stopped = await run(f.runId); + const sync = stopped.runnerProfileJson?.nativeWorkspaceSync as Record; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...stopped.runnerProfileJson, + nativeWorkspaceSync: { ...sync, finalHostSha256: "f".repeat(64) } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + await db.update(heartbeatRuns).set({ runnerProfileJson: stopped.runnerProfileJson }).where(eq(heartbeatRuns.id, f.runId)); + // The audit row and its no-replay hold stay intact for the previous owner. + expect(await action(f.actionId)).toMatchObject({ status: "resolved", outcome: "blocked", + evidence: { automaticRecovery: { replay: "blocked" } } }); + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, successors[0].id)); + await db.update(issues).set({ assigneeAgentId: previousOwnerId }).where(eq(issues.id, f.issueId)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.runId, workspaceRepairRequired: false }); + }, 30_000); + + it("keeps a snapshot-only recovery held for a new assignee because the task workspace lacks the saved files", async () => { + const f = await seed("native", false); + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "issue_reassigned" }).where(eq(heartbeatRuns.id, f.runId)); + const held = await action(f.actionId); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "blocked", resolvedAt: new Date(), + evidence: { ...held.evidence, automaticRecovery: { policy: "preserve_without_replay_v1", runId: f.runId, replay: "blocked" } }, + }).where(eq(issueRecoveryActions.id, f.actionId)); + const newOwnerId = randomUUID(); + await db.insert(agents).values({ id: newOwnerId, companyId: f.companyId, name: "Reviewer", adapterType: "paperclip_runner", status: "idle" }); + await db.update(issues).set({ assigneeAgentId: newOwnerId, status: "in_progress" }).where(eq(issues.id, f.issueId)); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + expect(await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, f.runId), + eq(workspaceOperations.phase, "workspace_recovery_snapshot"), eq(workspaceOperations.status, "succeeded")))).toHaveLength(1); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ recoveryActionId: f.actionId, runId: f.runId }); + }, 30_000); + it("projects active preservation without waiting input as working, and stale observations as failed", async () => { const f = await seed(); const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));