fix: reconcile skills.sh activation timeouts (#3352)

convex/schema.ts

@@ -3323,6 +3323,9 @@ const skillsShMirrorRuns = defineTable({
   }),
   actor: v.string(),
   reason: v.string(),
+  activatedTrendingRunId: v.optional(v.id("skillsShMirrorRuns")),
+  activationSnapshotId: v.optional(v.string()),
+  activatedAt: v.optional(v.number()),
   startedAt: v.number(),
   completedAt: v.optional(v.number()),
   updatedAt: v.number(),

convex/skillsShMirror.ts

@@ -967,6 +967,9 @@ type SummarizableMirrorRun = Pick<
   | "offset"
   | "counts"
   | "operations"
+  | "activatedTrendingRunId"
+  | "activationSnapshotId"
+  | "activatedAt"
   | "startedAt"
   | "completedAt"
   | "updatedAt"
@@ -988,6 +991,9 @@ function summarizeRun(run: SummarizableMirrorRun) {
     offset: run.offset,
     counts: runCounts(run.counts),
     operations: run.operations,
+    activatedTrendingRunId: run.activatedTrendingRunId ?? null,
+    activationSnapshotId: run.activationSnapshotId ?? null,
+    activatedAt: run.activatedAt ?? null,
     startedAt: run.startedAt,
     completedAt: run.completedAt ?? null,
     updatedAt: run.updatedAt,

convex/skillsShMirrorVisibility.test.ts

@@ -454,9 +454,20 @@ describe("skills.sh mirror visibility operations", () => {
           q.eq("kind", "skills").eq("status", "ready").gt("expiresAt", 0),
         )
         .collect(),
+      runs: await ctx.db.query("skillsShMirrorRuns").withIndex("by_started_at").collect(),
     }));
     expect(activationState.mirrorControl?.activationLockToken).toBeUndefined();
     expect(activationState.snapshots).toHaveLength(1);
