diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 40e50b67da0..1ae2fd4dbd5 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -151,7 +151,7 @@ describe("report-only ownership reconciliation", () => { expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) - .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "workspace_command_operation", "remote_snapshot_save", "execution_workspace_runtime_lease", "remote_workspace_unsaved_files", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]); + .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "workspace_command_operation", "remote_snapshot_save", "execution_workspace_runtime_lease", "remote_workspace_unsaved_files", "remote_snapshot_ledger", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); diff --git a/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts new file mode 100644 index 00000000000..6b532613118 --- /dev/null +++ b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts @@ -0,0 +1,143 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { + agents, + companies, + createDb, + executionWorkspaceSnapshots, + executionWorkspaces, + heartbeatRuns, + workspaceOperations, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { + createOwnershipReportState, + runOwnershipReconciliationPass, + type OwnershipReport, +} from "../services/recovery/ownership-invariant-report.js"; +import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); +const minutesAgo = (minutes: number) => new Date(Date.now() - minutes * 60_000); +const kindReport = (report: OwnershipReport, kind: string) => report.kinds.find(entry => entry.kind === kind)!; +const ledgerViolations = (report: OwnershipReport) => + report.violations.filter(violation => violation.kind === "remote_snapshot_ledger"); +const quiesced = { schema: "paperclip.remote-workspace-save-work.v1", activeIds: [], unknown: false, + owners: {}, quiescedAt: minutesAgo(30).toISOString() }; + +describe("remote snapshot ledger settlement", () => { + let database: Awaited> | undefined; + let db: ReturnType; + let companyId: string; + let agentId: string; + + beforeAll(async () => { + if (!external) database = await startEmbeddedPostgresTestDatabase("remote-snapshot-ledger-"); + db = createDb(external ?? database!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 120_000); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); + }); + + beforeEach(async () => { + await db.execute(sql`truncate execution_workspace_snapshots, activity_log, workspace_operations, execution_workspaces, + heartbeat_runs, agents, companies cascade`); + companyId = randomUUID(); agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Ledger fixture", issuePrefix: "LED" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", role: "engineer", adapterType: "codex_local" }); + }); + + async function run(status: string) { + const id = randomUUID(); + await db.insert(heartbeatRuns).values({ id, companyId, agentId, status, invocationSource: "manual", + ...(status === "running" ? {} : { finishedAt: minutesAgo(2), errorCode: "process_lost" }) }); + return id; + } + + async function workspace(authority: { epoch: number; writerRunId: string | null; dirty: boolean }) { + const id = randomUUID(); + await db.insert(executionWorkspaces).values({ id, companyId, mode: "isolated_workspace", strategyType: "test", name: id, + metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1", snapshotId: null, + source: { kind: "git", repoUrl: "https://example.invalid/repo.git", ref: null }, location: null, ...authority } } }); + return id; + } + + async function save(input: { workspaceId: string; runId: string; epoch?: number; status?: string; + saveWork?: Record; snapshot?: Partial; lastFailure?: boolean }) { + const operationId = randomUUID(), snapshotId = randomUUID(), epoch = input.epoch ?? 1; + await db.insert(workspaceOperations).values({ id: operationId, companyId, executionWorkspaceId: input.workspaceId, + heartbeatRunId: input.runId, phase: "remote_workspace_snapshot", status: input.status ?? "failed", + metadata: { epoch, boundary: "initial", generation: 1, saveWork: input.saveWork ?? quiesced, + exportCleanup: { state: "pending" }, + ...(input.lastFailure === false ? {} : { lastFailure: { code: "remote_workspace_save_abandoned", at: minutesAgo(20).toISOString() } }) } }); + await db.insert(executionWorkspaceSnapshots).values({ id: snapshotId, companyId, executionWorkspaceId: input.workspaceId, + sourceOperationId: operationId, sourceRunId: input.runId, generation: 1, epoch, + deleteAfter: new Date(Date.now() + 7 * 86400_000), createdAt: minutesAgo(30), updatedAt: minutesAgo(30), ...input.snapshot }); + return { operationId, snapshotId }; + } + + const ledger = async (id: string) => + (await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, id)))[0]!; + + it("retires a pending row an older build left behind once its save was abandoned and its epoch ended", async () => { + const failed = await run("failed"); + const workspaceId = await workspace({ epoch: 3, writerRunId: null, dirty: false }); + const { snapshotId } = await save({ workspaceId, runId: failed }); + + const state = createOwnershipReportState(); + const first = await runOwnershipReconciliationPass(db, { state }); + const second = await runOwnershipReconciliationPass(db, { state }); + + expect(kindReport(first, "remote_snapshot_ledger")).toMatchObject({ inProgress: 1, settled: 1 }); + expect(ledgerViolations(first)).toEqual([]); + expect(ledgerViolations(second)).toEqual([]); + expect(kindReport(second, "remote_snapshot_ledger").inProgress).toBe(0); + const row = await ledger(snapshotId); + expect(row).toMatchObject({ status: "deleted", deleteAfter: null, lastError: "abandoned_remote_snapshot_upload" }); + expect(row.metadata).toMatchObject({ abandonedUpload: { schema: "paperclip.remote-snapshot-ledger-abandonment.v1", + runId: failed, epoch: 1, workspaceEpoch: 3, uploadAttempted: false, previousStatus: "pending" } }); + }); + + it("leaves objects of an attempted upload to retention, and never retires an uploaded, committed or still-owned row", async () => { + const failed = await run("failed"), live = await run("running"), writer = await run("failed"); + const ended = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const held = await workspace({ epoch: 1, writerRunId: writer, dirty: true }); + const objectKey = `${companyId}/workspace-snapshots/${ended}/x/a.pcws`; + const rows = { + attempted: await save({ workspaceId: ended, runId: failed, snapshot: { metadata: { uploads: [{ objectKey }] } } }), + uploaded: await save({ workspaceId: await workspace({ epoch: 2, writerRunId: null, dirty: false }), runId: failed, + snapshot: { status: "uploaded", objectKey } }), + liveRun: await save({ workspaceId: await workspace({ epoch: 2, writerRunId: null, dirty: false }), runId: live }), + reopened: await save({ workspaceId: await workspace({ epoch: 2, writerRunId: null, dirty: false }), runId: failed, + status: "running", lastFailure: false }), + currentWriter: await save({ workspaceId: held, runId: writer }), + }; + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_ledger") }); + + expect(await ledger(rows.attempted.snapshotId)).toMatchObject({ status: "delete_requested", + metadata: { uploads: [{ objectKey }], abandonedUpload: { uploadAttempted: true } } }); + expect((await ledger(rows.attempted.snapshotId)).deleteAfter).not.toBeNull(); + expect((await ledger(rows.uploaded.snapshotId)).status).toBe("uploaded"); + expect((await ledger(rows.liveRun.snapshotId)).status).toBe("pending"); + expect((await ledger(rows.reopened.snapshotId)).status).toBe("pending"); + expect((await ledger(rows.currentWriter.snapshotId)).status).toBe("pending"); + }); + + it("retires the ledger row in the same settle that abandons a save a restart stopped mid-upload", async () => { + const failed = await run("failed"); + const workspaceId = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const { operationId, snapshotId } = await save({ workspaceId, runId: failed, status: "running", lastFailure: false }); + + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_save") }); + + expect(kindReport(report, "remote_snapshot_save")).toMatchObject({ settled: 1 }); + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, operationId)); + expect(operation!.metadata).toMatchObject({ lastFailure: { code: "remote_workspace_save_abandoned" } }); + expect(await ledger(snapshotId)).toMatchObject({ status: "deleted", + metadata: { abandonedUpload: { runId: failed, operationId } } }); + }); +}); diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index d576c4f48ea..fcf8b9195c3 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -39,6 +39,7 @@ import { import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; import { settleAbandonedRemoteSnapshotSave, settleAbandonedWorkspaceCommandOperation, settleAbandonedWorkspaceFinalizeAttempt } from "./workspace-finalize-attempt-settlement.js"; import { settlePreparationOnlyRemoteWorkspaceEpoch } from "./remote-workspace-unsaved-files-settlement.js"; +import { settleAbandonedRemoteSnapshotLedger } from "./remote-snapshot-ledger-settlement.js"; import { LOCAL_WORKSPACE_COMMAND_PHASES, runtimeControlOperationOwnerStamp, settleRuntimeControlOperation } from "../workspace-operations.js"; import { copybackOwnerHoldOwner, settleCopybackOwnerHold } from "./copyback-owner-hold.js"; import { applyObservedFenceProviderProof, observedWorkspaceFenceOwner, settleObservedWorkspaceFence } from "./observed-workspace-fence.js"; @@ -946,7 +947,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ }, { id: "remote_snapshot_ledger", title: "Remote snapshot ledger", obligation: "workspace_preservation", - storage: "database", settle: null, invariantChecks: [2], + storage: "database", settle: settleAbandonedRemoteSnapshotLedger, invariantChecks: [2], listInProgress: async (db, limit, offset) => { const rows = await db.select({ id: executionWorkspaceSnapshots.id, companyId: executionWorkspaceSnapshots.companyId, sourceRunId: executionWorkspaceSnapshots.sourceRunId, updatedAt: executionWorkspaceSnapshots.updatedAt }) diff --git a/server/src/services/recovery/remote-snapshot-ledger-settlement.ts b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts new file mode 100644 index 00000000000..f9e8b26bd27 --- /dev/null +++ b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts @@ -0,0 +1,93 @@ +import { and, eq, isNull, sql } from "drizzle-orm"; +import { executionWorkspaceSnapshots, executionWorkspaces, heartbeatRuns, workspaceOperations, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; +import type { OwnerProof } from "../owner-proof.js"; +import { readSaveWork } from "../remote-workspaces.js"; +import { notifyRemoteWorkspaceSnapshots } from "../remote-workspace-snapshot-retention.js"; +import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; + +export const ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON = "abandoned_remote_snapshot_upload"; +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; + +type Snapshot = typeof executionWorkspaceSnapshots.$inferSelect; + +export async function retireEndedRemoteSnapshotLedgerRow(tx: Db, input: { + companyId: string; snapshotId: string; now: Date; proof: string; +}): Promise<{ retired: true; snapshot: Snapshot; status: "deleted" | "delete_requested" } | { retired: false; reason: string }> { + const [snapshot] = await tx.select().from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, input.companyId), eq(executionWorkspaceSnapshots.id, input.snapshotId))); + if (!snapshot || snapshot.status !== "pending") return { retired: false, reason: "ledger_no_longer_pending" }; + if (snapshot.objectKey !== null) return { retired: false, reason: "upload_published" }; + if (!snapshot.sourceRunId) return { retired: false, reason: "owner_run_missing" }; + const [operation] = await tx.select().from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, snapshot.companyId), eq(workspaceOperations.id, snapshot.sourceOperationId))).for("update"); + if (operation) { + if (operation.phase !== "remote_workspace_snapshot") return { retired: false, reason: "not_a_workspace_save" }; + if (operation.status === "running" || operation.status === "pending") return { retired: false, reason: "save_in_progress" }; + const work = readSaveWork(operation.metadata); + if (work && (work.unknown || work.activeIds.length)) return { retired: false, reason: "save_work_unsettled" }; + } + const [run] = await tx.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, snapshot.companyId), eq(heartbeatRuns.id, snapshot.sourceRunId))); + if (run && !TERMINAL_RUN_STATUSES.includes(run.status)) return { retired: false, reason: "owner_run_active" }; + const [workspace] = await tx.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, snapshot.companyId), eq(executionWorkspaces.id, snapshot.executionWorkspaceId))); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (authority && authority.epoch === snapshot.epoch && authority.writerRunId === snapshot.sourceRunId) { + return { retired: false, reason: "writer_epoch_current" }; + } + const uploads = snapshot.metadata?.uploads; + const attempted = Array.isArray(uploads) ? uploads.length > 0 : uploads !== undefined && uploads !== null; + const status = attempted ? "delete_requested" as const : "deleted" as const; + const provenance = { schema: "paperclip.remote-snapshot-ledger-abandonment.v1", reason: ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON, + proof: input.proof, previousStatus: snapshot.status, runId: snapshot.sourceRunId, operationId: snapshot.sourceOperationId, + epoch: snapshot.epoch, workspaceEpoch: authority?.epoch ?? null, uploadAttempted: attempted, reconciledAt: input.now.toISOString() }; + await notifyRemoteWorkspaceSnapshots(tx, `workspace:${snapshot.executionWorkspaceId}`); + const [row] = await tx.update(executionWorkspaceSnapshots).set({ status, + deleteAfter: status === "deleted" ? null : snapshot.deleteAfter, lastError: ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON, + metadata: { ...snapshot.metadata, abandonedUpload: provenance }, updatedAt: input.now }) + .where(and(eq(executionWorkspaceSnapshots.id, snapshot.id), eq(executionWorkspaceSnapshots.status, "pending"), + isNull(executionWorkspaceSnapshots.objectKey), + sql`${executionWorkspaceSnapshots.metadata} is not distinct from ${snapshot.metadata === null ? null : JSON.stringify(snapshot.metadata)}::jsonb`)) + .returning(); + if (!row) return { retired: false, reason: "ledger_no_longer_pending" }; + return { retired: true, snapshot: row, status }; +} + +export async function settleAbandonedRemoteSnapshotLedger(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "run" || !record.companyId) return { settled: false, reason: "owner_not_provable" }; + const companyId = record.companyId; + let publication: ActivityPublication | undefined; + const outcome = await db.transaction(async (transaction): Promise => { + const tx = transaction as unknown as Db; + const result = await retireEndedRemoteSnapshotLedgerRow(tx, { companyId, snapshotId: record.recordId, + now: context.now, proof: context.proof.reason }); + if (!result.retired) { + return { settled: false, reason: result.reason, + ...(["upload_published", "not_a_workspace_save"].includes(result.reason) ? { watch: false } : {}) }; + } + const activity = await persistActivity(tx, { + companyId, actorType: "system", actorId: "ownership-reconciler", + action: "workspace_snapshot.reconciled", entityType: "execution_workspace", entityId: result.snapshot.executionWorkspaceId, + runId: result.snapshot.sourceRunId, details: { snapshotId: result.snapshot.id, status: result.status, + ...(result.snapshot.metadata?.abandonedUpload as Record) }, + }); + publication = activity.publication; + return { settled: true, reason: context.proof.reason }; + }); + if (outcome.settled && publication) publishActivity(publication); + return outcome; +} + +export async function retireAbandonedSaveLedgerRows(tx: Db, input: { + companyId: string; operationId: string; now: Date; proof: string; +}) { + const rows = await tx.select({ id: executionWorkspaceSnapshots.id }).from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, input.companyId), eq(executionWorkspaceSnapshots.sourceOperationId, input.operationId), + eq(executionWorkspaceSnapshots.status, "pending"))); + for (const row of rows) { + await retireEndedRemoteSnapshotLedgerRow(tx, { companyId: input.companyId, snapshotId: row.id, now: input.now, proof: input.proof }); + } +} diff --git a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts index 815102ac22b..5fcb5d945fd 100644 --- a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts +++ b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts @@ -9,6 +9,7 @@ import { notifyTerminalWorkspaceWork } from "../terminal-workspace-work.js"; import { RETAINED_SOURCE_FALLBACK_CHECK_KEY } from "../workspace-recovery-fallback-discovery.js"; import { LOCAL_WORKSPACE_COMMAND_PHASES, isOwnerStampedWorkspaceOperationLive } from "../workspace-operations.js"; import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; +import { retireAbandonedSaveLedgerRows } from "./remote-snapshot-ledger-settlement.js"; export const ABANDONED_WORKSPACE_FINALIZE_REASON = "abandoned_workspace_finalization"; export const ABANDONED_WORKSPACE_COMMAND_REASON = "abandoned_workspace_command"; @@ -164,6 +165,7 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre }).where(and(eq(workspaceOperations.id, operation.id), eq(workspaceOperations.status, operation.status))) .returning({ id: workspaceOperations.id }); if (!row) return { settled: false, reason: "attempt_no_longer_in_progress" }; + await retireAbandonedSaveLedgerRows(tx, { companyId: operation.companyId, operationId: operation.id, now, proof: context.proof.reason }); if (operation.executionWorkspaceId) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); await tx.update(issueRecoveryActions).set({ diff --git a/server/src/services/remote-workspace-snapshot-retention.test.ts b/server/src/services/remote-workspace-snapshot-retention.test.ts index 079d5a0206f..60986fff325 100644 --- a/server/src/services/remote-workspace-snapshot-retention.test.ts +++ b/server/src/services/remote-workspace-snapshot-retention.test.ts @@ -125,6 +125,23 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () => return { row: row!, companyId, workspaceId, issueId: issue!.id, operationId, keys }; } const service = () => remoteWorkspaceSnapshotRetentionService(db, { storage, now: () => now }); + it("deletes the attempted uploads of an abandoned save once its ledger row is retired, and keeps a failed save's retry pin", async () => { + const retired = await seed("pending"), held = await seed("pending"); + for (const f of [retired, held]) { + await db.insert(workspaceOperations).values({ id: f.operationId, companyId: f.companyId, executionWorkspaceId: f.workspaceId, + phase: "remote_workspace_snapshot", status: "failed", + metadata: { epoch: 1, generation: 1, lastFailure: { code: "remote_workspace_save_abandoned" } } }); + } + await db.update(executionWorkspaceSnapshots).set({ status: "delete_requested", objectKey: null, + metadata: { ...retired.row.metadata, abandonedUpload: { operationId: retired.operationId } } }) + .where(eq(executionWorkspaceSnapshots.id, retired.row.id)); + await db.update(executionWorkspaceSnapshots).set({ objectKey: null }).where(eq(executionWorkspaceSnapshots.id, held.row.id)); + expect(await service().tick()).toMatchObject({ deleted: 1, pinned: 1, failed: 0 }); + expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, retired.row.id)))[0]!.status).toBe("deleted"); + for (const key of retired.keys) expect((await storage.headObject(retired.companyId, key)).exists).toBe(false); + expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, held.row.id)))[0]!.status).toBe("pending"); + for (const key of held.keys) expect((await storage.headObject(held.companyId, key)).exists).toBe(true); + }); it("deletes every recorded attempt and winning object after grace", async () => { const f = await seed(); const owner = service(); expect(await owner.tick()).toMatchObject({ deleted: 1, failed: 0 }); diff --git a/server/src/services/remote-workspace-snapshot-retention.ts b/server/src/services/remote-workspace-snapshot-retention.ts index d4722c07d01..2003dce614e 100644 --- a/server/src/services/remote-workspace-snapshot-retention.ts +++ b/server/src/services/remote-workspace-snapshot-retention.ts @@ -17,6 +17,12 @@ export function notifyRemoteWorkspaceSnapshots(transaction: object, hint?: strin return notifyDeliveryWork(transaction, DELIVERY_QUEUES.remoteWorkspaceSnapshots, hint); } +function abandonedUploadOperationId(row: Snapshot) { + if (row.status !== "delete_requested") return null; + const abandoned = row.metadata?.abandonedUpload as { operationId?: unknown } | undefined; + return typeof abandoned?.operationId === "string" && abandoned.operationId === row.sourceOperationId ? abandoned.operationId : null; +} + async function pinned(db: ReadDb, row: Snapshot) { const [workspace] = await db.select({ id: executionWorkspaces.id, sourceIssueId: executionWorkspaces.sourceIssueId, metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( @@ -37,7 +43,8 @@ async function pinned(db: ReadDb, row: Snapshot) { // A failed save keeps its snapshot pinned for its retry, as while it ran, // until the final save retires it. sql`(${workspaceOperations.status} = 'failed' and ${workspaceOperations.metadata} ? 'lastFailure' - and ${workspaceOperations.metadata}->'retiredSave' is null)`, + and ${workspaceOperations.metadata}->'retiredSave' is null + and ${workspaceOperations.id}::text is distinct from ${abandonedUploadOperationId(row)})`, sql`(${workspaceOperations.metadata}->'saveWork'->>'schema' = 'paperclip.remote-workspace-save-work.v1' and (${workspaceOperations.metadata}#>'{saveWork,unknown}' is distinct from 'false'::jsonb or case when jsonb_typeof(${workspaceOperations.metadata}#>'{saveWork,activeIds}') = 'array'