Review each selected PR in its own item shard

AGENTS.md

@@ -29,9 +29,10 @@ not split reports into issue/PR subtrees.
   closing only unchanged, high-confidence proposals.
 - Repository-specific rules live in `src/repository-profiles.ts`; ClawHub apply
   may close only PRs that are certainly implemented on `main`.
-- Worker concurrency is shard-level: each shard processes its selected items
-  sequentially. Maximum parallel Codex sessions equals `shard_count`, not
-  `batch_size * shard_count`.
+- Review compute is item-sharded: every matrix job handles exactly one item.
+  `batch_size` limits planner selection, `shard_count` maps to matrix
+  `max-parallel`, and the durable queue permit is the authoritative cross-run
+  Codex concurrency limit.
 - `openclaw/clawsweeper-state` is the live status surface and generated state
   store. Check current Actions and that repo before trusting local generated
   timestamps.

README.md

@@ -380,10 +380,14 @@ scheduling, capacity, and monitoring behavior is documented in
 
 Review is proposal-only. It never closes items.
 
-- A planner scans open issues and PRs, then assigns exact item numbers to shards.
+- A planner scans open issues and PRs, then emits one matrix entry and one
+  cancellable compute job per item. `batch_size` limits planner selection;
+  `shard_count` controls matrix `max-parallel`.
 - Manual runs can pass `item_number` or comma-separated `item_numbers` to review
   exact Audit Health findings without scanning for a normal batch.
-- Each shard checks out the selected target repository at `main`.
+- Each item job checks out the selected target repository at `main`. Jobs for
+  different items have distinct concurrency groups; a newer generation cancels
+  only the older compute job for the same item.
 - Codex reviews with the internal model, high reasoning, the default service tier, and a
   10-minute per-item timeout.
 - Each item becomes a flat report under
@@ -777,16 +781,18 @@ ClawSweeper has one main capacity knob:
 This is a Codex worker budget, not a GitHub Actions runner limit. Deterministic
 exact-review publishers, comment routers, and lease reconcilers are
 control-plane workflows and do not consume these 128 slots.
-Lane limits are derived from that number: normal review defaults to 89 shards
+Lane limits are derived from that number: normal review defaults to 89 parallel item jobs
 for manual/backstop and scheduled runs, hot intake up to 44 shards, commit
 review 6 commits per page, and existing repair/issue implementation lanes use
 40% of `workers.max`, currently 51 live
 workers. Imported gitcrawl cluster repair allows 2 live workers by default.
 Exact-item review, repair, and issue implementation are priority work; normal
 review, hot intake, and commit review are background work and automatically
 yield when priority work is active. Exact-item runs use a durable Worker queue
-that coalesces item deliveries, leases at most 64 concurrent reviews, and admits
-up to 60 active exact reviews per target repository. Other lanes retain the
+that coalesces item deliveries and is the authoritative admission layer across
+batch, webhook, and re-review workflow runs. It leases at most 64 concurrent reviews
+and admits up to 60 active exact reviews per target repository; matrix
+`max-parallel` only bounds one workflow run. Other lanes retain the
 checked-in 128-worker scheduling model.
 Use `workers.max` first when turning total Codex usage up or down; use
 `lanes.repair.cluster_max_live_runs` to tune the imported legacy cluster-repair

dashboard/worker.ts

