From 2518ac85abce1316bdf6c3ed85b4b4aa7894c1f0 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 07:17:49 -0600 Subject: [PATCH] fix(recovery): continue verified failed native preparation (#18908) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Task Retry keeps a durable request while the system stops old work and saves its files. > - Native preparation can fail after it allocates a workspace, but before it starts a provider. > - Completed preservation can resolve the action while the original Retry still waits. > - The old classifier rejected every allocated failed preparation. The source check also returned before its canonical Retry fallback. > - This pull request accepts only complete server proof and moves the existing exact-source fallback before that return. > - The same Retry can then start one normal continuation. Current access and unknown execution still hold work. > - Saving files and admitting that continuation must use the same parent-row lock order; the composed head includes the supporting preservation fix. ## Linked Issues or Issue Description **What happened?** Task Retry accepted a failed native preparation and saved its request. The workspace had complete preservation and stop receipts. The request still waited for execution recovery and created no continuation. **Expected behavior** The same deliberate Retry must continue after the system verifies all saved files and source leases. It must preserve current task, actor, repository, Stop and execution checks. **Steps to reproduce** 1. Record a native-adapter claim that fails during unresolved legacy preparation, before provider execution. 2. Finish real workspace preservation and exact lease cleanup. 3. Submit normal HTTP task Retry and reconcile its canonical request. 4. On the base commit, no run is linked to that request. With this change, one continuation is linked to it. **Paperclip version or commit** Reproduced on master `d7283e875772ca8c684dc86e471e6b48d543b7b1`. **Deployment mode** Source checkout with real local PostgreSQL and filesystem archives. Object storage, sandbox transport and the native provider backend are controlled in the final execution test. This is not a live model or hosted object-storage claim. **Prerequisite: #18918** at `4c1d7769e6cfa9536a67c4e3b536cef155f19966`. This head includes that parent through normal history-preserving merges while retaining `master` as its target. The parent owns remote workspace save, prepare and provision lock order, legacy import reservation and publication, their concurrency regressions, and the database lock-order documentation. Merge the parent before this PR; no deployment is implied. Related: Refs #18892 for interrupted preparation. Refs #18882 for the separate connected-chat setup path. Refs #18898 for already-admitted cleanup waits. This change does not depend on #18882 or #18898. ## What Changed - Accept allocated failed preparation only with the server error and durable settlement records, no execution/session/checkpoint/output/result/coordinator evidence, all exact remote source leases stopped, and verified saved workspace proof. - Recognize only the host's closed accounting scope and typed pre-provider unsupported-harness warning. Unknown accounting, provider metrics and arbitrary diagnostics remain holds. - Check the canonical failed-preparation source before an empty resolved-action list can return. Reuse the normal locked continuation checks. - Add HTTP, preservation-worker and native execution coverage, including source failures produced by real heartbeat workspace preparation. Keep one canonical request and one provider turn. Preserve the original failure and real archive files. - Restore storage, environment and experimental settings even if native test setup or owned async cleanup fails. - Cover changed Stop, source, actor, controller, provider evidence, snapshot generation, writer and lease receipts. Keep current unavailable-catalog denial before provider startup. - Share one bounded host-settlement receipt reader between failed-native preparation and existing-directory readiness. Keep each caller's accounting, profile and additional warning restrictions. - Include parent #18918's company/task/run-before-workspace ordering for save, prepare and provision. Keep legacy import publication out of reverse native-owner waits with exact NOWAIT refusal. Retain foreign keys, ownership checks, uploaded work and transaction-aware notifications. - Document the failed-preparation continuation and parent lock-order contracts. ## Verification - Causal base comparison: the normal HTTP/preservation regression fails on the canonical request's missing successor. Both existing interrupted-preparation controls pass; the same cases pass with the classifier/fallback repair. - Actual heartbeat-produced catalog failures reproduce RED before the host-accounting fix, both with and without the real typed unsupported-harness warning. No source failure records are patched in those cases. - The native execution case keeps `paperclip_runner`, native runtime selection and the requested Codex model. A controlled native session reads real restored `saved.txt`, writes `continued.txt`, emits an accepted result and finishes through the real finalizer once. Only the external native backend is controlled. - Current unavailable-catalog admission and suspended authenticated membership remain denials before provider work. Extra provider/session/accounting material, Stop, newer source, writer/generation and lease-receipt mismatches remain held. - Both settlement consumers pass 53 focused real PostgreSQL/filesystem/native-boundary cases on Node24.22.2. Existing-directory warning presentation/provenance and durable-measurement restrictions are preserved. - Parent #18918 proves real PostgreSQL lock cycles through actual save, prepare, provision, reservation and import publication owners. Its native status peer test proves a busy refusal without a deadlock. Immutable task binding, writer/epoch fences, same-journal retry and unexpected-error rollback controls retain uploaded objects and original files. The original spontaneous peer identity is not claimed. - Historical head `6204f565` passed all 54 CI checks, but its local affected run passed only 572/573 with a real PostgreSQL preservation deadlock and its Paperclip review was 3/5. Those results remain historical evidence; this composed head includes the explicit parent repair. - Production-composed head `93a02d4cf74f4d8d2cc6c1b62884bc58280500bf`, Node24.22.2 / pnpm9.15.4: full `pnpm build` and recursive `pnpm -r typecheck` passed. The six affected recovery, continuation, Retry, migration and concurrency files passed 724 tests. Opt-in remote workspace integration passed 65/65 and ownership lifecycle passed 16/16, with no skips. - Final head `65dcd6c6a601448b5a436b7a7f2c84b8cfde761b` adds only the parent's unsupported-host test guard. Its 18 concurrency cases and generated server test-layer typecheck passed on the exact head. Production bytes match the qualified head above. - Exact-head CI passed on `65dcd6c6`: 54 success, 3 skipped, 1 neutral; no failed or pending jobs. Greptile reviewed this exact commit at 5/5 with no actionable issue and zero open threads. - The current-head Paperclip assessment is registered and queued with no run at the latest observation. It has not delivered a current-head score. The earlier 3/5 assessment and its resolved findings remain history. No extra review request or hosted task replay was used to force delivery. ## Risks - The new allocated-source proof is conservative. Extra or missing execution material keeps work held and needs inspection. - This changes admission after completed preservation. It does not certify unknown provider effects or restore repository access. - Administrative import can now return a named busy hold while an execution row is owned. It retains the archive and journal for a later administrative retry. - There is no schema migration or new API. Stop, task ownership, current actor access, budget and approval checks retain their normal owners. ## Model Used OpenAI Codex, GPT-6 family. The exact runtime model ID and context window are not exposed in this agent session. The agent used reasoning, tool calls, code edits, real local PostgreSQL and filesystem tests. A separate Codex agent reviewed the admission and execution boundaries. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/task-recovery.md | 12 + .../src/services/cancelled-native-startup.ts | 52 +- .../services/explicit-native-continuation.ts | 22 +- .../services/failed-preparation-settlement.ts | 38 ++ .../services/fresh-workspace-continuation.ts | 33 +- .../src/services/workspace-recovery.test.ts | 461 +++++++++++++++++- 6 files changed, 572 insertions(+), 46 deletions(-) create mode 100644 server/src/services/failed-preparation-settlement.ts 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);