mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(workspaces): reconcile superseded offline finalization attempts (#19179)
## Thinking Path > - Paperclip manages AI agents and preserves their work. > - Workspace migration must keep saved files and their recovery records. > - A crashed finalization attempt can remain running after a later attempt succeeds. > - The old row then blocks migration even when the saved result and export have exact receipts. > - Offline maintenance can reconcile the obsolete attempt after Cloud stops all old writers. > - This pull request fails only that attempt and records the receipt that supersedes it. > - Missing proof still blocks migration. ## Linked Issues or Issue Description Related storage migration work: #18841 and #18943. **What happened?** An old running workspace finalization row can block offline enrollment after a later finalization has saved the same run and workspace. **Expected behavior** Maintenance should reconcile the old attempt when exact saved receipts prove it is obsolete. It must preserve the accepted result, successful receipt, execution ownership, local files and recovery records. **Steps to reproduce** 1. Retain a running finalization attempt for a native run. 2. Commit an accepted run result, a finalized workspace reference and a later successful finalization receipt for the same company, task, run and workspace. 3. Stop all application and worker deployments. Run offline enrollment. 4. The old attempt blocks enrollment. This change records it as failed with receipt provenance before workspace import. ## What Changed - Add a bounded offline reconciliation service for local workspace sources that still need migration. - Require the stopped-writer acknowledgement, maintenance switch and existing active-execution gate. - Lock the finalizer ownership, run, coordinator, workspace, accepted result and operation receipts. Hold busy or incomplete records. - Fail only the obsolete attempt. Preserve its metadata and add the superseding receipt, result and workspace digests. Persist the activity audit in the same transaction. - Run reconciliation before workspace import. Keep unrelated, uncertain and saved-work obligations held. - Add real PostgreSQL tests for receipt proof, company scope, ownership races, queued work, audit rollback and repeated maintenance. ## Verification - `cd server && node ../node_modules/vitest/vitest.mjs run src/services/offline-workspace-operation-reconciliation.test.ts src/services/remote-workspace-migration.test.ts src/services/remote-workspace-migration-concurrency.test.ts src/maintenance/ephemeral-migration-service.test.ts`: 135 tests passed across four suites before the final integration test. The final focused suite passed all 33 cases, including a real local-files-to-object-storage workspace import after reconciliation. - The local server TypeScript check found one unrelated stale Runner declaration: `codexUsageBaselineForNextRun` is absent from the reused build output. It found no error in the changed files. CI must verify fresh build outputs. - Fresh GitHub Actions CI passed on `d302fbddd430d42346611302ad7307cf809c3449`, including typecheck, build, Runner, server and browser gates. Greptile gave 5/5 on the same head with no findings. - The external Paperclip Review check remains queued while it waits for its assigned reviewer agent. Do not merge until that check passes. - This PR does not claim live-data or deployed migration acceptance. A legacy preparation attempt without a superseding receipt remains held. ## Risks - The Cloud caller must establish the stop-all-deployments fence. An environment switch cannot prove that old writers stopped. - Only a succeeded native run with an unowned committed finalizer, an accepted result, a finalized matching workspace and a later completed successful receipt can pass. - Future controller leases, retained workspace owners, busy rows, uncertain work and independent save-work obligations stay held. - Expired legacy controller lease fields remain unchanged. The service changes no run, task, lease, recovery or source authority. ## Model Used OpenAI Codex, exact model ID `gpt-6.1-sol`. Context window size is not exposed. Used code reasoning, repository tools and local code execution. ## 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 (e.g. `docs/...`, `fix/...`) 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
a0250284e3
commit
fec0b57659
3 files changed
+374
No files matched your search
@@ -12,6 +12,7 @@ import { skillInstallationMaintenance } from "../services/skill-installation-mai
|
||||
import { skillPackageMaintenance } from "../services/skill-package-maintenance.js";
|
||||
import { hasUnsettledLocalInstructionCopies } from "../services/agent-instruction-working-copies.js";
|
||||
import { providerTraceStore } from "../services/provider-trace-store.js";
|
||||
import { reconcileOfflineWorkspaceOperations } from "../services/offline-workspace-operation-reconciliation.js";
|
||||
import { migrateWorkspaceTrees } from "../services/remote-workspace-migration-batch.js";
|
||||
import { migrateNativeHarnessBackups } from "../services/native-runtime/native-harness-backup-migration.js";
|
||||
import { migrateNativeControllers } from "../services/native-runtime/native-controller-migration.js";
|
||||
@@ -47,6 +48,10 @@ export async function migrateEphemeralServerData(db: Db, input: {
|
||||
counts.nativeControllersPreservedFailed = controllers.preservedFailed;
|
||||
phase("native_harness_backups");
|
||||
counts.nativeHarnessBackups = await migrateNativeHarnessBackups(db, { storage, writersStopped: true, signal });
|
||||
phase("workspace_operation_receipts");
|
||||
const operations = await reconcileOfflineWorkspaceOperations(db, { writersStopped: true, signal });
|
||||
counts.workspaceOperationsReconciled = operations.reconciled;
|
||||
counts.workspaceOperationsHeld = operations.held.length;
|
||||
phase("workspaces");
|
||||
counts.workspaces = await migrateWorkspaceTrees(db, { storage, signal, phase });
|
||||
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, agents, closeRegisteredClients, companies, completionContracts, createDb, executionWorkspaces,
|
||||
heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { reconcileOfflineWorkspaceOperations } from "./offline-workspace-operation-reconciliation.js";
|
||||
import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace } from "./remote-workspace-migration.js";
|
||||
import { createStorageService } from "../storage/service.js";
|
||||
import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("offline workspace operation receipts", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("offline-operation-receipts-");
|
||||
db = createDb(database.connectionString);
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
}, 60_000);
|
||||
afterAll(async () => {
|
||||
if (database) await closeRegisteredClients(database.connectionString);
|
||||
await database?.cleanup(); vi.unstubAllEnvs();
|
||||
});
|
||||
async function fixture() {
|
||||
const companyId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), workspaceId = randomUUID();
|
||||
const contractId = randomUUID(), resultId = randomUUID(), operationId = randomUUID(), receiptId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Historical receipts", issuePrefix: `O${companyId.slice(0, 6)}` });
|
||||
const [agent] = await db.insert(agents).values({ companyId, name: "Worker", adapterType: "paperclip_runner" }).returning();
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Saved work", status: "done" });
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId,
|
||||
mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", name: "Retained files", cwd: "/retained" });
|
||||
await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1,
|
||||
schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter",
|
||||
incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: {}, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" });
|
||||
const profile = { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "finalized", workspaceId,
|
||||
leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/work", descriptorSha256: "a".repeat(64),
|
||||
baselineSha256: "b".repeat(64), finalHostSha256: "c".repeat(64), resourceDisposition: "keep_running" } };
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId: agent.id, nativeIssueId: issueId, runtimeMode: "native",
|
||||
status: "succeeded", completionContractId: contractId, runnerProfileJson: profile, finishedAt: new Date() });
|
||||
await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId,
|
||||
serverFingerprint: resultId, schemaStatus: "accepted", resultJson: {}, canonicalSha256: resultId });
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, resultId, phase: "committed" });
|
||||
const metadata = { owningService: "native_workspace_finalizer", original: "keep" };
|
||||
await db.insert(workspaceOperations).values({ id: operationId, companyId, executionWorkspaceId: workspaceId,
|
||||
heartbeatRunId: runId, issueId, phase: "workspace_finalize", status: "running", metadata,
|
||||
createdAt: new Date("2026-01-01"), startedAt: new Date("2026-01-01") });
|
||||
await db.insert(workspaceOperations).values({ id: receiptId, companyId, executionWorkspaceId: workspaceId,
|
||||
heartbeatRunId: runId, issueId, phase: "workspace_finalize", status: "succeeded",
|
||||
createdAt: new Date("2026-01-02"), startedAt: new Date("2026-01-02"), finishedAt: new Date("2026-01-02") });
|
||||
return { companyId, issueId, runId, workspaceId, resultId, operationId, receiptId, metadata, profile };
|
||||
}
|
||||
const reconcile = (companyId: string) => reconcileOfflineWorkspaceOperations(db,
|
||||
{ companyId, writersStopped: true, signal: new AbortController().signal });
|
||||
const operation = async (id: string) => (await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, id)))[0]!;
|
||||
|
||||
it("fails only the superseded attempt and atomically retains receipt provenance and audit", async () => {
|
||||
const f = await fixture();
|
||||
const runBefore = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const coordinatorBefore = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const receiptBefore = await operation(f.receiptId), attemptBefore = await operation(f.operationId);
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ inspected: 1, reconciled: 1, held: [] });
|
||||
expect(await operation(f.operationId)).toMatchObject({ status: "failed", metadata: { ...attemptBefore.metadata,
|
||||
offlineReconciliation: { reason: "superseded_workspace_finalization", supersedingOperationId: f.receiptId,
|
||||
resultId: f.resultId, finalHostSha256: "c".repeat(64), previousStatus: "running" } } });
|
||||
expect(await operation(f.receiptId)).toEqual(receiptBefore);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId))).toEqual(runBefore);
|
||||
expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(coordinatorBefore);
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.operationId)))
|
||||
.toEqual([expect.objectContaining({ action: "workspace_operation.reconciled", details: expect.objectContaining({ supersedingOperationId: f.receiptId }) })]);
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 0 });
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.operationId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("unblocks the real workspace importer only after reconciling the obsolete attempt", async () => {
|
||||
const f = await fixture(), root = await mkdtemp(path.join(os.tmpdir(), "offline-receipt-import-"));
|
||||
try {
|
||||
const cwd = path.join(root, "source"); await mkdir(cwd);
|
||||
await writeFile(path.join(cwd, "saved.txt"), "Retained local work\n");
|
||||
await db.update(executionWorkspaces).set({ cwd, providerRef: cwd }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
await db.update(issues).set({ executionWorkspaceId: f.workspaceId }).where(eq(issues.id, f.issueId));
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")));
|
||||
const request = { db, storage, companyId: f.companyId, workspaceId: f.workspaceId, apply: true, writersStopped: true };
|
||||
expect((await inventoryRemoteWorkspaceMigration(db, request)).reasons).toContain("workspace_operation_unresolved");
|
||||
await expect(migrateRemoteWorkspace(request)).rejects.toThrow("workspace_operation_unresolved");
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 1 });
|
||||
await expect(migrateRemoteWorkspace(request)).resolves.toMatchObject({ applied: true, eligible: true });
|
||||
expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("Retained local work\n");
|
||||
expect((await operation(f.receiptId)).status).toBe("succeeded");
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
expect(workspace.metadata?.remoteWorkspace).toMatchObject({ schema: "paperclip.remote-workspace.v1", snapshotId: expect.any(String) });
|
||||
} finally { await rm(root, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
it.each(["missing_receipt", "failed_receipt", "earlier_receipt", "unfinished_receipt", "wrong_workspace", "wrong_run",
|
||||
"unexported", "missing_hash", "wrong_reference", "workspace_owner", "coordinator_owner", "uncommitted",
|
||||
"missing_result", "rejected_result", "controller_lease", "uncertain", "save_work", "legacy_prepare"])("holds incomplete proof: %s", async kind => {
|
||||
const f = await fixture();
|
||||
if (kind === "missing_receipt") await db.delete(workspaceOperations).where(eq(workspaceOperations.id, f.receiptId));
|
||||
if (kind === "failed_receipt") await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.id, f.receiptId));
|
||||
if (kind === "earlier_receipt") await db.update(workspaceOperations).set({ createdAt: new Date("2025-12-31") }).where(eq(workspaceOperations.id, f.receiptId));
|
||||
if (kind === "unfinished_receipt") await db.update(workspaceOperations).set({ finishedAt: null }).where(eq(workspaceOperations.id, f.receiptId));
|
||||
if (kind === "wrong_workspace") await db.update(workspaceOperations).set({ executionWorkspaceId: null }).where(eq(workspaceOperations.id, f.receiptId));
|
||||
if (kind === "wrong_run") await db.update(workspaceOperations).set({ heartbeatRunId: null }).where(eq(workspaceOperations.id, f.receiptId));
|
||||
if (["unexported", "missing_hash", "wrong_reference"].includes(kind)) await db.update(heartbeatRuns).set({ runnerProfileJson: {
|
||||
...f.profile, nativeWorkspaceSync: { ...f.profile.nativeWorkspaceSync,
|
||||
...(kind === "unexported" ? { state: "prepared" } : kind === "missing_hash" ? { finalHostSha256: null } : { workspaceId: randomUUID() }) },
|
||||
} }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (kind === "workspace_owner") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile,
|
||||
nativeWorkspaceFinalizationOwner: { token: "retained-owner" } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (kind === "controller_lease") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
if (kind === "coordinator_owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "retained-owner" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (kind === "uncommitted") await db.update(nativeRunFinalizations).set({ phase: "result_accepted" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (kind === "missing_result") await db.update(nativeRunFinalizations).set({ resultId: null }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (kind === "rejected_result") await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, f.resultId));
|
||||
if (kind === "uncertain" || kind === "save_work") await db.update(workspaceOperations).set({ metadata: { ...f.metadata,
|
||||
...(kind === "uncertain" ? { uncertain: true } : { saveWork: { unknown: true, activeIds: [randomUUID()] } }) },
|
||||
}).where(eq(workspaceOperations.id, f.operationId));
|
||||
if (kind === "legacy_prepare") {
|
||||
await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.update(workspaceOperations).set({ phase: "worktree_prepare", metadata: { refreshedUnstartedWorktree: true } })
|
||||
.where(eq(workspaceOperations.id, f.operationId));
|
||||
}
|
||||
const before = await operation(f.operationId);
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ inspected: 1, reconciled: 0, held: [{ operationId: f.operationId }] });
|
||||
expect(await operation(f.operationId)).toEqual(before);
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.operationId))).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("requires the explicit stopped-writer acknowledgement as well as the maintenance switch", async () => {
|
||||
const f = await fixture();
|
||||
await expect(reconcileOfflineWorkspaceOperations(db, { companyId: f.companyId, writersStopped: false,
|
||||
signal: new AbortController().signal })).rejects.toThrow("offline_maintenance_required");
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "false");
|
||||
try { await expect(reconcile(f.companyId)).rejects.toThrow("offline_maintenance_required"); }
|
||||
finally { vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); }
|
||||
expect((await operation(f.operationId)).status).toBe("running");
|
||||
});
|
||||
|
||||
it.each(["running", "queued"])("retains active %s execution as an offline blocker", async status => {
|
||||
const f = await fixture();
|
||||
await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, f.runId));
|
||||
try { await expect(reconcile(f.companyId)).rejects.toThrow("active_execution_requires_drain"); }
|
||||
finally { await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); }
|
||||
expect((await operation(f.operationId)).status).toBe("running");
|
||||
});
|
||||
|
||||
it("keeps a never-started queued turn and only changes the requested company", async () => {
|
||||
const f = await fixture(), other = await fixture();
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [queued] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: run.agentId, status: "queued" }).returning();
|
||||
const otherBefore = await operation(other.operationId);
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 1 });
|
||||
expect(await operation(other.operationId)).toEqual(otherBefore);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queued.id)))[0]).toEqual(queued);
|
||||
});
|
||||
|
||||
it("retains an expired legacy controller lease while reconciling its obsolete receipt", async () => {
|
||||
const f = await fixture();
|
||||
await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date("2026-01-01") }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const before = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 1 });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId))).toEqual(before);
|
||||
});
|
||||
|
||||
it("does not change historical operations after remote workspace authority is published", async () => {
|
||||
const f = await fixture();
|
||||
await db.update(executionWorkspaces).set({ metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1" } } })
|
||||
.where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const before = await operation(f.operationId);
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ inspected: 0, reconciled: 0 });
|
||||
expect(await operation(f.operationId)).toEqual(before);
|
||||
});
|
||||
|
||||
it.each(["coordinator", "run", "workspace", "receipt", "attempt"])("holds a busy %s row and succeeds after the actual writer releases it", async kind => {
|
||||
const f = await fixture();
|
||||
let release!: () => void, ready!: () => void;
|
||||
const barrier = new Promise<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(resolve => { ready = resolve; });
|
||||
const writer = db.transaction(async tx => {
|
||||
if (kind === "coordinator") await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)).for("update");
|
||||
else if (kind === "workspace") await tx.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)).for("update");
|
||||
else if (kind === "run") await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)).for("update");
|
||||
else await tx.select().from(workspaceOperations).where(eq(workspaceOperations.id, kind === "receipt" ? f.receiptId : f.operationId)).for("update");
|
||||
ready(); await barrier;
|
||||
});
|
||||
await locked;
|
||||
try {
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 0,
|
||||
held: [{ operationId: f.operationId, reason: "execution_rows_busy" }] });
|
||||
expect((await operation(f.operationId)).status).toBe("running");
|
||||
} finally { release(); await writer; }
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 1 });
|
||||
});
|
||||
|
||||
it("does not steal the finalizer's advisory ownership", async () => {
|
||||
const f = await fixture();
|
||||
let release!: () => void, ready!: () => void;
|
||||
const barrier = new Promise<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(resolve => { ready = resolve; });
|
||||
const writer = db.transaction(async tx => {
|
||||
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`native-workspace-finalization:${f.companyId}:${f.runId}`}, 0))`);
|
||||
ready(); await barrier;
|
||||
});
|
||||
await locked;
|
||||
try { expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 0,
|
||||
held: [{ operationId: f.operationId, reason: "finalization_owner_busy" }] }); }
|
||||
finally { release(); await writer; }
|
||||
expect(await reconcile(f.companyId)).toMatchObject({ reconciled: 1 });
|
||||
});
|
||||
|
||||
it("rolls back the attempt when the activity audit cannot persist", async () => {
|
||||
const f = await fixture();
|
||||
await db.execute(sql.raw(`create function reject_operation_audit() returns trigger language plpgsql as $$
|
||||
begin raise exception 'audit unavailable' using errcode = '23514'; end $$`));
|
||||
await db.execute(sql.raw(`create trigger reject_operation_audit before insert on activity_log for each row
|
||||
when (new.entity_id = '${f.operationId}') execute function reject_operation_audit()`));
|
||||
const before = await operation(f.operationId);
|
||||
try {
|
||||
await expect(reconcile(f.companyId)).rejects.toMatchObject({ cause: { code: "23514" } });
|
||||
expect(await operation(f.operationId)).toEqual(before);
|
||||
} finally {
|
||||
await db.execute(sql`drop trigger reject_operation_audit on activity_log`);
|
||||
await db.execute(sql`drop function reject_operation_audit()`);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,138 @@
|
||||
import { and, asc, desc, eq, gt, inArray, isNotNull, sql } from "drizzle-orm";
|
||||
import { executionWorkspaces, heartbeatRuns, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { persistActivity } from "./activity-log.js";
|
||||
import { offlineWorkspaceMigrationBlockers } from "./remote-workspace-migration-tree.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
|
||||
import { tryLockNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalization-ownership.js";
|
||||
|
||||
type Candidate = typeof workspaceOperations.$inferSelect;
|
||||
type Held = { operationId: string; reason: string };
|
||||
const unresolved = () => inArray(workspaceOperations.status, ["running", "pending"]);
|
||||
const localMigrationSource = () => and(isNotNull(executionWorkspaces.cwd),
|
||||
inArray(executionWorkspaces.providerType, ["local_fs", "git_worktree"]),
|
||||
sql`coalesce(${executionWorkspaces.metadata}->'remoteWorkspace'->>'schema', '') <> 'paperclip.remote-workspace.v1'`);
|
||||
|
||||
async function assertOffline(db: Db, writersStopped: boolean) {
|
||||
// Only the Cloud maintenance caller can establish that all old deployments
|
||||
// stopped. The environment switch alone is never that acknowledgement.
|
||||
if (!writersStopped) throw new Error("offline_maintenance_required");
|
||||
const blockers = await offlineWorkspaceMigrationBlockers(db);
|
||||
if (blockers.length) throw new Error(blockers[0]);
|
||||
}
|
||||
|
||||
function databaseCode(error: unknown): string | undefined {
|
||||
const value = error as { code?: string; cause?: unknown } | null;
|
||||
return value?.code ?? (value?.cause ? databaseCode(value.cause) : undefined);
|
||||
}
|
||||
|
||||
async function reconcileAttempt(db: Db, candidate: Candidate, input: { writersStopped: boolean; signal: AbortSignal }) {
|
||||
const held = (reason: string): Held => ({ operationId: candidate.id, reason });
|
||||
const { heartbeatRunId: runId, issueId, executionWorkspaceId: workspaceId } = candidate;
|
||||
if (candidate.phase !== "workspace_finalize" || !runId || !issueId || !workspaceId)
|
||||
return held("superseding_finalization_receipt_missing");
|
||||
try {
|
||||
return await db.transaction(async tx => {
|
||||
const database = tx as unknown as Db;
|
||||
input.signal.throwIfAborted();
|
||||
await assertOffline(database, input.writersStopped);
|
||||
if (!await tryLockNativeWorkspaceFinalization(database, { companyId: candidate.companyId, runId }))
|
||||
return held("finalization_owner_busy");
|
||||
// Follow the finalizer's coordinator-before-run order. NOWAIT keeps an
|
||||
// unexpected surviving writer visible instead of waiting or stealing it.
|
||||
const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.issueId, issueId),
|
||||
eq(nativeRunFinalizations.runId, runId),
|
||||
)).for("update", { noWait: true });
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, runId), eq(heartbeatRuns.nativeIssueId, issueId),
|
||||
)).for("update", { noWait: true });
|
||||
if (!run || run.runtimeMode !== "native" || run.status !== "succeeded"
|
||||
|| (run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date())
|
||||
|| run.runnerProfileJson?.nativeWorkspaceFinalizationOwner != null || !coordinator || coordinator.phase !== "committed"
|
||||
|| !coordinator.resultId || coordinator.leaseOwner !== null || coordinator.leaseExpiresAt !== null || coordinator.nextAttemptAt !== null)
|
||||
return held("committed_finalization_proof_missing");
|
||||
const reference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync);
|
||||
if (!reference || reference.state !== "finalized" || !reference.finalHostSha256 || reference.workspaceId !== workspaceId)
|
||||
return held("finalized_workspace_proof_missing");
|
||||
const [workspace] = await tx.select({ id: executionWorkspaces.id }).from(executionWorkspaces).where(and(
|
||||
eq(executionWorkspaces.companyId, candidate.companyId), eq(executionWorkspaces.id, workspaceId), localMigrationSource(),
|
||||
)).for("share", { noWait: true });
|
||||
if (!workspace) return held("workspace_authority_changed");
|
||||
const [result] = await tx.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and(
|
||||
eq(nativeRunResults.companyId, candidate.companyId), eq(nativeRunResults.issueId, issueId),
|
||||
eq(nativeRunResults.runId, runId), eq(nativeRunResults.id, coordinator.resultId),
|
||||
eq(nativeRunResults.completionContractId, run.completionContractId!), eq(nativeRunResults.schemaStatus, "accepted"),
|
||||
)).for("share", { noWait: true });
|
||||
if (!result) return held("accepted_result_proof_missing");
|
||||
const [current] = await tx.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.id, candidate.id), eq(workspaceOperations.companyId, candidate.companyId),
|
||||
eq(workspaceOperations.executionWorkspaceId, workspaceId), eq(workspaceOperations.issueId, issueId),
|
||||
eq(workspaceOperations.heartbeatRunId, runId), eq(workspaceOperations.phase, "workspace_finalize"), unresolved(),
|
||||
)).for("update", { noWait: true });
|
||||
if (!current) return held("attempt_changed");
|
||||
// These fields represent independent obligations. A successful later
|
||||
// export does not authorize dropping them or rewriting earlier history.
|
||||
if (current.metadata?.uncertain === true || current.metadata?.uncertain === "true"
|
||||
|| current.metadata?.saveWork != null || current.metadata?.offlineReconciliation != null)
|
||||
return held("independent_operation_obligation");
|
||||
const [receipt] = await tx.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, candidate.companyId), eq(workspaceOperations.executionWorkspaceId, workspaceId),
|
||||
eq(workspaceOperations.issueId, issueId), eq(workspaceOperations.heartbeatRunId, runId),
|
||||
eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded"),
|
||||
isNotNull(workspaceOperations.finishedAt),
|
||||
// Keep PostgreSQL timestamp precision when proving which attempt is later.
|
||||
sql`${workspaceOperations.createdAt} > (select created_at from workspace_operations where id = ${current.id}::uuid)`,
|
||||
)).orderBy(desc(workspaceOperations.createdAt), desc(workspaceOperations.id)).limit(1).for("share", { noWait: true });
|
||||
if (!receipt || receipt.metadata?.uncertain === true || receipt.metadata?.uncertain === "true")
|
||||
return held("superseding_finalization_receipt_missing");
|
||||
input.signal.throwIfAborted();
|
||||
await assertOffline(database, input.writersStopped);
|
||||
const now = new Date();
|
||||
const provenance = { schema: "paperclip.offline-workspace-operation-reconciliation.v1",
|
||||
reason: "superseded_workspace_finalization", reconciledAt: now.toISOString(), previousStatus: current.status,
|
||||
supersedingOperationId: receipt.id, runId, resultId: result.id,
|
||||
descriptorSha256: reference.descriptorSha256, finalHostSha256: reference.finalHostSha256 };
|
||||
await tx.update(workspaceOperations).set({ status: "failed", finishedAt: now, updatedAt: now,
|
||||
metadata: { ...current.metadata, offlineReconciliation: provenance } }).where(and(
|
||||
eq(workspaceOperations.id, current.id), eq(workspaceOperations.companyId, current.companyId), unresolved(),
|
||||
));
|
||||
// No live publication is needed in the stopped maintenance process. The
|
||||
// audit and the failed attempt commit (or roll back) together.
|
||||
await persistActivity(database, { companyId: current.companyId, actorType: "system", actorId: "ephemeral-migration",
|
||||
action: "workspace_operation.reconciled", entityType: "workspace_operation", entityId: current.id,
|
||||
runId, issueId, details: provenance });
|
||||
return null;
|
||||
});
|
||||
} catch (error) {
|
||||
if (databaseCode(error) === "55P03") return held("execution_rows_busy");
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/** Reconcile receipts only after Cloud has stopped all old application and
|
||||
* worker deployments. This never executes workspace work, declares success,
|
||||
* clears execution authority, or retires a recovery/source obligation. */
|
||||
export async function reconcileOfflineWorkspaceOperations(db: Db, input: {
|
||||
writersStopped: boolean; signal: AbortSignal; companyId?: string;
|
||||
}) {
|
||||
await assertOffline(db, input.writersStopped);
|
||||
let after: string | undefined, inspected = 0, reconciled = 0;
|
||||
const held: Held[] = [];
|
||||
for (;;) {
|
||||
input.signal.throwIfAborted();
|
||||
const rows = await db.select({ operation: workspaceOperations }).from(workspaceOperations).innerJoin(executionWorkspaces, and(
|
||||
eq(executionWorkspaces.companyId, workspaceOperations.companyId), eq(executionWorkspaces.id, workspaceOperations.executionWorkspaceId),
|
||||
)).where(and(unresolved(), localMigrationSource(),
|
||||
input.companyId ? eq(workspaceOperations.companyId, input.companyId) : undefined,
|
||||
after ? gt(workspaceOperations.id, after) : undefined,
|
||||
)).orderBy(asc(workspaceOperations.id)).limit(100);
|
||||
if (!rows.length) return { inspected, reconciled, held };
|
||||
for (const { operation } of rows) {
|
||||
input.signal.throwIfAborted();
|
||||
const outcome = await reconcileAttempt(db, operation, input);
|
||||
inspected++;
|
||||
if (outcome) held.push(outcome); else reconciled++;
|
||||
}
|
||||
after = rows.at(-1)!.operation.id;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user