fix: keep prepublication recovery draining (#3000)

convex/publishAttempts.test.ts

@@ -441,6 +441,56 @@ describe("publishAttempts", () => {
     );
   });
 
+  it("terminalizes ambiguous legacy fork slugs instead of retrying finalization", async () => {
+    const ctx = {
+      db: {
+        delete: vi.fn(),
+        get: vi.fn(async () => ({
+          _id: "publishAttempts:ambiguous-fork",
+          kind: "skill",
+          status: "finalizing",
+          skillInsertArgs: {
+            slug: "demo-skill",
+            version: "1.0.0",
+            forkOf: { slug: "shared-upstream" },
+          },
+          followup: {},
+          finalizationClaimId: "finalize:claim",
+        })),
+        insert: vi.fn(),
+        normalizeId: vi.fn(),
+        patch: vi.fn(),
+        query: vi.fn(),
+        replace: vi.fn(),
+        system: {},
+      },
+    };
+    const error =
+      "Uncaught ConvexError: Slug is used by multiple publishers. Use an owner-qualified skill URL.";
+
+    await expect(
+      releaseSkillFinalizationHandler(ctx, {
+        attemptId: "publishAttempts:ambiguous-fork",
+        claimId: "finalize:claim",
+        error,
+      }),
+    ).resolves.toEqual({
+      attemptId: "publishAttempts:ambiguous-fork",
+      status: "failed",
+    });
+
+    expect(ctx.db.patch).toHaveBeenCalledWith(
+      "publishAttempts:ambiguous-fork",
+      expect.objectContaining({
+        status: "failed",
+        checkClaimId: undefined,
+        finalizationClaimId: undefined,
+        finalizationLastError: error,
+        failedAt: expect.any(Number),
+      }),
+    );
+  });
+
   it("terminalizes duplicate package versions while preserving transient retries", async () => {
     const duplicateCtx = {
       db: {

convex/publishAttempts.ts

@@ -82,15 +82,16 @@ function scannerFailureSummary(args: {
   return "Pre-publication scanner failed before returning a verdict.";
 }
 
-function isTerminalVersionConflict(error: string | undefined) {
+function isTerminalFinalizationConflict(error: string | undefined) {
   return (
     typeof error === "string" &&
-    /Version .+ already exists\. Increment the version number and try again\./.test(error)
+    (/Version .+ already exists\. Increment the version number and try again\./.test(error) ||
+      error.includes("Slug is used by multiple publishers. Use an owner-qualified skill URL."))
   );
 }
 
 function releaseFinalizationClaimPatch(error: string | undefined, now: number) {
-  if (!isTerminalVersionConflict(error)) {
+  if (!isTerminalFinalizationConflict(error)) {
     return {
       status: "ready_to_finalize" as const,
       finalizationClaimId: undefined,

scripts/security/run-prepublication-worker.test.ts

@@ -5,6 +5,7 @@ import { join } from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { Id } from "../../convex/_generated/dataModel";
 import {
+  claimBatchDrainedQueue,
   claimPrePublicationBatch,
   configurePrePublicationCodexHome,
   processPrePublicationBatch,
@@ -48,6 +49,13 @@ const attempt = {
 };
 
 describe("pre-publication worker", () => {
+  it("keeps claiming after partial transient claim failures", () => {
+    expect(claimBatchDrainedQueue(0, 0, 6)).toBe(true);
+    expect(claimBatchDrainedQueue(0, 5, 6)).toBe(true);
+    expect(claimBatchDrainedQueue(1, 5, 6)).toBe(false);
+    expect(claimBatchDrainedQueue(0, 6, 6)).toBe(false);
+  });
+
   it("does not clear the Codex home configured by GitHub Actions login", () => {
     const env = {
       CI: "true",
@@ -213,7 +221,10 @@ describe("pre-publication worker", () => {
         .mockResolvedValueOnce(attempt),
     };
 
-    await expect(claimPrePublicationBatch(client, "worker-token", 2)).resolves.toEqual([attempt]);
+    await expect(claimPrePublicationBatch(client, "worker-token", 2)).resolves.toEqual({
+      attempts: [attempt],
+      claimFailures: 1,
+    });
     expect(client.action).toHaveBeenCalledTimes(2);
   });
 

scripts/security/run-prepublication-worker.ts

@@ -646,7 +646,15 @@ export async function claimPrePublicationBatch(
   if (attempts.length === 0 && failures.length > 0) {
     throw new AggregateError(failures, "Pre-publication claims failed without claiming work.");
   }
-  return attempts;
+  return { attempts, claimFailures: failures.length };
+}
+
+export function claimBatchDrainedQueue(
+  claimFailures: number,
+  claimedAttempts: number,
+  claimLimit: number,
+) {
+  return claimFailures === 0 && claimedAttempts < claimLimit;
 }
 
 async function main() {
@@ -676,18 +684,15 @@ async function main() {
     if (totalClaimed > 0 && claimDeadline - Date.now() < CLAIM_WINDOW_SHUTDOWN_BUFFER_MS) break;
     const remainingJobs = maxJobs === undefined ? batchLimit : Math.max(0, maxJobs - totalClaimed);
     if (remainingJobs === 0) break;
-    const attempts = await claimPrePublicationBatch(
-      client,
-      token,
-      Math.min(batchLimit, remainingJobs),
-    );
+    const claimLimit = Math.min(batchLimit, remainingJobs);
+    const { attempts, claimFailures } = await claimPrePublicationBatch(client, token, claimLimit);
     if (attempts.length === 0) break;
     totalClaimed += attempts.length;
     const results = await processPrePublicationBatch(attempts, (attempt) =>
       processPrePublicationAttempt(client, token, attempt),
     );
     totalCompleted += results.filter((result) => result.completed).length;
-    if (attempts.length < Math.min(batchLimit, remainingJobs)) break;
+    if (claimBatchDrainedQueue(claimFailures, attempts.length, claimLimit)) break;
   }
 
   logger.info(