fix: repair pending vt skill versions (#2442)

convex/schema.ts

@@ -729,6 +729,7 @@ const skillVersions = defineTable({
   .index("by_skill", ["skillId"])
   .index("by_skill_version", ["skillId", "version"])
   .index("by_active_created", ["softDeletedAt", "createdAt"])
+  .index("by_active_vt_status_created", ["softDeletedAt", "vtAnalysis.status", "createdAt"])
   .index("by_sha256hash", ["sha256hash"])
   .index("by_dep_registry_scan_status_and_created", ["depRegistryScanStatus", "createdAt"]);
 

convex/skills.pendingScanQueue.test.ts

@@ -9,6 +9,7 @@ import { MODERATION_ENGINE_VERSION } from "./lib/moderationReasonCodes";
 import {
   getActiveSkillBatchForStaticScanBackfillInternal,
   getPendingScanSkillsInternal,
+  getPendingVTSkillsInternal,
 } from "./skills";
 
 type PendingScanResult = Array<{
@@ -18,6 +19,18 @@ type PendingScanResult = Array<{
   checkCount: number;
 }>;
 
+type PendingVtRepairResult = {
+  skills: Array<{
+    skillId: string;
+    versionId: string;
+    sha256hash: string;
+    slug: string;
+    isLatest: boolean;
+  }>;
+  cursor: string | null;
+  done: boolean;
+};
+
 type WrappedHandler<TArgs, TResult> = {
   _handler: (ctx: unknown, args: TArgs) => Promise<TResult>;
 };
@@ -40,6 +53,13 @@ const getStaticScanBackfillBatchHandler = (
   >
 )._handler;
 
+const getPendingVTSkillsHandler = (
+  getPendingVTSkillsInternal as unknown as WrappedHandler<
+    { limit?: number; cursor?: string | null },
+    PendingVtRepairResult
+  >
+)._handler;
+
 describe("skills.getPendingScanSkillsInternal", () => {
   it("includes unresolved VT records from the oldest slice and skips finalized ones", async () => {
     const recentSkills = [
@@ -230,6 +250,118 @@ describe("skills.getPendingScanSkillsInternal", () => {
   });
 });
 
+describe("skills.getPendingVTSkillsInternal", () => {
+  it("selects pending VT cache rows from skill versions, not the skill moderation queue", async () => {
+    const page = [
+      {
+        _id: "skillVersions:historical",
+        skillId: "skills:demo",
+        sha256hash: "a".repeat(64),
+        vtAnalysis: { status: "pending" },
+      },
+      {
+        _id: "skillVersions:no-hash",
+        skillId: "skills:no-hash",
+        vtAnalysis: { status: "pending" },
+      },
+      {
+        _id: "skillVersions:deleted-skill",
+        skillId: "skills:deleted",
+        sha256hash: "b".repeat(64),
+        vtAnalysis: { status: "pending" },
+      },
+    ];
+    const skills = new Map<string, unknown>([
+      [
+        "skills:demo",
+        {
+          _id: "skills:demo",
+          slug: "demo",
+          latestVersionId: "skillVersions:latest",
+        },
+      ],
+      [
+        "skills:no-hash",
+        {
+          _id: "skills:no-hash",
+          slug: "no-hash",
+          latestVersionId: "skillVersions:no-hash",
+        },
+      ],
+      [
+        "skills:deleted",
+        {
+          _id: "skills:deleted",
+          slug: "deleted",
+          latestVersionId: "skillVersions:deleted-skill",
+          softDeletedAt: 123,
+        },
+      ],
+    ]);
+    const eqCalls: Array<[string, unknown]> = [];
+
+    const ctx = {
+      db: {
+        query: vi.fn((table: string) => {
+          if (table !== "skillVersions") throw new Error(`unexpected table ${table}`);
+          return {
+            withIndex: (
+              indexName: string,
+              builder: (q: { eq: (field: string, value: unknown) => unknown }) => unknown,
+            ) => {
+              if (indexName !== "by_active_vt_status_created") {
+                throw new Error(`unexpected index ${indexName}`);
+              }
+              type EqBuilder = { eq: (field: string, value: unknown) => EqBuilder };
+              const q: EqBuilder = {
+                eq: (field, value) => {
+                  eqCalls.push([field, value]);
+                  return q;
+                },
+              };
+              builder(q);
+              return {
+                paginate: async (paginationOpts: { cursor: string | null; numItems: number }) => {
+                  expect(paginationOpts).toEqual({ cursor: "cursor-1", numItems: 25 });
+                  return {
+                    page,
+                    continueCursor: "cursor-2",
+                    isDone: false,
+                  };
+                },
+              };
+            },
+          };
+        }),
+        get: vi.fn(async (id: string) => skills.get(id) ?? null),
+      },
+    };
+
+    const result = await getPendingVTSkillsHandler(ctx, {
+      limit: 25,
+      cursor: "cursor-1",
+    });
+
+    expect(eqCalls).toEqual([
+      ["softDeletedAt", undefined],
+      ["vtAnalysis.status", "pending"],
+    ]);
+    expect(result).toEqual({
+      skills: [
+        {
+          skillId: "skills:demo",
+          versionId: "skillVersions:historical",
+          slug: "demo",
+          sha256hash: "a".repeat(64),
+          isLatest: false,
+        },
+      ],
+      cursor: "cursor-2",
+      done: false,
+    });
+  });
+});
+
 describe("skills.getActiveSkillBatchForStaticScanBackfillInternal", () => {
   it("includes latest active skills with missing or stale static scan engine versions", async () => {
     const skills = [

convex/skills.ts

@@ -6640,8 +6640,7 @@ export const getSkillsWithStaleModerationReasonInternal = internalQuery({
 });
 
 /**
- * Get skills with scanner.vt.pending that need reanalysis.
- * Returns skills regardless of whether they have vtAnalysis cached.
+ * Get skill versions with pending VT cache rows that need reanalysis.
  */
 export const getPendingVTSkillsInternal = internalQuery({
   args: { limit: v.optional(v.number()), cursor: v.optional(v.union(v.string(), v.null())) },
@@ -6649,9 +6648,9 @@ export const getPendingVTSkillsInternal = internalQuery({
     const limit = args.limit ?? 100;
 
     const { page, continueCursor, isDone } = await ctx.db
-      .query("skills")
-      .withIndex("by_moderation", (q) =>
-        q.eq("moderationStatus", "active").eq("moderationReason", "scanner.vt.pending"),
+      .query("skillVersions")
+      .withIndex("by_active_vt_status_created", (q) =>
+        q.eq("softDeletedAt", undefined).eq("vtAnalysis.status", "pending"),
       )
       .paginate({ cursor: args.cursor ?? null, numItems: limit });
 
@@ -6660,18 +6659,20 @@ export const getPendingVTSkillsInternal = internalQuery({
       versionId: Id<"skillVersions">;
       slug: string;
       sha256hash: string;
+      isLatest: boolean;
     }> = [];
 
-    for (const skill of page) {
-      if (!skill.latestVersionId) continue;
-      const version = await ctx.db.get(skill.latestVersionId);
-      if (!version?.sha256hash) continue;
+    for (const version of page) {
+      if (!version.sha256hash) continue;
+      const skill = await ctx.db.get(version.skillId);
+      if (!skill || skill.softDeletedAt) continue;
 
       results.push({
         skillId: skill._id,
         versionId: version._id,
         slug: skill.slug,
         sha256hash: version.sha256hash,
+        isLatest: skill.latestVersionId === version._id,
       });
     }
 

convex/vt.test.ts

@@ -1250,6 +1250,129 @@ describe("vt pending repair", () => {
     );
   });
 
+  it("repairs historical pending VT cache rows without recomputing latest moderation", async () => {
+    process.env.VT_API_KEY = "test-key";
+    const hash = "a".repeat(64);
+    vi.stubGlobal(
+      "fetch",
+      vi.fn().mockResolvedValue({
+        ok: true,
+        json: async () => ({
+          data: {
+            attributes: {
+              last_analysis_stats: {
+                malicious: 0,
+                suspicious: 0,
+                harmless: 2,
+                undetected: 64,
+              },
+            },
+          },
+        }),
+      }),
+    );
+
+    const runMutation = vi.fn(async () => null);
+    const result = await repairPendingSkillVtAnalysisHandler(
+      {
+        runQuery: vi.fn().mockResolvedValue({
+          skills: [
+            {
+              skillId: "skills:pending",
+              versionId: "skillVersions:historical",
+              slug: "pending-skill",
+              sha256hash: hash,
+              isLatest: false,
+            },
+          ],
+          cursor: null,
+          done: true,
+        }),
+        runMutation,
+      } as never,
+      { dryRun: false, batchSize: 100 },
+    );
+
+    expect(result).toMatchObject({
+      wouldUpdate: 1,
+      updated: 1,
+      statusCounts: { clean: 1 },
+    });
+    expect(mutationPayloads(runMutation)).toContainEqual(
+      expect.objectContaining({
+        versionId: "skillVersions:historical",
+        sha256hash: hash,
+        vtAnalysis: expect.objectContaining({ status: "clean" }),
+      }),
+    );
+    expect(mutationPayloads(runMutation)).not.toContainEqual(
+      expect.objectContaining({ skillId: "skills:pending" }),
+    );
+    expect(mutationPayloads(runMutation)).not.toContainEqual(
+      expect.objectContaining({ source: "vt-update" }),
+    );
+  });
+
+  it("does not enqueue ClawScan follow-up for suspicious historical VT cache rows", async () => {
+    process.env.VT_API_KEY = "test-key";
+    const hash = "b".repeat(64);
+    vi.stubGlobal(
+      "fetch",
+      vi.fn().mockResolvedValue({
+        ok: true,
+        json: async () => ({
+          data: {
+            attributes: {
+              last_analysis_stats: {
+                malicious: 0,
+                suspicious: 1,
+                harmless: 1,
+                undetected: 64,
+              },
+            },
+          },
+        }),
+      }),
+    );
+
+    const runMutation = vi.fn(async () => null);
+    const result = await repairPendingSkillVtAnalysisHandler(
+      {
+        runQuery: vi.fn().mockResolvedValue({
+          skills: [
+            {
+              skillId: "skills:pending",
+              versionId: "skillVersions:historical",
+              slug: "pending-skill",
+              sha256hash: hash,
+              isLatest: false,
+            },
+          ],
+          cursor: null,
+          done: true,
+        }),
+        runMutation,
+      } as never,
+      { dryRun: false, batchSize: 100 },
+    );
+
+    expect(result).toMatchObject({
+      wouldUpdate: 1,
+      updated: 1,
+      statusCounts: { suspicious: 1 },
+    });
+    expect(mutationPayloads(runMutation)).toContainEqual(
+      expect.objectContaining({
+        versionId: "skillVersions:historical",
+        sha256hash: hash,
+        vtAnalysis: expect.objectContaining({ status: "suspicious" }),
+      }),
+    );
+    expect(mutationPayloads(runMutation)).not.toContainEqual(
+      expect.objectContaining({ source: "vt-update" }),
+    );
+  });
+
   it("returns pagination cursor when unresolved pending VT rows are skipped", async () => {
     process.env.VT_API_KEY = "test-key";
     vi.stubGlobal(

convex/vt.ts

@@ -307,6 +307,7 @@ type PendingVTSkill = {
   versionId: Id<"skillVersions">;
   sha256hash: string;
   slug: string;
+  isLatest?: boolean;
 };
 
 type NullModerationStatusSkill = {
@@ -1061,7 +1062,13 @@ export const repairPendingSkillVtAnalysis = internalAction({
     const statusCounts: Record<string, number> = {};
     const sampleUpdated: Array<{ slug: string; status: string }> = [];
 
-    async function repairSkill({ skillId, versionId, sha256hash, slug }: PendingVTSkill) {
+    async function repairSkill({
+      skillId,
+      versionId,
+      sha256hash,
+      slug,
+      isLatest = true,
+    }: PendingVTSkill) {
       try {
         const vtResult = await checkExistingFile(vtApiKey, sha256hash);
         if (!vtResult) {
@@ -1091,6 +1098,11 @@ export const repairPendingSkillVtAnalysis = internalAction({
             checkedAt: Date.now(),
           },
         });
+        if (!isLatest) {
+          updated++;
+          return;
+        }
+
         if (status === "malicious" || status === "suspicious") {
           await enqueueSkillCodexForVtSignal(ctx, versionId);
         } else {