diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index d1914ccbe21..4e7b4e352cd 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -159,8 +159,12 @@ checks under short coordinator, issue and run row locks, atomically with the owner claim. The task must still be open, assigned to the run or its authoritative review participant, and bound to that running native execution and accepted result. A different checkout, newer execution or queued/running legacy execution -prevents takeover. Publication checks repeat this validation; row locks are never -held across physical copyback I/O. +prevents takeover. A later wake for the same assignee that has never started, +such as the answer to the run's question, does not. Its claim waits for this +run's execution lock, so it waits for this copyback. Counting it would block both +runs. Another agent's run and a recovery-action wake still prevent takeover. +Publication checks repeat this validation, so a wake that starts during copyback +stops publication. Row locks are never held across physical copyback I/O. The harness endpoint is `POST /v1/tenant/workspace-worker-stop`. The fleet-wide server token authorizes a read request, never stop evidence: responses must be diff --git a/server/src/services/native-runtime/native-workspace-finalization-work.ts b/server/src/services/native-runtime/native-workspace-finalization-work.ts index 0152995b6b2..a6f85961f4b 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-work.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-work.ts @@ -2,6 +2,7 @@ import { and, asc, eq, gt, ne, notExists } from "drizzle-orm"; import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { logger } from "../../middleware/logger.js"; import { onNativeWorkspaceFinalizationRelease, type NativeWorkspaceFinalizationScope } from "../native-workspace-finalization-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { measureRunPhase, withRunWorkspaceFinalizationTiming } from "../execution-preparation-timing.js"; @@ -81,8 +82,14 @@ export function nativeWorkspaceFinalizationWork(db: Db, options: { })); if (outcome.kind === "retry") return { ...due, dueAt: outcome.retryAt.getTime() }; if (outcome.kind === "waiting") { - return outcome.reason === "busy" || outcome.reason === "platform_stop_pending" - ? { ...due, dueAt: Date.now() + OWNERSHIP_RECHECK_MS } : null; + if (outcome.reason === "busy" || outcome.reason === "platform_stop_pending") { + return { ...due, dueAt: Date.now() + OWNERSHIP_RECHECK_MS }; + } + // This run is revisited only after a restart or a release notification. + if (outcome.reason !== "owner_active") { + logger.info({ runId: current.id, reason: outcome.reason }, "native workspace finalization paused until the run is notified again"); + } + return null; } if (outcome.kind !== "completed" || outcome.operation.status === "succeeded") return null; const failure = await recordNativeWorkspaceFinalizationFailure(db, current.id, outcome.operation); diff --git a/server/src/services/native-runtime/native-workspace-finalizer.test.ts b/server/src/services/native-runtime/native-workspace-finalizer.test.ts index 0d57c171400..a60a0f05471 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.test.ts +++ b/server/src/services/native-runtime/native-workspace-finalizer.test.ts @@ -399,7 +399,7 @@ describe("native workspace finalization recovery", () => { it("does not reopen an old result after a newer task execution exists", async () => { const seed = await seedAbandonedCopyback(); - await db.insert(heartbeatRuns).values({ companyId, agentId, status: "queued", nativeIssueId: seed.issueId }); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running", startedAt: new Date(), nativeIssueId: seed.issueId }); await expect(resumeNativeWorkspaceAfterOwnerStop(seed)).rejects.toThrow("no longer current"); }); diff --git a/server/src/services/native-runtime/native-workspace-queued-wake-recovery.test.ts b/server/src/services/native-runtime/native-workspace-queued-wake-recovery.test.ts new file mode 100644 index 00000000000..fc2e8820834 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-queued-wake-recovery.test.ts @@ -0,0 +1,177 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { + agents, + companies, + completionContracts, + createDb, + executionWorkspaces, + heartbeatRuns, + issues, + nativeRunFinalizations, + nativeRunResults, + projects, + workspaceOperations, +} from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import { nativeSha256 } from "./canonical.js"; +import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js"; +import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js"; +import { resumeNativeWorkspaceAfterOwnerStop } from "./native-workspace-owner-recovery.js"; +import * as platformProof from "./native-workspace-platform-proof.js"; + +// A user's answer queues the assignee's next wake behind the saved run's +// execution lock. Recovery of that saved run must not wait for the wake. +describe("native workspace recovery with a queued answer wake", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let workspaceRoot: string; + let priorLogRoot: string | undefined; + const companyId = randomUUID(); + const agentId = randomUUID(); + const projectId = randomUUID(); + + beforeAll(async () => { + workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-native-queued-wake-recovery-")); + priorLogRoot = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = path.join(workspaceRoot, "operation-logs"); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_WORKSPACE_FINALIZER_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-queued-wake-recovery-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 }); + await db.insert(companies).values({ id: companyId, name: "Queued wake recovery", issuePrefix: "QWR" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native workspace agent", adapterType: "paperclip_runner", status: "running" }); + await db.insert(projects).values({ id: projectId, companyId, name: "Native workspace project", status: "active" }); + }, 30_000); + + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); + await temporary?.cleanup(); + await fs.rm(workspaceRoot, { recursive: true, force: true }); + if (priorLogRoot === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorLogRoot; + }); + + async function seedAbandonedCopyback() { + const issueId = randomUUID(), contractId = randomUUID(), runId = randomUUID(), resultId = randomUUID(); + const worker = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "d" }; + const owner = { token: randomUUID(), hostname: "replaced-controller", pid: 2, + processStartedAt: "2026-10-10T00:02:55.560Z", controllerBootId: randomUUID(), worker }; + await db.insert(issues).values({ id: issueId, companyId, title: "Saved copyback with an answer wake", + status: "in_progress", workMode: "standard", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "native-queued-wake-recovery-test-v1", + risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { objective: "Saved copyback" }, canonicalSha256: `contract-${contractId}`, + createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", + nativeIssueId: issueId, nativePhase: "workspace_finalizing", completionContractId: contractId, + runnerProfileJson: { nativeExecutionInput: { binding: { executionWorkspaceId: randomUUID() } }, + nativeWorkspaceFinalizationOwner: owner } }); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, + serverFingerprint: `result-${resultId}`, schemaStatus: "accepted", resultJson: {}, canonicalSha256: `result-${resultId}` }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "workspace_finalizing", resultId }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", ownerType: "board", cause: "native_workspace_finalization_owner_unverified", + fingerprint: nativeSha256({ runId, owner }), evidence: { runId, owner }, maxAttempts: 1, + nextAction: "Verify the previous controller and its workspace-copyback processes have stopped." }); + return { companyId, db, issueId, runId, owner, actionId: action.id, ownerToken: owner.token, actorId: "operator", + controllerAndCopybackStopped: true as const, + stopEvidence: "Deployment platform confirms the exact old container and all its processes exited." }; + } + + async function queueAnswerWake(seed: { runId: string; issueId: string }, values: Partial = {}) { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId)); + const [wake] = await db.insert(heartbeatRuns).values({ companyId, agentId, runtimeMode: "native", status: "queued", + contextSnapshot: { issueId: seed.issueId, wakeReason: "ask_user_questions.answered" }, + createdAt: new Date(run.createdAt.getTime() + 60_000), ...values }).returning(); + return wake; + } + + function proveStopped(seed: Awaited>) { + return vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { + v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, + ownerToken: seed.owner.token, worker: seed.owner.worker, stopped: true, iat: 1, exp: 61 } }); + } + + it("confirms owner recovery while the answer wake waits", async () => { + const seed = await seedAbandonedCopyback(); + await queueAnswerWake(seed); + await expect(resumeNativeWorkspaceAfterOwnerStop(seed)).resolves.toMatchObject({ runId: seed.runId, status: "queued" }); + }); + + it.each(["still queued", "starts during copyback"] as const)("adopts a proven stopped deployment behind the answer wake: %s", async state => { + const seed = await seedAbandonedCopyback(); + const wake = await queueAnswerWake(seed); + const probe = proveStopped(seed); + const write = vi.fn(); + try { + const copyback = withNativeWorkspaceFinalizationOwnership(seed, async ownership => { + if (state === "starts during copyback") await db.update(heartbeatRuns).set({ status: "running", startedAt: new Date() }).where(eq(heartbeatRuns.id, wake.id)); + await ownership.assertHeld(); + write(); + return "saved workspace copied"; + }); + if (state === "still queued") expect(await copyback).toEqual({ acquired: true, value: "saved workspace copied" }); + else await expect(copyback).rejects.toThrow("native_workspace_finalization_ownership_lost"); + expect(write).toHaveBeenCalledTimes(state === "still queued" ? 1 : 0); + } finally { probe.mockRestore(); } + }); + + // These runs have started, or can start without the saved run's execution lock. + it.each(["requeued started wake", "recovery action wake", "other agent's queued run"])("keeps the takeover fenced for %s", async changed => { + const seed = await seedAbandonedCopyback(); + if (changed === "requeued started wake") await queueAnswerWake(seed, { startedAt: new Date() }); + if (changed === "recovery action wake") await queueAnswerWake(seed, { contextSnapshot: { issueId: seed.issueId, wakeReason: "source_scoped_recovery_action" } }); + if (changed === "other agent's queued run") { + const [other] = await db.insert(agents).values({ companyId, name: "Mentioned agent", adapterType: "paperclip_runner", status: "running" }).returning(); + await queueAnswerWake(seed, { agentId: other.id }); + } + const probe = proveStopped(seed); + const copyback = vi.fn(); + try { + expect(await withNativeWorkspaceFinalizationOwnership(seed, copyback)).toEqual({ acquired: false, reason: "binding_changed" }); + expect(copyback).not.toHaveBeenCalled(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId)); + expect(run.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toEqual(seed.owner); + } finally { probe.mockRestore(); } + }); + + it("commits a retried copyback from a stopped deployment while the answer wake waits", async () => { + const seed = await seedAbandonedCopyback(); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1, + nextAttemptAt: new Date(Date.now() - 60_000), failureCode: "native_workspace_sync_out_failed", + }).where(eq(nativeRunFinalizations.runId, seed.runId)); + await db.update(heartbeatRuns).set({ nativePhase: "retryable_failure" }).where(eq(heartbeatRuns.id, seed.runId)); + const workspaceId = randomUUID(), cwd = path.join(workspaceRoot, seed.runId); + await fs.mkdir(cwd); + await fs.writeFile(path.join(cwd, "saved.txt"), "accepted work"); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, projectId, mode: "local", strategyType: "local_directory", name: "Retained workspace", cwd }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, seed.issueId)); + await db.update(nativeRunResults).set({ resultJson: { + result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", summary: "Wait for the answer.", + completionClaim: { contractRevision: "test", objectiveSatisfied: false, criteria: [], + remainingWork: [{ description: "Continue after the answer.", blocksCompletion: true }] }, + continuation: { kind: "response_wake", summary: "Wait for the answer.", idempotencyKey: "saved-wait" }, + evidence: [], verification: [], attentionRequests: [], artifacts: [] }, + terminal: { schema: "paperclip.prp.terminal.v1", turnTerminalState: "completed", + runTerminalState: "succeeded", reportedWorkDisposition: "yielded" }, + } }).where(eq(nativeRunResults.runId, seed.runId)); + const wake = await queueAnswerWake(seed); + const probe = proveStopped(seed); + try { await reconcileNativeFinalizations(db, [seed.runId]); } finally { probe.mockRestore(); } + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId)))[0].phase).toBe("committed"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId)))[0]).toMatchObject({ status: "succeeded", nativePhase: "committed" }); + expect(await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, seed.runId), + eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded")))).toHaveLength(1); + expect(await fs.readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("accepted work"); + // The saved run is terminal, so the wake's execution claim is no longer blocked. + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.id)))[0]).toMatchObject({ status: "queued", startedAt: null }); + }); +}); 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 20c4ee6861f..a9ab9f0d20b 100644 --- a/server/src/services/native-runtime/native-workspace-recovery-binding.ts +++ b/server/src/services/native-runtime/native-workspace-recovery-binding.ts @@ -1,4 +1,4 @@ -import { and, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; +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"; @@ -27,10 +27,19 @@ export async function currentNativeWorkspaceRecoveryResult(db: Db, input: { eq(nativeRunResults.issueId, issue.id), eq(nativeRunResults.runId, run.id), eq(nativeRunResults.schemaStatus, "accepted"), )); if (!result || result.completionContractId !== run.completionContractId) return null; + // The assignee's later wakes, such as an answer to this run's question, + // cannot claim while this run holds the execution lock. They wait for this + // copyback, so they are not newer work. Other agents' runs are not blocked. + const waitingWake = issue.assigneeAgentId === run.agentId ? and( + eq(heartbeatRuns.status, "queued"), isNull(heartbeatRuns.startedAt), + eq(heartbeatRuns.agentId, run.agentId), gt(heartbeatRuns.createdAt, run.createdAt), + sql`coalesce(${heartbeatRuns.contextSnapshot}->>'wakeReason', '') <> 'source_scoped_recovery_action'`, + ) : undefined; const [newer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), ne(heartbeatRuns.id, run.id), or(eq(heartbeatRuns.nativeIssueId, issue.id), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), or(gt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, ["queued", "running"])), + waitingWake ? not(waitingWake) : undefined, )).limit(1); return newer ? null : result; }