fix(sync,web): surface exhausted jobs and skip draft compass hop

packages/core/src/types/sync/diagnostic.contracts.test.ts

@@ -18,6 +18,8 @@ const sample = (): DiagnosticConnectionResponse => ({
   disconnectedAt: null,
   calendarCount: 2,
   pendingJobCount: 1,
+  failedJobCount: 0,
+  exhaustedJobCount: 0,
   pendingCommandCount: 0,
 });
 

packages/core/src/types/sync/diagnostic.contracts.ts

@@ -31,7 +31,13 @@ export const DiagnosticConnectionResponseSchema = z.strictObject({
   lastHealthyAt: DateTimeSchema.nullable(),
   disconnectedAt: DateTimeSchema.nullable(),
   calendarCount: z.number().int().nonnegative(),
+  // pending + claimed only — active work still eligible to run.
   pendingJobCount: z.number().int().nonnegative(),
+  // All failed jobs for the connection (including permanent and exhausted).
+  failedJobCount: z.number().int().nonnegative(),
+  // Failed jobs that exhausted the self-heal requeue budget and need an
+  // operator (`bun run cli manage-failed-jobs …`).
+  exhaustedJobCount: z.number().int().nonnegative(),
   pendingCommandCount: z.number().int().nonnegative(),
 });
 export type DiagnosticConnectionResponse = z.infer<

packages/scripts/src/cli.ts

@@ -1,4 +1,5 @@
 import { CliValidator } from "@scripts/cli.validator";
+import { runManageFailedJobs } from "@scripts/commands/manage-failed-jobs";
 import { runPurgeCorruptSyncEvents } from "@scripts/commands/purge-corrupt-sync-events";
 import { runPurgeUser } from "@scripts/commands/purge-user";
 import { runRefreshConnectionStates } from "@scripts/commands/refresh-connection-states";
@@ -25,6 +26,9 @@ export default class CompassCLI {
       case cmd === "refresh-connection-states":
         await runRefreshConnectionStates();
         break;
+      case cmd === "manage-failed-jobs":
+        await runManageFailedJobs();
+        break;
       case cmd === "purge-user":
         await runPurgeUser();
         break;
@@ -73,6 +77,14 @@ export default class CompassCLI {
         "Re-derive every provider connection's stored state from live evidence (--apply to write)",
       );
 
+    program
+      .command("manage-failed-jobs")
+      .helpOption(false)
+      .allowUnknownOption(true)
+      .description(
+        "List/clear/requeue Sync jobs that exhausted the self-heal budget (list | clear | requeue)",
+      );
+
     return program;
   }
 }

packages/scripts/src/commands/manage-failed-jobs.ts

