From a15f97470fae0f06dfe907079d1b68c5324bbb1a Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Fri, 31 Jul 2026 08:35:07 -0700 Subject: [PATCH] fix: reconcile skills.sh activation timeouts (#3352) --- convex/schema.ts | 3 + convex/skillsShMirror.ts | 6 ++ convex/skillsShMirrorVisibility.test.ts | 11 +++ convex/skillsShMirrorVisibility.ts | 29 +++++- scripts/skills-sh-catalog/sync.test.ts | 117 ++++++++++++++++++++++++ scripts/skills-sh-catalog/sync.ts | 53 ++++++++++- 6 files changed, 215 insertions(+), 4 deletions(-) diff --git a/convex/schema.ts b/convex/schema.ts index a3ad7b17..9080af3e 100644 --- a/convex/schema.ts +++ b/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(), diff --git a/convex/skillsShMirror.ts b/convex/skillsShMirror.ts index 33a781e7..abdeeda5 100644 --- a/convex/skillsShMirror.ts +++ b/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, diff --git a/convex/skillsShMirrorVisibility.test.ts b/convex/skillsShMirrorVisibility.test.ts index 6df3f1c5..fe17cffe 100644 --- a/convex/skillsShMirrorVisibility.test.ts +++ b/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({ diff --git a/convex/skillsShMirrorVisibility.ts b/convex/skillsShMirrorVisibility.ts index 49d457ed..0dd9e445 100644 --- a/convex/skillsShMirrorVisibility.ts +++ b/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, }; diff --git a/scripts/skills-sh-catalog/sync.test.ts b/scripts/skills-sh-catalog/sync.test.ts index dbd7ee69..1093854b 100644 --- a/scripts/skills-sh-catalog/sync.test.ts +++ b/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; + 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; + 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) => { diff --git a/scripts/skills-sh-catalog/sync.ts b/scripts/skills-sh-catalog/sync.ts index c148cb20..1553452b 100644 --- a/scripts/skills-sh-catalog/sync.ts +++ b/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; type SyncFetch = (input: string, init: RequestInit) => Promise; @@ -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) : 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 => { + 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 | 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" });