fix: recover scheduled temporal abuse scans (#3176)

* fix: recover scheduled temporal abuse scans

* test: strengthen temporal scan regression proof

Co-authored-by: Jesse Merhi <79823012+jesse-merhi@users.noreply.github.com>

* fix(ci): pin design system source commit

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

CHANGELOG.md

@@ -9,6 +9,7 @@
 
 ### Fixes
 
+- Security: recover scheduled temporal publisher-abuse scans from strict Convex payload validation failures without leaving zombie running runs (thanks @jesse-merhi).
 - CI: retry transient Convex preview provisioning failures under fresh deployment names during Vercel preview builds.
 - Web: keep CLI device login codes out of the GitHub OAuth code handler — the device page no longer loses its prefilled code, bounces through a surprise GitHub redirect, or drops an active session; device links now use `user_code`.
 - API: keep successful rate-limit checks available when retention metadata writes contend, while preserving fail-closed enforcement for authoritative counter conflicts.

bun.lock

@@ -14,7 +14,7 @@
         "@fontsource/manrope": "5.2.8",
         "@fontsource/noto-sans-sc": "5.2.9",
         "@monaco-editor/react": "4.7.0",
-        "@openclaw/design-system": "git+https://github.com/openclaw/design-system.git#v0.0.1",
+        "@openclaw/design-system": "git+https://github.com/openclaw/design-system.git#7b097d79eef6e9a0a4632f72727ac7450f07a1f2",
         "@openclaw/plugin-inspector": "0.3.17",
         "@radix-ui/react-avatar": "1.2.2",
         "@radix-ui/react-dialog": "1.1.19",
@@ -431,7 +431,7 @@
 
     "@openclaw/clawhub-admin": ["@openclaw/clawhub-admin@workspace:packages/clawhub-admin"],
 
-    "@openclaw/design-system": ["@openclaw/design-system@github:openclaw/design-system#b1774c3", {}, "openclaw-design-system-b1774c3"],
+    "@openclaw/design-system": ["@openclaw/design-system@github:openclaw/design-system#7b097d7", {}, "openclaw-carapace-7b097d7"],
 
     "@openclaw/plugin-inspector": ["@openclaw/plugin-inspector@0.3.17", "", { "bin": { "plugin-inspector": "src/cli.js" } }, "sha512-JPPHPhiXMsIvrV8UR8RQjhflMjRZX/uIhy9meE81dup7MMSnRJcsTGOXYACohv6e4z2P95z2QuE7nZkWT6Ysuw=="],
 

convex/publisherAbuseTemporalScan.test.ts

@@ -10,6 +10,7 @@ import {
 import type { TemporalSkillCandidate } from "./publisherAbuse";
 import {
   advanceScheduledTemporalCandidatesInternalHandler,
+  markScheduledTemporalScanFailedInternalHandler,
   percentileIndex,
   pruneExpiredTemporalScanRowsInternalHandler,
   runScheduledTemporalPublisherAbuseScanInternalHandler,
@@ -208,6 +209,90 @@ describe("scheduled temporal publisher abuse scan", () => {
     );
   });
 
+  it("passes only percentile inputs to the persisted benchmark sample validator", async () => {
+    const run = temporalRun();
+    const fullScore = temporalScore({ recent30Downloads: 3_000, spikeMultiplier: 4 });
+    const runQuery = vi
+      .fn()
+      .mockResolvedValueOnce(run)
+      .mockResolvedValueOnce({
+        benchmarkScores: [fullScore],
+        candidates: [],
+        cursor: "next-page",
+        isDone: false,
+        scannedSkills: 1,
+      });
+    const runMutation = vi.fn(async (_target: unknown, _args: unknown) => ({ applied: true }));
+    const scheduler = { runAfter: vi.fn(async () => null) };
+    const handler = runScheduledTemporalPublisherAbuseScanInternalHandler as unknown as (
+      ctx: {
+        runQuery: typeof runQuery;
+        runMutation: typeof runMutation;
+        scheduler: typeof scheduler;
+      },
+      args: { runId?: Id<"publisherAbuseScoreRuns"> },
+    ) => Promise<unknown>;
+
+    await handler({ runQuery, runMutation, scheduler }, { runId: run._id });
+
+    expect(runMutation.mock.calls[0]?.[1]).toEqual({
+      runId: run._id,
+      expectedCursor: undefined,
+      nextCursor: "next-page",
+      isDone: false,
+      benchmarkScores: [{ recent30Downloads: 3_000, spikeMultiplier: 4 }],
+      candidates: [],
+    });
+  });
+
+  it("marks a scheduled scan failed when a scan step throws", async () => {
+    const run = temporalRun();
+    const scanError = new Error("invalid benchmark payload");
+    const runQuery = vi.fn().mockResolvedValueOnce(run).mockRejectedValueOnce(scanError);
+    const runMutation = vi.fn(async (_target: unknown, _args: unknown) => ({ failed: true }));
+    const scheduler = { runAfter: vi.fn(async () => null) };
+    const handler = runScheduledTemporalPublisherAbuseScanInternalHandler as unknown as (
+      ctx: {
+        runQuery: typeof runQuery;
+        runMutation: typeof runMutation;
+        scheduler: typeof scheduler;
+      },
+      args: { runId?: Id<"publisherAbuseScoreRuns"> },
+    ) => Promise<unknown>;
+
+    await expect(handler({ runQuery, runMutation, scheduler }, { runId: run._id })).rejects.toThrow(
+      "invalid benchmark payload",
+    );
+
+    expect(runMutation).toHaveBeenCalledTimes(1);
+    expect(runMutation.mock.calls[0]?.[1]).toEqual({
+      runId: run._id,
+      errorMessage: "invalid benchmark payload",
+    });
+  });
+
+  it("persists a failed terminal state for an active scheduled scan", async () => {
+    const run = temporalRun();
+    const patch = vi.fn(async () => null);
+    const ctx = { db: { get: vi.fn(async () => run), patch } };
+
+    await expect(
+      markScheduledTemporalScanFailedInternalHandler(ctx as unknown as MutationCtx, {
+        runId: run._id,
+        errorMessage: "invalid benchmark payload",
+      }),
+    ).resolves.toEqual({ failed: true });
+
+    expect(patch).toHaveBeenCalledWith(
+      run._id,
+      expect.objectContaining({
+        status: "failed",
+        temporalScanComplete: false,
+        errorMessage: "invalid benchmark payload",
+      }),
+    );
+  });
+
   it("archives classified candidates with the completed full-platform benchmark", async () => {
     const benchmark = {
       scope: "all_active_skills" as const,

convex/publisherAbuseTemporalScan.ts

@@ -475,6 +475,29 @@ export const failExpiredScheduledTemporalScanInternal = internalMutation({
   handler: failExpiredScheduledTemporalScanInternalHandler,
 });
 
+export async function markScheduledTemporalScanFailedInternalHandler(
+  ctx: MutationCtx,
+  args: { runId: Id<"publisherAbuseScoreRuns">; errorMessage: string },
+) {
+  const run = await getScheduledTemporalScanStateInternalHandler(ctx, { runId: args.runId });
+  if (run.status !== "running") return { failed: false as const };
+  await ctx.db.patch(run._id, {
+    status: "failed",
+    temporalScanComplete: false,
+    errorMessage: args.errorMessage,
+    updatedAt: Date.now(),
+  });
+  return { failed: true as const };
+}
+
+export const markScheduledTemporalScanFailedInternal = internalMutation({
+  args: {
+    runId: v.id("publisherAbuseScoreRuns"),
+    errorMessage: v.string(),
+  },
+  handler: markScheduledTemporalScanFailedInternalHandler,
+});
+
 type TemporalSourcePage = {
   cursor?: string;
   isDone: boolean;
@@ -485,20 +508,28 @@ type TemporalSourcePage = {
 
 type PercentilePage = { values: number[]; cursor?: string; isDone: boolean };
 type CandidatePage = { candidates: TemporalSkillCandidate[]; cursor?: string; isDone: boolean };
+type ScheduledTemporalScanResult =
+  | { ok: true; runId: Id<"publisherAbuseScoreRuns">; completed: true }
+  | {
+      ok: false;
+      runId: Id<"publisherAbuseScoreRuns">;
+      completed: false;
+      expired: true;
+    }
+  | {
+      ok: true;
+      runId: Id<"publisherAbuseScoreRuns">;
+      completed: false;
+      phase: Exclude<TemporalScanRun["temporalPipelinePhase"], "completed">;
+    };
 
-export async function runScheduledTemporalPublisherAbuseScanInternalHandler(
+async function runScheduledTemporalPublisherAbuseScanStep(
   ctx: ActionCtx,
-  args: { runId?: Id<"publisherAbuseScoreRuns"> },
-) {
-  const start = args.runId
-    ? { runId: args.runId }
-    : await ctx.runMutation(
-        internal.publisherAbuseTemporalScan.getOrStartScheduledTemporalScanInternal,
-        {},
-      );
+  runId: Id<"publisherAbuseScoreRuns">,
+): Promise<ScheduledTemporalScanResult> {
   const run: TemporalScanRun = await ctx.runQuery(
     internal.publisherAbuseTemporalScan.getScheduledTemporalScanStateInternal,
-    { runId: start.runId },
+    { runId },
   );
   if (run.status !== "running" || run.temporalPipelinePhase === "completed") {
     return { ok: true as const, runId: run._id, completed: true as const };
@@ -526,6 +557,12 @@ export async function runScheduledTemporalPublisherAbuseScanInternalHandler(
         todayDay: run.temporalTodayDay,
       },
     );
+    const benchmarkScores = (
+      sourcePage.benchmarkScores ?? sourcePage.candidates.map(({ temporalScore }) => temporalScore)
+    ).map(({ recent30Downloads, spikeMultiplier }) => ({
+      recent30Downloads,
+      spikeMultiplier,
+    }));
     await ctx.runMutation(
       internal.publisherAbuseTemporalScan.storeScheduledTemporalScanPageInternal,
       {
@@ -533,12 +570,7 @@ export async function runScheduledTemporalPublisherAbuseScanInternalHandler(
         expectedCursor: run.temporalSourceCursor,
         nextCursor: sourcePage.cursor,
         isDone: sourcePage.isDone,
-        benchmarkScores:
-          sourcePage.benchmarkScores ??
-          sourcePage.candidates.map(({ temporalScore }) => ({
-            recent30Downloads: temporalScore.recent30Downloads,
-            spikeMultiplier: temporalScore.spikeMultiplier,
-          })),
+        benchmarkScores,
         candidates: sourcePage.candidates,
       },
     );
@@ -645,6 +677,35 @@ export async function runScheduledTemporalPublisherAbuseScanInternalHandler(
   };
 }
 
+export async function runScheduledTemporalPublisherAbuseScanInternalHandler(
+  ctx: ActionCtx,
+  args: { runId?: Id<"publisherAbuseScoreRuns"> },
+): Promise<ScheduledTemporalScanResult> {
+  const start: { runId: Id<"publisherAbuseScoreRuns"> } = args.runId
+    ? { runId: args.runId }
+    : await ctx.runMutation(
+        internal.publisherAbuseTemporalScan.getOrStartScheduledTemporalScanInternal,
+        {},
+      );
+  try {
+    return await runScheduledTemporalPublisherAbuseScanStep(ctx, start.runId);
+  } catch (error) {
+    const errorMessage = (error instanceof Error ? error.message : String(error)).slice(0, 2_000);
+    try {
+      await ctx.runMutation(
+        internal.publisherAbuseTemporalScan.markScheduledTemporalScanFailedInternal,
+        { runId: start.runId, errorMessage },
+      );
+    } catch (recordError) {
+      console.error("[publisher-temporal-abuse-scan] Failed to persist scan failure", {
+        runId: start.runId,
+        errorMessage: recordError instanceof Error ? recordError.message : String(recordError),
+      });
+    }
+    throw error;
+  }
+}
+
 export const runScheduledTemporalPublisherAbuseScanInternal = internalAction({
   args: { runId: v.optional(v.id("publisherAbuseScoreRuns")) },
   handler: runScheduledTemporalPublisherAbuseScanInternalHandler,

package.json

@@ -93,7 +93,7 @@
     "@fontsource/manrope": "5.2.8",
     "@fontsource/noto-sans-sc": "5.2.9",
     "@monaco-editor/react": "4.7.0",
-    "@openclaw/design-system": "git+https://github.com/openclaw/design-system.git#v0.0.1",
+    "@openclaw/design-system": "git+https://github.com/openclaw/design-system.git#7b097d79eef6e9a0a4632f72727ac7450f07a1f2",
     "@openclaw/plugin-inspector": "0.3.17",
     "@radix-ui/react-avatar": "1.2.2",
     "@radix-ui/react-dialog": "1.1.19",

specs/security-moderation.md

@@ -67,7 +67,9 @@ See also: [acceptable-usage.md](./acceptable-usage.md) for the marketplace polic
   opts into archived dry-run signal rows for the staff Signals tab. It persists
   bounded source pages, exact percentile samples, and review candidates, then
   resumes through percentile and classification phases. Temporary scan rows
-  expire after seven days. Explicitly bounded manual scans remain diagnostic-only.
+  expire after seven days. A scheduled scan step that fails validation or throws
+  must persist a terminal failed state instead of leaving a resumable running run.
+  Explicitly bounded manual scans remain diagnostic-only.
   The `review` label remains a calibration/manual-review signal. The
   `potential_ban_candidate` label is an
   enforcement signal only for pressure-score nominations: the first eligible