@@ -0,0 +1,153 @@
+import { loadCompassConfig } from "@core/config/compass.config";
+import { Logger } from "@core/logger/winston.logger";
+import { SyncJobIdSchema } from "@core/types/sync/identity.contracts";
+import { FAILED_JOB_MAX_REQUEUES } from "@sync/domain/failed-job-requeue.service";
+import { JobRepository } from "@sync/storage/repositories/job.repository";
+import { SyncMongoService } from "@sync/storage/sync-mongo.service";
+
+const logger = Logger("scripts.commands.manage-failed-jobs");
+
+function syncMongoUri(): string {
+  const fromEnv = process.env["SYNC_MONGO_URI"]?.trim();
+  if (fromEnv) return fromEnv;
+  const uri = loadCompassConfig().sync?.mongoUri?.trim();
+  if (!uri) {
+    throw new Error(
+      "Set SYNC_MONGO_URI or add sync.mongoUri to compass.yaml before manage-failed-jobs",
+    );
+  }
+  return uri;
+}
+
+function flagValue(args: string[], name: string): string | undefined {
+  const index = args.indexOf(name);
+  if (index < 0) return undefined;
+  return args[index + 1];
+}
+
+/**
+ * Operator tooling for Sync jobs that exhausted the self-heal requeue budget
+ * and still occupy a coalescing key. Default actions are dry-run; `--apply`
+ * persists.
+ *
+ *   bun run cli manage-failed-jobs list
+ *   bun run cli manage-failed-jobs clear --id <id> --coalescing-key <key> [--apply]
+ *   bun run cli manage-failed-jobs requeue --id <id> [--apply]
+ */
+export async function runManageFailedJobs(): Promise<void> {
+  const args = process.argv.slice(3);
+  const action = args[0];
+  if (action !== "list" && action !== "clear" && action !== "requeue") {
+    throw new Error(
+      "Usage: manage-failed-jobs <list|clear|requeue> [--id …] [--coalescing-key …] [--apply]",
+    );
+  }
+
+  const apply = args.includes("--apply");
+  const syncMongo = new SyncMongoService();
+  try {
+    await syncMongo.connect({
+      uri: syncMongoUri(),
+      enforceLeastPrivilege: false,
+      forbiddenDatabaseName: "prod_calendar",
+    });
+    const jobs = new JobRepository(syncMongo.db);
+
+    if (action === "list") {
+      const [count, sample] = await Promise.all([
+        jobs.countExhaustedFailed(FAILED_JOB_MAX_REQUEUES),
+        jobs.listExhaustedFailed(FAILED_JOB_MAX_REQUEUES),
+      ]);
+      const report = {
+        maxRequeues: FAILED_JOB_MAX_REQUEUES,
+        count,
+        sampleSize: sample.length,
+        truncated: count > sample.length,
+        jobs: sample.map((job) => ({
+          id: job.id,
+          coalescingKey: job.coalescingKey,
+          connectionId: job.connectionId,
+          failureClass: job.failureClass,
+          requeuedCount: job.requeuedCount,
+          updatedAt: job.updatedAt.toISOString(),
+        })),
+      };
+      process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
+      logger.info(
+        `manage-failed-jobs list count=${report.count} sample=${report.sampleSize} truncated=${report.truncated}`,
+      );
+      await syncMongo.disconnect();
+      process.exit(0);
+    }
+
+    const idRaw = flagValue(args, "--id");
+    if (!idRaw) {
+      throw new Error(`manage-failed-jobs ${action} requires --id <SyncJobId>`);
+    }
+    const id = SyncJobIdSchema.parse(idRaw);
+    const existing = await jobs.findByIdUnscoped(id);
+    if (!existing) {
+      throw new Error(`No job found for id=${id}`);
+    }
+    if (existing.state !== "failed") {
+      throw new Error(
+        `Job id=${id} is state=${existing.state}; ${action} only accepts failed jobs`,
+      );
+    }
+
+    if (action === "clear") {
+      const coalescingKey =
+        flagValue(args, "--coalescing-key") ?? existing.coalescingKey;
+      if (coalescingKey !== existing.coalescingKey) {
+        throw new Error(
+          `coalescing-key mismatch for id=${id}: expected ${existing.coalescingKey}`,
+        );
+      }
+      const report = {
+        dryRun: !apply,
+        action: "clear" as const,
+        id,
+        coalescingKey,
+        state: existing.state,
+      };
+      if (apply) {
+        const ok = await jobs.remove(id, coalescingKey);
+        if (!ok) {
+          throw new Error(`Failed to clear id=${id} key=${coalescingKey}`);
+        }
+      }
+      process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
+      logger.info(
+        `manage-failed-jobs clear dryRun=${report.dryRun} id=${id} key=${coalescingKey}`,
+      );
+      await syncMongo.disconnect();
+      process.exit(0);
+    }
+
+    const report = {
+      dryRun: !apply,
+      action: "requeue" as const,
+      id,
+      coalescingKey: existing.coalescingKey,
+      requeuedCount: existing.requeuedCount,
+    };
+    if (apply) {
+      const ok = await jobs.requeue(id, new Date());
+      if (!ok) {
+        throw new Error(`Failed to requeue id=${id}`);
+      }
+    }
+    process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
+    logger.info(`manage-failed-jobs requeue dryRun=${report.dryRun} id=${id}`);
+    await syncMongo.disconnect();
+    process.exit(0);
+  } catch (error) {
+    logger.error(error);
+    try {
+      await syncMongo.disconnect();
+    } catch {
+      // ignore
+    }
+    process.exit(1);
+  }
+}

