fix(scheduler): prioritize canonical coverage gaps

.github/workflows/sweep.yml

@@ -2525,6 +2525,7 @@ jobs:
             --workflow sweep.yml \
             --ref main \
             --cursor-store-url "$REVIEW_COVERAGE_URL" \
+            --coverage-tracked-items-manifest .artifacts/worker-records-manifest.json \
             --publish-url "$REVIEW_COVERAGE_URL"
 
       - name: Summarize trailing weekly review coverage
@@ -2972,6 +2973,7 @@ jobs:
             --codex-sandbox read-only \
             --min-active-shards "$MIN_ACTIVE_SHARDS" \
             --min-backfill-review-age-minutes "$MIN_BACKFILL_REVIEW_AGE_MINUTES" \
+            --coverage-tracked-items-manifest .artifacts/worker-records-manifest.json \
             "${hot_intake_arg[@]}" \
             "${item_arg[@]}" > plan.json
           pnpm run --silent workflow -- plan-output \

CHANGELOG.md

@@ -9,6 +9,7 @@ checkpoint, and status-only commits are intentionally omitted.
 
 ### Changed
 
+- Aligned normal fanout and planner priority with the dashboard's canonical tuple coverage identities, so legacy backfill reports no longer hide untracked open items behind canonical re-reviews.
 - Sized scheduled candidate batches from live free review capacity, apportioned fleet fanout by untracked backlog while retaining round-robin fairness, and skipped empty repositories before both normal and hot fanout so a dominant backlog can fill idle review slots without starving smaller targets.
 - Isolated review admission, pressure, and backpressure accounting from the publication lane so stale publication work cannot throttle review throughput, made top-level queue health review-specific, and added durable shed counters by reason.
 - Made exact-review reservation races and mid-generation supersession successful no-ops with bounded jittered retries, surfaced provider throttling separately from content/output failures, prioritized never-reviewed open items before oldest-reviewed canonical refreshes, and based dashboard coverage on signed live open-item inventory with explicit expired, untracked, protected, and unmanaged cohorts.

docs/scheduler.md

@@ -135,8 +135,8 @@ Target fanout dispatches review batches through `repository_dispatch` so each
 selected repository can carry its inventory default branch without consuming
 manual workflow inputs. Scheduled fanout uses:
 
-- hot intake: `4/15 * * * *`, 20 target repositories per cursor step
-- normal review: `41 * * * *`, 12 target repositories per cursor step
+- hot intake: `4/5 * * * *`, 20 target repositories per cursor step
+- normal review: `41/10 * * * *`, 12 target repositories per cursor step
 - audit: `37 */6 * * *`, 12 target repositories per cursor step
 
 Each mode's cursor lives in the authenticated ExactReviewQueue Durable Object,
@@ -153,6 +153,12 @@ the remaining candidate volume by backlog share. The rotating slice is
 dispatched first, so one large repository can fill otherwise-idle capacity
 without permanently consuming smaller repositories' scheduled-feed budget.
 
+Worker hydration also records the exact item identities present in the modern
+canonical tuple store. Normal fanout and each target planner use those identities
+for the same `untracked_open` boundary as the coverage endpoint. A hydrated
+legacy backfill report remains review context, but it does not count as coverage
+or yield a planner slot to a canonical re-review until a modern tuple exists.
+
 The six-hour audit fanout also writes a GitHub Actions summary with canonical
 open-item reports reviewed in the trailing seven days versus batched live open
 issue and PR totals across the complete dynamic inventory.

scripts/worker-records.ts