@@ -507,6 +507,12 @@ export default {
       return exactReviewQueueRequest(env, "/heartbeat", request);
     if (url.pathname === "/internal/exact-review/complete" && request.method === "POST")
       return exactReviewQueueRequest(env, "/complete", request);
+    if (url.pathname === "/internal/exact-review/generation/check" && request.method === "POST")
+      return authenticatedExactReviewQueueRequest(request, env, "/generation/check");
+    if (url.pathname === "/internal/exact-review/mutation/acquire" && request.method === "POST")
+      return authenticatedExactReviewQueueRequest(request, env, "/mutation/acquire");
+    if (url.pathname === "/internal/exact-review/mutation/release" && request.method === "POST")
+      return authenticatedExactReviewQueueRequest(request, env, "/mutation/release");
     if (url.pathname === "/internal/exact-review/claimed-runs" && request.method === "POST")
       return authenticatedExactReviewQueueRequest(request, env, "/claimed-runs");
     if (url.pathname === "/internal/exact-review/dead-letters/list" && request.method === "POST")

docs/pr-review-comments.md

@@ -27,10 +27,16 @@ Each synced comment includes the durable identity marker:
 ClawSweeper edits that comment in place instead of posting repeated comments.
 Report front matter stores the synced comment id, URL, hash, and sync time.
 
-When review starts and no ClawSweeper-owned comment exists yet, the review
-shard posts a short status placeholder with the same durable identity marker.
-The placeholder is intentionally light and crustacean-friendly, then the final
-review sync edits that exact comment in place.
+The review shard uses a separate transient status comment as its coordination
+lease. When a durable review already exists, acquiring that lease also edits
+the durable comment to show that a fresh review is in progress, marks the old
+review stale, and collapses it as previous context. The status projection has
+no active verdict or action markers, so downstream automation cannot act on the
+displaced review. A matching apply run replaces the projection with the new
+completed review; an abandoned or expired lease marks the refresh interrupted.
+
+When no durable review exists yet, the transient lease comment remains the only
+start status until the first completed review is published.
 
 For a PR that needs work, the visible comment starts with:
 

scripts/review-item-telemetry.mjs

@@ -0,0 +1,223 @@
+#!/usr/bin/env node
+
+import { createHmac } from "node:crypto";
+
+const PHASES = ["queue", "claim", "review", "publication", "total"];
+
+export function reviewTelemetryAttribution(input) {
+  const sourceAction = String(input.sourceAction || "").toLowerCase();
+  const sourceEvent = String(input.sourceEvent || "").toLowerCase();
+  const eventName = String(input.eventName || "").toLowerCase();
+  const recovery = sourceAction.includes("recovery") || input.recovery === true;
+  const command =
+    sourceAction.includes("command") ||
+    sourceAction.includes("router") ||
+    sourceEvent === "issue_comment";
+  const lane = recovery
+    ? "recovery"
+    : input.exact === true
+      ? "exact_event"
+      : input.hotIntake === true
+        ? "hot_intake"
+        : "normal_backfill";
+  const origin = command
+    ? "command"
+    : eventName === "schedule"
+      ? "schedule"
+      : eventName === "workflow_dispatch"
+        ? "manual"
+        : eventName === "repository_dispatch" &&
+            (sourceEvent === "issues" || sourceEvent === "pull_request")
+          ? "webhook"
+          : "system";
+  return { lane, origin };
+}
+
+export function buildReviewTelemetryRecord(input, existing, now = new Date()) {
+  const nowIso = now.toISOString();
+  const startedAt = existing?.started_at || input.startedAt || nowIso;
+  const phaseDurations = { ...existing?.phase_durations_ms };
+  for (const phase of PHASES) {
+    const value = input.phaseDurations?.[phase];
+    if (Number.isSafeInteger(value) && value >= 0) phaseDurations[phase] = value;
+  }
+  // Total is a live wall-clock duration, so carrying the start payload's value
+  // forward would make healthy heartbeats look permanently stalled.
+  phaseDurations.total =
+    input.phaseDurations?.total ?? Math.max(0, now.getTime() - Date.parse(startedAt));
+  const terminal = input.action === "terminal";
+  return {
+    repo: input.repo,
+    item_number: input.itemNumber,
+    run_id: input.runId,
+    run_attempt: input.runAttempt,
+    status: terminal ? "completed" : "refreshing",
+    outcome: terminal ? input.outcome : null,
+    started_at: startedAt,
+    updated_at: nowIso,
+    lease_expires_at: terminal ? null : input.leaseExpiresAt,
+    phase_durations_ms: phaseDurations,
+    generation: input.generation,
+    operation_id: input.operationId,
+    trigger_lane: input.triggerLane,
+    trigger_origin: input.triggerOrigin,
+    ...(input.sourceEvent && { source_event: input.sourceEvent }),
+    ...(input.sourceAction && { source_action: input.sourceAction }),
+    ...(terminal ? { terminal_reason: input.terminalReason, terminal_at: nowIso } : {}),
+  };
+}
+
+function elapsed(start, end) {
+  const startMs = Date.parse(String(start || ""));
+  const endMs = Date.parse(String(end || ""));
+  return Number.isFinite(startMs) && Number.isFinite(endMs) && endMs >= startMs
+    ? endMs - startMs
+    : undefined;
+}
+
+function positiveInteger(name, value) {
+  const parsed = Number(value);
+  if (!Number.isSafeInteger(parsed) || parsed < 1) throw new Error(`${name} must be positive`);
+  return parsed;
+}
+
+function optionalDuration(name) {
+  const value = process.env[name];
+  if (value === undefined || value === "") return undefined;
+  const parsed = Number(value);
+  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : undefined;
+}
+
+function required(name) {
+  const value = String(process.env[name] || "").trim();
+  if (!value) throw new Error(`${name} is required`);
+  return value;
+}
+
+async function existingRecord(queueUrl, input) {
+  if (input.action === "start") return undefined;
+  const url = new URL("/api/exact-review-queue/reviews", queueUrl);
+  url.searchParams.set("repo", input.repo);
+  url.searchParams.set("item_number", String(input.itemNumber));
+  url.searchParams.set("limit", "100");
+  const response = await fetch(url, { signal: AbortSignal.timeout(20_000) });
+  if (!response.ok) throw new Error(`telemetry read returned HTTP ${response.status}`);
+  const body = await response.json();
+  return body.reviews?.find(
+    (record) => record.run_id === input.runId && Number(record.run_attempt) === input.runAttempt,
+  );
+}
+
+async function writeRecord(queueUrl, secret, record) {
+  const payload = JSON.stringify(record);
+  const signature = `sha256=${createHmac("sha256", secret).update(payload).digest("hex")}`;
+  const response = await fetch(new URL("/internal/exact-review/review-telemetry", queueUrl), {
+    method: "POST",
+    headers: {
+      "content-type": "application/json",
+      "x-clawsweeper-exact-review-signature": signature,
+    },
+    body: payload,
+    signal: AbortSignal.timeout(20_000),
+  });
+  if (!response.ok) {
+    throw new Error(`telemetry write returned HTTP ${response.status}: ${await response.text()}`);
+  }
+}
+
+export async function runReviewTelemetryProducer() {
+  const action = required("REVIEW_TELEMETRY_ACTION");
+  if (!["start", "heartbeat", "terminal"].includes(action)) {
+    throw new Error("REVIEW_TELEMETRY_ACTION must be start, heartbeat, or terminal");
+  }
+  const sourceEvent = String(process.env.REVIEW_TELEMETRY_SOURCE_EVENT || "").trim();
+  const sourceAction = String(process.env.REVIEW_TELEMETRY_SOURCE_ACTION || "").trim();
+  const attribution = reviewTelemetryAttribution({
+    sourceEvent,
+    sourceAction,
+    eventName: process.env.GITHUB_EVENT_NAME,
+    exact: process.env.REVIEW_TELEMETRY_EXACT === "true",
+    hotIntake: process.env.REVIEW_TELEMETRY_HOT_INTAKE === "true",
+    recovery: process.env.REVIEW_TELEMETRY_RECOVERY === "true",
+  });
+  const leaseMs = optionalDuration("REVIEW_TELEMETRY_LEASE_MS");
+  const now = new Date();
+  const input = {
+    action,
+    repo: required("REVIEW_TELEMETRY_REPO"),
+    itemNumber: positiveInteger("item number", process.env.REVIEW_TELEMETRY_ITEM_NUMBER),
+    runId: required("REVIEW_TELEMETRY_RUN_ID"),
+    runAttempt: positiveInteger("run attempt", process.env.REVIEW_TELEMETRY_RUN_ATTEMPT),
+    generation: process.env.REVIEW_TELEMETRY_GENERATION
+      ? positiveInteger("generation", process.env.REVIEW_TELEMETRY_GENERATION)
+      : undefined,
+    operationId: String(process.env.REVIEW_TELEMETRY_OPERATION_ID || "").trim() || undefined,
+    triggerLane: String(process.env.REVIEW_TELEMETRY_TRIGGER_LANE || attribution.lane),
+    triggerOrigin: String(process.env.REVIEW_TELEMETRY_TRIGGER_ORIGIN || attribution.origin),
+    sourceEvent,
+    sourceAction,
+    startedAt:
+      String(
+        process.env.REVIEW_TELEMETRY_STARTED_AT || process.env.REVIEW_TELEMETRY_QUEUED_AT || "",
+      ).trim() || undefined,
+    leaseExpiresAt: leaseMs === undefined ? null : new Date(now.getTime() + leaseMs).toISOString(),
+    outcome: String(process.env.REVIEW_TELEMETRY_OUTCOME || "").trim(),
+    terminalReason: String(process.env.REVIEW_TELEMETRY_TERMINAL_REASON || "").trim(),
+    phaseDurations: Object.fromEntries(
+      PHASES.map((phase) => [
+        phase,
+        optionalDuration(`REVIEW_TELEMETRY_${phase.toUpperCase()}_MS`),
+      ]).filter(([, value]) => value !== undefined),
+    ),
+  };
+  const derivedPhases = {
+    queue: elapsed(
+      process.env.REVIEW_TELEMETRY_QUEUED_AT,
+      process.env.REVIEW_TELEMETRY_DISPATCHED_AT,
+    ),
+    claim: elapsed(
+      process.env.REVIEW_TELEMETRY_DISPATCHED_AT || process.env.REVIEW_TELEMETRY_QUEUED_AT,
+      process.env.REVIEW_TELEMETRY_CLAIMED_AT,
+    ),
+    review: elapsed(
+      process.env.REVIEW_TELEMETRY_REVIEW_STARTED_AT,
+      process.env.REVIEW_TELEMETRY_REVIEW_COMPLETED_AT,
+    ),
+    publication: elapsed(process.env.REVIEW_TELEMETRY_PUBLICATION_STARTED_AT, nowIso(now)),
+  };
+  for (const [phase, duration] of Object.entries(derivedPhases)) {
+    if (input.phaseDurations[phase] === undefined && duration !== undefined) {
+      input.phaseDurations[phase] = duration;
+    }
+  }
+  if (action === "terminal" && (!input.outcome || !input.terminalReason)) {
+    throw new Error("terminal telemetry requires outcome and terminal reason");
+  }
+  const queueUrl = required("REVIEW_TELEMETRY_QUEUE_URL").replace(/\/$/, "");
+  const existing = await existingRecord(queueUrl, input);
+  // First terminal truth is immutable in the Durable Object. Avoid a redundant
+  // write so retries also preserve that decision before crossing the network.
+  if (existing?.status === "completed") return;
+  if (action !== "start" && !existing) {
+    throw new Error("refreshing telemetry row is unavailable for update");
+  }
+  await writeRecord(
+    queueUrl,
+    required("CLAWSWEEPER_WEBHOOK_SECRET"),
+    buildReviewTelemetryRecord(input, existing, now),
+  );
+}
+
+function nowIso(now) {
+  return now.toISOString();
+}
+
+if (import.meta.url === `file://${process.argv[1]}`) {
+  runReviewTelemetryProducer().catch((error) => {
+    console.log(`::warning::Review telemetry producer skipped: ${error?.message || error}`);
+    // Workflow steps use continue-on-error so telemetry remains fail-open for
+    // review, while a nonzero outcome prevents dependent heartbeats from
+    // pretending that the refreshing row exists.
+    process.exitCode = 1;
+  });
+}