From 9cd7b6b46d6064d3bf69bd8f2a2d99561d074345 Mon Sep 17 00:00:00 2001 From: clawsweeper <274271284+clawsweeper[bot]@users.noreply.github.com> Date: Thu, 30 Jul 2026 06:27:55 -0700 Subject: [PATCH] fix(scheduler): prioritize canonical coverage gaps --- .github/workflows/sweep.yml | 2 + CHANGELOG.md | 1 + docs/scheduler.md | 10 ++- scripts/worker-records.ts | 70 +++++++++++++++++++- src/clawsweeper.ts | 29 +++++++- src/repair/target-fanout.ts | 25 +++++-- src/review-coverage-manifest.ts | 56 ++++++++++++++++ src/scheduler-policy.ts | 25 ++++--- test/repair/scheduled-review-enqueue.test.ts | 56 ++++++++++++++++ test/repair/target-fanout.test.ts | 24 +++++++ test/review-coverage-manifest.test.ts | 67 +++++++++++++++++++ test/scheduler-policy.test.ts | 46 +++++++++++++ test/sweep-workflow.test.ts | 12 ++++ test/worker-records-request.test.ts | 41 +++++++++++- 14 files changed, 440 insertions(+), 24 deletions(-) create mode 100644 src/review-coverage-manifest.ts create mode 100644 test/review-coverage-manifest.test.ts diff --git a/.github/workflows/sweep.yml b/.github/workflows/sweep.yml index 65c0616011..80446e1dd4 100644 --- a/.github/workflows/sweep.yml +++ b/.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 \ diff --git a/CHANGELOG.md b/CHANGELOG.md index 47c0e1261a..d1b2c9afc7 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/docs/scheduler.md b/docs/scheduler.md index 16fc5b3b92..4adde484f0 100644 --- a/docs/scheduler.md +++ b/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. diff --git a/scripts/worker-records.ts b/scripts/worker-records.ts index 9841ad750e..030795f049 100644 --- a/scripts/worker-records.ts +++ b/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,6 +297,12 @@ 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, @@ -301,12 +310,13 @@ export async function materializeWorkerRecords(options: { 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 { + const itemIds: number[] = []; + const seen = new Set(); + 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; diff --git a/src/clawsweeper.ts b/src/clawsweeper.ts index 091e8a4077..cdb1122400 100644 --- a/src/clawsweeper.ts +++ b/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, ): 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; }): 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[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; diff --git a/src/repair/target-fanout.ts b/src/repair/target-fanout.ts index 7530b87ed1..52b7714545 100644 --- a/src/repair/target-fanout.ts +++ b/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; @@ -187,7 +188,15 @@ export async function runTargetFanout(argv: string[]): Promise { ); } } - 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; recordsRoot?: string; + coverageTrackedCounts?: ReadonlyMap; }): 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, diff --git a/src/review-coverage-manifest.ts b/src/review-coverage-manifest.ts new file mode 100644 index 0000000000..b2221851ed --- /dev/null +++ b/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 { + 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 { + const repositories = readManifestRepositories(manifestPath); + const counts = new Map(); + 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 { + 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 { + const ids = new Set(); + 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 { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/scheduler-policy.ts b/src/scheduler-policy.ts index 7c794065ad..be85a334b4 100644 --- a/src/scheduler-policy.ts +++ b/src/scheduler-policy.ts @@ -40,6 +40,7 @@ export interface SchedulerDueCandidate< reviewedAt: number; nextDueAt: number; bucket: SchedulerBucket; + coverageTracked?: boolean | undefined; } const HOT_REVIEW_DAYS = 7; @@ -267,6 +268,13 @@ function weeklyCoverageReferenceMs(candidate: SchedulerDueCandidate): number { return Number.isFinite(createdAt) ? createdAt : 0; } +function isCoverageUntracked(candidate: SchedulerDueCandidate): boolean { + return ( + candidate.coverageTracked === false || + (candidate.coverageTracked === undefined && candidate.review === null) + ); +} + const SCHEDULER_BUCKET_WEIGHTS: ReadonlyArray = [ ["hot_issue", 4], ["hot_pull_request", 2], @@ -327,12 +335,11 @@ export function selectDueCandidates< } }; - // A missing canonical record means the live GitHub item has never had a - // review. Fill first-review coverage before renewing tracked records, while - // retaining the weekly-coverage and weighted bucket ordering within that - // cohort. - const neverReviewed = due.filter((candidate) => candidate.review === null); - const neverReviewedCoverageDue = neverReviewed + // A legacy backfill report can be useful review context without satisfying + // the canonical tuple coverage tracked by the Worker. Fill that operational + // coverage gap before renewing already-canonical records. + const coverageUntracked = due.filter(isCoverageUntracked); + const untrackedCoverageDue = coverageUntracked .filter( (candidate) => weeklyCoverageReferenceMs(candidate) + WEEKLY_COVERAGE_REVIEW_DAYS * DAY_MS <= now, @@ -343,9 +350,9 @@ export function selectDueCandidates< weeklyCoverageReferenceMs(left) - weeklyCoverageReferenceMs(right) || compare(left, right), ); - for (const candidate of neverReviewedCoverageDue) take(candidate); + for (const candidate of untrackedCoverageDue) take(candidate); takeWeighted( - neverReviewed.filter( + coverageUntracked.filter( (candidate) => !selectedKeys.has(schedulerItemKey(candidate.item.repo, candidate.item.number)), ), @@ -357,7 +364,7 @@ export function selectDueCandidates< const weeklyCoverageDue = due .filter( (candidate) => - candidate.review !== null && + !isCoverageUntracked(candidate) && weeklyCoverageReferenceMs(candidate) + WEEKLY_COVERAGE_REVIEW_DAYS * DAY_MS <= now, ) .sort( diff --git a/test/repair/scheduled-review-enqueue.test.ts b/test/repair/scheduled-review-enqueue.test.ts index b15f68e4db..90574b8f00 100644 --- a/test/repair/scheduled-review-enqueue.test.ts +++ b/test/repair/scheduled-review-enqueue.test.ts @@ -3,6 +3,62 @@ import { createHmac } from "node:crypto"; import test from "node:test"; import { enqueueScheduledReviewPlan } from "../../dist/repair/scheduled-review-enqueue.js"; +import { selectDueCandidates } from "../../dist/scheduler-policy.js"; + +test("coverage-untracked plans reach queue admission before canonical refreshes", async () => { + const repo = "openclaw/openclaw"; + const candidate = (number: number, coverageTracked: boolean, reviewedAt: string) => ({ + item: { + repo, + number, + kind: "issue" as const, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + }, + review: { reviewStatus: "complete", reviewedAt }, + bucket: "weekly_issue" as const, + priority: 6, + reviewedAt: Date.parse(reviewedAt), + nextDueAt: 0, + coverageTracked, + }); + const due = [ + ...Array.from({ length: 3_000 }, (_, index) => + candidate(index + 1, false, "2026-06-10T00:00:00Z"), + ), + ...Array.from({ length: 20 }, (_, index) => + candidate(3_001 + index, true, "2026-06-01T00:00:00Z"), + ), + ]; + const selected = selectDueCandidates(due, 128, undefined, Date.parse("2026-07-30T12:00:00Z")); + const queuedNumbers: number[] = []; + const summary = await enqueueScheduledReviewPlan({ + plan: { + candidates: selected.map(({ item }) => item), + selection: selected.map(() => ({ ageMs: 0 })), + }, + lane: "normal_backfill", + targetRepo: repo, + targetBranch: "main", + queueUrl: "https://queue.example", + secret: "secret", + deliveryPrefix: "scheduled:coverage:1", + fetchImpl: async (_input, init) => { + if (!init?.method) { + return Response.json({ scheduled_feed: { target_rate_per_hour: 600 } }); + } + const body = JSON.parse(String(init.body)) as { decision: { itemNumber: number } }; + queuedNumbers.push(body.decision.itemNumber); + return Response.json({ ok: true, queued: true }, { status: 202 }); + }, + }); + + assert.equal(summary.queued, 128); + assert.equal( + queuedNumbers.every((number) => number <= 3_000), + true, + ); +}); test("scheduled review enqueue reports the full selection-to-queue funnel and stops on rate limit", async () => { const secret = "scheduled-review-test-secret"; diff --git a/test/repair/target-fanout.test.ts b/test/repair/target-fanout.test.ts index 9aecc46144..4c51d57c97 100644 --- a/test/repair/target-fanout.test.ts +++ b/test/repair/target-fanout.test.ts @@ -115,6 +115,30 @@ test("review fanout gives a single repository enough candidates to saturate free assert.deepEqual(allocateReviewCandidateCapacity([huge], 128), new Map([[huge.targetRepo, 128]])); }); +test("review planning counts canonical coverage instead of legacy report files", () => { + const repositories = [ + { + targetRepo: "openclaw/openclaw", + defaultBranch: "main", + visibility: "PUBLIC", + }, + ]; + const planned = reviewPlanningRepositories({ + repositories, + openCounts: new Map([["openclaw/openclaw", { issues: 3_000, pullRequests: 20 }]]), + coverageTrackedCounts: new Map([["openclaw-openclaw", 20]]), + }); + + assert.deepEqual(planned, [ + { + ...repositories[0], + openItems: 3_020, + trackedRecords: 20, + untrackedOpen: 3_000, + }, + ]); +}); + test("dominant review backlog stays hot while all other 137 repositories rotate without starvation", () => { const dominant = planningRepository("openclaw/openclaw", 3_084); const small = Array.from({ length: 137 }, (_, index) => diff --git a/test/review-coverage-manifest.test.ts b/test/review-coverage-manifest.test.ts new file mode 100644 index 0000000000..9533a4f26a --- /dev/null +++ b/test/review-coverage-manifest.test.ts @@ -0,0 +1,67 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +import { + coverageTrackedCountsFromManifest, + coverageTrackedItemIdsFromManifest, + WORKER_RECORDS_MANIFEST_SCHEMA_VERSION, +} from "../dist/review-coverage-manifest.js"; + +test("coverage manifest exposes exact canonical identities and fleet counts", () => { + const root = mkdtempSync(join(tmpdir(), "clawsweeper-coverage-manifest-")); + const manifest = join(root, "worker-records-manifest.json"); + try { + writeFileSync( + manifest, + `${JSON.stringify({ + schemaVersion: WORKER_RECORDS_MANIFEST_SCHEMA_VERSION, + source: "worker", + repositories: { + "openclaw-openclaw": { coverageTrackedItemIds: [1, 5, 9] }, + "openclaw-clawsweeper": { coverageTrackedItemIds: [2] }, + }, + })}\n`, + ); + + assert.deepEqual( + [...coverageTrackedItemIdsFromManifest(manifest, "openclaw-openclaw")], + [1, 5, 9], + ); + assert.deepEqual( + coverageTrackedCountsFromManifest(manifest), + new Map([ + ["openclaw-openclaw", 3], + ["openclaw-clawsweeper", 1], + ]), + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("coverage manifest rejects duplicate canonical identities", () => { + const root = mkdtempSync(join(tmpdir(), "clawsweeper-coverage-manifest-")); + const manifest = join(root, "worker-records-manifest.json"); + try { + writeFileSync( + manifest, + `${JSON.stringify({ + schemaVersion: WORKER_RECORDS_MANIFEST_SCHEMA_VERSION, + source: "worker", + repositories: { + "openclaw-openclaw": { coverageTrackedItemIds: [1, 1] }, + }, + })}\n`, + ); + + assert.throws( + () => coverageTrackedItemIdsFromManifest(manifest, "openclaw-openclaw"), + /invalid coverage identities/, + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); diff --git a/test/scheduler-policy.test.ts b/test/scheduler-policy.test.ts index 1a497ac2ed..d17528a40c 100644 --- a/test/scheduler-policy.test.ts +++ b/test/scheduler-policy.test.ts @@ -18,6 +18,9 @@ function schedulerCandidate(candidate) { reviewedAt: candidate.reviewedAt ?? 0, nextDueAt: candidate.nextDueAt ?? 0, bucket: candidate.bucket, + ...(candidate.coverageTracked === undefined + ? {} + : { coverageTracked: candidate.coverageTracked }), }; } @@ -569,6 +572,49 @@ test("one repository's untracked backlog fills the queue-sized candidate limit", assert.equal(new Set(selected).size, 128); }); +test("legacy reports missing canonical coverage win every slot before tracked refreshes", () => { + const now = Date.parse("2026-07-30T12:00:00Z"); + const legacyUntracked = Array.from({ length: 3_000 }, (_, index) => ({ + item: item({ + number: index + 1, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + }), + review: { reviewStatus: "complete", reviewedAt: "2026-06-10T00:00:00Z" }, + bucket: "weekly_issue", + priority: 6, + reviewedAt: Date.parse("2026-06-10T00:00:00Z"), + nextDueAt: 0, + coverageTracked: false, + })); + const tracked = Array.from({ length: 20 }, (_, index) => ({ + item: item({ + number: 3_001 + index, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + }), + review: { reviewStatus: "complete", reviewedAt: "2026-06-01T00:00:00Z" }, + bucket: "weekly_issue", + priority: 6, + reviewedAt: Date.parse("2026-06-01T00:00:00Z"), + nextDueAt: 0, + coverageTracked: true, + })); + + const selected = selectDueCandidates( + [...legacyUntracked, ...tracked].map(schedulerCandidate), + 128, + undefined, + now, + ); + + assert.equal(selected.length, 128); + assert.equal( + selected.every((candidate) => candidate.coverageTracked === false), + true, + ); +}); + test("weekly freshness preselection still fills remaining scheduler capacity", () => { const now = Date.parse("2026-06-14T12:00:00Z"); const due = Array.from({ length: 5 }, (_, index) => ({ diff --git a/test/sweep-workflow.test.ts b/test/sweep-workflow.test.ts index be233ba8b2..52e7a6b09f 100644 --- a/test/sweep-workflow.test.ts +++ b/test/sweep-workflow.test.ts @@ -2603,12 +2603,20 @@ test("scheduled reviews feed the durable queue instead of one-item matrix worker workflow.indexOf("- name: Enqueue scheduled review candidates"), workflow.indexOf("\n - name: Prepare review runtime artifact"), ); + const selectBlock = workflow.slice( + workflow.indexOf("- id: select"), + workflow.indexOf("- name: Enqueue scheduled review candidates"), + ); assert.match(modeBlock, /queue_feed=.*clawsweeper_target_sweep/); assert.match(modeBlock, /requested_batch_size=.*client_payload\.batch_size/); assert.match(modeBlock, /requested_batch_size="\$queue_candidate_capacity"/); assert.match(modeBlock, /batch_size="\$requested_batch_size"[\s\S]*shard_count="1"/); assert.match(enqueueBlock, /repair:scheduled-review-enqueue/); + assert.match( + selectBlock, + /--coverage-tracked-items-manifest \.artifacts\/worker-records-manifest\.json/, + ); assert.match(enqueueBlock, /Scheduled review funnel/); assert.match(workflow, /Review scheduled hot item/); assert.match(workflow, /Review scheduled normal item/); @@ -2634,6 +2642,10 @@ test("target fanout uses the canonical cursor store without a git publisher", () assert.match(fanoutBlock, /hydrate-git-state: "false"/); assert.match(fanoutBlock, /--cursor-store-url "\$REVIEW_COVERAGE_URL"/); + assert.match( + fanoutBlock, + /--coverage-tracked-items-manifest \.artifacts\/worker-records-manifest\.json/, + ); assert.doesNotMatch(fanoutBlock, /Create state token/); assert.doesNotMatch(fanoutBlock, /repair:publish-main/); assert.doesNotMatch(fanoutBlock, /results\/target-fanout-cursors/); diff --git a/test/worker-records-request.test.ts b/test/worker-records-request.test.ts index 577880adda..271fefc522 100644 --- a/test/worker-records-request.test.ts +++ b/test/worker-records-request.test.ts @@ -1,7 +1,11 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { exportWorkerRecords, signedPost } from "../scripts/worker-records.ts"; +import { + exportWorkerRecords, + fetchWorkerCanonicalItemIds, + signedPost, +} from "../scripts/worker-records.ts"; const baseUrl = "http://127.0.0.1:8787"; const webhookSecret = "test-secret"; @@ -220,3 +224,38 @@ test("exportWorkerRecords rides out a transient 502 during export", async () => assert.deepEqual(snapshot.records, []); assert.equal(calls.length, 2); }); + +test("fetchWorkerCanonicalItemIds pages the exact coverage identity set", async () => { + const requests: Array> = []; + const responses = [ + jsonResponse(200, { + repoSlug: "openclaw-openclaw", + section: "items", + records: [{ id: 1 }, { id: 500 }], + nextCursor: 500, + }), + jsonResponse(200, { + repoSlug: "openclaw-openclaw", + section: "items", + records: [{ id: 501 }, { id: 3_020 }], + nextCursor: null, + }), + ]; + const ids = await fetchWorkerCanonicalItemIds({ + baseUrl, + webhookSecret, + repoSlug: "openclaw-openclaw", + fetch: async (_input, init) => { + requests.push(JSON.parse(String(init?.body)) as Record); + const response = responses.shift(); + if (!response) throw new Error("fetch stub exhausted"); + return response; + }, + }); + + assert.deepEqual(ids, [1, 500, 501, 3_020]); + assert.deepEqual( + requests.map((request) => request.cursor), + [0, 500], + ); +});