@@ -16,6 +16,8 @@ import {
 import path from "node:path";
 import { spawnSync } from "node:child_process";
 
+import { WORKER_RECORDS_MANIFEST_SCHEMA_VERSION } from "../src/review-coverage-manifest.ts";
+
 export const RECORD_SECTIONS = ["items", "closed", "plans", "decision-packets", "commits"] as const;
 export type RecordSection = (typeof RECORD_SECTIONS)[number];
 
@@ -229,6 +231,7 @@ export async function materializeWorkerRecords(options: {
       snapshotCache: "hit" | "miss" | "cold";
       deltaRecords: number;
       recordCount: number;
+      coverageTrackedItemIds: number[];
     }
   > = {};
   try {
@@ -294,19 +297,26 @@ export async function materializeWorkerRecords(options: {
           throw error;
         });
         applyWorkerRecords(stagedRepoRoot, journal.records);
+        const coverageTrackedItemIds = await fetchWorkerCanonicalItemIds({
+          baseUrl: options.baseUrl,
+          webhookSecret: options.webhookSecret,
+          repoSlug,
+          fetch: options.fetch,
+        });
         repositories[repoSlug] = {
           revision: journal.revision,
           snapshotRevision: storedSnapshot?.revisionWatermark ?? 0,
           snapshotBytes,
           snapshotCache,
           deltaRecords: journal.records.length,
           recordCount: countMaterializedRecords(stagingRoot, repoSlug),
+          coverageTrackedItemIds,
         };
         const entry = repositories[repoSlug];
         log(
           storedSnapshot
-            ? `[worker-records] snapshot hydrated repo=${repoSlug} revision=${entry.revision} snapshotRevision=${entry.snapshotRevision} snapshotBytes=${entry.snapshotBytes} cache=${entry.snapshotCache} deltaRecords=${entry.deltaRecords} records=${entry.recordCount}`
-            : `[worker-records] COLD HYDRATION repo=${repoSlug}: no stored snapshot, replayed the full journal from revision 0 (revision=${entry.revision} journalRecords=${entry.deltaRecords} records=${entry.recordCount} bound=${COLD_HYDRATION_MAX_RECORDS}); trigger a snapshot sweep to make future hydrations incremental`,
+            ? `[worker-records] snapshot hydrated repo=${repoSlug} revision=${entry.revision} snapshotRevision=${entry.snapshotRevision} snapshotBytes=${entry.snapshotBytes} cache=${entry.snapshotCache} deltaRecords=${entry.deltaRecords} records=${entry.recordCount} coverageTrackedItems=${entry.coverageTrackedItemIds.length}`
+            : `[worker-records] COLD HYDRATION repo=${repoSlug}: no stored snapshot, replayed the full journal from revision 0 (revision=${entry.revision} journalRecords=${entry.deltaRecords} records=${entry.recordCount} coverageTrackedItems=${entry.coverageTrackedItemIds.length} bound=${COLD_HYDRATION_MAX_RECORDS}); trigger a snapshot sweep to make future hydrations incremental`,
         );
       } catch (error) {
         // Re-wrap so the refusal that aborts a multi-slug hydration names the
@@ -334,7 +344,7 @@ export async function materializeWorkerRecords(options: {
   mkdirSync(path.dirname(manifestPath), { recursive: true });
   writeFileSync(
     manifestPath,
-    `${JSON.stringify({ schemaVersion: 2, source: "worker", repositories }, null, 2)}\n`,
+    `${JSON.stringify({ schemaVersion: WORKER_RECORDS_MANIFEST_SCHEMA_VERSION, source: "worker", repositories }, null, 2)}\n`,
     "utf8",
   );
   return { recordsRoot, manifestPath, repositories };
@@ -465,6 +475,60 @@ export async function discoverWorkerRecordRepoSlugs(options: {
   return repositories.sort((left, right) => left.repoSlug.localeCompare(right.repoSlug));
 }
 
+export async function fetchWorkerCanonicalItemIds(options: {
+  baseUrl: string;
+  webhookSecret: string;
+  repoSlug: string;
+  fetch?: typeof globalThis.fetch;
+}): Promise<number[]> {
+  const itemIds: number[] = [];
+  const seen = new Set<number>();
+  let nextCursor: number | null = 0;
+  while (nextCursor !== null) {
+    const cursor = nextCursor;
+    const page = await signedPost<{
+      repoSlug: string;
+      section: string;
+      records: Array<{ id: number }>;
+      nextCursor: number | null;
+    }>({
+      baseUrl: options.baseUrl,
+      path: "/internal/state/records/list",
+      webhookSecret: options.webhookSecret,
+      body: { repoSlug: options.repoSlug, section: "items", cursor, limit: 500 },
+      fetch: options.fetch,
+    });
+    if (
+      page.repoSlug !== options.repoSlug ||
+      page.section !== "items" ||
+      !Array.isArray(page.records)
+    ) {
+      throw new Error("Worker returned an invalid canonical item listing");
+    }
+    for (const record of page.records) {
+      const itemId = Number(record?.id);
+      if (!Number.isSafeInteger(itemId) || itemId <= cursor || seen.has(itemId)) {
+        throw new Error("Worker returned an invalid canonical item identity");
+      }
+      seen.add(itemId);
+      itemIds.push(itemId);
+    }
+    if (page.nextCursor === null) {
+      nextCursor = null;
+      continue;
+    }
+    if (
+      !Number.isSafeInteger(page.nextCursor) ||
+      page.nextCursor <= cursor ||
+      page.nextCursor !== itemIds.at(-1)
+    ) {
+      throw new Error("Worker returned an invalid canonical item cursor");
+    }
+    nextCursor = page.nextCursor;
+  }
+  return itemIds;
+}
+
 function countMaterializedRecords(root: string, repoSlug: string): number {
   const repoRoot = path.join(root, "records", repoSlug);
   if (!existsSync(repoRoot)) return 0;

src/clawsweeper.ts

@@ -54,6 +54,7 @@ import {
 } from "./github-retry.js";
 import { parseGhJson, parseGhJsonLinesWithRetry, parseGhJsonWithRetry } from "./github-json.js";
 import { stableJson } from "./stable-json.js";
+import { coverageTrackedItemIdsFromManifest } from "./review-coverage-manifest.js";
 import {
   LEGACY_FIXED_CLOSE_SKIP_ACTIONS,
   LIVE_RECHECK_CLOSE_GUARD_ACTIONS,
@@ -1145,6 +1146,7 @@ interface PlanCandidateResult {
 interface PlanSelectionTelemetry {
   itemNumber: number;
   bucket: SchedulerDueCandidate["bucket"];
+  coverageTracked: boolean;
   lastReviewedAt: string | null;
   ageMs: number;
   nextDueAt: string;
@@ -8485,15 +8487,20 @@ function dueCandidate(
   now = Date.now(),
   reviewPolicy?: string,
   reviewIndex?: ExistingReviewIndex,
+  coverageTrackedItemIds?: ReadonlySet<number>,
 ): DueCandidate | null {
   const review = indexedExistingReview(item, itemsDir, reviewIndex);
-  if (!shouldReviewItem(item, review, now, reviewPolicy)) return null;
+  const coverageTracked = coverageTrackedItemIds
+    ? coverageTrackedItemIds.has(item.number)
+    : review !== null;
+  if (coverageTracked && !shouldReviewItem(item, review, now, reviewPolicy)) return null;
   return {
     item,
     review,
+    coverageTracked,
     priority: reviewPriority(item, review, now, reviewPolicy),
     reviewedAt: reviewedAtMs(review) ?? 0,
-    nextDueAt: nextReviewDueAtMs(item, review, now, reviewPolicy),
+    nextDueAt: coverageTracked ? nextReviewDueAtMs(item, review, now, reviewPolicy) : 0,
     bucket: schedulerBucket(item, review, now),
   };
 }
@@ -9043,7 +9050,11 @@ function oldestUnreviewedAt(candidates: readonly DueCandidate[]): string | undef
   let oldest: string | undefined;
   let oldestMs = Number.POSITIVE_INFINITY;
   for (const candidate of candidates) {
-    if (candidate.review) continue;
+    const coverageTracked =
+      candidate.coverageTracked === undefined
+        ? candidate.review !== null
+        : candidate.coverageTracked;
+    if (coverageTracked) continue;
     const createdAtMs = Date.parse(candidate.item.createdAt);
     if (!Number.isFinite(createdAtMs) || createdAtMs >= oldestMs) continue;
     oldestMs = createdAtMs;
@@ -9087,6 +9098,10 @@ function planSelectionTelemetry(
     return {
       itemNumber: candidate.item.number,
       bucket: candidate.bucket,
+      coverageTracked:
+        candidate.coverageTracked === undefined
+          ? candidate.review !== null
+          : candidate.coverageTracked,
       lastReviewedAt: candidate.review?.reviewedAt ?? null,
       ageMs: Number.isFinite(referenceAt) ? Math.max(0, now - referenceAt) : 0,
       nextDueAt: new Date(candidate.nextDueAt).toISOString(),
@@ -9105,6 +9120,7 @@ function planCandidates(options: {
   hotIntake?: boolean;
   minimumActiveShards?: number;
   minimumBackfillReviewAgeMs?: number;
+  coverageTrackedItemIds?: ReadonlySet<number>;
 }): PlanCandidateResult {
   const shardCount = planShardCount(options.shardCount);
   const batchSize = Math.max(1, options.batchSize);
@@ -9175,6 +9191,7 @@ function planCandidates(options: {
         now,
         options.reviewPolicy,
         reviewIndex,
+        options.coverageTrackedItemIds,
       );
       if (candidate) due.push(candidate);
     }
@@ -9218,6 +9235,7 @@ function planCandidates(options: {
         now,
         options.reviewPolicy,
         reviewIndex,
+        options.coverageTrackedItemIds,
       );
       if (candidate) {
         due.push(candidate);
@@ -22697,6 +22715,10 @@ function planCommand(args: Args): void {
   const sandboxMode = stringArg(args.codex_sandbox, "read-only");
   const serviceTier = stringArg(args.codex_service_tier, DEFAULT_SERVICE_TIER);
   const reviewPolicy = reviewPolicyHash({ model, reasoningEffort, sandboxMode, serviceTier });
+  const coverageManifest = stringArg(args.coverage_tracked_items_manifest, "").trim();
+  const coverageTrackedItemIds = coverageManifest
+    ? coverageTrackedItemIdsFromManifest(resolve(coverageManifest), targetProfile().slug)
+    : undefined;
   const planOptions: Parameters<typeof planCandidates>[0] = {
     batchSize,
     maxPages,
@@ -22705,6 +22727,7 @@ function planCommand(args: Args): void {
     reviewPolicy,
     minimumActiveShards,
     minimumBackfillReviewAgeMs,
+    ...(coverageTrackedItemIds ? { coverageTrackedItemIds } : {}),
   };
   if (hasItemNumbersInput || itemNumbers.length > 0) planOptions.itemNumbers = itemNumbers;
   if (hotIntake) planOptions.hotIntake = true;

src/repair/target-fanout.ts

@@ -6,6 +6,7 @@ import { join } from "node:path";
 import { pathToFileURL } from "node:url";
 import { resolveCommand } from "../command.js";
 import { fetchExactReviewQueuePressure } from "../queue-pressure.js";
+import { coverageTrackedCountsFromManifest } from "../review-coverage-manifest.js";
 import { parseArgs, repoRoot } from "./lib.js";
 
 type JsonRecord = Record<string, unknown>;
@@ -187,7 +188,15 @@ export async function runTargetFanout(argv: string[]): Promise<void> {
           );
         }
       }
-      planningRepositories = reviewPlanningRepositories({ repositories, openCounts });
+      const coverageManifestPath = stringArg(args["coverage-tracked-items-manifest"], "");
+      const coverageTrackedCounts = coverageManifestPath
+        ? coverageTrackedCountsFromManifest(coverageManifestPath)
+        : undefined;
+      planningRepositories = reviewPlanningRepositories({
+        repositories,
+        openCounts,
+        ...(coverageTrackedCounts ? { coverageTrackedCounts } : {}),
+      });
     } else {
       planningRepositories = repositoriesWithOpenItems(repositories, openCounts);
     }
@@ -368,6 +377,7 @@ export function reviewPlanningRepositories(options: {
   repositories: readonly SelectedRepository[];
   openCounts: ReadonlyMap<string, RepositoryOpenCounts>;
   recordsRoot?: string;
+  coverageTrackedCounts?: ReadonlyMap<string, number>;
 }): ReviewPlanningRepository[] {
   const recordsRoot = options.recordsRoot ?? join(repoRoot(), "records");
   return options.repositories
@@ -382,11 +392,14 @@ export function reviewPlanningRepositories(options: {
         repository.targetRepo.toLowerCase().replace("/", "-"),
         "items",
       );
-      const trackedRecords = existsSync(itemsDir)
-        ? readdirSync(itemsDir, { withFileTypes: true }).filter(
-            (entry) => entry.isFile() && entry.name.endsWith(".md"),
-          ).length
-        : 0;
+      const repoSlug = repository.targetRepo.toLowerCase().replace("/", "-");
+      const trackedRecords = options.coverageTrackedCounts
+        ? (options.coverageTrackedCounts.get(repoSlug) ?? 0)
+        : existsSync(itemsDir)
+          ? readdirSync(itemsDir, { withFileTypes: true }).filter(
+              (entry) => entry.isFile() && entry.name.endsWith(".md"),
+            ).length
+          : 0;
       return {
         ...repository,
         openItems,

src/review-coverage-manifest.ts

@@ -0,0 +1,56 @@
+import { readFileSync } from "node:fs";
+
+export const WORKER_RECORDS_MANIFEST_SCHEMA_VERSION = 3;
+
+export function coverageTrackedItemIdsFromManifest(
+  manifestPath: string,
+  repoSlug: string,
+): Set<number> {
+  const repository = readManifestRepositories(manifestPath)[repoSlug];
+  if (!isRecord(repository) || !Array.isArray(repository.coverageTrackedItemIds)) {
+    throw new Error(`Worker records manifest has no coverage identities for ${repoSlug}`);
+  }
+  return coverageItemIds(repository.coverageTrackedItemIds, repoSlug);
+}
+
+export function coverageTrackedCountsFromManifest(
+  manifestPath: string,
+): ReadonlyMap<string, number> {
+  const repositories = readManifestRepositories(manifestPath);
+  const counts = new Map<string, number>();
+  for (const [repoSlug, value] of Object.entries(repositories)) {
+    if (!isRecord(value) || !Array.isArray(value.coverageTrackedItemIds)) {
+      throw new Error(`Worker records manifest has no coverage identities for ${repoSlug}`);
+    }
+    counts.set(repoSlug, coverageItemIds(value.coverageTrackedItemIds, repoSlug).size);
+  }
+  return counts;
+}
+
+function readManifestRepositories(manifestPath: string): Record<string, unknown> {
+  const parsed = JSON.parse(readFileSync(manifestPath, "utf8")) as unknown;
+  if (!isRecord(parsed)) throw new Error("Worker records manifest must be an object");
+  if (
+    parsed.schemaVersion !== WORKER_RECORDS_MANIFEST_SCHEMA_VERSION ||
+    parsed.source !== "worker" ||
+    !isRecord(parsed.repositories)
+  ) {
+    throw new Error("Worker records manifest has an unsupported schema");
+  }
+  return parsed.repositories;
+}
+
+function coverageItemIds(values: unknown[], repoSlug: string): Set<number> {
+  const ids = new Set<number>();
+  for (const value of values) {
+    if (!Number.isSafeInteger(value) || Number(value) < 1 || ids.has(Number(value))) {
+      throw new Error(`Worker records manifest has invalid coverage identities for ${repoSlug}`);
+    }
+    ids.add(Number(value));
+  }
+  return ids;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}