+    const activatedLeaderboard = activationState.runs.find(
+      (run) => run._id === activationState.mirrorControl?.latestCompletedLeaderboardRunId,
+    );
+    expect(activatedLeaderboard?.activatedTrendingRunId).toBe(
+      activationState.runs.find((run) => run.sourceView === "trending")?._id,
+    );
+    expect(activatedLeaderboard?.activationSnapshotId).toBe(
+      activationState.snapshots[0]?.snapshotId,
+    );
+    expect(activatedLeaderboard?.activatedAt).toEqual(expect.any(Number));
     await expect(
       t.query(internal.canonicalTrending.getPageInternal, { cursor: null, limit: 20 }),
     ).resolves.toMatchObject({

convex/skillsShMirrorVisibility.ts

@@ -755,6 +755,11 @@ export const finalizeActivationInternal = internalMutation({
     if (!actor || !reason) throw new Error("skills.sh public gate actor and reason are required");
     const now = Date.now();
     await writePublicGate(ctx, { enabled: true, actor, reason, now });
+    await ctx.db.patch("skillsShMirrorRuns", args.leaderboardRunId, {
+      activatedTrendingRunId: args.trendingRunId,
+      activationSnapshotId: args.snapshotId,
+      activatedAt: now,
+    });
     await ctx.db.patch(control._id, {
       activationLockToken: undefined,
       activationLockedAt: undefined,
@@ -764,7 +769,16 @@ export const finalizeActivationInternal = internalMutation({
       reason,
       updatedAt: now,
     });
-    return { ok: true as const, environment, enabled: true as const, updatedAt: now };
+    return {
+      ok: true as const,
+      environment,
+      enabled: true as const,
+      leaderboardRunId: args.leaderboardRunId,
+      trendingRunId: args.trendingRunId,
+      snapshotId: args.snapshotId,
+      activatedAt: now,
+      updatedAt: now,
+    };
   },
 });
 
@@ -964,7 +978,7 @@ export const verifyAndActivateInternal = internalAction({
       ) {
         throw new Error("skills.sh Trending activation snapshot failed source verification");
       }
-      await ctx.runMutation(
+      const publication = (await ctx.runMutation(
         internalRefs.skillsShMirrorVisibility.finalizeActivationInternal as never,
         {
           ...args,
@@ -974,7 +988,12 @@ export const verifyAndActivateInternal = internalAction({
           snapshotId: trendingSnapshot.snapshotId,
           expectedSkillsShTrending: trendingSnapshot.sourceCounts.skillsShTrending,
         } as never,
-      );
+      )) as {
+        leaderboardRunId: MirrorRun["_id"];
+        trendingRunId: MirrorRun["_id"];
+        snapshotId: string;
+        activatedAt: number;
+      };
       locked = false;
       return {
         ok: true as const,
@@ -984,6 +1003,10 @@ export const verifyAndActivateInternal = internalAction({
         trending,
         corpus: corpusAudit.counts,
         trendingSnapshot,
+        leaderboardRunId: publication.leaderboardRunId,
+        trendingRunId: publication.trendingRunId,
+        snapshotId: publication.snapshotId,
+        activatedAt: publication.activatedAt,
         scansPlanned: 0 as const,
         scansAdmitted: 0 as const,
       };

scripts/skills-sh-catalog/sync.test.ts

@@ -453,6 +453,123 @@ describe("skills.sh synchronization runner", () => {
     ]);
   });
 
+  it("reconciles an ambiguous activation timeout only from the exact durable activation receipt", async () => {
+    const operations: string[] = [];
+    let statusCalls = 0;
+    const sleep = vi.fn(async () => undefined);
+    const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      operations.push(String(body.operation));
+      if (body.operation === "status") {
+        statusCalls += 1;
+        if (statusCalls === 1) {
+          return response({ runs: [], invariants: { publicVisible: false } });
+        }
+        if (statusCalls === 2) {
+          return response({
+            control: {
+              activationLockToken: "skills-sh-activation:in-flight",
+              activationLeaderboardRunId: "run-leaderboard",
+              activationTrendingRunId: "run-trending",
+            },
+            runs: [],
+            invariants: { publicVisible: false },
+          });
+        }
+        return response({ runs: [], invariants: { publicVisible: true } });
+      }
+      if (body.operation === "prepare-native-trending") return nativeTrendingPreparation();
+      if (body.operation === "configure") return response({ ok: true });
+      if (body.operation === "start") return response(completedRun("leaderboard"));
+      if (body.operation === "start-trending") return response(completedRun("trending"));
+      if (body.operation === "verify-activate") {
+        throw new DOMException("The operation timed out.", "TimeoutError");
+      }
+      if (body.operation === "run") {
+        return statusCalls === 1
+          ? response(completedRun("leaderboard"))
+          : response({
+              ...completedRun("leaderboard"),
+              activatedTrendingRunId: "run-trending",
+              activationSnapshotId: "skills-activation",
+              activatedAt: 1_722_345_678_000,
+            });
+      }
+      throw new Error(`unexpected operation ${String(body.operation)}`);
+    });
+
+    await expect(
+      runSkillsShSync({
+        targetUrl: "https://clawhub.ai/ops/skills-sh/mirror",
+        authorization: "github-oidc",
+        reason: "scheduled proof",
+        fetchImpl,
+        sleep,
+      }),
+    ).resolves.toMatchObject({
+      ok: true,
+      activation: {
+        activated: true,
+        reconciledAfterTimeout: true,
+        leaderboardRunId: "run-leaderboard",
+        trendingRunId: "run-trending",
+        snapshotId: "skills-activation",
+      },
+    });
+    expect(sleep).toHaveBeenCalled();
+    expect(operations).toEqual([
+      "status",
+      "configure",
+      "prepare-native-trending",
+      "start",
+      "start-trending",
+      "verify-activate",
+      "run",
+      "status",
+      "run",
+      "configure",
+      "status",
+    ]);
+  });
+
+  it("does not wait past the final activation receipt reconciliation read", async () => {
+    const sleep = vi.fn(async () => undefined);
+    const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      if (body.operation === "status") {
+        return response({
+          control: {
+            activationLockToken: "skills-sh-activation:in-flight",
+            activationLeaderboardRunId: "run-leaderboard",
+            activationTrendingRunId: "run-trending",
+          },
+          runs: [],
+          invariants: { publicVisible: false },
+        });
+      }
+      if (body.operation === "prepare-native-trending") return nativeTrendingPreparation();
+      if (body.operation === "configure") return response({ ok: true });
+      if (body.operation === "start") return response(completedRun("leaderboard"));
+      if (body.operation === "start-trending") return response(completedRun("trending"));
+      if (body.operation === "verify-activate") {
+        throw new DOMException("The operation timed out.", "TimeoutError");
+      }
+      if (body.operation === "run") return response(completedRun("leaderboard"));
+      throw new Error(`unexpected operation ${String(body.operation)}`);
+    });
+
+    await expect(
+      runSkillsShSync({
+        targetUrl: "https://clawhub.ai/ops/skills-sh/mirror",
+        authorization: "github-oidc",
+        reason: "scheduled proof",
+        fetchImpl,
+        sleep,
+      }),
+    ).rejects.toThrow("did not produce an exact durable activation receipt");
+    expect(sleep).toHaveBeenCalledTimes(131);
+  });
+
   it("preserves the last verified public lane on a transient sync failure", async () => {
     const operations: string[] = [];
     const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => {

scripts/skills-sh-catalog/sync.ts

@@ -14,6 +14,8 @@ const MAX_STEPS = 2_000;
 const MAX_RATE_LIMIT_RETRIES = 30;
 const MAX_RATE_LIMIT_WAIT_MS = 30 * 60 * 1_000;
 const MAX_TRANSPORT_TIMEOUTS = 3;
+const ACTIVATION_RECONCILE_POLL_MS = 5_000;
+const MAX_ACTIVATION_RECONCILE_POLLS = 132;
 
 type MirrorRun = Record<string, unknown>;
 type SyncFetch = (input: string, init: RequestInit) => Promise<Response>;
@@ -47,6 +49,10 @@ function requiredInteger(value: unknown, name: string) {
   return Number(value);
 }
 
+function optionalRecord(value: unknown) {
+  return typeof value === "object" && value !== null ? (value as Record<string, unknown>) : null;
+}
+
 function jwtExpiresAt(jwt: string) {
   const payload = jwt.split(".")[1];
   if (!payload) throw new Error("GitHub OIDC returned a malformed token");
@@ -264,6 +270,44 @@ export async function runSkillsShSync(options: {
     };
   };
 
+  const reconcileTimedOutActivation = async (
+    leaderboardRunId: string,
+    trendingRunId: string,
+  ): Promise<MirrorRun> => {
+    for (let poll = 0; poll < MAX_ACTIVATION_RECONCILE_POLLS; poll += 1) {
+      const leaderboard = mirrorRunFromPayload(
+        await call({ operation: "run", runId: leaderboardRunId }),
+        "run",
+      );
+      if (leaderboard.activatedTrendingRunId === trendingRunId) {
+        return {
+          ok: true,
+          activated: true,
+          reconciledAfterTimeout: true,
+          leaderboardRunId,
+          trendingRunId,
+          snapshotId: requiredString(leaderboard.activationSnapshotId, "activationSnapshotId"),
+          activatedAt: requiredInteger(leaderboard.activatedAt, "activatedAt"),
+        };
+      }
+      const status = await call({ operation: "status" });
+      const control = optionalRecord(status.control);
+      if (!control || typeof control.activationLockToken !== "string") break;
+      if (
+        control.activationLeaderboardRunId !== leaderboardRunId ||
+        control.activationTrendingRunId !== trendingRunId
+      ) {
+        throw new Error("timed-out verify-activate is bound to different source runs");
+      }
+      if (poll + 1 < MAX_ACTIVATION_RECONCILE_POLLS) {
+        await sleep(ACTIVATION_RECONCILE_POLL_MS);
+      }
+    }
+    throw new Error(
+      "timed-out verify-activate did not produce an exact durable activation receipt",
+    );
+  };
+
   const startedAt = Date.now();
   const before = await call({ operation: "status" });
   const publicVisible = (before.invariants as Record<string, unknown> | undefined)?.publicVisible;
@@ -306,7 +350,14 @@ export async function runSkillsShSync(options: {
     try {
       activation = await call({ operation: "verify-activate", reason: options.reason });
     } catch (error) {
-      throw new UnsafeSkillsShCorpusError(error instanceof Error ? error.message : String(error));
+      if (isTransportTimeout(error)) {
+        activation = await reconcileTimedOutActivation(
+          requiredString((leaderboard as MirrorRun).runId, "leaderboard.runId"),
+          requiredString((trending as MirrorRun).runId, "trending.runId"),
+        );
+      } else {
+        throw new UnsafeSkillsShCorpusError(error instanceof Error ? error.message : String(error));
+      }
     }
     await call({ operation: "configure", enabled: false, reason: `${options.reason} complete` });
     const after = await call({ operation: "status" });