diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index c4e8328f7d9..33208b6d02a 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -19,6 +19,37 @@ cleanup run. This includes descriptor reads that fail inside runtime selection. preparation failure; another owner or a stale claim prevents cleanup. Settling that claim and recording file retention share a transaction. +Modern `remoteWorkspace` preparation references use the same failure-retention +producer as workspace-sync descriptors. A failed native run can also retain an +expired `awaiting_runner_reattach` coordinator after its task execution lock was +released. The orphan-lease owner may record a file-only preservation obligation +for that historical state only when the exact allocation remains active, +unreleased and ephemeral. It rechecks the task, run, coordinator generation, +prepared workspace writer, complete allocation history, current controller and +cleanup state under short row locks. A reusable allocation, accepted result, +competing writer, manual recovery state or ambiguous cleanup stays held. The +presence of a Stop-only marker remains a hold even when its value is invalid. + +The transaction pins the existing StopAndRetain intent before provider I/O. The +ordinary cleanup owner stops only the recorded allocation and rechecks all its +current owners before dispatch. Normal checkpoint restart must first publish its +task and coordinator claim; that claim prevents preservation intake. After a pin, +the released task execution lock prevents an observed run from resuming itself. +No row lock spans provider or archive I/O. Conflicting native or cleanup owners +produce a nonblocking busy result after transaction rollback. + +Only matching source and generation fingerprints plus the exact stopped-and-retained +receipt let the native environment settlement owner retire that stale coordinator +into manual terminal reconciliation. The existing workspace worker then verifies +and saves the original files. The interrupted provider checkpoint and its unknown +effects remain unknown; this path admits no provider turn, accepts no result and +changes no task status. A changed or malformed modern workspace claim remains a +cleanup hold even when it is ineligible for automatic preservation. It cannot fall +through to destructive orphan cleanup. Invalid workspace authority is refused +without throwing or stopping preservation of the next source in the batch. +Already released or mismatched historical +sources require their owning reconciliation and are not rearmed by this path. + The reconciliation candidate scan selects the complete workspace-sync record and the native status wake identifier from each saved run. It keeps their JSON types and reads full evidence later where finalization or recovery needs it. Saved diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 45aa4153326..29524aeb8f3 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -857,6 +857,23 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { return { pluginId, companyId, agentId, environment, runId, executionWorkspaceId, reusableLease }; } + it.each(["active", "released"] as const)("does not select an ephemeral allocation for warm reuse (%s)", async status => { + const seeded = await seedReusablePluginSandboxLease("paperclip_runner"); + await db.update(environmentLeases).set({ leasePolicy: "ephemeral", status, + releasedAt: status === "released" ? new Date() : null }).where(eq(environmentLeases.id, seeded.reusableLease.id)); + const workerManager = createFreshPluginLeaseWorker(seeded.pluginId); + const runtimeWithPlugin = environmentRuntimeService(db, { pluginWorkerManager: workerManager }); + const acquired = await runtimeWithPlugin.acquireRunLease({ + companyId: seeded.companyId, environment: seeded.environment, agentId: seeded.agentId, + issueId: null, adapterType: "paperclip_runner", heartbeatRunId: seeded.runId, + persistedExecutionWorkspace: { id: seeded.executionWorkspaceId, mode: "shared_workspace" }, + }); + expect(seeded.environment.config.reuseLease).toBe(true); + expect(acquired.lease.providerLeaseId).toBe("fresh-plugin-lease"); + expect(acquired.lease.providerLeaseId).not.toBe(seeded.reusableLease.providerLeaseId); + expect(workerManager.call).toHaveBeenCalledExactlyOnceWith(seeded.pluginId, "environmentAcquireLease", expect.any(Object), expect.any(Number)); + }); + it("gives a bootstrap retry a fresh allocation even when the failed lease was reusable", async () => { const seeded = await seedReusablePluginSandboxLease("paperclip_runner"); await environmentService(db).releaseLease(seeded.reusableLease.id, "expired"); diff --git a/server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts b/server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts new file mode 100644 index 00000000000..c0a24932cdd --- /dev/null +++ b/server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts @@ -0,0 +1,156 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; +import { eq } from "drizzle-orm"; +import { agents, authUsers, companyMemberships, environmentLeases, environments, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { beforeAll, expect, it, vi } from "vitest"; +import { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js"; +import * as storageBackend from "../storage/index.js"; +import { getStorageService } from "../storage/index.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { pinFailedObservedWorkspace } from "../services/native-runtime/native-observed-workspace-preservation.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { environmentRuntimeService, classifyEnvironmentCapabilities } from "../services/environment-runtime.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { buildNativeExecutionInput } from "../services/native-runtime/native-execution-input.js"; +import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; +import { registerNativeRunnerCheckpoint } from "../services/native-runtime/native-runner-checkpoint.js"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; + +describeEmbeddedPostgres("heartbeat native remote preparation failure retention", () => { + const fixture = nativeRestartFixture(); + let runtimeRoot: string, companyId: string, agentId: string; + beforeAll(() => { ({ runtimeRoot, companyId, agentId } = fixture.current); }); + + it.each(["modern", "legacy", "provider_history"] as const)("preserves the allocated workspace when restart preparation fails (%s)", async mode => { + const f = await fixture.seedRun(`PREPARATION-FAILURE-${mode}`); + const db = f.db, workspaceId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID(); + const settings = instanceSettingsService(db), previous = await settings.getExperimental(); + const priorDatabase = process.env.DATABASE_URL, priorKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + const localStorage = getStorageService(); + let storage: { mockRestore(): void } | undefined; + let heartbeat: ReturnType | undefined; + try { + storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" }); + process.env.DATABASE_URL = fixture.current.temporary.connectionString; + process.env.PAPERCLIP_SECRETS_MASTER_KEY = "71".repeat(32); + const userId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Original operator", email: `${userId}@example.invalid`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" }); + const remoteCwd = path.join(runtimeRoot, workspaceId), providerLeaseId = `sandbox://fake/${randomUUID()}`; + await mkdir(remoteCwd, { recursive: true }); + await writeFile(path.join(remoteCwd, "saved.txt"), "unfinished original work\n"); + await db.insert(environments).values({ id: environmentId, name: `Bound original allocation ${mode}`, driver: "sandbox", + config: { provider: "fake", image: "fixture:test" } }); + await db.update(agents).set({ defaultEnvironmentId: environmentId }).where(eq(agents.id, agentId)); + const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId, + remoteCwd, epoch: 1, state: "prepared", snapshotId: null }; + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: f.issueId, + mode: "isolated_workspace", strategyType: "project_primary", providerType: "cloud_sandbox", name: "Original files", + cwd: `/workspaces/${workspaceId}`, metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1", + epoch: 1, snapshotId: null, source: { kind: "empty" }, location: { environmentId, leaseId, providerLeaseId, remoteCwd }, + writerRunId: f.runId, dirty: true } } }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, f.issueId)); + await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId: f.issueId, heartbeatRunId: f.runId, + executionWorkspaceId: workspaceId, environmentId, provider: "fake", providerLeaseId, status: "active", + leasePolicy: "ephemeral", metadata: { driver: "sandbox", provider: "fake", image: "fixture:test", remoteCwd, agentId } }); + const execution = buildNativeExecutionInput({ companyId, agentId, runId: f.runId, + issue: { id: f.issueId, identifier: "NRR-MODERN-PREPARATION-FAILURE", title: "Preserve original work", description: null, workMode: "standard" }, + taskPrompt: "Continue the original task.", provider: "codex", + workspace: { id: workspaceId, cwd: remoteCwd, repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: f.native.normalizedSessionId, runtimeContext: nativeRuntimeContextFixture(), + completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", + contract: { revision: "1", objective: "Preserve original work", criteria: [{ id: "objective", requirement: "Retain files" }] } } }); + const identity = { companyId, issueId: f.issueId, runId: f.runId, agentId, sessionId: f.native.normalizedSessionId }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.preparedProfile, remoteWorkspace: reference, + nativeExecutionInput: execution, sessionCheckpoint: { identity, providerSessionId: "original-provider" }, + ...(mode === "legacy" ? { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId, leaseId, providerLeaseId, remoteCwd, resourceDisposition: "stop_and_retain" } } : {}) }, + responsibleUserId: userId, contextSnapshot: { issueId: f.issueId, executionWorkspaceId: workspaceId } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ controllerBootId: fixture.current.successor.bootId, + controllerGeneration: 1, attempt: 1, leaseOwner: "previous-controller", leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const authority = new DurablePrpControlPlane({ stateDirectory: path.join(runtimeRoot, f.runId, "control-plane"), + identity: fixture.transportOptions(f, runtimeRoot).prpIdentity, + expectedRunnerVersion: "retention-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + await registerNativeRunnerCheckpoint({ db, binding: { companyId, issueId: f.issueId, runId: f.runId }, + bootId: fixture.current.successor.bootId, authority }); + await authority.stop(); + if (mode === "provider_history") { + const event = { schema: "paperclip.prp.event.v1", sourceInstanceId: f.native.runnerInstanceId, + sourceEventId: randomUUID(), sourceSeq: 1, schemaVersion: 1, sourceKind: "runner", runId: f.runId, + normalizedSessionId: f.native.normalizedSessionId, turnId: f.native.turnId, eventType: "tool.execution.started", + priority: 0, emittedAt: new Date().toISOString(), payload: { toolName: "fixture.external_effect", toolCallId: randomUUID() } }; + await appendHeartbeatRunEvent(db, { companyId, agentId, runId: f.runId, eventType: event.eventType, + stream: "system", level: "info", message: "Original external action has an unknown outcome", payload: { prpEvent: event }, + nativeSource: { sourceEventId: event.sourceEventId, sourceInstanceId: event.sourceInstanceId, + sourceSeq: event.sourceSeq, protocolSchemaVersion: 1, canonicalPayload: event } }); + } + await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: agentId, cause: "native_session_interrupted", + fingerprint: `native-interruption:${f.runId}`, evidence: { runId: f.runId }, + nextAction: "Inspect the interrupted source and preserve its work.", maxAttempts: 3, wakePolicy: null, + supersedeOnIdentityChange: true }); + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const preparationFailure = new Error("Sandbox failed to become ready within the timeout period"); + const resumed = vi.fn(async () => { + const [claimedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [claimedOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const [claimedTask] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(claimedRun.status).toBe("running"); + expect(claimedTask.executionRunId).toBe(f.runId); + expect(claimedOwner.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now()); + expect(claimedOwner.controllerGeneration).toBe(2); + expect(await pinFailedObservedWorkspace(db, companyId, f.runId)).toBe(false); + throw preparationFailure; + }); + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", + acquireRunLease: async () => { throw new Error("Unexpected new allocation"); }, + resumeRunLease: async () => { throw new Error("Reattachment must not resume the allocation"); }, + releaseRunLease: async () => { throw new Error("Unexpected destructive cleanup"); }, + retryPendingSandboxTeardown: async () => { throw new Error("Stop must remain separately verified"); }, + realizeWorkspace: async () => { throw new Error("Unexpected workspace realization"); }, + execute: resumed, + syncIn: resumed, + syncOut: resumed, + supportsSync: () => true, + resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: ["environmentSyncIn", "environmentSyncOut", "environmentExecute"], declared: null, narrowing: null }), + }] }); + await settings.updateExperimental({ enableNativeRunner: true }); + heartbeat = heartbeatService(db, { environmentRuntime: runtime, + nativeSessionBackendFactory: () => { throw new Error("No provider backend may be created"); } }); + await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + await heartbeat.drainActiveRunExecutions(); + expect(resumed).toHaveBeenCalledOnce(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run.error).toBe(preparationFailure.message); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)); + expect({ status: run.status, phase: coordinator.phase, taskExecution: task.executionRunId, dirty: workspace.metadata?.remoteWorkspace }) + .toMatchObject({ status: "failed", phase: "terminal_failure", taskExecution: null, dirty: { writerRunId: f.runId, dirty: true } }); + expect(run.resultJson).toMatchObject({ workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { leaseIds: [leaseId] } }); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + expect(lease).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure", + metadata: { sandboxStopAndRetain: { runId: f.runId, leaseId } } }); + expect(await readFile(path.join(remoteCwd, "saved.txt"), "utf8")).toBe("unfinished original work\n"); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))) + .toEqual(expect.arrayContaining([expect.objectContaining({ evidence: expect.objectContaining({ workspaceRestoreRecovery: expect.any(Object) }) })])); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1); + if (mode === "provider_history") expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId))) + .toEqual(expect.arrayContaining([expect.objectContaining({ eventType: "tool.execution.started", sourceSeq: 1 })])); + } finally { + try { await heartbeat?.drainActiveRunExecutions(); } + finally { + storage?.mockRestore(); + if (priorDatabase === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = priorDatabase; + if (priorKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = priorKey; + await settings.updateExperimental(previous); + } + } + }, 60_000); +}); diff --git a/server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts b/server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts new file mode 100644 index 00000000000..6144f316570 --- /dev/null +++ b/server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts @@ -0,0 +1,389 @@ +import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { cp, mkdir, readFile, realpath, writeFile } from "node:fs/promises"; +import path from "node:path"; +import { promisify } from "node:util"; +import { and, eq, sql } from "drizzle-orm"; +import { agents, authUsers, companyMemberships, environmentLeases, environments, executionWorkspaceSnapshots, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { beforeAll, expect, it, vi } from "vitest"; +import * as storageBackend from "../storage/index.js"; +import { getStorageService } from "../storage/index.js"; +import { buildNativeExecutionInput } from "../services/native-runtime/native-execution-input.js"; +import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; +import { getSandboxProvider } from "../services/sandbox-provider-runtime.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerEnvironmentCleanupWork } from "../services/environment-cleanup-work.js"; +import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../services/environment-runtime.js"; +import { environmentService } from "../services/environments.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "../services/remote-workspace-snapshot-storage.js"; +import { restoreRemoteWorkspaceArchive } from "../services/remote-workspace-snapshot-archive.js"; +import { registerWorkspaceRecoveryWork } from "../services/workspace-recovery-work.js"; +import { hasVerifiedWorkspaceRepair } from "../services/workspace-recovery-receipt.js"; +import { pinFailedObservedWorkspace, settleStoppedObservedWorkspace, ObservedWorkspacePreservationBusyError } from "../services/native-runtime/native-observed-workspace-preservation.js"; +import { claimNativeRestartRecoveries } from "../services/native-runtime/native-restart-recovery.js"; +import { hasRemoteTerminationReceipt } from "../services/remote-execution-termination.js"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; + +const execute = promisify(execFile); +function barrier() { let resolve!: () => void; const promise = new Promise(ready => { resolve = ready; }); return { promise, resolve }; } + +describeEmbeddedPostgres("stopped observed native workspace preservation", () => { + const fixture = nativeRestartFixture(); + let runtimeRoot: string, companyId: string, agentId: string; + beforeAll(() => { ({ runtimeRoot, companyId, agentId } = fixture.current); }); + + async function seedSource(label: string) { + const f = await fixture.seedRun(label), db = f.db; + const workspaceId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox://fake/${randomUUID()}`; + await mkdir(path.join(runtimeRoot, workspaceId)); + const remoteCwd = await realpath(path.join(runtimeRoot, workspaceId)); + await writeFile(path.join(remoteCwd, "saved.txt"), "unfinished work with an unknown provider effect\n"); + await db.insert(environments).values({ id: environmentId, name: `Historical allocation ${environmentId}`, driver: "sandbox", + config: { provider: "fake", image: "fixture:test" } }); + const userId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Source operator", email: `${userId}@example.invalid`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" }); + await db.update(agents).set({ defaultEnvironmentId: environmentId, adapterConfig: { provider: "codex", model: "gpt-6.1-sol" } }).where(eq(agents.id, agentId)); + const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId, + remoteCwd, epoch: 1, state: "prepared", snapshotId: null }; + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: f.issueId, + mode: "isolated_workspace", strategyType: "project_primary", providerType: "cloud_sandbox", name: "Unsaved original files", + cwd: `/workspaces/${workspaceId}`, metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1", + epoch: 1, snapshotId: null, source: { kind: "empty" }, writerRunId: f.runId, dirty: true, + location: { environmentId, leaseId, providerLeaseId, remoteCwd } } } }); + await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null, executionWorkspaceId: workspaceId }) + .where(eq(issues.id, f.issueId)); + await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId: f.issueId, heartbeatRunId: f.runId, + executionWorkspaceId: workspaceId, environmentId, provider: "fake", providerLeaseId, status: "active", leasePolicy: "ephemeral", + metadata: { driver: "sandbox", provider: "fake", image: "fixture:test", remoteCwd, agentId }, updatedAt: new Date(Date.now() - 6 * 60_000) }); + const checkpoint = { identity: { companyId, issueId: f.issueId, runId: f.runId, agentId, sessionId: f.native.normalizedSessionId }, + providerSessionId: "historical-session", activeTurnId: "unfinished-turn", pendingRuntimeRequests: [{ requestId: "unknown-tool" }] }; + const execution = buildNativeExecutionInput({ companyId, agentId, runId: f.runId, + issue: { id: f.issueId, identifier: "NRR-HISTORICAL-OBSERVED", title: "Preserve original work", description: null, workMode: "standard" }, + taskPrompt: "Continue the original task.", provider: "codex", model: "gpt-6.1-sol", + workspace: { id: workspaceId, cwd: remoteCwd, repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: f.native.normalizedSessionId, runtimeContext: nativeRuntimeContextFixture(), + completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", + contract: { revision: "1", objective: "Preserve original work", criteria: [{ id: "objective", requirement: "Retain files" }] } } }); + const profile = { ...f.preparedProfile, remoteWorkspace: reference, sessionCheckpoint: checkpoint, + nativeExecutionInput: execution, nativeResourceDisposition: "stop_and_retain" }; + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(Date.now() - 6 * 60_000), + nativePhase: "observed", errorCode: "setup_failed", error: "Sandbox failed to become ready within the timeout period", + responsibleUserId: userId, runnerProfileJson: profile, contextSnapshot: { issueId: f.issueId, executionWorkspaceId: workspaceId } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "observed", attempt: 1, controllerGeneration: 3, recoveryState: "awaiting_runner_reattach", + controllerBootId: randomUUID(), controllerPid: 2_147_483_640, controllerProcessStartedAt: new Date(0), + leaseOwner: "old-observed-controller", leaseExpiresAt: new Date(Date.now() - 5 * 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", + ownerType: "board", returnOwnerAgentId: agentId, cause: "native_session_interrupted", fingerprint: `interrupted:${f.runId}`, + evidence: { runId: f.runId }, nextAction: "Reconcile the unknown provider outcome.", maxAttempts: 3, wakePolicy: null, + supersedeOnIdentityChange: true }); + return { ...f, workspaceId, environmentId, leaseId, providerLeaseId, remoteCwd, checkpoint, profile }; + } + + it("stops the exact stale allocation, preserves a cold-readable archive and never creates another native session", async () => { + const f = await seedSource("HISTORICAL-OBSERVED"), db = f.db; + const settings = instanceSettingsService(db), previous = await settings.getExperimental(); + const priorDatabase = process.env.DATABASE_URL, priorKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + const localStorage = getStorageService(); + let storage: { mockRestore(): void } | undefined; + let heartbeat: ReturnType | undefined; + let coordinator: ReturnType | undefined; + let workspaceWorker: ReturnType | undefined; + let providerStop: { mockRestore(): void } | undefined; + let providerDestroy: { mockRestore(): void } | undefined; + let cleanupWorker: ReturnType | undefined; + const stopAllowed = barrier(); + try { + storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" }); + process.env.DATABASE_URL = fixture.current.temporary.connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "72".repeat(32); + await settings.updateExperimental({ enableNativeRunner: true }); + const { workspaceId, environmentId, leaseId, providerLeaseId, remoteCwd, checkpoint, profile } = f; + const acquire = vi.fn(async () => { throw new Error("A new allocation is forbidden"); }); + const backend = vi.fn(() => { throw new Error("A native session is forbidden"); }); + const copy = async (input: { operations: Array<{ files: Array<{ sourcePath: string; targetPath: string }> }> }) => { + for (const operation of input.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 { operations: [], filesTransferred: input.operations.reduce((sum, operation) => sum + operation.files.length, 0), bytesTransferred: 0 }; + }; + const provider = getSandboxProvider("fake")!; + const originalStop = provider.stopLease!.bind(provider); + const stopped = vi.spyOn(provider, "stopLease").mockImplementation(async input => { + await stopAllowed.promise; return originalStop(input); + }); + providerStop = stopped; + const destroyed = vi.spyOn(provider, "destroyLease"); providerDestroy = destroyed; + const fileRuntime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", acquireRunLease: acquire, + resumeRunLease: async ({ lease }) => ({ providerLeaseId: lease.providerLeaseId!, metadata: { remoteCwd: lease.metadata?.remoteCwd } }), + releaseRunLease: async () => { throw new Error("Destructive release is forbidden"); }, + retryPendingSandboxTeardown: async () => { throw new Error("Use the actual lifecycle owner"); }, realizeWorkspace: async () => { throw new Error("New realization is forbidden"); }, + execute: async input => { + const result = await execute(input.command === "node" ? process.execPath : input.command, input.args ?? [], + { cwd: input.cwd, env: { ...process.env, ...input.env }, maxBuffer: 4 * 1024 * 1024 }); + return { stdout: result.stdout, stderr: result.stderr, exitCode: 0, timedOut: false, signal: null }; + }, + syncIn: copy, syncOut: copy, supportsSync: () => true, + resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: ["environmentSyncIn", "environmentSyncOut", "environmentExecute"] }), + }] }); + const runtime = { ...fileRuntime, retryPendingSandboxTeardown: environmentRuntimeService(db).retryPendingSandboxTeardown }; + heartbeat = heartbeatService(db, { environmentRuntime: runtime, nativeSessionBackendFactory: backend }); + const errors = vi.fn(); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors }); + workspaceWorker = registerWorkspaceRecoveryWork(db, coordinator, runtime, heartbeat.resumeRemoteStopComments); + const environmentWorker = heartbeat.registerNativeEnvironmentSettlementWork(coordinator, () => true); + const orphanWorker = heartbeat.registerResidualRecoveryWork(coordinator, () => true); + await Promise.all([workspaceWorker.ready, environmentWorker.ready, orphanWorker.ready]); + await expect.poll(async () => (await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)))[0].status, + { timeout: 10_000 }).toBe("pending_cleanup"); + const [beforeRestart] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))); + await coordinator.stop(); await workspaceWorker.stop(); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors }); + workspaceWorker = registerWorkspaceRecoveryWork(db, coordinator, runtime, heartbeat.resumeRemoteStopComments); + const restartedEnvironment = heartbeat.registerNativeEnvironmentSettlementWork(coordinator, () => true); + const restartedOrphan = heartbeat.registerResidualRecoveryWork(coordinator, () => true); + await Promise.all([workspaceWorker.ready, restartedEnvironment.ready, restartedOrphan.ready]); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, beforeRestart.id)))[0].evidence.observedWorkspacePreservation) + .toEqual(beforeRestart.evidence.observedWorkspacePreservation); + cleanupWorker = registerEnvironmentCleanupWork(db, coordinator, { canRun: () => true, + sweepPending: backoffMs => heartbeat!.sweepPendingCleanupLeases({ backoffMs }), minimumSpacingMs: 1 }); + await expect.poll(() => stopped.mock.calls.length, { timeout: 10_000 }).toBe(1); + const [pending] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + const [ownerBeforeStop] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(ownerBeforeStop).toMatchObject({ phase: "observed", leaseOwner: "old-observed-controller", controllerGeneration: 3, attempt: 1 }); + expect(pending).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure", + metadata: { sandboxStopAndRetain: { runId: f.runId, leaseId, providerLeaseId } } }); + expect(hasRemoteTerminationReceipt(pending)).toBe(false); + expect(await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, workspaceId))).toHaveLength(0); + expect(backend).not.toHaveBeenCalled(); expect(acquire).not.toHaveBeenCalled(); + stopAllowed.resolve(); + await cleanupWorker.ready; + const [waiting] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))); + const wait = waiting.evidence.systemRecovery as { state?: string; nextAttemptAt?: number; failure?: { code?: string } } | undefined; + if (wait?.nextAttemptAt && wait.nextAttemptAt > Date.now()) { + expect(wait).toMatchObject({ state: "awaiting_evidence", failure: { code: "agent_stop_unconfirmed" } }); + expect(wait.nextAttemptAt - Date.now()).toBeLessThanOrEqual(36_000); + } + await expect.poll(async () => hasVerifiedWorkspaceRepair(db, + (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]), { timeout: 40_000 }).toBe(true); + const [savedWorkspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)); + const [savedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [savedOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(savedOwner).toMatchObject({ phase: "terminal_failure", leaseOwner: null, attempt: 1, controllerGeneration: 3 }); + expect(savedRun).toMatchObject({ status: "failed", errorCode: "setup_failed", runnerProfileJson: { sessionCheckpoint: checkpoint, nativeExecutionInput: profile.nativeExecutionInput } }); + const snapshotId = (savedWorkspace.metadata?.remoteWorkspace as { snapshotId: string }).snapshotId; + const snapshot = await getWorkspaceSnapshot(db, { companyId, workspaceId, snapshotId }); + expect(snapshot).not.toBeNull(); + const cold = path.join(runtimeRoot, randomUUID()), archivePath = path.join(runtimeRoot, `${randomUUID()}.pcws`); + await loadWorkspaceSnapshotArchive({ storage: storageBackend.getStorageService(), reference: snapshot!, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir: cold, expectedManifestDigest: snapshot!.manifestDigest }); + expect(await readFile(path.join(cold, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n"); + expect(await readFile(path.join(remoteCwd, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toHaveLength(0); + const [preservation] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); + expect((await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.fingerprint, `interrupted:${f.runId}`))))[0]).toMatchObject({ status: "active", cause: "native_session_interrupted" }); + expect(preservation.evidence).toMatchObject({ automaticRecovery: { policy: "preserve_without_replay_v1", replay: "blocked", actionOutcome: "unknown" } }); + expect(backend).not.toHaveBeenCalled(); expect(acquire).not.toHaveBeenCalled(); + expect(destroyed).not.toHaveBeenCalled(); + expect(errors).not.toHaveBeenCalled(); + } finally { + stopAllowed.resolve(); cleanupWorker?.stop(); + try { await coordinator?.stop(); await workspaceWorker?.stop(); await heartbeat?.drainActiveRunExecutions(); } + finally { + providerStop?.mockRestore(); providerDestroy?.mockRestore(); + storage?.mockRestore(); + if (priorDatabase === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = priorDatabase; + if (priorKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = priorKey; + await settings.updateExperimental(previous); + } + } + }, 60_000); + + type Source = Awaited>; + async function readSource(f: Source) { + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [lease] = await f.db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + return { run, lease, owner }; + } + async function changeLease(f: Source, patch: Partial, metadata?: Record) { + const { lease } = await readSource(f); + await f.db.update(environmentLeases).set({ ...patch, ...(metadata ? { metadata: { ...lease.metadata, ...metadata } } : {}) }) + .where(eq(environmentLeases.id, f.leaseId)); + } + async function acceptedResult(f: Source) { + const { run } = await readSource(f); + await f.db.insert(nativeRunResults).values({ companyId, issueId: f.issueId, runId: f.runId, + completionContractId: run.completionContractId!, schemaStatus: "accepted", serverFingerprint: randomUUID(), + canonicalSha256: "a".repeat(64), resultJson: { outcome: "accepted" } }); + } + async function otherAllocationOwner(f: Source) { + const other = await fixture.seedRun(`OTHER-${randomUUID()}`, f.db); + await f.db.insert(environmentLeases).values({ companyId, issueId: other.issueId, heartbeatRunId: other.runId, + environmentId: f.environmentId, provider: "fake", providerLeaseId: f.providerLeaseId, + status: "active", leasePolicy: "ephemeral", metadata: { driver: "sandbox", provider: "fake", image: "fixture:test" } }); + return other; + } + async function sourceHeld(f: Source) { + const { run, owner } = await readSource(f); + expect(owner).toMatchObject({ phase: "observed", leaseOwner: "old-observed-controller", controllerGeneration: 3, attempt: 1 }); + expect(run).toMatchObject({ status: "failed", runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + expect(await readFile(path.join(f.remoteCwd, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n"); + expect(await f.db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, f.workspaceId))).toHaveLength(0); + } + + it.each(["reusable", "inflight_cleanup", "unknown_cleanup", "null_stop_intent", "null_export_intent", "unknown_recovery", "manual_failure", "live_controller", + "copyback_owner", "accepted_unbound", "other_allocation_owner", "wrong_epoch", "invalid_reference", "stopped_mismatch", "operator_stop"] as const)( + "keeps unsafe observed source held without Stop or Delete (%s)", async mode => { + const f = await seedSource(`HELD-${mode}`); + if (mode === "reusable") await changeLease(f, { leasePolicy: "reuse_by_environment" }); + if (mode === "inflight_cleanup") await changeLease(f, {}, { pendingCleanupInFlight: true }); + if (mode === "unknown_cleanup") await changeLease(f, {}, { pendingCleanupInFlight: "unknown" }); + if (mode === "null_stop_intent") await changeLease(f, {}, { sandboxStopAndRetain: null }); + if (mode === "null_export_intent") await changeLease(f, {}, { nativeWorkspaceExportResume: null }); + if (mode === "unknown_recovery") await f.db.update(nativeRunFinalizations).set({ recoveryState: "awaiting_evidence" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "manual_failure") await f.db.update(nativeRunFinalizations).set({ failureCode: "native_restart_recovery_blocked" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "live_controller") await f.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "copyback_owner") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, + nativeWorkspaceFinalizationOwner: { token: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "invalid_reference") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, remoteWorkspace: { schema: "unverified" } } }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "accepted_unbound") await acceptedResult(f); + if (mode === "other_allocation_owner") await otherAllocationOwner(f); + if (mode === "wrong_epoch") await f.db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '2')` }).where(eq(executionWorkspaces.id, f.workspaceId)); + if (mode === "stopped_mismatch") await changeLease(f, { status: "released", cleanupStatus: "success", releasedAt: new Date() }, { + remoteExecutionTermination: { schema: "paperclip.remote-execution-termination.v1", provider: "fake", providerLeaseId: `sandbox://fake/${randomUUID()}`, state: "stopped", confirmedAt: new Date().toISOString() }, + }); + if (mode === "operator_stop") await f.db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); + const provider = getSandboxProvider("fake")!, stopped = vi.spyOn(provider, "stopLease"), destroyed = vi.spyOn(provider, "destroyLease"); + const backend = vi.fn(() => { throw new Error("Unexpected native execution"); }); + const heartbeat = heartbeatService(f.db, { nativeSessionBackendFactory: backend }); + try { + await heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0, companyId, leaseId: f.leaseId }); + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + const current = await readSource(f); + expect(current.run.resultJson?.workspaceRestoreRecovery).toBeUndefined(); + if (mode === "null_stop_intent") expect(current.lease.metadata?.sandboxStopAndRetain).toBeNull(); + if (mode === "null_export_intent") expect(current.lease.metadata?.nativeWorkspaceExportResume).toBeNull(); + await sourceHeld(f); + expect(stopped.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0); + expect(destroyed.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0); expect(backend).not.toHaveBeenCalled(); + } finally { await heartbeat.drainActiveRunExecutions(); stopped.mockRestore(); destroyed.mockRestore(); } + }); + + it("holds malformed workspace authority without throwing or recording a new pin", async () => { + const f = await seedSource("MALFORMED-WORKSPACE"); + await f.db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '0')` }) + .where(eq(executionWorkspaces.id, f.workspaceId)); + await expect(pinFailedObservedWorkspace(f.db, companyId, f.runId)).resolves.toBe(false); + expect((await readSource(f)).run.resultJson?.workspaceRestoreRecovery).toBeUndefined(); + await sourceHeld(f); + }); + + it("keeps a poisoned workspace held and continues the actual orphan page to its next valid source", async () => { + const bad = await seedSource("POISONED-PAGE-FIRST"), good = await seedSource("VALID-PAGE-NEXT"), db = bad.db; + await db.update(environmentLeases).set({ updatedAt: new Date(Date.now() - 7 * 60_000) }).where(eq(environmentLeases.id, bad.leaseId)); + await db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '0')` }) + .where(eq(executionWorkspaces.id, bad.workspaceId)); + const provider = getSandboxProvider("fake")!, stopped = vi.spyOn(provider, "stopLease"), destroyed = vi.spyOn(provider, "destroyLease"); + const backend = vi.fn(() => { throw new Error("Unexpected native execution"); }); + const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: backend }); + try { + await expect(heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0, companyId })).resolves.toMatchObject({ recovered: 1 }); + expect((await readSource(bad)).run.resultJson?.workspaceRestoreRecovery).toBeUndefined(); + expect((await readSource(bad)).lease.status).toBe("active"); + expect((await readSource(good)).run.resultJson?.workspaceRestoreRecovery).toMatchObject({ leaseIds: [good.leaseId] }); + expect((await readSource(good)).lease).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" }); + await sourceHeld(bad); + expect(stopped.mock.calls.filter(([input]) => input.providerLeaseId === bad.providerLeaseId)).toHaveLength(0); + expect(destroyed.mock.calls.filter(([input]) => input.providerLeaseId === bad.providerLeaseId)).toHaveLength(0); + expect(backend).not.toHaveBeenCalled(); + } finally { await heartbeat.drainActiveRunExecutions(); stopped.mockRestore(); destroyed.mockRestore(); } + }); + + it.each(["before_pin", "after_pin"] as const)("serializes the real ephemeral checkpoint claim %s", async order => { + const f = await seedSource(`CLAIM-${order}`); + if (order === "before_pin") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + else expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true); + const dispositions = await claimNativeRestartRecoveries({ db: f.db, restartKind: "hard", runIds: [f.runId], controller: fixture.current.successor }); + if (order === "before_pin") { + expect(dispositions).toMatchObject([{ kind: "reattach_remote_runner", runId: f.runId }]); + expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(false); + const { run, lease, owner } = await readSource(f); + expect(run.status).toBe("running"); expect(owner.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now()); + expect(owner.controllerGeneration).toBe(4); expect(lease.status).toBe("active"); + expect(lease.metadata?.sandboxStopAndRetain).toBeUndefined(); + } else { + expect(dispositions).toMatchObject([{ kind: "blocked", runId: f.runId, reason: "issue_execution_lock_changed" }]); + expect((await readSource(f)).owner).toMatchObject({ phase: "observed", leaseOwner: "old-observed-controller", controllerGeneration: 3 }); + } + }); + + it("rechecks a new physical allocation owner after pin before the actual Stop dispatch", async () => { + const f = await seedSource("ARRIVING-OWNER"); + expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true); + const other = await otherAllocationOwner(f); + const provider = getSandboxProvider("fake")!, stopped = vi.spyOn(provider, "stopLease"), destroyed = vi.spyOn(provider, "destroyLease"); + const heartbeat = heartbeatService(f.db); + try { + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(stopped.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0); + expect(destroyed.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0); + expect(await settleStoppedObservedWorkspace(f.db, companyId, f.runId)).toBe(false); + expect((await readSource(f)).lease).toMatchObject({ status: "pending_cleanup", metadata: { pendingCleanupInFlight: false } }); + expect((await f.db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, other.runId)))[0].status).toBe("active"); + await sourceHeld(f); + } finally { await heartbeat.drainActiveRunExecutions(); stopped.mockRestore(); destroyed.mockRestore(); } + }); + + it.each(["generation", "checkpoint", "workspace_authority", "accepted_result", "other_owner", "lost_stop_ack"] as const)( + "does not settle observed ownership from stale or missing Stop evidence (%s)", async mode => { + const f = await seedSource(`STOP-FENCE-${mode}`); + expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true); + const original = (await f.db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))))[0]; + expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true); + expect((await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, original.id)))[0]).toEqual(original); + const provider = getSandboxProvider("fake")!; + const stop = mode === "lost_stop_ack" ? vi.spyOn(provider, "stopLease").mockRejectedValue(new Error("Controlled lost acknowledgement")) : null; + const heartbeat = heartbeatService(f.db); + try { + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(hasRemoteTerminationReceipt((await readSource(f)).lease)).toBe(mode !== "lost_stop_ack"); + if (mode === "generation") await f.db.update(nativeRunFinalizations).set({ controllerGeneration: 4 }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "checkpoint") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, sessionCheckpoint: { ...f.checkpoint, activeTurnId: "changed" } } }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "workspace_authority") await f.db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '0')` }).where(eq(executionWorkspaces.id, f.workspaceId)); + if (mode === "accepted_result") await acceptedResult(f); + if (mode === "other_owner") await otherAllocationOwner(f); + expect(await settleStoppedObservedWorkspace(f.db, companyId, f.runId)).toBe(false); + expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(mode === "lost_stop_ack"); + const current = (await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, original.id)))[0]; + expect(current.evidence.observedWorkspacePreservation).toEqual(original.evidence.observedWorkspacePreservation); + expect((await readSource(f)).owner.phase).toBe("observed"); + expect(await readFile(path.join(f.remoteCwd, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n"); + } finally { await heartbeat.drainActiveRunExecutions(); stop?.mockRestore(); } + }); + + it("backs off a real cleanup producer conflict only after rolling the preservation transaction back", async () => { + const f = await seedSource("CLEANUP-ROW-BUSY"), entered = barrier(), release = barrier(); + const peer = f.db.transaction(async tx => { + await environmentService(tx as unknown as typeof f.db).releaseLease(f.leaseId, "released", { cleanupStatus: "success" }); + entered.resolve(); await release.promise; + }); + try { + await entered.promise; + await expect(pinFailedObservedWorkspace(f.db, companyId, f.runId)).rejects.toBeInstanceOf(ObservedWorkspacePreservationBusyError); + expect((await readSource(f)).run.resultJson?.workspaceRestoreRecovery).toBeUndefined(); + } finally { release.resolve(); await peer; } + expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(false); + await sourceHeld(f); + }); + +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 8bde92adcda..07fa969603e 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1993,7 +1993,8 @@ export function heartbeatService( const preserveNativeWorkspaceOnFailure = async () => { const failureRun = await getRun(run.id); if (failureRun?.runtimeMode === "native" - && parseObject(failureRun.runnerProfileJson?.nativeWorkspaceSync).state === "prepared") { + && (parseObject(failureRun.runnerProfileJson?.nativeWorkspaceSync).state === "prepared" + || parseObject(failureRun.runnerProfileJson?.remoteWorkspace).state === "prepared")) { if (!nativeRunHasNoWorkspaceAccess(failureRun)) providerResourceDispositionForRun = "stop_and_retain"; try { await preserveNativeFailedWorkspace(db, failureRun, { diff --git a/server/src/services/native-runtime/native-environment-settlement-work.ts b/server/src/services/native-runtime/native-environment-settlement-work.ts index 43c925b63af..3e4cf2b6fc6 100644 --- a/server/src/services/native-runtime/native-environment-settlement-work.ts +++ b/server/src/services/native-runtime/native-environment-settlement-work.ts @@ -1,4 +1,4 @@ -import { and, asc, eq, exists, gt, inArray, or, sql } from "drizzle-orm"; +import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import type { ProviderResourceDisposition } from "../environment-runtime.js"; import { nextRunLeaseReleaseAt, pendingRunLeaseReleaseCondition } from "../environment-lease-release-state.js"; @@ -7,6 +7,9 @@ import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import type { NativeEnvironmentSettlementScope } from "../native-environment-settlement-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; +import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; +import { hasConfirmedSandboxStopAndRetain } from "../sandbox-stop-and-retain.js"; +import { settleStoppedObservedWorkspace } from "./native-observed-workspace-preservation.js"; type Run = Pick; export function nativeEnvironmentSettlementDisposition(run: Run, succeeded: boolean): ProviderResourceDisposition { @@ -38,17 +41,23 @@ export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settleme }).from(nativeRunFinalizations) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) - .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + .leftJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), eq(nativeRunResults.runId, nativeRunFinalizations.runId), eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), eq(nativeRunResults.schemaStatus, "accepted"))) .where(and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]), - or(eq(nativeRunFinalizations.phase, "committed"), - and(eq(nativeRunFinalizations.phase, "terminal_failure"), - sql`${nativeRunFinalizations.failureDetail} ? 'workspaceFinalizeAttempt'`)), - exists(db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( - eq(environmentLeases.companyId, heartbeatRuns.companyId), eq(environmentLeases.heartbeatRunId, heartbeatRuns.id), - pendingRunLeaseReleaseCondition(true)))), + or(and(isNotNull(nativeRunResults.id), + or(eq(nativeRunFinalizations.phase, "committed"), + and(eq(nativeRunFinalizations.phase, "terminal_failure"), + sql`${nativeRunFinalizations.failureDetail} ? 'workspaceFinalizeAttempt'`)), + exists(db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, heartbeatRuns.companyId), eq(environmentLeases.heartbeatRunId, heartbeatRuns.id), + pendingRunLeaseReleaseCondition(true))))), + and(eq(heartbeatRuns.status, "failed"), eq(nativeRunFinalizations.phase, "observed"), + isNull(nativeRunFinalizations.resultId), isNotNull(nativeRunFinalizations.leaseOwner), + lte(nativeRunFinalizations.leaseExpiresAt, new Date()), + sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,schema}' = 'paperclip.remote-workspace-run.v1'`, + sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,state}' = 'prepared'`)), scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, @@ -56,6 +65,12 @@ export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settleme )).orderBy(asc(heartbeatRuns.id)).limit(100); } async function entry(row: Awaited>[number]): Promise { + if (row.phase === "observed") { + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, row.companyId), eq(environmentLeases.heartbeatRunId, row.id))); + return leases.length === 1 && leases.every(lease => hasRemoteTerminationReceipt(lease) && hasConfirmedSandboxStopAndRetain(lease)) + ? { id: row.id, companyId: row.companyId, issueId: row.issueId, dueAt: 0 } : null; + } const dueAt = await nextRunLeaseReleaseAt(db, row.companyId, row.id, nativeEnvironmentSettlementDisposition(row, row.phase === "committed")); if (dueAt === null) { retryFloors.delete(row.id); return null; } @@ -86,6 +101,10 @@ export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settleme const scope = { companyId: current.companyId, issueId: current.issueId, runId: current.id }; const [row] = await candidates(scope); if (!row) { retryFloors.delete(current.id); return null; } + if (row.phase === "observed") { + await settleStoppedObservedWorkspace(db, current.companyId, current.id); + return null; + } const pending = await entry(row); if (!pending || pending.dueAt > Date.now()) return pending; if (signal?.aborted) throw new Error("Native environment settlement stopped"); diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index 8683cac4dc4..0900ecc6aa0 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -15,6 +15,7 @@ import { hasAcknowledgedNativeRunStopReceipt } from "../acknowledged-native-stop import { logger } from "../../middleware/logger.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { preserveRemoteWorkspacePreparationFailure } from "./native-workspace-finalization-state.js"; +import { pinFailedObservedWorkspace } from "./native-observed-workspace-preservation.js"; function workspaceRecoveryReference(profile: Record | null | undefined) { @@ -143,6 +144,7 @@ export async function preserveOrphanedNativeWorkspace(db: Db, lease: Pick(db: Db, operation: (tx: Db) => Promise): Promise { + try { + return await db.transaction(async transaction => { + const tx = transaction as unknown as Db; + await tx.execute(sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`); + return operation(tx); + }); + } catch (error) { + const failure = error as { code?: unknown; cause?: { code?: unknown } }; + if (failure?.code === "55P03" || failure?.cause?.code === "55P03") throw new ObservedWorkspacePreservationBusyError({ cause: error }); + throw error; + } +} + +function sourceSha256(run: Run) { + const profile = run.runnerProfileJson ?? {}; + return nativeSha256({ runId: run.id, companyId: run.companyId, agentId: run.agentId, issueId: run.nativeIssueId, + nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId, + processPid: run.processPid, processGroupId: run.processGroupId, processStartedAt: run.processStartedAt?.toISOString() ?? null, + remoteWorkspace: profile.remoteWorkspace, nativeExecutionInput: profile.nativeExecutionInput, + sessionCheckpoint: profile.sessionCheckpoint, nativeControllerRecovery: profile.nativeControllerRecovery, + errorCode: run.errorCode, error: run.error, finishedAt: run.finishedAt?.toISOString() ?? null }); +} + +function coordinatorSha256(coordinator: Coordinator) { + return nativeSha256({ controllerBootId: coordinator.controllerBootId, controllerPid: coordinator.controllerPid, + controllerProcessStartedAt: coordinator.controllerProcessStartedAt?.toISOString() ?? null, + controllerGeneration: coordinator.controllerGeneration, attempt: coordinator.attempt, + leaseOwner: coordinator.leaseOwner, leaseExpiresAt: coordinator.leaseExpiresAt?.toISOString() ?? null, + phase: coordinator.phase, resultId: coordinator.resultId, recoveryState: coordinator.recoveryState, + recoveryRequestId: coordinator.recoveryRequestId, recoveryHistory: coordinator.recoveryHistory, + failureCode: coordinator.failureCode, failureDetail: coordinator.failureDetail, + nextAttemptAt: coordinator.nextAttemptAt?.toISOString() ?? null, controlDeadlineAt: coordinator.controlDeadlineAt?.toISOString() ?? null }); +} + +function readFence(value: unknown): Fence | null { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const fence = value as Fence; + return fence.schema === SCHEMA && ["stop_pending", "stopped"].includes(fence.state) + && [fence.runId, fence.leaseOwner, fence.leaseId, fence.provider, fence.providerLeaseId, fence.environmentId, fence.workspaceId] + .every(item => typeof item === "string" && Boolean(item)) + && Number.isFinite(Date.parse(fence.leaseExpiresAt)) && /^[a-f0-9]{64}$/.test(fence.sourceSha256) && /^[a-f0-9]{64}$/.test(fence.coordinatorSha256) + && Number.isSafeInteger(fence.controllerGeneration) && fence.controllerGeneration >= 0 + && Number.isSafeInteger(fence.providerAttempt) && fence.providerAttempt >= 0 ? fence : null; +} + +function matchesObserved(run: Run, coordinator: Coordinator | undefined, now: Date) { + return run.runtimeMode === "native" && run.status === "failed" && run.nativePhase === "observed" && Boolean(run.nativeIssueId) + && run.errorCode !== NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE && run.errorCode !== NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT + && !adapterExecutionControls.has(run.id) + && run.resultJson?.nativeCancellation == null && run.resultJson?.startupCancellation == null + && run.resultJson?.executionCancellation == null && run.resultJson?.cancellation == null + && !(run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > now) + && run.runnerProfileJson?.nativeWorkspaceSync == null && run.runnerProfileJson?.nativeWorkspaceFinalizationOwner == null + && coordinator?.companyId === run.companyId && coordinator.issueId === run.nativeIssueId + && coordinator.runId === run.id && coordinator.phase === "observed" && coordinator.resultId === null + && coordinator.assessmentId === null && coordinator.decisionId === null && coordinator.failureCode === null + && coordinator.recoveryState === "awaiting_runner_reattach" && coordinator.nextAttemptAt === null && coordinator.controlDeadlineAt === null + && Boolean(coordinator.leaseOwner) && Boolean(coordinator.leaseExpiresAt && coordinator.leaseExpiresAt <= now); +} + +async function lockSource(db: Db, companyId: string, runId: string) { + await db.select({ id: companies.id }).from(companies).where(eq(companies.id, companyId)).for("key share", { noWait: true }); + const [identity] = await db.select({ issueId: heartbeatRuns.nativeIssueId }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); + if (!identity?.issueId) return null; + const [task] = await db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, identity.issueId))).for("update", { noWait: true }); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.issueId, identity.issueId))).for("update", { noWait: true }); + const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))).for("update", { noWait: true }); + if (!task || !run || task.status !== "blocked" || task.assigneeAgentId !== run.agentId || task.assigneeUserId + || task.executionRunId || task.checkoutRunId || !matchesObserved(run, coordinator, new Date())) return null; + const [accepted] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, companyId), eq(nativeRunResults.runId, runId), eq(nativeRunResults.schemaStatus, "accepted"))) + .limit(1).for("update", { noWait: true }); + if (accepted) return null; + const parsed = remoteWorkspaceRunReferenceSchema.safeParse(run.runnerProfileJson?.remoteWorkspace); + if (!parsed.success || parsed.data.state !== "prepared" || task.executionWorkspaceId !== parsed.data.workspaceId) return null; + const reference = parsed.data; + const [workspace] = await db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, companyId), eq(executionWorkspaces.id, reference.workspaceId))).for("update", { noWait: true }); + const parsedAuthority = remoteWorkspaceAuthoritySchema.safeParse(workspace?.metadata?.remoteWorkspace); + const authority = parsedAuthority.success ? parsedAuthority.data : null; + if (!workspace || workspace.sourceIssueId !== task.id || !authority?.location || authority.epoch !== reference.epoch + || authority.snapshotId !== reference.snapshotId || !authority.dirty || authority.writerRunId !== run.id + || authority.location.leaseId !== reference.leaseId || authority.location.providerLeaseId !== reference.providerLeaseId + || authority.location.remoteCwd !== reference.remoteCwd) return null; + const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), + inArray(workspaceOperations.status, ["pending", "running"]))).limit(1); + if (operation) return null; + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id))).orderBy(asc(environmentLeases.id)); + if (leases.length !== 1) return null; + const selected = leases[0]; + if (!selected.provider || !selected.providerLeaseId) return null; + const history = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.provider, selected.provider), eq(environmentLeases.providerLeaseId, selected.providerLeaseId))) + .orderBy(asc(environmentLeases.id)).for("update", { noWait: true }); + if (history.length !== 1 || history[0].id !== selected.id) return null; + const lease = history[0]; + if (lease.id !== reference.leaseId || lease.issueId !== task.id || lease.executionWorkspaceId !== workspace.id + || lease.environmentId !== authority.location.environmentId || !lease.environmentId + || !lease.provider || lease.provider === "local" || lease.providerLeaseId !== reference.providerLeaseId + || lease.metadata?.remoteCwd !== reference.remoteCwd || lease.metadata?.agentId !== run.agentId + || (lease.metadata?.pendingCleanupInFlight != null && lease.metadata.pendingCleanupInFlight !== false)) return null; + const [writer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), + sql`(${heartbeatRuns.nativeIssueId} = ${task.id}::uuid or ${heartbeatRuns.issueId} = ${task.id}::uuid + or ${heartbeatRuns.contextSnapshot}->>'issueId' = ${task.id} + or ${heartbeatRuns.contextSnapshot}->>'executionWorkspaceId' = ${workspace.id})`)).limit(1); + if (writer) return null; + return { run, task, coordinator: coordinator!, reference, workspace, lease }; +} + +function fenceMatches(fence: Fence, source: NonNullable>>, evidence: Record | undefined) { + const { run, coordinator, lease } = source; + const recorded = run.resultJson?.workspaceRestoreRecovery as { schema?: unknown; leaseIds?: unknown } | undefined; + const action = evidence?.workspaceRestoreRecovery as { schema?: unknown; leaseIds?: unknown } | undefined; + const retained = lease.metadata?.workspaceRestoreRecovery as { schema?: unknown; runId?: unknown } | undefined; + const exact = (value: typeof recorded) => value?.schema === "paperclip.workspace-restore-recovery.v1" + && Array.isArray(value.leaseIds) && value.leaseIds.length === 1 && value.leaseIds[0] === lease.id; + return lease.leasePolicy === "retain_on_failure" && exact(recorded) && exact(action) && retained?.schema === "paperclip.workspace-restore-recovery.v1" && retained.runId === run.id + && fence.runId === run.id && fence.sourceSha256 === sourceSha256(run) && fence.coordinatorSha256 === coordinatorSha256(coordinator) + && fence.leaseOwner === coordinator.leaseOwner && fence.leaseExpiresAt === coordinator.leaseExpiresAt?.toISOString() + && fence.controllerGeneration === coordinator.controllerGeneration && fence.providerAttempt === coordinator.attempt + && fence.leaseId === lease.id && fence.provider === lease.provider && fence.providerLeaseId === lease.providerLeaseId + && fence.environmentId === lease.environmentId && fence.workspaceId === source.workspace.id; +} + +export async function pinFailedObservedWorkspace(db: Db, companyId: string, runId: string): Promise { + return preservationTransaction(db, async tx => { + const source = await lockSource(tx, companyId, runId); + if (!source) return false; + const { run, task, coordinator, workspace, lease } = source; + const [existing] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId), + eq(issueRecoveryActions.sourceIssueId, task.id), eq(issueRecoveryActions.fingerprint, `native-workspace:${run.id}`))).for("update", { noWait: true }); + if (hasRequiredWorkspaceRecovery(run.resultJson)) { + const fence = readFence(existing?.evidence[FIELD]); + return Boolean(fence?.state === "stop_pending" && fenceMatches(fence, source, existing?.evidence)); + } + const active = lease.leasePolicy === "ephemeral" && lease.status === "active" && lease.releasedAt === null && lease.cleanupStatus === null + && !hasStopOnlyCleanup(lease) && lease.metadata?.sandboxStopAndRetainReceipt == null + && lease.metadata?.remoteExecutionTermination == null && lease.metadata?.pendingCleanupAttemptId == null + && (lease.metadata?.pendingCleanupLeaseExpiresAtMs == null || lease.metadata.pendingCleanupLeaseExpiresAtMs === 0) + && lease.metadata?.workspaceRecoveryCleanup == null; + if (!active) return false; + await notifyNativeEnvironmentSettlement(tx, { companyId, runId, issueId: task.id }); + const retained = await recordRequiredWorkspaceRestoreFailure(tx, run, { workspaceRestoreFailure: "restore_failed" }, lease); + if (!retained || !hasRequiredWorkspaceRecovery(retained.resultJson)) throw new Error("observed_workspace_retention_missing"); + const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId), + eq(issueRecoveryActions.sourceIssueId, task.id), eq(issueRecoveryActions.fingerprint, `native-workspace:${run.id}`))).for("update", { noWait: true }); + if (!action) throw new Error("observed_workspace_retention_action_missing"); + const fence: Fence = { schema: SCHEMA, state: "stop_pending", runId, leaseOwner: coordinator.leaseOwner!, + leaseExpiresAt: coordinator.leaseExpiresAt!.toISOString(), controllerGeneration: coordinator.controllerGeneration, + providerAttempt: coordinator.attempt, sourceSha256: sourceSha256(run), coordinatorSha256: coordinatorSha256(coordinator), leaseId: lease.id, + provider: lease.provider!, providerLeaseId: lease.providerLeaseId!, environmentId: lease.environmentId!, workspaceId: workspace.id }; + await tx.update(issueRecoveryActions).set({ evidence: { ...action.evidence, [FIELD]: fence }, updatedAt: new Date() }) + .where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.id, action.id))); + await logActivity(tx, { companyId, actorType: "system", actorId: "native-workspace-recovery", + action: "issue.observed_workspace_preservation_pinned", entityType: "issue", entityId: task.id, runId, + details: { recoveryActionId: action.id, sourceRunId: runId, controllerGeneration: fence.controllerGeneration } }); + return true; + }); +} + +export async function settleStoppedObservedWorkspace(db: Db, companyId: string, runId: string): Promise { + return preservationTransaction(db, async tx => { + const source = await lockSource(tx, companyId, runId); + if (!source) return false; + const { run, task, coordinator, lease } = source; + const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId), + eq(issueRecoveryActions.sourceIssueId, task.id), eq(issueRecoveryActions.fingerprint, `native-workspace:${run.id}`))).for("update", { noWait: true }); + const fence = readFence(action?.evidence[FIELD]); + if (!fence || fence.state !== "stop_pending" || !hasRequiredWorkspaceRecovery(run.resultJson) + || !fenceMatches(fence, source, action?.evidence) + || !hasRemoteTerminationReceipt(lease) || !hasConfirmedSandboxStopAndRetain(lease)) return false; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionFinalization); + const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, + leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null, recoveryState: "blocked", + failureCode: "native_continuation_requires_reconciliation", updatedAt: new Date() }).where(and( + eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.issueId, task.id), eq(nativeRunFinalizations.phase, "observed"), isNull(nativeRunFinalizations.resultId), + eq(nativeRunFinalizations.leaseOwner, fence.leaseOwner), eq(nativeRunFinalizations.controllerGeneration, fence.controllerGeneration), + eq(nativeRunFinalizations.attempt, fence.providerAttempt))).returning({ id: nativeRunFinalizations.runId }); + if (!settled) throw new Error("observed_workspace_preservation_fence_changed"); + await tx.update(heartbeatRuns).set({ nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date(), updatedAt: new Date() }) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); + await tx.update(issueRecoveryActions).set({ evidence: { ...action!.evidence, [FIELD]: { ...fence, state: "stopped" } }, updatedAt: new Date() }) + .where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.id, action!.id))); + await logActivity(tx, { companyId, actorType: "system", actorId: "native-workspace-recovery", + action: "issue.observed_workspace_stop_verified", entityType: "issue", entityId: task.id, runId, + details: { recoveryActionId: action!.id, sourceRunId: runId, controllerGeneration: fence.controllerGeneration } }); + return true; + }); +} diff --git a/server/src/services/native-runtime/native-workspace-finalization-state.ts b/server/src/services/native-runtime/native-workspace-finalization-state.ts index 00b783fe21c..250c070d968 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-state.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-state.ts @@ -46,13 +46,20 @@ export async function preserveRemoteWorkspacePreparationFailure(db: Db, lease: W export async function nativeWorkspaceExportPendingForLease(db: Db, lease: WorkspaceSourceLease): Promise { if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return false; - if (await preparedRemoteWorkspaceSource(db, lease)) return true; const input = sql`${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput'`; const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, driverKind: heartbeatRuns.driverKind, reference: sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`, + observedRemoteWorkspace: sql`${heartbeatRuns.status} = 'failed' + and ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace' is not null + and ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace' <> 'null'::jsonb + and exists (select 1 from ${nativeRunFinalizations} + where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} + and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.issueId} = ${heartbeatRuns.nativeIssueId} + and ${nativeRunFinalizations.phase} = 'observed' and ${nativeRunFinalizations.resultId} is null)`, executionScope: sql>`jsonb_build_object( 'schema', ${input}->'schema', 'binding', ${input}->'binding', 'workspace', ${input}->'workspace', 'provider', jsonb_build_object('kind', ${input}#>'{provider,kind}'))`, @@ -60,6 +67,8 @@ export async function nativeWorkspaceExportPendingForLease(db: Db, lease: Worksp eq(heartbeatRuns.id, lease.heartbeatRunId), eq(heartbeatRuns.companyId, lease.companyId), eq(heartbeatRuns.runtimeMode, "native"), )).limit(1); + if (run?.observedRemoteWorkspace) return true; + if (await preparedRemoteWorkspaceSource(db, lease)) return true; if (!run || nativeRunHasNoWorkspaceAccess({ ...run, runnerProfileJson: { nativeExecutionInput: run.executionScope } })) return false; const reference = readNativeWorkspaceSyncReference(run.reference); return reference?.state === "prepared" && reference.leaseId === lease.id