diff --git a/doc/task-recovery.md b/doc/task-recovery.md index 22ad527af6e..b4bdd73334c 100644 --- a/doc/task-recovery.md +++ b/doc/task-recovery.md @@ -64,6 +64,18 @@ workspace strategy for that run. It does not borrow another run's strategy or skip any hold's independent containment and effect checks. Attaching a Retry to active automatic recovery preserves its generation, attempt count and deadline. +A deliberate board Retry can continue a failed native preparation after its +saved remote workspace is verified. The source must retain its unresolved native +adapter claim and the server's failed-preparation settlement records, with no +provider execution, session, input or result evidence, and no active controller +or native coordinator. Every source +lease must have exact stop and cleanup proof. The saved snapshot must match the +source, workspace epoch, generation and clean writer state. Resolving the +preservation action does not discard the canonical Retry source. The same +request resumes through normal admission and current repository access checks. +Unknown execution, operator Stop, changed ownership and missing receipts keep +the request held. Preservation alone does not authorize automatic replay. + Agent Chat additionally requires the current conversation owner and enabled Agent Chat. Admission and dispatch both recheck these conditions under the issue lock. An unadmitted receipt from a former owner cannot own the current diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 88bb022617a..5c7699107d7 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -1,6 +1,7 @@ import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; import { and, eq, inArray, isNotNull, or } from "drizzle-orm"; import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { isUuidLike } from "@paperclipai/shared"; import { claimedAdapterType } from "./conversation-continuation.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; @@ -8,6 +9,7 @@ import { canContinueCancelledRun, readRunCancellation } from "./run-cancellation import { legacyControllerBootId } from "./legacy-controller-lease.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; +import { readFailedPreparationSettlement } from "./failed-preparation-settlement.js"; type Run = Pick new Date()) return false; - const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + const leases = await db.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), - )).limit(1); + )); const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( eq(nativeRunResults.companyId, run.companyId), eq(nativeRunResults.runId, run.id), )).limit(1); - return !lease && !result && !await nativeStartupHasExecutionEvidence(db, run); + if (result || await nativeStartupHasExecutionEvidence(db, run)) return false; + if (!leases.length && run.runnerProfileJson?.remoteWorkspace == null) return true; + const issueId = run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId; + return typeof issueId === "string" && leases.length > 0 && leases.every(lease => lease.issueId === issueId + && lease.metadata?.pendingCleanupInFlight !== true && hasRemoteTerminationReceipt(lease)) + && await hasSettledFailedPreparationEvidence(db, run) + && await hasVerifiedWorkspaceRepair(db, run); +} + +async function hasSettledFailedPreparationEvidence(db: Db, run: Run): Promise { + const [source] = await db.select({ agentId: heartbeatRuns.agentId, + runtimeModeResolverVersion: heartbeatRuns.runtimeModeResolverVersion, runtimeModeReason: heartbeatRuns.runtimeModeReason, + runnerInstanceId: heartbeatRuns.runnerInstanceId, nativePhase: heartbeatRuns.nativePhase, + driverKind: heartbeatRuns.driverKind, driverVersion: heartbeatRuns.driverVersion, + completionContractId: heartbeatRuns.completionContractId, completionContractSha256: heartbeatRuns.completionContractSha256, + sessionIdBefore: heartbeatRuns.sessionIdBefore, externalRunId: heartbeatRuns.externalRunId, + logStore: heartbeatRuns.logStore, logRef: heartbeatRuns.logRef, logBytes: heartbeatRuns.logBytes, + logSha256: heartbeatRuns.logSha256, logCompressed: heartbeatRuns.logCompressed, + stdoutExcerpt: heartbeatRuns.stdoutExcerpt, stderrExcerpt: heartbeatRuns.stderrExcerpt, + lastOutputAt: heartbeatRuns.lastOutputAt, lastOutputSeq: heartbeatRuns.lastOutputSeq, + lastOutputStream: heartbeatRuns.lastOutputStream, lastOutputBytes: heartbeatRuns.lastOutputBytes, + usageJson: heartbeatRuns.usageJson, exitCode: heartbeatRuns.exitCode, signal: heartbeatRuns.signal, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id))); + if (!source || source.logCompressed || source.lastOutputSeq !== 0 || Object.entries(source).some(([key, value]) => + !["agentId", "logCompressed", "lastOutputSeq", "usageJson"].includes(key) && value !== null) + || !hasOnlyPreparationAccountingContext(source.usageJson, run.issueId ?? run.contextSnapshot?.issueId) + || Object.keys(run.runnerProfileJson ?? {}).some(key => !["adapterDispatch", "remoteWorkspace", + "chatControlRecoveryAdmission"].includes(key)) + || run.resultJson?.stopReason !== "adapter_failed" + || Object.keys(run.resultJson).some(key => !["stopReason", "timeoutFired", "timeoutSource", "executionRecovery", + "timeoutConfigured", "effectiveTimeoutSec", "systemRecoveryAdoption", "workspaceRestoreFailure", + "workspaceRestoreRecovery"].includes(key))) return false; + return Boolean(await readFailedPreparationSettlement(db, { ...run, agentId: source.agentId })); +} + +function hasOnlyPreparationAccountingContext(usage: unknown, issueId: unknown): boolean { + if (usage === null) return true; + if (!usage || typeof usage !== "object" || Array.isArray(usage) + || Object.keys(usage).length !== 1 || !Object.hasOwn(usage, "ledgerScope")) return false; + const scope = (usage as Record).ledgerScope as Record | null; + return Boolean(scope && typeof scope === "object" && !Array.isArray(scope) && Object.keys(scope).length === 3 + && scope.issueId === issueId && (scope.projectId === null || typeof scope.projectId === "string" && isUuidLike(scope.projectId)) + && (scope.billingCode === null || typeof scope.billingCode === "string")); } /** Candidate state only. Admission must also verify ownership, execution records, diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index a153e4202b2..d4b04366d78 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -253,6 +253,17 @@ export async function admitExplicitNativeContinuation(input: { sourceChecks.push({ cause: "native_continuation_requires_reconciliation", evidence: { runId: latest.id } }); } } + if (taskRetry && !sourceChecks.some(source => (source.evidence.runId ?? source.evidence.sourceRunId) === taskRetry.sourceRunId)) { + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, taskRetry.sourceRunId), + )).for("update"); + const [source] = await db.select(continuationRunFields).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, taskRetry.sourceRunId), eq(heartbeatRuns.agentId, agentId), + )).for("update"); + if (!source || (source.nativeIssueId ?? source.issueId ?? source.contextSnapshot?.issueId) !== issueId + || adapterExecutionControls.has(source.id) || !await isFailedNativePreparation(db, source, coordinator)) return null; + sourceChecks.push({ cause: "legacy_execution_requires_reconciliation", evidence: { runId: source.id } }); + } if (!sourceChecks.length) return null; if (blocker && blocker.recoveryActionId === null) return null; const [pendingInteraction] = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( @@ -445,17 +456,6 @@ export async function admitExplicitNativeContinuation(input: { } sources.push(run); } - if (taskRetry && !sources.some(run => run.id === taskRetry.sourceRunId)) { - const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( - eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, taskRetry.sourceRunId), - )).for("update"); - const [source] = await db.select(continuationRunFields).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, taskRetry.sourceRunId), eq(heartbeatRuns.agentId, agentId), - )).for("update"); - if (!source || (source.nativeIssueId ?? source.issueId ?? source.contextSnapshot?.issueId) !== issueId - || adapterExecutionControls.has(source.id) || !await isFailedNativePreparation(db, source, coordinator)) return null; - sources.push(source); - } const nativeSources = sources.filter(run => run.runtimeMode === "native"); const executedSources = sources.filter(run => run.runtimeMode === "native" || run.errorCode !== "execution_reconciliation_required" || run.startedAt); if (!executedSources.length || (retry && !sources.some(run => run.id === input.failedRunId)) || diff --git a/server/src/services/failed-preparation-settlement.ts b/server/src/services/failed-preparation-settlement.ts new file mode 100644 index 00000000000..c46007a2c41 --- /dev/null +++ b/server/src/services/failed-preparation-settlement.ts @@ -0,0 +1,38 @@ +import { and, asc, eq } from "drizzle-orm"; +import { heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; +import { isUuidLike } from "@paperclipai/shared"; + +type Source = Pick; + +export async function readFailedPreparationSettlement(db: Db, source: Source): Promise<{ + unsupportedHarnessWarning: { message: string | null; revision: string } | null; +} | null> { + if (typeof source.error !== "string" || !source.error.length) return null; + const events = await db.select().from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, source.companyId), eq(heartbeatRunEvents.runId, source.id), + )).orderBy(asc(heartbeatRunEvents.seq)).limit(4); + const hostEvent = (event: typeof heartbeatRunEvents.$inferSelect) => event.agentId === source.agentId + && event.stream === "system" && event.sourceInstanceId === null && event.sourceEventId === null + && event.sourceSeq === null && event.sourcePayloadSha256 === null && event.protocolSchemaVersion === null; + let offset = 0; + let unsupportedHarnessWarning: { message: string | null; revision: string } | null = null; + const warning = events[0]; + if (warning?.eventType === "agent_observability.unavailable") { + if (!hostEvent(warning) || warning.seq !== 1 || warning.level !== "warn" + || warning.payload?.reason !== "unsupported_harness" || typeof warning.payload.revision !== "string" || !isUuidLike(warning.payload.revision) + || Object.keys(warning.payload).some(key => !["reason", "revision"].includes(key))) return null; + unsupportedHarnessWarning = { message: warning.message, revision: warning.payload.revision }; + offset = 1; + } + const [error, measured] = events.slice(offset); + if (events.length !== 2 + offset || !error || !measured || !hostEvent(error) || !hostEvent(measured) + || error.seq !== 1 + offset || error.eventType !== "error" || error.level !== "error" + || error.message !== source.error || error.payload !== null + || measured.seq !== 2 + offset || measured.eventType !== "lifecycle" || measured.level !== "info" || measured.message !== "task.run.measured" + || measured.payload?.phase !== "task.run.measured" || measured.payload.outcome !== "failed" + || measured.payload.durableSettlement !== true + || Object.keys(measured.payload).some(key => !["phase", "outcome", "durationMs", "durableSettlement"].includes(key)) + || typeof measured.payload.durationMs !== "number" + || !Number.isFinite(measured.payload.durationMs) || measured.payload.durationMs < 0) return null; + return { unsupportedHarnessWarning }; +} diff --git a/server/src/services/fresh-workspace-continuation.ts b/server/src/services/fresh-workspace-continuation.ts index 6789a12b928..aa49f00ddd1 100644 --- a/server/src/services/fresh-workspace-continuation.ts +++ b/server/src/services/fresh-workspace-continuation.ts @@ -1,5 +1,5 @@ -import { and, asc, eq, sql } from "drizzle-orm"; -import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; +import { and, eq, sql } from "drizzle-orm"; +import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; import { persistActivity } from "./activity-log.js"; import { readSystemRecovery } from "./system-recovery-state.js"; @@ -10,6 +10,7 @@ import { hasPendingRunCleanup } from "./execution-blocker.js"; import { toExecutionWorkspace } from "./execution-workspaces.js"; import { isBoundIsolatedTaskDirectory } from "./isolated-task-directory.js"; import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; +import { readFailedPreparationSettlement } from "./failed-preparation-settlement.js"; export type ContinuationReadiness = | { mode: "resume_preserved_workspace"; preservationReceiptId: string } @@ -74,30 +75,10 @@ async function reusableDirectoryAfterPreparationFailure(db: Db, source: typeof h Object.keys(source.usageJson ?? {}).some(key => key !== "ledgerScope") || Object.keys(source.runnerProfileJson ?? {}).some(key => !["adapterDispatch", "chatControlRecoveryAdmission", "aiConnectionNonAssigneeCommentWake"].includes(key)) || Object.keys(source.resultJson ?? {}).some(key => !["effectiveTimeoutSec", "executionRecovery", "stopReason", "timeoutConfigured", "timeoutFired", "timeoutSource", "workspaceValidation"].includes(key))) return null; - const events = await db.select().from(heartbeatRunEvents).where(and( - eq(heartbeatRunEvents.companyId, source.companyId), eq(heartbeatRunEvents.runId, source.id), - )).orderBy(asc(heartbeatRunEvents.seq)).limit(4); - let offset = 0; - const warning = events[0]; - if (warning?.eventType === "agent_observability.unavailable") { - if (warning.seq !== 1 || warning.level !== "warn" || - warning.message !== "Company observability was not applied: this harness is not supported." || - warning.payload?.reason !== "unsupported_harness" || !z.string().uuid().safeParse(warning.payload.revision).success || - Object.keys(warning.payload).some(key => !["reason", "revision"].includes(key))) return null; - offset = 1; - } - const diagnostic = events[offset], measured = events[offset + 1]; - if (events.length !== 2 + offset || !diagnostic || diagnostic.seq !== 1 + offset || - diagnostic.eventType !== "error" || diagnostic.level !== "error" || diagnostic.message !== source.error || - diagnostic.payload !== null || !measured || measured.seq !== 2 + offset || - measured.eventType !== "lifecycle" || measured.level !== "info" || measured.message !== "task.run.measured" || - measured.payload?.phase !== "task.run.measured" || measured.payload?.outcome !== "failed" || - measured.payload?.durableSettlement !== true || typeof measured.payload.durationMs !== "number" || - !Number.isFinite(measured.payload.durationMs) || measured.payload.durationMs < 0 || - Object.keys(measured.payload).some(key => !["phase", "outcome", "durableSettlement", "durationMs"].includes(key)) || - events.some(event => event.stream !== "system" || event.agentId !== source.agentId || - event.sourceInstanceId !== null || event.sourceEventId !== null || event.sourceSeq !== null || - event.sourcePayloadSha256 !== null || event.protocolSchemaVersion !== null)) return null; + const settlement = await readFailedPreparationSettlement(db, source); + const warning = settlement?.unsupportedHarnessWarning; + if (!settlement || warning && (warning.message !== "Company observability was not applied: this harness is not supported." + || !z.string().uuid().safeParse(warning.revision).success)) return null; const [executionEvidence] = await db.execute<{ present: boolean }>(sql`select exists(select 1 from ${environmentLeases} where ${environmentLeases.companyId} = ${source.companyId} and ${environmentLeases.heartbeatRunId} = ${source.id}) or exists(select 1 from ${nativeRunResults} where ${nativeRunResults.companyId} = ${source.companyId} and ${nativeRunResults.runId} = ${source.id}) diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index c93da025b86..b88e3f7fd19 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -22,6 +22,7 @@ import { DELIVERY_QUEUES, subscribeDeliveryWork } from "./delivery-work-notifica import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js"; import { createHeartbeatWorkspaceResolver, resolveExecutionWorkspaceReuseRequestForIssue } from "./heartbeat/workspaces.js"; import { environmentService } from "./environments.js"; +import { taskRepositoryRelativePath } from "./execution-workspace-repositories.js"; import { classifyEnvironmentCapabilities, environmentRuntimeService } from "./environment-runtime.js"; import { environmentRunOrchestrator } from "./environment-run-orchestrator.js"; import { executionWorkspaceService } from "./execution-workspaces.js"; @@ -33,25 +34,28 @@ import { GitShallowMetadataError } from "@paperclipai/adapter-utils/git-workspac import { readSystemRecovery } from "./system-recovery-state.js"; import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js"; import { getStorageService } from "../storage/index.js"; +import * as storageBackend from "../storage/index.js"; import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js"; import { createSystemRecovery, failSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; import { requestTaskRetry } from "./task-retry.js"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; +import { materializeIsolatedTaskDirectory } from "./isolated-task-directory.js"; import { execFile } from "node:child_process"; import { generateKeyPairSync, randomBytes, randomUUID, sign } from "node:crypto"; -import { cp, mkdir, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises"; +import { chmod, cp, lstat, mkdir, mkdtemp, readFile, readdir, realpath, rm, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import express from "express"; import request from "supertest"; import { and, eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, approvals, budgetPolicies, costEvents, completionContracts, issueApprovals, issueTreeHolds, issueTreeHoldMembers, companies, createDb, applyPendingMigrations, environmentLeases, environments, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, agentWakeupRequests, workspaceOperations } from "@paperclipai/db"; +import { companyObservability, executionWorkspaceRepositories, issueWorkProducts, authUsers, companyMemberships, agents, approvals, budgetPolicies, costEvents, completionContracts, issueApprovals, issueTreeHolds, issueTreeHoldMembers, companies, createDb, applyPendingMigrations, environmentLeases, environments, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, agentWakeupRequests, workspaceOperations } from "@paperclipai/db"; import { resumePreservedResultFinalization } from "./preserved-result-finalization.js"; import { finalizeNativeRun } from "./native-runtime/native-run-finalizer.js"; import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; +import type { NativeExecutionInput, NativeSession, NativeSessionBackend, PrpEvent } from "../vendor/paperclip-runner/index.js"; import { issueThreadInteractions } from "@paperclipai/db"; import { issueRecoveryActionService } from "./issue-recovery-actions.js"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; @@ -106,7 +110,16 @@ describe("automatic retained workspace recovery", () => { if (external) await db.execute(sql`TRUNCATE TABLE companies CASCADE`); }, 30_000); afterAll(async () => { - await db?.$client.end(); await temporary?.cleanup(); await rm(root, { recursive: true, force: true }); + await db?.$client.end(); await temporary?.cleanup(); + const makeWritable = async (directory: string): Promise => { + const entry = await lstat(directory); + if (!entry.isDirectory() || entry.isSymbolicLink()) return; + await chmod(directory, 0o700); + for (const child of await readdir(directory, { withFileTypes: true })) { + if (child.isDirectory()) await makeWritable(path.join(directory, child.name)); + } + }; + await makeWritable(root); await rm(root, { recursive: true, force: true }); 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; }); @@ -339,10 +352,10 @@ describe("automatic retained workspace recovery", () => { } }, 30_000); - function boardRecoveryApp() { + function boardRecoveryApp(actor: { userId: string; source: "local_implicit" | "session"; companyIds?: string[] } = { userId: "local-board", source: "local_implicit" }) { const app = express(); app.use(express.json()); - app.use((req, _res, next) => { req.actor = { type: "board", source: "local_implicit", userId: "local-board" }; next(); }); + app.use((req, _res, next) => { req.actor = { type: "board", ...actor }; next(); }); app.use("/api", issueRoutes(db, {} as Parameters[1])); app.use(errorHandler); return app; @@ -420,6 +433,444 @@ describe("automatic retained workspace recovery", () => { return f; } + async function failedRemotePreparation() { + const f = await interruptedRemotePreparation(); + const source = await run(f.runId); + const error = "Repository access is no longer available; refresh the repository catalog"; + await terminalizeLegacyExecution({ db, run: source, status: "failed", reconcileIfNeeded: false, + patch: { errorCode: "setup_failed", error, executionStage: "preparing", + startedAt: new Date(Date.now() - 1000), contextSnapshot: { ...source.contextSnapshot, source: "native_status_decision" }, + resultJson: { ...source.resultJson, stopReason: "adapter_failed", + executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } } }); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: source.agentId, + eventType: "error", stream: "system", level: "error", message: error }); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: source.agentId, + eventType: "lifecycle", stream: "system", level: "info", message: "task.run.measured", + payload: { phase: "task.run.measured", outcome: "failed", durationMs: 1000, durableSettlement: true } }); + return f; + } + + it("admits one HTTP Retry after a failed native preparation has verified saved work", async () => { + const f = await failedRemotePreparation(); + await recoverRetainedWorkspace(f.input); + const source = await run(f.runId); + expect(source).toMatchObject({ runtimeMode: "legacy", runtimeModeResolvedAt: null, executionStage: "preparing", + status: "failed", errorCode: "setup_failed", nativeIssueId: null, nativeSessionId: null, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }); + expect(await hasVerifiedWorkspaceRepair(db, source)).toBe(true); + const app = boardRecoveryApp(), heartbeat = heartbeatService(db); + const intent = { observedRunId: f.runId, idempotencyKey: randomUUID() }; + const receipt = (await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body; + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId), + heartbeat.resumeTaskRetry(f.companyId, receipt.requestId)]); + expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body.requestId).toBe(receipt.requestId); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "queued", contextSnapshot: { + previousRunId: f.runId, taskRetryRequestId: receipt.requestId } }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId)))[0]) + .toMatchObject({ status: "coalesced", runId: successors[0].id }); + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "resume_preserved_workspace" }); + const remote = source.runnerProfileJson!.remoteWorkspace as { snapshotId: string }; + const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId, + snapshotId: remote.snapshotId }); + try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); } + finally { await saved.cleanup(); } + expect((await run(f.runId)).error).toBe(source.error); + }, 30_000); + + it("resumes the same failed-preparation HTTP Retry from the real preservation worker", async () => { + const f = await failedRemotePreparation(); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + await db.update(environmentLeases).set({ status: "active", releasedAt: null, cleanupStatus: null, + metadata: { ...lease.metadata, remoteExecutionTermination: null } }).where(eq(environmentLeases.id, lease.id)); + let releaseStop!: () => void; + const stopped = new Promise(resolve => { releaseStop = resolve; }); + f.runtime.retryPendingSandboxTeardown.mockImplementation(async () => { + await stopped; + return { providerLeaseId: f.providerLeaseId, state: "stopped" }; + }); + const errors = vi.fn(), heartbeat = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors }); + const worker = registerWorkspaceRecoveryWork(db, coordinator, f.environmentRuntime, heartbeat.resumeRemoteStopComments); + const successors = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${requestId}`)); + let requestId: string; + try { + await worker.ready; + const app = boardRecoveryApp(), intent = { observedRunId: f.runId, idempotencyKey: randomUUID() }; + const receipt = (await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body; + requestId = receipt.requestId; + expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body.requestId).toBe(requestId); + await expect.poll(() => f.runtime.retryPendingSandboxTeardown.mock.calls.length, { timeout: 20_000 }).toBe(1); + expect(await successors()).toHaveLength(0); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0].metadata?.remoteExecutionTermination).toBeFalsy(); + releaseStop(); + await expect.poll(async () => (await successors()).length, { timeout: 20_000 }).toBe(1); + const [successor] = await successors(); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: requestId }); + expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ mode: "resume_preserved_workspace" }); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + const remote = (await run(f.runId)).runnerProfileJson!.remoteWorkspace as { snapshotId: string }; + const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId, snapshotId: remote.snapshotId }); + try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); } + finally { await saved.cleanup(); } + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, requestId), heartbeat.resumeRemoteStopComments(await run(f.runId))]); + expect(await successors()).toHaveLength(1); + expect(f.runtime.resumeRunLease).toHaveBeenCalledOnce(); + expect((await run(f.runId))).toMatchObject({ status: "failed", errorCode: "setup_failed" }); + expect(errors).not.toHaveBeenCalled(); + } finally { releaseStop(); await coordinator.stop(); await worker.stop(); } + }, 40_000); + + it.each(["delayed_provider_event", "native_input", "session_checkpoint", "controller_recovery", "process", "coordinator", + "runtime_selected", "live_controller", "missing_error", "missing_measurement", "foreign_error", "unknown_result", + "wrong_snapshot", "wrong_generation", "active_writer", "missing_stop", "foreign_stop", "additional_unstopped_lease", + "operator_stop", "newer_input", "reassigned", "wrong_actor", "actor_permission_revoked", + "provider_usage", "unknown_usage", "invalid_ledger_scope", "foreign_ledger_scope", + "array_usage", "false_usage", "zero_usage", "empty_usage_string", "empty_usage_object", + "active_lease", "pending_cleanup", "cleanup_failed", "cleanup_unknown", "cleanup_inflight", "unreleased", "foreign_lease_issue", "local_provider_claim"] as const)( + "holds a failed-preparation HTTP Retry when current proof changes: %s", async gate => { + const f = await failedRemotePreparation(); + await recoverRetainedWorkspace(f.input); + const source = await run(f.runId), heartbeat = heartbeatService(db); + const actorId = gate === "actor_permission_revoked" ? randomUUID() : "local-board"; + if (gate === "actor_permission_revoked") { + await db.insert(authUsers).values({ id: actorId, name: "Retry owner", email: `${actorId}@example.invalid`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId: f.companyId, principalType: "user", principalId: actorId, membershipRole: "owner" }); + } + const app = boardRecoveryApp({ userId: actorId, source: gate === "actor_permission_revoked" ? "session" : "local_implicit", companyIds: [f.companyId] }); + const receipt = (await request(app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body; + if (gate === "delayed_provider_event") await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: source.agentId, eventType: "session.started", stream: "system", payload: { sessionId: randomUUID() }, + nativeSource: { sourceInstanceId: randomUUID(), sourceEventId: randomUUID(), sourceSeq: 1, protocolSchemaVersion: 1, + canonicalPayload: { type: "session.started", sessionId: randomUUID() } } }); + if (["native_input", "session_checkpoint", "controller_recovery"].includes(gate)) { + const key = gate === "native_input" ? "nativeExecutionInput" : gate === "session_checkpoint" ? "sessionCheckpoint" : "nativeControllerRecovery"; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson, [key]: { sessionId: randomUUID() } } }) + .where(eq(heartbeatRuns.id, f.runId)); + } + if (gate === "process") await db.update(heartbeatRuns).set({ processPid: 1386 }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "coordinator") { + await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, + issueId: f.issueId, runId: f.runId, phase: "observed", attempt: 1 }); + } + if (gate === "runtime_selected") await db.update(heartbeatRuns).set({ runtimeModeResolvedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "live_controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, f.runId)); + if (gate === "missing_error" || gate === "missing_measurement") await db.delete(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, gate === "missing_error" ? 1 : 2))); + if (gate === "foreign_error") { + const [foreign] = await db.insert(agents).values({ companyId: f.companyId, name: "Other owner", adapterType: "paperclip_runner" }).returning(); + await db.update(heartbeatRunEvents).set({ agentId: foreign.id }) + .where(and(eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, 1))); + } + if (["array_usage", "false_usage", "zero_usage", "empty_usage_string", "empty_usage_object"].includes(gate)) { + const invalid = gate === "array_usage" ? [] : gate === "false_usage" ? false : gate === "zero_usage" ? 0 + : gate === "empty_usage_string" ? "" : {}; + await db.update(heartbeatRuns).set({ usageJson: sql`${JSON.stringify(invalid)}::jsonb` }).where(eq(heartbeatRuns.id, f.runId)); + } + if (["provider_usage", "unknown_usage", "invalid_ledger_scope", "foreign_ledger_scope"].includes(gate)) { + const ledgerScope = { issueId: gate === "foreign_ledger_scope" ? randomUUID() : f.issueId, projectId: null, billingCode: null }; + await db.update(heartbeatRuns).set({ usageJson: gate === "provider_usage" ? { ledgerScope, inputTokens: 0 } + : gate === "unknown_usage" ? { ledgerScope, unknown: true } + : { ledgerScope: gate === "invalid_ledger_scope" ? { ...ledgerScope, providerWorkStarted: false } : ledgerScope } }) + .where(eq(heartbeatRuns.id, f.runId)); + } + if (gate === "unknown_result") await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson, + output: "Unknown provider work must not be replayed" } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "wrong_snapshot") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson, + remoteWorkspace: { ...(source.runnerProfileJson!.remoteWorkspace as Record), snapshotId: randomUUID() } } }) + .where(eq(heartbeatRuns.id, f.runId)); + if (gate === "wrong_generation") await db.update(workspaceOperations).set({ metadata: sql`jsonb_set(${workspaceOperations.metadata}, '{generation}', '999')` }) + .where(and(eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "workspace_recovery"))); + if (gate === "active_writer") await db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,writerRunId}', ${JSON.stringify(randomUUID())}::jsonb)` }) + .where(eq(executionWorkspaces.id, f.workspaceId)); + if (gate === "missing_stop" || gate === "foreign_stop") { + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + await db.update(environmentLeases).set({ metadata: { ...lease.metadata, remoteExecutionTermination: gate === "missing_stop" + ? null : { ...(lease.metadata!.remoteExecutionTermination as Record), runId: randomUUID() } } }) + .where(eq(environmentLeases.id, lease.id)); + } + if (["active_lease", "pending_cleanup", "cleanup_failed", "cleanup_unknown", "cleanup_inflight", "unreleased", "local_provider_claim"].includes(gate)) { + await db.update(environmentLeases).set({ + ...(gate === "active_lease" ? { status: "active" } : gate === "pending_cleanup" ? { status: "pending_cleanup" } : {}), + ...(gate === "cleanup_failed" ? { cleanupStatus: "failed" } : gate === "cleanup_unknown" ? { cleanupStatus: null } : {}), + ...(gate === "unreleased" ? { releasedAt: null } : {}), + ...(gate === "cleanup_inflight" ? { metadata: sql`jsonb_set(${environmentLeases.metadata}, '{pendingCleanupInFlight}', 'true')` } : {}), + ...(gate === "local_provider_claim" ? { provider: "local", providerLeaseId: null, + metadata: sql`${environmentLeases.metadata} - 'remoteExecutionTermination'` } : {}), + }).where(eq(environmentLeases.id, f.lease.id)); + } + if (gate === "foreign_lease_issue") { + const [foreign] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task", status: "blocked", assigneeAgentId: source.agentId }).returning(); + await db.update(environmentLeases).set({ issueId: foreign.id }).where(eq(environmentLeases.id, f.lease.id)); + } + if (gate === "additional_unstopped_lease") await db.insert(environmentLeases).values({ companyId: f.companyId, + issueId: f.issueId, heartbeatRunId: f.runId, environmentId: f.lease.environmentId, + provider: "daytona", providerLeaseId: randomUUID(), status: "active" }); + if (gate === "operator_stop") await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "operator_interrupted", + resultJson: { ...source.resultJson, cancellation: requestedRunCancellation({ cancelledByActorType: "user", cancelledByUserId: "local-board" }, "Stop requested") } }) + .where(eq(heartbeatRuns.id, f.runId)); + if (gate === "newer_input") await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, + issueId: f.issueId, contextSnapshot: { issueId: f.issueId, source: "issue.comment.reopen" }, status: "running" }); + if (gate === "reassigned") await db.update(issues).set({ assigneeAgentId: null, assigneeUserId: "board" }).where(eq(issues.id, f.issueId)); + if (gate === "wrong_actor") await db.update(agentWakeupRequests).set({ requestedByActorId: "another-user" }) + .where(eq(agentWakeupRequests.id, receipt.requestId)); + if (gate === "actor_permission_revoked") await db.update(companyMemberships).set({ status: "suspended" }) + .where(and(eq(companyMemberships.companyId, f.companyId), eq(companyMemberships.principalId, actorId))); + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId), heartbeat.resumeTaskRetry(f.companyId, receipt.requestId)]); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`))).toHaveLength(0); + expect((await run(f.runId)).error).toBe(source.error); + }, 30_000); + + it.each(["settled", "active", "provider_handle"] as const)("keeps unsupported mixed local and remote source leases held: %s", async state => { + const f = await failedRemotePreparation(); + await recoverRetainedWorkspace(f.input); + await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.runId, issueId: f.issueId, + provider: "local", providerLeaseId: state === "provider_handle" ? randomUUID() : null, + status: state === "active" ? "active" : "released", releasedAt: new Date(), cleanupStatus: "success" }); + const receipt = (await request(boardRecoveryApp()).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body; + await heartbeatService(db).resumeTaskRetry(f.companyId, receipt.requestId); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`))).toHaveLength(0); + }, 30_000); + + it.each(["plain", "unsupported_warning", "warning_presentation", "provider_warning", "measurement"] as const)( + "keeps the existing-directory readiness settlement contract: %s", async mode => { + const companyId = randomUUID(), issueId = randomUUID(), agentId = randomUUID(), sourceRunId = randomUUID(), + successorRunId = randomUUID(), requestId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Preparation settlement", issuePrefix: `S${companyId.slice(0, 6)}` }); + fixtureCompanyIds.add(companyId); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue existing directory", status: "blocked", assigneeAgentId: agentId }); + const cwd = await materializeIsolatedTaskDirectory({ companyId, issueId }); + await writeFile(path.join(cwd, "saved.txt"), "existing task work\n"); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId, sourceIssueId: issueId, + mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", name: "Task directory", cwd }).returning(); + await db.update(issues).set({ executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing" }).where(eq(issues.id, issueId)); + const error = "The selected existing task directory has no Git checkout."; + await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId, issueId, runtimeMode: "legacy", + status: "failed", finishedAt: new Date(), errorCode: "workspace_validation_failed", error, executionStage: "preparing", + contextSnapshot: { issueId }, runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + usageJson: { ledgerScope: { issueId, projectId: null, billingCode: null } }, + resultJson: { stopReason: "adapter_failed", executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + workspaceValidation: { reason: "git_worktree_base_not_git_checkout", issueId, + resolvedWorkspaceSource: "task_session", requestedExecutionWorkspaceMode: "isolated_workspace", + issueExecutionWorkspacePreference: "reuse_existing", issueExecutionWorkspaceId: workspace.id, + issueProjectId: null, issueProjectWorkspaceId: null, resolvedWorkspaceCwd: cwd } } }); + if (mode.includes("warning")) await appendHeartbeatRunEvent(db, { companyId, agentId, runId: sourceRunId, + eventType: "agent_observability.unavailable", level: "warn", stream: "system", + message: mode === "warning_presentation" ? "Changed presentation" : "Company observability was not applied: this harness is not supported.", + payload: { reason: "unsupported_harness", revision: randomUUID() }, + ...(mode === "provider_warning" ? { nativeSource: { sourceInstanceId: randomUUID(), sourceEventId: randomUUID(), sourceSeq: 1, + protocolSchemaVersion: 1, canonicalPayload: { type: "agent_observability.unavailable" } } } : {}) }); + await appendHeartbeatRunEvent(db, { companyId, agentId, runId: sourceRunId, eventType: "error", level: "error", stream: "system", message: error }); + await appendHeartbeatRunEvent(db, { companyId, agentId, runId: sourceRunId, eventType: "lifecycle", level: "info", stream: "system", + message: "task.run.measured", payload: { phase: "task.run.measured", outcome: "failed", durationMs: 1, durableSettlement: mode !== "measurement" } }); + const [successor] = await db.insert(heartbeatRuns).values({ id: successorRunId, companyId, agentId, issueId, status: "queued", + contextSnapshot: { issueId, previousRunId: sourceRunId, taskRetryRequestId: requestId } }).returning(); + await db.insert(agentWakeupRequests).values({ id: requestId, companyId, agentId, source: "on_demand", triggerDetail: "manual", reason: "task_retry", + status: "coalesced", runId: successor.id, requestedByActorType: "user", requestedByActorId: "local-board", + idempotencyKey: `task-retry:${randomUUID()}`, payload: { issueId, taskRetry: { schema: "paperclip.task-retry.v1", + generationId: requestId, sourceRunId, actorId: "local-board", actorSource: "session", strategy: "evaluate" } } }); + const reuse = mode === "plain" || mode === "unsupported_warning"; + expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject(reuse + ? { mode: "reuse_current_workspace", retryIntentId: requestId, sourceRunId, workspaceId: workspace.id, cwd } + : { mode: "fresh_isolated_workspace", retryIntentId: requestId, sourceRunId, retainedSourceIds: [] }); + expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("existing task work\n"); + }); + + it.each(["complete", "catalog", "catalog_warning", "catalog_unknown_warning", "catalog_provider_warning"] as const)( + "dispatches the recovered failed-preparation Retry once through the native provider boundary (%s)", async mode => { + const unavailableCatalog = mode !== "complete", withWarning = mode.includes("warning"); + const localStorage = getStorageService(); + const databaseUrl = process.env.DATABASE_URL, masterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + const experimental = await instanceSettingsService(db).getExperimental(); + let storage: { mockRestore(): void } | undefined; + let ownedHeartbeat: ReturnType | undefined; + let recoveryCoordinator: ReturnType | undefined; + let recoveryWorker: ReturnType | undefined; + try { + storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" }); + process.env.DATABASE_URL = connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "63".repeat(32); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableIsolatedWorkspaces: true }); + const f = await failedRemotePreparation(); + await recoverRetainedWorkspace(f.input); + const source = await run(f.runId), nextDirectory = path.join(root, randomUUID(), "next"); + await mkdir(nextDirectory, { recursive: true }); + const nextRemote = await realpath(nextDirectory); + await db.insert(authUsers).values({ id: "local-board", name: "Local operator", email: "local-board@example.invalid", + createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId: f.companyId, principalType: "user", principalId: "local-board", membershipRole: "owner" }); + await db.update(agents).set({ defaultEnvironmentId: f.lease.environmentId, + adapterConfig: withWarning ? { provider: "acpx", acpxAgent: "pi", model: "openrouter/deepseek/deepseek-v4.1-flash-1017" } + : { provider: "codex", model: "gpt-6.1-sol" } }).where(eq(agents.id, source.agentId)); + if (withWarning) await db.insert(companyObservability).values({ companyId: f.companyId, revision: randomUUID(), enabled: true, + destination: { provider: "otlp", endpointBaseUrl: "https://collector.example.invalid/otel" }, + capture: { userPrompts: false, assistantResponses: false, toolDetails: false, toolContent: false } }); + await db.update(environments).set({ config: { provider: "fake", image: "fake:test" } }).where(eq(environments.id, f.lease.environmentId!)); + const environment = environmentService(db), oldTarget = targets.get(f.lease.id)!; + oldTarget.runner!.syncIn = async operations => { + for (const operation of operations) for (const file of operation.files) { + await mkdir(path.dirname(file.targetPath), { recursive: true }); + await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); + } + return { filesTransferred: operations.reduce((count, operation) => count + operation.files.length, 0), bytesTransferred: 0, operations: [] }; + }; + const executeCommand = oldTarget.runner!.execute; + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", + acquireRunLease: async input => { + const lease = await environment.acquireLease({ companyId: input.companyId, environmentId: input.environment.id, + issueId: input.issueId, heartbeatRunId: input.heartbeatRunId, executionWorkspaceId: input.executionWorkspaceId, + provider: "fake", providerLeaseId: randomUUID(), leasePolicy: "retain_on_failure", + metadata: { driver: "sandbox", remoteCwd: nextRemote, agentId: input.agentId } }); + targets.set(lease.id, { ...oldTarget, leaseId: lease.id, remoteCwd: nextRemote, + runner: { ...oldTarget.runner!, execute: executeCommand }, + sandboxLeaseAcquisition: { outcome: "created", providerLeaseId: lease.providerLeaseId! } }); + return lease; + }, + releaseRunLease: async input => environment.releaseLease(input.lease.id, input.status, { cleanupStatus: "success", + remoteExecutionTermination: remoteTerminationReceipt(input.lease, { providerLeaseId: input.lease.providerLeaseId, state: "stopped" }) }), + resumeRunLease: async ({ lease }) => ({ providerLeaseId: lease.providerLeaseId!, metadata: { remoteCwd: lease.metadata?.remoteCwd } }), + retryPendingSandboxTeardown: async ({ lease }) => ({ providerLeaseId: lease.providerLeaseId, state: "stopped" }), + realizeWorkspace: async () => ({ cwd: nextRemote, metadata: { remoteCwd: nextRemote } }), + execute: async input => executeCommand({ command: input.command, args: input.args, cwd: input.cwd, + env: input.env, stdin: input.stdin, timeoutMs: input.timeoutMs }), + resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: [], declared: null, narrowing: null }), + }] }); + const workProductId = randomUUID(), evidenceRef = `work_product:${workProductId}`; + await db.insert(issueWorkProducts).values({ id: workProductId, companyId: f.companyId, issueId: f.issueId, + type: "artifact", provider: "paperclip", title: "Continued saved work", status: "ready_for_review", reviewState: "approved" }); + const inputs: NativeExecutionInput[] = [], startTurn = vi.fn(); + const backendFactory = (execution: NativeExecutionInput): NativeSessionBackend => { + inputs.push(execution); + const identity = { companyId: f.companyId, agentId: source.agentId, issueId: f.issueId, + runId: execution.binding.runId, sessionId: execution.session.normalizedSessionId! }; + const turnId = `turn-${identity.runId}`, result = structuredClone(CONTROL_PLANE_CONFORMANCE_RESULT); + result.completionClaim.contractRevision = execution.completionContract.contract.revision; + result.completionClaim.criteria = execution.completionContract.contract.criteria.map(criterion => ({ + ...result.completionClaim.criteria[0]!, criterionId: criterion.id, evidenceRefs: [evidenceRef] })); + result.evidence = [{ kind: "work_product", ref: evidenceRef }]; result.verification[0]!.artifactRef = evidenceRef; + let started!: () => void; + const turnStarted = new Promise(resolve => { started = resolve; }); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { + await turnStarted; + const current = await run(identity.runId); + yield { schema: "paperclip.prp.event.v1", sourceEventId: `${identity.runId}:terminal`, sourceSeq: 1, + sourceInstanceId: current.runnerInstanceId!, sourceKind: "runner", runId: identity.runId, + normalizedSessionId: identity.sessionId, turnId, eventType: "turn.completed", schemaVersion: 1, + priority: 0, emittedAt: new Date().toISOString(), payload: {} } satisfies PrpEvent; + }, + async startTurn(input) { + expect(execution.provider).toMatchObject({ kind: "codex", model: "gpt-6.1-sol" }); + expect(await readFile(path.join(execution.workspace.cwd!, "saved.txt"), "utf8")).toBe("finished work\n"); + await writeFile(path.join(execution.workspace.cwd!, "continued.txt"), "continued from preserved work\n"); + startTurn(input); started(); return { turnId }; + }, + async result() { await turnStarted; return { result, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId }; }, + async snapshot() { return { backendKind: "mock", sessionId: "recovered-preparation", identity, + providerSessionId: "recovered-preparation", activeTurnId: null, pendingRuntimeRequests: [], lineage: [], cursor: "1" }; }, + async close() {}, + }; + return { async descriptor() { return { kind: "mock", name: "recovered-preparation", version: "1", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }; }, + async openSession() { return session; } }; + }; + const heartbeat = heartbeatService(db, { environmentRuntime: runtime, nativeSessionBackendFactory: backendFactory }); + ownedHeartbeat = heartbeat; + const receipt = (await request(boardRecoveryApp()).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body; + await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)); + expect(successors).toHaveLength(1); + if (unavailableCatalog) { + const repoUrl = "https://github.com/example/revoked.git"; + await db.insert(executionWorkspaceRepositories).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, + requestedByIssueId: f.issueId, repositoryIdentity: repoUrl, repoUrl, catalogRepositoryId: "revoked-before-dispatch", + relativePath: taskRepositoryRelativePath(repoUrl), requestKey: randomUUID(), state: "ready", pinnedCommit: f.commit }); + } + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(and( + eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "running"))); + await heartbeat.resumeQueuedRuns(); await heartbeat.drainActiveRunExecutions(); + const completed = await run(successors[0].id); + if (unavailableCatalog) { + expect(completed).toMatchObject({ status: "failed", errorCode: "setup_failed", + error: "Repository access is no longer available; refresh the repository catalog" }); + expect(inputs).toHaveLength(0); expect(startTurn).not.toHaveBeenCalled(); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + expect(await readFile(path.join(workspace.cwd!, "saved.txt"), "utf8")).toBe("finished work\n"); + expect((await run(f.runId)).error).toBe(source.error); + expect(completed).toMatchObject({ runtimeMode: "legacy", runtimeModeResolvedAt: null, executionStage: "preparing", + nativeIssueId: null, nativeSessionId: null, usageJson: { ledgerScope: expect.any(Object) }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }); + expect(completed.usageJson).toEqual({ ledgerScope: { issueId: f.issueId, projectId: null, billingCode: null } }); + if (withWarning) { + const [warning] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, completed.id), eq(heartbeatRunEvents.seq, 1))); + expect(warning).toMatchObject({ eventType: "agent_observability.unavailable", stream: "system", level: "warn", + payload: { reason: "unsupported_harness", revision: expect.any(String) }, sourceEventId: null, sourceInstanceId: null }); + if (mode === "catalog_unknown_warning") await db.update(heartbeatRunEvents).set({ payload: { ...warning.payload, reason: "unknown" } }).where(eq(heartbeatRunEvents.id, warning.id)); + if (mode === "catalog_provider_warning") await db.update(heartbeatRunEvents).set({ sourceEventId: randomUUID() }).where(eq(heartbeatRunEvents.id, warning.id)); + } + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, status: "running" }); + const errors = vi.fn(); + recoveryCoordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors }); + recoveryWorker = registerWorkspaceRecoveryWork(db, recoveryCoordinator, runtime, heartbeat.resumeRemoteStopComments); + await recoveryWorker.ready; + const retryIntent = { observedRunId: completed.id, idempotencyKey: randomUUID() }; + const retry = (await request(boardRecoveryApp()).post(`/api/issues/${f.issueId}/retry`).send(retryIntent).expect(202)).body; + const resumed = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${retry.requestId}`)); + await expect.poll(async () => await hasVerifiedWorkspaceRepair(db, await run(completed.id)), { timeout: 20_000 }).toBe(true); + if (mode === "catalog_unknown_warning" || mode === "catalog_provider_warning") { + await heartbeat.resumeTaskRetry(f.companyId, retry.requestId); + expect(await resumed()).toHaveLength(0); expect(startTurn).not.toHaveBeenCalled(); + return; + } + await expect.poll(async () => (await resumed()).length, { timeout: 20_000 }).toBe(1); + const [continuation] = await resumed(); + expect(continuation).toMatchObject({ status: "queued", contextSnapshot: { previousRunId: completed.id, + taskRetryRequestId: retry.requestId } }); + expect(await readFreshWorkspaceReadiness(db, continuation)).toMatchObject({ mode: "resume_preserved_workspace" }); + expect((await request(boardRecoveryApp()).post(`/api/issues/${f.issueId}/retry`).send(retryIntent).expect(202)).body.requestId).toBe(retry.requestId); + expect(await resumed()).toHaveLength(1); expect(startTurn).not.toHaveBeenCalled(); + expect(errors).not.toHaveBeenCalled(); + return; + } + expect({ status: completed.status, error: completed.error, runtimeMode: completed.runtimeMode }) + .toEqual({ status: "succeeded", error: null, runtimeMode: "native" }); + expect(inputs).toHaveLength(1); expect(startTurn).toHaveBeenCalledOnce(); + expect(completed.runnerProfileJson?.nativeExecutionInput).toMatchObject({ provider: { kind: "codex", model: "gpt-6.1-sol" } }); + expect(await readFile(path.join(inputs[0].workspace.cwd!, "continued.txt"), "utf8")).toBe("continued from preserved work\n"); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, completed.id))) + .toEqual([expect.objectContaining({ schemaStatus: "accepted" })]); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("done"); + await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId); expect(startTurn).toHaveBeenCalledOnce(); + expect((await run(f.runId)).error).toBe(source.error); + } finally { + try { + await recoveryCoordinator?.stop(); await recoveryWorker?.stop(); + await ownedHeartbeat?.drainActiveRunExecutions(); + } finally { + storage?.mockRestore(); + if (databaseUrl === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = databaseUrl; + if (masterKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = masterKey; + await instanceSettingsService(db).updateExperimental(experimental); + } + } + }, 40_000); + it.each(["shutdown", "lease"] as const)("admits HTTP Retry once after an interrupted Runner preparation recovers its remote workspace (%s)", async interruption => { const f = await interruptedRemotePreparation(interruption); await recoverRetainedWorkspace(f.input);