fix(moderation): cap signal scan retries (#3197)

* fix(moderation): replace stale signal scans

* fix(moderation): bound stale scan recovery

* fix(moderation): cap signal scan retries

* fix(management): show terminal signal scan failures

* docs: add signal failure UI proof

* fix(moderation): preserve signal retry status

convex/publisherAbuse.test.ts

@@ -683,7 +683,11 @@ function makeEmptyPublisherAbuseScoreRunsQuery() {
       indexName: string,
       build: (q: { eq: (field: string, value: unknown) => unknown }) => unknown,
     ) => {
-      expect(indexName).toBe("by_model_version_and_started_at");
+      expect([
+        "by_model_version_and_started_at",
+        "by_temporal_pipeline_kind_and_started_at",
+        "by_model_version_and_temporal_pipeline_kind_and_phase_started_at",
+      ]).toContain(indexName);
       const constraints: Record<string, unknown> = {};
       const q = {
         eq(field: string, value: unknown) {
@@ -692,7 +696,15 @@ function makeEmptyPublisherAbuseScoreRunsQuery() {
         },
       };
       build(q);
-      expect(constraints.modelVersion).toBeTypeOf("string");
+      if (indexName === "by_model_version_and_started_at") {
+        expect(constraints.modelVersion).toBeTypeOf("string");
+      } else if (indexName === "by_model_version_and_temporal_pipeline_kind_and_phase_started_at") {
+        expect(constraints.modelVersion).toBeTypeOf("string");
+        expect(constraints.temporalPipelineKind).toBeUndefined();
+        expect(constraints.temporalPipelinePhase).toBeTypeOf("string");
+      } else {
+        expect(constraints).toEqual({ temporalPipelineKind: "signals" });
+      }
       return {
         order: (direction: "asc" | "desc") => {
           expect(direction).toBe("desc");
@@ -807,6 +819,7 @@ describe("publisher abuse dry-run persistence", () => {
         listDashboardHandler({ db: { get: dbGet, query: dbQuery } }, {}),
       ).resolves.toEqual({
         latestRun: null,
+        latestSignalRun: null,
         pendingItems: [],
         pendingPotentialBanCandidateItems: [],
         pendingReviewItems: [],
@@ -872,6 +885,7 @@ describe("publisher abuse dry-run persistence", () => {
 
     await expect(listDashboardHandler({ db }, {})).resolves.toEqual({
       latestRun: null,
+      latestSignalRun: null,
       pendingItems: [],
       pendingPotentialBanCandidateItems: [],
       pendingReviewItems: [],
@@ -1622,6 +1636,7 @@ describe("publisher abuse dry-run persistence", () => {
 
     await expect(listDashboardHandler({ db }, {})).resolves.toEqual({
       latestRun: null,
+      latestSignalRun: null,
       pendingItems: [],
       pendingPotentialBanCandidateItems: [],
       pendingReviewItems: [],
@@ -1689,6 +1704,85 @@ describe("publisher abuse dry-run persistence", () => {
     expect(db.get).not.toHaveBeenCalled();
   });
 
+  it("keeps a resumable legacy signal scan visible behind a newer diagnostic run", async () => {
+    vi.mocked(requireUser).mockResolvedValue({
+      userId: "users:moderator",
+      user: { _id: "users:moderator", role: "moderator" },
+    } as never);
+    const legacySignalRun = {
+      _id: "publisherAbuseScoreRuns:legacy-signal",
+      modelVersion: "publisher-abuse-temporal.v1",
+      status: "running",
+      phase: "collecting",
+      trigger: "cron",
+      startedAt: 200,
+      updatedAt: 200,
+      temporalPipelinePhase: "collecting",
+    };
+    const newerDiagnosticRun = {
+      ...legacySignalRun,
+      _id: "publisherAbuseScoreRuns:newer-diagnostic",
+      startedAt: 300,
+      updatedAt: 300,
+      temporalPipelinePhase: undefined,
+    };
+    const db = {
+      get: vi.fn(async () => null),
+      query: vi.fn((table: string) => {
+        if (table === "publisherAbuseScoreRuns") {
+          return {
+            withIndex: (
+              indexName: string,
+              build: (q: { eq: (field: string, value: unknown) => unknown }) => unknown,
+            ) => {
+              const constraints: Record<string, unknown> = {};
+              const q = {
+                eq(field: string, value: unknown) {
+                  constraints[field] = value;
+                  return q;
+                },
+              };
+              build(q);
+              return {
+                order: () => ({
+                  first: async () => {
+                    if (indexName === "by_temporal_pipeline_kind_and_started_at") return null;
+                    if (
+                      indexName ===
+                      "by_model_version_and_temporal_pipeline_kind_and_phase_started_at"
+                    ) {
+                      return constraints.modelVersion === "publisher-abuse-temporal.v1" &&
+                        constraints.temporalPipelineKind === undefined &&
+                        constraints.temporalPipelinePhase === "collecting"
+                        ? legacySignalRun
+                        : null;
+                    }
+                    return constraints.modelVersion === "publisher-abuse-temporal.v1"
+                      ? newerDiagnosticRun
+                      : null;
+                  },
+                }),
+              };
+            },
+          };
+        }
+        if (table === "publisherAbuseReviewNominations") {
+          return makePublisherAbuseNominationCountQuery([]);
+        }
+        if (table === "publisherAbuseSignals") return makePublisherAbuseSignalCountQuery([]);
+        if (table === "officialPublishers") return makeEmptyOfficialPublishersQuery();
+        throw new Error(`unexpected table ${table}`);
+      }),
+    };
+
+    await expect(listDashboardHandler({ db }, {})).resolves.toEqual(
+      expect.objectContaining({
+        latestRun: expect.objectContaining({ _id: legacySignalRun._id }),
+        latestSignalRun: expect.objectContaining({ _id: legacySignalRun._id }),
+      }),
+    );
+  });
+
   it("counts only visible publisher abuse nominations on the dashboard", async () => {
     vi.mocked(requireUser).mockResolvedValue({
       userId: "users:moderator",
@@ -1830,6 +1924,7 @@ describe("publisher abuse dry-run persistence", () => {
 
     await expect(listDashboardHandler({ db }, {})).resolves.toEqual({
       latestRun: null,
+      latestSignalRun: null,
       pendingItems: [],
       pendingPotentialBanCandidateItems: [],
       pendingReviewItems: [],
@@ -1921,6 +2016,7 @@ describe("publisher abuse dry-run persistence", () => {
 
     await expect(listDashboardHandler({ db }, {})).resolves.toEqual({
       latestRun: null,
+      latestSignalRun: null,
       pendingItems: [],
       pendingPotentialBanCandidateItems: [],
       pendingReviewItems: [],

convex/publisherAbuse.ts

@@ -334,14 +334,21 @@ export const listReviewDashboard = query({
     const auth = await requirePublisherAbuseDashboardUser(ctx);
     if (!auth) return emptyPublisherAbuseReviewDashboard();
 
-    const [latestRun, nominationCountSummary, signalCountSummary] = await Promise.all([
-      getLatestPublisherAbuseScoreRun(ctx),
-      getPublisherAbuseReviewNominationCountSummary(ctx),
-      getPublisherAbuseSignalCountSummary(ctx),
-    ]);
+    const [latestPressureRun, latestSignalRun, nominationCountSummary, signalCountSummary] =
+      await Promise.all([
+        getLatestPublisherAbuseScoreRunForModel(
+          ctx,
+          DEFAULT_PUBLISHER_ABUSE_MODEL_CONFIG.modelVersion,
+        ),
+        getLatestPublisherAbuseSignalRun(ctx),
+        getPublisherAbuseReviewNominationCountSummary(ctx),
+        getPublisherAbuseSignalCountSummary(ctx),
+      ]);
+    const latestRun = newerPublisherAbuseRun(latestPressureRun, latestSignalRun);
 
     return {
       latestRun: latestRun ? summarizePublisherAbuseRun(latestRun) : null,
+      latestSignalRun: latestSignalRun ? summarizePublisherAbuseRun(latestSignalRun) : null,
       pendingItems: [],
       pendingPotentialBanCandidateItems: [],
       pendingReviewItems: [],
@@ -528,6 +535,7 @@ async function requirePublisherAbuseDashboardUser(ctx: QueryCtx) {
 function emptyPublisherAbuseReviewDashboard() {
   return {
     latestRun: null,
+    latestSignalRun: null,
     pendingItems: [],
     pendingPotentialBanCandidateItems: [],
     pendingReviewItems: [],
@@ -4119,18 +4127,10 @@ async function getPublisherAbuseReviewItemsPageFromPendingNominations(
   };
 }
 
-async function getLatestPublisherAbuseScoreRun(ctx: QueryCtx) {
-  const pressureRun = await getLatestPublisherAbuseScoreRunForModel(
-    ctx,
-    DEFAULT_PUBLISHER_ABUSE_MODEL_CONFIG.modelVersion,
-  );
-  const temporalRun = await getLatestPublisherAbuseScoreRunForModel(
-    ctx,
-    PUBLISHER_TEMPORAL_ABUSE_MODEL_VERSION,
-  );
-  if (!pressureRun) return temporalRun;
-  if (!temporalRun) return pressureRun;
-  return temporalRun.startedAt > pressureRun.startedAt ? temporalRun : pressureRun;
+function newerPublisherAbuseRun(left: ScoreRun | null, right: ScoreRun | null) {
+  if (!left) return right;
+  if (!right) return left;
+  return right.startedAt > left.startedAt ? right : left;
 }
 
 async function getLatestPublisherAbuseScoreRunForModel(ctx: QueryCtx, modelVersion: string) {
@@ -4141,6 +4141,42 @@ async function getLatestPublisherAbuseScoreRunForModel(ctx: QueryCtx, modelVersi
     .first();
 }
 
+async function getLatestPublisherAbuseSignalRun(ctx: QueryCtx) {
+  const legacyTemporalPhases = [
+    "collecting",
+    "downloads_percentiles",
+    "spike_percentiles",
+    "classifying",
+    "completed",
+  ] as const;
+  const [taggedRun, legacyRuns] = await Promise.all([
+    ctx.db
+      .query("publisherAbuseScoreRuns")
+      .withIndex("by_temporal_pipeline_kind_and_started_at", (q) =>
+        q.eq("temporalPipelineKind", "signals"),
+      )
+      .order("desc")
+      .first(),
+    Promise.all(
+      legacyTemporalPhases.map(
+        async (temporalPipelinePhase) =>
+          await ctx.db
+            .query("publisherAbuseScoreRuns")
+            .withIndex("by_model_version_and_temporal_pipeline_kind_and_phase_started_at", (q) =>
+              q
+                .eq("modelVersion", PUBLISHER_TEMPORAL_ABUSE_MODEL_VERSION)
+                .eq("temporalPipelineKind", undefined)
+                .eq("temporalPipelinePhase", temporalPipelinePhase),
+            )
+            .order("desc")
+            .first(),
+      ),
+    ),
+  ]);
+  const legacyRun = legacyRuns.reduce<ScoreRun | null>(newerPublisherAbuseRun, null);
+  return newerPublisherAbuseRun(taggedRun, legacyRun);
+}
+
 async function getRecentResolvedPublisherAbuseReviewItems(
   ctx: QueryCtx,
   limit: number,

convex/publisherAbuseTemporalScan.test.ts

@@ -13,6 +13,7 @@ import {
   advanceScheduledTemporalCandidatesInternalHandler,
   getOrStartScheduledTemporalScanInternalHandler,
   markScheduledTemporalScanFailedInternalHandler,
+  recordScheduledTemporalScanFailureInternalHandler,
   percentileIndex,
   pruneExpiredTemporalScanRowsInternalHandler,
   runScheduledTemporalPublisherAbuseScanInternalHandler,
@@ -197,6 +198,7 @@ describe("scheduled temporal publisher abuse scan", () => {
         })),
         insert,
       },
+      scheduler: { runAfter: vi.fn(async () => null) },
     };
 
     await expect(
@@ -220,10 +222,16 @@ describe("scheduled temporal publisher abuse scan", () => {
     );
   });
 
-  it("does not take over an active cron scan when a moderator requests a rescan", async () => {
+  it("does not replace a recently quiet cron scan when a moderator requests a rescan", async () => {
     const actorUserId = "users:moderator" as Id<"users">;
-    const existing = temporalRun({ trigger: "cron", actorUserId: undefined });
-    const patch = vi.fn(async () => null);
+    const now = Date.now();
+    const existing = temporalRun({
+      trigger: "cron",
+      actorUserId: undefined,
+      startedAt: now - 4 * 24 * 60 * 60 * 1_000,
+      updatedAt: now - 14 * 60 * 1_000,
+    });
+    const patch = vi.fn(async (_id: unknown, _value: unknown) => null);
     const ctx = {
       db: {
         query: vi.fn(() => ({
@@ -245,6 +253,55 @@ describe("scheduled temporal publisher abuse scan", () => {
     expect(patch).not.toHaveBeenCalled();
   });
 
+  it("retries the same signal scan after fifteen minutes without progress", async () => {
+    const now = Date.now();
+    const existing = temporalRun({
+      temporalPipelineKind: undefined,
+      startedAt: now - 4 * 24 * 60 * 60 * 1_000,
+      updatedAt: now - 16 * 60 * 1_000,
+    });
+    const patch = vi.fn(async (_id: unknown, _value: unknown) => null);
+    const insert = vi.fn();
+    const scheduler = { runAfter: vi.fn(async () => null) };
+    let queryCount = 0;
+    const ctx = {
+      db: {
+        query: vi.fn(() => ({
+          withIndex: vi.fn(() => ({
+            order: vi.fn(() => ({
+              first: vi.fn(async () => {
+                queryCount += 1;
+                return queryCount === 1 ? null : existing;
+              }),
+            })),
+          })),
+        })),
+        get: vi.fn(async () => existing),
+        patch,
+        insert,
+      },
+      scheduler,
+    };
+
+    await expect(
+      getOrStartScheduledTemporalScanInternalHandler(ctx as unknown as MutationCtx, {
+        trigger: "manual",
+        actorUserId: "users:moderator" as Id<"users">,
+      }),
+    ).resolves.toEqual({ runId: existing._id, resumed: true });
+
+    expect(patch).toHaveBeenCalledWith(
+      existing._id,
+      expect.objectContaining({
+        transientErrorCount: 1,
+        lastTransientError: expect.stringContaining("fifteen minutes"),
+      }),
+    );
+    expect(patch.mock.calls[0]?.[1]).not.toHaveProperty("status");
+    expect(insert).not.toHaveBeenCalled();
+    expect(scheduler.runAfter).toHaveBeenCalledTimes(2);
+  });
+
   it("does not launch a second worker for an already-running signal scan", async () => {
     const existing = temporalRun({ trigger: "cron" });
     const runMutation = vi.fn(async () => ({ runId: existing._id, resumed: true }));
@@ -347,8 +404,13 @@ describe("scheduled temporal publisher abuse scan", () => {
     });
   });
 
-  it("persists one bounded source page and advances its durable cursor", async () => {
-    const run = temporalRun();
+  it("persists one bounded source page, advances its cursor, and clears the failure streak", async () => {
+    const run = temporalRun({
+      transientErrorCount: 2,
+      lastTransientError: "previous failure",
+      lastTransientErrorAt: Date.now() - 1_000,
+      nextTransientRetryAt: Date.now() + 30_000,
+    });
     const insert = vi.fn(async () => "inserted");
     const patch = vi.fn(async () => null);
     const ctx = {
@@ -390,6 +452,10 @@ describe("scheduled temporal publisher abuse scan", () => {
         temporalSampleSize: 2,
         temporalDownloadsSum: 100,
         temporalPipelinePhase: "collecting",
+        transientErrorCount: 0,
+        lastTransientError: undefined,
+        lastTransientErrorAt: undefined,
+        nextTransientRetryAt: undefined,
       }),
     );
   });
@@ -459,11 +525,14 @@ describe("scheduled temporal publisher abuse scan", () => {
     });
   });
 
-  it("marks a scheduled scan failed when a scan step throws", async () => {
+  it("retries a scheduled scan step failure without discarding saved progress", 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 runMutation = vi.fn(async (_target: unknown, _args: unknown) => ({
+      outcome: "retry_scheduled",
+      failureCount: 1,
+    }));
     const scheduler = { runAfter: vi.fn(async () => null) };
     const handler = runScheduledTemporalPublisherAbuseScanInternalHandler as unknown as (
       ctx: {
@@ -474,17 +543,78 @@ describe("scheduled temporal publisher abuse scan", () => {
       args: { runId?: Id<"publisherAbuseScoreRuns"> },
     ) => Promise<unknown>;
 
-    await expect(handler({ runQuery, runMutation, scheduler }, { runId: run._id })).rejects.toThrow(
-      "invalid benchmark payload",
-    );
+    await expect(
+      handler({ runQuery, runMutation, scheduler }, { runId: run._id }),
+    ).resolves.toEqual({
+      ok: true,
+      runId: run._id,
+      completed: false,
+      phase: "collecting",
+      retrying: true,
+    });
 
     expect(runMutation).toHaveBeenCalledTimes(1);
     expect(runMutation.mock.calls[0]?.[1]).toEqual({
       runId: run._id,
+      expectedUpdatedAt: run.updatedAt,
       errorMessage: "invalid benchmark payload",
     });
   });
 
+  it("stops a scheduled scan after its fifth consecutive failed attempt", async () => {
+    const run = temporalRun({
+      transientErrorCount: 4,
+      lastTransientError: "fourth failure",
+    });
+    const patch = vi.fn(async () => null);
+    const scheduler = { runAfter: vi.fn(async () => null) };
+    const ctx = { db: { get: vi.fn(async () => run), patch }, scheduler };
+
+    await expect(
+      recordScheduledTemporalScanFailureInternalHandler(ctx as unknown as MutationCtx, {
+        runId: run._id,
+        expectedUpdatedAt: run.updatedAt,
+        errorMessage: "fifth failure",
+      }),
+    ).resolves.toEqual({ outcome: "failed", failureCount: 5 });
+
+    expect(patch).toHaveBeenCalledWith(
+      run._id,
+      expect.objectContaining({
+        status: "failed",
+        temporalScanComplete: false,
+        transientErrorCount: 5,
+        lastTransientError: "fifth failure",
+        errorMessage: "fifth failure",
+      }),
+    );
+    expect(scheduler.runAfter).not.toHaveBeenCalled();
+  });
+
+  it("schedules the next saved-page attempt after a non-terminal failure", async () => {
+    const run = temporalRun({ transientErrorCount: 2 });
+    const patch = vi.fn(async () => null);
+    const scheduler = { runAfter: vi.fn(async () => null) };
+    const ctx = { db: { get: vi.fn(async () => run), patch }, scheduler };
+
+    await expect(
+      recordScheduledTemporalScanFailureInternalHandler(ctx as unknown as MutationCtx, {
+        runId: run._id,
+        expectedUpdatedAt: run.updatedAt,
+        errorMessage: "third failure",
+      }),
+    ).resolves.toEqual({ outcome: "retry_scheduled", failureCount: 3 });
+
+    expect(patch).toHaveBeenCalledWith(
+      run._id,
+      expect.objectContaining({
+        transientErrorCount: 3,
+        lastTransientError: "third failure",
+      }),
+    );
+    expect(scheduler.runAfter).toHaveBeenCalledWith(120_000, expect.anything(), { runId: run._id });
+  });
+
   it("persists a failed terminal state for an active scheduled scan", async () => {
     const run = temporalRun();
     const patch = vi.fn(async () => null);

convex/publisherAbuseTemporalScan.ts

@@ -23,6 +23,10 @@ const SOURCE_PAGE_SIZE = 50;
 const PERCENTILE_PAGE_SIZE = 500;
 const CANDIDATE_PAGE_SIZE = 100;
 const TEMPORAL_SCAN_RETENTION_MS = 7 * 24 * 60 * 60 * 1000;
+const TEMPORAL_SCAN_HEARTBEAT_TIMEOUT_MS = 15 * 60 * 1000;
+const TEMPORAL_SCAN_RETRY_BASE_DELAY_MS = 30 * 1000;
+const TEMPORAL_SCAN_RETRY_MAX_DELAY_MS = 5 * 60 * 1000;
+const MAX_TEMPORAL_SCAN_FAILURE_ATTEMPTS = 5;
 
 const temporalCohortBandValidator = v.union(v.literal("p95"), v.literal("p99"));
 const temporalScoreValidator = v.object({
@@ -86,6 +90,19 @@ function isActiveScheduledTemporalRun(run: TemporalScanRun, now: number) {
   return run.status === "running" && now - run.startedAt < TEMPORAL_SCAN_RETENTION_MS;
 }
 
+function temporalScanRetryDelayMs(failureCount: number) {
+  return Math.min(
+    TEMPORAL_SCAN_RETRY_BASE_DELAY_MS * 2 ** Math.max(0, failureCount - 1),
+    TEMPORAL_SCAN_RETRY_MAX_DELAY_MS,
+  );
+}
+
+function temporalScanHeartbeatDueAt(run: TemporalScanRun) {
+  return (
+    Math.max(run.updatedAt, run.nextTransientRetryAt ?? 0) + TEMPORAL_SCAN_HEARTBEAT_TIMEOUT_MS
+  );
+}
+
 export async function getOrStartScheduledTemporalScanInternalHandler(
   ctx: MutationCtx,
   args: { trigger?: "cron" | "manual"; actorUserId?: Id<"users"> },
@@ -112,13 +129,38 @@ export async function getOrStartScheduledTemporalScanInternalHandler(
         .first();
   const existing =
     currentPipeline ?? (legacyCronPipeline?.temporalPipelinePhase ? legacyCronPipeline : null);
+  const withinWorkingStateRetention =
+    existing !== null && now - existing.startedAt < TEMPORAL_SCAN_RETENTION_MS;
+  const shouldRetryStaleRun =
+    existing?.temporalPipelinePhase !== undefined &&
+    existing.temporalPipelinePhase !== "completed" &&
+    withinWorkingStateRetention &&
+    now >= temporalScanHeartbeatDueAt(existing);
   if (
     existing?.temporalPipelinePhase &&
     existing.temporalPipelinePhase !== "completed" &&
-    now - existing.startedAt < TEMPORAL_SCAN_RETENTION_MS
+    withinWorkingStateRetention &&
+    !shouldRetryStaleRun
   ) {
     return { runId: existing._id, resumed: true as const };
   }
+  if (existing && shouldRetryStaleRun) {
+    const retry = await recordScheduledTemporalScanFailureInternalHandler(ctx, {
+      runId: existing._id,
+      expectedUpdatedAt: existing.updatedAt,
+      errorMessage: "Signal scan reported no progress for fifteen minutes.",
+    });
+    if (retry.outcome === "retry_scheduled") {
+      await ctx.scheduler.runAfter(
+        temporalScanRetryDelayMs(retry.failureCount) + TEMPORAL_SCAN_HEARTBEAT_TIMEOUT_MS,
+        internal.publisherAbuseTemporalScan.monitorScheduledTemporalScanInternal,
+        { runId: existing._id },
+      );
+    }
+    // On failure five, the action re-reads this run and surfaces its terminal
+    // error. A later independent trigger may start a new scan; this one stops.
+    return { runId: existing._id, resumed: true as const };
+  }
   if (existing) {
     await ctx.db.patch(existing._id, {
       status: "failed",
@@ -154,6 +196,11 @@ export async function getOrStartScheduledTemporalScanInternalHandler(
     temporalDownloadsProcessed: 0,
     temporalSpikeProcessed: 0,
   });
+  await ctx.scheduler.runAfter(
+    TEMPORAL_SCAN_HEARTBEAT_TIMEOUT_MS,
+    internal.publisherAbuseTemporalScan.monitorScheduledTemporalScanInternal,
+    { runId },
+  );
   return { runId, resumed: false as const };
 }
 
@@ -223,6 +270,10 @@ export async function storeScheduledTemporalScanPageInternalHandler(
       (run.temporalDownloadsSum ?? 0) +
       args.benchmarkScores.reduce((sum, score) => sum + Math.max(0, score.recent30Downloads), 0),
     temporalPipelinePhase: args.isDone ? "downloads_percentiles" : "collecting",
+    transientErrorCount: 0,
+    lastTransientError: undefined,
+    lastTransientErrorAt: undefined,
+    nextTransientRetryAt: undefined,
     updatedAt: now,
   });
   return { applied: true as const };
@@ -325,6 +376,10 @@ export async function advanceScheduledTemporalPercentileInternalHandler(
       temporalDownloadsP95: args.p95 ?? run.temporalDownloadsP95,
       temporalDownloadsP99: args.p99 ?? run.temporalDownloadsP99,
       temporalPipelinePhase: args.isDone ? "spike_percentiles" : args.phase,
+      transientErrorCount: 0,
+      lastTransientError: undefined,
+      lastTransientErrorAt: undefined,
+      nextTransientRetryAt: undefined,
       updatedAt: now,
     });
     return { applied: true as const };
@@ -341,6 +396,10 @@ export async function advanceScheduledTemporalPercentileInternalHandler(
     temporalSpikeP99: spikeP99,
     temporalBenchmark: benchmark,
     temporalPipelinePhase: args.isDone ? "classifying" : args.phase,
+    transientErrorCount: 0,
+    lastTransientError: undefined,
+    lastTransientErrorAt: undefined,
+    nextTransientRetryAt: undefined,
     updatedAt: now,
   });
   return { applied: true as const };
@@ -456,6 +515,10 @@ export async function advanceScheduledTemporalCandidatesInternalHandler(
     completedAt: args.isDone ? now : undefined,
     finalizedScores,
     reviewCount: finalizedScores,
+    transientErrorCount: 0,
+    lastTransientError: undefined,
+    lastTransientErrorAt: undefined,
+    nextTransientRetryAt: undefined,
     updatedAt: now,
   });
   return { applied: true as const };
@@ -518,6 +581,119 @@ export const markScheduledTemporalScanFailedInternal = internalMutation({
   handler: markScheduledTemporalScanFailedInternalHandler,
 });
 
+type ScheduledTemporalScanFailureResult =
+  | { outcome: "inactive" }
+  | { outcome: "retry_scheduled"; failureCount: number }
+  | { outcome: "failed"; failureCount: number };
+
+export async function recordScheduledTemporalScanFailureInternalHandler(
+  ctx: MutationCtx,
+  args: {
+    runId: Id<"publisherAbuseScoreRuns">;
+    expectedUpdatedAt: number;
+    errorMessage: string;
+  },
+): Promise<ScheduledTemporalScanFailureResult> {
+  const run = await getScheduledTemporalScanStateInternalHandler(ctx, { runId: args.runId });
+  if (run.status !== "running" || run.updatedAt !== args.expectedUpdatedAt) {
+    return { outcome: "inactive" };
+  }
+
+  const now = Date.now();
+  const failureCount = (run.transientErrorCount ?? 0) + 1;
+  const failureTelemetry = {
+    transientErrorCount: failureCount,
+    lastTransientError: args.errorMessage,
+    lastTransientErrorAt: now,
+    updatedAt: now,
+  };
+  if (failureCount >= MAX_TEMPORAL_SCAN_FAILURE_ATTEMPTS) {
+    await ctx.db.patch(run._id, {
+      ...failureTelemetry,
+      status: "failed",
+      temporalScanComplete: false,
+      errorMessage: args.errorMessage,
+      nextTransientRetryAt: undefined,
+    });
+    console.error("[publisher-temporal-abuse-scan] retry budget exhausted", {
+      event: "publisher_temporal_abuse_scan_failed",
+      runId: run._id,
+      failureCount,
+      errorMessage: args.errorMessage,
+    });
+    return { outcome: "failed", failureCount };
+  }
+
+  const retryDelayMs = temporalScanRetryDelayMs(failureCount);
+  await ctx.db.patch(run._id, {
+    ...failureTelemetry,
+    errorMessage: undefined,
+    nextTransientRetryAt: now + retryDelayMs,
+  });
+  await ctx.scheduler.runAfter(
+    retryDelayMs,
+    internal.publisherAbuseTemporalScan.runScheduledTemporalPublisherAbuseScanInternal,
+    { runId: run._id },
+  );
+  console.warn("[publisher-temporal-abuse-scan] scan step failed; retrying", {
+    runId: run._id,
+    failureCount,
+    maxFailureAttempts: MAX_TEMPORAL_SCAN_FAILURE_ATTEMPTS,
+    retryDelayMs,
+    errorMessage: args.errorMessage,
+  });
+  return { outcome: "retry_scheduled", failureCount };
+}
+
+export const recordScheduledTemporalScanFailureInternal = internalMutation({
+  args: {
+    runId: v.id("publisherAbuseScoreRuns"),
+    expectedUpdatedAt: v.number(),
+    errorMessage: v.string(),
+  },
+  handler: recordScheduledTemporalScanFailureInternalHandler,
+});
+
+export async function monitorScheduledTemporalScanInternalHandler(
+  ctx: MutationCtx,
+  args: { runId: Id<"publisherAbuseScoreRuns"> },
+) {
+  const run = await getScheduledTemporalScanStateInternalHandler(ctx, args);
+  if (run.status !== "running" || run.temporalPipelinePhase === "completed") {
+    return { outcome: "inactive" as const };
+  }
+
+  const now = Date.now();
+  const heartbeatDueAt = temporalScanHeartbeatDueAt(run);
+  if (now < heartbeatDueAt) {
+    await ctx.scheduler.runAfter(
+      heartbeatDueAt - now,
+      internal.publisherAbuseTemporalScan.monitorScheduledTemporalScanInternal,
+      args,
+    );
+    return { outcome: "waiting" as const };
+  }
+
+  const failure = await recordScheduledTemporalScanFailureInternalHandler(ctx, {
+    runId: run._id,
+    expectedUpdatedAt: run.updatedAt,
+    errorMessage: "Signal scan reported no progress for fifteen minutes.",
+  });
+  if (failure.outcome === "retry_scheduled") {
+    await ctx.scheduler.runAfter(
+      temporalScanRetryDelayMs(failure.failureCount) + TEMPORAL_SCAN_HEARTBEAT_TIMEOUT_MS,
+      internal.publisherAbuseTemporalScan.monitorScheduledTemporalScanInternal,
+      args,
+    );
+  }
+  return failure;
+}
+
+export const monitorScheduledTemporalScanInternal = internalMutation({
+  args: { runId: v.id("publisherAbuseScoreRuns") },
+  handler: monitorScheduledTemporalScanInternalHandler,
+});
+
 type TemporalSourcePage = {
   cursor?: string;
   isDone: boolean;
@@ -530,6 +706,14 @@ 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;
+      failed: true;
+      failureCount: number;
+      errorMessage: string;
+    }
   | {
       ok: false;
       runId: Id<"publisherAbuseScoreRuns">;
@@ -542,16 +726,19 @@ type ScheduledTemporalScanResult =
       completed: false;
       phase: Exclude<TemporalScanRun["temporalPipelinePhase"], "completed">;
       alreadyRunning?: true;
+      retrying?: true;
     };
 
 async function runScheduledTemporalPublisherAbuseScanStep(
   ctx: ActionCtx,
   runId: Id<"publisherAbuseScoreRuns">,
+  initialRun?: TemporalScanRun,
 ): Promise<ScheduledTemporalScanResult> {
-  const run: TemporalScanRun = await ctx.runQuery(
-    internal.publisherAbuseTemporalScan.getScheduledTemporalScanStateInternal,
-    { runId },
-  );
+  const run: TemporalScanRun =
+    initialRun ??
+    (await ctx.runQuery(internal.publisherAbuseTemporalScan.getScheduledTemporalScanStateInternal, {
+      runId,
+    }));
   if (run.status !== "running" || run.temporalPipelinePhase === "completed") {
     return { ok: true as const, runId: run._id, completed: true as const };
   }
@@ -584,7 +771,7 @@ async function runScheduledTemporalPublisherAbuseScanStep(
       recent30Downloads,
       spikeMultiplier,
     }));
-    await ctx.runMutation(
+    const stored: { applied: boolean } = await ctx.runMutation(
       internal.publisherAbuseTemporalScan.storeScheduledTemporalScanPageInternal,
       {
         runId: run._id,
@@ -595,6 +782,15 @@ async function runScheduledTemporalPublisherAbuseScanStep(
         candidates: sourcePage.candidates,
       },
     );
+    if (!stored.applied) {
+      return {
+        ok: true,
+        runId: run._id,
+        completed: false,
+        phase: run.temporalPipelinePhase,
+        alreadyRunning: true,
+      };
+    }
   } else if (
     run.temporalPipelinePhase === "downloads_percentiles" ||
     run.temporalPipelinePhase === "spike_percentiles"
@@ -629,7 +825,7 @@ async function runScheduledTemporalPublisherAbuseScanStep(
             targetIndex: percentileIndex(sampleSize, 0.5),
           })
         : undefined;
-    await ctx.runMutation(
+    const advanced: { applied: boolean } = await ctx.runMutation(
       internal.publisherAbuseTemporalScan.advanceScheduledTemporalPercentileInternal,
       {
         runId: run._id,
@@ -643,6 +839,15 @@ async function runScheduledTemporalPublisherAbuseScanStep(
         p99: sampleSize === 0 ? 0 : p99,
       },
     );
+    if (!advanced.applied) {
+      return {
+        ok: true,
+        runId: run._id,
+        completed: false,
+        phase: run.temporalPipelinePhase,
+        alreadyRunning: true,
+      };
+    }
   } else if (run.temporalPipelinePhase === "classifying") {
     if (!run.temporalBenchmark) throw new Error("Temporal scan benchmark is missing");
     const page: CandidatePage = await ctx.runQuery(
@@ -665,7 +870,7 @@ async function runScheduledTemporalPublisherAbuseScanStep(
         ({ temporalScore }) =>
           temporalScore.spike || temporalScore.sustained || temporalScore.nearConversion,
       );
-    await ctx.runMutation(
+    const advanced: { applied: boolean } = await ctx.runMutation(
       internal.publisherAbuseTemporalScan.advanceScheduledTemporalCandidatesInternal,
       {
         runId: run._id,
@@ -675,6 +880,15 @@ async function runScheduledTemporalPublisherAbuseScanStep(
         candidates: highCandidates,
       },
     );
+    if (!advanced.applied) {
+      return {
+        ok: true,
+        runId: run._id,
+        completed: false,
+        phase: run.temporalPipelinePhase,
+        alreadyRunning: true,
+      };
+    }
     if (page.isDone) {
       await ctx.scheduler.runAfter(
         0,
@@ -738,15 +952,53 @@ export async function runScheduledTemporalPublisherAbuseScanInternalHandler(
       alreadyRunning: true,
     };
   }
+  const runAtAttemptStart: TemporalScanRun = await ctx.runQuery(
+    internal.publisherAbuseTemporalScan.getScheduledTemporalScanStateInternal,
+    { runId: start.runId },
+  );
   try {
-    return await runScheduledTemporalPublisherAbuseScanStep(ctx, start.runId);
+    return await runScheduledTemporalPublisherAbuseScanStep(ctx, start.runId, runAtAttemptStart);
   } catch (error) {
     const errorMessage = (error instanceof Error ? error.message : String(error)).slice(0, 2_000);
+    const retryPhase =
+      runAtAttemptStart.temporalPipelinePhase === "completed"
+        ? "collecting"
+        : (runAtAttemptStart.temporalPipelinePhase ?? "collecting");
     try {
-      await ctx.runMutation(
-        internal.publisherAbuseTemporalScan.markScheduledTemporalScanFailedInternal,
-        { runId: start.runId, errorMessage },
+      const failure: ScheduledTemporalScanFailureResult = await ctx.runMutation(
+        internal.publisherAbuseTemporalScan.recordScheduledTemporalScanFailureInternal,
+        {
+          runId: start.runId,
+          expectedUpdatedAt: runAtAttemptStart.updatedAt,
+          errorMessage,
+        },
       );
+      if (failure.outcome === "retry_scheduled") {
+        return {
+          ok: true,
+          runId: start.runId,
+          completed: false,
+          phase: retryPhase,
+          retrying: true,
+        };
+      }
+      if (failure.outcome === "inactive") {
+        return {
+          ok: true,
+          runId: start.runId,
+          completed: false,
+          phase: retryPhase,
+          alreadyRunning: true,
+        };
+      }
+      return {
+        ok: false,
+        runId: start.runId,
+        completed: false,
+        failed: true,
+        failureCount: failure.failureCount,
+        errorMessage,
+      };
     } catch (recordError) {
       console.error("[publisher-temporal-abuse-scan] Failed to persist scan failure", {
         runId: start.runId,

convex/schema.ts

@@ -2830,6 +2830,13 @@ const publisherAbuseScoreRuns = defineTable({
     "status",
     "updatedAt",
   ])
+  .index("by_temporal_pipeline_kind_and_started_at", ["temporalPipelineKind", "startedAt"])
+  .index("by_model_version_and_temporal_pipeline_kind_and_phase_started_at", [
+    "modelVersion",
+    "temporalPipelineKind",
+    "temporalPipelinePhase",
+    "startedAt",
+  ])
   .index("by_model_status_phase_temporal_complete_started_at", [
     "modelVersion",
     "status",

specs/security-moderation.md

@@ -67,11 +67,17 @@ 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. A scheduled scan step that fails validation or throws
-  must persist a terminal failed state instead of leaving a resumable running run.
+  expire after seven days. A failed step retries from the last persisted cursor
+  with bounded backoff; any successfully persisted page resets the consecutive
+  failure count. A watchdog treats fifteen minutes without persisted progress as
+  a failed attempt. After five consecutive failed attempts, the run becomes
+  terminal, retains the last error for the staff UI, and emits the structured
+  `publisher_temporal_abuse_scan_failed` operator-alert event.
   Moderators can start this same full signal pipeline from the staff Signals tab.
   New manual starts record the actor; requests made while a temporal scan is
-  already active return that run without starting a competing worker.
+  already active return that run without starting a competing worker. Stale
+  recovery continues the same durable run instead of replacing it, and
+  cursor-guarded writes prevent overlapping late workers from duplicating work.
   Explicitly bounded manual scans remain diagnostic-only.
   The `review` label remains a calibration/manual-review signal. The
   `potential_ban_candidate` label is an

src/routes/-management.test.tsx

@@ -1024,6 +1024,73 @@ describe("Management", () => {
     expect(screen.getByText("Re-checks every active skill")).toBeTruthy();
   });
 
+  it("shows the terminal signal scan error after five failed attempts", () => {
+    searchState = { view: "abuse", tab: "signals" };
+    useQueryMock.mockImplementation((query, args) => {
+      if (args === "skip") return undefined;
+      const name = getFunctionName(query);
+      if (name === "skills:listRecentVersions") return [];
+      if (name === "skills:listReportedSkills") return [];
+      if (name === "skills:listDuplicateCandidates") return [];
+      if (name === "publisherAbuse:listReviewDashboard") {
+        return {
+          latestRun: null,
+          latestSignalRun: {
+            status: "failed",
+            scannedPublishers: 120,
+            scoredPublishers: 0,
+            transientErrorCount: 5,
+            errorMessage: "Query exceeded the document read limit.",
+          },
+          pendingItems: [],
+          pendingPotentialBanCandidateItems: [],
+          pendingReviewItems: [],
+          recentResolvedItems: [],
+        };
+      }
+      if (name === "users:list") return { items: [], total: 0 };
+      return undefined;
+    });
+
+    render(<Management />);
+
+    expect(screen.getByText("Stopped after 5 failed attempts")).toBeTruthy();
+    expect(screen.getByText("Query exceeded the document read limit.")).toBeTruthy();
+  });
+
+  it("does not show a retry warning for a completed signal scan", () => {
+    searchState = { view: "abuse", tab: "signals" };
+    useQueryMock.mockImplementation((query, args) => {
+      if (args === "skip") return undefined;
+      const name = getFunctionName(query);
+      if (name === "skills:listRecentVersions") return [];
+      if (name === "skills:listReportedSkills") return [];
+      if (name === "skills:listDuplicateCandidates") return [];
+      if (name === "publisherAbuse:listReviewDashboard") {
+        return {
+          latestRun: null,
+          latestSignalRun: {
+            status: "completed",
+            scannedPublishers: 120,
+            scoredPublishers: 12,
+            transientErrorCount: 1,
+            lastTransientError: "Temporary timeout.",
+          },
+          pendingItems: [],
+          pendingPotentialBanCandidateItems: [],
+          pendingReviewItems: [],
+          recentResolvedItems: [],
+        };
+      }
+      if (name === "users:list") return { items: [], total: 0 };
+      return undefined;
+    });
+
+    render(<Management />);
+
+    expect(screen.queryByText("Retrying after 1 of 5 failed attempts")).toBeNull();
+  });
+
   it("shows users as a separate management view", () => {
     searchState = { view: "users" };
 

src/routes/-management/AbusePage.tsx

@@ -131,6 +131,9 @@ export function AbusePage({
     }
   }, [selectedSignalId, signalItems, signalPageStatus, tab]);
   const latestRun = dashboard?.latestRun ?? null;
+  const latestSignalRun = dashboard?.latestSignalRun ?? null;
+  const displayedRun = tab === "signals" ? latestSignalRun : latestRun;
+  const signalFailureCount = latestSignalRun?.transientErrorCount ?? 0;
   const selectedScore = selectedItem?.latestScore ?? null;
   const selectedPublisher = selectedItem?.publisher ?? null;
   const canBanSelectedUser = canBanPublisherAbuseOwner(selectedItem, currentUserId);
@@ -267,15 +270,15 @@ export function AbusePage({
               <dt>Last scan</dt>
               <dd
                 className={
-                  latestRun?.status === "completed"
+                  displayedRun?.status === "completed"
                     ? "pa-run-ok"
-                    : latestRun?.status === "failed"
+                    : displayedRun?.status === "failed"
                       ? "pa-run-bad"
                       : undefined
                 }
               >
-                {latestRun
-                  ? formatPublisherAbuseRunStatus(latestRun.status)
+                {displayedRun
+                  ? formatPublisherAbuseRunStatus(displayedRun.status)
                   : dashboardLoaded
                     ? "No scans yet"
                     : "Loading"}
@@ -283,11 +286,11 @@ export function AbusePage({
             </div>
             <div>
               <dt>Scanned</dt>
-              <dd>{formatWholeNumber(latestRun?.scannedPublishers)}</dd>
+              <dd>{formatWholeNumber(displayedRun?.scannedPublishers)}</dd>
             </div>
             <div>
               <dt>Scored</dt>
-              <dd>{formatWholeNumber(latestRun?.scoredPublishers)}</dd>
+              <dd>{formatWholeNumber(displayedRun?.scoredPublishers)}</dd>
             </div>
           </dl>
           <div className="pa-rescan">
@@ -318,6 +321,32 @@ export function AbusePage({
         </div>
       </header>
 
+      {tab === "signals" && latestSignalRun?.status === "failed" ? (
+        <div className="pa-scan-failure" role="alert">
+          <XCircle aria-hidden="true" size={18} />
+          <div>
+            <strong>
+              {signalFailureCount > 0
+                ? `Stopped after ${signalFailureCount} failed attempts`
+                : "Signal scan failed"}
+            </strong>
+            <span>
+              {latestSignalRun.errorMessage ?? "The signal scan failed without an error."}
+            </span>
+          </div>
+        </div>
+      ) : tab === "signals" && latestSignalRun?.status === "running" && signalFailureCount > 0 ? (
+        <div className="pa-scan-retrying" role="status">
+          <RefreshCcw aria-hidden="true" size={18} />
+          <div>
+            <strong>Retrying after {signalFailureCount} of 5 failed attempts</strong>
+            <span>
+              {latestSignalRun?.lastTransientError ?? "The previous signal scan attempt failed."}
+            </span>
+          </div>
+        </div>
+      ) : null}
+
       <div className="pa-tabs" role="tablist" aria-label="Publisher abuse queue">
         <PublisherAbuseTabButton
           active={tab === "potential_ban_candidate"}

src/styles.css

@@ -14351,6 +14351,37 @@ code {
   color: var(--status-error-fg);
 }
 
+.pa-scan-failure,
+.pa-scan-retrying {
+  display: flex;
+  align-items: flex-start;
+  gap: var(--space-2);
+  padding: var(--space-3) var(--space-4);
+  border: 1px solid color-mix(in srgb, var(--status-error-fg) 30%, var(--line));
+  background: var(--status-error-bg);
+  color: var(--status-error-fg);
+}
+
+.pa-scan-failure > div,
+.pa-scan-retrying > div {
+  display: grid;
+  gap: 2px;
+  min-width: 0;
+}
+
+.pa-scan-failure span,
+.pa-scan-retrying span {
+  overflow-wrap: anywhere;
+  color: var(--ink-soft);
+  font-size: var(--fs-sm);
+}
+
+.pa-scan-retrying {
+  border-color: color-mix(in srgb, var(--oc-status-warning-fg) 30%, var(--line));
+  background: var(--oc-status-warning-bg);
+  color: var(--oc-status-warning-fg);
+}
+
 .pa-tabs {
   display: flex;
   flex-wrap: wrap;