diff --git a/scripts/skills-sh-catalog/sync.test.ts b/scripts/skills-sh-catalog/sync.test.ts index 70501462..a10d5b3b 100644 --- a/scripts/skills-sh-catalog/sync.test.ts +++ b/scripts/skills-sh-catalog/sync.test.ts @@ -72,7 +72,8 @@ describe("skills.sh synchronization runner", () => { const fetchImpl = vi .fn() .mockResolvedValueOnce(response({ value: token("first", now + 5 * 60_000) })) - .mockResolvedValueOnce(response({ value: token("second", now + 10 * 60_000) })); + .mockResolvedValueOnce(response({ value: token("second", now + 10 * 60_000) })) + .mockResolvedValueOnce(response({ value: token("forced", now + 10 * 60_000) })); const authorization = createGitHubActionsOidcAuthorization({ requestUrl: "https://token.actions.example/id-token?job=sync", requestToken: "request-token", @@ -85,7 +86,8 @@ describe("skills.sh synchronization runner", () => { await expect(authorization()).resolves.toContain(".first"); now += 2 * 60_000; await expect(authorization()).resolves.toContain(".second"); - expect(fetchImpl).toHaveBeenCalledTimes(2); + await expect(authorization(true)).resolves.toContain(".forced"); + expect(fetchImpl).toHaveBeenCalledTimes(3); }); it("completes leaderboard and Trending before automatic activation", async () => { @@ -436,6 +438,87 @@ describe("skills.sh synchronization runner", () => { ]); }); + it("refreshes authorization and reconciles the durable cursor after an operator 401", async () => { + const operations: string[] = []; + let refreshed = false; + const authorization = vi.fn(async (forceRefresh?: boolean) => { + if (forceRefresh) refreshed = true; + return refreshed ? "fresh-github-oidc" : "stale-github-oidc"; + }); + const running = { + runId: "run-leaderboard", + snapshotId: "skills-sh:leaderboard:durable", + sourceView: "leaderboard", + sourceTotal: 1, + sourcePageSize: 500, + sourceMeasuredAt: "2026-08-01T05:32:30.080Z", + page: 17, + offset: 300, + status: "running", + startedAt: 1, + }; + const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { + const body = JSON.parse(String(init.body)) as Record; + const bearer = String((init.headers as Record).Authorization); + operations.push(String(body.operation)); + switch (body.operation) { + case "status": + return response({ runs: [running], invariants: { publicVisible: true } }); + case "configure": + return response({ ok: true, enabled: body.enabled }); + case "step": + if (bearer === "Bearer stale-github-oidc") { + return response( + { + error: "skills_sh_mirror_test_failed", + message: "Convex Test mirror operator returned HTTP 401: Unauthorized", + }, + 502, + ); + } + return response(completedRun("leaderboard")); + case "run": + expect(bearer).toBe("Bearer fresh-github-oidc"); + return response(running); + case "start-trending": + return response(completedRun("trending")); + case "verify-activate": + return response({ ok: true, activated: true }); + default: + throw new Error(`unexpected operation ${String(body.operation)}`); + } + }); + + await expect( + runSkillsShSync({ + targetUrl: "https://clawhub.ai/ops/skills-sh/mirror", + authorization, + reason: "scheduled recovery", + fetchImpl, + }), + ).resolves.toMatchObject({ + ok: true, + leaderboard: { + runId: "run-leaderboard", + syncProof: { authorizationRetries: 1 }, + }, + scansPlanned: 0, + scansAdmitted: 0, + }); + expect(authorization).toHaveBeenCalledWith(true); + expect(operations).toEqual([ + "status", + "configure", + "step", + "run", + "step", + "start-trending", + "verify-activate", + "configure", + "status", + ]); + }); + it("retries the exact durable cursor after a rate-limit backoff and ambiguous timeout", async () => { const requests: string[] = []; const sleep = vi.fn(async () => undefined); diff --git a/scripts/skills-sh-catalog/sync.ts b/scripts/skills-sh-catalog/sync.ts index 49ec22b1..74b6e842 100644 --- a/scripts/skills-sh-catalog/sync.ts +++ b/scripts/skills-sh-catalog/sync.ts @@ -14,13 +14,14 @@ 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_AUTHORIZATION_RETRIES = 3; const MAX_NATIVE_TRENDING_RECONCILE_POLLS = 360; const ACTIVATION_RECONCILE_POLL_MS = 5_000; const MAX_ACTIVATION_RECONCILE_POLLS = 360; type MirrorRun = Record; type SyncFetch = (input: string, init: RequestInit) => Promise; -type SyncAuthorization = string | (() => Promise); +type SyncAuthorization = string | ((forceRefresh?: boolean) => Promise); const OIDC_REFRESH_SKEW_MS = 2 * 60_000; @@ -107,9 +108,11 @@ export function createGitHubActionsOidcAuthorization(options: { const fetchImpl = options.fetchImpl ?? fetch; const now = options.now ?? Date.now; let cached: { token: string; expiresAt: number } | null = null; - return async () => { + return async (forceRefresh = false) => { const currentTime = now(); - if (cached && currentTime < cached.expiresAt - OIDC_REFRESH_SKEW_MS) return cached.token; + if (!forceRefresh && cached && currentTime < cached.expiresAt - OIDC_REFRESH_SKEW_MS) { + return cached.token; + } const separator = requestUrl.includes("?") ? "&" : "?"; const response = await fetchImpl(`${requestUrl}${separator}audience=clawhub`, { headers: { Authorization: `Bearer ${requestToken}` }, @@ -171,11 +174,11 @@ export async function runSkillsShSync(options: { const fetchImpl = options.fetchImpl ?? fetch; const sleep = options.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); - const callRaw = async (body: Record) => { + const callRaw = async (body: Record, forceAuthorizationRefresh = false) => { const authorization = typeof options.authorization === "string" ? options.authorization - : await options.authorization(); + : await options.authorization(forceAuthorizationRefresh); const response = await fetchImpl(options.targetUrl, { method: "POST", headers: { @@ -194,6 +197,15 @@ export async function runSkillsShSync(options: { } return { response, payload }; }; + const isAuthorizationFailure = (result: Awaited>) => { + if (result.response.status === 401) return true; + const message = result.payload.message; + return ( + result.response.status === 502 && + typeof message === "string" && + /operator returned HTTP 401\b/i.test(message) + ); + }; const call = async (body: Record) => { const result = await callRaw(body); if (!result.response.ok) { @@ -209,6 +221,7 @@ export async function runSkillsShSync(options: { let rateLimitRetries = 0; let rateLimitWaitMs = 0; let transportTimeouts = 0; + let authorizationRetries = 0; if (run.status === "paused") { run = mirrorRunFromPayload( await call({ @@ -259,6 +272,35 @@ export async function runSkillsShSync(options: { continue; } if (!result.response.ok) { + if ( + isAuthorizationFailure(result) && + typeof options.authorization === "function" && + authorizationRetries < MAX_AUTHORIZATION_RETRIES + ) { + authorizationRetries += 1; + const authoritativeResult = await callRaw( + { operation: "run", runId: request.runId }, + true, + ); + if (!authoritativeResult.response.ok) { + throw new Error( + `run returned HTTP ${authoritativeResult.response.status}: ${JSON.stringify(authoritativeResult.payload)}`, + ); + } + const authoritativeRun = mirrorRunFromPayload(authoritativeResult.payload, "run"); + if ( + requiredString(authoritativeRun.runId, "runId") !== request.runId || + (authoritativeRun.sourceView ?? "leaderboard") !== sourceView + ) { + throw new Error( + `unauthorized ${request.operation} reconciled to a different durable run`, + ); + } + run = authoritativeRun; + const cursorAdvanced = run.page !== request.page || run.offset !== request.offset; + if (cursorAdvanced) steps += 1; + continue; + } const delayMs = mirrorRateLimitRetryDelayMs( result.response.status, result.response.headers.get("retry-after"), @@ -293,6 +335,7 @@ export async function runSkillsShSync(options: { rateLimitRetries, rateLimitWaitMs, transportTimeouts, + authorizationRetries, }, }; };