diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index 4f3a1a7c67e..f4a64d19ad8 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -154,12 +154,22 @@ The checkpoint owner authenticates/decrypts that header and delegates version an inspection to Runner; the identity must match the run, session and runner. Ordinary sandbox runs bind their workspace identity; plugin tasks bind their active dispatched environment lease and need no workspace-sync descriptor. +Accepted terminal results can also qualify through their existing finalization +owner. This requires the exact accepted result and contract, a remote workspace, +and its retained provider lease or successful workspace barrier. Committed work +also needs its applied decision and assessment. Unknown physical owners and +finalizer leases beyond the five-minute budget remain blockers. The observation +does not request stop proof or take a lease. Copyback still requires the old +worker's signed stop receipt and an atomic owner claim. Committed projections +use the existing accepted-result and decision receipts. Legacy/local executions, missing identities/checkpoints, foreign/expired owners, in-flight admission, and an incomplete census prevent a positive report. It returns aggregate counts only. The request does not stop admission, detach a runner, or authorize new work. Ordinary drain reads do not perform this SQL. -The explicit probe uses a read-only transaction, a two-second statement timeout, -and at most 1,001 run rows. Header reads are capped at 256 KiB each and 8 MiB total; +The explicit probe uses read-only transactions, a two-second statement timeout, +and at most 1,001 run rows. Post-result inspections share a two-second scheduling +budget; after it expires, unchecked candidates remain blocking. An in-flight +inspection retains its SQL statement timeout. Header reads are capped at 256 KiB each and 8 MiB total; oversized, malformed, legacy-format or local evidence falls back to the ordinary drain. This bounded header probe does not validate every journal record or prove a live sandbox. This is an observation; the existing shutdown and startup owners still perform the actual handoff and enforce Stop and fencing. diff --git a/server/src/services/heartbeat/upgrade-readiness-post-result.test.ts b/server/src/services/heartbeat/upgrade-readiness-post-result.test.ts new file mode 100644 index 00000000000..ddd456c07a1 --- /dev/null +++ b/server/src/services/heartbeat/upgrade-readiness-post-result.test.ts @@ -0,0 +1,262 @@ +import { generateKeyPairSync, randomUUID, sign } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq, inArray, sql } from "drizzle-orm"; +import { agents, companies, completionContracts, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, environments, executionWorkspaces, + heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, workAssessments, workspaceOperations } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; +import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js"; +import { nativeRunnerCheckpointWriter } from "../native-runtime/native-runner-checkpoint.js"; +import { readUpgradeReadiness } from "./upgrade-readiness.js"; +import { currentWorkspaceWorker } from "../native-runtime/native-workspace-platform-proof.js"; +import { withNativeWorkspaceFinalizationOwnership } from "../native-runtime/native-workspace-finalization-ownership.js"; +import { resumeNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js"; +import { projectCommittedRun } from "../native-runtime/native-run-projection.js"; + +describe("post-result upgrade readiness", () => { + let database: Awaited>; + let db: ReturnType; + const runs: string[] = []; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("post-result-upgrade-"); + db = createDb(database.connectionString); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "a".repeat(64)); + for (const [key, value] of Object.entries({ RAILWAY_PROJECT_ID: "project", RAILWAY_ENVIRONMENT_ID: "environment", + RAILWAY_SERVICE_ID: "service", RAILWAY_DEPLOYMENT_ID: "deployment", PAPERCLIP_CLOUD_STACK_ID: "stack", + PAPERCLIP_CLOUD_API_ORIGIN: "https://cloud.example", PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "fixture-token" })) vi.stubEnv(key, value); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterEach(async () => { if (runs.length) await db.update(heartbeatRuns).set({ status: "succeeded" }).where(inArray(heartbeatRuns.id, runs)); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); vi.unstubAllEnvs(); }); + + async function seed() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const contractId = randomUUID(), workspaceId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID(); + const resultId = randomUUID(), runnerInstanceId = randomUUID(), nativeSessionId = randomUUID(); + const controller = await currentNativeControllerIdentity(); + const leaseOwner = `execution:${randomUUID()}`; + const remote = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId: "sandbox", + remoteCwd: "/workspace", epoch: 1, state: "prepared", snapshotId: null }; + const authority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: null, source: { kind: "empty" }, + location: { environmentId, leaseId, providerLeaseId: "sandbox", remoteCwd: "/workspace" }, writerRunId: runId, dirty: true }; + await db.insert(companies).values({ id: companyId, name: "Upgrade", issuePrefix: `P${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Runner", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Accepted work", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "fixture", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: {}, + canonicalSha256: "a".repeat(64), createdByActorType: "system", createdByActorId: "test" }); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, mode: "isolated_workspace", + strategyType: "local_directory", name: "Remote", cwd: "/workspace", providerType: "cloud_sandbox", metadata: { remoteWorkspace: authority } }); + await db.insert(environments).values({ id: environmentId, name: `Sandbox ${environmentId}`, driver: "sandbox" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, status: "running", + runtimeMode: "native", nativePhase: "observed", runnerInstanceId, nativeSessionId, completionContractId: contractId, + processPid: process.pid, processStartedAt: controller.processStartedAt, + runnerProfileJson: { remoteWorkspace: remote, nativeExecutionInput: { binding: { executionWorkspaceId: workspaceId } } }, + contextSnapshot: { executionWorkspaceId: workspaceId } }); + runs.push(runId); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId, executionWorkspaceId: workspaceId }).where(eq(issues.id, issueId)); + await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, executionWorkspaceId: workspaceId, + issueId, heartbeatRunId: runId, provider: "daytona", providerLeaseId: "sandbox", leasePolicy: "retain_on_failure" }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", leaseOwner, + leaseExpiresAt: new Date(Date.now() + 60_000), controllerBootId: controller.bootId, controllerPid: controller.pid, + controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: 1 }); + const root = await mkdtemp(path.join(os.tmpdir(), "post-result-header-")); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity: { runId, runnerInstanceId, + normalizedSessionId: nativeSessionId, environmentLeaseId: workspaceId, turnId: "turn", itemId: "item" }, + expectedRunnerVersion: "0.3.0", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + const writer = nativeRunnerCheckpointWriter(db, { companyId, issueId, runId, leaseOwner, bootId: controller.bootId, + authorityRunId: runId, generation: 1 }); + core.setCheckpointPersistence(writer); + try { core.issueBootstrapTicket(); await core.flushCheckpoint(); } + finally { await core.stop(); await rm(root, { recursive: true, force: true }); } + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, + serverFingerprint: resultId, schemaStatus: "accepted", resultJson: { terminal: { runTerminalState: "succeeded" } }, canonicalSha256: "b".repeat(64) }); + await db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing", resultId }).where(eq(nativeRunFinalizations.runId, runId)); + return { companyId, issueId, runId, agentId, contractId, resultId, workspaceId, environmentId, leaseId, remote, authority, controller, + probe: () => readUpgradeReadiness(db, { activeRunIds: new Set([runId]), counters: () => ({ activeRuns: 1, pendingWakes: 0, draining: false }) }) }; + } + + it("certifies accepted remote work without changing its lease, result or task admission", async () => { + const f = await seed(); + const before = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await f.probe()).toMatchObject({ ready: true, recoverableRuns: 1, blockingRuns: 0, draining: false }); + expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before); + }); + + it.each(["result_accepted", "workspace_finalizing", "retryable_failure"])("retains the real recovery owner for %s", async phase => { + const f = await seed(); + await db.update(nativeRunFinalizations).set({ phase, leaseOwner: null, leaseExpiresAt: null, + ...(phase === "retryable_failure" ? { failureCode: "native_workspace_sync_out_failed", nextAttemptAt: new Date(Date.now() + 60_000) } : {}), + }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await f.probe()).toMatchObject({ ready: true, recoverableRuns: 1 }); + }); + + it("requires the accepted terminal result, current task, remote lease and exact physical owner", async () => { + const f = await seed(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const profile = run.runnerProfileJson!; + const owner = { token: randomUUID(), controllerBootId: f.controller.bootId, hostname: os.hostname(), + pid: f.controller.pid, processStartedAt: f.controller.processStartedAt.toISOString(), worker: currentWorkspaceWorker() }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, f.runId)); + expect((await f.probe()).ready).toBe(true); + for (const physical of [null, { ...owner, token: "" }, { ...owner, worker: undefined }, { ...owner, controllerBootId: randomUUID() }, + { ...owner, hostname: "another-host" }, { ...owner, processStartedAt: new Date(0).toISOString() }, + { ...owner, worker: { ...owner.worker, deploymentId: "another-deployment" } }, { ...owner, worker: { ...owner.worker, extra: true } }]) { + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeWorkspaceFinalizationOwner: physical ?? {} } }).where(eq(heartbeatRuns.id, f.runId)); + expect((await f.probe()).ready).toBe(false); + } + await db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + for (const patch of [{ controllerGeneration: 0 }, { controllerBootId: "another-boot" }, { phase: "unknown" }, { phase: "terminal_failure" }]) { + await db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, f.runId)); + expect((await f.probe()).ready).toBe(false); + await db.update(nativeRunFinalizations).set({ controllerGeneration: 1, controllerBootId: f.controller.bootId, phase: "workspace_finalizing" }).where(eq(nativeRunFinalizations.runId, f.runId)); + } + await expect(db.update(heartbeatRuns).set({ completionContractId: null }).where(eq(heartbeatRuns.id, f.runId))) + .rejects.toMatchObject({ cause: { code: "23503" } }); + expect((await f.probe()).ready).toBe(true); + for (const terminal of [null, {}, { runTerminalState: "active" }, { runTerminalState: "unknown" }]) { + await db.update(nativeRunResults).set({ resultJson: { terminal } }).where(eq(nativeRunResults.id, f.resultId)); + expect((await f.probe()).ready).toBe(false); + } + await db.update(nativeRunResults).set({ resultJson: { terminal: { runTerminalState: "succeeded" } } }).where(eq(nativeRunResults.id, f.resultId)); + await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, f.resultId)); + expect((await f.probe()).ready).toBe(false); + await db.update(nativeRunResults).set({ schemaStatus: "accepted" }).where(eq(nativeRunResults.id, f.resultId)); + const foreignCheckout = randomUUID(); + await db.insert(heartbeatRuns).values({ id: foreignCheckout, companyId: f.companyId, agentId: f.agentId, status: "succeeded" }); + for (const patch of [{ executionRunId: null }, { checkoutRunId: foreignCheckout }, { status: "cancelled" }, { assigneeAgentId: null }]) { + await db.update(issues).set(patch).where(eq(issues.id, f.issueId)); + expect((await f.probe()).ready).toBe(false); + await db.update(issues).set({ executionRunId: f.runId, checkoutRunId: f.runId, status: "in_progress", assigneeAgentId: f.agentId }).where(eq(issues.id, f.issueId)); + } + for (const patch of [{ cleanupStatus: "pending" }, { status: "expired" }, { status: "released" }, { status: "unknown" }, { expiresAt: new Date(0) }, + { providerLeaseId: "another-sandbox" }, { heartbeatRunId: null }]) { + await db.update(environmentLeases).set(patch).where(eq(environmentLeases.id, f.leaseId)); + expect((await f.probe()).ready).toBe(false); + await db.update(environmentLeases).set({ status: "active", cleanupStatus: null, expiresAt: null, + providerLeaseId: "sandbox", heartbeatRunId: f.runId }).where(eq(environmentLeases.id, f.leaseId)); + } + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeWorkspaceSync: { state: "prepared" } } }).where(eq(heartbeatRuns.id, f.runId)); + expect((await f.probe()).ready).toBe(false); + await db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ leaseOwner: `native-finalizer:${randomUUID()}`, leaseExpiresAt: new Date(Date.now() + 600_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect((await f.probe()).ready).toBe(false); + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect((await f.probe()).ready).toBe(true); + vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", ""); + expect((await f.probe()).ready).toBe(false); + vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "fixture-token"); + }); + + it("allows independent admitted work while observing both saved results", async () => { + const first = await seed(), second = await seed(); + expect(await readUpgradeReadiness(db, { activeRunIds: new Set([first.runId, second.runId]), + counters: () => ({ activeRuns: 2, pendingWakes: 0, draining: false }) })).toMatchObject({ ready: true, recoverableRuns: 2 }); + }); + + it("bounds saved-result inspection across several locked candidates", async () => { + const fixtures = [await seed(), await seed(), await seed()]; + const probeDb = createDb(database.connectionString, { maxConnections: 1 }); + const lock = await db.$client.reserve(); + try { + await lock.unsafe("begin"); + await lock.unsafe("lock table workspace_operations in access exclusive mode"); + const started = performance.now(); + const report = readUpgradeReadiness(probeDb, { activeRunIds: new Set(fixtures.map(f => f.runId)), + counters: () => ({ activeRuns: 3, pendingWakes: 0, draining: false }) }); + expect(await db.execute(sql`select 1 as value`)).toMatchObject([{ value: 1 }]); + expect(await report).toMatchObject({ ready: false, activeRuns: 3, recoverableRuns: 0, blockingRuns: 3 }); + expect(performance.now() - started).toBeLessThan(5_000); + expect(await probeDb.execute(sql`show statement_timeout`)).toMatchObject([{ statement_timeout: "0" }]); + } finally { + await lock.unsafe("rollback"); lock.release(); await probeDb.$client.end(); + } + expect(await readUpgradeReadiness(db, { activeRunIds: new Set(fixtures.map(f => f.runId)), + counters: () => ({ activeRuns: 3, pendingWakes: 0, draining: false }) })) + .toMatchObject({ ready: true, recoverableRuns: 3, blockingRuns: 0 }); + }, 15_000); + + it("requires the applied decision and saved workspace before certifying committed projection", async () => { + const f = await seed(), assessmentId = randomUUID(), decisionId = randomUUID(); + await db.insert(workAssessments).values({ id: assessmentId, companyId: f.companyId, issueId: f.issueId, runId: f.runId, + contractId: f.contractId, resultId: f.resultId, triggerKind: "native_result", triggerActorCompanyId: f.companyId, + priorIssueStatus: "in_progress", priorStatusVersion: 0, policyVersion: "fixture", assessmentJson: {}, inputDigest: assessmentId }); + await db.insert(statusDecisions).values({ id: decisionId, companyId: f.companyId, issueId: f.issueId, runId: f.runId, + assessmentId, decisionVersion: 1, policyVersion: "fixture", fromStatus: "in_progress", toStatus: "done", + reasonCode: "completion_contract_satisfied", decisionJson: {}, decisionDigest: decisionId, applicationState: "applied", appliedAt: new Date() }); + await db.update(nativeRunFinalizations).set({ phase: "committed", assessmentId, decisionId, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(issues).set({ status: "done", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + expect((await f.probe()).ready).toBe(false); + const [barrier] = await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, executionWorkspaceId: f.workspaceId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date() }).returning(); + expect((await f.probe()).ready).toBe(true); + for (const patch of [{ applicationState: "proposed" }, { applicationState: "superseded" }]) { + await db.update(statusDecisions).set(patch).where(eq(statusDecisions.id, decisionId)); + expect((await f.probe()).ready).toBe(false); + } + await db.update(statusDecisions).set({ applicationState: "applied" }).where(eq(statusDecisions.id, decisionId)); + await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.id, barrier.id)); + expect((await f.probe()).ready).toBe(false); + await db.update(workspaceOperations).set({ status: "succeeded" }).where(eq(workspaceOperations.id, barrier.id)); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await projectCommittedRun({ db, run, coordinator }); + await projectCommittedRun({ db, run, coordinator }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toMatchObject({ status: "succeeded", nativePhase: "committed" }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toHaveLength(1); + }); + + it("waits for a real signed stopped-worker receipt, then resumes workspace ownership once", async () => { + const f = await seed(); + const worker = currentWorkspaceWorker()!; + const owner = { token: randomUUID(), controllerBootId: f.controller.bootId, hostname: os.hostname(), + pid: f.controller.pid, processStartedAt: f.controller.processStartedAt.toISOString(), worker }; + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, f.runId)); + const { privateKey, publicKey } = generateKeyPairSync("ed25519"); + vi.stubEnv("PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS", JSON.stringify({ keys: [{ ...publicKey.export({ format: "jwk" }), kid: "fixture", alg: "EdDSA", use: "sig" }] })); + let stopped = false; + const fetch = vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => { + expect(JSON.parse(String(init?.body))).toEqual({ runId: f.runId, ownerToken: owner.token, worker }); + if (!stopped) return new Response(null, { status: 409 }); + const header = Buffer.from(JSON.stringify({ alg: "EdDSA", typ: "paperclip-workspace-worker-stop+jwt", kid: "fixture" })).toString("base64url"); + const iat = Math.floor(Date.now() / 1000); + const payload = Buffer.from(JSON.stringify({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", + sub: "stack", runId: f.runId, ownerToken: owner.token, worker, stopped: true, iat, exp: iat + 60 })).toString("base64url"); + return Response.json({ receipt: `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`), privateKey).toString("base64url")}` }); + }); + try { + expect((await f.probe()).ready).toBe(true); + expect(fetch).not.toHaveBeenCalled(); + vi.stubEnv("RAILWAY_DEPLOYMENT_ID", "replacement"); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, + nativeWorkspaceFinalizationOwner: { ...owner, hostname: "previous-container", controllerBootId: "previous-boot" } } }).where(eq(heartbeatRuns.id, f.runId)); + const scope = { db, companyId: f.companyId, runId: f.runId }; + const copyback = vi.fn(); + expect(await withNativeWorkspaceFinalizationOwnership(scope, copyback)).toEqual({ acquired: false, reason: "platform_stop_pending" }); + expect(copyback).not.toHaveBeenCalled(); + stopped = true; + let enter!: () => void, release!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const joined = new Promise(resolve => { release = resolve; }); + const first = withNativeWorkspaceFinalizationOwnership(scope, async ownership => { + await ownership.assertHeld(); copyback(); enter(); await joined; + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId, + executionWorkspaceId: f.workspaceId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date() }); + }); + try { + await entered; + expect(await resumeNativeWorkspaceFinalization({ db, runId: f.runId })).toBeNull(); + } finally { release(); await first; } + expect(await resumeNativeWorkspaceFinalization({ db, runId: f.runId })).toMatchObject({ status: "succeeded" }); + expect(copyback).toHaveBeenCalledOnce(); + const [recovered] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(recovered.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeUndefined(); + expect(recovered.runnerProfileJson?.nativeWorkspaceOwnerStopProof).toMatchObject({ runId: f.runId, ownerToken: owner.token, stopped: true, worker }); + } finally { fetch.mockRestore(); vi.stubEnv("RAILWAY_DEPLOYMENT_ID", "deployment"); } + }); +}); diff --git a/server/src/services/heartbeat/upgrade-readiness.ts b/server/src/services/heartbeat/upgrade-readiness.ts index 99da469b2ec..e8a2d5663f3 100644 --- a/server/src/services/heartbeat/upgrade-readiness.ts +++ b/server/src/services/heartbeat/upgrade-readiness.ts @@ -7,8 +7,10 @@ import { readNativeWorkspaceSyncReference } from "../native-runtime/native-works import { REMOTE_RUNNER_RECONNECT_GRACE_MS } from "../native-runtime/native-runner-recovery-contract.js"; import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js"; import { readNativeControllerRecoveryInventory } from "../native-runtime/native-controller-recovery-inventory.js"; +import { readNativeFinalizationUpgradeReadiness } from "../native-runtime/native-finalization-upgrade-readiness.js"; const MAX_UPGRADE_RUNS = 1_000; +const POST_RESULT_INSPECTION_BUDGET_MS = 2_000; type Counters = { activeRuns: number; pendingWakes: number; draining: boolean }; type Candidate = { id: string; recoverable: boolean }; @@ -20,17 +22,19 @@ export async function readUpgradeReadiness(db: Db, input: { activeRunIds: ReadonlySet; counters: () => Counters; }) { + const inspectionDeadline = performance.now() + POST_RESULT_INSPECTION_BUDGET_MS; const before = input.counters(); const beforeIds = [...input.activeRunIds]; const owner = await currentNativeControllerIdentity(); const rows = await db.transaction(async tx => { await tx.execute(sql`set local statement_timeout = '2s'`); - return tx.execute | null }>(sql` + return tx.execute | null }>(sql` select r.id, r.company_id as "companyId", r.native_issue_id as "issueId", r.runner_instance_id as "runnerId", r.native_session_id as "sessionId", r.runner_profile_json->'nativeWorkspaceSync' as workspace, r.runner_profile_json->'remoteWorkspace' as "remoteWorkspace", r.runner_profile_json ? 'remoteWorkspace' as "hasRemoteWorkspace", p.id as "pluginLeaseId", + f.result_id is not null as "hasResult", case when pg_column_size(c.material) <= ${NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES} and sum(pg_column_size(c.material)) over () <= ${NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES} then c.material else null end as material, ( r.runtime_mode = 'native' and a.adapter_type = 'paperclip_runner' @@ -58,19 +62,25 @@ export async function readUpgradeReadiness(db: Db, input: { where r.status = 'running' limit ${MAX_UPGRADE_RUNS + 1}`); }, { accessMode: "read only" }); - const candidates = rows.map(row => { + const candidates: Candidate[] = []; + for (const row of rows) { const remote = remoteWorkspaceRunReferenceSchema.safeParse(row.remoteWorkspace); // A remote reference owns the logical workspace identity. Corruption or conflicting // plugin authority must not fall back to a different recoverable-looking record. const workspaceId = row.hasRemoteWorkspace ? remote.success && !row.pluginLeaseId ? remote.data.workspaceId : null : row.pluginLeaseId ?? readNativeWorkspaceSyncReference(row.workspace)?.workspaceId; - const identity = row.recoverable && row.material + const identity = (row.recoverable || row.hasResult) && row.material ? inspectNativeRunnerUpgradeHeader({ companyId: row.companyId, issueId: row.issueId, runId: row.id }, row.material) : null; - return { id: row.id, recoverable: Boolean(workspaceId && identity && identity.runId === row.id + const bound = Boolean(workspaceId && identity && identity.runId === row.id && identity.runnerInstanceId === row.runnerId && identity.normalizedSessionId === row.sessionId - && identity.environmentLeaseId === workspaceId) }; - }); + && identity.environmentLeaseId === workspaceId); + const finalization = bound && row.hasResult && remote.success && !row.pluginLeaseId + && performance.now() < inspectionDeadline + ? await readNativeFinalizationUpgradeReadiness(db, { companyId: row.companyId, runId: row.id, + controller: owner, supportedOutageMs: 300_000 }).catch(() => false) : false; + candidates.push({ id: row.id, recoverable: bound && (row.recoverable || finalization) }); + } // Additive evidence does not strengthen or redefine the existing ready flag. const retained = await readNativeControllerRecoveryInventory(db).catch(() => ({ version: 1 as const, observedAt: new Date().toISOString(), coverage: "unknown" as const, diff --git a/server/src/services/native-runtime/native-finalization-upgrade-readiness.ts b/server/src/services/native-runtime/native-finalization-upgrade-readiness.ts new file mode 100644 index 00000000000..148d9c4059b --- /dev/null +++ b/server/src/services/native-runtime/native-finalization-upgrade-readiness.ts @@ -0,0 +1,107 @@ +import { and, desc, eq, sql } from "drizzle-orm"; +import { agents, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, + nativeRunResults, statusDecisions, workAssessments, workspaceOperations, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; +import type { NativeControllerIdentity } from "./native-restart-recovery.js"; +import { currentNativeWorkspaceRecoveryResult } from "./native-workspace-recovery-binding.js"; +import { nativeWorkspaceFinalizationOwnerCanUpgrade } from "./native-workspace-finalization-ownership.js"; +import { currentWorkspaceWorker, workspaceWorkerStopProofSupported } from "./native-workspace-platform-proof.js"; +import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; + +export async function readNativeFinalizationUpgradeReadiness(db: Db, input: { + companyId: string; runId: string; controller: NativeControllerIdentity; supportedOutageMs: number; +}): Promise { + const worker = currentWorkspaceWorker(); + if (!workspaceWorkerStopProofSupported(worker)) return false; + return db.transaction(async tx => { + await tx.execute(sql`set local statement_timeout = '2s'`); + const [row] = await tx.select({ + run: { id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, nativePhase: heartbeatRuns.nativePhase, completionContractId: heartbeatRuns.completionContractId, + createdAt: heartbeatRuns.createdAt, + contextSnapshot: sql>`jsonb_strip_nulls(jsonb_build_object( + 'nativeReviewInteractionId', ${heartbeatRuns.contextSnapshot}->'nativeReviewInteractionId', + 'nativeReviewDecisionId', ${heartbeatRuns.contextSnapshot}->'nativeReviewDecisionId'))`, + runnerProfileJson: sql>`jsonb_build_object( + 'remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace', + 'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync' is not null + and ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync' <> 'null'::jsonb, + 'nativeWorkspaceFinalizationOwner', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceFinalizationOwner')` }, + coordinator: { companyId: nativeRunFinalizations.companyId, issueId: nativeRunFinalizations.issueId, + runId: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase, resultId: nativeRunFinalizations.resultId, + controllerBootId: nativeRunFinalizations.controllerBootId, controllerPid: nativeRunFinalizations.controllerPid, + controllerProcessStartedAt: nativeRunFinalizations.controllerProcessStartedAt, controllerGeneration: nativeRunFinalizations.controllerGeneration, + leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + assessmentId: nativeRunFinalizations.assessmentId, decisionId: nativeRunFinalizations.decisionId }, + issue: { id: issues.id, companyId: issues.companyId, status: issues.status, executionRunId: issues.executionRunId, + checkoutRunId: issues.checkoutRunId, assigneeAgentId: issues.assigneeAgentId, executionWorkspaceId: issues.executionWorkspaceId }, + adapter: agents.adapterType, + }) + .from(heartbeatRuns).innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id), + eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId))) + .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, heartbeatRuns.companyId))) + .innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, heartbeatRuns.companyId))) + .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))).limit(1); + if (!row) return false; + const { run, coordinator, issue } = row; + if (run.runtimeMode !== "native" || run.status !== "running" || row.adapter !== "paperclip_runner" + || isNativeRunnerOwnershipHeld(run) || !coordinator.resultId || run.nativeIssueId !== issue.id + || coordinator.controllerBootId !== input.controller.bootId || coordinator.controllerPid !== input.controller.pid + || coordinator.controllerProcessStartedAt?.getTime() !== input.controller.processStartedAt.getTime() + || coordinator.controllerGeneration < 1 + || !["result_accepted", "workspace_finalizing", "retryable_failure", "committed"].includes(coordinator.phase)) return false; + if (coordinator.leaseOwner?.startsWith("native-finalizer:") && coordinator.leaseExpiresAt + && coordinator.leaseExpiresAt.getTime() > Date.now() + input.supportedOutageMs) return false; + const profile = run.runnerProfileJson; + const reference = remoteWorkspaceRunReferenceSchema.safeParse(profile?.remoteWorkspace); + if (!reference.success || profile.nativeWorkspaceSync === true + || !nativeWorkspaceFinalizationOwnerCanUpgrade(profile, input.controller, worker)) return false; + const remote = reference.data; + const [workspace] = await tx.select({ id: executionWorkspaces.id, providerType: executionWorkspaces.providerType, + metadata: sql>`jsonb_build_object('remoteWorkspace', ${executionWorkspaces.metadata}->'remoteWorkspace')` }) + .from(executionWorkspaces).where(and(eq(executionWorkspaces.id, remote.workspaceId), + eq(executionWorkspaces.companyId, input.companyId))).limit(1); + if (!workspace || workspace.providerType !== "cloud_sandbox") return false; + let authority; + try { authority = readRemoteWorkspaceAuthority(workspace.metadata); } catch { return false; } + if (!authority) return false; + const [barrier] = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(eq(workspaceOperations.companyId, input.companyId), + eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.issueId, issue.id), + eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.phase, "workspace_finalize"), + eq(workspaceOperations.status, "succeeded"))).orderBy(desc(workspaceOperations.createdAt)).limit(1); + const [result] = await tx.select({ id: nativeRunResults.id, completionContractId: nativeRunResults.completionContractId, + terminalState: sql`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState'` }).from(nativeRunResults).where(and(eq(nativeRunResults.id, coordinator.resultId), + eq(nativeRunResults.companyId, input.companyId), eq(nativeRunResults.runId, run.id), eq(nativeRunResults.issueId, issue.id), + eq(nativeRunResults.schemaStatus, "accepted"))).limit(1); + if (!result || result.completionContractId !== run.completionContractId) return false; + if (!["succeeded", "failed", "cancelled"].includes(result.terminalState)) return false; + if (coordinator.phase === "committed") { + if (!barrier || !coordinator.decisionId || !coordinator.assessmentId) return false; + const [decision] = await tx.select({ id: statusDecisions.id }).from(statusDecisions).innerJoin(workAssessments, + and(eq(workAssessments.id, statusDecisions.assessmentId), eq(workAssessments.companyId, statusDecisions.companyId), + eq(workAssessments.issueId, statusDecisions.issueId), eq(workAssessments.runId, statusDecisions.runId))) + .where(and(eq(statusDecisions.id, coordinator.decisionId), eq(statusDecisions.companyId, input.companyId), + eq(statusDecisions.issueId, issue.id), eq(statusDecisions.runId, run.id), eq(statusDecisions.applicationState, "applied"), + eq(workAssessments.id, coordinator.assessmentId), eq(workAssessments.resultId, result.id), + eq(workAssessments.contractId, result.completionContractId))).limit(1); + return Boolean(decision); + } + if (!await currentNativeWorkspaceRecoveryResult(tx as unknown as Db, { companyId: input.companyId, run, coordinator, issue })) return false; + if (barrier) return true; + const location = authority.location; + if (!location || authority.writerRunId !== run.id || authority.epoch !== remote.epoch || location.leaseId !== remote.leaseId + || location.providerLeaseId !== remote.providerLeaseId || location.remoteCwd !== remote.remoteCwd + || issue.executionWorkspaceId !== workspace.id) return false; + const [lease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).innerJoin(environments, + eq(environments.id, environmentLeases.environmentId)).where(and(eq(environmentLeases.id, location.leaseId), + eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, location.environmentId), + eq(environmentLeases.executionWorkspaceId, workspace.id), eq(environmentLeases.heartbeatRunId, run.id), + eq(environmentLeases.issueId, issue.id), eq(environmentLeases.providerLeaseId, location.providerLeaseId), + eq(environments.driver, "sandbox"), eq(environments.status, "active"), + eq(environmentLeases.status, "active"), + sql`${environmentLeases.releasedAt} is null and ${environmentLeases.cleanupStatus} is null`, + sql`(${environmentLeases.expiresAt} is null or ${environmentLeases.expiresAt} > clock_timestamp())`)).limit(1); + return Boolean(lease); + }, { accessMode: "read only" }); +} diff --git a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts index 3dd2316b76b..9d51af94999 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts @@ -17,6 +17,7 @@ import { withWorkspaceLockKeepalive } from "./native-workspace-lock-keepalive.js import { currentNativeControllerIdentity, evaluateNativeControllerTakeover, + type NativeControllerIdentity, } from "./native-restart-recovery.js"; import { currentWorkspaceWorker, requestWorkspaceWorkerStopProof, type WorkspaceWorker, type WorkspaceWorkerStopProof } from "./native-workspace-platform-proof.js"; @@ -75,6 +76,18 @@ function readOwner(value: unknown): Owner | null { ? owner as Owner : null; } +export function nativeWorkspaceFinalizationOwnerCanUpgrade(profile: Record | null, + identity: NativeControllerIdentity, worker: WorkspaceWorker | undefined): boolean { + const raw = profile?.[OWNER_KEY]; + if (raw == null) return true; + const owner = readOwner(raw); + return Boolean(owner && owner.token.length > 0 && owner.token.length <= 200 && worker && owner.worker + && Object.keys(owner.worker).length === Object.keys(worker).length && owner.controllerBootId === identity.bootId + && owner.hostname === os.hostname() && owner.pid === identity.pid + && Date.parse(owner.processStartedAt) === identity.processStartedAt.getTime() + && Object.keys(worker).every(key => owner.worker![key as keyof WorkspaceWorker] === worker[key as keyof WorkspaceWorker])); +} + /** * The advisory lock serializes controllers without holding row locks across I/O. * The durable receipt additionally prevents a disconnected lock session from diff --git a/server/src/services/native-runtime/native-workspace-platform-proof.ts b/server/src/services/native-runtime/native-workspace-platform-proof.ts index 2126f605f36..e5003686f14 100644 --- a/server/src/services/native-runtime/native-workspace-platform-proof.ts +++ b/server/src/services/native-runtime/native-workspace-platform-proof.ts @@ -62,6 +62,15 @@ export type WorkspaceWorkerStopCheck = | { status: "pending" } | { status: "unsupported" }; +export function workspaceWorkerStopProofSupported(worker: WorkspaceWorker | undefined, env: NodeJS.ProcessEnv = process.env): boolean { + const context = getCloudStackContext(env); + if (!context?.stackId || !context.cloudOrigin || !env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN?.trim() || !validWorker(worker)) return false; + try { + const origin = new URL(context.cloudOrigin); + return origin.protocol === "https:" && !origin.username && !origin.password; + } catch { return false; } +} + export async function requestWorkspaceWorkerStopProof(binding: WorkspaceStopBinding, options: { env?: NodeJS.ProcessEnv; fetchImpl?: typeof fetch; now?: Date; @@ -69,7 +78,7 @@ export async function requestWorkspaceWorkerStopProof(binding: WorkspaceStopBind const env = options.env ?? process.env; const context = getCloudStackContext(env); const token = env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN?.trim(); - if (!context?.stackId || !context.cloudOrigin || !token || !validWorker(binding.worker)) return { status: "unsupported" }; + if (!workspaceWorkerStopProofSupported(binding.worker, env) || !context?.stackId || !context.cloudOrigin || !token) return { status: "unsupported" }; let origin: URL; try { origin = new URL(context.cloudOrigin); } catch { return { status: "unsupported" }; } if (origin.protocol !== "https:" || origin.username || origin.password) return { status: "unsupported" }; diff --git a/server/src/services/native-runtime/native-workspace-recovery-binding.ts b/server/src/services/native-runtime/native-workspace-recovery-binding.ts index a9ab9f0d20b..7c6a4475406 100644 --- a/server/src/services/native-runtime/native-workspace-recovery-binding.ts +++ b/server/src/services/native-runtime/native-workspace-recovery-binding.ts @@ -2,13 +2,12 @@ import { and, eq, gt, inArray, isNull, ne, not, or, sql } from "drizzle-orm"; import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { getNativeReviewAssignment } from "./native-review-participant.js"; -/** Both operator and platform recovery must still own the task's accepted work. - * Call with coordinator -> issue -> run rows locked in a short transaction. */ export async function currentNativeWorkspaceRecoveryResult(db: Db, input: { companyId: string; - issue: typeof issues.$inferSelect; - run: typeof heartbeatRuns.$inferSelect; - coordinator: typeof nativeRunFinalizations.$inferSelect; + issue: Pick; + run: Pick; + coordinator: Pick; }) { const { companyId, issue, run, coordinator } = input; if (issue.companyId !== companyId || run.companyId !== companyId || coordinator.companyId !== companyId @@ -22,7 +21,7 @@ export async function currentNativeWorkspaceRecoveryResult(db: Db, input: { contextSnapshot: run.contextSnapshot, actingRunId: run.id, issueExecutionRunId: issue.executionRunId, allowResolvedByRunId: run.id, })) return null; - const [result] = await db.select().from(nativeRunResults).where(and( + const [result] = await db.select({ id: nativeRunResults.id, completionContractId: nativeRunResults.completionContractId }).from(nativeRunResults).where(and( eq(nativeRunResults.id, coordinator.resultId), eq(nativeRunResults.companyId, companyId), eq(nativeRunResults.issueId, issue.id), eq(nativeRunResults.runId, run.id), eq(nativeRunResults.schemaStatus, "accepted"), ));