diff --git a/server/src/services/native-runtime/native-workspace-finalization-supersession.test.ts b/server/src/services/native-runtime/native-workspace-finalization-supersession.test.ts new file mode 100644 index 00000000000..7b2a0cbf651 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-finalization-supersession.test.ts @@ -0,0 +1,148 @@ +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 } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { + activityLog, + agents, + companies, + completionContracts, + createDb, + heartbeatRuns, + issues, + nativeRunFinalizations, + nativeRunResults, + workspaceOperations, +} from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js"; +import { resumeNativeWorkspaceFinalization } from "./native-workspace-finalizer.js"; + +// RC-64: a crash loop killed the controller after it recorded a workspace +// finalize attempt as `running`. A later owner finalized the same run, and the +// orphaned attempt stayed `running` forever. +describe("superseded native workspace finalization attempts", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let workspaceRoot: string; + let priorLogRoot: string | undefined; + const companyId = randomUUID(); + const agentId = randomUUID(); + + beforeAll(async () => { + workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-finalize-supersession-")); + 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-finalize-supersession-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 }); + await db.insert(companies).values({ id: companyId, name: "Finalize supersession", issuePrefix: "FSS" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Finalize agent", + adapterType: "paperclip_runner", status: "running" }); + }, 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; + }); + + /** A succeeded native run with an accepted result and a dead controller's running finalize. */ + async function seedOrphanedFinalize(options: { cwdExists?: boolean; metadata?: Record } = {}) { + const issueId = randomUUID(); + const contractId = randomUUID(); + const runId = randomUUID(); + const resultId = randomUUID(); + const cwd = path.join(workspaceRoot, `orphan-${randomUUID()}`); + if (options.cwdExists !== false) await fs.mkdir(cwd); + await db.insert(issues).values({ id: issueId, companyId, title: "Crash-loop finalize", status: "in_progress", + workMode: "standard", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "finalize-supersession-test-v1", + risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { objective: "finalize" }, canonicalSha256: `contract-${contractId}`, + createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: issueId, completionContractId: contractId, + runnerProfileJson: { nativeExecutionInput: { binding: { executionWorkspaceId: randomUUID() } } } }); + 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: "result_accepted", resultId }); + const startedAt = new Date(Date.now() - 60_000); + const [orphan] = await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: runId, issueId, + phase: "workspace_finalize", status: "running", cwd, startedAt, createdAt: startedAt, updatedAt: startedAt, + metadata: { owningService: "native_workspace_finalizer", observation: "workspace_directory", + resumedFromOperationId: null, ...options.metadata }, + }).returning(); + return { issueId, runId, orphan }; + } + + const read = async (id: string) => + (await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, id)))[0]!; + + it("settles the orphan once a resumed attempt succeeds", async () => { + const seeded = await seedOrphanedFinalize(); + const operation = await resumeNativeWorkspaceFinalization({ db, runId: seeded.runId }); + expect(operation).toMatchObject({ status: "succeeded", + metadata: expect.objectContaining({ resumedFromOperationId: seeded.orphan.id }) }); + const orphan = await read(seeded.orphan.id); + expect(orphan).toMatchObject({ status: "failed", finishedAt: expect.any(Date) }); + expect(orphan.metadata).toMatchObject({ owningService: "native_workspace_finalizer", + supersededFinalization: { reason: "superseded_workspace_finalization", supersedingOperationId: operation!.id, + supersedingStatus: "succeeded", previousStatus: "running" } }); + expect(await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, seeded.runId), + eq(workspaceOperations.status, "running")))).toHaveLength(0); + expect(await db.select().from(activityLog).where(and(eq(activityLog.entityId, seeded.orphan.id), + eq(activityLog.action, "workspace_operation.reconciled")))).toHaveLength(1); + }); + + it("settles the orphan when the resumed attempt itself fails", async () => { + const seeded = await seedOrphanedFinalize({ cwdExists: false }); + const operation = await resumeNativeWorkspaceFinalization({ db, runId: seeded.runId }); + expect(operation?.status).toBe("failed"); + const orphan = await read(seeded.orphan.id); + expect(orphan.status).toBe("failed"); + expect(orphan.metadata).toMatchObject({ supersededFinalization: { + supersedingOperationId: operation!.id, supersedingStatus: "failed" } }); + }); + + it("settles an orphan older than an already succeeded attempt", async () => { + const seeded = await seedOrphanedFinalize(); + const [success] = await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: seeded.runId, + issueId: seeded.issueId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(), + metadata: { owningService: "native_workspace_finalizer", resumedFromOperationId: seeded.orphan.id } }).returning(); + const operation = await resumeNativeWorkspaceFinalization({ db, runId: seeded.runId }); + expect(operation?.id).toBe(success.id); + const orphan = await read(seeded.orphan.id); + expect(orphan.status).toBe("failed"); + expect(orphan.metadata).toMatchObject({ supersededFinalization: { supersedingOperationId: success.id } }); + }); + + it("leaves a running attempt alone while another controller owns the run", async () => { + const seeded = await seedOrphanedFinalize(); + const result = await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, + () => resumeNativeWorkspaceFinalization({ db, runId: seeded.runId })); + expect(result).toEqual({ acquired: true, value: null }); + expect((await read(seeded.orphan.id)).status).toBe("running"); + }); + + it("does not settle a running finalize recorded by another service", async () => { + const seeded = await seedOrphanedFinalize({ metadata: { owningService: "heartbeat" } }); + expect((await resumeNativeWorkspaceFinalization({ db, runId: seeded.runId }))?.status).toBe("succeeded"); + expect((await read(seeded.orphan.id)).status).toBe("running"); + }); + + it.each([ + ["saveWork", { saveWork: { requested: true } }], + ["uncertain", { uncertain: true }], + ])("keeps an attempt with an independent %s obligation running", async (_name, metadata) => { + const seeded = await seedOrphanedFinalize({ metadata }); + expect((await resumeNativeWorkspaceFinalization({ db, runId: seeded.runId }))?.status).toBe("succeeded"); + expect((await read(seeded.orphan.id)).status).toBe("running"); + }); +}); diff --git a/server/src/services/native-runtime/native-workspace-finalization-supersession.ts b/server/src/services/native-runtime/native-workspace-finalization-supersession.ts new file mode 100644 index 00000000000..40e5e191336 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-finalization-supersession.ts @@ -0,0 +1,91 @@ +import { and, asc, eq, ne, sql } from "drizzle-orm"; +import { workspaceOperations, type Db } from "@paperclipai/db"; +import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; +import { isNativeWorkspaceFinalizationOperationActive } from "../workspace-operations.js"; +import { notifyTerminalWorkspaceWork } from "../terminal-workspace-work.js"; + +export const SUPERSEDED_NATIVE_FINALIZATION_REASON = "superseded_workspace_finalization"; + +/** + * Settle earlier `running` native workspace-finalization operations for one run + * once a later attempt for that run has reached a terminal status (RC-64). + * + * The caller must hold the run's native workspace-finalization ownership. That + * ownership is the proof that no other controller is executing an export for + * this run: it holds the advisory lock and has claimed the durable owner receipt, + * which requires the previous controller to have released it or to have been + * proven stopped (same-host process proof, platform proof, or a completed + * copyback barrier). The only remaining live executor could be this process, + * which the in-process registry rules out. Age is never used. + */ +export async function settleSupersededNativeWorkspaceFinalizations(db: Db, input: { + companyId: string; + runId: string; + issueId: string; + superseding: { id: string; status: string }; +}): Promise { + if (input.superseding.status === "running") return []; + const candidates = await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, input.companyId), + eq(workspaceOperations.heartbeatRunId, input.runId), + eq(workspaceOperations.issueId, input.issueId), + eq(workspaceOperations.phase, "workspace_finalize"), + eq(workspaceOperations.status, "running"), + ne(workspaceOperations.id, input.superseding.id), + sql`${workspaceOperations.metadata}->>'owningService' = 'native_workspace_finalizer'`, + // Keep PostgreSQL timestamp precision when proving which attempt is later. + sql`${workspaceOperations.createdAt} < (select created_at from workspace_operations where id = ${input.superseding.id}::uuid)`, + )).orderBy(asc(workspaceOperations.createdAt), asc(workspaceOperations.id)); + + const settled: string[] = []; + for (const candidate of candidates) { + if (isNativeWorkspaceFinalizationOperationActive({ + operationId: candidate.id, companyId: input.companyId, runId: input.runId, issueId: input.issueId, + })) continue; + const metadata = candidate.metadata ?? {}; + // Independent obligations are never retired by a later export. + if (metadata.uncertain === true || metadata.uncertain === "true" || metadata.saveWork != null) continue; + const now = new Date(); + const provenance = { + schema: "paperclip.native-workspace-finalization-supersession.v1", + reason: SUPERSEDED_NATIVE_FINALIZATION_REASON, + proof: "native_workspace_finalization_ownership", + previousStatus: candidate.status, + supersedingOperationId: input.superseding.id, + supersedingStatus: input.superseding.status, + runId: input.runId, + reconciledAt: now.toISOString(), + }; + const message = "Workspace finalization was interrupted before it finished; " + + `a later finalization attempt (${input.superseding.id}) ${input.superseding.status}.\n`; + let publication: ActivityPublication | undefined; + const updated = await db.transaction(async (tx) => { + // Register the wake intent before the write so drain and sleep checks see it. + await notifyTerminalWorkspaceWork(tx, candidate.executionWorkspaceId + ? `workspace:${candidate.executionWorkspaceId}` : `issue:${input.issueId}`); + const [row] = await tx.update(workspaceOperations).set({ + status: "failed", + finishedAt: now, + updatedAt: now, + stderrExcerpt: `${candidate.stderrExcerpt ?? ""}${message}`.slice(-4096), + metadata: { ...metadata, supersededFinalization: provenance }, + }).where(and( + eq(workspaceOperations.id, candidate.id), + eq(workspaceOperations.companyId, input.companyId), + eq(workspaceOperations.status, "running"), + )).returning({ id: workspaceOperations.id }); + if (!row) return null; + const activity = await persistActivity(tx as unknown as Db, { + companyId: input.companyId, actorType: "system", actorId: "native-workspace-finalizer", + action: "workspace_operation.reconciled", entityType: "workspace_operation", entityId: candidate.id, + runId: input.runId, issueId: input.issueId, details: provenance, + }); + publication = activity.publication; + return row; + }); + if (!updated) continue; + if (publication) publishActivity(publication); + settled.push(candidate.id); + } + return settled; +} diff --git a/server/src/services/native-runtime/native-workspace-finalizer.ts b/server/src/services/native-runtime/native-workspace-finalizer.ts index cd0967e63be..9bdd2c2aaf7 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.ts +++ b/server/src/services/native-runtime/native-workspace-finalizer.ts @@ -5,7 +5,9 @@ import { recordNativeFinalizationFailure } from "./native-run-finalizer.js"; import { inspectPluginWorkspaceForFinalization } from "./native-plugin-workspace-finalization.js"; import type { WorkspaceWorkers } from "../plugin-workspaces.js"; import { restoreNativeWorkspaceBestEffort } from "./native-workspace-best-effort.js"; -import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js"; +import { withNativeWorkspaceFinalizationOwnership, type NativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js"; +import { settleSupersededNativeWorkspaceFinalizations } from "./native-workspace-finalization-supersession.js"; +import { logger } from "../../middleware/logger.js"; import fs from "node:fs/promises"; import { and, desc, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -83,6 +85,20 @@ export async function attemptNativeWorkspaceFinalization(input: { throw new Error("native_workspace_finalization_binding_missing"); } + // An attempt interrupted by a controller crash stays `running`; once this + // owner records a terminal attempt, settle the orphan under the same proof. + const settleSuperseded = async (ownership: NativeWorkspaceFinalizationOwnership, + superseding: { id: string; status: string }) => { + try { + await ownership.assertHeld(); + await settleSupersededNativeWorkspaceFinalizations(input.db, { + companyId: bound.companyId, runId: input.runId, issueId: bound.issueId, superseding, + }); + } catch (error) { + // Settlement is bookkeeping; it must not change the finalization outcome. + logger.warn({ runId: input.runId, error }, "Superseded workspace finalization could not be settled"); + } + }; const owned = await withNativeWorkspaceFinalizationOwnership({ db: input.db, companyId: bound.companyId, runId: input.runId, }, async (ownership) => { @@ -113,7 +129,10 @@ export async function attemptNativeWorkspaceFinalization(input: { eq(workspaceOperations.status, "succeeded"), )).orderBy(desc(workspaceOperations.createdAt)).limit(1).then((rows) => rows[0] ?? null); // A stale failure from a pre-fencing controller cannot invalidate exported work. - if (successful) return { kind: "completed", operation: successful } as const; + if (successful) { + await settleSuperseded(ownership, successful); + return { kind: "completed", operation: successful } as const; + } const previous = await input.db.select().from(workspaceOperations).where(and( eq(workspaceOperations.companyId, bound.companyId), eq(workspaceOperations.heartbeatRunId, input.runId), @@ -345,6 +364,7 @@ export async function attemptNativeWorkspaceFinalization(input: { }; }, }); + await settleSuperseded(ownership, operation); return { kind: "completed", operation } as const; }); return owned.acquired ? owned.value : { kind: "waiting", reason: owned.reason } as const;