fix: reconcile native trending preflight timeouts (#3358)

convex/httpApiV1/skillsShCatalogV1.test.ts

@@ -76,6 +76,44 @@ function artifact(externalId: string, content: string) {
 }
 
 describe("skills.sh catalog Test HTTP API", () => {
+  it("includes native Trending readiness in mirror status", async () => {
+    vi.stubEnv("CLAWHUB_ENV", "production");
+    vi.stubEnv("CLAWHUB_DEPLOYMENT_NAME", "wry-manatee-359");
+    vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "production");
+    vi.mocked(verifyGitHubActionsSkillsShSyncJwt).mockResolvedValue({
+      actor: "github-actions[bot]",
+      eventName: "schedule",
+      runId: "603",
+      runAttempt: "1",
+      sha: "a".repeat(40),
+    } as never);
+    const nativeTrending = {
+      status: "ready",
+      snapshotId: "skills-native-ready",
+      sourceCounts: { clawhubTrending: 10, clawhubRising: 5, skillsShTrending: 0 },
+    };
+    const runQuery = vi
+      .fn()
+      .mockResolvedValueOnce({ runs: [], control: null })
+      .mockResolvedValueOnce(nativeTrending);
+
+    const response = await skillsShCatalogTestV1Handler(
+      { runQuery } as never,
+      new Request("https://wry-manatee-359.convex.site/api/v1/operator/skills-sh/mirror", {
+        method: "POST",
+        headers: { Authorization: "Bearer github-oidc-token" },
+        body: JSON.stringify({ operation: "mirror-status" }),
+      }),
+    );
+
+    await expect(response.json()).resolves.toEqual({
+      runs: [],
+      control: null,
+      nativeTrending,
+    });
+    expect(runQuery).toHaveBeenCalledTimes(2);
+  });
+
   it("accepts only the exact GitHub Actions production sync identity", async () => {
     const workflowSha = "a".repeat(40);
     vi.stubEnv("CLAWHUB_ENV", "production");

convex/httpApiV1/skillsShCatalogV1.ts

@@ -26,6 +26,9 @@ import {
 import { json, requireAdminOrResponse, requireApiTokenUserOrResponse, text } from "./shared";
 
 const internalRefs = internal as unknown as {
+  canonicalTrending: {
+    getReadyNativeSnapshotInternal: unknown;
+  };
   githubSkillSources: {
     getSkillsShAliasTargetInternal: unknown;
   };
@@ -666,11 +669,17 @@ export async function skillsShCatalogTestV1Handler(ctx: ActionCtx, request: Requ
       return text("Not found", 404, rate.headers);
     }
     if (operation === "mirror-status") {
-      return json(
-        await runQueryRef(ctx, internalRefs.skillsShMirror.getStatusInternal, {}),
-        200,
-        rate.headers,
-      );
+      const [status, nativeTrending] = await Promise.all([
+        runQueryRef<Record<string, unknown>>(
+          ctx,
+          internalRefs.skillsShMirror.getStatusInternal,
+          {},
+        ),
+        runQueryRef(ctx, internalRefs.canonicalTrending.getReadyNativeSnapshotInternal, {
+          now: Date.now(),
+        }),
+      ]);
+      return json({ ...status, nativeTrending }, 200, rate.headers);
     }
     if (operation === "mirror-isolation") {
       return json(

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

@@ -160,6 +160,171 @@ describe("skills.sh synchronization runner", () => {
     ]);
   });
 
+  it("waits for a timed-out native preflight to finish without starting it twice", async () => {
+    const operations: string[] = [];
+    const sleep = vi.fn(async () => undefined);
+    let statusCalls = 0;
+    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-native-trending:in-flight",
+              activationLockedAt: 1_722_345_678_000,
+              reason: "scheduled proof native-only preflight",
+            },
+            nativeTrending: null,
+            runs: [],
+            invariants: { publicVisible: false },
+          });
+        }
+        if (statusCalls === 3) {
+          return response({
+            control: {},
+            nativeTrending: {
+              status: "ready",
+              snapshotId: "skills-native-after-timeout",
+              sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 },
+            },
+            runs: [],
+            invariants: { publicVisible: false },
+          });
+        }
+        return response({ runs: [], invariants: { publicVisible: true } });
+      }
+      if (body.operation === "prepare-native-trending") {
+        throw new DOMException("The operation timed out.", "TimeoutError");
+      }
+      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") {
+        return response({ ok: true, activated: true });
+      }
+      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,
+      nativeBefore: {
+        nativeTrending: {
+          status: "ready",
+          snapshotId: "skills-native-after-timeout",
+          sourceCounts: { skillsShTrending: 0 },
+        },
+        reconciledAfterTimeout: true,
+      },
+    });
+    expect(sleep).toHaveBeenCalledExactlyOnceWith(5_000);
+    expect(operations).toEqual([
+      "status",
+      "configure",
+      "prepare-native-trending",
+      "status",
+      "status",
+      "start",
+      "start-trending",
+      "verify-activate",
+      "configure",
+      "status",
+    ]);
+  });
+
+  it("fails closed when a timed-out native preflight releases without a ready snapshot", async () => {
+    const operations: string[] = [];
+    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") {
+        return operations.length === 1
+          ? response({ runs: [], invariants: { publicVisible: false } })
+          : response({
+              control: {},
+              nativeTrending: null,
+              runs: [],
+              invariants: { publicVisible: false },
+            });
+      }
+      if (body.operation === "prepare-native-trending") {
+        throw new DOMException("The operation timed out.", "TimeoutError");
+      }
+      if (body.operation === "configure") return response({ ok: true });
+      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,
+      }),
+    ).rejects.toThrow("native-only Trending preflight finished without a ready snapshot");
+    expect(operations).toEqual([
+      "status",
+      "configure",
+      "prepare-native-trending",
+      "status",
+      "configure",
+    ]);
+  });
+
+  it("fails closed when a timed-out native preflight is replaced by another lock", async () => {
+    const operations: string[] = [];
+    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") {
+        return operations.length === 1
+          ? response({ runs: [], invariants: { publicVisible: false } })
+          : response({
+              control: {
+                activationLockToken: "skills-sh-activation:different-operation",
+                reason: "another activation",
+              },
+              nativeTrending: null,
+              runs: [],
+              invariants: { publicVisible: false },
+            });
+      }
+      if (body.operation === "prepare-native-trending") {
+        throw new DOMException("The operation timed out.", "TimeoutError");
+      }
+      if (body.operation === "configure") return response({ ok: true });
+      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,
+      }),
+    ).rejects.toThrow("timed-out native Trending preflight is bound to a different lock");
+    expect(operations).toEqual([
+      "status",
+      "configure",
+      "prepare-native-trending",
+      "status",
+      "configure",
+    ]);
+  });
+
   it("resumes an interrupted durable run before starting the next source view", async () => {
     const operations: string[] = [];
     const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => {
@@ -567,7 +732,7 @@ describe("skills.sh synchronization runner", () => {
         sleep,
       }),
     ).rejects.toThrow("did not produce an exact durable activation receipt");
-    expect(sleep).toHaveBeenCalledTimes(131);
+    expect(sleep).toHaveBeenCalledTimes(359);
   });
 
   it("preserves the last verified public lane on a transient sync failure", async () => {

scripts/skills-sh-catalog/sync.ts

@@ -14,8 +14,9 @@ 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 MAX_NATIVE_TRENDING_RECONCILE_POLLS = 360;
 const ACTIVATION_RECONCILE_POLL_MS = 5_000;
-const MAX_ACTIVATION_RECONCILE_POLLS = 132;
+const MAX_ACTIVATION_RECONCILE_POLLS = 360;
 
 type MirrorRun = Record<string, unknown>;
 type SyncFetch = (input: string, init: RequestInit) => Promise<Response>;
@@ -308,6 +309,40 @@ export async function runSkillsShSync(options: {
     );
   };
 
+  const reconcileTimedOutNativePreparation = async (
+    preflightReason: string,
+  ): Promise<MirrorRun> => {
+    for (let poll = 0; poll < MAX_NATIVE_TRENDING_RECONCILE_POLLS; poll += 1) {
+      const status = await call({ operation: "status" });
+      const control = optionalRecord(status.control);
+      const lockToken = control?.activationLockToken;
+      if (typeof lockToken === "string") {
+        if (
+          !lockToken.startsWith("skills-sh-native-trending:") ||
+          control?.reason !== preflightReason
+        ) {
+          throw new Error("timed-out native Trending preflight is bound to a different lock");
+        }
+        if (poll + 1 < MAX_NATIVE_TRENDING_RECONCILE_POLLS) {
+          await sleep(ACTIVATION_RECONCILE_POLL_MS);
+          continue;
+        }
+        break;
+      }
+      const nativeTrending = optionalRecord(status.nativeTrending);
+      const sourceCounts = optionalRecord(nativeTrending?.sourceCounts);
+      if (nativeTrending?.status !== "ready" || sourceCounts?.skillsShTrending !== 0) {
+        throw new Error("native-only Trending preflight finished without a ready snapshot");
+      }
+      return {
+        ok: true,
+        nativeTrending,
+        reconciledAfterTimeout: true,
+      };
+    }
+    throw new Error("timed-out native Trending preflight did not release its exact durable lock");
+  };
+
   const startedAt = Date.now();
   const before = await call({ operation: "status" });
   const publicVisible = (before.invariants as Record<string, unknown> | undefined)?.publicVisible;
@@ -320,12 +355,18 @@ export async function runSkillsShSync(options: {
   let nativeBefore: Record<string, unknown> | null = null;
   await call({ operation: "configure", enabled: true, reason: options.reason });
   try {
-    nativeBefore = publicVisible
-      ? null
-      : await call({
+    if (!publicVisible) {
+      const preflightReason = `${options.reason} native-only preflight`;
+      try {
+        nativeBefore = await call({
           operation: "prepare-native-trending",
-          reason: `${options.reason} native-only preflight`,
+          reason: preflightReason,
         });
+      } catch (error) {
+        if (!isTransportTimeout(error)) throw error;
+        nativeBefore = await reconcileTimedOutNativePreparation(preflightReason);
+      }
+    }
     if (nativeBefore) {
       const nativeTrending = nativeBefore.nativeTrending as Record<string, unknown> | undefined;
       const sourceCounts = nativeTrending?.sourceCounts as Record<string, unknown> | undefined;