diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 3f0ff6d81ab..eaa6e14e0a8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1726,6 +1726,54 @@ that thread is tracked in [#19047](https://github.com/paperclipai/paperclip/issu A permanently unusable native runner session may be replaced only with evidence that its predecessor is stopped and fenced, completed results and workspace state are preserved, required task history is available, and pending effects have been reconciled. A provider-native shell command or external write without a reliable outcome receipt is unknown. Unknown effects, integrity failures, and unverified process ownership never authorize speculative replay. Once automatic recovery is ruled out, Paperclip selects a conservative default: preserve recorded work, stop the affected task, and retain a durable no-replay hold. Unknown action outcomes remain unknown. No reconciliation form or user diagnosis is required. +Normal retained-file recovery can re-enter finalization of the same immutable +accepted result. It does not start another provider turn. A published external +response wait may retain its passive waiting disposition after the source run +becomes terminal. This requires complete verified preservation, stopped old +execution, the same result and completion contract, no newer run or current +execution, unchanged task ownership, and current source, channel, principal and +review binding authorization. Stop, pause, cancellation, changed review head and +revoked access retain their normal gates. The status committer rechecks the +preserved-result proof and current authority under its existing locks. +If recovery folds an already saved Retry into that same result, its history +records the preservation generation separately from the Retry request identity. +These identities do not grant execution or ancestry authority. New generation +proof must match exactly; an absent or foreign field cannot certify a folded +legacy Retry. Ordinary legacy history that already names the exact preservation +generation retains only its existing passive proof, without status ownership. + +An exhausted workspace export records the exact Blocked status version it +creates, bound to its source run, result, contract and identity. Recovery may +restore the accepted passive wait only while that receipt still owns the task +status. Missing historical receipts and later status decisions preserve the +current status. A missing receipt is a status-provenance limit, not evidence of +connector revocation. A yielded published review remains a wait; recovery never +converts it to completed work or queues another provider turn. +GitHub response maintenance recognizes this committed no-publication wait while +the task status remains unproven. It preserves the published assessment and +requires the same succeeded native run, yielded response-wake result and +server-committed presentation proof. Failed, missing or malformed presentation +does not qualify. In Review status selectors retain their separate status gate. + +The old export recovery notice settles separately after its accepted result, +preservation receipts, finalization, status effects and run projection settle +and old Runner ownership is absent. This closes only that source action. Newer +authorized work may own the task and workspace by then; closing the old notice +does not change that work, its status, execution owner, files or result. +Explicit export Retry uses its own settlement proof. Its unchanged request must +bind the same accepted result, source run and retained lease in both action and +run records. The saved source and successful native copyback barrier must agree. +A generic resume wake policy cannot substitute for either this proof or normal +Recover preservation receipts. Both paths wait for old effects and projection +to settle before closing the notice, without releasing physical resources. + +Recover transfers an accepted export's repair to the system without removing +the board's explicit export Retry path. A system-owned action must carry valid +recovery records for that same source run. Explicit export Retry still requires +the unchanged accepted result, exact stopped allocation, current task ownership +and the shared physical export lock. It cannot take over an active exporter or +start another provider turn. + An ACPX sidecar transport failure carries its redacted cause and code in the terminal event's structured `error` field. The session facade must retain that cause instead of reporting a failed turn with `error: null`. A transport failure diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 03c61c93c98..9ff1ae6b8eb 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -3084,7 +3084,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const [review] = await readChatGitHubReviewsByEndpointId(f.endpoint.id); expect(review.assessment?.summary).toBe(assessment.summary); }); - it("preserves the published GitHub assessment during passive-response maintenance", async () => { + it.each(["external_chat_response_waiting", "preserved_result_status_unproven"] as const)( + "preserves the published GitHub assessment during passive-response maintenance: %s", async reason => { const f = await publicationRetryFixture(); await f.service.execute(f.session, "submit_review", f.assessment, "published-before-passive-reply"); await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.run.id)); @@ -3102,13 +3103,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const [next] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, status: "running", runtimeMode: "native", contextSnapshot: await chatWakeContext({ endpointId: f.endpoint.id, issueId: f.conversation.issueId, provider: "github", providerMessageId: "41802", - }), resultJson: { finalizationPhase: "committed", finalizationReasonCode: "external_chat_response_waiting", + }), resultJson: { finalizationPhase: "committed", finalizationReasonCode: reason, nativeResult: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", continuation: { kind: "response_wake", idempotencyKey: "passive-author-reply" } }, presentationDecision: { schema: "paperclip.run_presentation_decision.v1", resolverVersion: "2", activityDisposition: "collapse", chosenSource: "none", commentAction: "none", commentId: null, reasonCodes: ["yielded_control_plane_wait"] } } }).returning(); - await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, f.conversation.issueId!)); + const expectedStatus = reason === "preserved_result_status_unproven" ? "blocked" : "in_review"; + await db.update(issues).set({ status: expectedStatus }).where(eq(issues.id, f.conversation.issueId!)); await responses.processPending(); expect(f.responseComments.get(originalId)!.body).toBe(originalBody); expect(f.responseWrites).toHaveLength(writesBefore); @@ -3116,6 +3118,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await responses.processPending(); await responses.processPending(); expect(f.responseComments.get(originalId)!.body).toBe(originalBody); + expect(f.responseWrites).toHaveLength(writesBefore); + expect((await db.select({ status: issues.status }).from(issues).where(eq(issues.id, f.conversation.issueId!)))[0].status) + .toBe(expectedStatus); expect(await readChatGitHubReviewsByEndpointId(f.endpoint.id)).toEqual(reviewBefore); expect([...f.responseComments.keys()]).toEqual([originalId]); expect(await githubRunReplyState(db, { ...f.session, runId: next.id, endpointId: f.endpoint.id })).toBe("none"); @@ -3134,8 +3139,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { body: "This turn ended without publishing a final response." })).rejects.toMatchObject({ status: 403 }); expect(f.responseWrites).toHaveLength(writes); }); - it.each(["completed", "failed", "missing_decision", "comment_requested", "missing_semantic", "legacy", "uncommitted_native", "pending_publication", "missing_decision_then_missing", "uncommitted_then_failed", "malformed_null", "malformed_scalar", "malformed_schema", "malformed_action"] as const)( - "keeps GitHub response maintenance truthful for a new request: %s", async scenario => { + it.each((["completed", "failed", "missing_decision", "comment_requested", "missing_semantic", "legacy", "uncommitted_native", "pending_publication", "missing_decision_then_missing", "uncommitted_then_failed", "malformed_null", "malformed_scalar", "malformed_schema", "malformed_action"] as const) + .flatMap(scenario => ["external_chat_response_waiting", "preserved_result_status_unproven"].map(reason => ({ scenario, reason }))))( + "keeps GitHub response maintenance truthful for a new request: $scenario ($reason)", async ({ scenario, reason }) => { const pendingPresentation = scenario.startsWith("missing_decision") || scenario.startsWith("uncommitted") || scenario.startsWith("malformed"); const f = await publicationRetryFixture(); await f.service.execute(f.session, "submit_review", f.assessment, "review-before-new-request"); @@ -3149,7 +3155,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { status: scenario === "failed" ? "failed" : "succeeded", runtimeMode: scenario === "legacy" ? "legacy" : "native", nativePhase: scenario.startsWith("uncommitted") ? "observed" : "committed", contextSnapshot: await chatWakeContext({ endpointId: f.endpoint.id, issueId: f.conversation.issueId, provider: "github", providerMessageId: "41802", - }), resultJson: { finalizationPhase: "committed", finalizationReasonCode: "external_chat_response_waiting", + }), resultJson: { finalizationPhase: "committed", finalizationReasonCode: reason, ...(scenario === "missing_semantic" ? {} : { nativeResult: { schema: "paperclip.run_result.v1", reportedWorkDisposition: scenario === "completed" ? "completed" : "yielded", continuation: { kind: "response_wake", idempotencyKey: "new-request-wait" } } }), diff --git a/server/src/services/chat-channels/github/response-run-state.ts b/server/src/services/chat-channels/github/response-run-state.ts index 2568c73d366..e0d4bc23669 100644 --- a/server/src/services/chat-channels/github/response-run-state.ts +++ b/server/src/services/chat-channels/github/response-run-state.ts @@ -29,7 +29,8 @@ export const responseRunColumns = { and ${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.nativePhase} = 'committed' and ${heartbeatRuns.resultJson}->>'finalizationPhase' = 'committed' - and ${heartbeatRuns.resultJson}->>'finalizationReasonCode' = 'external_chat_response_waiting' + and ${heartbeatRuns.resultJson}->>'finalizationReasonCode' in + ('external_chat_response_waiting', 'preserved_result_status_unproven') and ${heartbeatRuns.resultJson}->'nativeResult'->>'schema' = 'paperclip.run_result.v1' and ${heartbeatRuns.resultJson}->'nativeResult'->>'reportedWorkDisposition' = 'yielded' and ${heartbeatRuns.resultJson}->'nativeResult'->'continuation'->>'kind' = 'response_wake' diff --git a/server/src/services/native-runtime/chat-attachment-reuse.ts b/server/src/services/native-runtime/chat-attachment-reuse.ts index dfc7343168b..e4be5acb926 100644 --- a/server/src/services/native-runtime/chat-attachment-reuse.ts +++ b/server/src/services/native-runtime/chat-attachment-reuse.ts @@ -38,6 +38,7 @@ import { readInheritedGitHubReviewWait } from "../chat-channels/github/review-wa import { boundExternalChatProvider } from "./external-chat-provider.js"; import { resolveExternalChatQuestionResponse } from "./external-chat-question-response.js"; import { githubRunReplyState } from "../chat-run-publications.js"; +import { readPreservedNativeResultFinalization } from "./preserved-native-result-finalization.js"; // This boundary reads identity and wake authority, never launch renderings, // saved provider profiles, or run output. Keep the nested wake proof narrow too. @@ -754,12 +755,14 @@ export async function resolveExternalChatResponseWaitAuthorizationInTransaction( : lockMode === "nonblocking" ? actorQuery.for("update", { noWait: true }).limit(1) : actorQuery.for("update").limit(1)); + const preserved = run && run.status !== "running" + ? await readPreservedNativeResultFinalization(tx, binding) : null; if ( !issue || !actor || - run.status !== "running" || + (!preserved && run.status !== "running") || issue.assigneeAgentId !== binding.agentId || - issue.executionRunId !== binding.runId || + (!preserved && issue.executionRunId !== binding.runId) || ["paused", "terminated", "pending_approval", "error"].includes( actor.status, ) @@ -879,7 +882,7 @@ export async function readPublishedGitHubReviewCompletionContinuationInTransacti eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId), eq(heartbeatRuns.runtimeMode, "native"), )).limit(1); - if (!run || run.status !== "running") return null; + if (!run || run.status !== "running" && !await readPreservedNativeResultFinalization(tx, binding)) return null; const wait = await readInheritedGitHubReviewWait(tx, run); if (wait?.kind !== "published") return null; if (wait.reviewRunId !== run.id) { diff --git a/server/src/services/native-runtime/native-recovery-notices.ts b/server/src/services/native-runtime/native-recovery-notices.ts index defba4b33f8..65248a5d0fe 100644 --- a/server/src/services/native-runtime/native-recovery-notices.ts +++ b/server/src/services/native-runtime/native-recovery-notices.ts @@ -1,18 +1,71 @@ import { and, asc, eq, gt, inArray, isNotNull, not, or, sql } from "drizzle-orm"; -import { heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentLeases, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import type { NativeRecoveryNoticeScope } from "../native-recovery-notice-notifications.js"; import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js"; import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; import { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; +import { completedWorkspaceRecoveryCondition } from "../workspace-recovery-receipt.js"; + +const workspaceExportFailureCauses = ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]; + +function recoveredExportNotice(runId: string | typeof nativeRunFinalizations.runId) { + return and( + inArray(issueRecoveryActions.cause, workspaceExportFailureCauses), + completedWorkspaceRecoveryCondition(runId), + ); +} + +function completedExportRetryNotice(runId: string | typeof nativeRunFinalizations.runId) { + return and( + inArray(issueRecoveryActions.cause, workspaceExportFailureCauses), + sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`, + sql`${issueRecoveryActions.wakePolicy}->>'runId' = ${runId}::text`, + sql`exists (select 1 from ${heartbeatRuns} retry_run + join ${nativeRunFinalizations} retry_final on retry_final.company_id = retry_run.company_id + and retry_final.run_id = retry_run.id and retry_final.issue_id = retry_run.native_issue_id + join ${nativeRunResults} retry_result on retry_result.company_id = retry_run.company_id + and retry_result.run_id = retry_run.id and retry_result.issue_id = retry_run.native_issue_id + and retry_result.id = retry_final.result_id and retry_result.schema_status = 'accepted' + and retry_result.completion_contract_id = retry_run.completion_contract_id + join ${environmentLeases} retry_lease on retry_lease.company_id = retry_run.company_id + and retry_lease.heartbeat_run_id = retry_run.id and retry_lease.issue_id = retry_run.native_issue_id + and retry_lease.id::text = ${issueRecoveryActions.evidence}#>>'{workspaceExportRetry,leaseId}' + where retry_run.company_id = ${issueRecoveryActions.companyId} + and retry_run.native_issue_id = ${issueRecoveryActions.sourceIssueId} and retry_run.id::text = ${runId}::text + and retry_run.runtime_mode = 'native' and retry_run.status in ('succeeded', 'failed', 'cancelled', 'interrupted', 'timed_out') + and retry_run.native_phase = 'committed' and retry_final.phase = 'committed' + and ${issueRecoveryActions.evidence}#>>'{workspaceExportRetry,runId}' = retry_run.id::text + and ${issueRecoveryActions.evidence}#>>'{workspaceExportRetry,resultId}' = retry_result.id::text + and ${issueRecoveryActions.evidence}#>>'{workspaceExportRetry,status}' = 'queued' + and ${issueRecoveryActions.evidence}#>>'{workspaceExportRetry,requestId}' ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' + and retry_run.result_json->'workspaceExportRetry' = ${issueRecoveryActions.evidence}->'workspaceExportRetry' + and retry_run.runner_profile_json#>>'{nativeWorkspaceSync,schema}' in + ('paperclip.native-workspace-sync/v1', 'paperclip.native-workspace-sync/v2', 'paperclip.native-workspace-sync/v3') + and retry_run.runner_profile_json#>>'{nativeWorkspaceSync,state}' = 'finalized' + and retry_run.runner_profile_json#>>'{nativeWorkspaceSync,finalHostSha256}' ~ '^[0-9a-f]{64}$' + and retry_run.runner_profile_json#>>'{nativeWorkspaceSync,leaseId}' = retry_lease.id::text + and retry_run.runner_profile_json#>>'{nativeWorkspaceSync,providerLeaseId}' = retry_lease.provider_lease_id + and coalesce(retry_run.runner_profile_json->'nativeWorkspaceFinalizationOwner', 'null'::jsonb) = 'null'::jsonb + and exists (select 1 from ${workspaceOperations} op where op.company_id = retry_run.company_id + and op.issue_id = retry_run.native_issue_id and op.heartbeat_run_id = retry_run.id + and op.phase = 'workspace_finalize' and op.status = 'succeeded' + and op.metadata->>'owningService' = 'native_workspace_finalizer' + and op.metadata#>>'{workspaceSync,schema}' = retry_run.runner_profile_json#>>'{nativeWorkspaceSync,schema}' + and op.metadata#>>'{workspaceSync,workspaceId}' = retry_run.runner_profile_json#>>'{nativeWorkspaceSync,workspaceId}' + and op.metadata#>>'{workspaceSync,leaseId}' = retry_lease.id::text))`, + ); +} function matchingNotice(runId: string | typeof nativeRunFinalizations.runId) { return and( eq(issueRecoveryActions.kind, "active_run_watchdog"), inArray(issueRecoveryActions.status, ["active", "escalated"]), sql`${issueRecoveryActions.evidence}->>'runId' = ${runId}::text`, - or(sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`, - eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified")), + or(and(not(inArray(issueRecoveryActions.cause, workspaceExportFailureCauses)), + sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`), + eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified"), + recoveredExportNotice(runId), completedExportRetryNotice(runId)), ); } @@ -23,6 +76,9 @@ export function pendingNativeRecoveryNotices(db: Db, scope?: NativeRecoveryNotic .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), + eq(nativeRunResults.runId, nativeRunFinalizations.runId), eq(nativeRunResults.schemaStatus, "accepted"))) .where(and(eq(nativeRunFinalizations.phase, "committed"), isNotNull(nativeRunFinalizations.resultId), eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition(), not(pendingNativeStatusEffectsCondition(db)), not(pendingNativeRunProjectionCondition(db)), @@ -60,10 +116,11 @@ export async function resolveCommittedNativeRecoveryNotices(db: Db, eq(heartbeatRuns.nativeIssueId, scope.issueId), eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition())).for("update"); if (!run) return; - if (onlyWhenSettled && !(await pendingNativeRecoveryNotices(tx, scope)).length) return; - const actions = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + const actions = await tx.select({ id: issueRecoveryActions.id, + exportSettled: sql`coalesce(${or(recoveredExportNotice(scope.runId), completedExportRetryNotice(scope.runId))}, false)` }).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, scope.companyId), eq(issueRecoveryActions.sourceIssueId, scope.issueId), matchingNotice(scope.runId))).for("update"); + if ((onlyWhenSettled || actions.some(action => action.exportSettled)) && !(await pendingNativeRecoveryNotices(tx, scope)).length) return; for (const action of actions) await issueRecoveryActionService(tx).resolveActiveForIssue({ companyId: scope.companyId, sourceIssueId: scope.issueId, actionId: action.id, status: "resolved", outcome: "restored", diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index e80246865b0..b1c903b0239 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -65,6 +65,7 @@ import { type NativeStatusCommitFailpoint, } from "./status-decision-committer.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import { preservedResultHasStop, readPreservedNativeResultFinalization } from "./preserved-native-result-finalization.js"; import { issueService } from "../issues.js"; import { publishChatPublicationCommitSignal } from "../chat-publication-reconciliation.js"; import { nativeSha256 } from "./canonical.js"; @@ -337,6 +338,8 @@ async function recordRetryableFailure(input: { const issue = await tx .select({ lastStatusDecisionId: issues.lastStatusDecisionId, + status: issues.status, statusVersion: issues.statusVersion, + assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, }) .from(issues) .where( @@ -493,6 +496,7 @@ async function recordRetryableFailure(input: { .returning(); if (projectsTerminalStatus && updatedRun) await settleNativeTerminalWake(tx as unknown as Db, updatedRun); if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null; + let workspaceFinalizationBlock: Record | undefined; if (supersededByNewerRun) { await issueRecoveryActionService( tx as unknown as Db, @@ -513,7 +517,7 @@ async function recordRetryableFailure(input: { tx, ); } else if (exhausted) { - await issueService(tx as unknown as Db).update( + const blocked = await issueService(tx as unknown as Db).update( input.coordinator.issueId, { status: @@ -523,6 +527,25 @@ async function recordRetryableFailure(input: { }, tx, ); + if (input.failureScope === "workspace" && issue && blocked?.status === "blocked" + && !["blocked", "done", "cancelled"].includes(issue.status) + && issue.assigneeAgentId === input.run.agentId && !issue.assigneeUserId + && input.run.status !== "cancelled" && !preservedResultHasStop(input.run.resultJson) + && input.coordinator.resultId && input.run.completionContractId && input.run.completionContractSha256) { + const [accepted] = await tx.select({ id: nativeRunResults.id, sha256: nativeRunResults.canonicalSha256 }) + .from(nativeRunResults).where(and(eq(nativeRunResults.companyId, input.run.companyId), + eq(nativeRunResults.issueId, input.coordinator.issueId), eq(nativeRunResults.runId, input.run.id), + eq(nativeRunResults.id, input.coordinator.resultId), eq(nativeRunResults.schemaStatus, "accepted"), + eq(nativeRunResults.completionContractId, input.run.completionContractId))).limit(1); + if (accepted) workspaceFinalizationBlock = { + schema: "paperclip.native-workspace-finalization-block.v1", + companyId: input.run.companyId, issueId: input.coordinator.issueId, + runId: input.run.id, agentId: input.run.agentId, resultId: accepted.id, resultSha256: accepted.sha256, + contractId: input.run.completionContractId, contractSha256: input.run.completionContractSha256, + identityContextId: input.run.activeIdentityContextId, responsibleUserId: input.run.responsibleUserId, + statusVersion: Number(blocked.statusVersion), priorDecisionId: issue.lastStatusDecisionId, + }; + } } if (!supersededByNewerRun) { await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({ @@ -536,6 +559,7 @@ async function recordRetryableFailure(input: { fingerprint: nativeSha256({ runId: input.run.id, failureCode }), evidence: { runId: input.run.id, + ...(workspaceFinalizationBlock ? { workspaceFinalizationBlock } : {}), coordinatorAttempt: input.coordinator.attempt, ...(workspaceFinalizeAttempt === null ? {} @@ -1325,6 +1349,10 @@ export async function finalizeNativeRun(input: { .then((rows) => rows[0] ?? null); if (!authoritativeIssue) throw new Error("native_finalization_issue_missing"); + const preservedResultFinalization = run.status !== "running" + ? await readPreservedNativeResultFinalization(input.db, { + companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId, + }) : null; const publishedReviewContinuation = assessment.reportedDisposition === "yielded" && isPublishedGitHubReviewCompletionContinuation(assessment.continuation) ? await readPublishedGitHubReviewCompletionContinuation({ db: input.db, @@ -1465,6 +1493,7 @@ export async function finalizeNativeRun(input: { dependencyReadiness.unresolvedBlockerCount > 0, governanceResolvedForRun: resolvedInteraction !== null, externalChatResponseWaitAuthorization, + hasUnprovenPreservedResultStatus: preservedResultFinalization?.statusRestoration === "unproven", publishedExternalReview: assessment.reportedDisposition === "yielded" && (externalChatResponseWaitAuthorization === "authorized" || monitorWaitAt !== null) ? await hasPublishedGitHubReview(input.db, { @@ -1548,6 +1577,7 @@ export async function finalizeNativeRun(input: { priorStatusVersion: Number(authoritativeIssue.statusVersion), priorDecisionId: authoritativeIssue.lastStatusDecisionId, decision, + requirePreservedResultFinalization: preservedResultFinalization ?? undefined, supersededExternalReviewRequests: supersededReviews.length && !governanceGate && ["completion_claim_policy_accepted", "completion_contract_satisfied", "external_chat_response_waiting", "scheduled_monitor_waiting"].includes(decision.reasonCode ?? "") ? { agentId: run.agentId, resultId: resultRow.id, interactionIds: supersededReviews.map(({ interaction }) => interaction.id) } : undefined, @@ -1578,7 +1608,7 @@ export async function finalizeNativeRun(input: { ? (boardResponseWaitOrigin ?? undefined) : undefined, requireExternalChatResponseWaitAuthorization: - decision.reasonCode === "external_chat_response_waiting" + ["external_chat_response_waiting", "preserved_result_status_unproven"].includes(decision.reasonCode ?? "") ? { agentId: run.agentId, ...(publishedReviewContinuation ? { completionContinuationKey: publishedReviewContinuation.idempotencyKey } : {}) } : undefined, reviewResponsePresentation: diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index 41bc23f45a7..4da42937557 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -12,6 +12,7 @@ import { pendingNativeGovernance, resolveNativeFinalizerStatus } from "./native- import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, type NativeStatusDecision } from "./status-arbiter.js"; import { commitNativeStatusDecision, NativeStatusRaceError } from "./status-decision-committer.js"; import { recordNativeWorkAssessment } from "./work-assessments.js"; +import { readPreservedNativeResultFinalization } from "./preserved-native-result-finalization.js"; export type NativeStatusReassessmentSource = { runId: string; companyId: string; agentId: string; issueId: string; @@ -123,6 +124,11 @@ export async function reassessNativeStatusFromSource( assessment.priorIssueStatus !== row.issueStatus || Number(assessment.priorStatusVersion) !== Number(row.issueStatusVersion) ); + if (authoritativeStatusChanged && row.coordinatorPhase !== "committed") { + const preserved = await readPreservedNativeResultFinalization(db, row); + if (preserved?.statusRestoration === "owned" && preserved.resultId === assessment.resultId + && preserved.contractId === assessment.contractId) return { kind: "unchanged" }; + } const retiredAutomaticReview = issueMatchesCurrentDecision && currentDecision ? await decisionHasRetiredAutomaticReview(db, currentDecision) : false; let reassessment = null; diff --git a/server/src/services/native-runtime/native-workspace-export-retry.test.ts b/server/src/services/native-runtime/native-workspace-export-retry.test.ts index 5b4ef55853c..5256df77534 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.test.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.test.ts @@ -18,6 +18,7 @@ import { recordNativeFinalizationFailure } from "./native-run-finalizer.js"; import { recoveryService } from "../recovery/service.js"; import { retryNativeWorkspaceExport } from "./native-workspace-export-retry.js"; import { releaseCompletedNativeWorkspaceExportRetention } from "./native-workspace-export-resume.js"; +import { requestWorkspaceRecovery } from "../workspace-recovery.js"; describe("board retry of accepted workspace export", () => { let temporary: Awaited> | undefined; @@ -80,6 +81,34 @@ describe("board retry of accepted workspace export", () => { expect(resume).toHaveBeenCalledOnce(); expect(resume).toHaveBeenCalledWith(expect.objectContaining({ lease: expect.objectContaining({ id: f.leaseId, providerLeaseId: lease.providerLeaseId }) })); }); + it("keeps explicit export repair available after Recover adopts the same accepted source", async () => { + const f = await seed(); + const before = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + await requestWorkspaceRecovery({ ...f.request, idempotencyKey: randomUUID() }); + const [adopted] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + expect(adopted.ownerType).toBe("system"); + expect(await retryNativeWorkspaceExport(f.request)).toMatchObject({ + status: "queued", runId: f.runId, resultId: f.resultId, leaseId: f.leaseId, + }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toEqual(before); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId))).toHaveLength(0); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1); + expect((f.request.environmentRuntime as { resumeRunLease: ReturnType }).resumeRunLease).toHaveBeenCalledOnce(); + }); + it.each(["missing_recovery", "wrong_system_source", "wrong_workspace_source", "agent_owner"])("rejects unrelated recovery ownership before provider access: %s", async kind => { + const f = await seed(); + await requestWorkspaceRecovery({ ...f.request, idempotencyKey: randomUUID() }); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + const evidence = { ...action.evidence }; + if (kind === "missing_recovery") delete evidence.systemRecovery; + if (kind === "wrong_system_source") evidence.systemRecovery = { ...(evidence.systemRecovery as object), sourceRunId: randomUUID() }; + if (kind === "wrong_workspace_source") evidence.workspaceRecovery = { ...(evidence.workspaceRecovery as object), runId: randomUUID() }; + await db.update(issueRecoveryActions).set({ evidence, ...(kind === "agent_owner" ? { ownerType: "agent" } : {}) }).where(eq(issueRecoveryActions.id, f.actionId)); + await expect(retryNativeWorkspaceExport(f.request)).rejects.toThrow("no longer current"); + expect(probe).not.toHaveBeenCalled(); + expect((f.request.environmentRuntime as { resumeRunLease: ReturnType }).resumeRunLease).not.toHaveBeenCalled(); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toMatchObject({ phase: "terminal_failure", resultId: f.resultId }); + }); it("retries exhausted transient export of an ephemeral allocation without another provider turn", async () => { const f = await seed(); await db.update(nativeRunFinalizations).set({ failureCode: "native_workspace_sync_out_retry_exhausted", failureDetail: { workspaceFinalizeAttempt: 3 } }).where(eq(nativeRunFinalizations.runId, f.runId)); @@ -258,12 +287,14 @@ describe("board retry of accepted workspace export", () => { await expect(retryNativeWorkspaceExport(f.request)).rejects.toThrow("no longer current"); expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("terminal_failure"); }); - it("does not take ownership from an active exporter", async () => { + it.each(["board", "system"])("does not take ownership from an active exporter after %s admission", async owner => { const f = await seed(); + if (owner === "system") await requestWorkspaceRecovery({ ...f.request, idempotencyKey: randomUUID() }); await withNativeWorkspaceFinalizationOwnership(f, async () => { await expect(retryNativeWorkspaceExport(f.request)).rejects.toThrow("still owned"); }); expect(probe).not.toHaveBeenCalled(); + expect((f.request.environmentRuntime as { resumeRunLease: ReturnType }).resumeRunLease).not.toHaveBeenCalled(); }); it.each(["native_workspace_sync_out_retry_exhausted", "native_workspace_sync_out_unsafe_archive"])("does not dispatch another provider turn from a pending accepted export: %s", async cause => { const f = await seed(cause); diff --git a/server/src/services/native-runtime/native-workspace-export-retry.ts b/server/src/services/native-runtime/native-workspace-export-retry.ts index 5a79c9e782e..e33aff46c73 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.ts @@ -14,6 +14,8 @@ import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-fin import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; import type { EnvironmentRuntimeService } from "../environment-runtime.js"; import { resolveEnvironmentExecutionTarget } from "../environment-execution-target.js"; +import { readSystemRecovery } from "../system-recovery-state.js"; +import { readWorkspaceRecovery } from "../workspace-recovery-receipt.js"; const record = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; const changed = () => conflict("The recorded export repair is no longer current. Refresh the task and inspect its run.", { code: "workspace_export_retry_stale" }); @@ -31,8 +33,12 @@ export async function retryNativeWorkspaceExport(input: { const coordinator = await one(db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId), eq(nativeRunFinalizations.issueId, input.issueId))).limit(1)); const run = await one(db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.nativeIssueId, input.issueId))).limit(1)); const action = await one(db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId), eq(issueRecoveryActions.id, input.actionId))).limit(1)); + const system = action && readSystemRecovery(action.evidence); + const recovery = action && readWorkspaceRecovery(action.evidence); + const repairOwner = action?.ownerType === "board" || (action?.ownerType === "system" + && system?.sourceRunId === run?.id && recovery?.runId === run?.id); if (!issue || !run || !coordinator || !action || action.evidence?.runId !== run.id - || action.kind !== "active_run_watchdog" || !isNativeWorkspaceExportRepairCause(action.cause) || action.ownerType !== "board" + || action.kind !== "active_run_watchdog" || !isNativeWorkspaceExportRepairCause(action.cause) || !repairOwner || !["active", "escalated"].includes(action.status) || run.runtimeMode !== "native" || !(issue.status === "blocked" || (issue.status === "in_review" && action.cause === "native_workspace_sync_out_retry_exhausted")) || issue.assigneeAgentId !== run.agentId || (issue.executionRunId && issue.executionRunId !== run.id) diff --git a/server/src/services/native-runtime/preserved-native-response-wait.integration.test.ts b/server/src/services/native-runtime/preserved-native-response-wait.integration.test.ts new file mode 100644 index 00000000000..80ebbb01c4a --- /dev/null +++ b/server/src/services/native-runtime/preserved-native-response-wait.integration.test.ts @@ -0,0 +1,669 @@ +import { execFile } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import os from "node:os"; +import path from "node:path"; +import { and, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { agentWakeupRequests, agents, chatActions, chatConversations, chatEndpointResources, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, + chatGitHubReviews, companies, completionContracts, createDb, environmentLeases, environments, executionWorkspaceSnapshots, executionWorkspaces, heartbeatRuns, + issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, runIdentityContexts, + statusDecisionEffects, statusDecisions, toolConnections, workspaceOperations } from "@paperclipai/db"; +import { defaultGitHubReviewPolicy, readRemoteWorkspaceAuthority, type EnvironmentLease } from "@paperclipai/shared"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { seedBoundChatAuthorization } from "../../__tests__/helpers/bound-chat-authorization.js"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../../vendor/paperclip-runner/testing.js"; +import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../environment-runtime.js"; +import { issueService } from "../issues.js"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { resumePreservedResultFinalization } from "../preserved-result-finalization.js"; +import { recoverRetainedWorkspace, requestWorkspaceRecovery } from "../workspace-recovery.js"; +import { readSystemRecovery } from "../system-recovery-state.js"; +import { readTaskRetryIntent, requestTaskRetry } from "../task-retry.js"; +import { readPublishedGitHubReviewCompletionContinuation, resolveExternalChatResponseWaitAuthorization } from "./chat-attachment-reuse.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { recordNativeWorkspaceFinalizationFailure } from "./native-workspace-finalizer.js"; +import { prepareNativeWorkspaceSync } from "./native-workspace-sync.js"; +import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { readPreservedNativeResultFinalization } from "./preserved-native-result-finalization.js"; +import { pendingNativeRecoveryNotices, resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; +import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js"; +import { recordNativeWorkAssessment } from "./work-assessments.js"; +import { classifyNativeEvidence } from "./evidence-classifier.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "./status-arbiter.js"; +import { retryNativeWorkspaceExport } from "./native-workspace-export-retry.js"; + +const targets = vi.hoisted(() => new Map()); +const commitRace = vi.hoisted(() => ({ before: null as null | ((input: + Parameters[0]) => Promise), rejected: [] as string[] })); +vi.mock("./status-decision-committer.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, commitNativeStatusDecision: async (input: Parameters[0]) => { + const before = commitRace.before; + commitRace.before = null; + await before?.(input); + try { return await actual.commitNativeStatusDecision(input); } + catch (error) { + if (before) commitRace.rejected.push(error instanceof Error ? error.name : "unknown"); + throw error; + } + } }; +}); +vi.mock("../environment-execution-target.js", async importOriginal => ({ + ...await importOriginal(), + resolveEnvironmentExecutionTarget: async (input: { leaseId?: string }) => targets.get(input.leaseId!)!, +})); + +describe("preserved native response wait", () => { + let temporary: Awaited>; + let db: ReturnType, root: string; + const servers = new Set(); + beforeAll(async () => { + root = await mkdtemp(path.join(os.tmpdir(), "preserved-native-wait-")); + temporary = await startEmbeddedPostgresTestDatabase("preserved-native-wait-db-"); + db = createDb(temporary.connectionString); + }, 90_000); + afterAll(async () => { + for (const server of servers) { + server.closeAllConnections(); + await new Promise(resolve => server.close(() => resolve())); + } + await db?.$client.end(); + await temporary?.cleanup(); + if (root) await rm(root, { recursive: true, force: true }); + }); + afterEach(() => { commitRace.before = null; commitRace.rejected.length = 0; vi.restoreAllMocks(); }); + + type Scenario = "valid" | "legacy" | "blocked_reasserted" | "principal_revoked" | "resource_disabled" + | "connection_disabled" | "review_head_changed" | "stop" | "cancelled_run" | "cancelled_task" + | "owner_changed" | "newer_cancelled" | "newer_active" | "incomplete_preservation" | "missing_receipt" + | "defer_notice" | "http_reconcile" | "existing_assessment" | "commit_stop" | "commit_status" | "commit_identity" + | "commit_newer" | "commit_principal" | "commit_preservation" | "commit_head" | "commit_cancelled" + | "saved_retry" | "saved_chat_retry" | "export_retry" | "export_retry_defer"; + async function recovered(scenario: Scenario = "valid") { + const f = await seedBoundChatAuthorization(db); + const prefix = `PW${randomUUID().slice(0, 6)}`; + await db.update(companies).set({ issuePrefix: prefix }).where(eq(companies.id, f.companyId)); + await db.update(issues).set({ identifier: `${prefix}-1` }).where(eq(issues.id, f.issueId)); + await db.update(issues).set({ identifier: `${prefix}-2` }).where(eq(issues.id, f.otherIssueId)); + const context = { ...f.context, source: "chat:github", issueId: f.issueId, + wakeCommentId: f.commentId, paperclipWake: { checkedOutByHarness: true, + externalChatProvider: "github", issue: { id: f.issueId }, commentIds: [f.commentId] } }; + await db.update(chatEndpoints).set({ provider: "github", + ...(scenario === "saved_retry" ? { externalExecutionPolicy: "agent" as const } : {}) }).where(eq(chatEndpoints.id, f.endpointId)); + await db.update(chatExternalPrincipals).set({ provider: "github" }).where(eq(chatExternalPrincipals.id, f.principalId)); + await db.update(chatEndpointResources).set({ type: "repository", providerResourceId: "example/repository", + metadata: { providerRepositoryId: "101" } }).where(eq(chatEndpointResources.id, f.resourceId)); + await db.update(chatConversations).set({ externalThreadId: "github:example/repository:42" }) + .where(eq(chatConversations.id, f.conversationId)); + const contractId = randomUUID(), sourceInstanceId = randomUUID(), identityId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId, + revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "preserved-wait-test", + risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.contractRevision, + objective: "Review the requested change", criteria: [{ id: "objective", requirement: "Review the change" }] }, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: f.runId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, issueId: f.issueId, runtimeMode: "native", status: "running", + nativeSessionId: f.runId, runnerInstanceId: sourceInstanceId, responsibleUserId: f.userId, + completionContractId: contractId, completionContractSha256: contractId, contextSnapshot: context }); + await db.insert(runIdentityContexts).values({ id: identityId, companyId: f.companyId, runId: f.runId, + revision: 1, responsibleUserId: f.userId, messageId: f.commentId, cause: "run_admission", + correlationId: f.runId, status: "accepted", acceptedAt: new Date() }); + await db.update(heartbeatRuns).set({ activeIdentityContextId: identityId }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ status: "in_progress", executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + const head = "b".repeat(40), reviewId = randomUUID(); + await db.insert(chatGitHubReviews).values({ id: reviewId, companyId: f.companyId, endpointId: f.endpointId, + issueId: f.issueId, runId: f.runId, repositoryId: "101", repository: "example/repository", pullNumber: 42, + headSha: head, deliveryId: f.deliveryId, configurationRevision: 1, policySnapshot: defaultGitHubReviewPolicy(), + event: { event: "mention", deliveryId: f.deliveryId, repositoryId: "101", repository: "example/repository", + pullNumber: 42, title: "Review this change", body: "Review request", baseSha: "a".repeat(40), + headSha: head, baseBranch: "master", author: { id: "user-1", login: "author", isBot: false }, + sender: { id: "user-1", login: "author" }, draft: false, labels: [] }, + state: "completed", assessment: { reviewedCommit: head, score: 3, complete: true, + summary: "Review delivered", rationale: "Reviewed the change", findings: [], + coverage: { reviewedPaths: ["src/example.ts"], omittedPaths: [], limitations: [] } } }); + await db.insert(chatActions).values({ companyId: f.companyId, endpointId: f.endpointId, + conversationId: f.conversationId, kind: "github_review_publication", status: "processed", + providerActionId: `github_publication:${f.issueId}:assessment:${createHash("sha256").update(JSON.stringify(reviewId)).digest("hex")}` }); + const continuation = await readPublishedGitHubReviewCompletionContinuation({ db, binding: f.binding }); + expect(continuation?.kind).toBe("response_wake"); + expect(await resolveExternalChatResponseWaitAuthorization({ db, binding: f.binding })).toBe("authorized"); + const binding = { ...f.binding, sessionId: f.runId, completionContractId: contractId, + completionContractSha256: contractId, sourceInstanceId, controlPlaneSourceInstanceId: "preserved-wait-test" }; + await new PaperclipControlPlanePort(db, binding).completeRun({ + result: { ...CONTROL_PLANE_CONFORMANCE_RESULT, reportedWorkDisposition: "yielded", continuation: continuation! }, + terminal: { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, reportedWorkDisposition: "yielded" }, + callerResultId: `${f.runId}:result` }); + const [accepted] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + expect(accepted.schemaStatus).toBe("accepted"); + if (scenario === "existing_assessment") { + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.id, contractId)); + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const assessment = await recordNativeWorkAssessment({ db, companyId: f.companyId, issueId: f.issueId, runId: f.runId, + turnId: accepted.turnId, contractId, contractCanonicalSha256: contract.canonicalSha256, + resultId: accepted.id, resultCanonicalSha256: accepted.canonicalSha256, + priorIssueStatus: issue.status, priorStatusVersion: Number(issue.statusVersion), priorDecisionId: issue.lastStatusDecisionId, + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, assessment: await classifyNativeEvidence({ db, + companyId: f.companyId, issueId: f.issueId, runId: f.runId, contract: contract.contractJson, + result: accepted.resultJson.result as Record }) }); + await db.update(nativeRunFinalizations).set({ assessmentId: assessment.id }).where(eq(nativeRunFinalizations.runId, f.runId)); + } + const host = path.join(root, f.runId, "host"), remote = path.join(root, f.runId, "remote"); + await mkdir(host, { recursive: true }); await writeFile(path.join(host, "existing.txt"), "base\n"); + const [environment] = await db.insert(environments).values({ name: f.runId, driver: "sandbox" }).returning(); + const providerLeaseId = randomUUID(); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, environmentId: environment.id, provider: "daytona", providerLeaseId, + status: "released", cleanupStatus: "success", releasedAt: new Date(), leasePolicy: "retain_on_failure", + metadata: { remoteCwd: remote, driver: "sandbox", pluginId: randomUUID(), sandboxProviderPlugin: true } }).returning(); + const target: AdapterSandboxExecutionTarget = { kind: "remote", transport: "sandbox", providerKey: "test", + leaseId: lease.id, remoteCwd: remote, timeoutMs: 30_000, runner: { execute: input => new Promise(resolve => { + const child = execFile(input.command, input.args ?? [], { cwd: input.cwd, env: { ...process.env, ...input.env }, + timeout: input.timeoutMs, maxBuffer: 16 * 1024 * 1024 }, (error, stdout, stderr) => resolve({ + exitCode: error ? 1 : 0, signal: error?.signal ?? null, timedOut: error?.killed ?? false, + stdout, stderr, pid: child.pid ?? null, startedAt: new Date().toISOString() })); + child.stdin?.on("error", () => undefined); child.stdin?.end(input.stdin ?? ""); + }) } }; + targets.set(lease.id, target); + await prepareNativeWorkspaceSync({ db, companyId: f.companyId, workspaceId: f.issueId, + workspaceLocalDir: host, runId: f.runId, lease: lease as EnvironmentLease, target, resourceDisposition: "keep_running" }); + await writeFile(path.join(remote, "review.txt"), "Original reviewed work\n"); + await db.update(environmentLeases).set({ metadata: { ...lease.metadata, + remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId, state: "stopped" }) } }) + .where(eq(environmentLeases.id, lease.id)); + for (let attempt = 0; attempt < 3; attempt += 1) { + await recordNativeWorkspaceFinalizationFailure(db, f.runId, { stderrExcerpt: "workspace_sync_out_failed" }); + } + await issueService(db).clearExecutionRunIfTerminal(f.issueId); + await issueService(db).clearCheckoutRunIfTerminal(f.issueId); + const [action] = await db.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, f.companyId), eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.status, "active"))); + expect(action).toMatchObject({ status: "active", cause: "native_workspace_sync_out_retry_exhausted", wakePolicy: null }); + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", acquireRunLease: vi.fn(), + releaseRunLease: vi.fn(), resumeRunLease: vi.fn(async () => ({ providerLeaseId, metadata: { remoteCwd: remote } })), + resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: [], declared: null, narrowing: null }), + retryPendingSandboxTeardown: vi.fn(async () => ({ providerLeaseId, state: "stopped" as const })) }] }); + const input = { ...f.binding, actionId: action.id, db, environmentRuntime: runtime }; + if (scenario === "export_retry" || scenario === "export_retry_defer") { + expect(await retryNativeWorkspaceExport({ ...input, actorId: f.userId, + repairNote: "The retained transport is available for exact saved-file export." })) + .toMatchObject({ runId: f.runId, resultId: accepted.id, leaseId: lease.id, status: "queued" }); + const finalized = (await reconcileNativeFinalizations(db, [f.runId], { environmentRuntime: runtime, + skipNativeRecoveryNotices: scenario === "export_retry_defer" })) + .find(value => value.runId === f.runId); + expect(finalized?.phase).toBe("committed"); + expect(await readFile(path.join(host, "review.txt"), "utf8")).toBe("Original reviewed work\n"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_review"); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toEqual([accepted]); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + expect(await db.select().from(chatActions).where(eq(chatActions.companyId, f.companyId))).toHaveLength(1); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].status) + .toBe(scenario === "export_retry_defer" ? "active" : "resolved"); + return { ...f, actionId: action.id }; + } + let replayRecovery: (() => Promise) | undefined; + if (scenario === "http_reconcile") { + const [{ default: express }, { default: request }, { issueRoutes }, { errorHandler }] = await Promise.all([ + import("express"), import("supertest"), import("../../routes/issues.js"), import("../../middleware/error-handler.js"), + ]); + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { + req.actor = { type: "board", userId: f.userId, source: "local_implicit", companyIds: [f.companyId], isInstanceAdmin: false }; + next(); + }); + app.use("/api", issueRoutes(db, { provider: "local_disk", putFile: async () => { throw new Error("unused"); }, + getObject: async () => { throw new Error("unused"); }, headObject: async () => ({ exists: false }), deleteObject: async () => {} })); + app.use(errorHandler); + const server = createServer(app); + servers.add(server); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { server.removeListener("error", reject); resolve(); }); + }); + const api = request(`http://127.0.0.1:${(server.address() as AddressInfo).port}`); + const body = { actionId: action.id, runId: f.runId, idempotencyKey: randomUUID() }; + expect((await api.post(`/api/issues/${f.issueId}/recovery-actions/recover-workspace`).send(body)).status).toBe(202); + replayRecovery = async () => { + const response = await api.post(`/api/issues/${f.issueId}/recovery-actions/recover-workspace`).send(body); + expect(response.status).toBe(202); + expect(response.body).toMatchObject({ actionId: action.id, status: "recovered" }); + const refreshed = await api.get(`/api/issues/${f.issueId}/recovery-actions`); + expect(refreshed.status).toBe(200); + expect(refreshed.body.active).toBeNull(); + expect(refreshed.body.actions.some((value: { id: string; status: string }) => value.id === action.id + && ["active", "escalated"].includes(value.status))).toBe(false); + }; + } else await requestWorkspaceRecovery({ ...input, idempotencyKey: randomUUID(), actorId: f.userId }); + await recoverRetainedWorkspace(input); + const [preserved] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)); + expect(readSystemRecovery(preserved.evidence)?.state).toBe("complete"); + expect(await readFile(path.join(host, "review.txt"), "utf8")).toBe("Original reviewed work\n"); + let retryRequestId: string | null = null; + if (scenario === "saved_retry") { + const retry = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, + actorId: f.userId, actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(retry.accepted).toBe(true); + retryRequestId = retry.requestId; + } + if (scenario === "saved_chat_retry") { + retryRequestId = randomUUID(); + const payload = { issueId: f.issueId, taskRetry: { schema: "paperclip.task-retry.v1", generationId: retryRequestId, + sourceRunId: f.runId, actorId: f.userId, actorSource: "local_implicit", strategy: "evaluate" } }; + expect(readTaskRetryIntent(payload)?.sourceRunId).toBe(f.runId); + await db.insert(agentWakeupRequests).values({ id: retryRequestId, companyId: f.companyId, agentId: f.agentId, + source: "on_demand", triggerDetail: "manual", reason: "task_retry", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: f.userId, idempotencyKey: `task-retry:${randomUUID()}`, payload }); + } + if (retryRequestId) expect(retryRequestId).not.toBe(readSystemRecovery(preserved.evidence)?.generationId); + if (scenario === "legacy") { + const { workspaceFinalizationBlock: _block, ...evidence } = preserved.evidence; + await db.update(issueRecoveryActions).set({ evidence }).where(eq(issueRecoveryActions.id, action.id)); + } + if (scenario === "blocked_reasserted") await issueService(db).update(f.issueId, { status: "blocked" }); + if (scenario === "principal_revoked") await db.update(chatIdentityLinks).set({ status: "revoked" }) + .where(eq(chatIdentityLinks.endpointId, f.endpointId)); + if (scenario === "resource_disabled") await db.update(chatEndpointResources).set({ enabled: false }) + .where(eq(chatEndpointResources.id, f.resourceId)); + if (scenario === "connection_disabled") { + const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId)); + await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, endpoint.connectionId)); + } + if (scenario === "review_head_changed") await db.update(chatGitHubReviews).set({ headSha: "c".repeat(40) }) + .where(eq(chatGitHubReviews.id, reviewId)); + if (scenario === "stop") { + const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ resultJson: { ...current.resultJson, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", companyId: f.companyId, + issueId: f.issueId, runId: f.runId, scope: "run", dispatchState: "requested" } } }).where(eq(heartbeatRuns.id, f.runId)); + } + if (scenario === "cancelled_run") await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "cancelled_task") await issueService(db).update(f.issueId, { status: "cancelled" }); + if (scenario === "owner_changed") { + const peerId = randomUUID(); + await db.insert(agents).values({ id: peerId, companyId: f.companyId, name: "New owner", adapterType: "paperclip_runner" }); + await db.update(issues).set({ assigneeAgentId: peerId }).where(eq(issues.id, f.issueId)); + } + if (scenario === "newer_cancelled" || scenario === "newer_active") { + const [original] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const newerId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: newerId, companyId: f.companyId, issueId: f.issueId, + agentId: f.agentId, runtimeMode: "legacy", status: scenario === "newer_cancelled" ? "cancelled" : "running", + createdAt: new Date(original.createdAt.getTime() + 1), contextSnapshot: { issueId: f.issueId } }); + if (scenario === "newer_active") await db.update(issues).set({ executionRunId: newerId }).where(eq(issues.id, f.issueId)); + } + if (scenario === "incomplete_preservation") await db.update(issueRecoveryActions).set({ evidence: { + ...preserved.evidence, workspaceRecovery: { ...preserved.evidence.workspaceRecovery as Record, status: "recovering" }, + } }).where(eq(issueRecoveryActions.id, action.id)); + if (scenario === "missing_receipt") await db.delete(workspaceOperations).where(eq(workspaceOperations.id, + readSystemRecovery(preserved.evidence)!.preservationOperationIds[0])); + const denied = ["stop", "cancelled_run", "cancelled_task", "owner_changed", "newer_cancelled", "newer_active", + "incomplete_preservation", "missing_receipt"].includes(scenario); + const [beforeIssue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(await resumePreservedResultFinalization(db, f.companyId, f.runId)).toBe(!denied); + if (denied) { + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(beforeIssue); + expect(await readPreservedNativeResultFinalization(db, f.binding)).toBeNull(); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toEqual([accepted]); + return { ...f, actionId: action.id }; + } + expect((await readPreservedNativeResultFinalization(db, f.binding))?.statusRestoration) + .toBe(["legacy", "blocked_reasserted"].includes(scenario) ? "unproven" : "owned"); + let injected = false; + if (scenario.startsWith("commit_")) commitRace.before = async value => { + injected = true; + expect(value.requirePreservedResultFinalization?.statusRestoration).toBe("owned"); + if (scenario === "commit_stop") await db.update(heartbeatRuns).set({ resultJson: { nativeCancellation: { dispatchState: "requested" } } }) + .where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "commit_cancelled") await db.update(heartbeatRuns).set({ status: "cancelled" }) + .where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "commit_status") await issueService(db).update(f.issueId, { status: "blocked" }); + if (scenario === "commit_identity") await db.update(heartbeatRuns).set({ activeIdentityContextId: null }).where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "commit_newer") { + const newerId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: newerId, companyId: f.companyId, nativeIssueId: f.issueId, + agentId: f.agentId, runtimeMode: "native", status: "running" }); + await db.update(issues).set({ status: "in_progress", executionRunId: newerId }).where(eq(issues.id, f.issueId)); + } + if (scenario === "commit_principal") await db.update(chatIdentityLinks).set({ status: "revoked" }) + .where(eq(chatIdentityLinks.endpointId, f.endpointId)); + if (scenario === "commit_preservation") await db.delete(workspaceOperations).where(eq(workspaceOperations.id, + readSystemRecovery(preserved.evidence)!.preservationOperationIds[0])); + if (scenario === "commit_head") await db.update(chatGitHubReviews).set({ headSha: "d".repeat(40) }) + .where(eq(chatGitHubReviews.id, reviewId)); + }; + const finalized = ["http_reconcile", "existing_assessment", "saved_retry", "saved_chat_retry"].includes(scenario) + ? (await reconcileNativeFinalizations(db, [f.runId], { environmentRuntime: runtime })).find(row => row.runId === f.runId)! + : await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true, + skipNativeRecoveryNotices: scenario === "defer_notice" }); + const [afterFinalization] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario.startsWith("commit_")) expect(injected).toBe(true); + expect(finalized.phase, JSON.stringify(afterFinalization.failureDetail)).toBe("committed"); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toEqual([accepted]); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(scenario === "commit_newer" ? 2 : 1); + expect(await db.select().from(chatActions).where(eq(chatActions.companyId, f.companyId))).toHaveLength(1); + const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(wakes).toHaveLength(retryRequestId ? 1 : 0); + if (retryRequestId) { + expect(wakes[0]).toMatchObject({ id: retryRequestId, status: "coalesced", runId: f.runId }); + expect(readTaskRetryIntent(wakes[0].payload)?.strategy).toBe("finalize_result"); + expect(afterFinalization.recoveryHistory).toEqual(expect.arrayContaining([expect.objectContaining({ + kind: "preserved_result_finalization", generationId: retryRequestId, + preservationGenerationId: readSystemRecovery(preserved.evidence)?.generationId, + })])); + } + if (scenario.startsWith("commit_")) { + expect(commitRace.rejected).toEqual(["NativeStatusRaceError"]); + const [currentIssue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(currentIssue.status).toBe(scenario === "commit_newer" ? "in_progress" : "blocked"); + expect(currentIssue.lastStatusDecisionId).toBe(beforeIssue.lastStatusDecisionId); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); + expect(decisions).toHaveLength(1); + expect(decisions[0].reasonCode).not.toBe("external_chat_response_waiting"); + expect(await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, decisions[0].id))).toHaveLength(0); + if (scenario === "commit_newer") expect(currentIssue.executionRunId).not.toBeNull(); + if (scenario === "commit_cancelled") expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("cancelled"); + if (scenario === "commit_stop") expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].resultJson?.nativeCancellation) + .toEqual({ dispatchState: "requested" }); + } + if (replayRecovery) { + const [settledIssue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + await replayRecovery(); + await reconcileNativeFinalizations(db, [f.runId], { environmentRuntime: runtime }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(settledIssue); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toEqual([accepted]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + } + return { ...f, actionId: action.id }; + } + + it("restores the same accepted published wait after normal file recovery", async () => { + const f = await recovered(); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.id, coordinator.decisionId!)); + expect(decision).toMatchObject({ reasonCode: "external_chat_response_waiting", toStatus: "in_review" }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_review"); + }, 60_000); + + it("resolves the exact obsolete exhausted export notice after normal file recovery", async () => { + const f = await recovered(); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0]) + .toMatchObject({ status: "resolved", outcome: "restored" }); + }, 60_000); + + it("accepts HTTP Recover and background finalization once across a repeated receipt", async () => { + const f = await recovered("http_reconcile"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_review"); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("resolved"); + }, 60_000); + + it("reconciles the failure-owned block after an earlier work assessment", async () => { + const f = await recovered("existing_assessment"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_review"); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("resolved"); + }, 60_000); + + it("settles an explicit export Retry after its own verified export and committed result", async () => { + await recovered("export_retry"); + }, 60_000); + + it.each(["missing_receipt", "mismatched_request", "mismatched_result", "mismatched_allocation", "unfinalized", + "foreign_barrier", "unsupported_schema", "pending_effect", "unknown_owner"] as const)( + "holds explicit export Retry bookkeeping with %s", async scenario => { + const f = await recovered("export_retry_defer"); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(1); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const retry = action.evidence.workspaceExportRetry as Record; + const source = run.runnerProfileJson!.nativeWorkspaceSync as Record; + if (["missing_receipt", "mismatched_request", "mismatched_result"].includes(scenario)) { + const value = scenario === "missing_receipt" ? null : { ...retry, + ...(scenario === "mismatched_request" ? { requestId: randomUUID() } : { resultId: randomUUID() }) }; + await db.update(issueRecoveryActions).set({ evidence: { ...action.evidence, workspaceExportRetry: value } }) + .where(eq(issueRecoveryActions.id, action.id)); + if (scenario === "mismatched_result") await db.update(heartbeatRuns).set({ resultJson: { ...run.resultJson, workspaceExportRetry: value } }) + .where(eq(heartbeatRuns.id, f.runId)); + } + if (["mismatched_allocation", "unfinalized", "unsupported_schema", "unknown_owner"].includes(scenario)) + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceSync: { ...source, + ...(scenario === "mismatched_allocation" ? { providerLeaseId: randomUUID() } : {}), + ...(scenario === "unfinalized" ? { state: "prepared" } : {}), + ...(scenario === "unsupported_schema" ? { schema: "unsupported" } : {}), + }, ...(scenario === "unknown_owner" ? { nativeWorkspaceFinalizationOwner: { ownerId: randomUUID() } } : {}) } }) + .where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "foreign_barrier") await db.update(workspaceOperations).set({ issueId: f.otherIssueId }) + .where(and(eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "workspace_finalize"))); + if (scenario === "unsupported_schema") { + const operations = await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "workspace_finalize"))); + for (const operation of operations) await db.update(workspaceOperations).set({ metadata: { ...operation.metadata, + workspaceSync: { ...operation.metadata?.workspaceSync as Record, schema: "unsupported" } } }) + .where(eq(workspaceOperations.id, operation.id)); + } + if (scenario === "pending_effect") { + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.insert(statusDecisionEffects).values({ companyId: f.companyId, issueId: f.issueId, decisionId: coordinator.decisionId!, + ordinal: 0, effectKind: "release_checkout", targetType: "issue", targetId: f.issueId, + idempotencyKey: randomUUID(), payload: {}, deliveryState: "pending" }); + } + const [before] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(0); + await resolveCommittedNativeRecoveryNotices(db, f.binding); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0]).toEqual(before); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + }, 60_000); + + it.each(["paperclip.native-workspace-sync/v1", "paperclip.native-workspace-sync/v2"])( + "recognizes the existing finalized export receipt schema %s", async schema => { + const f = await recovered("export_retry_defer"); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, + nativeWorkspaceSync: { ...run.runnerProfileJson!.nativeWorkspaceSync as Record, schema } } }) + .where(eq(heartbeatRuns.id, f.runId)); + const operations = await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "workspace_finalize"))); + for (const operation of operations) await db.update(workspaceOperations).set({ metadata: { ...operation.metadata, + workspaceSync: { ...operation.metadata?.workspaceSync as Record, schema } } }) + .where(eq(workspaceOperations.id, operation.id)); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(1); + await resolveCommittedNativeRecoveryNotices(db, f.binding); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("resolved"); + }, 60_000); + + it.each(["saved_retry", "saved_chat_retry"] as const)("settles the same wait after ordinary Recover folds %s", async scenario => { + const f = await recovered(scenario); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_review"); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("resolved"); + }, 60_000); + + it.each(["mismatched", "malformed", "missing_folded"] as const)("rejects %s preservation generation proof", async scenario => { + const f = await recovered(scenario === "missing_folded" ? "saved_retry" : "legacy"); + expect((await readPreservedNativeResultFinalization(db, f.binding))?.statusRestoration).toBe("unproven"); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const history = coordinator.recoveryHistory.map(entry => { + if (entry.kind !== "preserved_result_finalization") return entry; + const value = { ...entry }; + if (scenario === "missing_folded") delete value.preservationGenerationId; + else value.preservationGenerationId = scenario === "mismatched" ? randomUUID() : { generationId: randomUUID() }; + return value; + }); + await db.update(nativeRunFinalizations).set({ recoveryHistory: history }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await readPreservedNativeResultFinalization(db, f.binding)).toBeNull(); + }, 60_000); + + it("retains exact normal legacy generation proof without granting status ownership", async () => { + const f = await recovered("legacy"); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(nativeRunFinalizations).set({ recoveryHistory: coordinator.recoveryHistory.map(entry => { + const value = { ...entry }; delete value.preservationGenerationId; return value; + }) }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect((await readPreservedNativeResultFinalization(db, f.binding))?.statusRestoration).toBe("unproven"); + expect(await resolveExternalChatResponseWaitAuthorization({ db, binding: f.binding })).toBe("authorized"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); + }, 60_000); + + it.each(["commit_stop", "commit_status", "commit_identity", "commit_newer", "commit_principal", "commit_preservation", "commit_head", "commit_cancelled"] as const)( + "rechecks %s at the actual status commit", async scenario => { await recovered(scenario); }, 60_000); + + it.each(["legacy", "blocked_reasserted"] as const)("preserves %s status without claiming connector revocation", async scenario => { + const f = await recovered(scenario); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.id, coordinator.decisionId!)); + expect(decision).toMatchObject({ reasonCode: "preserved_result_status_unproven", toStatus: "blocked" }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); + }, 60_000); + + it.each(["principal_revoked", "resource_disabled", "connection_disabled", "review_head_changed"] as const)( + "retains the actual current %s gate", async scenario => { + const f = await recovered(scenario); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.id, coordinator.decisionId!)); + expect(decision).toMatchObject({ reasonCode: "external_chat_response_wait_authorization_lost", toStatus: "blocked" }); + }, 60_000); + + it.each(["stop", "cancelled_run", "cancelled_task", "owner_changed", "newer_cancelled", "newer_active", + "incomplete_preservation", "missing_receipt"] as const)("does not re-enter old finalization with %s", async scenario => { + await recovered(scenario); + }, 60_000); + + it("settles only the old notice while newer authorized work owns the task and files", async () => { + const f = await recovered("defer_notice"); + const newerId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: newerId, companyId: f.companyId, nativeIssueId: f.issueId, + agentId: f.agentId, runtimeMode: "native", status: "running", nativePhase: "observed" }); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId: newerId, + phase: "observed", leaseOwner: "current-native-owner", leaseExpiresAt: new Date(Date.now() + 60_000) }); + await db.update(issues).set({ status: "in_progress", executionRunId: newerId, checkoutRunId: newerId }) + .where(eq(issues.id, f.issueId)); + const currentPath = path.join(root, f.runId, "host", "review.txt"); + await writeFile(currentPath, "Newer authorized work\n"); + const [currentIssue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const [currentRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, newerId)); + const [currentOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, newerId)); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(1); + await resolveCommittedNativeRecoveryNotices(db, f.binding, true); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0]) + .toMatchObject({ status: "resolved", outcome: "restored" }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(currentIssue); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, newerId)))[0]).toEqual(currentRun); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, newerId)))[0]).toEqual(currentOwner); + expect(await readFile(currentPath, "utf8")).toBe("Newer authorized work\n"); + }, 60_000); + + it("settles the old notice without changing an explicit later Done decision", async () => { + const f = await recovered("defer_notice"); + await issueService(db).update(f.issueId, { status: "done" }); + const [before] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const [result] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + expect(before.status).toBe("done"); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(1); + await resolveCommittedNativeRecoveryNotices(db, f.binding, true); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("resolved"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(before); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)))[0]).toEqual(result); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + }, 60_000); + + it("settles the old notice when newer committed work supersedes its saved snapshot", async () => { + const f = await recovered("defer_notice"); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + const operationId = readSystemRecovery(action.evidence)!.preservationOperationIds[0]; + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, operationId)); + const newerId = randomUUID(), snapshotId = randomUUID(), currentSnapshotId = randomUUID(); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, f.runId)); + const location = { environmentId: lease.environmentId, leaseId: lease.id, providerLeaseId: lease.providerLeaseId!, + remoteCwd: path.join(root, f.runId, "remote") }; + await db.insert(heartbeatRuns).values({ id: newerId, companyId: f.companyId, nativeIssueId: f.issueId, + agentId: f.agentId, runtimeMode: "native", status: "running", nativePhase: "observed" }); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId: f.companyId, sourceIssueId: f.issueId, + mode: "task", strategyType: "isolated", name: "Retained generations", metadata: { remoteWorkspace: { + schema: "paperclip.remote-workspace.v1", epoch: 2, snapshotId: currentSnapshotId, + source: { kind: "empty" }, location, writerRunId: newerId, dirty: true } } }).returning(); + expect(readRemoteWorkspaceAuthority(workspace.metadata)?.writerRunId).toBe(newerId); + const [oldSave, newSave] = await db.insert(workspaceOperations).values([ + { companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId, executionWorkspaceId: workspace.id, + phase: "workspace_snapshot", status: "succeeded", metadata: { boundary: "final", epoch: 1, generation: 2, snapshotId, sha256: "a".repeat(64) } }, + { companyId: f.companyId, issueId: f.issueId, heartbeatRunId: newerId, executionWorkspaceId: workspace.id, + phase: "workspace_snapshot", status: "succeeded", metadata: { boundary: "final", epoch: 2, generation: 3, + snapshotId: currentSnapshotId, sha256: "b".repeat(64) } }, + ]).returning(); + await db.insert(executionWorkspaceSnapshots).values([ + { id: snapshotId, companyId: f.companyId, executionWorkspaceId: workspace.id, sourceOperationId: oldSave.id, + sourceRunId: f.runId, generation: 2, epoch: 1, status: "superseded", sha256: "a".repeat(64), byteSize: 10 }, + { id: currentSnapshotId, companyId: f.companyId, executionWorkspaceId: workspace.id, sourceOperationId: newSave.id, + sourceRunId: newerId, parentSnapshotId: snapshotId, generation: 3, epoch: 2, status: "committed", sha256: "b".repeat(64), byteSize: 20 }, + ]); + await db.update(workspaceOperations).set({ executionWorkspaceId: workspace.id, metadata: { ...operation.metadata, + storage: "remote_workspace", snapshotId, generation: 2, epoch: 1, sha256: "a".repeat(64) } }) + .where(eq(workspaceOperations.id, operationId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { remoteWorkspace: { + schema: "paperclip.remote-workspace-run.v1", workspaceId: workspace.id, ...location, + epoch: 1, state: "saved", snapshotId } } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ status: "in_progress", executionRunId: newerId }).where(eq(issues.id, f.issueId)); + const [beforeIssue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const beforeSnapshots = await db.select().from(executionWorkspaceSnapshots) + .where(eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id)); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(1); + await resolveCommittedNativeRecoveryNotices(db, f.binding, true); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("resolved"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(beforeIssue); + expect((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id)))[0]).toEqual(workspace); + expect(await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id))) + .toEqual(beforeSnapshots); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + }, 60_000); + + it.each(["missing_operation", "foreign_issue", "different_action", "wrong_barrier", "failed_operation", + "nonarray_operations", "empty_operations", "legacy_resume_policy"] as const)("retains a notice with %s preservation proof", async scenario => { + const f = await recovered("defer_notice"); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + const policy = readSystemRecovery(action.evidence)!; + const operationId = policy.preservationOperationIds[0]; + if (scenario === "missing_operation" || scenario === "legacy_resume_policy") await db.delete(workspaceOperations).where(eq(workspaceOperations.id, operationId)); + if (scenario === "legacy_resume_policy") await db.update(issueRecoveryActions).set({ wakePolicy: { kind: "resume_native_run", runId: f.runId } }) + .where(eq(issueRecoveryActions.id, action.id)); + if (scenario === "foreign_issue") await db.update(workspaceOperations).set({ issueId: f.otherIssueId }) + .where(eq(workspaceOperations.id, operationId)); + if (scenario === "different_action") { + const [op] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, operationId)); + await db.update(workspaceOperations).set({ metadata: { ...op.metadata, recoveryActionId: randomUUID() } }) + .where(eq(workspaceOperations.id, operationId)); + } + if (scenario === "wrong_barrier") await db.update(workspaceOperations).set({ metadata: { owningService: "another_service" } }) + .where(and(eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "workspace_finalize"))); + if (scenario === "failed_operation") await db.update(workspaceOperations).set({ status: "failed" }) + .where(eq(workspaceOperations.id, operationId)); + if (scenario === "nonarray_operations" || scenario === "empty_operations") await db.update(issueRecoveryActions).set({ + evidence: { ...action.evidence, systemRecovery: { ...policy, preservationOperationIds: scenario === "empty_operations" ? [] : {} } }, + }).where(eq(issueRecoveryActions.id, action.id)); + const [before] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(0); + await resolveCommittedNativeRecoveryNotices(db, f.binding); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0]).toEqual(before); + }, 60_000); + + it.each(["pending_effect", "failed_effect", "pending_projection", "unknown_old_owner"] as const)( + "retains the old notice with its own %s", async scenario => { + const f = await recovered("defer_notice"); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario === "pending_effect" || scenario === "failed_effect") await db.insert(statusDecisionEffects).values({ + companyId: f.companyId, issueId: f.issueId, decisionId: coordinator.decisionId!, ordinal: 0, + effectKind: "release_checkout", targetType: "issue", targetId: f.issueId, + idempotencyKey: randomUUID(), payload: {}, deliveryState: scenario === "pending_effect" ? "pending" : "failed", + }); + if (scenario === "pending_projection") await db.update(heartbeatRuns).set({ nativePhase: "workspace_finalizing" }) + .where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "unknown_old_owner") await db.update(heartbeatRuns).set({ status: "running", + nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" }).where(eq(heartbeatRuns.id, f.runId)); + const [before] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + expect(await pendingNativeRecoveryNotices(db, f.binding)).toHaveLength(0); + await resolveCommittedNativeRecoveryNotices(db, f.binding, true); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0]).toEqual(before); + }, 60_000); +}); diff --git a/server/src/services/native-runtime/preserved-native-result-finalization.ts b/server/src/services/native-runtime/preserved-native-result-finalization.ts new file mode 100644 index 00000000000..d39c00459ca --- /dev/null +++ b/server/src/services/native-runtime/preserved-native-result-finalization.ts @@ -0,0 +1,135 @@ +import { and, eq, gte, inArray, ne, or, sql } from "drizzle-orm"; +import { completionContracts, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, + nativeRunResults, type Db } from "@paperclipai/db"; +import { z } from "zod"; +import { issueTreeControlService } from "../issue-tree-control.js"; +import { remoteExecutionHasStopped } from "../remote-execution-termination.js"; +import { readSystemRecovery } from "../system-recovery-state.js"; +import { completedWorkspaceRecoveryCondition, readWorkspaceRecovery, verifiedWorkspaceRepairReceipt } from "../workspace-recovery-receipt.js"; +import { nativeSha256 } from "./canonical.js"; + +export const nativeWorkspaceFinalizationBlockSchema = z.object({ + schema: z.literal("paperclip.native-workspace-finalization-block.v1"), + companyId: z.uuid(), issueId: z.uuid(), runId: z.uuid(), agentId: z.uuid(), + resultId: z.uuid(), resultSha256: z.string().min(1), + contractId: z.uuid(), contractSha256: z.string().min(1), + identityContextId: z.uuid().nullable(), responsibleUserId: z.string().nullable(), + statusVersion: z.number().int().nonnegative(), priorDecisionId: z.uuid().nullable(), +}).strict(); + +export type PreservedNativeResultFinalization = { + companyId: string; issueId: string; runId: string; agentId: string; + actionId: string; generationId: string; resultId: string; resultSha256: string; + contractId: string; contractSha256: string; statusVersion: number; + identityContextId: string | null; responsibleUserId: string | null; + priorDecisionId: string | null; statusRestoration: "owned" | "unproven"; + fingerprint: string; +}; + +export function preservedResultHasStop(result: Record | null) { + if (!result) return false; + const cancellation = result.cancellation as { source?: unknown } | undefined; + return result.nativeCancellation != null || typeof result.cancelledByUserId === "string" + || ["user", "board", "agent"].includes(String(result.cancelledByActorType ?? result.interruptedByActorType)) + || result.reassignmentStopRequested === true || ["operator", "queued_message"].includes(String(cancellation?.source)); +} + +export async function readPreservedNativeResultFinalization(db: Db, binding: { + companyId: string; issueId: string; runId: string; agentId: string; +}, lockProof = false): Promise { + const query = db.select({ + issue: { status: issues.status, statusVersion: issues.statusVersion, decisionId: issues.lastStatusDecisionId, + assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, + executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }, + run: { id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, nativeIssueId: heartbeatRuns.nativeIssueId, + issueId: heartbeatRuns.issueId, status: heartbeatRuns.status, createdAt: heartbeatRuns.createdAt, + identityContextId: heartbeatRuns.activeIdentityContextId, responsibleUserId: heartbeatRuns.responsibleUserId, + contractSha256: heartbeatRuns.completionContractSha256, + contextSnapshot: sql>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId')`, + resultJson: sql>`jsonb_build_object( + 'nativeCancellation', ${heartbeatRuns.resultJson}->'nativeCancellation', + 'executionCancellation', ${heartbeatRuns.resultJson}->'executionCancellation', + 'cancellation', ${heartbeatRuns.resultJson}->'cancellation', + 'cancelledByUserId', ${heartbeatRuns.resultJson}->'cancelledByUserId', + 'cancelledByActorType', ${heartbeatRuns.resultJson}->'cancelledByActorType', + 'interruptedByActorType', ${heartbeatRuns.resultJson}->'interruptedByActorType', + 'reassignmentStopRequested', ${heartbeatRuns.resultJson}->'reassignmentStopRequested')`, + runnerProfileJson: sql>`jsonb_build_object( + 'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync', + 'remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace', + 'legacyWorkspaceRecoveryMigration', ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration', + 'nativeWorkspaceFinalizationOwner', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceFinalizationOwner')` }, + result: { id: nativeRunResults.id, sha256: nativeRunResults.canonicalSha256, + disposition: sql`${nativeRunResults.resultJson}#>>'{result,reportedWorkDisposition}'`, + continuation: sql`${nativeRunResults.resultJson}#>>'{result,continuation,kind}'`, + terminalDisposition: sql`${nativeRunResults.resultJson}#>>'{terminal,reportedWorkDisposition}'`, + terminalState: sql`${nativeRunResults.resultJson}#>>'{terminal,runTerminalState}'` }, + contract: { id: completionContracts.id, sha256: completionContracts.canonicalSha256 }, + actionId: issueRecoveryActions.id, + evidence: sql>`jsonb_build_object('runId', ${issueRecoveryActions.evidence}->'runId', + 'systemRecovery', ${issueRecoveryActions.evidence}->'systemRecovery', + 'workspaceRecovery', ${issueRecoveryActions.evidence}->'workspaceRecovery', + 'workspaceFinalizationBlock', ${issueRecoveryActions.evidence}->'workspaceFinalizationBlock')`, + hasRecoveryHistory: sql`exists (select 1 from jsonb_array_elements(${nativeRunFinalizations.recoveryHistory}) entry + where entry->>'kind' = 'preserved_result_finalization' and entry->>'resultId' = ${nativeRunResults.id}::text + and case when entry ? 'preservationGenerationId' + then entry->>'preservationGenerationId' = ${issueRecoveryActions.evidence}#>>'{systemRecovery,generationId}' + else entry->>'generationId' = ${issueRecoveryActions.evidence}#>>'{systemRecovery,generationId}' end)`, + }).from(heartbeatRuns) + .innerJoin(issues, and(eq(issues.companyId, binding.companyId), eq(issues.id, binding.issueId))) + .innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.companyId, binding.companyId), + eq(nativeRunFinalizations.issueId, binding.issueId), eq(nativeRunFinalizations.runId, binding.runId))) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.companyId, binding.companyId), + eq(nativeRunResults.issueId, binding.issueId), eq(nativeRunResults.runId, binding.runId), + eq(nativeRunResults.id, nativeRunFinalizations.resultId), eq(nativeRunResults.schemaStatus, "accepted"))) + .innerJoin(completionContracts, and(eq(completionContracts.companyId, binding.companyId), + eq(completionContracts.issueId, binding.issueId), eq(completionContracts.id, heartbeatRuns.completionContractId), + eq(completionContracts.id, nativeRunResults.completionContractId))) + .innerJoin(issueRecoveryActions, and(eq(issueRecoveryActions.companyId, binding.companyId), + eq(issueRecoveryActions.sourceIssueId, binding.issueId), + eq(issueRecoveryActions.kind, "active_run_watchdog"), + inArray(issueRecoveryActions.cause, ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]), + completedWorkspaceRecoveryCondition(binding.runId), + sql`${issueRecoveryActions.evidence}->>'runId' = ${binding.runId}`)) + .where(and(eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.id, binding.runId), + eq(heartbeatRuns.nativeIssueId, binding.issueId), eq(heartbeatRuns.agentId, binding.agentId), + eq(heartbeatRuns.runtimeMode, "native"), ne(nativeRunFinalizations.phase, "terminal_failure"))) + .orderBy(sql`${issueRecoveryActions.createdAt} desc`).limit(1); + const [row] = await (lockProof ? query.for("share", { noWait: true }) : query); + if (!row || !["failed", "succeeded"].includes(row.run.status) || preservedResultHasStop(row.run.resultJson) + || row.issue.assigneeAgentId !== binding.agentId || row.issue.assigneeUserId + || ["done", "cancelled"].includes(row.issue.status) || row.issue.executionRunId || row.issue.checkoutRunId + || row.run.runnerProfileJson.nativeWorkspaceFinalizationOwner || !row.hasRecoveryHistory + || row.run.contractSha256 !== row.contract.sha256 || row.result.disposition !== "yielded" + || row.result.terminalDisposition !== "yielded" || row.result.terminalState !== "succeeded" + || row.result.continuation !== "response_wake") return null; + const recovery = readSystemRecovery(row.evidence), workspace = readWorkspaceRecovery(row.evidence); + if (recovery?.state !== "complete" || recovery.sourceRunId !== binding.runId || !recovery.preservationOperationIds.length + || workspace?.status !== "recovered" || workspace.runId !== binding.runId) return null; + const [newer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, binding.companyId), ne(heartbeatRuns.id, binding.runId), + gte(heartbeatRuns.createdAt, row.run.createdAt), or(eq(heartbeatRuns.issueId, binding.issueId), + eq(heartbeatRuns.nativeIssueId, binding.issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${binding.issueId}`), + )).limit(1); + if (newer || await issueTreeControlService(db).getActivePauseHoldGate(binding.companyId, binding.issueId) + || !await remoteExecutionHasStopped(db, binding.companyId, binding.runId)) return null; + const receiptId = await verifiedWorkspaceRepairReceipt(db, row.run); + if (!receiptId || !recovery.preservationOperationIds.includes(receiptId)) return null; + const block = nativeWorkspaceFinalizationBlockSchema.safeParse(row.evidence.workspaceFinalizationBlock); + const statusRestoration = row.issue.status === "blocked" && block.success + && block.data.companyId === binding.companyId && block.data.issueId === binding.issueId + && block.data.runId === binding.runId && block.data.agentId === binding.agentId + && block.data.resultId === row.result.id && block.data.resultSha256 === row.result.sha256 + && block.data.contractId === row.contract.id && block.data.contractSha256 === row.contract.sha256 + && block.data.identityContextId === row.run.identityContextId && block.data.responsibleUserId === row.run.responsibleUserId + && block.data.statusVersion === Number(row.issue.statusVersion) + && block.data.priorDecisionId === row.issue.decisionId ? "owned" : "unproven"; + const source: Omit = { + companyId: binding.companyId, issueId: binding.issueId, runId: binding.runId, agentId: binding.agentId, + actionId: row.actionId, generationId: recovery.generationId, + resultId: row.result.id, resultSha256: row.result.sha256, contractId: row.contract.id, + contractSha256: row.contract.sha256, statusVersion: Number(row.issue.statusVersion), + identityContextId: row.run.identityContextId, responsibleUserId: row.run.responsibleUserId, + priorDecisionId: row.issue.decisionId, statusRestoration }; + return { ...source, fingerprint: nativeSha256({ ...source, receiptId }) }; +} diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index 6cd7e8e68a3..008ca97b44b 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -113,6 +113,7 @@ export function arbitrateNativeStatus(input: { governanceResolvedForRun?: boolean; externalChatResponseWaitAuthorization?: "authorized" | "revoked" | "not_applicable"; + hasUnprovenPreservedResultStatus?: boolean; publishedExternalReview?: boolean; /** Server-verified accepted plan request and normal provider terminal. */ planWaitAuthorized?: boolean; @@ -526,6 +527,11 @@ export function arbitrateNativeStatus(input: { input.assessment.continuation?.kind === "response_wake" && input.externalChatResponseWaitAuthorization === "authorized" ) { + if (input.hasUnprovenPreservedResultStatus) return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", toStatus: input.priorIssueStatus, + reasonCode: "preserved_result_status_unproven", unblockDescriptor: null, effects: [], + }; return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: input.publishedExternalReview ? "in_review" : "in_progress", diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 3ea3096b710..ba4413f61ad 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -5,6 +5,7 @@ import { hasNativeWorkspaceFinalizationOwner } from "./native-workspace-finaliza import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; import { lockCurrentNativeCompletionContract } from "./completion-contracts.js"; import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; +import { readPreservedNativeResultFinalization, type PreservedNativeResultFinalization } from "./preserved-native-result-finalization.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js"; @@ -1609,6 +1610,7 @@ export async function commitNativeStatusDecision(input: { supersedesCommittedDecisionId?: string; supersededExternalReviewRequests?: { agentId: string; resultId: string; interactionIds: string[] }; requireExternalChatResponseWaitAuthorization?: { agentId: string; completionContinuationKey?: string }; + requirePreservedResultFinalization?: PreservedNativeResultFinalization; requirePlanWaitSource?: NativePlanWaitSource; requireMonitorWait?: { agentId: string; nextCheckAt: string }; requireMonitorDelivery?: NativeMonitorDelivery; @@ -1844,6 +1846,16 @@ export async function commitNativeStatusDecision(input: { throw new NativeStatusRaceError(); } } + if (input.requirePreservedResultFinalization) { + let current; + try { + current = await readPreservedNativeResultFinalization(tx as unknown as Db, input.requirePreservedResultFinalization, true); + } catch (error) { + if (isExternalChatWaitAuthorizationContention(error)) throw new NativeStatusRaceError(); + throw error; + } + if (!current || current.fingerprint !== input.requirePreservedResultFinalization.fingerprint) throw new NativeStatusRaceError(); + } if (input.requireExternalChatResponseWaitAuthorization) { let authorization; try { diff --git a/server/src/services/preserved-result-finalization.ts b/server/src/services/preserved-result-finalization.ts index 627f0e87bd5..e6aa14d61a9 100644 --- a/server/src/services/preserved-result-finalization.ts +++ b/server/src/services/preserved-result-finalization.ts @@ -2,7 +2,8 @@ import { notifyNativeFinalizationWork } from "./native-finalization-notification import { and, desc, eq, gte, ne, or, sql } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { readSystemRecovery } from "./system-recovery-state.js"; -import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; +import { preservedResultHasStop } from "./native-runtime/preserved-native-result-finalization.js"; +import { readWorkspaceRecovery, verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; /** Re-enter the existing finalizer only after preservation. This does not * recreate a result, grant an executor budget, or apply it over newer work. */ @@ -17,7 +18,8 @@ export async function resumePreservedResultFinalization(db: Db, companyId: strin eq(issues.id, coordinator.issueId))).for("update"); const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))).for("update"); - if (!issue || !run || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status) || + if (!issue || !run || run.status === "cancelled" || preservedResultHasStop(run.resultJson) + || issue.assigneeAgentId !== run.agentId || issue.assigneeUserId || ["done", "cancelled"].includes(issue.status) || issue.executionRunId || issue.checkoutRunId) return false; const [newer] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), ne(heartbeatRuns.id, runId), @@ -37,8 +39,10 @@ export async function resumePreservedResultFinalization(db: Db, companyId: strin sql`${issueRecoveryActions.evidence}->>'runId' = ${runId}`, )).orderBy(desc(issueRecoveryActions.createdAt)).limit(1); const recovery = readSystemRecovery(action?.evidence); - if (!result || recovery?.state !== "complete" || !recovery.preservationOperationIds.length) return false; - if (!await hasVerifiedWorkspaceRepair(tx as unknown as Db, run)) return false; + const workspace = readWorkspaceRecovery(action?.evidence ?? null); + if (!result || recovery?.state !== "complete" || recovery.sourceRunId !== run.id || !recovery.preservationOperationIds.length + || workspace?.status !== "recovered" || workspace.runId !== run.id + || !await verifiedWorkspaceRepairReceipt(tx as unknown as Db, run)) return false; const [intent] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.reason, "task_retry"), @@ -54,6 +58,7 @@ export async function resumePreservedResultFinalization(db: Db, companyId: strin nextAttemptAt: new Date(), failureCode: null, failureDetail: null, updatedAt: new Date(), recoveryHistory: [...coordinator.recoveryHistory, { kind: "preserved_result_finalization", generationId, + preservationGenerationId: recovery.generationId, resultId: result.id, previousFailureCode: coordinator.failureCode, previousFailureDetail: coordinator.failureDetail, at: new Date().toISOString(), }], diff --git a/server/src/services/workspace-recovery-receipt.ts b/server/src/services/workspace-recovery-receipt.ts index be127ec9798..50d3835f11e 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 } from "drizzle-orm"; +import { and, desc, eq, sql, type AnyColumn } 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"; @@ -24,6 +24,33 @@ export function readWorkspaceRecovery(evidence: Record | null) const parsed = workspaceRecoverySchema.safeParse(evidence?.workspaceRecovery); return parsed.success ? parsed.data : null; } + +export function completedWorkspaceRecoveryCondition(runId: string | AnyColumn) { + const operationIds = sql`case when jsonb_typeof(${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}') = 'array' + then ${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}' else '[]'::jsonb end`; + return and( + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,schema}' = 'paperclip.system-recovery.v1'`, + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,policyVersion}' = '1'`, + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,sourceRunId}' = ${runId}::text`, + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete'`, + sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,schema}' = 'paperclip.workspace-recovery.v1'`, + sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,runId}' = ${runId}::text`, + sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' = 'recovered'`, + sql`jsonb_array_length(${operationIds}) > 0`, + sql`not exists (select 1 from jsonb_array_elements_text(${operationIds}) saved(id) where not exists ( + select 1 from ${workspaceOperations} op where op.company_id = ${issueRecoveryActions.companyId} + and op.issue_id = ${issueRecoveryActions.sourceIssueId} and op.heartbeat_run_id::text = ${runId}::text + and op.id::text = saved.id and op.status = 'succeeded' + and op.metadata->>'recoveryActionId' = ${issueRecoveryActions.id}::text + and op.phase in ('workspace_recovery', 'workspace_recovery_snapshot')))`, + sql`exists (select 1 from ${workspaceOperations} op where op.company_id = ${issueRecoveryActions.companyId} + and op.issue_id = ${issueRecoveryActions.sourceIssueId} and op.heartbeat_run_id::text = ${runId}::text + and op.phase = 'workspace_finalize' and op.status = 'succeeded' + and op.metadata->>'owningService' = 'workspace_recovery' + and op.metadata->'preservationOperationIds' = ${operationIds})`, + ); +} + export function workspaceRecoveryDisplay(evidence: Record, runId: string | null): ExecutionBlocker["workspaceRecovery"] { const state = readWorkspaceRecovery(evidence); if (!state || state.runId !== runId) return undefined;