fix: persist native trending activation pool (#3360)

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;

convex/canonicalTrending.test.ts

@@ -104,6 +104,47 @@ async function insertEligibleNativeSource(t: ReturnType<typeof convexTest>, slug
   });
 }
 
+async function insertReadyNativePool(
+  t: ReturnType<typeof convexTest>,
+  input: {
+    poolId: string;
+    skillId: Awaited<ReturnType<typeof insertEligibleNativeSource>>["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",
+        }),
+      ],
+    });
   });
 });

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,
     };
   },

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()

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.",
     {

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,

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");

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 {

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<string, unknown>;
+      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) => {

scripts/skills-sh-catalog/sync.ts

@@ -54,6 +54,32 @@ function optionalRecord(value: unknown) {
   return typeof value === "object" && value !== null ? (value as Record<string, unknown>) : 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<string, unknown> | undefined;
-      const sourceCounts = nativeTrending?.sourceCounts as Record<string, unknown> | 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;