packages/sync/src/app.ts

@@ -9,7 +9,10 @@ import {
   CONNECTION_CACHE_RETENTION_MS,
   purgeExpiredDisconnectedConnections,
 } from "@sync/domain/connection-retention.service";
-import { requeueFailedJobs } from "@sync/domain/failed-job-requeue.service";
+import {
+  FAILED_JOB_MAX_REQUEUES,
+  requeueFailedJobs,
+} from "@sync/domain/failed-job-requeue.service";
 import { reconcileStaleCalendars } from "@sync/domain/reconcile.service";
 import { retryStaleCommands } from "@sync/domain/stale-command-retry.service";
 import { maintainExpiringSubscriptions } from "@sync/domain/subscription-sweep.service";
@@ -262,9 +265,6 @@ async function start(): Promise<void> {
 // at least this long — long enough that a real provider outage has had a
 // chance to clear before we burn another retry ladder on it.
 const FAILED_JOB_REQUEUE_COOLDOWN_MS = 30 * 60_000;
-// How many times the self-heal sweep will requeue the same job before
-// leaving it failed for an operator instead.
-const FAILED_JOB_MAX_REQUEUES = 3;
 // A resource not synced within this window is swept for a reconcile pull.
 const RECONCILE_STALE_AFTER_MS = 15 * 60_000;
 // A cloud command left nonterminal past this long since its last update is
@@ -467,6 +467,16 @@ function buildSchedulers(
         if (result.exhausted > 0) {
           logger.error(
             `Sync self-heal sweep: ${result.exhausted} failed job(s) exhausted their requeue budget and need operator attention`,
+            {
+              exhaustedJobs: result.exhaustedJobs.map((job) => ({
+                id: job.id,
+                coalescingKey: job.coalescingKey,
+                connectionId: job.connectionId,
+                failureClass: job.failureClass,
+                requeuedCount: job.requeuedCount,
+                updatedAt: job.updatedAt.toISOString(),
+              })),
+            },
           );
         }
         return result.requeued;

packages/sync/src/domain/connection-diagnostic.service.ts

@@ -2,6 +2,7 @@ import {
   type DiagnosticConnectionResponse,
   DiagnosticConnectionResponseSchema,
 } from "@core/types/sync/diagnostic.contracts";
+import { FAILED_JOB_MAX_REQUEUES } from "@sync/domain/failed-job-requeue.service";
 import { type CommandRepository } from "@sync/storage/repositories/command.repository";
 import { type JobRepository } from "@sync/storage/repositories/job.repository";
 import { type ProviderCalendarRepository } from "@sync/storage/repositories/provider-calendar.repository";
@@ -19,6 +20,7 @@ export interface ConnectionDiagnosticDeps {
 export async function resolveDiagnosticConnection(
   deps: ConnectionDiagnosticDeps,
   diagnosticKey: string,
+  maxRequeues: number = FAILED_JOB_MAX_REQUEUES,
 ): Promise<DiagnosticConnectionResponse | null> {
   const connection = await deps.connections.findByDiagnosticKey(diagnosticKey);
   if (!connection) return null;
@@ -28,12 +30,28 @@ export async function resolveDiagnosticConnection(
     connection.principalId,
     connection._id,
   );
-  const [pendingJobCount, pendingCommandCount] = await Promise.all([
+  const [
+    pendingJobCount,
+    failedJobCount,
+    exhaustedJobCount,
+    pendingCommandCount,
+  ] = await Promise.all([
     deps.jobs.countOutstandingByConnection(
       connection.tenantId,
       connection.principalId,
       connection._id,
     ),
+    deps.jobs.countFailedByConnection(
+      connection.tenantId,
+      connection.principalId,
+      connection._id,
+    ),
+    deps.jobs.countExhaustedFailedByConnection(
+      connection.tenantId,
+      connection.principalId,
+      connection._id,
+      maxRequeues,
+    ),
     deps.commands.countNonterminalByConnection(
       connection.tenantId,
       connection.principalId,
@@ -56,6 +74,8 @@ export async function resolveDiagnosticConnection(
     disconnectedAt: connection.disconnectedAt?.toISOString() ?? null,
     calendarCount: calendars.length,
     pendingJobCount,
+    failedJobCount,
+    exhaustedJobCount,
     pendingCommandCount,
   });
 }

packages/sync/src/domain/failed-job-requeue.service.db.test.ts

@@ -49,7 +49,7 @@ describe("requeueFailedJobs", () => {
 
     const result = await requeueFailedJobs(deps(), cooldownBefore, now, 3);
 
-    expect(result).toEqual({ requeued: 1, exhausted: 0 });
+    expect(result).toEqual({ requeued: 1, exhausted: 0, exhaustedJobs: [] });
     const raw = await storage
       .db()
       .collection("jobs")
@@ -64,7 +64,7 @@ describe("requeueFailedJobs", () => {
 
     const result = await requeueFailedJobs(deps(), cooldownBefore, now, 3);
 
-    expect(result).toEqual({ requeued: 0, exhausted: 0 });
+    expect(result).toEqual({ requeued: 0, exhausted: 0, exhaustedJobs: [] });
   });
 
   it("stops requeuing once a job hits the cap and reports it as exhausted", async () => {
@@ -84,13 +84,22 @@ describe("requeueFailedJobs", () => {
 
     const result = await requeueFailedJobs(deps(), cooldownBefore, now, 2);
 
-    expect(result).toEqual({ requeued: 0, exhausted: 1 });
+    expect(result.requeued).toBe(0);
+    expect(result.exhausted).toBe(1);
+    expect(result.exhaustedJobs).toEqual([
+      expect.objectContaining({
+        id,
+        failureClass: "retryableTransient",
+        requeuedCount: 2,
+      }),
+    ]);
   });
 
   it("does nothing when there are no failed jobs", async () => {
     expect(await requeueFailedJobs(deps(), cooldownBefore, now, 3)).toEqual({
       requeued: 0,
       exhausted: 0,
+      exhaustedJobs: [],
     });
   });
 });

packages/sync/src/domain/failed-job-requeue.service.ts

@@ -1,4 +1,12 @@
-import { type JobRepository } from "@sync/storage/repositories/job.repository";
+import {
+  type ExhaustedFailedJob,
+  type JobRepository,
+} from "@sync/storage/repositories/job.repository";
+
+// How many times the self-heal sweep will requeue the same job before leaving
+// it failed for an operator instead. Shared with diagnostics so exhausted
+// counts use the same budget the sweep enforces.
+export const FAILED_JOB_MAX_REQUEUES = 3;
 
 export interface FailedJobRequeueDeps {
   jobs: JobRepository;
@@ -10,6 +18,8 @@ export interface FailedJobRequeueResult {
   // How many failed jobs have hit the requeue cap and re-failed anyway — the
   // sweep will not touch them again; they need an operator.
   exhausted: number;
+  // Bounded sample of exhausted rows for operator-facing logs / CLI.
+  exhaustedJobs: ExhaustedFailedJob[];
 }
 
 // The self-heal sweep for jobs terminalized as state:"failed". A worker marks
@@ -34,7 +44,7 @@ export async function requeueFailedJobs(
   deps: FailedJobRequeueDeps,
   before: Date,
   now: () => Date,
-  maxRequeues: number,
+  maxRequeues: number = FAILED_JOB_MAX_REQUEUES,
   limit = 100,
 ): Promise<FailedJobRequeueResult> {
   const candidates = await deps.jobs.listFailedForRequeue(
@@ -45,6 +55,13 @@ export async function requeueFailedJobs(
   for (const job of candidates) {
     await deps.jobs.requeue(job._id, now());
   }
-  const exhausted = await deps.jobs.countExhaustedFailed(maxRequeues);
-  return { requeued: candidates.length, exhausted };
+  const [exhausted, exhaustedJobs] = await Promise.all([
+    deps.jobs.countExhaustedFailed(maxRequeues),
+    deps.jobs.listExhaustedFailed(maxRequeues),
+  ]);
+  return {
+    requeued: candidates.length,
+    exhausted,
+    exhaustedJobs,
+  };
 }