From f5e5b05a333ad8ee5d5dbbca2c3c67693a4844aa Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 30 Oct 2026 10:57:53 -0500 Subject: [PATCH] Preserve native workspace recovery across Stop and sandbox reuse (#18332) Land the thirteen-file native recovery fix independently from #18245. Retain verified Stop evidence and queued user comments. Retire stale preparation ownership atomically and require paired sandbox lease handoff receipts before workspace recovery. Co-Authored-By: Paperclip --- .../modules/run-dispatch/adapters/postgres.ts | 90 +- .../modules/wake-queue/adapters/postgres.ts | 37 + .../modules/wake-queue/application/ports.ts | 4 +- .../wake-queue/application/use-cases.test.ts | 42 + .../wake-queue/application/use-cases.ts | 4 +- .../src/services/environment-lease-handoff.ts | 74 ++ server/src/services/environments.ts | 23 +- server/src/services/heartbeat/run-control.ts | 8 +- .../heartbeat/runtime-selection.test.ts | 67 ++ .../services/heartbeat/runtime-selection.ts | 3 + .../src/services/workspace-recovery-source.ts | 29 +- .../src/services/workspace-recovery.test.ts | 783 +++++++++++++++++- server/src/services/workspace-recovery.ts | 84 +- 13 files changed, 1220 insertions(+), 28 deletions(-) create mode 100644 server/src/services/environment-lease-handoff.ts diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index efa70ef5dab..cf9946bafe9 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,8 @@ +import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restore-recovery-state.js"; +import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js"; +import { isPreDispatchExecutionWaitVerified } from "../../../services/pre-dispatch-execution-wait.js"; +import { isUserCommentContinuationReason } from "../../../services/explicit-native-continuation.js"; +import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; @@ -7,7 +12,7 @@ import { hasConversationContinuationPolicy } from "../../../services/conversatio import { getExecutionBlocker } from "../../../services/execution-blocker.js"; import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js"; import { getNativeReviewAssignment } from "../../../services/native-runtime/native-review-participant.js"; -import { and, asc, eq, gte, inArray, lte, or, sql } from "drizzle-orm"; +import { and, asc, eq, gte, inArray, isNull, lte, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agentWakeupRequests, @@ -22,7 +27,7 @@ import { nativeRunFinalizations, projects, } from "@paperclipai/db"; -import { ISSUE_DISPOSITION_REPAIR_RETRY_REASON } from "@paperclipai/shared"; +import { ISSUE_DISPOSITION_REPAIR_RETRY_REASON, isUuidLike } from "@paperclipai/shared"; import { withAccountingTransaction } from "../../../services/accounting-transaction.js"; import type { ActivityPublication } from "../../../services/activity-log.js"; import { parseObject } from "../../../adapters/utils.js"; @@ -918,6 +923,80 @@ export function createPostgresRunDispatchAdapter( return transactionResult.outcome; } + async function deferUnstartedUserComment(tx: Db, run: HeartbeatRun, issueId: string, now: Date) { + if (!run.wakeupRequestId || !await isPreDispatchExecutionWaitVerified(tx, run)) return null; + const wait = parseObject(run.resultJson?.executionWait); + const [preservation] = await tx.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + sql`${issueRecoveryActions.id}::text = ${wait.recoveryActionId}`, + )).for("update"); + const [source] = preservation ? await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "cancelled"), + sql`${heartbeatRuns.id}::text = ${preservation.evidence.runId}`, + )) : []; + const baseline = readNativeWorkspaceSyncReference(source?.runnerProfileJson?.nativeWorkspaceSync); + if (!preservation || !source || !hasRequiredWorkspaceRecovery(source.resultJson) || !baseline || baseline.state === "finalized" || + preservation.fingerprint !== `native-workspace:${source.id}`) return null; + const context = parseObject(run.contextSnapshot), commentId = deriveCommentId(context); + const commentIds = queuedCommentIdsFromRunContext(context); + if (!commentId || !commentIds.includes(commentId) || !isUserCommentContinuationReason(readNonEmptyString(context.wakeReason))) return null; + const [wake] = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id), + inArray(agentWakeupRequests.status, ["queued", "claimed"]), + eq(agentWakeupRequests.requestedByActorType, "user"), + )).for("update"); + const payload = parseObject(wake?.payload); + const originalCommentId = readNonEmptyString(payload.commentId); + if (!wake?.requestedByActorId || payload.issueId !== issueId || + !originalCommentId || !commentIds.includes(originalCommentId)) return null; + const comments = await tx.select().from(issueComments).where(and( + eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), + inArray(issueComments.id, commentIds), eq(issueComments.authorType, "user"), + sql`${issueComments.deletedAt} is null`, sql`${issueComments.createdByRunId} is null`, + )); + if (comments.length !== commentIds.length || comments.some(comment => + !comment.body.trim() || !comment.authorUserId) || + comments.find(comment => comment.id === originalCommentId)?.authorUserId !== wake.requestedByActorId) return null; + const sourceIds = context.queuedCommentSourceWakeIds; + if (sourceIds !== undefined && (!Array.isArray(sourceIds) || sourceIds.some(id => typeof id !== "string" || !isUuidLike(id)))) return null; + const sourceWakeIds = Array.isArray(sourceIds) ? [...new Set(sourceIds as string[])] : []; + const adopted = sourceWakeIds.length ? await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "coalesced"), + inArray(agentWakeupRequests.id, sourceWakeIds), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, + )).for("update") : []; + if (adopted.length !== sourceWakeIds.length) return null; + const mergeTargets = [wake.id, ...adopted.map(receipt => receipt.id)]; + const coalesced = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.requestedByActorType, "user"), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, + or(and(eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.reason, "issue_execution_same_name")), + and(or(isNull(agentWakeupRequests.runId), eq(agentWakeupRequests.runId, run.id)), + inArray(sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId'`, mergeTargets))), + )).for("update"); + const authors = new Map(comments.map(comment => [comment.id, comment.authorUserId])); + const recordedCommentIds = new Set([wake, ...adopted, ...coalesced].flatMap(receipt => { + const recordedPayload = parseObject(receipt.payload), actorId = readNonEmptyString(receipt.requestedByActorId); + const recordedCommentId = readNonEmptyString(recordedPayload.commentId); + return receipt.requestedByActorType === "user" && recordedPayload.issueId === issueId && actorId && + recordedCommentId && authors.get(recordedCommentId) === actorId ? [recordedCommentId] : []; + })); + if (commentIds.some(id => !recordedCommentIds.has(id))) return null; + const [deferred] = await tx.update(agentWakeupRequests).set({ + status: "deferred_issue_execution", runId: null, claimedAt: null, finishedAt: null, error: null, + payload: withQueuedCommentIdsInWakePayload({ ...payload, _paperclipWakeContext: context, + executionWait: { ...parseObject(run.resultJson?.executionWait), reason: "execution_recovery", + message: "Waiting for execution recovery. Your message is saved." } }, commentIds), updatedAt: now, + }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId), + eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id), + inArray(agentWakeupRequests.status, ["queued", "claimed"]))).returning({ id: agentWakeupRequests.id }); + return deferred?.id ?? null; + } + async function cancelStaleRunInTx( tx: Db, run: HeartbeatRun, @@ -969,7 +1048,9 @@ export function createPostgresRunDispatchAdapter( // this write must not overwrite whatever status won it. if (!row) return { outcome: "lost_race" }; - if (row.wakeupRequestId) { + const deferredWakeRequestId = expectedStatus === "queued" && decision.errorCode === "execution_reconciliation_required" + ? await deferUnstartedUserComment(tx, row, issueId, now) : null; + if (row.wakeupRequestId && !deferredWakeRequestId) { await tx .update(agentWakeupRequests) .set({ status: "skipped", finishedAt: now, error: decision.reason, updatedAt: now }) @@ -977,6 +1058,9 @@ export function createPostgresRunDispatchAdapter( and( eq(agentWakeupRequests.id, row.wakeupRequestId), eq(agentWakeupRequests.companyId, row.companyId), + eq(agentWakeupRequests.agentId, row.agentId), + eq(agentWakeupRequests.runId, row.id), + inArray(agentWakeupRequests.status, ["queued", "claimed"]), ), ); } diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 5b5cdb89519..b888ee51fb3 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1001,6 +1001,40 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { async mergeIntoExistingDeferredWake(scope, input) { const tx = requireAdmissionTx(scope, input.companyId); if (input.manualUserWakeActorId) await notifyChatInboundWakeNotices(tx, { companyId: input.companyId }); + const [prior] = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.existingDeferredWakeId), + eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS), + )).for("update"); + if (!prior) throw new Error("wake-queue: the deferred wake to merge into was not found for this company"); + const priorPayload = parseObject(prior.payload), issueId = readNonEmptyString(priorPayload.issueId); + const priorCommentId = readNonEmptyString(priorPayload.commentId), priorActor = readNonEmptyString(prior.requestedByActorId); + const [priorComment] = prior.requestedByActorType === "user" && priorActor && issueId && isUuidLike(issueId) && + priorCommentId && isUuidLike(priorCommentId) ? await tx.select({ id: issueComments.id, body: issueComments.body }).from(issueComments).where(and( + eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, issueId), eq(issueComments.id, priorCommentId), + eq(issueComments.authorType, "user"), eq(issueComments.authorUserId, priorActor), + isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), + )) : []; + if ((input.coalescedReceipt || priorComment?.body.trim()) && + (readNonEmptyString(input.mergedPayload.issueId) !== issueId || (input.coalescedReceipt && + (input.coalescedReceipt.agentId !== prior.agentId || readNonEmptyString(input.coalescedReceipt.payload.issueId) !== issueId)))) + throw new Error("wake-queue: merged receipt does not match the saved queue scope"); + if (priorComment?.body.trim()) { + const [recorded] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, prior.agentId), + eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.requestedByActorType, "user"), + eq(agentWakeupRequests.requestedByActorId, priorActor!), + prior.runId ? eq(agentWakeupRequests.runId, prior.runId) : isNull(agentWakeupRequests.runId), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + sql`${agentWakeupRequests.payload}->>'commentId' = ${priorComment.id}`, + sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId' = ${prior.id}`, + )).limit(1); + if (!recorded) await tx.insert(agentWakeupRequests).values({ + companyId: input.companyId, agentId: prior.agentId, source: prior.source, triggerDetail: prior.triggerDetail, + reason: prior.reason, payload: { issueId, commentId: priorComment.id, coalescedIntoWakeupRequestId: prior.id }, + status: "coalesced", coalescedCount: 1, requestedByActorType: "user", requestedByActorId: priorActor, + requestedAt: prior.requestedAt, finishedAt: new Date(), runId: prior.runId, + }); + } const rows = await tx .update(agentWakeupRequests) .set({ @@ -1032,6 +1066,9 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { await tx.insert(agentWakeupRequests).values({ ...input.coalescedReceipt, companyId: input.companyId, + agentId: prior.agentId, + runId: prior.runId, + payload: { ...input.coalescedReceipt.payload, coalescedIntoWakeupRequestId: prior.id }, status: "coalesced", coalescedCount: 1, finishedAt: new Date(), diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index c37aa995e0e..8827c82e4ca 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -322,7 +322,9 @@ export type DurableWakeAdmissionReceipt = { requestedAt: Date; }; -export type CoalescedDeferredAdmissionReceipt = DurableWakeAdmissionReceipt & { +export type CoalescedDeferredAdmissionReceipt = + Omit & + Partial> & { agentId: string; source: string; triggerDetail: string | null; diff --git a/server/src/modules/wake-queue/application/use-cases.test.ts b/server/src/modules/wake-queue/application/use-cases.test.ts index 0c7e66f9dd1..7f605e70f30 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -1129,6 +1129,48 @@ describe("admitWakeBehindIssueExecution", () => { expect(writer.coalesceIntoActiveExecutionRun).not.toHaveBeenCalled(); }); + it("records an ordinary merged comment under its own actor without requiring durable receipt identity", async () => { + const writer = createFakeAdmissionWriter(); + const reader = createFakeAdmissionReader({ findExistingDeferredWake: vi.fn(async () => ({ + id: "existing-deferred-wake", runId: null, payload: { issueId: "issue-1" }, deferredContext: {}, coalescedCount: 1, + })) }); + const admit = createAdmitWakeBehindIssueExecution({ reader, writer, + helpers: createFakeAdmissionHelpers({ shouldDeferFollowupWakeForSameIssue: vi.fn(() => true) }) }); + await expect(admit(SCOPE, admissionInput({ wakeCommentId: "second-comment", requestedByActorId: "second-user", + payload: { commentId: "second-comment", coalescedIntoWakeupRequestId: "forged-target" }, + }))).resolves.toEqual({ kind: "deferred" }); + const input = vi.mocked(writer.mergeIntoExistingDeferredWake).mock.calls[0][1]; + expect(input.coalescedReceipt).toEqual({ agentId: "wake-agent", source: "on_demand", triggerDetail: null, + reason: null, payload: { issueId: "issue-1", commentId: "second-comment", coalescedIntoWakeupRequestId: "existing-deferred-wake" }, + requestedByActorType: "user", requestedByActorId: "second-user", idempotencyKey: null, runId: null }); + expect(input.coalescedReceipt).not.toHaveProperty("id"); + expect(input.coalescedReceipt).not.toHaveProperty("requestedAt"); + expect(reader.matchesActiveWakeActor).not.toHaveBeenCalled(); + }); + + it("binds authoritative issue scope when durable merged input omits it from payload", async () => { + const writer = createFakeAdmissionWriter(); + const durableReceipt = { id: "own-request", requestedAt: new Date("2026-09-10T00:00:00Z") }; + const admit = createAdmitWakeBehindIssueExecution({ writer, + reader: createFakeAdmissionReader({ findExistingDeferredWake: vi.fn(async () => ({ + id: "saved-wake", runId: null, payload: { issueId: "issue-1" }, deferredContext: {}, coalescedCount: 0, + })) }), helpers: createFakeAdmissionHelpers({ shouldDeferFollowupWakeForSameIssue: vi.fn(() => true) }) }); + await admit(SCOPE, admissionInput({ durableReceipt, payload: { text: "saved input" } })); + expect(vi.mocked(writer.mergeIntoExistingDeferredWake).mock.calls[0][1].coalescedReceipt).toEqual(expect.objectContaining({ + ...durableReceipt, payload: { text: "saved input", issueId: "issue-1", coalescedIntoWakeupRequestId: "saved-wake" }, + })); + }); + + it.each(["agent", "system"])("does not create a user-comment authority receipt for an ordinary %s merge", async actor => { + const writer = createFakeAdmissionWriter(); + const admit = createAdmitWakeBehindIssueExecution({ writer, + reader: createFakeAdmissionReader({ findExistingDeferredWake: vi.fn(async () => ({ + id: "existing-deferred-wake", runId: null, payload: { issueId: "issue-1" }, deferredContext: {}, coalescedCount: 1, + })) }), helpers: createFakeAdmissionHelpers({ shouldDeferFollowupWakeForSameIssue: vi.fn(() => true) }) }); + await admit(SCOPE, admissionInput({ wakeCommentId: "second-comment", requestedByActorType: actor })); + expect(vi.mocked(writer.mergeIntoExistingDeferredWake).mock.calls[0][1]).not.toHaveProperty("coalescedReceipt"); + }); + it("gives manual input its own run boundary even when the active receipt has the same requester", async () => { const writer = createFakeAdmissionWriter(); const admit = createAdmitWakeBehindIssueExecution({ diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 1e3ffa93609..069143cca94 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -923,7 +923,8 @@ export function createAdmitWakeBehindIssueExecution(deps: { mergedPayload, nextCoalescedCount: (existingDeferred.coalescedCount ?? 0) + 1, ...(manualUserWakeActorId ? { manualUserWakeActorId } : {}), - ...(input.durableReceipt + ...(input.durableReceipt || (input.requestedByActorType === "user" && + readNonEmptyString(input.requestedByActorId) && readNonEmptyString(input.wakeCommentId)) ? { coalescedReceipt: { ...input.durableReceipt, @@ -933,6 +934,7 @@ export function createAdmitWakeBehindIssueExecution(deps: { reason: input.reason ?? null, payload: { ...(input.payload ?? {}), + issueId: input.issueId, coalescedIntoWakeupRequestId: existingDeferred.id, }, requestedByActorType: input.requestedByActorType, diff --git a/server/src/services/environment-lease-handoff.ts b/server/src/services/environment-lease-handoff.ts new file mode 100644 index 00000000000..41df34ec05b --- /dev/null +++ b/server/src/services/environment-lease-handoff.ts @@ -0,0 +1,74 @@ +import { randomUUID } from "node:crypto"; +import type { environmentLeases } from "@paperclipai/db"; + +type Lease = typeof environmentLeases.$inferSelect; +type Binding = Pick; +const outgoingKey = "reusableLeaseSuccessor"; +const incomingKey = "reusableLeasePredecessor"; + +export function withoutReusableLeaseHandoff(metadata: Record | null | undefined) { + if (metadata == null) return null; + const result = { ...metadata }; + delete result[outgoingKey]; + delete result[incomingKey]; + return result; +} + +export function preserveReusableLeaseHandoff(previous: Record | null, + metadata: Record | null | undefined) { + const result = withoutReusableLeaseHandoff(metadata); + const proof = Object.fromEntries([outgoingKey, incomingKey] + .filter(key => Object.hasOwn(previous ?? {}, key)).map(key => [key, previous![key]])); + return Object.keys(proof).length ? { ...result, ...proof } : result; +} + +function handoffBinding(previous: Binding, current: Binding) { + return { schema: "paperclip.reusable-lease-handoff.v1", + previousLeaseId: previous.id, successorLeaseId: current.id, + previousRunId: previous.heartbeatRunId, successorRunId: current.heartbeatRunId, + companyId: current.companyId, environmentId: current.environmentId, + workspaceId: current.executionWorkspaceId, issueId: current.issueId, + agentId: current.metadata?.agentId ?? null, provider: current.provider, + providerLeaseId: current.providerLeaseId, recordedAt: current.acquiredAt.toISOString() }; +} + +export function createReusableLeaseHandoff(previous: Binding, current: Binding) { + return { ...handoffBinding(previous, current), handoffId: randomUUID() }; +} + +export function retiredReusableLeasePredecessors(current: Lease, history: Lease[]): Lease[] | null { + if (Object.hasOwn(current.metadata ?? {}, outgoingKey)) return null; + const remaining = new Map(history.map(lease => [lease.id, lease])); + if (remaining.size !== history.length) return null; + const result: Lease[] = []; + let successor = current; + while (remaining.size) { + const raw = successor.metadata?.[incomingKey]; + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const receipt = raw as Record; + const previous = remaining.get(String(receipt.previousLeaseId)); + if (!previous || previous.status !== "expired" || previous.cleanupStatus !== "success" || + previous.leasePolicy !== "reuse_by_environment" || !previous.heartbeatRunId || + !successor.heartbeatRunId || previous.heartbeatRunId === successor.heartbeatRunId || + typeof receipt.handoffId !== "string" || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(receipt.handoffId)) return null; + const expected = { ...handoffBinding(previous, successor), handoffId: receipt.handoffId }; + const outgoing = previous.metadata?.[outgoingKey]; + if (!outgoing || typeof outgoing !== "object" || Array.isArray(outgoing) || + Object.keys(receipt).length !== Object.keys(expected).length || + Object.keys(outgoing).length !== Object.keys(expected).length || + Object.entries(expected).some(([key, value]) => receipt[key] !== value || + (outgoing as Record)[key] !== value) || + !previous.environmentId || !previous.issueId || typeof previous.metadata?.agentId !== "string" || + previous.companyId !== current.companyId || previous.environmentId !== current.environmentId || + previous.executionWorkspaceId !== current.executionWorkspaceId || previous.issueId !== current.issueId || + previous.metadata.agentId !== current.metadata?.agentId || + previous.provider !== current.provider || previous.providerLeaseId !== current.providerLeaseId || + previous.releasedAt?.getTime() !== successor.acquiredAt.getTime() || + previous.acquiredAt.getTime() > successor.acquiredAt.getTime()) return null; + result.push(previous); + remaining.delete(previous.id); + successor = previous; + } + return Object.hasOwn(successor.metadata ?? {}, incomingKey) ? null : result; +} diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index b6da210efa5..ea72243e464 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1,3 +1,4 @@ +import { createReusableLeaseHandoff, preserveReusableLeaseHandoff, withoutReusableLeaseHandoff } from "./environment-lease-handoff.js"; import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; @@ -1465,7 +1466,9 @@ export function environmentService(db: Db) { assertCompanyBinding?: boolean; }): Promise => { const now = new Date(); + const replacementLeaseId = input.replacesReusableLeaseId ? randomUUID() : null; const values = { + ...(replacementLeaseId ? { id: replacementLeaseId } : {}), companyId: input.companyId, environmentId: input.environmentId, executionWorkspaceId: input.executionWorkspaceId ?? null, @@ -1481,7 +1484,7 @@ export function environmentService(db: Db) { releasedAt: null, failureReason: null, cleanupStatus: null, - metadata: withoutCleanupVerification(input.metadata), + metadata: withoutReusableLeaseHandoff(withoutCleanupVerification(input.metadata)), createdAt: now, updatedAt: now, }; @@ -1571,6 +1574,7 @@ export function environmentService(db: Db) { eq(environmentLeases.id, input.replacesReusableLeaseId), eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, input.environmentId), + sql`${environmentLeases.provider} is not distinct from ${input.provider ?? null}`, input.executionWorkspaceId ? eq(environmentLeases.executionWorkspaceId, input.executionWorkspaceId) : and( @@ -1589,14 +1593,21 @@ export function environmentService(db: Db) { ]), ), ) - .returning({ id: environmentLeases.id }); + .returning(); if (retired.length !== 1) { throw conflict( "Reusable sandbox lease ownership changed during acquisition.", ); } + const handoff = createReusableLeaseHandoff(retired[0]!, { ...values, id: replacementLeaseId! }); + values.metadata = { ...values.metadata, reusableLeasePredecessor: handoff }; + await tx.update(environmentLeases).set({ metadata: { + ...retired[0]!.metadata, reusableLeaseSuccessor: handoff, + } }).where(eq(environmentLeases.id, retired[0]!.id)); } if (input.reusesReusableLeaseId) { + const [previous] = await tx.select({ metadata: environmentLeases.metadata }).from(environmentLeases) + .where(and(eq(environmentLeases.id, input.reusesReusableLeaseId), eq(environmentLeases.companyId, input.companyId))).for("update"); const reacquired = await tx .update(environmentLeases) .set({ @@ -1606,7 +1617,7 @@ export function environmentService(db: Db) { expiresAt: input.expiresAt ?? null, failureReason: null, cleanupStatus: null, - metadata: withoutCleanupVerification(input.metadata), + metadata: preserveReusableLeaseHandoff(previous?.metadata ?? null, withoutCleanupVerification(input.metadata)), updatedAt: now, }) .where( @@ -1789,7 +1800,7 @@ export function environmentService(db: Db) { releasedAt: now, failureReason: input.failureReason, cleanupStatus: "failed", - metadata: withoutCleanupVerification(input.metadata), + metadata: withoutReusableLeaseHandoff(withoutCleanupVerification(input.metadata)), createdAt: now, updatedAt: now, }) @@ -1807,13 +1818,13 @@ export function environmentService(db: Db) { metadata: Record | null, ): Promise => { const row = await db.transaction(async tx => { - const [lease] = await tx.select({ companyId: environmentLeases.companyId, heartbeatRunId: environmentLeases.heartbeatRunId }) + const [lease] = await tx.select({ companyId: environmentLeases.companyId, heartbeatRunId: environmentLeases.heartbeatRunId, metadata: environmentLeases.metadata }) .from(environmentLeases).where(eq(environmentLeases.id, id)).for("update"); if (!lease) return null; await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); return tx.update(environmentLeases) .set({ - metadata: withoutCleanupVerification(metadata), + metadata: preserveReusableLeaseHandoff(lease.metadata, withoutCleanupVerification(metadata)), lastUsedAt: new Date(), updatedAt: new Date(), }) diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index f8931615322..820f1e50c61 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import { observeStoppedWorkspaceRecovery } from "../workspace-recovery.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { resumeDeferredChatWake } from "../durable-chat-wakeup.js"; @@ -545,11 +546,12 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont )) : []; // The ordinary path still enforces task holds, ownership, and native session // cleanup before starting a provider. - const stoppedNativeContinuation = currentRun && isAcknowledgedNativeStop(currentRun) && + const stoppedNativeExecution = currentRun && isAcknowledgedNativeStop(currentRun) && !activeRunExecutions.has(run.id) && !coordinator?.leaseOwner && ["terminal_failure", "applied"].includes(coordinator?.phase ?? "") && - await acknowledgedNativeStopExecutionHasStopped(db, currentRun) && - !(await getExecutionBlocker(db, run.companyId, issueId)); + await acknowledgedNativeStopExecutionHasStopped(db, currentRun); + if (stoppedNativeExecution) await observeStoppedWorkspaceRecovery(db, { companyId: run.companyId, issueId, runId: run.id }); + const stoppedNativeContinuation = stoppedNativeExecution && !(await getExecutionBlocker(db, run.companyId, issueId)); const legacyContinuation = run.runtimeMode === "legacy" && hasConversationContinuationPolicy((await getRunExecutionWait(run.companyId, run.id))?.resultJson) && !(await getExecutionBlocker(db, run.companyId, issueId)); diff --git a/server/src/services/heartbeat/runtime-selection.test.ts b/server/src/services/heartbeat/runtime-selection.test.ts index c7708d5e078..cdfae206f44 100644 --- a/server/src/services/heartbeat/runtime-selection.test.ts +++ b/server/src/services/heartbeat/runtime-selection.test.ts @@ -14,6 +14,7 @@ import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mod import { claimNativeRestartRecoveries } from "../native-runtime/native-restart-recovery.js"; import { canonicalNativeRuntimeContextDigest } from "../../vendor/paperclip-runner/index.js"; import { selectHeartbeatRuntime, type HeartbeatRuntimeSelectionInput } from "./runtime-selection.js"; +import { legacyControllerClaim } from "../legacy-controller-lease.js"; // Keep the real contract, session, and runtime-selection transactions. Replace only // filesystem materialization and sandbox synchronization at their external boundaries. @@ -371,4 +372,70 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () = expect(input.stopControllerLease).not.toHaveBeenCalled(); expect(buildNativeRuntimeContext).not.toHaveBeenCalled(); }); + async function claimPreparation(f: Awaited>) { + await db.update(heartbeatRuns).set(legacyControllerClaim("legacy")).where(eq(heartbeatRuns.id, f.run.id)); + f.input.run = await read(f.run.id); + expect(f.input.run.controllerBootId).toBeTruthy(); + expect(f.input.run.controllerLeaseExpiresAt!.getTime()).toBeGreaterThan(Date.now()); + expect(f.input.run.executionStage).toBe("preparing"); + } + + it.each(["local", "remote"] as const)("retires actual preparation ownership atomically at native admission: %s", async target => { + const f = await fixture(); + await claimPreparation(f); + if (target === "remote") f.input.workspace.executionTarget = sandbox; + vi.mocked(prepareNativeWorkspaceSync).mockImplementationOnce(async () => { + const saved = await read(f.run.id); + expect(saved).toMatchObject({ runtimeMode: "native", controllerBootId: null, controllerLeaseExpiresAt: null, executionStage: null }); + expect(await coordinators(f.run.id)).toHaveLength(1); + expect(f.input.stopControllerLease).toHaveBeenCalledOnce(); + return null; + }); + expect((await selectHeartbeatRuntime(db, f.input)).selected).toBe(true); + expect(await read(f.run.id)).toMatchObject({ runtimeMode: "native", controllerBootId: null, controllerLeaseExpiresAt: null, executionStage: null }); + }); + + it.each(["foreign_boot", "expired", "cancelled"] as const)("keeps preparation ownership when native admission is denied: %s", async denied => { + const f = await fixture(); + await claimPreparation(f); + const patch = denied === "foreign_boot" ? { controllerBootId: randomUUID() } : denied === "expired" ? { controllerLeaseExpiresAt: new Date(0) } : { status: "cancelled" as const }; + await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.run.id)); + const before = await read(f.run.id); + expect(await selectHeartbeatRuntime(db, f.input)).toEqual({ selected: false }); + const after = await read(f.run.id); + expect(after).toMatchObject({ runtimeMode: "legacy", controllerBootId: before.controllerBootId, controllerLeaseExpiresAt: before.controllerLeaseExpiresAt, executionStage: before.executionStage }); + expect(await coordinators(f.run.id)).toHaveLength(0); + expect(f.input.stopControllerLease).not.toHaveBeenCalled(); + expect(prepareNativeWorkspaceSync).not.toHaveBeenCalled(); + }); + + it("rolls the preparation handoff back if the native coordinator cannot commit", async () => { + const f = await fixture(); + await claimPreparation(f); + const before = await read(f.run.id); + await db.execute(sql`CREATE FUNCTION reject_test_native_coordinator() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'controlled_native_coordinator_rejected'; END $$`); + await db.execute(sql`CREATE TRIGGER reject_test_native_coordinator BEFORE INSERT ON native_run_finalizations FOR EACH ROW EXECUTE FUNCTION reject_test_native_coordinator()`); + try { + await expect(selectHeartbeatRuntime(db, f.input)).rejects.toMatchObject({ cause: { code: "P0001", message: "controlled_native_coordinator_rejected" } }); + expect(await read(f.run.id)).toMatchObject({ runtimeMode: "legacy", controllerBootId: before.controllerBootId, controllerLeaseExpiresAt: before.controllerLeaseExpiresAt, executionStage: before.executionStage, nativeSessionId: null }); + expect(await coordinators(f.run.id)).toHaveLength(0); + expect(f.input.stopControllerLease).not.toHaveBeenCalled(); + expect(prepareNativeWorkspaceSync).not.toHaveBeenCalled(); + } finally { + await db.execute(sql`DROP TRIGGER reject_test_native_coordinator ON native_run_finalizations`); + await db.execute(sql`DROP FUNCTION reject_test_native_coordinator()`); + } + }); + + it("keeps committed native authority after a later workspace synchronization failure", async () => { + const f = await fixture(); + await claimPreparation(f); + f.input.workspace.executionTarget = sandbox; + vi.mocked(prepareNativeWorkspaceSync).mockRejectedValueOnce(new Error("controlled_workspace_sync_failure")); + await expect(selectHeartbeatRuntime(db, f.input)).rejects.toThrow("controlled_workspace_sync_failure"); + expect(await read(f.run.id)).toMatchObject({ runtimeMode: "native", controllerBootId: null, controllerLeaseExpiresAt: null, executionStage: null }); + expect(await coordinators(f.run.id)).toHaveLength(1); + expect(f.input.stopControllerLease).toHaveBeenCalledOnce(); + }); + }); diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index f5909a970c5..c3402d4d348 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -946,6 +946,9 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele .update(heartbeatRuns) .set({ runtimeMode: "native", + controllerBootId: null, + controllerLeaseExpiresAt: null, + executionStage: null, runtimeModeResolverVersion: lockedRun.runtimeModeResolverVersion ?? nativeRuntimeResolution.resolverVersion, diff --git a/server/src/services/workspace-recovery-source.ts b/server/src/services/workspace-recovery-source.ts index 5afb3274608..ce0c26de49a 100644 --- a/server/src/services/workspace-recovery-source.ts +++ b/server/src/services/workspace-recovery-source.ts @@ -1,3 +1,4 @@ +import { retiredReusableLeasePredecessors } from "./environment-lease-handoff.js"; import { and, eq, inArray, ne, or, sql } from "drizzle-orm"; import { environmentLeases, environments, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { adapterExecutionControls } from "./adapter-execution-control.js"; @@ -131,10 +132,32 @@ export async function inspectWorkspaceRecoverySource(db: Db, input: WorkspaceRec if (descriptor && remoteCwd !== descriptor.remoteCwd) throw invalid("cwd_reference_mismatch"); if (descriptor && lease.executionWorkspaceId && lease.executionWorkspaceId !== descriptor.workspaceId) throw invalid("workspace_reference_mismatch"); - const [other] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + const history = await db.select().from(environmentLeases).where(and( ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider), - eq(environmentLeases.providerLeaseId, lease.providerLeaseId))).limit(1); - if (other) throw new SystemRecoveryFailure("allocation_ownership_ambiguous", "service_required"); + eq(environmentLeases.providerLeaseId, lease.providerLeaseId))); + const predecessors = retiredReusableLeasePredecessors(lease, history); + if (!predecessors || (predecessors.length && (run.runtimeMode !== "native" || !descriptor || lease.metadata?.agentId !== run.agentId))) + throw new SystemRecoveryFailure("allocation_ownership_ambiguous", "service_required"); + const predecessorRuns = predecessors.map(item => item.heartbeatRunId!); + const priors = predecessorRuns.length ? await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), inArray(heartbeatRuns.id, predecessorRuns))) : []; + const priorOwners = predecessorRuns.length ? await db.select({ runId: nativeRunFinalizations.runId, + phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), inArray(nativeRunFinalizations.runId, predecessorRuns), + eq(nativeRunFinalizations.issueId, input.issueId))) : []; + for (const predecessor of predecessors) { + const prior = priors.find(item => item.id === predecessor.heartbeatRunId); + const priorReference = readNativeWorkspaceSyncReference(prior?.runnerProfileJson?.nativeWorkspaceSync); + const priorOwner = priorOwners.find(item => item.runId === predecessor.heartbeatRunId); + if (!prior || prior.runtimeMode !== "native" || prior.agentId !== run.agentId || prior.nativeIssueId !== input.issueId || + !terminalWorkspaceRunStates.includes(prior.status) || adapterExecutionControls.has(prior.id) || + (prior.controllerLeaseExpiresAt && prior.controllerLeaseExpiresAt > new Date()) || + !priorOwner || priorOwner.leaseOwner || !["committed", "terminal_failure", "applied"].includes(priorOwner.phase) || + !priorReference || priorReference.state !== "finalized" || !priorReference.finalHostSha256 || + priorReference.leaseId !== predecessor.id || priorReference.providerLeaseId !== lease.providerLeaseId || + priorReference.workspaceId !== descriptor!.workspaceId || priorReference.remoteCwd !== remoteCwd) + throw new SystemRecoveryFailure("allocation_ownership_ambiguous", "service_required"); + } // Snapshot export writes only a new staging directory. A native baseline // merge additionally needs exclusive ownership of its host destination. if (descriptor) { diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 5d044015334..38d869f3bc5 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -1,3 +1,22 @@ +import { createAdmissionTransactionScope, createWakeAdmissionReader, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; +import { createAdmitWakeBehindIssueExecution } from "../modules/wake-queue/application/use-cases.js"; +import { filterZombieCoalesceTarget } from "./heartbeat.js"; +import { mergeCoalescedContextSnapshot } from "./heartbeat/run-preparation.js"; +import { recoveryService } from "./recovery/index.js"; +import { inspectWorkspaceSource } from "./workspace-source-inspection.js"; +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { isPreDispatchExecutionWaitVerified } from "./pre-dispatch-execution-wait.js"; +import { heartbeatRunEvents } from "@paperclipai/db"; +import { createHeartbeatQueue, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake } from "./heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "../__tests__/helpers/heartbeat-queue.js"; +import { recordRequiredWorkspaceRestoreFailure } from "./legacy-execution-recovery.js"; +import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; +import { withQueuedCommentIdsInWakePayload } from "./issue-queued-comment-queue.js"; +import * as recoverySources from "./workspace-recovery-source.js"; +import { requestedRunCancellation } from "./run-cancellation.js"; +import { createHeartbeatRunControl } from "./heartbeat/run-control.js"; +import { callbacks } from "./heartbeat/__tests__/run-control-fixtures.js"; +import { adapterExecutionControls, createAdapterExecutionControl } from "./adapter-execution-control.js"; import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; import { DELIVERY_QUEUES, subscribeDeliveryWork } from "./delivery-work-notifications.js"; import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js"; @@ -12,7 +31,7 @@ import { projectTaskInterruption } from "./task-interruption-presentation.js"; import * as nativeWorkspaceSync from "./native-runtime/native-workspace-sync.js"; import { GitShallowMetadataError } from "@paperclipai/adapter-utils/git-workspace-sync"; import { readSystemRecovery } from "./system-recovery-state.js"; -import { adoptSystemRecovery } from "./system-recovery.js"; +import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js"; import { getStorageService } from "../storage/index.js"; import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js"; @@ -33,10 +52,10 @@ import { resumePreservedResultFinalization } from "./preserved-result-finalizati import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import type { EnvironmentLease } from "@paperclipai/shared"; import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; -import { prepareNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js"; +import { prepareNativeWorkspaceSync, resumeNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js"; import { nativeWorkspaceFinalizationLockKey } from "./native-runtime/native-workspace-finalization-ownership.js"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; -import { prepareStoppedSandboxForRecovery } from "./sandbox-stop-and-retain.js"; +import { prepareStoppedSandboxForRecovery, prepareSandboxStopAndRetain, settleStopOnlyCleanup, readStopOnlyCleanup } from "./sandbox-stop-and-retain.js"; import { recoverRetainedWorkspace, requestWorkspaceRecovery, scheduleRecoveredWorkspaceCleanup, reconcileRetainedWorkspaces, retainedWorkspaceCandidates } from "./workspace-recovery.js"; import * as activity from "./activity-log.js"; import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; @@ -79,8 +98,8 @@ describe("automatic retained workspace recovery", () => { if (priorHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = priorHome; if (priorInstance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = priorInstance; }); - async function seed(mode = "native", nativeDescriptor = true, resourceDisposition: "destroy" | "keep_running" = "destroy") { - const companyId = randomUUID(), issueId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), actionId = randomUUID(); + async function seed(mode = "native", nativeDescriptor = true, resourceDisposition: "destroy" | "keep_running" = "destroy", reusable = false, ephemeral = false) { + const companyId = randomUUID(), issueId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), actionId: string = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Workspace recovery", defaultResponsibleUserId: "board", issuePrefix: `R${companyId.slice(0, 6)}` }); await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner", status: "idle", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); await db.insert(issues).values({ id: issueId, companyId, assigneeAgentId: agentId, status: "blocked", title: "Recover existing work" }); @@ -91,10 +110,13 @@ describe("automatic retained workspace recovery", () => { const [environment] = await db.insert(environments).values({ name: runId, driver: "sandbox" }).returning(); const providerLeaseId = randomUUID(), remote = path.join(root, runId, "remote"), host = path.join(root, runId, "host"); await mkdir(host, { recursive: true }); await writeFile(path.join(host, "existing.txt"), "base\n"); - const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, - environmentId: environment.id, provider: "daytona", providerLeaseId, leasePolicy: "retain_on_failure", status: "released", - cleanupStatus: "success", releasedAt: new Date(), metadata: { remoteCwd: remote, driver: "sandbox", pluginId: randomUUID(), sandboxProviderPlugin: true }, - }).returning(); + const leaseValues = { companyId, issueId, heartbeatRunId: runId, + environmentId: environment.id, provider: "daytona", providerLeaseId, + metadata: { remoteCwd: remote, driver: "sandbox", pluginId: randomUUID(), sandboxProviderPlugin: true, ...(reusable ? { agentId } : {}) } }; + const acquired = reusable ? await environmentService(db).acquireLease({ ...leaseValues, leasePolicy: "reuse_by_environment" }) : null; + const [lease] = acquired ? await db.select().from(environmentLeases).where(eq(environmentLeases.id, acquired.id)) + : await db.insert(environmentLeases).values({ ...leaseValues, leasePolicy: ephemeral ? "ephemeral" : "retain_on_failure", status: "released", + cleanupStatus: "success", releasedAt: new Date() }).returning(); const target: AdapterSandboxExecutionTarget = { kind: "remote", transport: "sandbox", providerKey: "test", leaseId: lease.id, remoteCwd: remote, timeoutMs: 30_000, sandboxLeaseAcquisition: { outcome: "created", providerLeaseId }, runner: { execute: input => new Promise((resolve, reject) => { @@ -144,6 +166,207 @@ describe("automatic retained workspace recovery", () => { const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + async function warmHandoffSource(hops = 1, beforeStop?: (lease: typeof environmentLeases.$inferSelect) => Promise) { + const prior = await seed("native", true, "keep_running", true), source = await run(prior.runId); + let previousTarget = targets.get(prior.lease.id)!, previousLeaseId = prior.lease.id; + let previousRunId: string = prior.runId; + let lease = prior.lease, current = source; + const leaseIds = [previousLeaseId]; + for (let hop = 0; hop < hops; hop++) { + expect(await resumeNativeWorkspaceSync({ db, runId: previousRunId, target: previousTarget })).toBe(true); + await db.update(heartbeatRuns).set({ status: "succeeded", errorCode: null, resultJson: null }).where(eq(heartbeatRuns.id, previousRunId)); + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, previousRunId)); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, prior.actionId)); + await environmentService(db).releaseLease(previousLeaseId, "released", { cleanupStatus: "success" }); + const [released] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, previousLeaseId)); + [current] = await db.insert(heartbeatRuns).values({ companyId: prior.companyId, agentId: source.agentId, + issueId: prior.issueId, nativeIssueId: prior.issueId, runtimeMode: "native", status: "cancelled", finishedAt: new Date(), + contextSnapshot: { issueId: prior.issueId } }).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: prior.companyId, issueId: prior.issueId, runId: current.id, phase: "terminal_failure" }); + const acquired = await environmentService(db).acquireLease({ companyId: prior.companyId, environmentId: prior.lease.environmentId!, + issueId: prior.issueId, heartbeatRunId: current.id, provider: "daytona", providerLeaseId: prior.providerLeaseId, + leasePolicy: "reuse_by_environment", replacesReusableLeaseId: previousLeaseId, + metadata: { ...released.metadata, sandboxLeaseAcquisition: { outcome: "resumed" } } }); + [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, acquired.id)); + const target = { ...previousTarget, leaseId: lease.id, sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId: prior.providerLeaseId } }; + targets.set(lease.id, target); + expect(await prepareNativeWorkspaceSync({ db, companyId: prior.companyId, workspaceId: prior.issueId, + workspaceLocalDir: prior.host, runId: current.id, lease: lease as EnvironmentLease, target, resourceDisposition: "keep_running" })).not.toBeNull(); + leaseIds.push(lease.id); previousRunId = current.id; previousLeaseId = lease.id; previousTarget = target; + } + [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)); + await beforeStop?.(lease); + await writeFile(path.join(prior.remote, "after-warm-stop.txt"), "stopped turn work\n"); + await recordRequiredWorkspaceRestoreFailure(db, await run(current.id), { workspaceRestoreFailure: "restore_failed" }, lease); + const [action] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, prior.companyId), + eq(issueRecoveryActions.fingerprint, `native-workspace:${current.id}`))); + const [retained] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)); + const stopping = readStopOnlyCleanup(retained) ? retained : await prepareSandboxStopAndRetain(db, retained); + expect(stopping).not.toBeNull(); + expect(await settleStopOnlyCleanup(db, stopping!, { attemptId: String(stopping!.metadata!.pendingCleanupAttemptId), + receipt: { providerLeaseId: prior.providerLeaseId, state: "stopped" } })).not.toBeNull(); + return { ...prior, priorRunId: prior.runId, priorLeaseId: prior.lease.id, leaseIds, runId: current.id, lease, actionId: action.id, + input: { ...prior.input, runId: current.id, actionId: action.id } }; + } + + describe("verified warm allocation handoffs", () => { + let original: { db: typeof db; root: string; connectionString: string; home: string | undefined }; + let isolated: Awaited> | undefined; + let isolatedDb: typeof db | undefined; + let isolatedRoot: string | undefined; + beforeAll(async () => { + original = { db, root, connectionString, home: process.env.PAPERCLIP_HOME }; + isolated = await startEmbeddedPostgresTestDatabase("workspace-warm-handoff-"); + isolatedDb = createDb(isolated.connectionString); + isolatedRoot = await mkdtemp(path.join(os.tmpdir(), "workspace-warm-handoff-files-")); + db = isolatedDb; root = isolatedRoot; connectionString = isolated.connectionString; + process.env.PAPERCLIP_HOME = path.join(root, "home"); + }, 30_000); + afterAll(async () => { + await isolatedDb?.$client.end(); await isolated?.cleanup(); + if (isolatedRoot) await rm(isolatedRoot, { recursive: true, force: true }); + if (original) { + db = original.db; root = original.root; connectionString = original.connectionString; + if (original.home === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = original.home; + } + }); + it.each([1, 2])("recovers a stopped native source after %s verified warm predecessor handoffs", async hops => { + const f = await warmHandoffSource(hops); + expect((await run(f.priorRunId)).runnerProfileJson?.nativeWorkspaceSync).toMatchObject({ state: "finalized", leaseId: f.priorLeaseId }); + const [previous] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.priorLeaseId)); + expect(previous).toMatchObject({ status: "expired", cleanupStatus: "success", providerLeaseId: f.providerLeaseId }); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + expect(await readFile(path.join(f.host, "after-warm-stop.txt"), "utf8")).toBe("stopped turn work\n"); + const source = await run(f.runId), freshProviderLeaseId = randomUUID(); + const [fresh] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, + issueId: f.issueId, nativeIssueId: f.issueId, runtimeMode: "native", status: "succeeded", finishedAt: new Date() }).returning(); + const remoteCwd = path.join(root, fresh.id, "remote"); + const freshLease = await environmentService(db).acquireLease({ companyId: f.companyId, environmentId: f.lease.environmentId!, + issueId: f.issueId, heartbeatRunId: fresh.id, provider: "daytona", providerLeaseId: freshProviderLeaseId, + metadata: { ...f.lease.metadata, remoteCwd } }); + const freshTarget = { ...targets.get(f.lease.id)!, leaseId: freshLease.id, remoteCwd, + sandboxLeaseAcquisition: { outcome: "created" as const, providerLeaseId: freshProviderLeaseId } }; + expect(await prepareNativeWorkspaceSync({ db, companyId: f.companyId, workspaceId: f.issueId, + workspaceLocalDir: f.host, runId: fresh.id, lease: freshLease, target: freshTarget, resourceDisposition: "keep_running" })).not.toBeNull(); + expect(await resumeNativeWorkspaceSync({ db, runId: fresh.id, target: freshTarget })).toBe(true); + expect(await inspectWorkspaceSource(db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId, actionId: f.actionId })) + .toMatchObject({ status: "verified" }); + }, 30_000); + + it.each(["active", "active_run", "active_control", "foreign_company", "environment", "workspace", "task", "agent", "unfinalized", "owner", "missing_proof", "conflicting_proof", "dangling_proof", "extra_history"])( + "keeps ambiguous warm source history fenced: %s", async fault => { + const f = await warmHandoffSource(); + const [previous] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.priorLeaseId)); + if (fault === "active") await db.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, previous.id)); + if (fault === "active_run") await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.priorRunId)); + if (fault === "active_control") adapterExecutionControls.set(f.priorRunId, createAdapterExecutionControl()); + if (fault === "foreign_company") { + const [foreign] = await db.insert(companies).values({ name: "Foreign owner", issuePrefix: `F${randomUUID().slice(0, 6)}` }).returning(); + await db.update(environmentLeases).set({ companyId: foreign.id }).where(eq(environmentLeases.id, previous.id)); + } + if (fault === "environment") { + const [other] = await db.insert(environments).values({ name: randomUUID(), driver: "sandbox" }).returning(); + await db.update(environmentLeases).set({ environmentId: other.id }).where(eq(environmentLeases.id, previous.id)); + } + if (fault === "workspace") { + const [other] = await db.insert(executionWorkspaces).values({ companyId: f.companyId, mode: "isolated_workspace", strategyType: "test", name: "Other workspace" }).returning(); + await db.update(environmentLeases).set({ executionWorkspaceId: other.id }).where(eq(environmentLeases.id, previous.id)); + } + if (fault === "task") { + const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task" }).returning(); + await db.update(environmentLeases).set({ issueId: other.id }).where(eq(environmentLeases.id, previous.id)); + } + if (fault === "agent") { + const [other] = await db.insert(agents).values({ companyId: f.companyId, name: "Other agent", adapterType: "paperclip_runner" }).returning(); + await db.update(heartbeatRuns).set({ agentId: other.id }).where(eq(heartbeatRuns.id, f.priorRunId)); + } + if (fault === "unfinalized") { + const prior = await run(f.priorRunId), reference = prior.runnerProfileJson!.nativeWorkspaceSync as Record; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...prior.runnerProfileJson, nativeWorkspaceSync: { ...reference, state: "prepared", finalHostSha256: null } } }).where(eq(heartbeatRuns.id, prior.id)); + } + if (fault === "owner") await db.update(nativeRunFinalizations).set({ leaseOwner: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.priorRunId)); + if (fault === "missing_proof") { + const metadata = { ...previous.metadata }; delete metadata.reusableLeaseSuccessor; + await db.update(environmentLeases).set({ metadata }).where(eq(environmentLeases.id, previous.id)); + } + if (fault === "conflicting_proof") await db.update(environmentLeases).set({ metadata: { ...previous.metadata, + reusableLeaseSuccessor: { ...(previous.metadata!.reusableLeaseSuccessor as Record), successorRunId: randomUUID() } } }).where(eq(environmentLeases.id, previous.id)); + if (fault === "dangling_proof") await db.update(environmentLeases).set({ metadata: { ...previous.metadata, reusableLeasePredecessor: {} } }).where(eq(environmentLeases.id, previous.id)); + if (fault === "extra_history") await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId: previous.environmentId, + heartbeatRunId: f.priorRunId, issueId: f.issueId, provider: "daytona", providerLeaseId: f.providerLeaseId, + status: "expired", leasePolicy: "reuse_by_environment", cleanupStatus: "success", metadata: previous.metadata }); + try { + expect(await inspectWorkspaceSource(db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId, actionId: f.actionId })) + .toMatchObject({ status: "blocked", failureCode: "allocation_ownership_ambiguous" }); + await expect(readFile(path.join(f.host, "after-warm-stop.txt"))).rejects.toMatchObject({ code: "ENOENT" }); + } finally { adapterExecutionControls.get(f.priorRunId)?.finish(); adapterExecutionControls.delete(f.priorRunId); } + }, 30_000); + + it("keeps server handoff receipts across metadata updates and same-run reacquisition", async () => { + const f = await warmHandoffSource(1, async lease => { + const proof = lease.metadata!.reusableLeasePredecessor; + expect(proof).toMatchObject({ schema: "paperclip.reusable-lease-handoff.v1", successorLeaseId: lease.id }); + const forged = { schema: "forged-provider-metadata" }; + const updated = await environmentService(db).updateLeaseMetadata(lease.id, { ...lease.metadata, + reusableLeasePredecessor: forged, reusableLeaseSuccessor: forged, marker: "updated" }); + expect(updated!.metadata?.reusableLeasePredecessor).toEqual(proof); + expect(updated!.metadata?.reusableLeaseSuccessor).toBeUndefined(); + await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success" }); + const reacquired = await environmentService(db).acquireLease({ companyId: lease.companyId, environmentId: lease.environmentId!, + issueId: lease.issueId, heartbeatRunId: lease.heartbeatRunId, provider: lease.provider, providerLeaseId: lease.providerLeaseId, + leasePolicy: "reuse_by_environment", reusesReusableLeaseId: lease.id, + metadata: { ...updated!.metadata, reusableLeasePredecessor: forged, reusableLeaseSuccessor: forged } }); + expect(reacquired.id).toBe(lease.id); + expect(reacquired.metadata?.reusableLeasePredecessor).toEqual(proof); + expect(reacquired.metadata?.reusableLeaseSuccessor).toBeUndefined(); + }); + await recoverRetainedWorkspace(f.input); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + }, 30_000); + + it("does not manufacture handoff authority from injected fresh or orphan lease metadata", async () => { + const f = await warmHandoffSource(), source = await run(f.runId); + const forged = { schema: "paperclip.reusable-lease-handoff.v1", successorLeaseId: f.lease.id }; + const values = { companyId: f.companyId, environmentId: f.lease.environmentId!, issueId: f.issueId, + heartbeatRunId: source.id, provider: "daytona", providerLeaseId: randomUUID(), + metadata: { agentId: source.agentId, reusableLeasePredecessor: forged, reusableLeaseSuccessor: forged } }; + const fresh = await environmentService(db).acquireLease(values); + const orphan = await environmentService(db).insertPendingCleanupLease({ ...values, providerLeaseId: randomUUID(), failureReason: "test" }); + for (const lease of [fresh, orphan]) { + expect(lease.metadata?.reusableLeasePredecessor).toBeUndefined(); + expect(lease.metadata?.reusableLeaseSuccessor).toBeUndefined(); + } + }, 30_000); + + it("rolls back both handoff links and keeps provider mismatches or retained sources fenced", async () => { + const f = await warmHandoffSource(), source = await run(f.runId), service = environmentService(db); + const values = { companyId: f.companyId, environmentId: f.lease.environmentId!, issueId: f.issueId, + heartbeatRunId: f.priorRunId, provider: "daytona", providerLeaseId: randomUUID(), + leasePolicy: "reuse_by_environment" as const, metadata: { agentId: source.agentId } }; + const prior = await service.acquireLease(values); + await service.releaseLease(prior.id, "released", { cleanupStatus: "success" }); + const [before] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, prior.id)); + const rollback = new Error("rollback_handoff"); + await expect(db.transaction(async tx => { + const acquired = await environmentService(tx as unknown as typeof db).acquireLease({ ...values, heartbeatRunId: f.runId, replacesReusableLeaseId: prior.id }); + expect(acquired.metadata?.reusableLeasePredecessor).toMatchObject({ previousLeaseId: prior.id, successorLeaseId: acquired.id }); + throw rollback; + })).rejects.toBe(rollback); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, prior.id)))[0]).toEqual(before); + expect(await db.select().from(environmentLeases).where(eq(environmentLeases.providerLeaseId, values.providerLeaseId))).toHaveLength(1); + await expect(service.acquireLease({ ...values, heartbeatRunId: f.runId, provider: "foreign-provider", replacesReusableLeaseId: prior.id })).rejects.toThrow("ownership changed"); + await expect(service.acquireLease({ ...values, providerLeaseId: f.providerLeaseId, replacesReusableLeaseId: f.lease.id })).rejects.toThrow("ownership changed"); + const [later] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, issueId: f.issueId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "succeeded", finishedAt: new Date() }).returning(); + await service.acquireLease({ ...values, heartbeatRunId: later.id, providerLeaseId: f.providerLeaseId }); + expect(await inspectWorkspaceSource(db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId, actionId: f.actionId })) + .toMatchObject({ status: "blocked", failureCode: "allocation_ownership_ambiguous" }); + }, 30_000); + + }); + it("unblocks a pending workspace selection only after a superseded retry's real files are preserved", async () => { const f = await seed(); await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); @@ -1211,4 +1434,546 @@ describe("automatic retained workspace recovery", () => { await restarted.$client.end(); } }, 30_000); + async function queuedResetFixture(nativeDescriptor = true, ephemeral = false) { + const f = await seed("native", nativeDescriptor, "keep_running", false, ephemeral), original = await run(f.runId); + await instanceSettingsService(db).updateExperimental({ enableAgentChat: true }); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + await db.update(issues).set({ conversationAgentId: original.agentId, conversationUserId: "operator", conversationState: "active", status: "in_progress" }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: "cancelled", nativePhase: "terminal_failure", errorCode: "cancelled", resultJson: { + cancelledByActorType: "user", cancelledByUserId: "operator", + cancellation: requestedRunCancellation({ cancelledByActorType: "user", cancelledByUserId: "operator" }, "Stop"), + executionCancellation: { state: "requested", requestedAt: new Date().toISOString() }, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId: f.runId, companyId: f.companyId, issueId: f.issueId, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID() }, + } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(environmentLeases).set({ status: "active", releasedAt: null, cleanupStatus: null, metadata: f.lease.metadata }).where(eq(environmentLeases.id, f.lease.id)); + const [reset] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "operator", body: "/new" }).returning(); + expect(await getExecutionBlocker(db, f.companyId, f.issueId, { conversationResetCommentId: reset.id })).toBeNull(); + const context = { source: "issue.comment", issueId: f.issueId, wakeReason: "issue_commented", wakeCommentId: reset.id, wakeCommentIds: [reset.id] }; + const payload = withQueuedCommentIdsInWakePayload({ issueId: f.issueId, commentId: reset.id, mutation: "comment", _paperclipWakeContext: context }, [reset.id]); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "on_demand", triggerDetail: "manual", reason: "issue_commented", status: "claimed", claimedAt: new Date(), + requestedByActorType: "user", requestedByActorId: "operator", payload }).returning(); + const [queued] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, + invocationSource: "on_demand", triggerDetail: "manual", status: "queued", runtimeMode: "legacy", responsibleUserId: "operator", + wakeupRequestId: wake.id, sessionIdBefore: "saved-native-session-before-reset", contextSnapshot: context }).returning(); + await db.update(agentWakeupRequests).set({ runId: queued.id }).where(eq(agentWakeupRequests.id, wake.id)); + await db.update(issues).set({ executionRunId: queued.id, executionAgentNameKey: "Worker", executionLockedAt: new Date() }).where(eq(issues.id, f.issueId)); + const deps = callbacks(db), control = createHeartbeatRunControl(db, deps), execution = createAdapterExecutionControl(); + const queueDeps = createHeartbeatQueueTestDependencies(db); + queueDeps.activeRunExecutions = deps.activeRunExecutions; + queueDeps.getAgent.mockResolvedValue((await db.select().from(agents).where(eq(agents.id, original.agentId)))[0]); + queueDeps.instanceSettings.getExperimental.mockResolvedValue({ enableAgentChat: true, enableNativeRunner: true } as Awaited>); + const queue = createHeartbeatQueue(db, queueDeps); + vi.mocked(deps.enqueueWakeup).mockImplementation(queue.enqueueWakeup); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, status: "running", invocationSource: "assignment" }); + deps.activeRunExecutions.add(f.runId); adapterExecutionControls.set(f.runId, execution); + return { f, original, reset, wake, queued, deps, control, execution, queueDeps, queue }; + } + + async function coalesceQueuedReset(t: Awaited>) { + const { f, original, reset, wake, queued } = t; + const [followup] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "operator", body: "/new" }).returning(); + const incoming = { source: "issue.comment", issueId: f.issueId, wakeReason: "issue_commented", + wakeCommentId: followup.id, wakeCommentIds: [followup.id] }; + await db.transaction(async tx => { + const scope = createAdmissionTransactionScope(f.companyId, tx as unknown as typeof db); + const actor = { companyId: f.companyId, wakeupRequestId: wake.id, requestedByActorType: "user", requestedByActorId: "operator" }; + expect(await createWakeAdmissionReader().matchesActiveWakeActor(scope, actor)).toBe(true); + expect(await createWakeAdmissionReader().matchesActiveWakeActor(scope, { ...actor, requestedByActorId: "other-user" })).toBe(false); + await createWakeAdmissionWriter().coalesceIntoActiveExecutionRun(scope, { + companyId: f.companyId, agentId: original.agentId, activeExecutionRunId: queued.id, + mergedContextSnapshot: mergeCoalescedContextSnapshot((await run(queued.id)).contextSnapshot, incoming, + { preserveExistingInteractionContinuation: true }), + source: "on_demand", triggerDetail: "manual", + payload: withQueuedCommentIdsInWakePayload({ issueId: f.issueId, mutation: "comment", _paperclipWakeContext: incoming }, [followup.id]), + requestedByActorType: "user", requestedByActorId: "operator", idempotencyKey: null, + }); + }); + const [originalWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + const [receipt] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.runId, queued.id), eq(agentWakeupRequests.status, "coalesced"))); + expect(originalWake.payload?.commentId).toBe(reset.id); + expect((await run(queued.id)).contextSnapshot).toMatchObject({ wakeCommentId: followup.id, + wakeCommentIds: [reset.id, followup.id] }); + return { followup, receipt }; + } + + async function ordinaryMixedQueuedFixture(kind: "active" | "deferred" | "deferred-different" | "system-deferred") { + const t = await queuedResetFixture(true, kind !== "active"), { f, original, reset, wake, queued, queueDeps, queue, control } = t; + await db.update(issues).set({ conversationAgentId: null, conversationUserId: null, conversationState: null, + executionRunId: null, executionAgentNameKey: null, executionLockedAt: null }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ wakeupRequestId: null }).where(eq(heartbeatRuns.id, queued.id)); + await db.update(agentWakeupRequests).set({ runId: null }).where(eq(agentWakeupRequests.id, wake.id)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, queued.id)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + const reader = createWakeAdmissionReader(), sameActor = vi.spyOn(reader, "matchesActiveWakeActor"); + queueDeps.wakeQueue.createAdmissionTransactionScope.mockImplementation(createAdmissionTransactionScope); + queueDeps.wakeQueue.admitWakeBehindIssueExecution.mockImplementation(createAdmitWakeBehindIssueExecution({ + reader, writer: createWakeAdmissionWriter(), helpers: { filterZombieCoalesceTarget, mergeCoalescedContextSnapshot, + shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake }, + })); + queueDeps.liveRunExecutions.has.mockImplementation(id => t.deps.activeRunExecutions.has(id)); + const stopped = await run(f.runId); + if (kind !== "active") { + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null, errorCode: null, + nativePhase: "running", resultJson: null }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "running" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(issues).set({ executionRunId: f.runId, executionAgentNameKey: "worker", executionLockedAt: new Date() }).where(eq(issues.id, f.issueId)); + } + const enqueue = async (comment: typeof reset) => queue.enqueueWakeup(original.agentId, { + source: "automation", triggerDetail: "system", reason: "issue_commented", requestedByActorType: "user", requestedByActorId: comment.authorUserId!, + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment", resumeIntent: true, followUpRequested: true }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, + wakeCommentIds: [comment.id], source: "issue.comment", wakeReason: "issue_commented", resumeIntent: true, followUpRequested: true }, + }); + if (kind === "system-deferred") expect(await queue.enqueueWakeup(original.agentId, { + source: "automation", triggerDetail: "system", reason: "issue_updated", requestedByActorType: "system", requestedByActorId: "second-user", + payload: { issueId: f.issueId, mutation: "update" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, source: "issue.updated", wakeReason: "issue_updated", forceFreshSession: true }, + })).toBeNull(); + const first = await enqueue(reset); + const [followup] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "second-user", body: "Please also preserve the second user's input." }).returning(); + const second = await enqueue(followup); + const ids = [reset.id, followup.id]; + let admitted = first; + if (kind === "active") { expect(first?.status).toBe("queued"); expect(second?.id).toBe(first!.id); } + else { + expect(first).toBeNull(); expect(second).toBeNull(); + const [pending] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.agentId, original.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"))); + expect(pending.payload?._paperclipWakeContext).toMatchObject({ wakeCommentIds: ids }); + expect(pending.requestedByActorType).toBe(kind === "system-deferred" ? "system" : "user"); + expect(pending.requestedByActorId).toBe(kind === "system-deferred" ? "second-user" : "operator"); + await db.update(heartbeatRuns).set({ status: "cancelled", nativePhase: "terminal_failure", finishedAt: new Date(), + errorCode: stopped.errorCode, resultJson: stopped.resultJson }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + const [currentLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + const stopping = await prepareSandboxStopAndRetain(db, currentLease); + expect(stopping).not.toBeNull(); + expect(await settleStopOnlyCleanup(db, stopping!, { attemptId: String(stopping!.metadata!.pendingCleanupAttemptId), + receipt: { providerLeaseId: f.providerLeaseId, state: "stopped" } })).not.toBeNull(); + await control.acknowledgeRemoteStop(f.runId, f.companyId); + const [last] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: kind === "deferred-different" ? "third-user" : "operator", body: "Resume after Stop and preserve all three inputs in their original order." }).returning(); + expect(last.createdAt.getTime()).toBeGreaterThan((await run(f.runId)).finishedAt!.getTime()); + ids.push(last.id); admitted = await enqueue(last); + } + expect(admitted?.status).toBe("queued"); + const current = await run(admitted!.id); + const [claimedWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, current.wakeupRequestId!)); + expect(current.contextSnapshot?.wakeCommentIds).toEqual(ids); + expect(sameActor).not.toHaveBeenCalled(); + expect(queueDeps.wakeQueue.admitWakeBehindIssueExecution.mock.calls.every(([, input]) => input.durableReceipt === undefined)).toBe(true); + expect(queueDeps.executeRun).not.toHaveBeenCalled(); + return { ...t, queued: current, wake: claimedWake, followup, expectedIds: ids, latestAuthor: kind === "deferred-different" ? "third-user" : kind === "active" ? "second-user" : "operator" }; + } + + async function exerciseQueuedReset(nativeDescriptor: boolean, coalesced = false, ordinary?: "active" | "deferred" | "deferred-different" | "system-deferred") { + const mixed = ordinary ? await ordinaryMixedQueuedFixture(ordinary) : null; + const t = mixed ?? await queuedResetFixture(nativeDescriptor); + const { f, original, reset, wake, queued, deps, control, execution, queueDeps, queue } = t; + const followup = coalesced ? (await coalesceQueuedReset(t)).followup : null; + const expectedIds = mixed?.expectedIds ?? (followup ? [reset.id, followup.id] : [reset.id]); + const responsible = queued.responsibleUserId; + try { + expect(await queue.claimQueuedRun(queued)).toBeNull(); + expect((await run(queued.id)).status).toBe("queued"); + await recordRequiredWorkspaceRestoreFailure(db, await run(f.runId), { workspaceRestoreFailure: "restore_failed" }, f.lease); + const [preservation] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId), + eq(issueRecoveryActions.sourceIssueId, f.issueId), eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))); + f.actionId = preservation.id; f.input.actionId = preservation.id; + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.failure?.code).toBe("agent_stop_unconfirmed"); + await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", metadata: { ...f.lease.metadata, + remoteExecutionTermination: remoteTerminationReceipt(f.lease, { providerLeaseId: f.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, f.lease.id)); + await control.acknowledgeRemoteStop(f.runId, f.companyId); + expect((await run(f.runId)).resultJson?.executionCancellation).toMatchObject({ state: "acknowledged", proof: "provider_termination_receipt" }); + deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); + await control.resumeRemoteStopComments(await run(f.runId)); + expect(readSystemRecovery((await action(f.actionId)).evidence)!.nextAttemptAt!).toBeLessThanOrEqual(Date.now()); + const onRecovered = vi.fn(async (current: Awaited>) => { + if (nativeDescriptor) expect(await inspectWorkspaceSource(db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, actionId: f.actionId })).toMatchObject({ status: "verified", scope: "recorded_source_identity" }); + await Promise.all([control.resumeRemoteStopComments(current), control.resumeRemoteStopComments(current)]); + }); + const repair = async () => { await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered) ?? []); }; + if (!nativeDescriptor) await repair(); + expect(await queue.startNextQueuedRunForAgent(original.agentId)).toEqual([]); + expect(await queue.claimQueuedRun(await run(queued.id))).toBeNull(); + const [heldWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(await isPreDispatchExecutionWaitVerified(db, await run(queued.id))).toBe(true); + const admissionEvents = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, queued.id)); + expect(admissionEvents.map(event => event.eventType)).toEqual(["lifecycle"]); + expect((await run(queued.id)).nextEventSeq).toBe(2); + expect((await run(queued.id)).sessionIdBefore).toBe(queued.sessionIdBefore); + expect((await run(queued.id)).responsibleUserId).toBe(responsible); + const scheduleRecoveryRetry = vi.fn(async (_runId: string) => null); + const recoveryDependencies = { enqueueWakeup: vi.fn(async () => null), scheduleRecoveryRetry, liveRunExecutions: { has: () => false } }; + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const recover = (database: typeof db) => recoveryService(database, recoveryDependencies).reconcileStrandedAssignedIssues({ issueCreatedAtGte: task.createdAt }); + if (nativeDescriptor) { + const rollback = new Error("restore_counterfactual_skipped_wake"); + await expect(db.transaction(async tx => { + await tx.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, wake.id)); + await recover(tx as unknown as typeof db); + expect(scheduleRecoveryRetry).toHaveBeenCalledWith(queued.id); + throw rollback; + })).rejects.toBe(rollback); + scheduleRecoveryRetry.mockClear(); + await recover(db); + expect(scheduleRecoveryRetry).not.toHaveBeenCalledWith(queued.id); + } + expect(queueDeps.executeRun).not.toHaveBeenCalled(); + if (nativeDescriptor) await repair(); + expect(onRecovered).toHaveBeenCalled(); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + if (nativeDescriptor) expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + await Promise.all([control.resumeRemoteStopComments(await run(f.runId)), + control.resumeRemoteStopComments(await run(f.runId))]); + const [afterWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(heldWake.status).toBe(nativeDescriptor ? "deferred_issue_execution" : "skipped"); + if (!nativeDescriptor) { + expect(afterWake.status).toBe("skipped"); + expect(deps.enqueueWakeup).not.toHaveBeenCalled(); + expect(queueDeps.executeRun).not.toHaveBeenCalled(); + return; + } + expect(afterWake).toMatchObject({ status: "coalesced", runId: expect.any(String) }); + expect(afterWake.runId).not.toBe(queued.id); + expect((await action(f.actionId)).evidence).toMatchObject({ automaticRecovery: { replay: "explicit_user_continuation" } }); + expect(await run(queued.id)).toMatchObject({ status: "cancelled", errorCode: "execution_reconciliation_required", startedAt: null }); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${f.issueId}`, eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot?.wakeCommentIds).toEqual(expectedIds); + if (mixed) expect(deps.enqueueWakeup).toHaveBeenCalledWith(original.agentId, expect.objectContaining({ requestedByActorType: "user", requestedByActorId: mixed.latestAuthor }), wake.id); + await recover(db); + expect(scheduleRecoveryRetry).not.toHaveBeenCalledWith(queued.id); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, queued.id))).toHaveLength(0); + expect((await run(queued.id)).nextEventSeq).toBe(2); + expect(queueDeps.executeRun).not.toHaveBeenCalled(); + } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); } + } + + it.each([ + { nativeDescriptor: true, title: "preserves an already queued reset until real native copyback replays its saved wake" }, + { nativeDescriptor: false, title: "does not replay a snapshot-only source without a verified native baseline" }, + ])("$title", async ({ nativeDescriptor }) => exerciseQueuedReset(nativeDescriptor), 30_000); + + it("preserves coalesced queued user comments until real native copyback replays the original wake", async () => { + await exerciseQueuedReset(true, true); + }, 30_000); + + it("preserves each ordinary mixed-author comment across queued Stop and native copyback", async () => { + await exerciseQueuedReset(true, false, "active"); + }, 30_000); + + it("preserves merged-deferred mixed-author comments across queued Stop and native copyback", async () => { + await exerciseQueuedReset(true, false, "deferred"); + }, 30_000); + + it("preserves mixed-author comments merged into a system-owned deferred carrier across Stop and native copyback", async () => { + await exerciseQueuedReset(true, false, "system-deferred"); + }, 30_000); + + it("preserves mixed-author deferred inputs when the resuming third user differs from the carrier", async () => { + await exerciseQueuedReset(true, false, "deferred-different"); + }, 30_000); + + it.each(["missing_primary", "agent_namespace", "system_namespace", "foreign_link", "foreign_carrier_run", "foreign_receipt_user", "array_only"] as const)("rejects mixed-author queue authority forgery: %s", async fault => { + const t = await ordinaryMixedQueuedFixture("system-deferred"), { f, wake, queued, deps, execution, queueDeps, queue, followup } = t; + try { + await recordRequiredWorkspaceRestoreFailure(db, await run(f.runId), { workspaceRestoreFailure: "restore_failed" }, f.lease); + deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); + const sourceIds = queued.contextSnapshot!.queuedCommentSourceWakeIds as string[]; + const [receipt] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.requestedByActorType, "user"), + sql`${agentWakeupRequests.payload}->>'commentId' = ${followup.id}`)); + expect(receipt).toBeDefined(); + if (fault === "missing_primary") await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "agent_namespace" || fault === "system_namespace") await db.update(agentWakeupRequests).set({ requestedByActorType: fault === "agent_namespace" ? "agent" : "system", requestedByActorId: "second-user" }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "foreign_link") await db.update(agentWakeupRequests).set({ payload: { ...receipt.payload, coalescedIntoWakeupRequestId: randomUUID() } }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "foreign_carrier_run") await db.update(agentWakeupRequests).set({ runId: f.runId }).where(eq(agentWakeupRequests.id, sourceIds[0])); + if (fault === "foreign_receipt_user") await db.update(agentWakeupRequests).set({ requestedByActorId: "operator" }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "array_only") { + const [injected] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "second-user", body: "An array is not a saved admission receipt." }).returning(); + await db.update(heartbeatRuns).set({ contextSnapshot: mergeCoalescedContextSnapshot(queued.contextSnapshot, + { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentId: injected.id, wakeCommentIds: [injected.id] }, { preserveExistingInteractionContinuation: true }) }).where(eq(heartbeatRuns.id, queued.id)); + await db.update(agentWakeupRequests).set({ payload: withQueuedCommentIdsInWakePayload(receipt.payload!, [followup.id, injected.id]) }).where(eq(agentWakeupRequests.id, receipt.id)); + } + expect(await queue.claimQueuedRun(await run(queued.id))).toBeNull(); + const [after] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(after.status).toBe("skipped"); + expect(queueDeps.executeRun).not.toHaveBeenCalled(); expect(deps.enqueueWakeup).not.toHaveBeenCalled(); + } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); } + }, 30_000); + + it("preserves ordinary no-issue deferred merge semantics", async () => { + const f = await seed(), original = await run(f.runId), writer = createWakeAdmissionWriter(); + const [carrier] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "automation", status: "deferred_issue_execution", payload: { text: "original" }, requestedByActorType: "system" }).returning(); + await db.transaction(tx => writer.mergeIntoExistingDeferredWake(createAdmissionTransactionScope(f.companyId, tx as unknown as typeof db), { + companyId: f.companyId, existingDeferredWakeId: carrier.id, mergedPayload: { text: "merged" }, nextCoalescedCount: 1, + coalescedReceipt: { agentId: original.agentId, source: "automation", triggerDetail: null, reason: null, payload: { text: "saved" }, requestedByActorType: "system", requestedByActorId: null, idempotencyKey: null, runId: null }, + })); + const [after] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, carrier.id)); + expect(after.payload).toEqual({ text: "merged" }); + const [receipt] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.status, "coalesced"))); + expect(receipt.payload).toEqual({ text: "saved", coalescedIntoWakeupRequestId: carrier.id }); + }); + + it("rolls back saved primary and carrier overwrite if incoming mixed-author receipt persistence fails", async () => { + const f = await seed(), original = await run(f.runId), writer = createWakeAdmissionWriter(); + const [first] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "operator", body: "First saved input" }).returning(); + const [second] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "second-user", body: "Second saved input" }).returning(); + const [carrier] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "automation", status: "deferred_issue_execution", payload: { issueId: f.issueId, commentId: first.id }, requestedByActorType: "user", requestedByActorId: "operator" }).returning(); + const merge = (tx: Parameters[0]>[0], id?: string) => writer.mergeIntoExistingDeferredWake(createAdmissionTransactionScope(f.companyId, tx as unknown as typeof db), { + companyId: f.companyId, existingDeferredWakeId: carrier.id, mergedPayload: { issueId: f.issueId, commentId: second.id }, nextCoalescedCount: 1, + coalescedReceipt: { ...(id ? { id } : {}), agentId: original.agentId, source: "automation", triggerDetail: null, reason: "issue_commented", payload: { issueId: f.issueId, commentId: second.id, coalescedIntoWakeupRequestId: "untrusted" }, requestedByActorType: "user", requestedByActorId: "second-user", idempotencyKey: null, runId: null }, + }); + await expect(db.transaction(tx => merge(tx, carrier.id))).rejects.toThrow(); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)))).toEqual([carrier]); + await db.transaction(tx => merge(tx)); await db.transaction(tx => merge(tx)); + const proofs = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.status, "coalesced"))); + expect(proofs.filter(receipt => receipt.payload?.commentId === first.id)).toHaveLength(1); + expect(proofs.every(receipt => receipt.payload?.coalescedIntoWakeupRequestId === carrier.id)).toBe(true); + expect(proofs.find(receipt => receipt.payload?.commentId === first.id)?.requestedByActorId).toBe("operator"); + }); + + it.each(["missing_receipt", "wrong_receipt_run", "wrong_receipt_agent", "wrong_receipt_company", "wrong_receipt_user", "wrong_receipt_issue", "terminal_receipt", "foreign_comment_user", "deleted_followup", "unrecorded_comment", "unrecorded_original"] as const)("keeps coalesced Stop recovery input closed without saved receipt authority: %s", async fault => { + const t = await queuedResetFixture(), { f, wake, queued, deps, execution, queueDeps, queue } = t; + try { + const { followup, receipt } = await coalesceQueuedReset(t); + await recordRequiredWorkspaceRestoreFailure(db, await run(f.runId), { workspaceRestoreFailure: "restore_failed" }, f.lease); + deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); + if (fault === "missing_receipt") await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "wrong_receipt_run") await db.update(agentWakeupRequests).set({ runId: f.runId }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "wrong_receipt_agent") await db.update(agentWakeupRequests).set({ agentId: (await db.insert(agents).values({ companyId: f.companyId, name: "Other", role: "engineer" }).returning())[0].id }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "wrong_receipt_company") await db.update(agentWakeupRequests).set({ companyId: (await db.insert(companies).values({ name: "Other", issuePrefix: `OT${randomUUID().slice(0, 6)}` }).returning())[0].id }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "wrong_receipt_user") await db.update(agentWakeupRequests).set({ requestedByActorId: "other-user" }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "wrong_receipt_issue") await db.update(agentWakeupRequests).set({ payload: { ...receipt.payload, issueId: randomUUID() } }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "terminal_receipt") await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, receipt.id)); + if (fault === "foreign_comment_user") await db.update(issueComments).set({ authorUserId: "other-user" }).where(eq(issueComments.id, followup.id)); + if (fault === "deleted_followup") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, followup.id)); + if (fault === "unrecorded_comment") { + const [injected] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "operator", body: "/new" }).returning(); + await db.update(heartbeatRuns).set({ contextSnapshot: mergeCoalescedContextSnapshot((await run(queued.id)).contextSnapshot, + { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentId: injected.id, wakeCommentIds: [injected.id] }, + { preserveExistingInteractionContinuation: true }) }).where(eq(heartbeatRuns.id, queued.id)); + } + if (fault === "unrecorded_original") await db.update(agentWakeupRequests).set({ payload: { ...wake.payload, commentId: randomUUID() } }).where(eq(agentWakeupRequests.id, wake.id)); + expect(await queue.claimQueuedRun(await run(queued.id))).toBeNull(); + const [afterWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(afterWake.status).toBe("skipped"); + expect(queueDeps.executeRun).not.toHaveBeenCalled(); + expect(deps.enqueueWakeup).not.toHaveBeenCalled(); + expect((await run(queued.id)).status).toBe("cancelled"); + } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); } + }); + + it.each(["system_actor", "wrong_actor", "deleted_comment", "foreign_comment", "foreign_agent", "foreign_company", "foreign_run", "terminal_wake", "started", "provider_event", "provider_lease", "resolved_runtime"] as const)("keeps queued Stop recovery input closed without paired predispatch authority: %s", async fault => { + const t = await queuedResetFixture(), { f, original, reset, wake, queued, deps, execution, queueDeps, queue } = t; + try { + await recordRequiredWorkspaceRestoreFailure(db, await run(f.runId), { workspaceRestoreFailure: "restore_failed" }, f.lease); + deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); + if (fault === "system_actor") await db.update(agentWakeupRequests).set({ requestedByActorType: "system" }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "wrong_actor") await db.update(agentWakeupRequests).set({ requestedByActorId: "other-user" }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "deleted_comment") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, reset.id)); + if (fault === "foreign_comment") await db.update(issueComments).set({ issueId: (await db.insert(issues).values({ companyId: f.companyId, title: "Other" }).returning())[0].id }).where(eq(issueComments.id, reset.id)); + if (fault === "foreign_agent") await db.update(agentWakeupRequests).set({ agentId: (await db.insert(agents).values({ companyId: f.companyId, name: "Other", role: "engineer" }).returning())[0].id }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "foreign_company") await db.update(agentWakeupRequests).set({ companyId: (await db.insert(companies).values({ name: "Other", issuePrefix: "OT" }).returning())[0].id }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "foreign_run") await db.update(agentWakeupRequests).set({ runId: f.runId }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "terminal_wake") await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date(), error: "Operator cancelled" }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "started") await db.update(heartbeatRuns).set({ startedAt: new Date() }).where(eq(heartbeatRuns.id, queued.id)); + if (fault === "provider_event") await appendHeartbeatRunEvent(db, { companyId: f.companyId, agentId: original.agentId, runId: queued.id, + eventType: "adapter.invoke", stream: "system", level: "info", message: "Controlled provider execution evidence" }); + if (fault === "provider_lease") await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: queued.id, environmentId: f.lease.environmentId, provider: "daytona", providerLeaseId: randomUUID() }); + if (fault === "resolved_runtime") await db.update(heartbeatRuns).set({ runtimeModeResolvedAt: new Date() }).where(eq(heartbeatRuns.id, queued.id)); + const [beforeWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(await queue.claimQueuedRun(await run(queued.id))).toBeNull(); + const [afterWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(afterWake.status).not.toBe("deferred_issue_execution"); + if (["foreign_agent", "foreign_company", "foreign_run", "terminal_wake"].includes(fault)) expect(afterWake).toEqual(beforeWake); + else expect(afterWake.status).toBe("skipped"); + expect(queueDeps.executeRun).not.toHaveBeenCalled(); + expect(deps.enqueueWakeup).not.toHaveBeenCalled(); + } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); } + }); + + async function stoppedEvidenceFixture() { + const f = await seed(); + const original = await run(f.runId); + await db.update(heartbeatRuns).set({ status: "cancelled", nativePhase: "terminal_failure", errorCode: "cancelled", resultJson: { + ...original.resultJson, cancelledByActorType: "user", cancelledByUserId: "operator", + cancellation: requestedRunCancellation({ cancelledByActorType: "user", cancelledByUserId: "operator" }, "Stop"), + executionCancellation: { state: "requested", requestedAt: new Date().toISOString() }, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId: f.runId, companyId: f.companyId, issueId: f.issueId, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID() }, + } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ conversationAgentId: original.agentId, conversationUserId: "operator", conversationState: "active", status: "in_progress" }).where(eq(issues.id, f.issueId)); + const [reset] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "operator", body: "/new" }).returning(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + await recordRequiredWorkspaceRestoreFailure(db, await run(f.runId), { workspaceRestoreFailure: "restore_failed" }, f.lease); + const [preservation] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId), + eq(issueRecoveryActions.sourceIssueId, f.issueId), eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))); + expect(preservation).toMatchObject({ status: "resolved", evidence: { automaticRecovery: { policy: "preserve_without_replay_v1", replay: "blocked", runId: f.runId } } }); + f.actionId = preservation.id; f.input.actionId = preservation.id; + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "on_demand", reason: "issue_execution_deferred", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: "operator", + payload: withQueuedCommentIdsInWakePayload({ issueId: f.issueId, commentId: reset.id, + _paperclipWakeContext: { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentId: reset.id, wakeCommentIds: [reset.id] } }, [reset.id]) }).returning(); + const deps = callbacks(db), control = createHeartbeatRunControl(db, deps); + const execution = createAdapterExecutionControl(); + deps.activeRunExecutions.add(f.runId); adapterExecutionControls.set(f.runId, execution); + const cleanup = () => { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); }; + return { f, original, reset, wake, deps, control, execution, cleanup }; + } + + async function heldStoppedEvidence() { + const prepared = await stoppedEvidenceFixture(); + await recoverRetainedWorkspace(prepared.f.input); + const held = readSystemRecovery((await action(prepared.f.actionId)).evidence)!; + expect(held.state).toBe("awaiting_evidence"); + await prepared.control.acknowledgeRemoteStop(prepared.f.runId, prepared.f.companyId); + prepared.cleanup(); + return { ...prepared, held }; + } + + it("advances saved workspace recovery on new Stop evidence after executor cleanup", async () => { + const { f, original, reset, deps, control, execution } = await stoppedEvidenceFixture(); + try { + await recoverRetainedWorkspace(f.input); + const held = readSystemRecovery((await action(f.actionId)).evidence)!; + expect(held).toMatchObject({ state: "awaiting_evidence", failure: { code: "agent_stop_unconfirmed", retryClass: "recheck_evidence" } }); + expect(held.nextAttemptAt!).toBeGreaterThan(Date.now() + 20_000); + expect(await getExecutionBlocker(db, f.companyId, f.issueId, { conversationResetCommentId: reset.id })).not.toBeNull(); + await control.acknowledgeRemoteStop(f.runId, f.companyId); + expect((await run(f.runId)).resultJson?.executionCancellation).toMatchObject({ state: "acknowledged", proof: "provider_termination_receipt" }); + await control.resumeRemoteStopComments(await run(f.runId)); + expect(readSystemRecovery((await action(f.actionId)).evidence)).toEqual(held); + deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); + const at = Date.now(); let events = 0; + const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.workspaceRecovery, () => { events += 1; }); + try { await control.resumeRemoteStopComments(await run(f.runId)); } finally { stop(); } + const advanced = readSystemRecovery((await action(f.actionId)).evidence)!; + expect(advanced.nextAttemptAt!).toBeLessThanOrEqual(Date.now()); + expect(advanced).toMatchObject({ generationId: held.generationId, attempts: held.attempts, retriesUsed: held.retriesUsed, + firstFailureAt: held.firstFailureAt, deadlineAt: held.deadlineAt, providerCooldownAt: held.providerCooldownAt }); + expect(events).toBeGreaterThan(0); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + const successorRunId = randomUUID(); + const admitted = await admitExplicitNativeContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: original.agentId, + actorType: "user", actorId: "operator", reason: "issue_commented", commentId: reset.id, successorRunId }); + expect(admitted).toMatchObject({ previousRunId: f.runId, commentId: reset.id }); + expect((await action(f.actionId)).evidence).toMatchObject({ automaticRecovery: { replay: "explicit_user_continuation" } }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId, { conversationResetCommentId: reset.id })).toBeNull(); + expect(Date.now() - at).toBeLessThan(30_000); + expect(deps.enqueueWakeup).not.toHaveBeenCalled(); + } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); } + }); + + it.each(["active_adapter", "active_execution", "missing_native_ack", "wrong_native_run", "missing_physical_receipt", "wrong_allocation", "coordinator_owner", "coordinator_observed", "foreign_action_run", "expired_deadline", "exhausted", "retry_budget", "unrelated_failure"] as const)("does not advance Stop-evidence work without current authority: %s", async fault => { + const t = await heldStoppedEvidence(), { f, deps, control, execution } = t; + try { + if (fault === "active_adapter") adapterExecutionControls.set(f.runId, execution); + if (fault === "active_execution") deps.activeRunExecutions.add(f.runId); + if (fault === "missing_native_ack") await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{nativeCancellation,dispatchState}', '"pending"'::jsonb)` }).where(eq(heartbeatRuns.id, f.runId)); + if (fault === "wrong_native_run") await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{nativeCancellation,runId}', ${JSON.stringify(randomUUID())}::jsonb)` }).where(eq(heartbeatRuns.id, f.runId)); + if (fault === "missing_physical_receipt") await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} - 'remoteExecutionTermination'` }).where(eq(environmentLeases.id, f.lease.id)); + if (fault === "wrong_allocation") await db.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, '{remoteExecutionTermination,providerLeaseId}', ${JSON.stringify(randomUUID())}::jsonb)` }).where(eq(environmentLeases.id, f.lease.id)); + if (fault === "coordinator_owner") await db.update(nativeRunFinalizations).set({ leaseOwner: randomUUID(), leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (fault === "coordinator_observed") await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (fault === "foreign_action_run") await db.update(issueRecoveryActions).set({ evidence: sql`jsonb_set(${issueRecoveryActions.evidence}, '{runId}', ${JSON.stringify(randomUUID())}::jsonb)` }).where(eq(issueRecoveryActions.id, f.actionId)); + if (fault === "expired_deadline") await updateSystemRecovery(db, f.input, t.held, { ...t.held, revision: t.held.revision + 1, deadlineAt: Date.now() - 1 }); + if (fault === "exhausted") await updateSystemRecovery(db, f.input, t.held, { ...t.held, revision: t.held.revision + 1, state: "exhausted", nextAttemptAt: null }); + if (fault === "retry_budget") await updateSystemRecovery(db, f.input, t.held, { ...t.held, revision: t.held.revision + 1, retriesUsed: 5 }); + if (fault === "unrelated_failure") await updateSystemRecovery(db, f.input, t.held, failSystemRecovery(t.held, new SystemRecoveryFailure("provider_unavailable", "retry_transient"))); + const before = (await action(f.actionId)).evidence; + await control.resumeRemoteStopComments(await run(f.runId)); + expect((await action(f.actionId)).evidence).toEqual(before); + if (fault !== "exhausted" && fault !== "retry_budget") expect(deps.enqueueWakeup).not.toHaveBeenCalled(); + } finally { t.cleanup(); } + }); + + it("consumes identical Stop-evidence only once across another actual failed attempt", async () => { + const t = await heldStoppedEvidence(), { f, control, execution } = t; + try { + await control.resumeRemoteStopComments(await run(f.runId)); + expect((await action(f.actionId)).evidence.workspaceStopEvidence).toMatchObject({ consumed: true }); + adapterExecutionControls.set(f.runId, execution); + await recoverRetainedWorkspace(f.input); + t.cleanup(); + const before = (await action(f.actionId)).evidence; + expect(readSystemRecovery(before)?.nextAttemptAt).toBeGreaterThan(Date.now() + 20_000); + await control.resumeRemoteStopComments(await run(f.runId)); + await control.resumeRemoteStopComments(await run(f.runId)); + expect((await action(f.actionId)).evidence).toEqual(before); + } finally { t.cleanup(); } + }); + + it("keeps the provider cooldown and retry window when advancing new Stop-evidence", async () => { + const t = await heldStoppedEvidence(), { f, control } = t; + try { + const cooldown = Date.now() + 5_000; + const state = failSystemRecovery(t.held, new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence"), { retryAfterAt: cooldown }); + await updateSystemRecovery(db, f.input, t.held, state); + await control.resumeRemoteStopComments(await run(f.runId)); + const after = readSystemRecovery((await action(f.actionId)).evidence)!; + expect(after.nextAttemptAt).toBe(cooldown); + expect(after).toMatchObject({ generationId: state.generationId, attempts: state.attempts, retriesUsed: state.retriesUsed, firstFailureAt: state.firstFailureAt, deadlineAt: state.deadlineAt, providerCooldownAt: cooldown }); + await control.resumeRemoteStopComments(await run(f.runId)); + expect(readSystemRecovery((await action(f.actionId)).evidence)).toEqual(after); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)).toEqual(after); + } finally { t.cleanup(); } + }); + + it.each(["current", "stale_generation", "lost_receipt"] as const)("retains a post-cleanup Stop-evidence event before worker failure commits: %s", async variant => { + const t = await stoppedEvidenceFixture(), { f, control } = t; + const originalReader = recoverySources.inspectWorkspaceRecoverySource; + let observed = false, release!: () => void; + const barrier = new Promise(resolve => { release = resolve; }); + const reader = vi.spyOn(recoverySources, "inspectWorkspaceRecoverySource").mockImplementationOnce(async (...args) => { + try { return await originalReader(...args); } + catch (error) { observed = true; await barrier; throw error; } + }); + const worker = recoverRetainedWorkspace(f.input); + try { + await expect.poll(() => observed, { timeout: 5_000 }).toBe(true); + const running = readSystemRecovery((await action(f.actionId)).evidence)!; + expect(running.state).toBe("running"); + await control.acknowledgeRemoteStop(f.runId, f.companyId); t.cleanup(); + await control.resumeRemoteStopComments(await run(f.runId)); + const pending = await action(f.actionId); + expect(readSystemRecovery(pending.evidence)).toEqual(running); + expect(pending.evidence.workspaceStopEvidence).toMatchObject({ sourceRunId: f.runId, generationId: running.generationId, consumed: false }); + if (variant === "stale_generation") await db.update(issueRecoveryActions).set({ evidence: sql`jsonb_set(${issueRecoveryActions.evidence}, '{workspaceStopEvidence,generationId}', ${JSON.stringify(randomUUID())}::jsonb)` }).where(eq(issueRecoveryActions.id, f.actionId)); + if (variant === "lost_receipt") await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} - 'remoteExecutionTermination'` }).where(eq(environmentLeases.id, f.lease.id)); + release(); await worker; + const after = await action(f.actionId), state = readSystemRecovery(after.evidence)!; + expect(state.state).toBe("awaiting_evidence"); + expect(state).toMatchObject({ attempts: running.attempts, retriesUsed: running.retriesUsed, generationId: running.generationId }); + if (variant === "current") { + expect(state.nextAttemptAt!).toBeLessThanOrEqual(Date.now()); + expect(after.evidence.workspaceStopEvidence).toMatchObject({ consumed: true }); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + } else { + expect(state.nextAttemptAt!).toBeGreaterThan(Date.now() + 20_000); + expect(after.evidence.workspaceStopEvidence).toMatchObject({ consumed: false }); + } + } finally { release(); await worker; reader.mockRestore(); t.cleanup(); } + }); + }); diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 1eb9b1bf959..1ffd159d139 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -1,4 +1,7 @@ import { classifyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-merge"; +import { createHash } from "node:crypto"; +import { adapterExecutionControls } from "./adapter-execution-control.js"; +import { isAcknowledgedNativeStop } from "./acknowledged-native-stop.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; @@ -7,7 +10,7 @@ import { beginIdleTrackedWork } from "./task-admission.js"; import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { resumePreservedResultFinalization } from "./preserved-result-finalization.js"; import { and, asc, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm"; -import { agentWakeupRequests, environmentLeases, heartbeatRuns, issues, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, environmentLeases, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, workspaceOperations, type Db } from "@paperclipai/db"; import type { Environment, EnvironmentLease } from "@paperclipai/shared"; import { conflict } from "../errors.js"; import { logger } from "../middleware/logger.js"; @@ -24,7 +27,7 @@ import { readNativeWorkspaceSyncReference, readNativeWorkspaceSyncLocalCwd, resu import { classifyNativeWorkspaceFailure } from "./native-runtime/native-workspace-failure.js"; import { adoptSystemRecovery, loadSystemRecovery, updateSystemRecovery } from "./system-recovery.js"; -import { createSystemRecovery, failSystemRecovery, readSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure, type SystemRecoveryState } from "./system-recovery-state.js"; +import { SYSTEM_RECOVERY_MAX_RETRIES, createSystemRecovery, failSystemRecovery, readSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure, type SystemRecoveryState } from "./system-recovery-state.js"; import { inspectWorkspaceRecoverySource, workspaceRecoverySourceEligible, workspaceSourceBindingFailure } from "./workspace-recovery-source.js"; import { preserveWorkspaceSnapshot } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries } from "./workspace-recovery-backfill.js"; @@ -33,6 +36,80 @@ const exportCauses = ["native_workspace_sync_out_unrecoverable", "native_workspa const terminal = ["failed", "interrupted", "timed_out", "cancelled", "succeeded"]; type Source = { companyId: string; issueId: string; runId: string; actionId: string }; +type StopEvidenceSource = Omit & { actionId?: string }; + +async function recheckStoppedWorkspaceRecovery(db: Db, source: StopEvidenceSource, observe: boolean) { + return db.transaction(async tx => { + const [issue] = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.id, source.issueId), eq(issues.companyId, source.companyId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.runId), + eq(heartbeatRuns.companyId, source.companyId))).for("update"); + if (!issue || !run || run.runtimeMode !== "native" || run.nativeIssueId !== issue.id || + !isAcknowledgedNativeStop(run) || adapterExecutionControls.has(run.id)) return 0; + const cancellation = run.resultJson?.executionCancellation as { state?: unknown; proof?: unknown; acknowledgedAt?: unknown } | undefined; + if (cancellation?.state !== "acknowledged" || cancellation.proof !== "provider_termination_receipt") return 0; + const [coordinator] = await tx.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner }) + .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, source.companyId), + eq(nativeRunFinalizations.runId, source.runId), eq(nativeRunFinalizations.issueId, source.issueId))); + if (!coordinator || coordinator.leaseOwner || !["terminal_failure", "applied"].includes(coordinator.phase)) return 0; + const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, source.companyId), + eq(environmentLeases.heartbeatRunId, source.runId))); + if (!leases.length || leases.some(lease => !lease.provider || lease.provider === "local" || !hasRemoteTerminationReceipt(lease))) return 0; + const intent = run.resultJson!.nativeCancellation as Record; + const fingerprint = createHash("sha256").update(JSON.stringify({ companyId: source.companyId, issueId: source.issueId, + runId: source.runId, intentAuditId: intent.intentAuditId, acknowledgementAuditId: intent.acknowledgementAuditId, + acknowledgedAt: cancellation.acknowledgedAt, leases: leases.sort((a, b) => a.id.localeCompare(b.id)).map(lease => { + const receipt = lease.metadata!.remoteExecutionTermination as Record; + return { leaseId: lease.id, provider: lease.provider, providerLeaseId: lease.providerLeaseId, + state: receipt.state, confirmedAt: receipt.confirmedAt }; + }) })).digest("hex"); + const actions = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, source.companyId), + eq(issueRecoveryActions.sourceIssueId, source.issueId), or(inArray(issueRecoveryActions.status, ["active", "escalated"]), and( + eq(issueRecoveryActions.status, "resolved"), sql`${issueRecoveryActions.evidence}#>>'{automaticRecovery,policy}' = 'preserve_without_replay_v1'`, + sql`${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'blocked'`, + sql`${issueRecoveryActions.evidence}#>>'{automaticRecovery,runId}' = ${source.runId}`)), + sql`${issueRecoveryActions.evidence}->>'runId' = ${source.runId}`, + source.actionId ? eq(issueRecoveryActions.id, source.actionId) : undefined)).for("update"); + let changed = 0; + for (const action of actions) { + const current = readSystemRecovery(action.evidence), workspace = readWorkspaceRecovery(action.evidence), now = Date.now(); + if (!current || current.sourceRunId !== source.runId || current.phase !== "verify_execution" || + !workspace || workspace.runId !== source.runId || ["complete", "exhausted"].includes(current.state) || current.retriesUsed >= SYSTEM_RECOVERY_MAX_RETRIES || + (current.deadlineAt !== null && now >= current.deadlineAt)) continue; + const stopWait = current.failure?.retryClass === "recheck_evidence" && + ["agent_stop_unconfirmed", "copyback_stop_unconfirmed"].includes(current.failure.code); + if (current.state !== "running" && current.failure && !stopWait) continue; + const prior = action.evidence.workspaceStopEvidence as Record | undefined; + const same = prior?.schema === "paperclip.workspace-stop-evidence.v1" && prior.sourceRunId === source.runId && + prior.generationId === current.generationId && prior.fingerprint === fingerprint; + if ((same && prior!.consumed === true) || (!observe && (!same || prior!.consumed !== false))) continue; + const due = Math.max(now, current.providerCooldownAt ?? 0); + const advance = current.state === "awaiting_evidence" && stopWait && current.attempts > 0 && + current.deadlineAt !== null && due < current.deadlineAt && current.nextAttemptAt !== null && current.nextAttemptAt > due; + const consumed = current.state !== "running"; + if (!advance && same && prior!.consumed === consumed) continue; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); + if (advance && !await updateSystemRecovery(tx as unknown as Db, { ...source, actionId: action.id }, current, + { ...current, revision: current.revision + 1, nextAttemptAt: due })) continue; + const marker = { schema: "paperclip.workspace-stop-evidence.v1", sourceRunId: source.runId, + generationId: current.generationId, fingerprint, consumed }; + const recorded = await tx.update(issueRecoveryActions).set({ evidence: sql`${issueRecoveryActions.evidence} || + ${JSON.stringify({ workspaceStopEvidence: marker })}::jsonb`, updatedAt: new Date() }).where(and( + eq(issueRecoveryActions.companyId, source.companyId), eq(issueRecoveryActions.id, action.id), + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,generationId}' = ${current.generationId}`, + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,revision}' = ${String(current.revision + (advance ? 1 : 0))}`)) + .returning({ id: issueRecoveryActions.id }); + if (!recorded.length) throw new Error("workspace_stop_evidence_changed"); + changed += 1; + } + return changed; + }); +} + +export function observeStoppedWorkspaceRecovery(db: Db, source: StopEvidenceSource) { + return recheckStoppedWorkspaceRecovery(db, source, true); +} + /** Admission records intent only. No request handler runs physical recovery. */ export async function requestWorkspaceRecovery(input: Source & { db: Db; idempotencyKey: string; actorId: string }) { return input.db.transaction(async tx => { @@ -283,6 +360,8 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ await updateSystemRecovery(db, input, current!, failSystemRecovery(current!, failure, { ...(typeof retryAfterAt === "number" && Number.isFinite(retryAfterAt) ? { retryAfterAt } : {}), })); + if (failure.retryClass === "recheck_evidence" && ["agent_stop_unconfirmed", "copyback_stop_unconfirmed"].includes(failure.code)) + await recheckStoppedWorkspaceRecovery(db, input, false); logger.warn({ err: error, runId: input.runId, failureCode: failure.code, phase: current!.phase }, "Workspace preservation deferred; source retained"); } finally { clearInterval(observer); await observing; } }; @@ -293,6 +372,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ const reserved = current && reserveSystemRecoveryAttempt(current); if (current && reserved) await updateSystemRecovery(db, input, current, failSystemRecovery(reserved, new SystemRecoveryFailure("copyback_stop_unconfirmed", "recheck_evidence"))); + await recheckStoppedWorkspaceRecovery(db, input, false); } }