From bee0f0e2fa9f2d6f24e60e2a81c165856b21b7ca Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 17:30:28 -0600 Subject: [PATCH] fix(server): recognize recoverable result finalization during upgrades (#19651) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native execution saves an accepted result before it finishes workspace export and status projection. > - The upgrade report currently recognizes only a live execution in the observed phase. > - A durable accepted result can therefore block an upgrade after its execution lease is released. > - Existing finalization owners already resume that saved work with task binding, physical worker stop proof, and an atomic owner claim. > - This pull request lets the report inspect those existing contracts without taking authority. > - Qualified remote work can continue through a normal server replacement. ## Linked Issues or Issue Description Refs #18037 and #19596. **What happened?** An accepted remote native result remains in result acceptance, workspace finalization, retry, or committed projection. The upgrade report marks it blocking because it is no longer a result-less observed execution with a live execution lease. **Expected behavior** Report a saved result as recoverable only when the existing replacement owners can resume it. Preserve all task, result, contract, remote workspace, provider lease, physical owner, decision, and workspace barrier checks. Keep unsupported or unknown work blocking. **Steps to reproduce** Run the new PostgreSQL post-result readiness fixture. It persists a real encrypted Runner header and an accepted terminal result with a bound remote workspace. The unchanged report returns one blocker. **Paperclip version or commit** Regression base: ee6c101854. Related shutdown recovery is in #19596. ## What Changed - Delegate saved-result eligibility to the native finalization owner. - Require the accepted terminal result and contract, current task binding, remote authority, retained provider lease or successful workspace receipt, and a supported physical worker identity. - Require an applied decision and assessment for committed projection. Keep finalizer leases beyond the supported outage budget blocking. - Reuse the existing result recovery binding with selected metadata. Do not load complete run or result payloads for the new observation. - Keep real signed stop verification, owner compare-and-swap, retry deadlines, and actual finalization with their existing owners. - Share a two-second scheduling budget across saved-result inspections, leaving unchecked candidates blocking after the budget expires. - Add PostgreSQL, encrypted-header, signed stop-proof, competing-owner, projection, and independent-work cases. Clarify the operator contract. ## Verification - The valid accepted-result fixture reproduced the original blocking report, then passed with this change. - The initial eight post-result cases passed against PostgreSQL. They include a real Ed25519 stop receipt, the existing verifier, a competing PostgreSQL ownership claim, one copyback, and receipt reuse. - Seventeen focused existing PostgreSQL shutdown and workspace ownership cases passed. - A broader five-file run passed 91 cases. Its one failure was an invalid negative fixture: PostgreSQL correctly refused a contract change protected by the accepted-result foreign key. The test now asserts that refusal. - A subsequent standalone run timed out in PostgreSQL bootstrap before tests ran. The final run used the existing shared PostgreSQL command and isolated per-suite template databases: all 92 tests passed across five files (159.95 seconds). It covers both readiness suites, platform stop proof, workspace finalization, and graceful shutdown. That scope passed on `ce4e3995b6`. A follow-up restricts provider leases to the active status and adds released/unknown-status controls. The new fixture now uses the existing PostgreSQL bootstrap timeout constant. The affected eight-case rerun passed on `850859b44d`. The explicit configured-stack guard also passed its six platform-proof cases. Broad local source typecheck still has stale borrowed dependency declarations; fresh CI supplies that coverage. - Review's locked-query regression reproduced RED: three saved-result candidates took 6.081 seconds. With the shared scheduling budget, both readiness suites passed all 16 cases (49.36 seconds) on final code committed as `70b0b7b8f7`. The lock case leaves all three candidates blocking, releases its connection, and observes all three as recoverable after the lock is released. - Greptile reviewed final `70b0b7b8f7` at 5/5 with no open threads. Current-head CI typecheck, build, tests, and native review are pending. No hosted replacement or enrollment acceptance is claimed. ## Risks A false positive could select an unsafe upgrade. Missing evidence and read errors remain blocking. Readiness does not stop admission, detach an agent, acquire a lease, request a stop receipt, or authorize takeover. The replacement still verifies physical stop and rechecks current task ownership before copyback. Saved results retaining a previous controller boot remain conservatively blocking, including finalizers resumed by a later boot. This PR does not change enrollment policy or migrate data. ## Model Used OpenAI GPT-6.1, model ID `gpt-6.1-sol`, with reasoning, tool use, and code execution. The host did not expose the context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/testing-native-remote-continuity.md | 14 +- .../upgrade-readiness-post-result.test.ts | 262 ++++++++++++++++++ .../services/heartbeat/upgrade-readiness.ts | 22 +- .../native-finalization-upgrade-readiness.ts | 107 +++++++ ...native-workspace-finalization-ownership.ts | 13 + .../native-workspace-platform-proof.ts | 11 +- .../native-workspace-recovery-binding.ts | 11 +- 7 files changed, 425 insertions(+), 15 deletions(-) create mode 100644 server/src/services/heartbeat/upgrade-readiness-post-result.test.ts create mode 100644 server/src/services/native-runtime/native-finalization-upgrade-readiness.ts 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"), ));