diff --git a/convex/_generated/api.d.ts b/convex/_generated/api.d.ts index bd85890c..d71134dd 100644 --- a/convex/_generated/api.d.ts +++ b/convex/_generated/api.d.ts @@ -86,6 +86,7 @@ import type * as lib_githubIdentity from "../lib/githubIdentity.js"; import type * as lib_githubImport from "../lib/githubImport.js"; import type * as lib_githubOrgMemberships from "../lib/githubOrgMemberships.js"; import type * as lib_githubProfileSync from "../lib/githubProfileSync.js"; +import type * as lib_githubRepositoryDispatch from "../lib/githubRepositoryDispatch.js"; import type * as lib_githubSkillScans from "../lib/githubSkillScans.js"; import type * as lib_githubSkillSync from "../lib/githubSkillSync.js"; import type * as lib_globalStats from "../lib/globalStats.js"; @@ -294,6 +295,7 @@ declare const fullApi: ApiFromModules<{ "lib/githubImport": typeof lib_githubImport; "lib/githubOrgMemberships": typeof lib_githubOrgMemberships; "lib/githubProfileSync": typeof lib_githubProfileSync; + "lib/githubRepositoryDispatch": typeof lib_githubRepositoryDispatch; "lib/githubSkillScans": typeof lib_githubSkillScans; "lib/githubSkillSync": typeof lib_githubSkillSync; "lib/globalStats": typeof lib_globalStats; diff --git a/convex/canonicalTrending.test.ts b/convex/canonicalTrending.test.ts index 62dda6b3..2267caec 100644 --- a/convex/canonicalTrending.test.ts +++ b/convex/canonicalTrending.test.ts @@ -104,6 +104,47 @@ async function insertEligibleNativeSource(t: ReturnType, slug }); } +async function insertReadyNativePool( + t: ReturnType, + input: { + poolId: string; + skillId: Awaited>["skillId"]; + now: number; + }, +) { + await t.mutation(internal.canonicalTrending.startNativePoolInternal, { + poolId: input.poolId, + generatedAt: input.now - 1_000, + expiresAt: input.now + 24 * 60 * 60 * 1_000, + windowStartHour: 100, + windowEndHour: 123, + sealedGeneration: 7, + }); + await t.mutation(internal.canonicalTrending.writeNativePoolItemsInternal, { + poolId: input.poolId, + lane: "clawhub-trending", + items: [ + { + identity: `clawhub:${input.poolId}`, + publisherKey: "user:patrick", + installs24h: 8, + bookmarks24h: 1, + createdAt: input.now - 10_000, + updatedAt: input.now - 1_000, + upstreamRank: null, + sourceRef: { kind: "clawhub", skillId: input.skillId }, + card: nativeCard(`clawhub:${input.poolId}`, 8), + }, + ], + }); + await t.mutation(internal.canonicalTrending.finalizeNativePoolInternal, { + poolId: input.poolId, + completedAt: input.now - 500, + sourceCounts: { clawhubTrending: 1, clawhubRising: 0 }, + operations: { documentsRead: 10, documentsWritten: 4, functionCalls: 3 }, + }); +} + describe("canonical Trending snapshot storage", () => { it("selects the newest completed Trending run even when no digest references it", async () => { const t = convexTest(schema, modules); @@ -557,12 +598,19 @@ describe("canonical Trending snapshot storage", () => { const t = convexTest(schema, modules); const now = Date.now(); const source = await insertEligibleNativeSource(t, "native-preflight-ready"); + await insertReadyNativePool(t, { + poolId: "skills-native-preflight-ready", + skillId: source.skillId, + now, + }); await t.mutation(internal.canonicalTrending.startSnapshotInternal, { snapshotId: "skills-native-preflight-ready", generatedAt: now - 1_000, expiresAt: now + 24 * 60 * 60 * 1_000, windowStartDay: 40, windowEndDay: 40, + windowStartHour: 100, + windowEndHour: 123, }); await t.mutation(internal.canonicalTrending.writeItemsInternal, { snapshotId: "skills-native-preflight-ready", @@ -587,6 +635,7 @@ describe("canonical Trending snapshot storage", () => { totalItems: 1, sourceCounts: { clawhubTrending: 1, clawhubRising: 0, skillsShTrending: 0 }, operations: { documentsRead: 10, documentsWritten: 2, functionCalls: 3 }, + nativePoolId: "skills-native-preflight-ready", }); await expect( @@ -600,10 +649,105 @@ describe("canonical Trending snapshot storage", () => { totalItems: 1, sourceCounts: { clawhubTrending: 1, clawhubRising: 0, skillsShTrending: 0 }, operations: { documentsRead: 10, documentsWritten: 2, functionCalls: 3 }, + nativePool: { + poolId: "skills-native-preflight-ready", + sourceCounts: { clawhubTrending: 1, clawhubRising: 0 }, + operations: { documentsRead: 10, documentsWritten: 4, functionCalls: 3 }, + }, reused: true, }); }); + it("does not expose an orphan native pool as ready for mixed activation", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + const source = await insertEligibleNativeSource(t, "orphan-native-pool"); + await insertReadyNativePool(t, { + poolId: "skills-orphan-native-pool", + skillId: source.skillId, + now, + }); + + await expect( + t.query(internal.canonicalTrending.getReadyNativePoolInternal, { now }), + ).resolves.toBeNull(); + }); + + it("reuses an older verified native pool when a newer orphan exists", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + const source = await insertEligibleNativeSource(t, "verified-before-orphan"); + await insertReadyNativePool(t, { + poolId: "skills-verified-before-orphan", + skillId: source.skillId, + now, + }); + await t.mutation(internal.canonicalTrending.startSnapshotInternal, { + snapshotId: "skills-verified-before-orphan", + generatedAt: now - 1_000, + expiresAt: now + 24 * 60 * 60 * 1_000, + windowStartDay: 40, + windowEndDay: 40, + windowStartHour: 100, + windowEndHour: 123, + }); + await t.mutation(internal.canonicalTrending.finalizeSnapshotInternal, { + snapshotId: "skills-verified-before-orphan", + completedAt: now - 500, + totalItems: 0, + sourceCounts: { clawhubTrending: 1, clawhubRising: 0, skillsShTrending: 0 }, + operations: { documentsRead: 10, documentsWritten: 2, functionCalls: 3 }, + nativePoolId: "skills-verified-before-orphan", + }); + await insertReadyNativePool(t, { + poolId: "skills-newer-orphan", + skillId: source.skillId, + now: now + 500, + }); + + await expect( + t.query(internal.canonicalTrending.getReadyNativePoolInternal, { now }), + ).resolves.toMatchObject({ poolId: "skills-verified-before-orphan" }); + }); + + it("keeps a native snapshot ready but marks a mismatched pool unusable", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + const source = await insertEligibleNativeSource(t, "mismatched-native-pool"); + await insertReadyNativePool(t, { + poolId: "skills-mismatched-native-pool", + skillId: source.skillId, + now, + }); + await t.mutation(internal.canonicalTrending.startSnapshotInternal, { + snapshotId: "skills-mismatched-native-pool", + generatedAt: now - 1_000, + expiresAt: now + 24 * 60 * 60 * 1_000, + windowStartDay: 40, + windowEndDay: 40, + windowStartHour: 100, + windowEndHour: 123, + }); + await t.mutation(internal.canonicalTrending.finalizeSnapshotInternal, { + snapshotId: "skills-mismatched-native-pool", + completedAt: now - 500, + totalItems: 0, + sourceCounts: { clawhubTrending: 0, clawhubRising: 0, skillsShTrending: 0 }, + operations: { documentsRead: 1, documentsWritten: 2, functionCalls: 2 }, + nativePoolId: "skills-mismatched-native-pool", + }); + + await expect( + t.query(internal.canonicalTrending.getReadyNativeSnapshotInternal, { now }), + ).resolves.toMatchObject({ + snapshotId: "skills-mismatched-native-pool", + nativePool: null, + }); + await expect( + t.query(internal.canonicalTrending.getReadyNativePoolInternal, { now }), + ).resolves.toBeNull(); + }); + it("does not reuse a native-only snapshot from the pre-download ranking version", async () => { const t = convexTest(schema, modules); const now = Date.now(); @@ -701,20 +845,47 @@ describe("canonical Trending snapshot storage", () => { }, ], }); + await t.mutation(internal.canonicalTrending.startNativePoolInternal, { + poolId: "skills-expired-cleanup", + generatedAt: 1_000, + expiresAt: Date.now() - 1, + windowStartHour: 100, + windowEndHour: 123, + sealedGeneration: 1, + }); + await t.mutation(internal.canonicalTrending.writeNativePoolItemsInternal, { + poolId: "skills-expired-cleanup", + lane: "clawhub-trending", + items: [ + { + identity: "clawhub:old", + publisherKey: "user:patrick", + installs24h: 1, + bookmarks24h: 0, + createdAt: 1_000, + updatedAt: 1_000, + upstreamRank: null, + sourceRef: { kind: "clawhub", skillId: source.skillId }, + card: nativeCard("clawhub:old", 1), + }, + ], + }); const result = await t.action(internal.canonicalTrending.pruneExpiredActionInternal, {}); const rows = await t.run(async (ctx) => ({ snapshots: await ctx.db.query("canonicalTrendingSnapshots").collect(), items: await ctx.db.query("canonicalTrendingItems").collect(), + nativePools: await ctx.db.query("canonicalTrendingNativePools").collect(), + nativePoolItems: await ctx.db.query("canonicalTrendingNativePoolItems").collect(), })); expect(result).toEqual({ - itemsDeleted: 1, - snapshotsDeleted: 1, + itemsDeleted: 2, + snapshotsDeleted: 2, batches: 1, continuationScheduled: false, }); - expect(rows).toEqual({ snapshots: [], items: [] }); + expect(rows).toEqual({ snapshots: [], items: [], nativePools: [], nativePoolItems: [] }); }); it("materializes hourly native metrics with verified skills.sh rows under the activation lock", async () => { @@ -910,6 +1081,16 @@ describe("canonical Trending snapshot storage", () => { }); }); + const nativePreflight = await t.action(internal.canonicalTrending.materializeInternal, { + activationLockToken: "activation-lock", + }); + expect(nativePreflight).toMatchObject({ + status: "ready", + totalItems: 1, + sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + nativePool: { reused: false }, + }); + const result = await t.action(internal.canonicalTrending.materializeInternal, { activationLockToken: "activation-lock", }); @@ -917,6 +1098,7 @@ describe("canonical Trending snapshot storage", () => { status: "ready", totalItems: 2, sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 1 }, + nativePool: { poolId: nativePreflight.snapshotId, reused: true }, sample: [ { rank: 1, @@ -1033,6 +1215,11 @@ describe("canonical Trending snapshot storage", () => { status: "ready", totalItems: 1, sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + nativePool: { + reused: false, + sourceCounts: { clawhubTrending: 1, clawhubRising: 1 }, + operations: { documentsWritten: 6 }, + }, }); await expect( t.query(internal.canonicalTrending.getPageInternal, { cursor: null, limit: 20 }), @@ -1040,5 +1227,44 @@ describe("canonical Trending snapshot storage", () => { status: "ok", page: { items: [{ source: "clawhub" }] }, }); + + await t.run(async (ctx) => { + const hourlyRows = await ctx.db.query("skillHourlyStats").collect(); + for (const row of hourlyRows) await ctx.db.delete(row._id); + const mirrorControl = await ctx.db + .query("skillsShMirrorControls") + .withIndex("by_key", (q) => q.eq("key", "global")) + .unique(); + if (!mirrorControl) throw new Error("mirror control missing"); + await ctx.db.patch(mirrorControl._id, { + activationLockToken: "mixed-pool-lock", + activationLockedAt: Date.now(), + }); + }); + + const mixedFromPool = await t.action(internal.canonicalTrending.materializeInternal, { + activationLockToken: "mixed-pool-lock", + }); + expect(mixedFromPool).toMatchObject({ + status: "ready", + totalItems: 2, + sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 1 }, + nativePool: { + reused: true, + poolId: nativeOnly.snapshotId, + sourceCounts: { clawhubTrending: 1, clawhubRising: 1 }, + }, + sample: [ + expect.objectContaining({ + lane: "clawhub-trending", + trending24hDownloads: 18, + trending24hInstalls: 12, + }), + expect.objectContaining({ + lane: "skills-sh-trending", + id: "skills-sh:patrick/repo/external", + }), + ], + }); }); }); diff --git a/convex/canonicalTrending.ts b/convex/canonicalTrending.ts index 88782710..5b73f29a 100644 --- a/convex/canonicalTrending.ts +++ b/convex/canonicalTrending.ts @@ -37,10 +37,15 @@ const WRITE_BATCH_SIZE = 100; const NATIVE_SOURCE_BATCH_SIZE = 100; const SNAPSHOT_RETENTION_MS = 48 * 60 * 60 * 1_000; const SNAPSHOT_MAX_SERVING_AGE_MS = 2 * 60 * 60 * 1_000; +const NATIVE_POOL_MAX_AGE_MS = SNAPSHOT_MAX_SERVING_AGE_MS; const EXTERNAL_SOURCE_MAX_AGE_MS = 2 * 60 * 60 * 1_000; const RISING_MAX_AGE_MS = 30 * 24 * 60 * 60 * 1_000; const PRUNE_BATCH_SIZE = 500; const PRUNE_MAX_BATCHES = 20; +const NATIVE_POOL_REUSE_SCAN_LIMIT = 100; +const NATIVE_SNAPSHOT_REUSE_SCAN_LIMIT = 100; +const MAX_NATIVE_POOL_ITEMS_PER_LANE = + CANONICAL_TRENDING_LANE_LIMIT + CANONICAL_TRENDING_FIRST_PAGE_SIZE; const internalRefs = internal as unknown as { canonicalTrending: { @@ -51,9 +56,15 @@ const internalRefs = internal as unknown as { getMaterializationModeInternal: unknown; getNativeSourceBatchInternal: unknown; getLatestCompletedTrendingRunInternal: unknown; + getNativePoolPageInternal: unknown; + getReadyNativePoolInternal: unknown; + failNativePoolInternal: unknown; + finalizeNativePoolInternal: unknown; pruneExpiredInternal: unknown; pruneExpiredActionInternal: unknown; + startNativePoolInternal: unknown; startSnapshotInternal: unknown; + writeNativePoolItemsInternal: unknown; writeItemsInternal: unknown; }; skillHourlyStats: { @@ -86,6 +97,20 @@ const laneValidator = v.union( v.literal("skills-sh-trending"), ); +const nativeLaneValidator = v.union(v.literal("clawhub-trending"), v.literal("clawhub-rising")); + +const nativePoolItemValidator = v.object({ + identity: v.string(), + publisherKey: v.string(), + installs24h: v.number(), + bookmarks24h: v.number(), + createdAt: v.number(), + updatedAt: v.number(), + upstreamRank: v.union(v.number(), v.null()), + sourceRef: canonicalTrendingSourceRefValidator, + card: canonicalTrendingCardValidator, +}); + const sourceCountsValidator = v.object({ clawhubTrending: v.number(), clawhubRising: v.number(), @@ -241,6 +266,220 @@ export const getLatestCompletedTrendingRunInternal = internalQuery({ }, }); +export const getReadyNativePoolInternal = internalQuery({ + args: { now: v.number() }, + handler: async (ctx, args) => { + const pools = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_status_and_generated_at", (q) => + q.eq("status", "ready").gte("generatedAt", args.now - NATIVE_POOL_MAX_AGE_MS), + ) + .order("desc") + .take(NATIVE_POOL_REUSE_SCAN_LIMIT); + for (const pool of pools) { + if ( + pool.expiresAt <= args.now || + pool.rankingVersion !== CANONICAL_TRENDING_RANKING_VERSION || + pool.completedAt === undefined || + !pool.sourceCounts || + !pool.operations || + pool.sourceCounts.clawhubTrending !== pool.writtenTrendingItems || + pool.sourceCounts.clawhubRising !== pool.writtenRisingItems || + pool.writtenTrendingItems > MAX_NATIVE_POOL_ITEMS_PER_LANE || + pool.writtenRisingItems > MAX_NATIVE_POOL_ITEMS_PER_LANE + ) { + continue; + } + const snapshot = await ctx.db + .query("canonicalTrendingSnapshots") + .withIndex("by_snapshot_id", (q) => q.eq("snapshotId", pool.poolId)) + .unique(); + if ( + !snapshot || + snapshot.status !== "ready" || + snapshot.nativePoolId !== pool.poolId || + snapshot.rankingVersion !== pool.rankingVersion || + snapshot.generatedAt !== pool.generatedAt || + snapshot.windowStartHour !== pool.windowStartHour || + snapshot.windowEndHour !== pool.windowEndHour || + snapshot.sourceCounts?.skillsShTrending !== 0 || + snapshot.sourceCounts.clawhubTrending !== pool.sourceCounts.clawhubTrending || + snapshot.sourceCounts.clawhubRising !== pool.sourceCounts.clawhubRising + ) { + continue; + } + return { + status: "ready" as const, + poolId: pool.poolId, + rankingVersion: pool.rankingVersion, + generatedAt: pool.generatedAt, + completedAt: pool.completedAt, + expiresAt: pool.expiresAt, + windowStartHour: pool.windowStartHour, + windowEndHour: pool.windowEndHour, + sealedGeneration: pool.sealedGeneration, + sourceCounts: pool.sourceCounts, + operations: pool.operations, + }; + } + return null; + }, +}); + +export const getNativePoolPageInternal = internalQuery({ + args: { + poolId: v.string(), + paginationOpts: paginationOptsValidator, + }, + handler: async (ctx, args) => { + const pool = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_pool_id", (q) => q.eq("poolId", args.poolId)) + .unique(); + if ( + !pool || + pool.status !== "ready" || + !pool.sourceCounts || + pool.sourceCounts.clawhubTrending !== pool.writtenTrendingItems || + pool.sourceCounts.clawhubRising !== pool.writtenRisingItems + ) { + throw new Error("native Trending candidate pool is not ready"); + } + const page = await ctx.db + .query("canonicalTrendingNativePoolItems") + .withIndex("by_pool_id_and_lane_and_position", (q) => q.eq("poolId", args.poolId)) + .paginate(args.paginationOpts); + return { ...page, documentsRead: page.page.length + 1 }; + }, +}); + +export const startNativePoolInternal = internalMutation({ + args: { + poolId: v.string(), + generatedAt: v.number(), + expiresAt: v.number(), + windowStartHour: v.number(), + windowEndHour: v.number(), + sealedGeneration: v.number(), + }, + handler: async (ctx, args) => { + const existing = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_pool_id", (q) => q.eq("poolId", args.poolId)) + .unique(); + if (existing) throw new Error("native Trending candidate pool already exists"); + return await ctx.db.insert("canonicalTrendingNativePools", { + ...args, + status: "building", + rankingVersion: CANONICAL_TRENDING_RANKING_VERSION, + writtenTrendingItems: 0, + writtenRisingItems: 0, + }); + }, +}); + +export const writeNativePoolItemsInternal = internalMutation({ + args: { + poolId: v.string(), + lane: nativeLaneValidator, + items: v.array(nativePoolItemValidator), + }, + handler: async (ctx, args) => { + if (args.items.length < 1 || args.items.length > WRITE_BATCH_SIZE) { + throw new Error("Invalid native Trending candidate-pool batch size"); + } + const pool = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_pool_id", (q) => q.eq("poolId", args.poolId)) + .unique(); + if (!pool || pool.status !== "building") { + throw new Error("native Trending candidate pool is not writable"); + } + const written = + args.lane === "clawhub-trending" ? pool.writtenTrendingItems : pool.writtenRisingItems; + if (written + args.items.length > MAX_NATIVE_POOL_ITEMS_PER_LANE) { + throw new Error("native Trending candidate pool exceeded its lane bound"); + } + for (const [batchIndex, item] of args.items.entries()) { + if ( + item.sourceRef.kind !== "clawhub" || + item.card.source !== "clawhub" || + item.card.id !== item.identity + ) { + throw new Error("native Trending candidate pool contains a non-native identity"); + } + await ctx.db.insert("canonicalTrendingNativePoolItems", { + poolId: args.poolId, + lane: args.lane, + position: written + batchIndex, + ...item, + expiresAt: pool.expiresAt, + }); + } + const nextWritten = written + args.items.length; + await ctx.db.patch( + pool._id, + args.lane === "clawhub-trending" + ? { writtenTrendingItems: nextWritten } + : { writtenRisingItems: nextWritten }, + ); + return { lane: args.lane, writtenItems: nextWritten }; + }, +}); + +export const finalizeNativePoolInternal = internalMutation({ + args: { + poolId: v.string(), + completedAt: v.number(), + sourceCounts: v.object({ + clawhubTrending: v.number(), + clawhubRising: v.number(), + }), + operations: operationsValidator, + }, + handler: async (ctx, args) => { + const pool = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_pool_id", (q) => q.eq("poolId", args.poolId)) + .unique(); + if (!pool || pool.status !== "building") { + throw new Error("native Trending candidate pool cannot be finalized"); + } + if ( + pool.writtenTrendingItems !== args.sourceCounts.clawhubTrending || + pool.writtenRisingItems !== args.sourceCounts.clawhubRising || + args.sourceCounts.clawhubTrending > MAX_NATIVE_POOL_ITEMS_PER_LANE || + args.sourceCounts.clawhubRising > MAX_NATIVE_POOL_ITEMS_PER_LANE + ) { + throw new Error("native Trending candidate-pool count mismatch"); + } + await ctx.db.patch(pool._id, { + status: "ready", + completedAt: args.completedAt, + sourceCounts: args.sourceCounts, + operations: args.operations, + }); + return { poolId: args.poolId, status: "ready" as const }; + }, +}); + +export const failNativePoolInternal = internalMutation({ + args: { poolId: v.string(), completedAt: v.number(), error: v.string() }, + handler: async (ctx, args) => { + const pool = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_pool_id", (q) => q.eq("poolId", args.poolId)) + .unique(); + if (!pool || pool.status !== "building") return { changed: false }; + await ctx.db.patch(pool._id, { + status: "failed", + completedAt: args.completedAt, + error: args.error.slice(0, 500), + }); + return { changed: true }; + }, +}); + export const startSnapshotInternal = internalMutation({ args: { snapshotId: v.string(), @@ -319,6 +558,7 @@ export const finalizeSnapshotInternal = internalMutation({ totalItems: v.number(), sourceCounts: sourceCountsValidator, operations: operationsValidator, + nativePoolId: v.optional(v.string()), activationLockToken: v.optional(v.string()), }, handler: async (ctx, args) => { @@ -349,6 +589,7 @@ export const finalizeSnapshotInternal = internalMutation({ totalItems: args.totalItems, sourceCounts: args.sourceCounts, operations: args.operations, + nativePoolId: args.nativePoolId, }); return { snapshotId: args.snapshotId, status: "ready" as const }; }, @@ -379,12 +620,21 @@ export const pruneExpiredInternal = internalMutation({ args: { now: v.number(), batchSize: v.number() }, handler: async (ctx, args) => { const batchSize = Math.min(Math.max(Math.trunc(args.batchSize), 1), PRUNE_BATCH_SIZE); - const items = await ctx.db + const snapshotItems = await ctx.db .query("canonicalTrendingItems") .withIndex("by_expires_at", (q) => q.lte("expiresAt", args.now)) .take(batchSize); - for (const item of items) await ctx.db.delete(item._id); - const remaining = batchSize - items.length; + for (const item of snapshotItems) await ctx.db.delete(item._id); + let remaining = batchSize - snapshotItems.length; + const nativePoolItems = + remaining > 0 + ? await ctx.db + .query("canonicalTrendingNativePoolItems") + .withIndex("by_expires_at", (q) => q.lte("expiresAt", args.now)) + .take(remaining) + : []; + for (const item of nativePoolItems) await ctx.db.delete(item._id); + remaining -= nativePoolItems.length; const snapshots = remaining > 0 ? await ctx.db @@ -393,10 +643,21 @@ export const pruneExpiredInternal = internalMutation({ .take(remaining) : []; for (const snapshot of snapshots) await ctx.db.delete(snapshot._id); + remaining -= snapshots.length; + const nativePools = + remaining > 0 + ? await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_expires_at", (q) => q.lte("expiresAt", args.now)) + .take(remaining) + : []; + for (const pool of nativePools) await ctx.db.delete(pool._id); + const itemsDeleted = snapshotItems.length + nativePoolItems.length; + const snapshotsDeleted = snapshots.length + nativePools.length; return { - itemsDeleted: items.length, - snapshotsDeleted: snapshots.length, - fullBatch: items.length + snapshots.length === batchSize, + itemsDeleted, + snapshotsDeleted, + fullBatch: itemsDeleted + snapshotsDeleted === batchSize, }; }, }); @@ -452,91 +713,253 @@ export const materializeInternal = internalAction({ type HourlyWindow = { startHour: number; endHour: number; - startAt: number; - endAt: number; - lastAggregationCompletedAt: number; sealedGeneration: number; }; - const proofWindow = - args.proofSnapshotId !== undefined - ? { - ...getCompletedRolling24HourWindow(startedAt), - lastAggregationCompletedAt: startedAt, - sealedGeneration: 0, - } - : null; - const hourlyWindow = proofWindow - ? proofWindow - : ((await ctx.runMutation( - internalRefs.skillHourlyStats.sealForSnapshotInternal as never, + type ReadyNativePool = { + status: "ready"; + poolId: string; + rankingVersion: string; + generatedAt: number; + completedAt: number; + expiresAt: number; + windowStartHour: number; + windowEndHour: number; + sealedGeneration: number; + sourceCounts: { clawhubTrending: number; clawhubRising: number }; + operations: { + documentsRead: number; + documentsWritten: number; + functionCalls: number; + }; + }; + const canReuseNativePool = + args.activationLockToken !== undefined && + args.skillsShMode !== "native-only" && + args.proofSnapshotId === undefined; + const readyNativePool = canReuseNativePool + ? ((await ctx.runQuery( + internalRefs.canonicalTrending.getReadyNativePoolInternal as never, { now: startedAt } as never, - )) as HourlyWindow | null); - if (!proofWindow) functionCalls += 1; - if (!hourlyWindow) { - return { status: "unavailable" as const, reason: "hourly-stats-not-ready" as const }; - } - - const usageBySkill: RollingHourlyStatTotals = new Map(); - const hourlySource = await forEachCanonicalTrendingSourcePage( - ctx, - internalRefs.canonicalTrending.getHourlySourcePageInternal, - { - startHour: hourlyWindow.startHour, - endHour: hourlyWindow.endHour, - maxGeneration: hourlyWindow.sealedGeneration, - }, - (page) => accumulateRollingHourlyStats(usageBySkill, page as Doc<"skillHourlyStats">[]), - ); - finalizeRollingHourlyStats(usageBySkill); + )) as ReadyNativePool | null) + : null; + if (canReuseNativePool) functionCalls += 1; + const persistOnlyNativePreflight = canReuseNativePool && readyNativePool === null; + let hourlyWindow: HourlyWindow; let nativeCandidates: CanonicalTrendingMaterializationCandidate[] = []; let risingCandidates: CanonicalTrendingMaterializationCandidate[] = []; - const nativeSource = { documentsRead: 0, functionCalls: 0 }; - const risingCutoff = startedAt - RISING_MAX_AGE_MS; - let pendingNativeSkillIds: Id<"skills">[] = []; - const flushNativeSourceBatch = async () => { - if (pendingNativeSkillIds.length === 0) return; - const skillIds = pendingNativeSkillIds; - pendingNativeSkillIds = []; - const sourceBatch = (await ctx.runQuery( - internalRefs.canonicalTrending.getNativeSourceBatchInternal as never, - { skillIds } as never, - )) as { page: Doc<"skillSearchDigest">[]; documentsRead: number }; - nativeSource.documentsRead += sourceBatch.documentsRead; - nativeSource.functionCalls += 1; - for (const digest of sourceBatch.page) { - const usage = usageBySkill.get(String(digest.skillId)); - if (!usage || usage.downloads + usage.installs + usage.bookmarks <= 0) continue; - const candidate = buildNativeCanonicalTrendingCandidate(digest, usage); - if (!candidate) continue; - nativeCandidates.push(candidate); - if (candidate.createdAt >= risingCutoff) { - risingCandidates.push({ ...candidate, lane: "clawhub-rising" }); + let nativePool: { + poolId: string; + reused: boolean; + sourceCounts: { clawhubTrending: number; clawhubRising: number }; + operations: { + documentsRead: number; + documentsWritten: number; + functionCalls: number; + }; + }; + + if (readyNativePool) { + hourlyWindow = { + startHour: readyNativePool.windowStartHour, + endHour: readyNativePool.windowEndHour, + sealedGeneration: readyNativePool.sealedGeneration, + }; + const poolSource = await forEachCanonicalTrendingSourcePage( + ctx, + internalRefs.canonicalTrending.getNativePoolPageInternal, + { poolId: readyNativePool.poolId }, + (page) => { + for (const row of page as Doc<"canonicalTrendingNativePoolItems">[]) { + const candidate: CanonicalTrendingMaterializationCandidate = { + identity: row.identity, + lane: row.lane, + publisherKey: row.publisherKey, + installs24h: row.installs24h, + bookmarks24h: row.bookmarks24h, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + upstreamRank: row.upstreamRank, + sourceRef: row.sourceRef, + card: row.card, + }; + if (row.lane === "clawhub-trending") nativeCandidates.push(candidate); + else risingCandidates.push(candidate); + } + }, + ); + documentsRead += poolSource.documentsRead; + functionCalls += poolSource.functionCalls; + if ( + nativeCandidates.length !== readyNativePool.sourceCounts.clawhubTrending || + risingCandidates.length !== readyNativePool.sourceCounts.clawhubRising + ) { + throw new Error("native Trending candidate-pool read count mismatch"); + } + nativePool = { + poolId: readyNativePool.poolId, + reused: true, + sourceCounts: readyNativePool.sourceCounts, + operations: readyNativePool.operations, + }; + } else { + const proofWindow = + args.proofSnapshotId !== undefined + ? { ...getCompletedRolling24HourWindow(startedAt), sealedGeneration: 0 } + : null; + const sealedWindow = proofWindow + ? proofWindow + : ((await ctx.runMutation( + internalRefs.skillHourlyStats.sealForSnapshotInternal as never, + { now: startedAt } as never, + )) as HourlyWindow | null); + if (!proofWindow) functionCalls += 1; + if (!sealedWindow) { + return { status: "unavailable" as const, reason: "hourly-stats-not-ready" as const }; + } + hourlyWindow = sealedWindow; + + const usageBySkill: RollingHourlyStatTotals = new Map(); + const hourlySource = await forEachCanonicalTrendingSourcePage( + ctx, + internalRefs.canonicalTrending.getHourlySourcePageInternal, + { + startHour: hourlyWindow.startHour, + endHour: hourlyWindow.endHour, + maxGeneration: hourlyWindow.sealedGeneration, + }, + (page) => accumulateRollingHourlyStats(usageBySkill, page as Doc<"skillHourlyStats">[]), + ); + finalizeRollingHourlyStats(usageBySkill); + + const nativeSource = { documentsRead: 0, functionCalls: 0 }; + const risingCutoff = startedAt - RISING_MAX_AGE_MS; + let pendingNativeSkillIds: Id<"skills">[] = []; + const flushNativeSourceBatch = async () => { + if (pendingNativeSkillIds.length === 0) return; + const skillIds = pendingNativeSkillIds; + pendingNativeSkillIds = []; + const sourceBatch = (await ctx.runQuery( + internalRefs.canonicalTrending.getNativeSourceBatchInternal as never, + { skillIds } as never, + )) as { page: Doc<"skillSearchDigest">[]; documentsRead: number }; + nativeSource.documentsRead += sourceBatch.documentsRead; + nativeSource.functionCalls += 1; + for (const digest of sourceBatch.page) { + const usage = usageBySkill.get(String(digest.skillId)); + if (!usage || usage.downloads + usage.installs + usage.bookmarks <= 0) continue; + const candidate = buildNativeCanonicalTrendingCandidate(digest, usage); + if (!candidate) continue; + nativeCandidates.push(candidate); + if (candidate.createdAt >= risingCutoff) { + risingCandidates.push({ ...candidate, lane: "clawhub-rising" }); + } + } + // The fetched batch is capped at 100, so each lane stays within 100 rows of its limit. + nativeCandidates = retainTopCanonicalTrendingCandidates( + nativeCandidates, + "clawhub-trending", + CANONICAL_TRENDING_LANE_LIMIT, + LANE_DIVERSITY_RESERVE, + ); + risingCandidates = retainTopCanonicalTrendingCandidates( + risingCandidates, + "clawhub-rising", + CANONICAL_TRENDING_LANE_LIMIT, + LANE_DIVERSITY_RESERVE, + ); + for (const skillId of skillIds) usageBySkill.delete(String(skillId)); + }; + for (const skillId of usageBySkill.keys()) { + pendingNativeSkillIds.push(skillId as Id<"skills">); + if (pendingNativeSkillIds.length === NATIVE_SOURCE_BATCH_SIZE) { + await flushNativeSourceBatch(); } } - // The fetched batch is capped at 100, so each lane stays within 100 rows of its limit. - nativeCandidates = retainTopCanonicalTrendingCandidates( - nativeCandidates, - "clawhub-trending", - CANONICAL_TRENDING_LANE_LIMIT, - LANE_DIVERSITY_RESERVE, - ); - risingCandidates = retainTopCanonicalTrendingCandidates( - risingCandidates, - "clawhub-rising", - CANONICAL_TRENDING_LANE_LIMIT, - LANE_DIVERSITY_RESERVE, - ); - // Each digest is unique by skill, so its rolling totals are no longer needed. - for (const skillId of skillIds) usageBySkill.delete(String(skillId)); - }; - for (const skillId of usageBySkill.keys()) { - pendingNativeSkillIds.push(skillId as Id<"skills">); - if (pendingNativeSkillIds.length === NATIVE_SOURCE_BATCH_SIZE) { - await flushNativeSourceBatch(); + await flushNativeSourceBatch(); + documentsRead += nativeSource.documentsRead + hourlySource.documentsRead; + functionCalls += nativeSource.functionCalls + hourlySource.functionCalls; + + const poolId = snapshotId; + let poolStarted = false; + try { + await ctx.runMutation( + internalRefs.canonicalTrending.startNativePoolInternal as never, + { + poolId, + generatedAt: startedAt, + expiresAt: startedAt + SNAPSHOT_RETENTION_MS, + windowStartHour: hourlyWindow.startHour, + windowEndHour: hourlyWindow.endHour, + sealedGeneration: hourlyWindow.sealedGeneration, + } as never, + ); + poolStarted = true; + functionCalls += 1; + documentsWritten += 1; + for (const [lane, candidates] of [ + ["clawhub-trending", nativeCandidates], + ["clawhub-rising", risingCandidates], + ] as const) { + for (let index = 0; index < candidates.length; index += WRITE_BATCH_SIZE) { + const batch = candidates.slice(index, index + WRITE_BATCH_SIZE); + await ctx.runMutation( + internalRefs.canonicalTrending.writeNativePoolItemsInternal as never, + { + poolId, + lane, + items: batch.map((candidate) => ({ + identity: candidate.identity, + publisherKey: candidate.publisherKey, + installs24h: candidate.installs24h, + bookmarks24h: candidate.bookmarks24h, + createdAt: candidate.createdAt, + updatedAt: candidate.updatedAt, + upstreamRank: candidate.upstreamRank, + sourceRef: candidate.sourceRef, + card: candidate.card, + })), + } as never, + ); + functionCalls += 1; + documentsWritten += batch.length + 1; + } + } + const sourceCounts = { + clawhubTrending: nativeCandidates.length, + clawhubRising: risingCandidates.length, + }; + const poolWriteBatches = + Math.ceil(nativeCandidates.length / WRITE_BATCH_SIZE) + + Math.ceil(risingCandidates.length / WRITE_BATCH_SIZE); + const poolOperations = { + documentsRead: nativeSource.documentsRead + hourlySource.documentsRead, + documentsWritten: + nativeCandidates.length + risingCandidates.length + poolWriteBatches + 2, + functionCalls: + nativeSource.functionCalls + hourlySource.functionCalls + poolWriteBatches + 2, + }; + await ctx.runMutation( + internalRefs.canonicalTrending.finalizeNativePoolInternal as never, + { poolId, completedAt: Date.now(), sourceCounts, operations: poolOperations } as never, + ); + poolStarted = false; + functionCalls += 1; + documentsWritten += 1; + nativePool = { poolId, reused: false, sourceCounts, operations: poolOperations }; + } finally { + if (poolStarted) { + await ctx.runMutation( + internalRefs.canonicalTrending.failNativePoolInternal as never, + { + poolId, + completedAt: Date.now(), + error: "native Trending candidate-pool persistence failed", + } as never, + ); + } } } - await flushNativeSourceBatch(); type TrendingRun = { runId: Doc<"skillsShMirrorRuns">["_id"] | null; completedAt: number | null; @@ -549,6 +972,7 @@ export const materializeInternal = internalAction({ }; let externalCandidates: CanonicalTrendingMaterializationCandidate[] = []; if ( + !persistOnlyNativePreflight && args.skillsShMode !== "native-only" && getRuntimeRolloutCapabilities().skillsSh.runtimeEnabled ) { @@ -583,10 +1007,8 @@ export const materializeInternal = internalAction({ ); } } - documentsRead += - nativeSource.documentsRead + hourlySource.documentsRead + externalSource.documentsRead; - functionCalls += - nativeSource.functionCalls + hourlySource.functionCalls + externalSource.functionCalls; + documentsRead += externalSource.documentsRead; + functionCalls += externalSource.functionCalls; if (latestTrendingRun) { const confirmedTrendingRun = (await ctx.runQuery( @@ -658,6 +1080,7 @@ export const materializeInternal = internalAction({ completedAt: Date.now(), totalItems: blended.length, sourceCounts, + nativePoolId: nativePool.poolId, operations, activationLockToken: args.activationLockToken, } as never, @@ -673,6 +1096,7 @@ export const materializeInternal = internalAction({ rankingVersion: CANONICAL_TRENDING_RANKING_VERSION, totalItems: blended.length, sourceCounts, + nativePool, operations: { documentsRead, documentsWritten, @@ -709,13 +1133,18 @@ export const materializeInternal = internalAction({ export const getReadyNativeSnapshotInternal = internalQuery({ args: { now: v.number() }, handler: async (ctx, args) => { - const snapshot = await ctx.db + const snapshots = await ctx.db .query("canonicalTrendingSnapshots") .withIndex("by_kind_and_status_and_expires_at", (q) => q.eq("kind", "skills").eq("status", "ready").gt("expiresAt", args.now), ) .order("desc") - .first(); + .take(NATIVE_SNAPSHOT_REUSE_SCAN_LIMIT); + const snapshot = snapshots.find( + (candidate) => + candidate.sourceCounts?.skillsShTrending === 0 && + candidate.generatedAt + SNAPSHOT_MAX_SERVING_AGE_MS > args.now, + ); if ( !snapshot || snapshot.generatedAt + SNAPSHOT_MAX_SERVING_AGE_MS <= args.now || @@ -727,6 +1156,36 @@ export const getReadyNativeSnapshotInternal = internalQuery({ ) { return null; } + const poolId = snapshot.nativePoolId ?? snapshot.snapshotId; + const pool = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_pool_id", (q) => q.eq("poolId", poolId)) + .unique(); + const nativePool = + pool && + pool.status === "ready" && + pool.expiresAt > args.now && + pool.generatedAt + NATIVE_POOL_MAX_AGE_MS > args.now && + pool.rankingVersion === snapshot.rankingVersion && + pool.completedAt !== undefined && + pool.sourceCounts !== undefined && + pool.operations !== undefined && + pool.poolId === snapshot.nativePoolId && + pool.generatedAt === snapshot.generatedAt && + pool.windowStartHour === snapshot.windowStartHour && + pool.windowEndHour === snapshot.windowEndHour && + pool.sourceCounts.clawhubTrending === snapshot.sourceCounts.clawhubTrending && + pool.sourceCounts.clawhubRising === snapshot.sourceCounts.clawhubRising && + pool.sourceCounts.clawhubTrending === pool.writtenTrendingItems && + pool.sourceCounts.clawhubRising === pool.writtenRisingItems && + pool.writtenTrendingItems <= MAX_NATIVE_POOL_ITEMS_PER_LANE && + pool.writtenRisingItems <= MAX_NATIVE_POOL_ITEMS_PER_LANE + ? { + poolId: pool.poolId, + sourceCounts: pool.sourceCounts, + operations: pool.operations, + } + : null; return { status: "ready" as const, snapshotId: snapshot.snapshotId, @@ -736,6 +1195,7 @@ export const getReadyNativeSnapshotInternal = internalQuery({ totalItems: snapshot.totalItems, sourceCounts: snapshot.sourceCounts, operations: snapshot.operations, + nativePool, reused: true as const, }; }, diff --git a/convex/httpApiV1/skillsShCatalogV1.test.ts b/convex/httpApiV1/skillsShCatalogV1.test.ts index b2cf4615..8d8a36c4 100644 --- a/convex/httpApiV1/skillsShCatalogV1.test.ts +++ b/convex/httpApiV1/skillsShCatalogV1.test.ts @@ -91,6 +91,11 @@ describe("skills.sh catalog Test HTTP API", () => { status: "ready", snapshotId: "skills-native-ready", sourceCounts: { clawhubTrending: 10, clawhubRising: 5, skillsShTrending: 0 }, + nativePool: { + poolId: "skills-native-ready", + sourceCounts: { clawhubTrending: 10, clawhubRising: 5 }, + operations: { documentsRead: 100, documentsWritten: 20, functionCalls: 5 }, + }, }; const runQuery = vi .fn() diff --git a/convex/lib/retentionPolicy.ts b/convex/lib/retentionPolicy.ts index d3b6c5ff..e5120a7e 100644 --- a/convex/lib/retentionPolicy.ts +++ b/convex/lib/retentionPolicy.ts @@ -232,6 +232,24 @@ export const RETENTION_POLICIES = { retention: "Forty-eight hours after snapshot generation.", }, ), + canonicalTrendingNativePools: ephemeral( + "Verified bounded native Trending candidate-pool headers expire with their source window.", + { + expirationField: "expiresAt", + expirationIndex: "by_expires_at", + prune: "canonicalTrending.pruneExpiredActionInternal", + retention: "Forty-eight hours after candidate-pool generation.", + }, + ), + canonicalTrendingNativePoolItems: ephemeral( + "Verified bounded native Trending candidates expire with their pool header.", + { + expirationField: "expiresAt", + expirationIndex: "by_expires_at", + prune: "canonicalTrending.pruneExpiredActionInternal", + retention: "Forty-eight hours after candidate-pool generation.", + }, + ), skillStatEvents: ephemeral( "Skill stat event log is retained only after both consumers pass it.", { diff --git a/convex/schema.ts b/convex/schema.ts index 9080af3e..9457f6a6 100644 --- a/convex/schema.ts +++ b/convex/schema.ts @@ -2722,6 +2722,7 @@ const canonicalTrendingSnapshots = defineTable({ windowEndDay: v.number(), windowStartHour: v.optional(v.number()), windowEndHour: v.optional(v.number()), + nativePoolId: v.optional(v.string()), writtenItems: v.number(), totalItems: v.optional(v.number()), sourceCounts: v.optional( @@ -2759,6 +2760,55 @@ const canonicalTrendingItems = defineTable({ .index("by_snapshot_id_and_position", ["snapshotId", "position"]) .index("by_expires_at", ["expiresAt"]); +const canonicalTrendingNativePools = defineTable({ + poolId: v.string(), + status: v.union(v.literal("building"), v.literal("ready"), v.literal("failed")), + rankingVersion: v.string(), + generatedAt: v.number(), + completedAt: v.optional(v.number()), + expiresAt: v.number(), + windowStartHour: v.number(), + windowEndHour: v.number(), + sealedGeneration: v.number(), + writtenTrendingItems: v.number(), + writtenRisingItems: v.number(), + sourceCounts: v.optional( + v.object({ + clawhubTrending: v.number(), + clawhubRising: v.number(), + }), + ), + operations: v.optional( + v.object({ + documentsRead: v.number(), + documentsWritten: v.number(), + functionCalls: v.number(), + }), + ), + error: v.optional(v.string()), +}) + .index("by_pool_id", ["poolId"]) + .index("by_status_and_generated_at", ["status", "generatedAt"]) + .index("by_expires_at", ["expiresAt"]); + +const canonicalTrendingNativePoolItems = defineTable({ + poolId: v.string(), + lane: v.union(v.literal("clawhub-trending"), v.literal("clawhub-rising")), + position: v.number(), + identity: v.string(), + publisherKey: v.string(), + installs24h: v.number(), + bookmarks24h: v.number(), + createdAt: v.number(), + updatedAt: v.number(), + upstreamRank: v.union(v.number(), v.null()), + sourceRef: canonicalTrendingSourceRefValidator, + card: canonicalTrendingCardValidator, + expiresAt: v.number(), +}) + .index("by_pool_id_and_lane_and_position", ["poolId", "lane", "position"]) + .index("by_expires_at", ["expiresAt"]); + const skillStatEvents = defineTable({ skillId: v.id("skills"), kind: v.union( @@ -4263,6 +4313,8 @@ export default defineSchema({ rankingMetricImports, canonicalTrendingSnapshots, canonicalTrendingItems, + canonicalTrendingNativePools, + canonicalTrendingNativePoolItems, skillStatEvents, skillStatUpdateCursors, skillStatDocSyncLeases, diff --git a/convex/skillsShMirrorVisibility.test.ts b/convex/skillsShMirrorVisibility.test.ts index fe17cffe..4c0dad83 100644 --- a/convex/skillsShMirrorVisibility.test.ts +++ b/convex/skillsShMirrorVisibility.test.ts @@ -182,7 +182,7 @@ describe("skills.sh mirror visibility operations", () => { const t = convexTest(schema, modules); const now = Date.now(); const window = getCompletedRolling24HourWindow(now); - await t.run(async (ctx) => { + const { nativeSkillId } = await t.run(async (ctx) => { const leaderboardRunId = await ctx.db.insert("skillsShMirrorRuns", { snapshotId: "skills-sh:leaderboard:verified", sourceView: "leaderboard", @@ -353,7 +353,7 @@ describe("skills.sh mirror visibility operations", () => { createdAt: now - 10_000, updatedAt: now, }); - const nativeSkillId = await ctx.db.insert("skills", { + const createdNativeSkillId = await ctx.db.insert("skills", { slug: "native-ready", displayName: "Native ready", summary: "Native canonical Trending fixture", @@ -365,7 +365,7 @@ describe("skills.sh mirror visibility operations", () => { updatedAt: now, }); const nativeVersionId = await ctx.db.insert("skillVersions", { - skillId: nativeSkillId, + skillId: createdNativeSkillId, version: "1.0.0", changelog: "Initial", files: [], @@ -374,7 +374,7 @@ describe("skills.sh mirror visibility operations", () => { createdAt: now - 10_000, }); await ctx.db.insert("skillSearchDigest", { - skillId: nativeSkillId, + skillId: createdNativeSkillId, slug: "native-ready", displayName: "Native ready", summary: "Native canonical Trending fixture", @@ -384,7 +384,7 @@ describe("skills.sh mirror visibility operations", () => { ownerName: "Native owner", ownerDisplayName: "Native owner", latestVersionId: nativeVersionId, - latestVersionSkillId: nativeSkillId, + latestVersionSkillId: createdNativeSkillId, publicVersion: { status: "available", versionId: nativeVersionId }, tags: {}, statsInstallsAllTime: 900, @@ -392,6 +392,18 @@ describe("skills.sh mirror visibility operations", () => { createdAt: now - 10_000, updatedAt: now, }); + return { nativeSkillId: createdNativeSkillId }; + }); + + await expect( + t.action(internal.skillsShMirrorVisibility.verifyAndActivateInternal, { + actor: "codex-test", + reason: "CLAW-603 unavailable native source", + confirm: "activate-skills-sh-public-test", + }), + ).rejects.toThrow("skills.sh Trending activation snapshot failed source verification"); + + await t.run(async (ctx) => { await ctx.db.insert("skillHourlyStatStates", { key: "canonical_trending", liveStartedAt: now - 3_600_000, @@ -414,6 +426,20 @@ describe("skills.sh mirror visibility operations", () => { }); }); + await expect( + t.action(internal.skillsShMirrorVisibility.prepareNativeTrendingInternal, { + actor: "codex-test", + reason: "CLAW-603 persist native candidate pool", + confirm: "deactivate-skills-sh-public-test", + }), + ).resolves.toMatchObject({ + nativeTrending: { + status: "ready", + sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + nativePool: { reused: false }, + }, + }); + await expect( t.action(internal.skillsShMirrorVisibility.verifyAndActivateInternal, { actor: "codex-test", @@ -439,6 +465,7 @@ describe("skills.sh mirror visibility operations", () => { trendingSnapshot: { status: "ready", sourceCounts: { skillsShTrending: 1 }, + nativePool: { reused: true }, }, scansPlanned: 0, scansAdmitted: 0, @@ -457,7 +484,7 @@ describe("skills.sh mirror visibility operations", () => { runs: await ctx.db.query("skillsShMirrorRuns").withIndex("by_started_at").collect(), })); expect(activationState.mirrorControl?.activationLockToken).toBeUndefined(); - expect(activationState.snapshots).toHaveLength(1); + expect(activationState.snapshots).toHaveLength(2); const activatedLeaderboard = activationState.runs.find( (run) => run._id === activationState.mirrorControl?.latestCompletedLeaderboardRunId, ); @@ -465,7 +492,8 @@ describe("skills.sh mirror visibility operations", () => { activationState.runs.find((run) => run.sourceView === "trending")?._id, ); expect(activatedLeaderboard?.activationSnapshotId).toBe( - activationState.snapshots[0]?.snapshotId, + activationState.snapshots.find((snapshot) => snapshot.sourceCounts?.skillsShTrending === 1) + ?.snapshotId, ); expect(activatedLeaderboard?.activatedAt).toEqual(expect.any(Number)); await expect( @@ -489,14 +517,70 @@ describe("skills.sh mirror visibility operations", () => { reason: "CLAW-603 fail-closed rollback", confirm: "deactivate-skills-sh-public-test", }), - ).rejects.toThrow("native-only canonical Trending did not become ready"); + ).resolves.toMatchObject({ + enabled: false, + nativeTrending: { + status: "ready", + sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + nativePool: { poolId: expect.any(String) }, + reused: true, + }, + }); await expect( t.query(internal.canonicalTrending.getPageInternal, { cursor: null, limit: 20 }), ).resolves.toMatchObject({ status: "ok", page: { items: [{ source: "clawhub" }] }, }); + }); + + it("keeps the public gate closed when activation has to build the native pool", async () => { + const t = convexTest(schema, modules); + const now = Date.now(); + const window = getCompletedRolling24HourWindow(now); await t.run(async (ctx) => { + const leaderboardRunId = await ctx.db.insert( + "skillsShMirrorRuns", + mirrorRun({ + sourceView: "leaderboard", + sourceTotal: 1, + counts: mirrorRunCounts({ observed: 1, inserted: 1, detailsInserted: 1 }), + startedAt: now - 4_000, + completedAt: now - 3_000, + }), + ); + const trendingRunId = await ctx.db.insert( + "skillsShMirrorRuns", + mirrorRun({ + sourceView: "trending", + sourceTotal: 1, + counts: mirrorRunCounts({ observed: 1, trendingJoined: 1 }), + startedAt: now - 2_000, + completedAt: now - 1_000, + }), + ); + await ctx.db.insert( + "skillsShMirrorDigests", + digest({ + lastObservedRunId: leaderboardRunId, + trendingObservedRunId: trendingRunId, + trendingRank: 1, + trendingLifetimeInstalls: 10, + trendingObservedAt: now - 1_000, + }), + ); + await ctx.db.insert("skillsShMirrorControls", { + key: "global", + enabled: true, + paused: false, + maxRowsPerRun: 50_000, + maxRowsPerBatch: 50, + maxDetailBytes: 65_536, + latestCompletedLeaderboardRunId: leaderboardRunId, + updatedBy: "codex-test", + reason: "CLAW-603 activation fallback test", + updatedAt: now - 3_000, + }); await ctx.db.insert("skillHourlyStatStates", { key: "canonical_trending", liveStartedAt: now - 3_600_000, @@ -510,26 +594,44 @@ describe("skills.sh mirror visibility operations", () => { }); await expect( - t.action(internal.skillsShMirrorVisibility.deactivateAndMaterializeInternal, { + t.action(internal.skillsShMirrorVisibility.verifyAndActivateInternal, { actor: "codex-test", - reason: "CLAW-603 native-only rollback", - confirm: "deactivate-skills-sh-public-test", + reason: "CLAW-603 must reuse native preflight", + confirm: "activate-skills-sh-public-test", + }), + ).rejects.toThrow("skills.sh activation must reuse the verified native candidate pool"); + await expect( + t.run(async (ctx) => + ctx.db + .query("canonicalTrendingSnapshots") + .withIndex("by_kind_and_status_and_expires_at", (q) => + q.eq("kind", "skills").eq("status", "ready").gt("expiresAt", now), + ) + .order("desc") + .first(), + ), + ).resolves.toMatchObject({ sourceCounts: { skillsShTrending: 0 } }); + await expect( + t.run(async (ctx) => + ctx.db + .query("skillsShCatalogControls") + .withIndex("by_key", (q) => q.eq("key", "global")) + .unique(), + ), + ).resolves.toBeNull(); + + await expect( + t.action(internal.skillsShMirrorVisibility.verifyAndActivateInternal, { + actor: "codex-test", + reason: "CLAW-603 retry with verified native pool", + confirm: "activate-skills-sh-public-test", }), ).resolves.toMatchObject({ - ok: true, - enabled: false, - nativeTrending: { - status: "ready", - sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + activated: true, + trendingSnapshot: { + sourceCounts: { skillsShTrending: 1 }, + nativePool: { reused: true }, }, - scansPlanned: 0, - scansAdmitted: 0, - }); - await expect( - t.query(internal.canonicalTrending.getPageInternal, { cursor: null, limit: 20 }), - ).resolves.toMatchObject({ - status: "ok", - page: { items: [{ source: "clawhub" }] }, }); }); @@ -636,13 +738,30 @@ describe("skills.sh mirror visibility operations", () => { expiresAt: now + 24 * 60 * 60 * 1_000, windowStartDay: 40, windowEndDay: 40, + windowStartHour: 960, + windowEndHour: 983, + }); + await t.mutation(internal.canonicalTrending.startNativePoolInternal, { + poolId: "skills-native-preflight-existing", + generatedAt: now - 1_000, + expiresAt: now + 24 * 60 * 60 * 1_000, + windowStartHour: 960, + windowEndHour: 983, + sealedGeneration: 1, + }); + await t.mutation(internal.canonicalTrending.finalizeNativePoolInternal, { + poolId: "skills-native-preflight-existing", + completedAt: now - 500, + sourceCounts: { clawhubTrending: 0, clawhubRising: 0 }, + operations: { documentsRead: 10, documentsWritten: 2, functionCalls: 3 }, }); await t.mutation(internal.canonicalTrending.finalizeSnapshotInternal, { snapshotId: "skills-native-preflight-existing", completedAt: now - 500, totalItems: 0, - sourceCounts: { clawhubTrending: 3, clawhubRising: 2, skillsShTrending: 0 }, + sourceCounts: { clawhubTrending: 0, clawhubRising: 0, skillsShTrending: 0 }, operations: { documentsRead: 10, documentsWritten: 2, functionCalls: 3 }, + nativePoolId: "skills-native-preflight-existing", }); await expect( @@ -657,6 +776,7 @@ describe("skills.sh mirror visibility operations", () => { status: "ready", snapshotId: "skills-native-preflight-existing", sourceCounts: { skillsShTrending: 0 }, + nativePool: { poolId: "skills-native-preflight-existing" }, reused: true, }, }); @@ -766,6 +886,7 @@ describe("skills.sh mirror visibility operations", () => { leaderboardRunId, trendingRunId, snapshotId: "stale-snapshot", + nativePoolId: "stale-native-pool", expectedSkillsShTrending: 0, }), ).rejects.toThrow("activation lock or source run changed before publication"); diff --git a/convex/skillsShMirrorVisibility.ts b/convex/skillsShMirrorVisibility.ts index 0dd9e445..9faa3537 100644 --- a/convex/skillsShMirrorVisibility.ts +++ b/convex/skillsShMirrorVisibility.ts @@ -510,11 +510,16 @@ async function materializeNativeTrending( status: "ready"; snapshotId: string; sourceCounts: { clawhubTrending: number; clawhubRising: number; skillsShTrending: 0 }; + nativePool: { + poolId: string; + sourceCounts: { clawhubTrending: number; clawhubRising: number }; + operations: { documentsRead: number; documentsWritten: number; functionCalls: number }; + } | null; reused: true; } | null; // Native-only data is independent of skills.sh run chronology. The activation path // always materializes its mixed snapshot after verifying the exact imported runs. - if (reusable) return reusable; + if (reusable?.nativePool) return reusable; const nativeTrending = (await ctx.runAction( internalRefs.canonicalTrending.materializeInternal as never, { activationLockToken: lockToken, skillsShMode: "native-only" } as never, @@ -709,6 +714,7 @@ export const finalizeActivationInternal = internalMutation({ leaderboardRunId: v.id("skillsShMirrorRuns"), trendingRunId: v.id("skillsShMirrorRuns"), snapshotId: v.string(), + nativePoolId: v.string(), expectedSkillsShTrending: v.number(), }, handler: async (ctx, args) => { @@ -742,18 +748,38 @@ export const finalizeActivationInternal = internalMutation({ ) .order("desc") .first(); + const now = Date.now(); if ( !snapshot || snapshot.status !== "ready" || snapshot.snapshotId !== latestReady?.snapshotId || + snapshot.nativePoolId !== args.nativePoolId || snapshot.sourceCounts?.skillsShTrending !== args.expectedSkillsShTrending ) { throw new Error("verified skills.sh Trending snapshot is not the current ready snapshot"); } + const nativePool = await ctx.db + .query("canonicalTrendingNativePools") + .withIndex("by_pool_id", (q) => q.eq("poolId", args.nativePoolId)) + .unique(); + if ( + !nativePool || + nativePool.status !== "ready" || + nativePool.completedAt === undefined || + nativePool.expiresAt <= now || + !nativePool.sourceCounts || + !nativePool.operations || + nativePool.rankingVersion !== snapshot.rankingVersion || + nativePool.sourceCounts.clawhubTrending !== nativePool.writtenTrendingItems || + nativePool.sourceCounts.clawhubRising !== nativePool.writtenRisingItems || + nativePool.sourceCounts.clawhubTrending !== snapshot.sourceCounts?.clawhubTrending || + nativePool.sourceCounts.clawhubRising !== snapshot.sourceCounts.clawhubRising + ) { + throw new Error("verified skills.sh Trending native candidate pool is not ready"); + } const actor = args.actor.trim(); const reason = args.reason.trim(); 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, @@ -967,14 +993,25 @@ export const verifyAndActivateInternal = internalAction({ internalRefs.canonicalTrending.materializeInternal as never, { activationLockToken: lockToken } as never, )) as { - status: "ready"; - snapshotId: string; - sourceCounts: { skillsShTrending: number }; + status: "ready" | "unavailable"; + snapshotId?: string; + sourceCounts?: { skillsShTrending: number }; + nativePool?: { poolId: string; reused: boolean }; }; if ( trendingSnapshot.status !== "ready" || + !trendingSnapshot.nativePool || + !trendingSnapshot.sourceCounts || + !trendingSnapshot.snapshotId + ) { + throw new Error("skills.sh Trending activation snapshot failed source verification"); + } + if (trendingSnapshot.nativePool.reused !== true) { + throw new Error("skills.sh activation must reuse the verified native candidate pool"); + } + if ( trendingSnapshot.sourceCounts.skillsShTrending !== - corpusAudit.counts.activationRunTrendingEligible + corpusAudit.counts.activationRunTrendingEligible ) { throw new Error("skills.sh Trending activation snapshot failed source verification"); } @@ -986,6 +1023,7 @@ export const verifyAndActivateInternal = internalAction({ leaderboardRunId: leaderboardRun._id, trendingRunId: trendingRun._id, snapshotId: trendingSnapshot.snapshotId, + nativePoolId: trendingSnapshot.nativePool.poolId, expectedSkillsShTrending: trendingSnapshot.sourceCounts.skillsShTrending, } as never, )) as { diff --git a/scripts/skills-sh-catalog/sync.test.ts b/scripts/skills-sh-catalog/sync.test.ts index b67ccf87..70501462 100644 --- a/scripts/skills-sh-catalog/sync.test.ts +++ b/scripts/skills-sh-catalog/sync.test.ts @@ -52,6 +52,10 @@ function nativeTrendingPreparation() { status: "ready", snapshotId: "skills-native-before-import", sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + nativePool: { + poolId: "skills-native-before-import", + sourceCounts: { clawhubTrending: 1, clawhubRising: 1 }, + }, }, }); } @@ -191,6 +195,10 @@ describe("skills.sh synchronization runner", () => { status: "ready", snapshotId: "skills-native-after-timeout", sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + nativePool: { + poolId: "skills-native-after-timeout", + sourceCounts: { clawhubTrending: 1, clawhubRising: 1 }, + }, }, runs: [], invariants: { publicVisible: false }, @@ -244,6 +252,44 @@ describe("skills.sh synchronization runner", () => { ]); }); + it("fails closed when timed-out native readiness has a mismatched candidate pool", async () => { + const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { + const body = JSON.parse(String(init.body)) as Record; + if (body.operation === "status") { + return fetchImpl.mock.calls.length === 1 + ? response({ runs: [], invariants: { publicVisible: false } }) + : response({ + control: {}, + nativeTrending: { + status: "ready", + snapshotId: "skills-native-after-timeout", + sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + nativePool: { + poolId: "skills-different-native-pool", + sourceCounts: { clawhubTrending: 1, clawhubRising: 1 }, + }, + }, + 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 snapshot and candidate pool do not match"); + }); + 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) => { diff --git a/scripts/skills-sh-catalog/sync.ts b/scripts/skills-sh-catalog/sync.ts index 27dac3d9..49ec22b1 100644 --- a/scripts/skills-sh-catalog/sync.ts +++ b/scripts/skills-sh-catalog/sync.ts @@ -54,6 +54,32 @@ function optionalRecord(value: unknown) { return typeof value === "object" && value !== null ? (value as Record) : null; } +function assertReadyNativeTrending(value: unknown, unavailableMessage: string) { + const nativeTrending = optionalRecord(value); + const sourceCounts = optionalRecord(nativeTrending?.sourceCounts); + const snapshotId = nativeTrending?.snapshotId; + if ( + nativeTrending?.status !== "ready" || + typeof snapshotId !== "string" || + !snapshotId || + sourceCounts?.skillsShTrending !== 0 + ) { + throw new Error(unavailableMessage); + } + const nativePool = optionalRecord(nativeTrending.nativePool); + if (nativePool?.poolId !== snapshotId) { + throw new Error("native-only Trending snapshot and candidate pool do not match"); + } + const poolSourceCounts = optionalRecord(nativePool.sourceCounts); + if ( + poolSourceCounts?.clawhubTrending !== sourceCounts.clawhubTrending || + poolSourceCounts?.clawhubRising !== sourceCounts.clawhubRising + ) { + throw new Error("native-only Trending snapshot and candidate pool counts do not match"); + } + return nativeTrending; +} + function jwtExpiresAt(jwt: string) { const payload = jwt.split(".")[1]; if (!payload) throw new Error("GitHub OIDC returned a malformed token"); @@ -329,11 +355,10 @@ export async function runSkillsShSync(options: { } 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"); - } + const nativeTrending = assertReadyNativeTrending( + status.nativeTrending, + "native-only Trending preflight finished without a ready snapshot", + ); return { ok: true, nativeTrending, @@ -368,11 +393,10 @@ export async function runSkillsShSync(options: { } } if (nativeBefore) { - const nativeTrending = nativeBefore.nativeTrending as Record | undefined; - const sourceCounts = nativeTrending?.sourceCounts as Record | undefined; - if (nativeTrending?.status !== "ready" || sourceCounts?.skillsShTrending !== 0) { - throw new Error("native-only canonical Trending preflight did not become ready"); - } + assertReadyNativeTrending( + nativeBefore.nativeTrending, + "native-only canonical Trending preflight did not become ready", + ); } const recoveredSourceView = recoverable?.sourceView ?? "leaderboard"; const recovered = recoverable ? await completeRun(recoverable, recoveredSourceView) : null;