diff --git a/.github/workflows/sweep.yml b/.github/workflows/sweep.yml index af84a58c8f..b7925adb64 100644 --- a/.github/workflows/sweep.yml +++ b/.github/workflows/sweep.yml @@ -17,7 +17,7 @@ run-name: >- github.event.action == 'clawsweeper_target_sweep') && format('Review target repo {0}', github.event.client_payload.target_repo || 'openclaw/openclaw') || (github.event_name == 'repository_dispatch' && github.event.client_payload.queue_lease_id != '') && - format('Review event item {0}', github.event.client_payload.queue_claim.item_key || github.event.client_payload.item_key || '?') || + format('Review exact item {0} rev {1} head {2}', github.event.client_payload.queue_claim.item_key || github.event.client_payload.item_key || '?', github.event.client_payload.queue_claim.lease_revision || github.event.client_payload.lease_revision || '?', github.event.client_payload.source_head_sha || 'na') || (github.event_name == 'repository_dispatch' && github.event.client_payload.dispatch_key != '') && format('Review event item {0}#{1} [{2}]', github.event.client_payload.target_repo || 'openclaw/openclaw', github.event.client_payload.item_number || '?', github.event.client_payload.dispatch_key) || github.event_name == 'repository_dispatch' && @@ -278,6 +278,9 @@ jobs: sourceEvent, sourceAction: payload.source_action || "legacy_dispatch", supersedesInProgress: payload.supersedes_in_progress === true, + ...(/^[0-9a-f]{40}$/.test(String(payload.source_head_sha || "").trim().toLowerCase()) + ? { sourceHeadSha: String(payload.source_head_sha).trim().toLowerCase() } + : {}), ...(Number.isFinite(Number(payload.codex_timeout_ms)) ? { codexTimeoutMs: Number(payload.codex_timeout_ms) } : {}), @@ -413,6 +416,9 @@ jobs: sourceEvent: String(dispatch.source_event || ""), sourceAction: String(dispatch.source_action || "legacy_dispatch"), supersedesInProgress: dispatch.supersedes_in_progress === true, + ...(/^[0-9a-f]{40}$/.test(String(dispatch.source_head_sha || "").trim().toLowerCase()) + ? { sourceHeadSha: String(dispatch.source_head_sha).trim().toLowerCase() } + : {}), ...(Number.isFinite(Number(reviewOptions.codex_timeout_ms)) ? { codexTimeoutMs: Number(reviewOptions.codex_timeout_ms) } : {}), @@ -785,6 +791,12 @@ jobs: ITEM_NUMBER: ${{ steps.target.outputs.item_number }} CODEX_TIMEOUT_MS: ${{ steps.target.outputs.codex_timeout_ms }} MEDIA_PROOF_TIMEOUT_MS: ${{ steps.target.outputs.media_proof_timeout_ms }} + EXACT_REVIEW_ITEM_KEY: ${{ steps.claim-exact-review-queue.outputs.item_key }} + EXACT_REVIEW_LEASE_ID: ${{ steps.claim-exact-review-queue.outputs.lease_id }} + EXACT_REVIEW_LEASE_REVISION: ${{ steps.claim-exact-review-queue.outputs.lease_revision }} + EXACT_REVIEW_CLAIM_GENERATION: ${{ steps.claim-exact-review-queue.outputs.claim_generation }} + EXACT_REVIEW_SOURCE_HEAD_SHA: ${{ fromJSON(steps.claim-exact-review-queue.outputs.decision).sourceHeadSha || '' }} + EXACT_REVIEW_QUEUE_URL: ${{ vars.CLAWSWEEPER_EXACT_REVIEW_QUEUE_URL || 'https://clawsweeper.openclaw.ai' }} run: | set -euo pipefail test -n "$GH_TOKEN" @@ -849,6 +861,8 @@ jobs: EXACT_REVIEW_ITEM_KEY: ${{ steps.claim-exact-review-queue.outputs.item_key }} EXACT_REVIEW_LEASE_ID: ${{ steps.claim-exact-review-queue.outputs.lease_id }} EXACT_REVIEW_LEASE_REVISION: ${{ steps.claim-exact-review-queue.outputs.lease_revision }} + EXACT_REVIEW_CLAIM_GENERATION: ${{ steps.claim-exact-review-queue.outputs.claim_generation }} + EXACT_REVIEW_SOURCE_HEAD_SHA: ${{ fromJSON(steps.claim-exact-review-queue.outputs.decision).sourceHeadSha || '' }} QUEUE_URL: ${{ vars.CLAWSWEEPER_EXACT_REVIEW_QUEUE_URL || 'https://clawsweeper.openclaw.ai' }} REVIEW_LEASE_OWNER: ${{ steps.reserve-exact-review-lease.outputs.owner }} REVIEW_LEASE_COMMENT_ID: ${{ steps.reserve-exact-review-lease.outputs.comment_id }} @@ -856,6 +870,32 @@ jobs: set -euo pipefail test -n "$GH_TOKEN" heartbeat_pid="" + review_pid="" + review_pgid="" + superseded_marker="$RUNNER_TEMP/exact-review-superseded-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" + rm -f "$superseded_marker" + terminate_review_group() { + [ -n "$review_pgid" ] || return 0 + kill -TERM -- "-$review_pgid" 2>/dev/null || true + } + wait_for_review_group() { + [ -n "$review_pgid" ] || return 0 + for _ in {1..30}; do + if ! kill -0 -- "-$review_pgid" 2>/dev/null; then + return 0 + fi + sleep 1 + done + kill -KILL -- "-$review_pgid" 2>/dev/null || true + for _ in {1..50}; do + if ! kill -0 -- "-$review_pgid" 2>/dev/null; then + return 0 + fi + sleep 0.1 + done + echo "::error::Exact-review process group $review_pgid did not terminate." + return 1 + } heartbeat_loop() { while true; do heartbeat_status="$(curl --silent --connect-timeout 5 --max-time 20 \ @@ -867,9 +907,11 @@ jobs: "${QUEUE_URL%/}/internal/exact-review/heartbeat" 2>/dev/null)" || heartbeat_status="" if [ "$heartbeat_status" = "409" ]; then echo "::notice::Exact-review heartbeat lost its lease tuple." + : > "$superseded_marker" + terminate_review_group return fi - sleep 300 + sleep 60 done } start_heartbeat() { @@ -877,13 +919,22 @@ jobs: # claim/complete), so this Codex-adjacent step never sees the webhook secret. heartbeat_payload="$(node -e ' const leaseRevision = Number(process.env.EXACT_REVIEW_LEASE_REVISION); + const claimGeneration = Number(process.env.EXACT_REVIEW_CLAIM_GENERATION); + const runAttempt = Number(process.env.GITHUB_RUN_ATTEMPT); + const sourceHeadSha = String(process.env.EXACT_REVIEW_SOURCE_HEAD_SHA || "").trim().toLowerCase(); if (!process.env.EXACT_REVIEW_ITEM_KEY || !process.env.EXACT_REVIEW_LEASE_ID) process.exit(1); if (!Number.isInteger(leaseRevision) || leaseRevision < 1) process.exit(1); + if (!Number.isInteger(claimGeneration) || claimGeneration < 1) process.exit(1); + if (!Number.isInteger(runAttempt) || runAttempt < 1) process.exit(1); + if (sourceHeadSha && !/^[0-9a-f]{40}$/.test(sourceHeadSha)) process.exit(1); process.stdout.write(JSON.stringify({ item_key: process.env.EXACT_REVIEW_ITEM_KEY, lease_id: process.env.EXACT_REVIEW_LEASE_ID, lease_revision: leaseRevision, + claim_generation: claimGeneration, run_id: process.env.GITHUB_RUN_ID, + run_attempt: runAttempt, + ...(sourceHeadSha ? { source_head_sha: sourceHeadSha } : {}), })); ')" || return 0 heartbeat_loop & @@ -894,7 +945,6 @@ jobs: kill "$heartbeat_pid" 2>/dev/null || true wait "$heartbeat_pid" 2>/dev/null || true } - start_heartbeat trap cleanup_heartbeat EXIT codex_timeout_ms="${{ steps.target.outputs.codex_timeout_ms }}" if ! [[ "$codex_timeout_ms" =~ ^[0-9]+$ ]] || [ "$codex_timeout_ms" -lt 1 ]; then @@ -912,7 +962,7 @@ jobs: review_timeout_seconds=$((codex_timeout_seconds + media_preprocessing_reserve_seconds + 180)) echo "::notice::Exact event review timeout is ${review_timeout_seconds}s for per-item Codex timeout ${codex_timeout_seconds}s, reserved media preprocessing ${media_preprocessing_reserve_seconds}s, and detected media allowance ${media_proof_timeout_seconds}s." set +e - timeout --kill-after=30s "${review_timeout_seconds}s" pnpm run review -- \ + setsid timeout --kill-after=30s "${review_timeout_seconds}s" pnpm run review -- \ --target-repo "${{ steps.target.outputs.target_repo }}" \ --target-dir "${{ steps.target.outputs.target_checkout_dir }}" \ --artifact-dir artifacts/event \ @@ -929,8 +979,17 @@ jobs: --review-lease-comment-id "$REVIEW_LEASE_COMMENT_ID" \ --shard-index 0 \ --shard-count 1 \ - "${additional_prompt_arg[@]}" + "${additional_prompt_arg[@]}" & + review_pid=$! + review_pgid=$review_pid + start_heartbeat + wait "$review_pid" review_exit_code=$? + cleanup_heartbeat + if [ -f "$superseded_marker" ]; then + wait_for_review_group + review_exit_code=143 + fi set -e echo "exit_code=$review_exit_code" >> "$GITHUB_OUTPUT" if [ "$review_exit_code" -ne 0 ]; then diff --git a/dashboard/exact-review-queue.ts b/dashboard/exact-review-queue.ts index cb4d0008b7..3c994fbeec 100644 --- a/dashboard/exact-review-queue.ts +++ b/dashboard/exact-review-queue.ts @@ -56,6 +56,10 @@ export type ExactReviewBaseDecision = { sourceEvent: "issues" | "pull_request"; sourceAction: string; supersedesInProgress: boolean; + sourceHeadSha?: string; + sourceHeadVerified?: boolean; + sourceAuthoritySeq?: number; + sourceUpdatedAt?: string; codexTimeoutMs?: number; mediaProofTimeoutMs?: number; commandStatusMarker?: string; @@ -210,7 +214,7 @@ type ExactReviewQueueStorageMeta = { shed_since_reset?: number; }; type ExactReviewQueueMetricTotals = { - review: { enqueued: number; completed: number }; + review: { enqueued: number; completed: number; superseded: number }; publication: { enqueued: number; completed: number; @@ -222,9 +226,18 @@ type ExactReviewQueueMetricTotals = { refreshed: number; }; }; +type ExactReviewSourceAuthorityReservation = { + deliveryId: string; + decision: ExactReviewDecision; + installationId: number; + sourceAuthoritySeq: number; + attempts: number; + nextAttemptAt: number; +}; type ExactReviewQueueMetricDelta = { reviewEnqueued?: number; reviewCompleted?: number; + reviewSuperseded?: number; reviewRetried?: number; reviewShed?: number; publicationEnqueued?: number; @@ -253,6 +266,14 @@ type StateAppendWindowRow = { produced_at: string; delivery_id: string; }; +type ExactReviewSupersessionAudit = { + itemKey: string; + priorRevision: number; + nextRevision: number; + supersededRunId: string | null; + sourceAction: string; + supersededAt: number; +}; export type DurableObjectStub = { fetch: (request: Request) => Promise }; export type DurableObjectNamespace = { idFromName: (name: string) => unknown; @@ -285,7 +306,7 @@ const DEFAULT_EXACT_REVIEW_EXECUTION_LEASE_MS = 130 * 60 * 1000; const DEFAULT_EXACT_REVIEW_HEARTBEAT_GRACE_MS = 20 * 60 * 1000; const DEFAULT_EXACT_REVIEW_RETRY_MS = 30_000; const DEFAULT_EXACT_REVIEW_WORKFLOW_PAUSED_RETRY_MS = 60_000; -const DEFAULT_EXACT_REVIEW_DISPATCH_DEBOUNCE_MS = 45_000; +const DEFAULT_EXACT_REVIEW_DISPATCH_DEBOUNCE_MS = 90_000; const DEFAULT_EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS = 3 * 60_000; const DEFAULT_EXACT_REVIEW_PENDING_SOFT_LIMIT = 300; const EXACT_REVIEW_COMPLETION_RETRY_MAX_MS = 2 * 60 * 60 * 1000; @@ -316,11 +337,18 @@ const EXACT_REVIEW_QUEUE_LEGACY_RECEIPT_SHIFT_MS = 2 * 24 * 60 * 60 * 1000; const EXACT_REVIEW_QUEUE_ROLLBACK_CLOCK_SKEW_MS = 5 * 60 * 1000; const EXACT_REVIEW_QUEUE_LEGACY_GENERATION_PREFIX = "__clawsweeper_sql_generation:"; const EXACT_REVIEW_QUEUE_STATE_KEY = "exact-review-queue"; +const EXACT_REVIEW_SOURCE_AUTHORITY_SEQUENCE_KEY = "exact-review-source-authority-sequence:v1"; +const EXACT_REVIEW_SOURCE_AUTHORITY_RESERVATION_PREFIX = + "exact-review-source-authority-reservation:v1:"; +const EXACT_REVIEW_SOURCE_AUTHORITY_RETRY_LIMIT = 16; +const EXACT_REVIEW_SOURCE_AUTHORITY_RETRY_BASE_MS = 15_000; +const EXACT_REVIEW_SOURCE_AUTHORITY_RETRY_MAX_MS = 15 * 60_000; const EXACT_REVIEW_QUEUE_META_TABLE = "exact_review_queue_meta"; const EXACT_REVIEW_QUEUE_ITEM_TABLE = "exact_review_queue_items"; const EXACT_REVIEW_QUEUE_DELIVERY_TABLE = "exact_review_queue_deliveries"; const EXACT_REVIEW_QUEUE_METRICS_TABLE = "exact_review_queue_metrics"; const EXACT_REVIEW_QUEUE_METRIC_BUCKET_TABLE = "exact_review_queue_metric_buckets"; +const EXACT_REVIEW_QUEUE_SUPERSESSION_TABLE = "exact_review_queue_supersessions"; const EXACT_REVIEW_QUEUE_DEAD_LETTER_TABLE = "exact_review_queue_dead_letters"; const EXACT_REVIEW_PUBLICATION_HEAD_TABLE = "exact_review_publication_heads"; const STATE_APPEND_WINDOW_TABLE = "state_append_window"; @@ -350,6 +378,7 @@ const EXACT_REVIEW_QUEUE_DEAD_LETTER_LIMIT = 5_000; const EXACT_REVIEW_QUEUE_DEAD_LETTER_RESOLVED_TTL_MS = 30 * 24 * 60 * 60 * 1000; const EXACT_REVIEW_QUEUE_METRIC_BUCKET_MS = 5 * 60 * 1000; const EXACT_REVIEW_QUEUE_METRIC_BUCKET_TTL_MS = 48 * 60 * 60 * 1000; +const EXACT_REVIEW_QUEUE_SUPERSESSION_RETENTION_MS = 7 * 24 * 60 * 60 * 1000; const EXACT_REVIEW_PUBLICATION_CONTROL_KEY = "exact-review-publication-control:v1"; export const EXACT_REVIEW_QUEUE_NAME = "global"; const EXACT_REVIEW_COMMAND_STATUS_MARKER_PATTERN = @@ -396,6 +425,111 @@ export class ExactReviewQueue { await this.ready; this.cleanupLegacyCompatibilitySync(); const url = new URL(request.url); + if (request.method === "POST" && url.pathname === "/source-authority") { + const body = objectValue(await request.json().catch(() => null)); + const deliveryId = String(body.delivery_id || "").trim(); + const decision = exactReviewDecisionFrom(body.decision); + const installationId = Number(body.installation_id); + if ( + !deliveryId || + deliveryId.length > 200 || + !decision || + decision.itemKind !== "pull_request" || + decision.publication || + !Number.isInteger(installationId) || + installationId <= 0 + ) { + return json({ error: "invalid_source_authority_reservation" }, 400); + } + const reservationKey = exactReviewSourceAuthorityReservationKey(deliveryId); + const now = Date.now(); + try { + const reserved = this.storage.transactionSync(() => { + const completed = Array.from( + this.storage.sql.exec( + `SELECT delivery_id FROM ${EXACT_REVIEW_QUEUE_DELIVERY_TABLE} + WHERE delivery_id = ?`, + deliveryId, + ), + ).length; + if (completed) return { deduped: true as const }; + const existing = exactReviewSourceAuthorityReservationFrom( + this.storage.kv.get(reservationKey), + ); + if (existing) { + if ( + existing.deliveryId !== deliveryId || + existing.installationId !== installationId || + stableJson(exactReviewDecisionWithoutSourceAuthority(existing.decision)) !== + stableJson(decision) + ) { + throw new Error("conflicting exact-review source authority reservation"); + } + return { deduped: false as const, reservation: existing }; + } + const stored = this.storage.kv.get(EXACT_REVIEW_SOURCE_AUTHORITY_SEQUENCE_KEY); + const current = stored === undefined ? 0 : Number(stored); + if (!Number.isSafeInteger(current) || current < 0 || current >= Number.MAX_SAFE_INTEGER) { + throw new Error("invalid exact-review source authority sequence"); + } + const next = current + 1; + this.storage.kv.put(EXACT_REVIEW_SOURCE_AUTHORITY_SEQUENCE_KEY, next); + const created: ExactReviewSourceAuthorityReservation = { + deliveryId, + decision: { ...decision, sourceAuthoritySeq: next }, + installationId, + sourceAuthoritySeq: next, + attempts: 0, + nextAttemptAt: now, + }; + this.storage.kv.put(reservationKey, created); + return { deduped: false as const, reservation: created }; + }); + if (reserved.deduped) return json({ ok: true, deduped: true }); + await this.scheduleSourceAuthorityVerification(reserved.reservation.nextAttemptAt); + return json({ + ok: true, + source_authority_seq: reserved.reservation.sourceAuthoritySeq, + }); + } catch { + return json({ error: "source_authority_unavailable" }, 409); + } + } + if (request.method === "POST" && url.pathname === "/source-authority/complete") { + const body = objectValue(await request.json().catch(() => null)); + const deliveryId = String(body.delivery_id || "").trim(); + const sourceAuthoritySeq = Number(body.source_authority_seq); + const disposition = String(body.disposition || ""); + if ( + !deliveryId || + !Number.isSafeInteger(sourceAuthoritySeq) || + sourceAuthoritySeq <= 0 || + (disposition !== "enqueued" && disposition !== "mismatch") + ) { + return json({ error: "invalid_source_authority_completion" }, 400); + } + const result = this.storage.transactionSync(() => { + const reservationKey = exactReviewSourceAuthorityReservationKey(deliveryId); + const reservation = exactReviewSourceAuthorityReservationFrom( + this.storage.kv.get(reservationKey), + ); + if (!reservation) return "missing" as const; + if (reservation.sourceAuthoritySeq !== sourceAuthoritySeq) return "conflict" as const; + if (disposition === "mismatch") { + this.storage.sql.exec( + `INSERT OR IGNORE INTO ${EXACT_REVIEW_QUEUE_DELIVERY_TABLE} + (delivery_id, received_at) VALUES (?, ?)`, + deliveryId, + Date.now(), + ); + } + this.storage.kv.delete(reservationKey); + return "completed" as const; + }); + if (result === "conflict") return json({ error: "source_authority_conflict" }, 409); + await this.scheduleNext(this.readStateSync(), Date.now()); + return json({ ok: true, completed: result === "completed" }); + } if (request.method === "POST" && url.pathname === "/state/append") { const body = objectValue(await request.json().catch(() => null)); const deliveryId = String(body.delivery_id || "").trim(); @@ -818,19 +952,71 @@ export class ExactReviewQueue { } const key = exactReviewItemKey(decision); const current = state.items[key]; + let supersededRunId: string | null = null; + let supersessionAudit: ExactReviewSupersessionAudit | null = null; if (current) { - const ignoredRecovery = - decision.sourceAction === FAILED_REVIEW_SHARD_RECOVERY_SOURCE_ACTION; + const ignoredRecovery = isLowPriorityExactReviewDecision(decision); // A recovery is only a one-shot repair of a failed shard. It may create a queue item, // but must never supersede an existing pending, dispatching, or leased decision: doing // so can leave either ordinary work or another recovery as a stale follow-up revision. // Ordinary source events retain normal replacement behavior, including the // command-context merge for pending items. if (!ignoredRecovery) { + // Explicit commands arrive through repository_dispatch without a webhook authority + // tuple. Bind them to the current verified decision via the merge below. An active + // review keeps its lease and exposes the command as a follow-up revision on completion. + const bindsCommandToCurrentAuthority = + decision.itemKind === "pull_request" && + (current.leaseDecision || current.decision).itemKind === "pull_request" && + Boolean(decision.commandStatusMarker) && + !Object.hasOwn(decision, "sourceHeadSha") && + !Object.hasOwn(decision, "sourceAuthoritySeq"); + const queuesCommandFollowUp = + bindsCommandToCurrentAuthority && + (current.state === "dispatching" || current.state === "leased"); + const attemptsReviewSupersession = + !exactReviewQueueIsPublication(current) && + decision.itemKind === "pull_request" && + !bindsCommandToCurrentAuthority; + const sourceAuthorityIsNewer = + !attemptsReviewSupersession || + exactReviewDecisionCanSupersedeReview(current, decision); + if (attemptsReviewSupersession && !sourceAuthorityIsNewer) { + this.writeStateSync(state); + return { + deduped: true as const, + staleSource: true as const, + key, + state, + }; + } + const supersedesActiveReview = + !bindsCommandToCurrentAuthority && + sourceAuthorityIsNewer && + decision.supersedesInProgress && + (current.state === "dispatching" || current.state === "leased"); + if (supersedesActiveReview) { + const priorRevision = current.revision; + supersededRunId = current.claimedRunId || null; + supersessionAudit = { + itemKey: key, + priorRevision, + nextRevision: priorRevision + 1, + supersededRunId, + sourceAction: decision.sourceAction, + supersededAt: now, + }; + clearExactReviewLease(current); + current.state = "pending"; + current.createdAt = now; + current.parkedReason = undefined; + } const mergeable = current.state === "pending" || current.state === "parked"; - current.decision = mergeable - ? mergePendingExactReviewDecision(current.decision, decision) - : decision; + current.decision = supersedesActiveReview + ? decision + : mergeable || queuesCommandFollowUp + ? mergePendingExactReviewDecision(current.decision, decision) + : decision; current.revision += 1; current.updatedAt = now; // Immediacy must come from the merged decision: a pending explicit command @@ -883,7 +1069,17 @@ export class ExactReviewQueue { publicationSuperseded: supersededPublications, }); } - return { deduped: false as const, key, state, supersededPublications }; + if (supersessionAudit) { + this.insertSupersessionAuditSync(supersessionAudit); + this.incrementQueueMetricsSync({ reviewSuperseded: 1 }); + } + return { + deduped: false as const, + key, + state, + supersededPublications, + supersededRunId, + }; }); if (accepted.deduped) { if ("state" in accepted) await this.scheduleNext(accepted.state, now); @@ -901,6 +1097,7 @@ export class ExactReviewQueue { semantic_duplicates_removed: accepted.semanticDuplicatesRemoved, } : {}), + ...("staleSource" in accepted && accepted.staleSource ? { stale_source: true } : {}), ...(accepted.superseded ? { superseded: true, @@ -1050,11 +1247,28 @@ export class ExactReviewQueue { const leaseId = String(body.lease_id || "").trim(); const leaseRevision = Number(body.lease_revision); const runId = String(body.run_id || "").trim(); + const hasRunAttempt = body.run_attempt !== undefined; + const runAttempt = hasRunAttempt ? exactReviewRunAttempt(body.run_attempt) : null; + const hasClaimGeneration = body.claim_generation !== undefined; + const claimGeneration = hasClaimGeneration ? Number(body.claim_generation) : null; + const hasSourceHeadSha = body.source_head_sha !== undefined; + const sourceHeadSha = hasSourceHeadSha + ? String(body.source_head_sha || "") + .trim() + .toLowerCase() + : null; if (!itemKey || !leaseId || !runId) return json({ error: "missing_lease_tuple" }, 400); if (!Number.isInteger(leaseRevision) || leaseRevision < 1) { return json({ error: "invalid_lease_revision" }, 400); } if (!/^\d+$/.test(runId)) return json({ error: "invalid_run_id" }, 400); + if (hasRunAttempt && runAttempt === null) return json({ error: "invalid_run_attempt" }, 400); + if (hasClaimGeneration && (!Number.isInteger(claimGeneration) || claimGeneration < 1)) { + return json({ error: "invalid_claim_generation" }, 400); + } + if (hasSourceHeadSha && !/^[0-9a-f]{40}$/.test(sourceHeadSha || "")) { + return json({ error: "invalid_source_head_sha" }, 400); + } const now = Date.now(); const state = this.readStateSync(); @@ -1080,6 +1294,12 @@ export class ExactReviewQueue { item.leaseId !== leaseId || item.leaseRevision !== leaseRevision || item.claimedRunId !== runId || + (hasRunAttempt && item.claimedRunAttempt !== runAttempt) || + (hasClaimGeneration && + exactReviewClaimGeneration(item.claimGeneration) !== claimGeneration) || + (item.leaseDecision?.sourceHeadSha && + (!hasSourceHeadSha || + item.leaseDecision.sourceHeadSha.toLowerCase() !== sourceHeadSha)) || !isLiveExactReviewLease( item, now, @@ -1647,6 +1867,7 @@ export class ExactReviewQueue { ...stats.lanes.review, enqueued_total: metrics.review.enqueued, completed_total: metrics.review.completed, + superseded_total: metrics.review.superseded, flow: reviewFlow, }, publication: { @@ -1718,6 +1939,7 @@ export class ExactReviewQueue { this.cleanupLegacyCompatibilitySync(); const startedAt = Date.now(); await this.storage.deleteAlarm(); + await this.processSourceAuthorityReservations(startedAt); this.storage.transactionSync(() => { this.pruneDeliveryReceiptsSync(startedAt); this.pruneStateAppendReceiptsSync(startedAt); @@ -3406,6 +3628,8 @@ export class ExactReviewQueue { singleton_id INTEGER PRIMARY KEY CHECK (singleton_id = 1), review_enqueued_total INTEGER NOT NULL DEFAULT 0 CHECK (review_enqueued_total >= 0), review_completed_total INTEGER NOT NULL DEFAULT 0 CHECK (review_completed_total >= 0), + review_superseded_total INTEGER NOT NULL DEFAULT 0 + CHECK (review_superseded_total >= 0), publication_enqueued_total INTEGER NOT NULL DEFAULT 0 CHECK (publication_enqueued_total >= 0), publication_completed_total INTEGER NOT NULL CHECK (publication_completed_total >= 0) @@ -3414,6 +3638,7 @@ export class ExactReviewQueue { for (const column of [ "review_enqueued_total", "review_completed_total", + "review_superseded_total", "publication_enqueued_total", "publication_published_total", "publication_superseded_total", @@ -3445,11 +3670,27 @@ export class ExactReviewQueue { `CREATE INDEX IF NOT EXISTS exact_review_queue_deliveries_received_at ON ${EXACT_REVIEW_QUEUE_DELIVERY_TABLE} (received_at, delivery_id)`, ); + this.storage.sql.exec( + `CREATE TABLE IF NOT EXISTS ${EXACT_REVIEW_QUEUE_SUPERSESSION_TABLE} ( + item_key TEXT NOT NULL, + prior_revision INTEGER NOT NULL CHECK (prior_revision >= 1), + next_revision INTEGER NOT NULL CHECK (next_revision > prior_revision), + superseded_run_id TEXT, + source_action TEXT NOT NULL, + superseded_at INTEGER NOT NULL, + PRIMARY KEY (item_key, prior_revision, next_revision) + ) STRICT`, + ); + this.storage.sql.exec( + `CREATE INDEX IF NOT EXISTS exact_review_queue_supersessions_at + ON ${EXACT_REVIEW_QUEUE_SUPERSESSION_TABLE} (superseded_at, item_key)`, + ); this.storage.sql.exec( `CREATE TABLE IF NOT EXISTS ${EXACT_REVIEW_QUEUE_METRIC_BUCKET_TABLE} ( bucket_start INTEGER PRIMARY KEY, review_enqueued INTEGER NOT NULL DEFAULT 0 CHECK (review_enqueued >= 0), review_completed INTEGER NOT NULL DEFAULT 0 CHECK (review_completed >= 0), + review_superseded INTEGER NOT NULL DEFAULT 0 CHECK (review_superseded >= 0), review_retried INTEGER NOT NULL DEFAULT 0 CHECK (review_retried >= 0), review_shed INTEGER NOT NULL DEFAULT 0 CHECK (review_shed >= 0), publication_enqueued INTEGER NOT NULL DEFAULT 0 CHECK (publication_enqueued >= 0), @@ -3466,6 +3707,7 @@ export class ExactReviewQueue { for (const column of [ "review_enqueued", "review_completed", + "review_superseded", "review_retried", "review_shed", "publication_semantic_deduped", @@ -3965,7 +4207,7 @@ export class ExactReviewQueue { private queueMetricTotalsSync(): ExactReviewQueueMetricTotals { const row = Array.from( this.storage.sql.exec( - `SELECT review_enqueued_total, review_completed_total, + `SELECT review_enqueued_total, review_completed_total, review_superseded_total, publication_enqueued_total, publication_completed_total, publication_published_total, publication_superseded_total, publication_semantic_deduped_total, @@ -3978,6 +4220,7 @@ export class ExactReviewQueue { | { review_enqueued_total?: number; review_completed_total?: number; + review_superseded_total?: number; publication_enqueued_total?: number; publication_completed_total?: number; publication_published_total?: number; @@ -3992,6 +4235,7 @@ export class ExactReviewQueue { review: { enqueued: exactReviewMetricTotal(row?.review_enqueued_total), completed: exactReviewMetricTotal(row?.review_completed_total), + superseded: exactReviewMetricTotal(row?.review_superseded_total), }, publication: { enqueued: exactReviewMetricTotal(row?.publication_enqueued_total), @@ -4009,6 +4253,7 @@ export class ExactReviewQueue { private incrementQueueMetricsSync(delta: ExactReviewQueueMetricDelta) { const reviewEnqueued = exactReviewMetricDelta(delta.reviewEnqueued); const reviewCompleted = exactReviewMetricDelta(delta.reviewCompleted); + const reviewSuperseded = exactReviewMetricDelta(delta.reviewSuperseded); const reviewRetried = exactReviewMetricDelta(delta.reviewRetried); const reviewShed = exactReviewMetricDelta(delta.reviewShed); const publicationEnqueued = exactReviewMetricDelta(delta.publicationEnqueued); @@ -4022,6 +4267,7 @@ export class ExactReviewQueue { if ( !reviewEnqueued && !reviewCompleted && + !reviewSuperseded && !reviewRetried && !reviewShed && !publicationEnqueued && @@ -4039,6 +4285,7 @@ export class ExactReviewQueue { `UPDATE ${EXACT_REVIEW_QUEUE_METRICS_TABLE} SET review_enqueued_total = review_enqueued_total + ?, review_completed_total = review_completed_total + ?, + review_superseded_total = review_superseded_total + ?, publication_enqueued_total = publication_enqueued_total + ?, publication_completed_total = publication_completed_total + ?, publication_published_total = publication_published_total + ?, @@ -4050,6 +4297,7 @@ export class ExactReviewQueue { WHERE singleton_id = 1`, reviewEnqueued, reviewCompleted, + reviewSuperseded, publicationEnqueued, publicationCompleted, publicationPublished, @@ -4062,6 +4310,7 @@ export class ExactReviewQueue { this.incrementQueueMetricBucketSync({ reviewEnqueued, reviewCompleted, + reviewSuperseded, reviewRetried, reviewShed, publicationEnqueued, @@ -4077,6 +4326,7 @@ export class ExactReviewQueue { private incrementQueueMetricBucketSync({ reviewEnqueued, reviewCompleted, + reviewSuperseded, reviewRetried, reviewShed, publicationEnqueued, @@ -4089,6 +4339,7 @@ export class ExactReviewQueue { }: { reviewEnqueued: number; reviewCompleted: number; + reviewSuperseded: number; reviewRetried: number; reviewShed: number; publicationEnqueued: number; @@ -4102,6 +4353,7 @@ export class ExactReviewQueue { if ( !reviewEnqueued && !reviewCompleted && + !reviewSuperseded && !reviewRetried && !reviewShed && !publicationEnqueued && @@ -4119,14 +4371,16 @@ export class ExactReviewQueue { EXACT_REVIEW_QUEUE_METRIC_BUCKET_MS; this.storage.sql.exec( `INSERT INTO ${EXACT_REVIEW_QUEUE_METRIC_BUCKET_TABLE} - (bucket_start, review_enqueued, review_completed, review_retried, review_shed, + (bucket_start, review_enqueued, review_completed, review_superseded, review_retried, + review_shed, publication_enqueued, publication_resolved, publication_published, publication_superseded, publication_semantic_deduped, publication_retried, publication_dead_lettered) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(bucket_start) DO UPDATE SET review_enqueued = review_enqueued + excluded.review_enqueued, review_completed = review_completed + excluded.review_completed, + review_superseded = review_superseded + excluded.review_superseded, review_retried = review_retried + excluded.review_retried, review_shed = review_shed + excluded.review_shed, publication_enqueued = publication_enqueued + excluded.publication_enqueued, @@ -4141,6 +4395,7 @@ export class ExactReviewQueue { bucketStart, reviewEnqueued, reviewCompleted, + reviewSuperseded, reviewRetried, reviewShed, publicationEnqueued, @@ -4203,6 +4458,21 @@ export class ExactReviewQueue { ); } + private insertSupersessionAuditSync(audit: ExactReviewSupersessionAudit) { + this.storage.sql.exec( + `INSERT OR IGNORE INTO ${EXACT_REVIEW_QUEUE_SUPERSESSION_TABLE} + (item_key, prior_revision, next_revision, superseded_run_id, + source_action, superseded_at) + VALUES (?, ?, ?, ?, ?, ?)`, + audit.itemKey, + audit.priorRevision, + audit.nextRevision, + audit.supersededRunId, + audit.sourceAction, + audit.supersededAt, + ); + } + private drainParkedDeadLettersSync(state: ExactReviewQueueState, now: number) { const openRow = Array.from( this.storage.sql.exec( @@ -4280,6 +4550,10 @@ export class ExactReviewQueue { WHERE status != 'open' AND resolved_at < ?`, now - EXACT_REVIEW_QUEUE_DEAD_LETTER_RESOLVED_TTL_MS, ); + this.storage.sql.exec( + `DELETE FROM ${EXACT_REVIEW_QUEUE_SUPERSESSION_TABLE} WHERE superseded_at < ?`, + now - EXACT_REVIEW_QUEUE_SUPERSESSION_RETENTION_MS, + ); this.storage.sql.exec( `DELETE FROM ${EXACT_REVIEW_STATE_WRITER_OPERATION_TABLE} WHERE observed_at < ?`, now - EXACT_REVIEW_STATE_WRITER_RETENTION_MS, @@ -4966,6 +5240,116 @@ export class ExactReviewQueue { this.storage.kv.delete(EXACT_REVIEW_QUEUE_STATE_KEY); } + private async processSourceAuthorityReservations(now: number) { + const reservations = (await this.sourceAuthorityReservations()) + .filter((reservation) => reservation.nextAttemptAt <= now) + .sort( + (left, right) => + left.nextAttemptAt - right.nextAttemptAt || + left.sourceAuthoritySeq - right.sourceAuthoritySeq, + ) + .slice(0, 8); + for (const reservation of reservations) { + try { + const liveHeadSha = await exactReviewSourceAuthorityLiveHead(this.env, reservation); + const reservedHeadSha = String(reservation.decision.sourceHeadSha || "").toLowerCase(); + if (liveHeadSha !== reservedHeadSha) { + this.completeSourceAuthorityReservationSync(reservation, "mismatch"); + continue; + } + const response = await this.fetch( + new Request("https://clawsweeper-exact-review-queue/enqueue", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + delivery_id: reservation.deliveryId, + decision: { + ...reservation.decision, + sourceHeadVerified: true, + }, + }), + }), + ); + if (!response.ok) throw new Error(`source authority enqueue failed: ${response.status}`); + this.completeSourceAuthorityReservationSync(reservation, "enqueued"); + } catch (error) { + console.warn( + "exact-review source authority verification deferred", + error instanceof Error ? error.message : String(error), + ); + this.deferSourceAuthorityReservationSync(reservation, Date.now()); + } + } + } + + private completeSourceAuthorityReservationSync( + expected: ExactReviewSourceAuthorityReservation, + disposition: "enqueued" | "mismatch", + ) { + this.storage.transactionSync(() => { + const key = exactReviewSourceAuthorityReservationKey(expected.deliveryId); + const current = exactReviewSourceAuthorityReservationFrom(this.storage.kv.get(key)); + if (current?.sourceAuthoritySeq === expected.sourceAuthoritySeq) { + if (disposition === "mismatch") { + this.storage.sql.exec( + `INSERT OR IGNORE INTO ${EXACT_REVIEW_QUEUE_DELIVERY_TABLE} + (delivery_id, received_at) VALUES (?, ?)`, + expected.deliveryId, + Date.now(), + ); + } + this.storage.kv.delete(key); + } + }); + } + + private deferSourceAuthorityReservationSync( + expected: ExactReviewSourceAuthorityReservation, + now: number, + ) { + this.storage.transactionSync(() => { + const key = exactReviewSourceAuthorityReservationKey(expected.deliveryId); + const current = exactReviewSourceAuthorityReservationFrom(this.storage.kv.get(key)); + if (current?.sourceAuthoritySeq !== expected.sourceAuthoritySeq) return; + const attempts = Math.min(EXACT_REVIEW_SOURCE_AUTHORITY_RETRY_LIMIT, current.attempts + 1); + const backoffMs = Math.min( + EXACT_REVIEW_SOURCE_AUTHORITY_RETRY_MAX_MS, + EXACT_REVIEW_SOURCE_AUTHORITY_RETRY_BASE_MS * 2 ** Math.max(0, attempts - 1), + ); + this.storage.kv.put(key, { + ...current, + attempts, + nextAttemptAt: now + backoffMs, + }); + }); + } + + private async sourceAuthorityReservations() { + const values = await this.storage.list({ + prefix: EXACT_REVIEW_SOURCE_AUTHORITY_RESERVATION_PREFIX, + }); + return Array.from(values.values()) + .map(exactReviewSourceAuthorityReservationFrom) + .filter( + (reservation): reservation is ExactReviewSourceAuthorityReservation => reservation !== null, + ); + } + + private async nextSourceAuthorityVerificationAt() { + return (await this.sourceAuthorityReservations()).reduce( + (next, reservation) => + next === null ? reservation.nextAttemptAt : Math.min(next, reservation.nextAttemptAt), + null, + ); + } + + private async scheduleSourceAuthorityVerification(nextAttemptAt: number) { + const scheduled = await this.storage.getAlarm(); + if (scheduled === null || scheduled <= Date.now() || nextAttemptAt < scheduled) { + await this.storage.setAlarm(nextAttemptAt); + } + } + private async scheduleNext(state: ExactReviewQueueState, now: number) { const publicationControl = this.refreshPublicationControlSync(state, now); const batchOwnership = this.batchStore.activeLeaseSnapshot(now); @@ -5004,11 +5388,13 @@ export class ExactReviewQueue { now, new Set(batchOwnership.itemKeys), ); + const sourceAuthorityNext = await this.nextSourceAuthorityVerificationAt(); const next = [ queueNext, reviewNext, batchOwnership.nextLeaseExpiresAt, batchDeparture?.dueAt ?? null, + sourceAuthorityNext, ] .filter((candidate): candidate is number => candidate !== null) .reduce( @@ -5052,6 +5438,58 @@ function exactReviewDecisionFrom(value): ExactReviewDecision | null { return { ...base, ...(publication ? { publication } : {}) }; } +function exactReviewDecisionWithoutSourceAuthority(decision: ExactReviewDecision) { + const { + sourceAuthoritySeq: _sourceAuthoritySeq, + sourceHeadVerified: _sourceHeadVerified, + ...rest + } = decision; + return rest; +} + +function exactReviewSourceAuthorityReservationKey(deliveryId: string) { + return `${EXACT_REVIEW_SOURCE_AUTHORITY_RESERVATION_PREFIX}${encodeURIComponent(deliveryId)}`; +} + +function exactReviewSourceAuthorityReservationFrom( + value, +): ExactReviewSourceAuthorityReservation | null { + const reservation = objectValue(value); + const deliveryId = String(reservation.deliveryId || "").trim(); + const decision = exactReviewDecisionFrom(reservation.decision); + const installationId = Number(reservation.installationId); + const sourceAuthoritySeq = Number(reservation.sourceAuthoritySeq); + const attempts = Number(reservation.attempts); + const nextAttemptAt = Number(reservation.nextAttemptAt); + if ( + !deliveryId || + deliveryId.length > 200 || + !decision || + decision.itemKind !== "pull_request" || + decision.publication || + !Number.isInteger(installationId) || + installationId <= 0 || + !Number.isSafeInteger(sourceAuthoritySeq) || + sourceAuthoritySeq <= 0 || + decision.sourceAuthoritySeq !== sourceAuthoritySeq || + !Number.isInteger(attempts) || + attempts < 0 || + attempts > EXACT_REVIEW_SOURCE_AUTHORITY_RETRY_LIMIT || + !Number.isSafeInteger(nextAttemptAt) || + nextAttemptAt < 0 + ) { + return null; + } + return { + deliveryId, + decision, + installationId, + sourceAuthoritySeq, + attempts, + nextAttemptAt, + }; +} + function exactReviewBaseDecisionFrom(value): ExactReviewBaseDecision | null { const decision = objectValue(value); const targetRepo = String(decision.targetRepo || "").trim(); @@ -5060,6 +5498,21 @@ function exactReviewBaseDecisionFrom(value): ExactReviewBaseDecision | null { const itemKind = String(decision.itemKind || ""); const sourceEvent = String(decision.sourceEvent || ""); const sourceAction = String(decision.sourceAction || ""); + const hasSourceHeadSha = Object.hasOwn(decision, "sourceHeadSha"); + const sourceHeadSha = hasSourceHeadSha + ? String(decision.sourceHeadSha || "") + .trim() + .toLowerCase() + : undefined; + const hasSourceHeadVerified = Object.hasOwn(decision, "sourceHeadVerified"); + const hasSourceAuthoritySeq = Object.hasOwn(decision, "sourceAuthoritySeq"); + const sourceAuthoritySeq = hasSourceAuthoritySeq + ? Number(decision.sourceAuthoritySeq) + : undefined; + const hasSourceUpdatedAt = Object.hasOwn(decision, "sourceUpdatedAt"); + const sourceUpdatedAt = hasSourceUpdatedAt + ? String(decision.sourceUpdatedAt || "").trim() + : undefined; const hasCommandStatusMarker = Object.hasOwn(decision, "commandStatusMarker"); const commandStatusMarker = hasCommandStatusMarker ? decision.commandStatusMarker : undefined; const hasStatusCommentId = Object.hasOwn(decision, "statusCommentId"); @@ -5072,6 +5525,15 @@ function exactReviewBaseDecisionFrom(value): ExactReviewBaseDecision | null { if (itemKind !== "issue" && itemKind !== "pull_request") return null; if (sourceEvent !== "issues" && sourceEvent !== "pull_request") return null; if (!sourceAction) return null; + if (hasSourceHeadSha && !/^[0-9a-f]{40}$/.test(sourceHeadSha || "")) return null; + if (hasSourceHeadVerified && typeof decision.sourceHeadVerified !== "boolean") return null; + if ( + hasSourceAuthoritySeq && + (!Number.isSafeInteger(sourceAuthoritySeq) || Number(sourceAuthoritySeq) <= 0) + ) { + return null; + } + if (hasSourceUpdatedAt && !Number.isFinite(Date.parse(sourceUpdatedAt || ""))) return null; if ( hasCommandStatusMarker && (typeof commandStatusMarker !== "string" || @@ -5101,6 +5563,10 @@ function exactReviewBaseDecisionFrom(value): ExactReviewBaseDecision | null { sourceEvent, sourceAction, supersedesInProgress: Boolean(decision.supersedesInProgress), + ...(hasSourceHeadSha ? { sourceHeadSha } : {}), + ...(hasSourceHeadVerified ? { sourceHeadVerified: decision.sourceHeadVerified } : {}), + ...(hasSourceAuthoritySeq ? { sourceAuthoritySeq } : {}), + ...(hasSourceUpdatedAt ? { sourceUpdatedAt } : {}), ...(Number.isFinite(Number(decision.codexTimeoutMs)) ? { codexTimeoutMs: Number(decision.codexTimeoutMs) } : {}), @@ -5231,6 +5697,51 @@ function mergePendingExactReviewDecision( return merged; } +function exactReviewDecisionCanSupersedeReview( + current: ExactReviewQueueItem, + incoming: ExactReviewDecision, +): boolean { + const active = current.leaseDecision || current.decision; + if (active.itemKind !== "pull_request" || incoming.itemKind !== "pull_request") return true; + + const activeHead = String(active.sourceHeadSha || "").toLowerCase(); + const incomingHead = String(incoming.sourceHeadSha || "").toLowerCase(); + if (!/^[0-9a-f]{40}$/.test(incomingHead)) return false; + const incomingAuthoritySeq = Number(incoming.sourceAuthoritySeq || 0); + const activeSourceAuthoritySeq = Number(active.sourceAuthoritySeq || 0); + const activeHasAuthority = + Number.isSafeInteger(activeSourceAuthoritySeq) && activeSourceAuthoritySeq > 0; + if (!/^[0-9a-f]{40}$/.test(activeHead)) { + return ( + incoming.sourceHeadVerified === true && + Number.isSafeInteger(incomingAuthoritySeq) && + incomingAuthoritySeq > 0 + ); + } + if (incomingHead !== activeHead && incoming.sourceHeadVerified !== true) { + return false; + } + if (!activeHasAuthority) { + if (incomingHead !== activeHead) { + return incoming.sourceHeadVerified === true; + } + return Number.isSafeInteger(incomingAuthoritySeq) && incomingAuthoritySeq > 0; + } + if (!Number.isSafeInteger(incomingAuthoritySeq) || incomingAuthoritySeq <= 0) return false; + + const activeUpdatedAt = Date.parse(String(active.sourceUpdatedAt || "")); + const incomingUpdatedAt = Date.parse(String(incoming.sourceUpdatedAt || "")); + if ( + Number.isFinite(activeUpdatedAt) && + Number.isFinite(incomingUpdatedAt) && + incomingUpdatedAt !== activeUpdatedAt + ) { + return incomingUpdatedAt > activeUpdatedAt; + } + + return incomingAuthoritySeq > activeSourceAuthoritySeq; +} + function exactReviewItemKey(decision: ExactReviewDecision) { const base = `${decision.targetRepo}#${decision.itemNumber}`; return decision.publication @@ -7114,8 +7625,8 @@ function exactReviewExecutionLeaseMs(env) { } function exactReviewHeartbeatGraceMs(env) { - // Floor must stay above the 5-minute worker heartbeat interval (plus request time and - // scheduling jitter) or a configured grace would reclaim healthy leases between beats. + // Keep a conservative floor above the one-minute worker heartbeat so scheduler + // or network stalls cannot reclaim a healthy lease between beats. return Math.max( 420_000, numberFrom(env.EXACT_REVIEW_HEARTBEAT_GRACE_MS, DEFAULT_EXACT_REVIEW_HEARTBEAT_GRACE_MS), @@ -7373,6 +7884,32 @@ async function exactReviewDispatchToken(env) { return exactReviewRepositoryToken(env, { actions: "write", contents: "write" }); } +async function exactReviewSourceAuthorityLiveHead( + env, + reservation: ExactReviewSourceAuthorityReservation, +) { + const credentials = githubAppCredentials(env); + if (!credentials) throw new Error("github app is not configured"); + const appJwt = await signGithubAppJwt(credentials.issuer, credentials.privateKey); + const token = await createGithubAppTokenFor({ + appJwt, + installationId: reservation.installationId, + label: reservation.decision.targetRepo, + repositories: [repoName(reservation.decision.targetRepo)], + permissions: { pull_requests: "read" }, + }); + const pull = await githubTokenJson({ + token, + path: `/repos/${reservation.decision.targetRepo}/pulls/${reservation.decision.itemNumber}`, + method: "GET", + body: undefined, + errorLabel: "live pull request head", + }); + return String(objectValue(objectValue(pull).head).sha || "") + .trim() + .toLowerCase(); +} + export async function exactReviewActionsReadToken(env) { return exactReviewRepositoryToken(env, { actions: "read" }); } @@ -7579,6 +8116,7 @@ async function dispatchClawsweeperItem({ source_event: decision.sourceEvent, source_action: decision.sourceAction, supersedes_in_progress: decision.supersedesInProgress, + ...(decision.sourceHeadSha ? { source_head_sha: decision.sourceHeadSha } : {}), ...(Object.keys(reviewOptions).length > 0 ? { review_options: reviewOptions } : {}), }, }, @@ -7623,7 +8161,8 @@ async function githubTokenJson({ token, path, method = "GET", body, errorLabel } ); } if (response.status === 204) return {}; - return response.json(); + const text = await response.text(); + return text ? JSON.parse(text) : {}; } function exactReviewPublicationBatchId(value) { diff --git a/dashboard/worker.ts b/dashboard/worker.ts index cf5509aa00..17a891be28 100644 --- a/dashboard/worker.ts +++ b/dashboard/worker.ts @@ -1010,12 +1010,69 @@ async function githubWebhook(request, env, ctx) { if ("type" in decision && decision.type === "item") { const deliveryId = request.headers.get("x-github-delivery") || ""; + const itemDecision = decision as ExactReviewDecision & { installationId?: number }; + const sourceAuthority = + itemDecision.itemKind === "pull_request" + ? await reserveExactReviewSourceAuthority(env, { + deliveryId, + decision: itemDecision, + }) + : null; + if (itemDecision.itemKind === "pull_request" && sourceAuthority === null) { + return json({ error: "exact_review_queue_not_configured" }, 503); + } + if (sourceAuthority && "deduped" in sourceAuthority) { + return json({ + ok: true, + deduped: true, + item_key: `${itemDecision.targetRepo}#${itemDecision.itemNumber}`, + }); + } + const sourceAuthoritySeq = + sourceAuthority && "sourceAuthoritySeq" in sourceAuthority + ? sourceAuthority.sourceAuthoritySeq + : null; + let exactReviewDecision: ExactReviewDecision | null; + try { + exactReviewDecision = await bindLivePullRequestHeadAuthority({ + env, + decision: itemDecision, + sourceAuthoritySeq, + }); + } catch { + return json( + { + ok: true, + accepted: true, + deferred: true, + reason: "pull request head verification deferred", + }, + 202, + ); + } + if (!exactReviewDecision) { + await completeExactReviewSourceAuthority( + env, + deliveryId, + Number(sourceAuthoritySeq), + "mismatch", + ).catch(() => undefined); + return json({ ok: true, accepted: false, reason: "stale pull request head" }, 202); + } const queued = await enqueueExactReview({ env, deliveryId, - decision: decision as ExactReviewDecision, + decision: exactReviewDecision, }); if (!queued) return json({ error: "exact_review_queue_not_configured" }, 503); + if (sourceAuthoritySeq !== null) { + await completeExactReviewSourceAuthority( + env, + deliveryId, + sourceAuthoritySeq, + "enqueued", + ).catch(() => undefined); + } return json({ ok: true, ...queued }, 202); } @@ -1261,10 +1318,15 @@ function classifyGithubItemWebhook({ event, payload }) { if (action === "unlabeled" && !isCloseGuardLabel(payload.label)) { return { accepted: false, reason: "unsupported action" }; } - const itemNumber = Number(objectValue(payload.pull_request).number); + const pullRequest = objectValue(payload.pull_request); + const itemNumber = Number(pullRequest.number); if (!Number.isInteger(itemNumber) || itemNumber <= 0) { return { accepted: false, reason: "missing pull request number" }; } + const sourceHeadSha = String(objectValue(pullRequest.head).sha || "") + .trim() + .toLowerCase(); + const sourceUpdatedAt = exactWebhookTimestamp(pullRequest.updated_at); return { accepted: true, type: "item", @@ -1275,6 +1337,8 @@ function classifyGithubItemWebhook({ event, payload }) { installationId, sourceEvent: "pull_request", sourceAction: action, + ...(/^[0-9a-f]{40}$/.test(sourceHeadSha) ? { sourceHeadSha } : {}), + ...(sourceUpdatedAt ? { sourceUpdatedAt } : {}), supersedesInProgress: [ "edited", "synchronize", @@ -1288,6 +1352,55 @@ function classifyGithubItemWebhook({ event, payload }) { return { accepted: false, reason: "unsupported event" }; } +async function bindLivePullRequestHeadAuthority({ + env, + decision, + sourceAuthoritySeq, +}: { + env: DashboardEnv; + decision: ExactReviewDecision & { installationId?: number }; + sourceAuthoritySeq: number | null; +}): Promise { + if (decision.itemKind !== "pull_request") return decision; + if (!Number.isSafeInteger(sourceAuthoritySeq) || Number(sourceAuthoritySeq) <= 0) { + throw new Error("exact-review source authority unavailable"); + } + const sourceHeadSha = String(decision.sourceHeadSha || "").toLowerCase(); + if (!/^[0-9a-f]{40}$/.test(sourceHeadSha)) return null; + + let token = stringEnv(env.GITHUB_TOKEN); + if (!token) { + const credentials = githubAppCredentials(env); + if ( + !credentials || + !Number.isInteger(decision.installationId) || + decision.installationId! <= 0 + ) { + throw new Error("GitHub App credentials are required to verify a pull request head"); + } + const appJwt = await signGithubAppJwt(credentials.issuer, credentials.privateKey); + token = await createGithubAppTokenFor({ + appJwt, + installationId: decision.installationId!, + label: decision.targetRepo, + repositories: [repoName(decision.targetRepo)], + permissions: { pull_requests: "read" }, + }); + } + const pull = await githubTokenJson({ + token, + path: `/repos/${decision.targetRepo}/pulls/${decision.itemNumber}`, + body: undefined, + errorLabel: "live pull request head", + }); + const liveHeadSha = String(objectValue(objectValue(pull).head).sha || "") + .trim() + .toLowerCase(); + return liveHeadSha === sourceHeadSha + ? { ...decision, sourceHeadVerified: true, sourceAuthoritySeq: Number(sourceAuthoritySeq) } + : null; +} + function isCloseGuardLabel(value) { const label = String(objectValue(value).name || "") .trim() @@ -1339,6 +1452,66 @@ function exactReviewQueueStub(env): DurableObjectStub | null { return namespace ? namespace.get(namespace.idFromName(EXACT_REVIEW_QUEUE_NAME)) : null; } +async function reserveExactReviewSourceAuthority( + env, + { + deliveryId, + decision, + }: { + deliveryId: string; + decision: ExactReviewDecision & { installationId?: number }; + }, +): Promise<{ deduped: true } | { sourceAuthoritySeq: number } | null> { + const queue = exactReviewQueueStub(env); + if (!queue) return null; + const response = await queue.fetch( + new Request("https://clawsweeper-exact-review-queue/source-authority", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + delivery_id: deliveryId, + decision, + installation_id: decision.installationId, + }), + }), + ); + const body = objectValue(await response.json().catch(() => null)); + if (!response.ok) { + throw new Error(String(body.error || "exact-review source authority unavailable")); + } + if (body.deduped === true) return { deduped: true as const }; + const sourceAuthoritySeq = Number(body.source_authority_seq); + if (!Number.isSafeInteger(sourceAuthoritySeq) || sourceAuthoritySeq <= 0) { + throw new Error("exact-review source authority unavailable"); + } + return { sourceAuthoritySeq }; +} + +async function completeExactReviewSourceAuthority( + env, + deliveryId: string, + sourceAuthoritySeq: number, + disposition: "enqueued" | "mismatch", +) { + const queue = exactReviewQueueStub(env); + if (!queue) throw new Error("exact-review queue not configured"); + const response = await queue.fetch( + new Request("https://clawsweeper-exact-review-queue/source-authority/complete", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + delivery_id: deliveryId, + source_authority_seq: sourceAuthoritySeq, + disposition, + }), + }), + ); + if (!response.ok) { + const body = objectValue(await response.json().catch(() => null)); + throw new Error(String(body.error || "exact-review source authority completion failed")); + } +} + async function exactReviewQueueRequest(env, path, request?: Request) { const queue = exactReviewQueueStub(env); if (!queue) return json({ error: "exact_review_queue_not_configured" }, 503); diff --git a/dashboard/wrangler.toml b/dashboard/wrangler.toml index c87aae1c12..b13d72b196 100644 --- a/dashboard/wrangler.toml +++ b/dashboard/wrangler.toml @@ -53,7 +53,7 @@ EXACT_REVIEW_DISPATCH_LEASE_MS = "360000" EXACT_REVIEW_EXECUTION_LEASE_MS = "7800000" EXACT_REVIEW_HEARTBEAT_GRACE_MS = "1200000" EXACT_REVIEW_WORKFLOW_PAUSED_RETRY_MS = "60000" -EXACT_REVIEW_DISPATCH_DEBOUNCE_MS = "45000" +EXACT_REVIEW_DISPATCH_DEBOUNCE_MS = "90000" EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS = "180000" EXACT_REVIEW_PENDING_SOFT_LIMIT = "300" # Keep bounded four-worker preparation, but return the live mutation lease to 8: diff --git a/docs/limits.md b/docs/limits.md index 939ffc80e9..f745078dee 100644 --- a/docs/limits.md +++ b/docs/limits.md @@ -25,7 +25,7 @@ budget: | `CLAWSWEEPER_BULK_FILER_THRESHOLD` | 10 | Recent authored-issue count that marks bulk filing. | | `CLAWSWEEPER_BULK_FILER_WINDOW_DAYS` | 7 | Lookback window for authored issue filing rate. | | `CLAWSWEEPER_STALE_VERSION_BUG_CLOSE_ENABLED` | `false` | Enables stale-version bug closes after 120 days. | -| `CLAWSWEEPER_OBSOLETE_FIX_PR_CLOSE_ENABLED` | `false` | Enables obsolete small-fix PR closes after 90 days. | +| `CLAWSWEEPER_OBSOLETE_FIX_PR_CLOSE_ENABLED` | `false` | Enables obsolete small-fix PR closes after 90 days. | See [`author-pr-budget-close-policy.md`](author-pr-budget-close-policy.md) for the rating, proof, inactivity, engagement, and fail-closed gates. @@ -176,11 +176,23 @@ backlog drain. Exact capacity is consumed only while queue work is pending. As those priority workers start, normal, hot-intake, and commit-review planners count them and reduce their next background wave. -Fresh webhook work waits for `EXACT_REVIEW_DISPATCH_DEBOUNCE_MS` (45 seconds by +Fresh webhook work waits for `EXACT_REVIEW_DISPATCH_DEBOUNCE_MS` (90 seconds by default) so rapid edits and pushes coalesce before dispatch. Repeated pending revisions extend that delay up to `EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS` (three -minutes by default) from the item's first enqueue. Explicit command work and -publication work bypass the delay. When pending depth reaches +minutes by default) from the item's first enqueue. A superseding source event +immediately revokes the old queue lease and starts a fresh debounce window for +the latest revision. The old workflow's tuple heartbeat then returns `409`; its +review step terminates the local review process instead of using GitHub's +run-id-only cancellation API, which cannot safely distinguish a later rerun +attempt. The replacement is durably scheduled before the old worker observes +the revocation, and +`review_superseded_total` records the terminalized review generation. Recovery +events never supersede an existing item; only a fresh source revision can revoke +an active review. Duplicate workflow deliveries remain non-cancelling and rely +on the lease claim tuple, so they cannot terminate the sole valid owner. An older +unclaimed workflow cannot pass the replacement lease tuple and exits before +review compute. Explicit command work and publication work bypass the delay. +When pending depth reaches `EXACT_REVIEW_PENDING_SOFT_LIMIT` (300 by default), new recovery-only work is shed; existing items, webhook events, commands, and publications remain admitted. @@ -200,7 +212,10 @@ Codex and does not deduct these control-plane workflows from `workers.max`. Each dispatched workflow claims its opaque lease before checkout. Protocol v2 binds claim and completion to the item key, lease revision, run attempt, claim -generation, and an immutable decision snapshot. During the rolling-upgrade +generation, source head (for pull requests), and an immutable decision snapshot. +For pull requests, the review-start reservation rejects a claimed source head +that no longer matches the live head, then heartbeats that full queue tuple +before deleting any different-revision placeholder. During the rolling-upgrade window, dispatches nest the strict tuple under `queue_claim`, also carry the immutable v1 snapshot, and the Worker accepts lease-id-only finalization only for claims recorded as protocol v1. Duplicate dispatches and stale workflows cannot claim the same lease, and a completion @@ -219,7 +234,7 @@ workflows holding the expired lease cannot claim it. Run-attempt binding and a per-claim generation check keep delayed terminal decisions from releasing a later rerun; queued and in-progress runs are never released. If a workflow never claims or completes, the Durable Object reclaims -the expired lease. Claimed review workers heartbeat every five minutes; after the first +the expired lease. Claimed review workers heartbeat every minute; after the first heartbeat, `EXACT_REVIEW_HEARTBEAT_GRACE_MS` bounds liveness to 20 minutes by default while never extending the original 130-minute execution lease. Leases created before heartbeat support was deployed retain their original execution expiry. This keeps capacity waiting and @@ -287,9 +302,8 @@ hot intake `14`, and commit review `2`. Existing repair lanes keep their review-start placeholder must be before recovery; the default is 2 hours and the maximum is 720 hours. - `REVIEW_PLACEHOLDER_MAX_RECOVERIES` overrides the number of orphaned review - placeholders enqueued per recovery pass; the default is 5 and the maximum is - 100. -- `EXACT_REVIEW_DISPATCH_DEBOUNCE_MS` overrides the 45,000 ms coalescing delay + placeholders enqueued per recovery pass; the default is 5 and the maximum is 100. +- `EXACT_REVIEW_DISPATCH_DEBOUNCE_MS` overrides the 90,000 ms coalescing delay for fresh non-command exact-review events. - `EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS` overrides the 180,000 ms maximum coalescing window measured from the item's first enqueue. @@ -297,7 +311,7 @@ hot intake `14`, and commit review `2`. Existing repair lanes keep their shedding new recovery-only exact-review work; the default is 300. - `EXACT_REVIEW_HEARTBEAT_GRACE_MS` overrides the 1,200,000 ms exact-review worker heartbeat grace. It is clamped to at least 420,000 ms so a configured grace can never dip - below the five-minute worker heartbeat interval plus request time and jitter. + near the one-minute worker heartbeat interval during scheduler or network stalls. - `CLAWSWEEPER_COMMIT_REVIEW_PAGE_SIZE` overrides `commit_review.page_size_default`. - `CLAWSWEEPER_FEATURE_CLUSTER_REPAIR_ENABLED=1` enables the scheduled diff --git a/docs/pr-review-comments.md b/docs/pr-review-comments.md index f14332b01d..8ea8d168da 100644 --- a/docs/pr-review-comments.md +++ b/docs/pr-review-comments.md @@ -32,12 +32,28 @@ 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. +After a newer source revision wins its lease, ClawSweeper may delete dedicated +review-start placeholders for older revisions. The candidate comment snapshot +is captured first, then the worker must still own the exact queue +item/lease/revision/generation/run tuple and the live item revision must match +its lease. For pull requests, the claimed queue source head must also match the +live head. A stale worker therefore cannot treat a newer lease as superseded +just because the SHAs differ. Same-revision contenders still use the +server-assigned comment-id election, and expired leftovers retain the existing +conservative cleanup path. + For a PR that needs work, the visible comment starts with: ```text Codex review: needs changes before merge. ``` +The visible `Summary` also includes `Reviewed head: `. This makes the +human-facing verdict self-identifying without requiring maintainers to inspect +hidden markers. Publication still verifies the durable tuple against live state; +the visible SHA is evidence of the captured review revision, not a substitute +for that guard. + For an external PR that lacks after-fix real behavior proof, the visible comment starts with: diff --git a/src/clawsweeper.ts b/src/clawsweeper.ts index 8ca45e1dab..ec511bf653 100644 --- a/src/clawsweeper.ts +++ b/src/clawsweeper.ts @@ -127,6 +127,7 @@ import { import { expiredReviewStartStatusLeases, freshExactHeadReviewStartLease, + supersededReviewStartStatusLeases, } from "./repair/comment-router-core.js"; import { AUTOMERGE_LABEL, @@ -517,6 +518,17 @@ type AcquiredReviewStartLease = { comment?: Record; }; +type ExactReviewQueueAuthority = { + queueUrl: string; + itemKey: string; + leaseId: string; + leaseRevision: number; + claimGeneration: number; + runId: string; + runAttempt: number; + sourceHeadSha: string | null; +}; + type ReviewStartStatusCommentResult = | { status: "posted"; lease: AcquiredReviewStartLease; didMutate: true } | { status: "held"; lease: null; retryAt: string; didMutate: boolean }; @@ -18539,10 +18551,12 @@ function publicPrSummaryBody( summaryLine: string, reproductionAssessment: string, prSurfaceSummary: string, + reviewedHeadSha: string, ): string { return [ summaryLine, prSurfaceSummary ? `PR surface: ${prSurfaceSummary}` : "", + reviewedHeadSha ? `Reviewed head: \`${reviewedHeadSha}\`` : "", publicReproducibilityLine(reproductionAssessment), ] .filter(Boolean) @@ -18816,7 +18830,12 @@ function renderKeepOpenCommentFromReport( appendPublicSection( lines, "Summary", - publicPrSummaryBody(changeSummaryLine, reproductionAssessment, prSurfaceSummary), + publicPrSummaryBody( + changeSummaryLine, + reproductionAssessment, + prSurfaceSummary, + pullHeadShaFromReport(markdown) ?? "", + ), ); lines.push(renderReviewMetricsDigest(reviewMetrics), ""); const dataModelWarning = renderDataModelWarningFromReport(markdown); @@ -20599,6 +20618,101 @@ export function newReviewStartLeaseOwnerForTest( return newReviewStartLeaseOwner(env, fallback); } +function exactReviewQueueAuthorityFromEnv( + env: NodeJS.ProcessEnv = process.env, +): ExactReviewQueueAuthority | null { + const raw = { + queueUrl: String(env.EXACT_REVIEW_QUEUE_URL ?? "") + .trim() + .replace(/\/$/, ""), + itemKey: String(env.EXACT_REVIEW_ITEM_KEY ?? "").trim(), + leaseId: String(env.EXACT_REVIEW_LEASE_ID ?? "").trim(), + leaseRevision: String(env.EXACT_REVIEW_LEASE_REVISION ?? "").trim(), + claimGeneration: String(env.EXACT_REVIEW_CLAIM_GENERATION ?? "").trim(), + runId: String(env.GITHUB_RUN_ID ?? "").trim(), + runAttempt: String(env.GITHUB_RUN_ATTEMPT ?? "").trim(), + sourceHeadSha: String(env.EXACT_REVIEW_SOURCE_HEAD_SHA ?? "") + .trim() + .toLowerCase(), + }; + if ( + ![raw.queueUrl, raw.itemKey, raw.leaseId, raw.leaseRevision, raw.claimGeneration].some(Boolean) + ) { + return null; + } + + let queueUrl: URL; + try { + queueUrl = new URL(raw.queueUrl); + } catch { + throw new UserFacingCommandError("EXACT_REVIEW_QUEUE_URL must be an HTTP(S) URL."); + } + const leaseRevision = Number(raw.leaseRevision); + const claimGeneration = Number(raw.claimGeneration); + const runAttempt = Number(raw.runAttempt); + if ( + !["http:", "https:"].includes(queueUrl.protocol) || + !/^[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+#[1-9]\d*$/.test(raw.itemKey) || + !/^[A-Za-z0-9._:-]{1,200}$/.test(raw.leaseId) || + !Number.isSafeInteger(leaseRevision) || + leaseRevision < 1 || + !Number.isSafeInteger(claimGeneration) || + claimGeneration < 1 || + !/^[1-9]\d{0,29}$/.test(raw.runId) || + !Number.isSafeInteger(runAttempt) || + runAttempt < 1 || + (raw.sourceHeadSha !== "" && !/^[0-9a-f]{40}$/.test(raw.sourceHeadSha)) + ) { + throw new UserFacingCommandError("Exact-review queue authority context is incomplete."); + } + return { + queueUrl: queueUrl.toString().replace(/\/$/, ""), + itemKey: raw.itemKey, + leaseId: raw.leaseId, + leaseRevision, + claimGeneration, + runId: raw.runId, + runAttempt, + sourceHeadSha: raw.sourceHeadSha || null, + }; +} + +function exactReviewQueueAuthorityIsLive(authority: ExactReviewQueueAuthority): boolean { + const payload = JSON.stringify({ + item_key: authority.itemKey, + lease_id: authority.leaseId, + lease_revision: authority.leaseRevision, + claim_generation: authority.claimGeneration, + run_id: authority.runId, + run_attempt: authority.runAttempt, + ...(authority.sourceHeadSha ? { source_head_sha: authority.sourceHeadSha } : {}), + }); + const result = spawnSync( + "curl", + [ + "--silent", + "--show-error", + "--connect-timeout", + "5", + "--max-time", + "20", + "--output", + "/dev/null", + "--write-out", + "%{http_code}", + "--request", + "POST", + "--header", + "content-type: application/json", + "--data-binary", + payload, + `${authority.queueUrl}/internal/exact-review/heartbeat`, + ], + { encoding: "utf8" }, + ); + return result.status === 0 && result.stdout.trim() === "200"; +} + function freshDedicatedReviewStartLeases(options: { comments: Record[]; itemNumber: number; @@ -20656,6 +20770,8 @@ function postReviewStartStatusComment(options: { shardIndex: number; shardCount: number; purpose?: "review" | "apply"; + queueAuthority?: ExactReviewQueueAuthority | null; + allowSupersededLeaseCleanup?: boolean; }): ReviewStartStatusCommentResult { const startedAtMs = Date.now(); const leaseOwner = newReviewStartLeaseOwner(); @@ -20725,8 +20841,9 @@ function postReviewStartStatusComment(options: { ); } const acquired = { owner: leaseOwner, commentId: createdCommentId, headSha: normalizedHead }; + const confirmedState = issueReviewCommentState(options.item.number); const confirmed = freshDedicatedReviewStartLeases({ - comments: issueReviewCommentState(options.item.number).leaseComments, + comments: confirmedState.leaseComments, itemNumber: options.item.number, headSha: normalizedHead, nowMs: Date.now(), @@ -20751,6 +20868,38 @@ function postReviewStartStatusComment(options: { deleteOwnedDedicatedReviewStartLease(options.item.number, acquired); return heldReviewStartStatusCommentResult(winner.expiresAt, true); } + if (options.queueAuthority) { + const authoritativeHead = currentReviewRevision(options.item); + if (authoritativeHead !== normalizedHead) { + deleteOwnedDedicatedReviewStartLease(options.item.number, acquired); + throw new Error( + `review revision changed while reserving #${options.item.number}; retry required`, + ); + } + if (!exactReviewQueueAuthorityIsLive(options.queueAuthority)) { + deleteOwnedDedicatedReviewStartLease(options.item.number, acquired); + throw new Error( + `exact-review queue authority changed while reserving #${options.item.number}; retry required`, + ); + } + // The candidate snapshot predates both authority checks. A newer worker + // cannot be selected by a stale caller: if its lease is already present, + // the live revision/queue tuple has moved; if it starts later, it is absent + // from this immutable snapshot. + if (options.allowSupersededLeaseCleanup) { + reapSupersededDedicatedReviewStartLeases( + options.item.number, + confirmedState.dedicatedLeaseComments, + normalizedHead, + authoritativeHead, + ); + } else if (pullRequestHeadSha(options.item.number) !== normalizedHead) { + deleteOwnedDedicatedReviewStartLease(options.item.number, acquired); + throw new Error( + `review revision changed while reserving #${options.item.number}; retry required`, + ); + } + } return { status: "posted", lease: { ...acquired, comment: winner.comment }, @@ -20830,6 +20979,46 @@ function reapExpiredDedicatedReviewStartLeases( } } +function reapSupersededDedicatedReviewStartLeases( + itemNumber: number, + dedicatedLeaseComments: Record[], + currentHeadSha: string, + authoritativeHeadSha: string, +): void { + const superseded = supersededReviewStartStatusLeases({ + comments: dedicatedLeaseComments, + itemNumber, + headSha: currentHeadSha, + authoritativeHeadSha, + trustedAuthors: new Set( + [...PATCHABLE_REVIEW_COMMENT_AUTHORS].map((author) => author.toLowerCase()), + ), + }); + for (const lease of superseded) { + try { + ghObservedMutationCommand({ + identity: `review_lease_supersede:${itemNumber}:${lease.commentId}`, + args: [ + "api", + `repos/${targetRepo()}/issues/comments/${lease.commentId}`, + "--method", + "DELETE", + ], + }); + console.error( + `[review] deleted superseded review lease comment ${lease.commentId} for #${itemNumber} (reviewed head ${lease.headSha}, current head ${currentHeadSha})`, + ); + } catch (error) { + // A failed cleanup must never block the current revision from acquiring its lease. + console.error( + `[review] could not delete superseded review lease comment ${lease.commentId} for #${itemNumber}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } +} + const REVIEW_PLACEHOLDER_BODY_PATTERN = /^ClawSweeper status: review started\./i; export function supersededReviewPlaceholderCommentIds(options: { @@ -20901,6 +21090,12 @@ function pullRequestHeadSha(number: number): string { return typeof sha === "string" ? sha.trim().toLowerCase() : ""; } +function currentReviewRevision(item: Item): string { + if (item.kind === "pull_request") return pullRequestHeadSha(item.number); + const revision = collectItemContext(item, { fullTimelineForRelations: true }).sourceRevision; + return typeof revision === "string" ? revision : ""; +} + function closeItem(options: { number: number; kind: ItemKind; reason: CloseReason }): void { if (options.kind === "pull_request") { ghObservedMutationCommand({ @@ -22634,15 +22829,30 @@ function reserveReviewLeaseCommand(args: Args): void { `Cannot reserve a review lease for #${itemNumber}: state is ${state}.`, ); } - const currentRevision = - item.kind === "pull_request" - ? pullRequestHeadSha(itemNumber) - : collectItemContext(item, { fullTimelineForRelations: true }).sourceRevision; + const queueAuthority = exactReviewQueueAuthorityFromEnv(); + const expectedItemKey = `${targetRepo()}#${itemNumber}`.toLowerCase(); + if (queueAuthority && queueAuthority.itemKey.toLowerCase() !== expectedItemKey) { + throw new UserFacingCommandError("Exact-review queue authority item does not match target."); + } + const currentRevision = currentReviewRevision(item); if (!currentRevision || !/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/.test(currentRevision)) { throw new UserFacingCommandError( `Could not resolve the current review revision for #${itemNumber}.`, ); } + if ( + queueAuthority?.sourceHeadSha && + item.kind === "pull_request" && + queueAuthority.sourceHeadSha !== currentRevision + ) { + throw new UserFacingCommandError( + "Exact-review queue authority source head does not match the current pull request.", + ); + } + const reservationAuthority = + queueAuthority && item.kind === "pull_request" && !queueAuthority.sourceHeadSha + ? { ...queueAuthority, sourceHeadSha: currentRevision } + : queueAuthority; const result = postReviewStartStatusComment({ item, headSha: currentRevision, @@ -22651,6 +22861,9 @@ function reserveReviewLeaseCommand(args: Args): void { total: 1, shardIndex: 0, shardCount: 1, + queueAuthority: reservationAuthority, + allowSupersededLeaseCleanup: + item.kind !== "pull_request" || Boolean(queueAuthority?.sourceHeadSha), }); if (result.status === "held") { console.log(JSON.stringify({ status: "held", retryAt: result.retryAt })); diff --git a/src/repair/comment-router-core.ts b/src/repair/comment-router-core.ts index 054a45fb4c..9a9aa54a4d 100644 --- a/src/repair/comment-router-core.ts +++ b/src/repair/comment-router-core.ts @@ -1953,6 +1953,57 @@ export function expiredReviewStartStatusLeases({ return expired; } +export function supersededReviewStartStatusLeases({ + comments, + itemNumber, + headSha, + authoritativeHeadSha, + trustedAuthors = new Set(), +}: { + comments: LooseRecord[]; + itemNumber: number; + headSha: string; + authoritativeHeadSha: string; + trustedAuthors?: ReadonlySet; +}): Array<{ commentId: number; headSha: string }> { + const normalizedHead = String(headSha ?? "") + .trim() + .toLowerCase(); + const normalizedAuthoritativeHead = String(authoritativeHeadSha ?? "") + .trim() + .toLowerCase(); + if ( + !Number.isInteger(itemNumber) || + itemNumber <= 0 || + !/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/.test(normalizedHead) || + normalizedAuthoritativeHead !== normalizedHead + ) { + return []; + } + const superseded: Array<{ commentId: number; headSha: string }> = []; + for (const comment of comments) { + const author = String(comment?.user?.login ?? "") + .trim() + .toLowerCase(); + if (!author || !trustedAuthors.has(author)) continue; + const body = String(comment?.body ?? ""); + if (!body.includes(``)) continue; + const canonical = canonicalReviewStartStatusMarker(body); + if (!canonical || canonical.itemNumber !== itemNumber) continue; + if (String(canonical.marker.attrs.v ?? "") !== "1") continue; + const reviewedHead = String(canonical.marker.attrs.sha ?? "") + .trim() + .toLowerCase(); + if (!/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/.test(reviewedHead) || reviewedHead === normalizedHead) { + continue; + } + const rawCommentId = Number(comment?.id); + if (!Number.isInteger(rawCommentId) || rawCommentId <= 0) continue; + superseded.push({ commentId: rawCommentId, headSha: reviewedHead }); + } + return superseded; +} + export function trustedAutomationPredatesReviewStartLease({ command, currentHeadSha, diff --git a/src/repair/comment-webhook.ts b/src/repair/comment-webhook.ts index 3b69fd69c8..8a6d702873 100644 --- a/src/repair/comment-webhook.ts +++ b/src/repair/comment-webhook.ts @@ -57,6 +57,7 @@ type AcceptedItemWebhook = { sourceEvent: "issues" | "pull_request"; sourceAction: string; supersedesInProgress: boolean; + sourceHeadSha?: string; codexTimeoutMs?: number; mediaProofTimeoutMs?: number; }; @@ -302,6 +303,9 @@ export function classifyItemWebhook({ event, payload }: { event: string; payload if (!Number.isInteger(itemNumber) || itemNumber <= 0) { return { accepted: false, reason: "missing pull request number" }; } + const sourceHeadSha = String(asRecord(pull.head).sha ?? "") + .trim() + .toLowerCase(); const reviewBudget = adaptiveReviewBudgetForPullRequest(pull); return { accepted: true, @@ -313,6 +317,7 @@ export function classifyItemWebhook({ event, payload }: { event: string; payload installationId, sourceEvent: "pull_request", sourceAction: action, + ...(/^[0-9a-f]{40}$/.test(sourceHeadSha) ? { sourceHeadSha } : {}), supersedesInProgress: [ "edited", "synchronize", @@ -725,6 +730,7 @@ async function dispatchItemReview({ source_event: accepted.sourceEvent, source_action: accepted.sourceAction, supersedes_in_progress: accepted.supersedesInProgress, + ...(accepted.sourceHeadSha ? { source_head_sha: accepted.sourceHeadSha } : {}), ...(accepted.codexTimeoutMs ? { codex_timeout_ms: accepted.codexTimeoutMs } : {}), ...(accepted.mediaProofTimeoutMs ? { media_proof_timeout_ms: accepted.mediaProofTimeoutMs } diff --git a/test/command.test.ts b/test/command.test.ts index 7712e94367..16d792ba37 100644 --- a/test/command.test.ts +++ b/test/command.test.ts @@ -141,23 +141,43 @@ test("only the exact supplied lease is externally owned", () => { ); }); -test("reserve-review-lease creates and confirms a durable pre-review tuple", () => { +test("reserve-review-lease hydrates a legacy queue claim without cross-head cleanup", () => { const root = mkdtempSync(join(tmpdir(), "cmd-reserve-lease-")); const binDir = join(root, "bin"); const ghPath = join(binDir, "gh.js"); + const curlPath = join(binDir, "curl"); const leasePath = join(root, "lease.json"); + const deleteLogPath = join(root, "deletes.log"); + const curlLogPath = join(root, "curl.log"); const headSha = "0123456789abcdef0123456789abcdef01234567"; + const oldHeadSha = "f".repeat(40); try { mkdirSync(binDir, { recursive: true }); writeFileSync( ghPath, ` -const { existsSync, readFileSync, writeFileSync } = require("node:fs"); +const { appendFileSync, existsSync, readFileSync, writeFileSync } = require("node:fs"); const leasePath = ${JSON.stringify(leasePath)}; +const deleteLogPath = ${JSON.stringify(deleteLogPath)}; const headSha = ${JSON.stringify(headSha)}; +const oldHeadSha = ${JSON.stringify(oldHeadSha)}; const args = process.argv.slice(2); const path = args[1] || ""; -const comments = () => existsSync(leasePath) ? [JSON.parse(readFileSync(leasePath, "utf8"))] : []; +const oldLease = { + id: 9990, + html_url: "https://github.com/openclaw/openclaw/pull/357#issuecomment-9990", + created_at: "2026-07-15T00:00:00Z", + updated_at: "2026-07-15T00:00:00Z", + user: { login: "clawsweeper[bot]" }, + body: [ + "ClawSweeper status: review started.", + \`\`, + "", + ].join("\\n"), +}; +const comments = () => existsSync(leasePath) + ? [oldLease, JSON.parse(readFileSync(leasePath, "utf8"))] + : [oldLease]; if (args[0] === "api" && path === "repos/openclaw/openclaw/issues/357") { console.log(JSON.stringify({ number: 357, @@ -191,6 +211,9 @@ if (args[0] === "api" && path === "repos/openclaw/openclaw/issues/357") { }; writeFileSync(leasePath, JSON.stringify(lease)); console.log(JSON.stringify(lease)); +} else if (args[0] === "api" && /repos\\/openclaw\\/openclaw\\/issues\\/comments\\/\\d+$/.test(path) && args.includes("DELETE")) { + appendFileSync(deleteLogPath, path + "\\n"); + console.log(""); } else { console.error("unexpected gh args", JSON.stringify(args)); process.exit(1); @@ -198,6 +221,15 @@ if (args[0] === "api" && path === "repos/openclaw/openclaw/issues/357") { `, "utf8", ); + writeFileSync( + curlPath, + `#!/usr/bin/env node +require("node:fs").appendFileSync(${JSON.stringify(curlLogPath)}, process.argv.slice(2).join(" ") + "\\n"); +process.stdout.write("200"); +`, + "utf8", + ); + chmodSync(curlPath, 0o755); const result = spawnSync( process.execPath, [ @@ -210,7 +242,20 @@ if (args[0] === "api" && path === "repos/openclaw/openclaw/issues/357") { "--review-timeout-ms", "600000", ], - { encoding: "utf8", env: { ...process.env, ...mockGhBinEnv(ghPath) } }, + { + encoding: "utf8", + env: { + ...process.env, + ...mockGhBinEnv(ghPath, binDir), + GITHUB_RUN_ID: "999", + GITHUB_RUN_ATTEMPT: "1", + EXACT_REVIEW_QUEUE_URL: "https://queue.example.invalid", + EXACT_REVIEW_ITEM_KEY: "openclaw/openclaw#357", + EXACT_REVIEW_LEASE_ID: "lease-357", + EXACT_REVIEW_LEASE_REVISION: "1", + EXACT_REVIEW_CLAIM_GENERATION: "1", + }, + }, ); assert.equal(result.status, 0, result.stderr); @@ -223,6 +268,141 @@ if (args[0] === "api" && path === "repos/openclaw/openclaw/issues/357") { assert.match(lease.body, /clawsweeper-review-status:started/); assert.match(lease.body, /clawsweeper-review-lease item=357/); assert.match(lease.body, new RegExp(`sha=${headSha}`)); + assert.match(readFileSync(curlLogPath, "utf8"), new RegExp(`source_head_sha.*${headSha}`)); + assert.equal(existsSync(deleteLogPath), false); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("reserve-review-lease stale A preserves newer-head lease B", () => { + const root = mkdtempSync(join(tmpdir(), "cmd-reserve-lease-race-")); + const binDir = join(root, "bin"); + const ghPath = join(binDir, "gh.js"); + const curlPath = join(binDir, "curl"); + const postedPath = join(root, "posted.json"); + const pullCountPath = join(root, "pull-count.txt"); + const deleteLogPath = join(root, "deletes.log"); + const curlLogPath = join(root, "curl.log"); + const staleHead = "a".repeat(40); + const newerHead = "b".repeat(40); + try { + mkdirSync(binDir, { recursive: true }); + writeFileSync( + ghPath, + ` +const { appendFileSync, existsSync, readFileSync, writeFileSync } = require("node:fs"); +const postedPath = ${JSON.stringify(postedPath)}; +const pullCountPath = ${JSON.stringify(pullCountPath)}; +const deleteLogPath = ${JSON.stringify(deleteLogPath)}; +const staleHead = ${JSON.stringify(staleHead)}; +const newerHead = ${JSON.stringify(newerHead)}; +const args = process.argv.slice(2); +const path = args[1] || ""; +const newerLease = { + id: 9992, + html_url: "https://github.com/openclaw/openclaw/pull/357#issuecomment-9992", + created_at: "2026-07-23T13:00:02Z", + updated_at: "2026-07-23T13:00:02Z", + user: { login: "clawsweeper[bot]" }, + body: [ + "ClawSweeper status: review started.", + \`\`, + "", + ].join("\\n"), +}; +const comments = () => existsSync(postedPath) + ? [JSON.parse(readFileSync(postedPath, "utf8")), newerLease] + : []; +if (args[0] === "api" && path === "repos/openclaw/openclaw/issues/357") { + console.log(JSON.stringify({ + number: 357, + title: "Fence stale review cleanup", + html_url: "https://github.com/openclaw/openclaw/pull/357", + created_at: "2026-07-23T12:00:00Z", + updated_at: "2026-07-23T13:00:00Z", + closed_at: null, + state: "open", + locked: false, + active_lock_reason: null, + author_association: "CONTRIBUTOR", + user: { login: "reporter" }, + labels: [], + pull_request: {} + })); +} else if (args[0] === "api" && path === "repos/openclaw/openclaw/pulls/357") { + const count = existsSync(pullCountPath) ? Number(readFileSync(pullCountPath, "utf8")) : 0; + writeFileSync(pullCountPath, String(count + 1)); + console.log(JSON.stringify({ head: { sha: count < 2 ? staleHead : newerHead } })); +} else if (args[0] === "api" && path.startsWith("repos/openclaw/openclaw/issues/357/comments") && !args.includes("--method")) { + const value = comments(); + console.log(JSON.stringify(args.includes("--slurp") ? [value] : value)); +} else if (args[0] === "api" && path === "repos/openclaw/openclaw/issues/357/comments" && args.includes("POST")) { + const body = JSON.parse(readFileSync(args[args.indexOf("--input") + 1], "utf8")).body; + const lease = { + id: 9991, + html_url: "https://github.com/openclaw/openclaw/pull/357#issuecomment-9991", + created_at: "2026-07-23T13:00:01Z", + updated_at: "2026-07-23T13:00:01Z", + user: { login: "clawsweeper[bot]" }, + body + }; + writeFileSync(postedPath, JSON.stringify(lease)); + console.log(JSON.stringify(lease)); +} else if (args[0] === "api" && /repos\\/openclaw\\/openclaw\\/issues\\/comments\\/\\d+$/.test(path) && args.includes("DELETE")) { + appendFileSync(deleteLogPath, path + "\\n"); + console.log(""); +} else { + console.error("unexpected gh args", JSON.stringify(args)); + process.exit(1); +} +`, + "utf8", + ); + writeFileSync( + curlPath, + `#!/usr/bin/env node +require("node:fs").appendFileSync(${JSON.stringify(curlLogPath)}, process.argv.slice(2).join(" ") + "\\n"); +process.stdout.write("200"); +`, + "utf8", + ); + chmodSync(curlPath, 0o755); + + const result = spawnSync( + process.execPath, + [ + CLI, + "reserve-review-lease", + "--target-repo", + "openclaw/openclaw", + "--item-number", + "357", + "--review-timeout-ms", + "600000", + ], + { + encoding: "utf8", + env: { + ...process.env, + ...mockGhBinEnv(ghPath, binDir), + GITHUB_RUN_ID: "111", + GITHUB_RUN_ATTEMPT: "1", + EXACT_REVIEW_QUEUE_URL: "https://queue.example.invalid", + EXACT_REVIEW_ITEM_KEY: "openclaw/openclaw#357", + EXACT_REVIEW_LEASE_ID: "lease-357", + EXACT_REVIEW_LEASE_REVISION: "1", + EXACT_REVIEW_CLAIM_GENERATION: "1", + }, + }, + ); + + assert.equal(result.status, 1); + assert.match(result.stderr, /review revision changed while reserving #357/); + assert.deepEqual(readFileSync(deleteLogPath, "utf8").trim().split("\n"), [ + "repos/openclaw/openclaw/issues/comments/9991", + ]); + assert.match(readFileSync(curlLogPath, "utf8"), new RegExp(`source_head_sha.*${staleHead}`)); } finally { rmSync(root, { recursive: true, force: true }); } diff --git a/test/dashboard-worker.test.ts b/test/dashboard-worker.test.ts index cf5d826415..c612d0f920 100644 --- a/test/dashboard-worker.test.ts +++ b/test/dashboard-worker.test.ts @@ -39,6 +39,51 @@ test("exact-review queue defaults to 64 of the 128 global workers", () => { ); }); +test("exact-review source authority sequence survives queue restarts", async () => { + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue({ storage }, {}); + const reserve = (target: ExactReviewQueue, deliveryId: string) => + target.fetch( + new Request("https://clawsweeper-exact-review-queue/source-authority", { + method: "POST", + body: JSON.stringify({ + delivery_id: deliveryId, + installation_id: 123, + decision: { + targetRepo: "openclaw/openclaw", + targetBranch: "main", + itemNumber: 749, + itemKind: "pull_request", + sourceEvent: "pull_request", + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: "a".repeat(40), + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + }), + }), + ); + + assert.deepEqual(await (await reserve(queue, "authority-delivery-1")).json(), { + ok: true, + source_authority_seq: 1, + }); + assert.deepEqual(await (await reserve(queue, "authority-delivery-1")).json(), { + ok: true, + source_authority_seq: 1, + }); + assert.deepEqual(await (await reserve(queue, "authority-delivery-2")).json(), { + ok: true, + source_authority_seq: 2, + }); + + const restarted = new ExactReviewQueue({ storage }, {}); + assert.deepEqual(await (await reserve(restarted, "authority-delivery-3")).json(), { + ok: true, + source_authority_seq: 3, + }); +}); + test("automerge reliability summarizes failures, recovery, duration, and stalled runs", () => { const run = ( id: number, @@ -548,6 +593,813 @@ test("exact-review queue protects an active batch lineage from a later duplicate ); }); +test("superseding source revisions revoke the old lease without Actions cancellation", async () => { + const originalNow = Date.now; + const now = 3_000_000; + Date.now = () => now; + const storage = new MemoryDurableStorage(); + const staleHeadSha = "a".repeat(40); + const currentHeadSha = "b".repeat(40); + const staleBase = leasedExactReviewQueueItem(753, "7530"); + const stale = { + ...staleBase, + createdAt: now - 10 * 60_000, + updatedAt: now - 10 * 60_000, + decision: { + ...staleBase.decision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceHeadSha: staleHeadSha, + sourceUpdatedAt: "2026-07-23T13:00:01Z", + }, + leaseDecision: { + ...staleBase.leaseDecision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceHeadSha: staleHeadSha, + sourceUpdatedAt: "2026-07-23T13:00:01Z", + commandStatusMarker: "", + }, + }; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#753": stale }, + }); + try { + const queue = new ExactReviewQueue( + { storage }, + { + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "90000", + EXACT_REVIEW_DISPATCH_DEBOUNCE_MAX_MS: "180000", + }, + ); + const response = await queue.fetch( + buildExactReviewQueueRequest( + "superseding-head-753", + 753, + "synchronize", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + ), + ); + + assert.equal(response.status, 202); + const state = (await storage.get("exact-review-queue")) as { + items: Record>; + }; + const current = state.items["openclaw/openclaw#753"]; + assert.ok(current); + assert.equal(current.state, "pending"); + assert.equal(current.revision, 2); + assert.equal(current.createdAt, now); + assert.equal(current.nextAttemptAt, now + 90_000); + assert.equal(current.leaseId, undefined); + assert.equal(current.claimedRunId, undefined); + assert.equal(current.leaseDecision, undefined); + assert.equal((current.decision as Record).sourceAction, "synchronize"); + assert.equal((current.decision as Record).commandStatusMarker, undefined); + const stats = await ( + await queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.lanes.review.superseded_total, 1); + assert.deepEqual( + Array.from( + storage.sql.exec( + `SELECT item_key, prior_revision, next_revision, superseded_run_id, + source_action, superseded_at + FROM exact_review_queue_supersessions`, + ), + (row) => ({ ...row }), + ), + [ + { + item_key: "openclaw/openclaw#753", + prior_revision: 1, + next_revision: 2, + superseded_run_id: "7530", + source_action: "synchronize", + superseded_at: now, + }, + ], + ); + + const staleCompletion = await queue.fetch( + new Request("https://clawsweeper-exact-review-queue/complete", { + method: "POST", + body: JSON.stringify({ + lease_id: "lease-753", + item_key: "openclaw/openclaw#753", + lease_revision: 1, + claim_generation: 1, + run_id: "7530", + run_attempt: 1, + outcome: "success", + }), + }), + ); + assert.equal(staleCompletion.status, 409); + assert.deepEqual(await staleCompletion.json(), { error: "lease_not_claimed" }); + } finally { + Date.now = originalNow; + } +}); + +test("source timestamp and sequence ordering protect a newer-head lease", async () => { + const storage = new MemoryDurableStorage(); + const staleHeadSha = "a".repeat(40); + const currentHeadSha = "b".repeat(40); + const activeBase = leasedExactReviewQueueItem(756, "7560"); + const active = { + ...activeBase, + decision: { + ...activeBase.decision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + leaseDecision: { + ...activeBase.leaseDecision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + }; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#756": active }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + + const response = await queue.fetch( + buildExactReviewQueueRequest( + "delayed-stale-head-756", + 756, + "synchronize", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: staleHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + ), + ); + + assert.equal(response.status, 202); + assert.deepEqual(await response.json(), { + ok: true, + deduped: true, + item_key: "openclaw/openclaw#756", + stale_source: true, + }); + const olderSourceResponse = await queue.fetch( + buildExactReviewQueueRequest( + "later-reserved-older-head-756", + 756, + "synchronize", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: staleHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 3, + sourceUpdatedAt: "2026-07-23T13:00:01Z", + }, + ), + ); + assert.equal(olderSourceResponse.status, 202); + assert.equal((await olderSourceResponse.json()).stale_source, true); + const state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + const current = state.items["openclaw/openclaw#756"]; + assert.equal(current.state, "leased"); + assert.equal(current.revision, 1); + assert.equal(current.claimedRunId, "7560"); + assert.equal(current.decision.sourceHeadSha, currentHeadSha); + assert.equal(current.leaseDecision.sourceHeadSha, currentHeadSha); + const stats = await ( + await queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.lanes.review.superseded_total, 0); + assert.deepEqual( + Array.from(storage.sql.exec(`SELECT item_key FROM exact_review_queue_supersessions`)), + [], + ); +}); + +test("delayed opened pull request delivery cannot replace a newer pending synchronize", async () => { + const storage = new MemoryDurableStorage(); + const staleHeadSha = "a".repeat(40); + const currentHeadSha = "b".repeat(40); + const now = Date.now(); + await storage.put("exact-review-queue", { + deliveries: {}, + items: { + "openclaw/openclaw#757": { + key: "openclaw/openclaw#757", + decision: { + targetRepo: "openclaw/openclaw", + targetBranch: "main", + itemNumber: 757, + itemKind: "pull_request", + sourceEvent: "pull_request", + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + state: "pending", + revision: 1, + createdAt: now, + updatedAt: now, + nextAttemptAt: now + 90_000, + attempts: 0, + }, + }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + + const response = await queue.fetch( + buildExactReviewQueueRequest( + "delayed-opened-pending-head-757", + 757, + "opened", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: staleHeadSha, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + ), + ); + + assert.equal(response.status, 202); + assert.equal((await response.json()).stale_source, true); + const state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items["openclaw/openclaw#757"].revision, 1); + assert.equal(state.items["openclaw/openclaw#757"].decision.sourceHeadSha, currentHeadSha); +}); + +test("explicit pull request commands bind to pending source authority", async () => { + const originalNow = Date.now; + const now = 4_000_000; + Date.now = () => now; + const storage = new MemoryDurableStorage(); + const currentHeadSha = "b".repeat(40); + try { + await storage.put("exact-review-queue", { + deliveries: {}, + items: { + "openclaw/openclaw#758": { + key: "openclaw/openclaw#758", + decision: { + targetRepo: "openclaw/openclaw", + targetBranch: "main", + itemNumber: 758, + itemKind: "pull_request", + sourceEvent: "pull_request", + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + state: "pending", + revision: 1, + createdAt: now, + updatedAt: now, + nextAttemptAt: now + 90_000, + attempts: 0, + }, + }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + const commandStatusMarker = + ""; + + const commandResponse = await queue.fetch( + buildExactReviewQueueRequest( + "explicit-command-758", + 758, + "legacy_dispatch", + "pull_request", + "openclaw/openclaw", + { + commandStatusMarker, + statusCommentId: 9001, + }, + ), + ); + + assert.equal(commandResponse.status, 202); + assert.equal((await commandResponse.json()).queued, true); + let state = (await storage.get("exact-review-queue")) as { + items: Record< + string, + { + revision: number; + nextAttemptAt: number; + decision: Record; + } + >; + }; + let current = state.items["openclaw/openclaw#758"]; + assert.equal(current.revision, 2); + assert.equal(current.nextAttemptAt, now); + assert.equal(current.decision.sourceHeadSha, currentHeadSha); + assert.equal(current.decision.sourceHeadVerified, true); + assert.equal(current.decision.sourceAuthoritySeq, 2); + assert.equal(current.decision.sourceUpdatedAt, "2026-07-23T13:00:02Z"); + assert.equal(current.decision.commandStatusMarker, commandStatusMarker); + assert.equal(current.decision.statusCommentId, 9001); + + const staleWebhookResponse = await queue.fetch( + buildExactReviewQueueRequest( + "stale-after-command-758", + 758, + "opened", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: "a".repeat(40), + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:01Z", + }, + ), + ); + assert.equal(staleWebhookResponse.status, 202); + assert.equal((await staleWebhookResponse.json()).stale_source, true); + state = (await storage.get("exact-review-queue")) as typeof state; + current = state.items["openclaw/openclaw#758"]; + assert.equal(current.revision, 2); + assert.equal(current.decision.sourceHeadSha, currentHeadSha); + assert.equal(current.decision.commandStatusMarker, commandStatusMarker); + } finally { + Date.now = originalNow; + } +}); + +test("explicit pull request commands queue behind a dispatching authoritative review", async () => { + const storage = new MemoryDurableStorage(); + const currentHeadSha = "b".repeat(40); + const item = unclaimedExactReviewQueueItem(759); + item.decision = { + ...item.decision, + itemKind: "pull_request", + sourceEvent: "pull_request", + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }; + item.leaseDecision = { ...item.decision }; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#759": item }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + const commandStatusMarker = + ""; + const command = buildExactReviewQueueRequest( + "explicit-command-dispatching-759", + 759, + "legacy_dispatch", + "pull_request", + "openclaw/openclaw", + { + commandStatusMarker, + statusCommentId: 9002, + additionalPrompt: "Inspect the command-requested dispatching follow-up.", + }, + ); + + const response = await queue.fetch(command.clone()); + assert.equal(response.status, 202); + assert.equal((await response.json()).queued, true); + const afterCommand = structuredClone(await storage.get("exact-review-queue")) as { + items: Record< + string, + { + state: string; + revision: number; + leaseId?: string; + leaseRevision?: number; + decision: Record; + leaseDecision?: Record; + } + >; + }; + const current = afterCommand.items["openclaw/openclaw#759"]; + assert.equal(current.state, "dispatching"); + assert.equal(current.revision, 2); + assert.equal(current.leaseId, "lease-759"); + assert.equal(current.leaseRevision, 1); + assert.equal(current.leaseDecision?.commandStatusMarker, undefined); + assert.equal(current.leaseDecision?.sourceHeadSha, currentHeadSha); + assert.equal(current.decision.sourceHeadSha, currentHeadSha); + assert.equal(current.decision.sourceAuthoritySeq, 2); + assert.equal(current.decision.commandStatusMarker, commandStatusMarker); + assert.equal(current.decision.statusCommentId, 9002); + assert.equal( + current.decision.additionalPrompt, + "Inspect the command-requested dispatching follow-up.", + ); + + const redelivery = await queue.fetch(command); + assert.equal(redelivery.status, 202); + assert.equal((await redelivery.json()).deduped, true); + assert.deepEqual(await storage.get("exact-review-queue"), afterCommand); +}); + +test("explicit pull request commands survive an active authoritative lease completion", async () => { + const storage = new MemoryDurableStorage(); + const currentHeadSha = "c".repeat(40); + const item = leasedExactReviewQueueItem(760, "7600"); + item.decision = { + ...item.decision, + itemKind: "pull_request", + sourceEvent: "pull_request", + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 3, + sourceUpdatedAt: "2026-07-23T13:00:03Z", + }; + item.leaseDecision = { ...item.decision }; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#760": item }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + const commandStatusMarker = + ""; + const command = buildExactReviewQueueRequest( + "explicit-command-leased-760", + 760, + "legacy_dispatch", + "pull_request", + "openclaw/openclaw", + { + commandStatusMarker, + statusCommentId: 9003, + additionalPrompt: "Inspect the command-requested leased follow-up.", + }, + ); + + const response = await queue.fetch(command.clone()); + assert.equal(response.status, 202); + assert.equal((await response.json()).queued, true); + const afterCommand = structuredClone(await storage.get("exact-review-queue")) as { + items: Record< + string, + { + state: string; + revision: number; + leaseId?: string; + leaseRevision?: number; + decision: Record; + leaseDecision?: Record; + } + >; + }; + const active = afterCommand.items["openclaw/openclaw#760"]; + assert.equal(active.state, "leased"); + assert.equal(active.revision, 2); + assert.equal(active.leaseId, "lease-760"); + assert.equal(active.leaseRevision, 1); + assert.equal(active.leaseDecision?.commandStatusMarker, undefined); + assert.equal(active.leaseDecision?.sourceHeadSha, currentHeadSha); + assert.equal(active.decision.sourceHeadSha, currentHeadSha); + assert.equal(active.decision.sourceAuthoritySeq, 3); + assert.equal(active.decision.commandStatusMarker, commandStatusMarker); + assert.equal(active.decision.statusCommentId, 9003); + assert.equal(active.decision.additionalPrompt, "Inspect the command-requested leased follow-up."); + + const redelivery = await queue.fetch(command); + assert.equal(redelivery.status, 202); + assert.equal((await redelivery.json()).deduped, true); + assert.deepEqual(await storage.get("exact-review-queue"), afterCommand); + + const completion = await queue.fetch( + new Request("https://clawsweeper-exact-review-queue/complete", { + method: "POST", + body: JSON.stringify({ + lease_id: "lease-760", + item_key: "openclaw/openclaw#760", + lease_revision: 1, + claim_generation: 1, + run_id: "7600", + run_attempt: 1, + outcome: "success", + }), + }), + ); + assert.equal(completion.status, 200); + assert.deepEqual(await completion.json(), { ok: true, requeued: true }); + const completed = (await storage.get("exact-review-queue")) as typeof afterCommand; + const followUp = completed.items["openclaw/openclaw#760"]; + assert.equal(followUp.state, "pending"); + assert.equal(followUp.revision, 2); + assert.equal(followUp.leaseId, undefined); + assert.equal(followUp.leaseDecision, undefined); + assert.equal(followUp.decision.sourceHeadSha, currentHeadSha); + assert.equal(followUp.decision.sourceAuthoritySeq, 3); + assert.equal(followUp.decision.commandStatusMarker, commandStatusMarker); + assert.equal(followUp.decision.statusCommentId, 9003); + assert.equal( + followUp.decision.additionalPrompt, + "Inspect the command-requested leased follow-up.", + ); +}); + +test("same-timestamp verified successor replaces the current pull request head", async () => { + const storage = new MemoryDurableStorage(); + const currentHeadSha = "b".repeat(40); + const successorHeadSha = "c".repeat(40); + const activeBase = leasedExactReviewQueueItem(7571, "75710"); + const active = { + ...activeBase, + decision: { + ...activeBase.decision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + leaseDecision: { + ...activeBase.leaseDecision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + }; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#7571": active }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + + const response = await queue.fetch( + buildExactReviewQueueRequest( + "same-timestamp-successor-7571", + 7571, + "synchronize", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: successorHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 3, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + ), + ); + + assert.equal(response.status, 202); + assert.equal((await response.json()).queued, true); + const state = (await storage.get("exact-review-queue")) as { + items: Record< + string, + { state: string; revision: number; decision: { sourceHeadSha?: string } } + >; + }; + assert.equal(state.items["openclaw/openclaw#7571"].state, "pending"); + assert.equal(state.items["openclaw/openclaw#7571"].revision, 2); + assert.equal(state.items["openclaw/openclaw#7571"].decision.sourceHeadSha, successorHeadSha); +}); + +test("verified live head supersedes a timestamp-less rolling lease", async () => { + const originalNow = Date.now; + const storage = new MemoryDurableStorage(); + const rolling = leasedExactReviewQueueItem(758, "7580"); + rolling.decision.itemKind = "pull_request"; + rolling.decision.sourceEvent = "pull_request"; + rolling.leaseDecision.itemKind = "pull_request"; + rolling.leaseDecision.sourceEvent = "pull_request"; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#758": rolling }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + const currentHeadSha = "c".repeat(40); + Date.now = () => rolling.updatedAt + 1_000; + + try { + const heartbeat = await queue.fetch( + new Request("https://clawsweeper-exact-review-queue/heartbeat", { + method: "POST", + body: JSON.stringify({ + item_key: "openclaw/openclaw#758", + lease_id: "lease-758", + lease_revision: 1, + claim_generation: 1, + run_id: "7580", + run_attempt: 1, + }), + }), + ); + assert.equal(heartbeat.status, 200); + const response = await queue.fetch( + buildExactReviewQueueRequest( + "current-head-after-rolling-758", + 758, + "synchronize", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: currentHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:03Z", + }, + ), + ); + + assert.equal(response.status, 202); + const state = (await storage.get("exact-review-queue")) as { + items: Record< + string, + { state: string; revision: number; decision: { sourceHeadSha?: string } } + >; + }; + assert.equal(state.items["openclaw/openclaw#758"].state, "pending"); + assert.equal(state.items["openclaw/openclaw#758"].revision, 2); + assert.equal(state.items["openclaw/openclaw#758"].decision.sourceHeadSha, currentHeadSha); + } finally { + Date.now = originalNow; + } +}); + +test("same-head edit supersedes a timestamp-less rolling decision", async () => { + const storage = new MemoryDurableStorage(); + const currentHeadSha = "d".repeat(40); + const rolling = leasedExactReviewQueueItem(759, "7590"); + rolling.decision.itemKind = "pull_request"; + rolling.decision.sourceEvent = "pull_request"; + rolling.decision.sourceHeadSha = currentHeadSha; + rolling.leaseDecision.itemKind = "pull_request"; + rolling.leaseDecision.sourceEvent = "pull_request"; + rolling.leaseDecision.sourceHeadSha = currentHeadSha; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#759": rolling }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + + const response = await queue.fetch( + buildExactReviewQueueRequest( + "same-head-edit-after-rolling-759", + 759, + "edited", + "pull_request", + "openclaw/openclaw", + { + sourceHeadSha: currentHeadSha, + sourceAuthoritySeq: 1, + }, + ), + ); + + assert.equal(response.status, 202); + const state = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items["openclaw/openclaw#759"].state, "pending"); + assert.equal(state.items["openclaw/openclaw#759"].revision, 2); + assert.equal(state.items["openclaw/openclaw#759"].decision.sourceAction, "edited"); +}); + +test("exact-review heartbeat binds a claimed pull request source head", async () => { + const storage = new MemoryDurableStorage(); + const staleHeadSha = "c".repeat(40); + const staleBase = leasedExactReviewQueueItem(755, "7550"); + const stale = { + ...staleBase, + decision: { + ...staleBase.decision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceHeadSha: staleHeadSha, + }, + leaseDecision: { + ...staleBase.leaseDecision, + itemKind: "pull_request" as const, + sourceEvent: "pull_request" as const, + sourceHeadSha: staleHeadSha, + }, + }; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#755": stale }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + const heartbeat = (sourceHeadSha?: string) => + queue.fetch( + new Request("https://clawsweeper-exact-review-queue/heartbeat", { + method: "POST", + body: JSON.stringify({ + item_key: "openclaw/openclaw#755", + lease_id: "lease-755", + lease_revision: 1, + claim_generation: 1, + run_id: "7550", + run_attempt: 1, + ...(sourceHeadSha ? { source_head_sha: sourceHeadSha } : {}), + }), + }), + ); + + assert.equal((await heartbeat()).status, 409); + assert.equal((await heartbeat("d".repeat(40))).status, 409); + assert.equal((await heartbeat(staleHeadSha)).status, 200); +}); + +test("recovery revisions cannot supersede an active authoritative exact review", async () => { + const originalFetch = globalThis.fetch; + const storage = new MemoryDurableStorage(); + const active = leasedExactReviewQueueItem(754, "7540"); + active.decision.sourceAction = "synchronize"; + active.leaseDecision.sourceAction = "synchronize"; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#754": active }, + }); + globalThis.fetch = async (input) => { + throw new Error(`unexpected cancellation request ${String(input)}`); + }; + + try { + const queue = new ExactReviewQueue({ storage }, {}); + for (const sourceAction of [ + "failed_review_shard_recovery", + "artifact_retention_recovery", + "source_drift_requeue", + ]) { + const response = await queue.fetch( + buildExactReviewQueueRequest( + `stale-recovery-${sourceAction}`, + 754, + sourceAction, + "pull_request", + "openclaw/openclaw", + ), + ); + assert.equal(response.status, 202); + } + + const state = (await storage.get("exact-review-queue")) as { + items: Record>; + }; + const current = state.items["openclaw/openclaw#754"]; + assert.equal(current.state, "leased"); + assert.equal(current.revision, 1); + assert.equal(current.claimedRunId, "7540"); + assert.equal((current.decision as Record).sourceAction, "synchronize"); + const stats = await ( + await queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.lanes.review.superseded_total, 0); + } finally { + globalThis.fetch = originalFetch; + } +}); + test("exact-review queue sheds only new recovery work above the pending soft limit", async () => { const storage = new MemoryDurableStorage(); const env = { @@ -680,7 +1532,9 @@ test("exact-review heartbeat refreshes only the matching live lease tuple", asyn item_key: "openclaw/openclaw#700", lease_id: "lease-700", lease_revision: 1, + claim_generation: 1, run_id: "7000", + run_attempt: 1, }); const response = await worker.fetch( new Request("https://clawsweeper.openclaw.ai/internal/exact-review/heartbeat", { @@ -703,8 +1557,10 @@ test("exact-review heartbeat refreshes only the matching live lease tuple", asyn const mismatchBody = JSON.stringify({ item_key: "openclaw/openclaw#700", lease_id: "lease-700", - lease_revision: 2, + lease_revision: 1, + claim_generation: 2, run_id: "7000", + run_attempt: 1, }); const mismatch = await worker.fetch( new Request("https://clawsweeper.openclaw.ai/internal/exact-review/heartbeat", { @@ -3186,74 +4042,27 @@ test("exact-review queue coalesces deliveries, dispatches a bound rollout snapsh }), }), ); - assert.equal(claimed.status, 200); - assert.deepEqual(await claimed.json(), { - ok: true, - claimed: true, - protocol_version: 2, - item_key: "openclaw/gogcli#597", - lease_revision: 2, - claim_generation: 1, - decision: { - targetRepo: "openclaw/gogcli", - targetBranch: "main", - itemNumber: 597, - itemKind: "issue", - sourceEvent: "issues", - sourceAction: "edited", - supersedesInProgress: true, - commandStatusMarker, - statusCommentId: 9001, - additionalPrompt: "Check the maintainer-requested regression path.", - codexTimeoutMs: 1_200_000, - mediaProofTimeoutMs: 480_000, - }, - }); + assert.equal(claimed.status, 409); + assert.deepEqual(await claimed.json(), { error: "lease_not_active" }); stats = await ( await queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) ).json(); assert.equal(stats.dispatching, 0); - assert.equal(stats.leased, 1); - assert.equal(stats.handoff_health.phases.leased.count, 1); - assert.equal(typeof stats.oldest_leased_age_seconds, "number"); - assert.equal( - ( - await queue.fetch( - new Request("https://clawsweeper-exact-review-queue/claim", { - method: "POST", - body: JSON.stringify({ - lease_id: leaseId, - item_key: "openclaw/gogcli#597", - lease_revision: 2, - run_id: "101", - run_attempt: 1, - }), - }), - ) - ).status, - 409, - ); - - const completed = await queue.fetch( - new Request("https://clawsweeper-exact-review-queue/complete", { - method: "POST", - body: JSON.stringify({ - lease_id: leaseId, - item_key: "openclaw/gogcli#597", - lease_revision: 2, - claim_generation: 1, - run_id: "100", - run_attempt: 1, - }), - }), - ); - assert.deepEqual(await completed.json(), { ok: true, requeued: true }); + assert.equal(stats.leased, 0); const requeued = (await storage.get("exact-review-queue")) as { items: Record< string, - { attempts: number; nextAttemptAt: number; decision: Record } + { + state: string; + revision: number; + attempts: number; + nextAttemptAt: number; + decision: Record; + } >; }; + assert.equal(requeued.items["openclaw/gogcli#597"].state, "pending"); + assert.equal(requeued.items["openclaw/gogcli#597"].revision, 3); assert.equal(requeued.items["openclaw/gogcli#597"].decision.commandStatusMarker, undefined); assert.equal(requeued.items["openclaw/gogcli#597"].decision.statusCommentId, undefined); assert.equal(requeued.items["openclaw/gogcli#597"].decision.additionalPrompt, undefined); @@ -3310,12 +4119,14 @@ test("exact-review queue upgrades flow metrics without losing publication comple { review_enqueued: stats.lanes.review.enqueued_total, review_completed: stats.lanes.review.completed_total, + review_superseded: stats.lanes.review.superseded_total, publication_enqueued: stats.lanes.publication.enqueued_total, publication_completed: stats.lanes.publication.completed_total, }, { review_enqueued: 0, review_completed: 0, + review_superseded: 0, publication_enqueued: 0, publication_completed: 42, }, @@ -3795,23 +4606,14 @@ test("exact-review imports an active rollback before expiring an old bridge", as assert.equal(storage.rawHas("exact-review-queue"), false); }); -test("exact-review claim preserves its immutable decision across a newer enqueue", async () => { - const storage = new MemoryDurableStorage(); - const item = unclaimedExactReviewQueueItem(620); - await storage.put("exact-review-queue", { - deliveries: {}, - items: { "openclaw/openclaw#620": item }, - }); - const queue = new ExactReviewQueue({ storage }, {}); - - const newer = buildExactReviewQueueRequest( - "newer-620", - 620, - "edited", - "pull_request", - "openclaw/openclaw", - ); - assert.equal((await queue.fetch(newer)).status, 202); +test("a newer exact-review enqueue revokes a claimed immutable decision", async () => { + const storage = new MemoryDurableStorage(); + const item = unclaimedExactReviewQueueItem(620); + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/openclaw#620": item }, + }); + const queue = new ExactReviewQueue({ storage }, {}); const claim = await queue.fetch( new Request("https://clawsweeper-exact-review-queue/claim", { @@ -3836,16 +4638,27 @@ test("exact-review claim preserves its immutable decision across a newer enqueue decision: item.leaseDecision, }); + const newer = buildExactReviewQueueRequest( + "newer-620", + 620, + "edited", + "pull_request", + "openclaw/openclaw", + ); + assert.equal((await queue.fetch(newer)).status, 202); + const claimedState = (await storage.get("exact-review-queue")) as { items: Record< string, { revision: number; decision: { sourceAction: string; itemKind: string }; - leaseDecision: { sourceAction: string; itemKind: string }; + state: string; + leaseDecision?: { sourceAction: string; itemKind: string }; } >; }; + assert.equal(claimedState.items["openclaw/openclaw#620"].state, "pending"); assert.equal(claimedState.items["openclaw/openclaw#620"].revision, 2); assert.deepEqual(claimedState.items["openclaw/openclaw#620"].decision, { targetRepo: "openclaw/openclaw", @@ -3856,7 +4669,7 @@ test("exact-review claim preserves its immutable decision across a newer enqueue sourceAction: "edited", supersedesInProgress: true, }); - assert.deepEqual(claimedState.items["openclaw/openclaw#620"].leaseDecision, item.leaseDecision); + assert.equal(claimedState.items["openclaw/openclaw#620"].leaseDecision, undefined); const complete = await queue.fetch( new Request("https://clawsweeper-exact-review-queue/complete", { @@ -3872,22 +4685,11 @@ test("exact-review claim preserves its immutable decision across a newer enqueue }), }), ); - assert.equal(complete.status, 200); - assert.deepEqual(await complete.json(), { ok: true, requeued: true }); - - const requeued = (await storage.get("exact-review-queue")) as { - items: Record>; - }; - assert.equal(requeued.items["openclaw/openclaw#620"].state, "pending"); - assert.equal(requeued.items["openclaw/openclaw#620"].revision, 2); - assert.equal( - (requeued.items["openclaw/openclaw#620"].decision as { sourceAction: string }).sourceAction, - "edited", - ); - assert.equal(requeued.items["openclaw/openclaw#620"].leaseDecision, undefined); + assert.equal(complete.status, 409); + assert.deepEqual(await complete.json(), { error: "lease_not_claimed" }); }); -test("new exact-review queue serves legacy workflow claims during rolling deploys", async () => { +test("a newer exact-review enqueue revokes a claimed legacy workflow lease", async () => { const storage = new MemoryDurableStorage(); const item = unclaimedExactReviewQueueItem(624); await storage.put("exact-review-queue", { @@ -3896,21 +4698,6 @@ test("new exact-review queue serves legacy workflow claims during rolling deploy }); const queue = new ExactReviewQueue({ storage }, {}); - assert.equal( - ( - await queue.fetch( - buildExactReviewQueueRequest( - "newer-624", - 624, - "edited", - "pull_request", - "openclaw/openclaw", - ), - ) - ).status, - 202, - ); - const legacyClaim = await queue.fetch( new Request("https://clawsweeper-exact-review-queue/claim", { method: "POST", @@ -3933,6 +4720,21 @@ test("new exact-review queue serves legacy workflow claims during rolling deploy decision: item.leaseDecision, }); + assert.equal( + ( + await queue.fetch( + buildExactReviewQueueRequest( + "newer-624", + 624, + "edited", + "pull_request", + "openclaw/openclaw", + ), + ) + ).status, + 202, + ); + const strictCompletion = await queue.fetch( new Request("https://clawsweeper-exact-review-queue/complete", { method: "POST", @@ -3948,7 +4750,7 @@ test("new exact-review queue serves legacy workflow claims during rolling deploy }), ); assert.equal(strictCompletion.status, 409); - assert.deepEqual(await strictCompletion.json(), { error: "lease_protocol_not_claimed" }); + assert.deepEqual(await strictCompletion.json(), { error: "lease_not_claimed" }); const legacyCompletion = await queue.fetch( new Request("https://clawsweeper-exact-review-queue/complete", { @@ -3961,8 +4763,8 @@ test("new exact-review queue serves legacy workflow claims during rolling deploy }), }), ); - assert.equal(legacyCompletion.status, 200); - assert.deepEqual(await legacyCompletion.json(), { ok: true, requeued: true }); + assert.equal(legacyCompletion.status, 409); + assert.deepEqual(await legacyCompletion.json(), { error: "lease_not_claimed" }); const requeued = (await storage.get("exact-review-queue")) as { items: Record>; }; @@ -11277,8 +12079,17 @@ test("hosted webhook rejects label additions before exact-review intake", async } }); -test("hosted webhook enqueues item events with the repository default branch", async () => { +test("hosted issue webhook enqueues without completing pull request authority", async () => { const queue = new ExactReviewQueue({ storage: new MemoryDurableStorage() }, {}); + let authorityCompletionCalls = 0; + const queueStub = { + fetch(request: Request) { + if (new URL(request.url).pathname === "/source-authority/complete") { + authorityCompletionCalls += 1; + } + return queue.fetch(request); + }, + }; const response = await worker.fetch( signedGithubWebhookRequest({ event: "issues", @@ -11299,7 +12110,7 @@ test("hosted webhook enqueues item events with the repository default branch", a }), { CLAWSWEEPER_WEBHOOK_SECRET: "test-secret", - EXACT_REVIEW_QUEUE: new MemoryDurableNamespace(queue), + EXACT_REVIEW_QUEUE: new MemoryDurableNamespace(queueStub), }, ); @@ -11310,37 +12121,35 @@ test("hosted webhook enqueues item events with the repository default branch", a item_key: "openclaw/gogcli#597", superseded_publications: 0, }); + assert.equal(authorityCompletionCalls, 0); }); -test("hosted webhook requeues unlocked and close-guard removal events", async () => { - const closeGuardLabels = [ - "security", - "beta-blocker", - "release-blocker", - "maintainer", - "clawsweeper:human-review", - "clawsweeper:manual-only", - "clawsweeper:automerge", - "clawsweeper:autofix", - ]; - const cases = [ - { event: "issues", action: "unlocked" }, - { event: "pull_request", action: "unlocked" }, - ...closeGuardLabels.flatMap((name) => [ - { event: "issues", action: "unlabeled", label: { name } }, - { event: "pull_request", action: "unlabeled", label: { name } }, - ]), - ]; - for (const [index, { event, action, label }] of cases.entries()) { - const number = 598 + index; - const storage = new MemoryDurableStorage(); - const queue = new ExactReviewQueue({ storage }, {}); - const response = await worker.fetch( +test("hosted synchronize webhook binds and enqueues only the live pull request head", async () => { + const originalFetch = globalThis.fetch; + const originalNow = Date.now; + const now = 3_000_000; + Date.now = () => now; + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue({ storage }, {}); + const staleHeadSha = "a".repeat(40); + const sourceHeadSha = "b".repeat(40); + let verificationCalls = 0; + globalThis.fetch = async (input) => { + verificationCalls += 1; + assert.equal(String(input), "https://api.github.com/repos/openclaw/gogcli/pulls/596"); + return new Response(JSON.stringify({ head: { sha: sourceHeadSha } }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + const send = (headSha: string, updatedAt: string, deliveryId: string) => + worker.fetch( signedGithubWebhookRequest({ - event, + event: "pull_request", secret: "test-secret", + deliveryId, payload: { - action, + action: "synchronize", repository: { full_name: "openclaw/gogcli", default_branch: "trunk", @@ -11349,29 +12158,338 @@ test("hosted webhook requeues unlocked and close-guard removal events", async () fork: false, has_issues: true, }, - ...(event === "issues" ? { issue: { number } } : { pull_request: { number } }), - ...(label ? { label } : {}), + pull_request: { + number: 596, + head: { sha: headSha }, + updated_at: updatedAt, + }, installation: { id: 123 }, }, }), { CLAWSWEEPER_WEBHOOK_SECRET: "test-secret", + GITHUB_TOKEN: "test-token", EXACT_REVIEW_QUEUE: new MemoryDurableNamespace(queue), }, ); + try { + const response = await send(sourceHeadSha, "2026-07-23T13:00:02Z", "synchronize-current-596"); + assert.equal(response.status, 202); + const duplicateResponse = await send( + sourceHeadSha, + "2026-07-23T13:00:02Z", + "synchronize-current-596", + ); + assert.deepEqual(await duplicateResponse.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#596", + }); + assert.equal(verificationCalls, 1); + const staleResponse = await send(staleHeadSha, "2026-07-23T13:00:01Z", "synchronize-stale-596"); + assert.equal(staleResponse.status, 202); + assert.deepEqual(await staleResponse.json(), { + ok: true, + accepted: false, + reason: "stale pull request head", + }); + assert.equal(verificationCalls, 2); + const staleDuplicateResponse = await send( + staleHeadSha, + "2026-07-23T13:00:01Z", + "synchronize-stale-596", + ); + assert.deepEqual(await staleDuplicateResponse.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#596", + }); + assert.equal(verificationCalls, 2); + assert.equal(storage.rawGet("exact-review-source-authority-sequence:v1"), 2); + const stored = (await storage.get("exact-review-queue")) as { + items: Record< + string, + { + decision: { + sourceHeadSha?: string; + sourceHeadVerified?: boolean; + sourceAuthoritySeq?: number; + sourceUpdatedAt?: string; + }; + } + >; + }; + assert.deepEqual(stored.items["openclaw/gogcli#596"].decision, { + targetRepo: "openclaw/gogcli", + targetBranch: "trunk", + itemNumber: 596, + itemKind: "pull_request", + sourceEvent: "pull_request", + sourceAction: "synchronize", + supersedesInProgress: true, + sourceHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }); + } finally { + globalThis.fetch = originalFetch; + Date.now = originalNow; + } +}); + +test("hosted pull request verification survives a transient failure and queue restart", async () => { + const originalFetch = globalThis.fetch; + const originalNow = Date.now; + let now = 3_500_000; + Date.now = () => now; + const storage = new MemoryDurableStorage(); + const sourceHeadSha = "c".repeat(40); + const { privateKey } = generateKeyPairSync("rsa", { + modulusLength: 2048, + privateKeyEncoding: { type: "pkcs8", format: "pem" }, + publicKeyEncoding: { type: "spki", format: "pem" }, + }); + const queueEnv = { + CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", + CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + }; + const queue = new ExactReviewQueue({ storage }, queueEnv); + let verificationFailures = 2; + globalThis.fetch = async (input) => { + const url = new URL(String(input)); + if (url.pathname === "/repos/openclaw/gogcli/pulls/595") { + if (verificationFailures > 0) { + verificationFailures -= 1; + throw new Error("transient GitHub failure"); + } + return jsonResponse({ head: { sha: sourceHeadSha } }); + } + if (url.pathname === "/app/installations/123/access_tokens") { + return jsonResponse({ token: "target-token" }); + } + throw new Error(`unexpected fetch ${url}`); + }; + + try { + const response = await worker.fetch( + signedGithubWebhookRequest({ + event: "pull_request", + secret: "test-secret", + payload: { + action: "synchronize", + repository: { + full_name: "openclaw/gogcli", + default_branch: "trunk", + private: false, + archived: false, + fork: false, + has_issues: true, + }, + pull_request: { + number: 595, + head: { sha: sourceHeadSha }, + updated_at: "2026-07-23T13:00:02Z", + }, + installation: { id: 123 }, + }, + }), + { + CLAWSWEEPER_WEBHOOK_SECRET: "test-secret", + GITHUB_TOKEN: "test-token", + EXACT_REVIEW_QUEUE: new MemoryDurableNamespace(queue), + }, + ); assert.equal(response.status, 202); assert.deepEqual(await response.json(), { ok: true, - queued: true, - item_key: `openclaw/gogcli#${number}`, - superseded_publications: 0, + accepted: true, + deferred: true, + reason: "pull request head verification deferred", }); + assert.equal( + storage.rawHas("exact-review-source-authority-reservation:v1:test-delivery"), + true, + ); + + const restarted = new ExactReviewQueue({ storage }, queueEnv); + await restarted.alarm(); + const deferred = storage.rawGet( + "exact-review-source-authority-reservation:v1:test-delivery", + ) as { + attempts: number; + nextAttemptAt: number; + sourceAuthoritySeq: number; + }; + assert.equal(deferred.attempts, 1); + assert.equal(deferred.sourceAuthoritySeq, 1); + assert.equal(deferred.nextAttemptAt, now + 15_000); + assert.equal(await storage.getAlarm(), deferred.nextAttemptAt); + + now = deferred.nextAttemptAt; + await restarted.alarm(); + const stored = (await storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(stored.items["openclaw/gogcli#595"].decision.sourceHeadSha, sourceHeadSha); + assert.equal(stored.items["openclaw/gogcli#595"].decision.sourceHeadVerified, true); + assert.equal( + storage.rawHas("exact-review-source-authority-reservation:v1:test-delivery"), + false, + ); + } finally { + globalThis.fetch = originalFetch; + Date.now = originalNow; + } +}); + +test("hosted reopened webhook advances to its verified current head", async () => { + const originalFetch = globalThis.fetch; + const storage = new MemoryDurableStorage(); + const previousHeadSha = "d".repeat(40); + const sourceHeadSha = "e".repeat(40); + const existing = leasedExactReviewQueueItem(594, "5940"); + existing.decision.itemKind = "pull_request"; + existing.decision.sourceEvent = "pull_request"; + existing.decision.sourceHeadSha = previousHeadSha; + existing.leaseDecision.itemKind = "pull_request"; + existing.leaseDecision.sourceEvent = "pull_request"; + existing.leaseDecision.sourceHeadSha = previousHeadSha; + await storage.put("exact-review-queue", { + deliveries: {}, + items: { "openclaw/gogcli#594": existing }, + }); + const queue = new ExactReviewQueue({ storage }, {}); + globalThis.fetch = async (input) => { + assert.equal(String(input), "https://api.github.com/repos/openclaw/gogcli/pulls/594"); + return jsonResponse({ head: { sha: sourceHeadSha } }); + }; + + try { + const response = await worker.fetch( + signedGithubWebhookRequest({ + event: "pull_request", + secret: "test-secret", + payload: { + action: "reopened", + repository: { + full_name: "openclaw/gogcli", + default_branch: "trunk", + private: false, + archived: false, + fork: false, + has_issues: true, + }, + pull_request: { + number: 594, + head: { sha: sourceHeadSha }, + updated_at: "2026-07-23T13:00:03Z", + }, + installation: { id: 123 }, + }, + }), + { + CLAWSWEEPER_WEBHOOK_SECRET: "test-secret", + GITHUB_TOKEN: "test-token", + EXACT_REVIEW_QUEUE: new MemoryDurableNamespace(queue), + }, + ); + assert.equal(response.status, 202); const stored = (await storage.get("exact-review-queue")) as { - items: Record; + items: Record; }; - assert.equal(stored.items[`openclaw/gogcli#${number}`].decision.sourceAction, action); - assert.equal(stored.items[`openclaw/gogcli#${number}`].decision.supersedesInProgress, true); + assert.equal(stored.items["openclaw/gogcli#594"].decision.sourceAction, "reopened"); + assert.equal(stored.items["openclaw/gogcli#594"].decision.sourceHeadSha, sourceHeadSha); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("hosted webhook requeues unlocked and close-guard removal events", async () => { + const originalFetch = globalThis.fetch; + globalThis.fetch = async () => jsonResponse({ head: { sha: "e".repeat(40) } }); + const closeGuardLabels = [ + "security", + "beta-blocker", + "release-blocker", + "maintainer", + "clawsweeper:human-review", + "clawsweeper:manual-only", + "clawsweeper:automerge", + "clawsweeper:autofix", + ]; + const cases = [ + { event: "issues", action: "unlocked" }, + { event: "pull_request", action: "unlocked" }, + ...closeGuardLabels.flatMap((name) => [ + { event: "issues", action: "unlabeled", label: { name } }, + { event: "pull_request", action: "unlabeled", label: { name } }, + ]), + ]; + try { + for (const [index, { event, action, label }] of cases.entries()) { + const number = 598 + index; + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue({ storage }, {}); + const response = await worker.fetch( + signedGithubWebhookRequest({ + event, + secret: "test-secret", + payload: { + action, + repository: { + full_name: "openclaw/gogcli", + default_branch: "trunk", + private: false, + archived: false, + fork: false, + has_issues: true, + }, + ...(event === "issues" + ? { issue: { number } } + : { pull_request: { number, head: { sha: "e".repeat(40) } } }), + ...(label ? { label } : {}), + installation: { id: 123 }, + }, + }), + { + CLAWSWEEPER_WEBHOOK_SECRET: "test-secret", + GITHUB_TOKEN: "test-token", + EXACT_REVIEW_QUEUE: new MemoryDurableNamespace(queue), + }, + ); + + assert.equal(response.status, 202); + assert.deepEqual(await response.json(), { + ok: true, + queued: true, + item_key: `openclaw/gogcli#${number}`, + superseded_publications: 0, + }); + const stored = (await storage.get("exact-review-queue")) as { + items: Record< + string, + { + decision: { + sourceAction: string; + supersedesInProgress: boolean; + sourceHeadSha?: string; + }; + } + >; + }; + assert.equal(stored.items[`openclaw/gogcli#${number}`].decision.sourceAction, action); + assert.equal(stored.items[`openclaw/gogcli#${number}`].decision.supersedesInProgress, true); + if (event === "pull_request") { + assert.equal( + stored.items[`openclaw/gogcli#${number}`].decision.sourceHeadSha, + "e".repeat(40), + ); + } + } + } finally { + globalThis.fetch = originalFetch; } }); @@ -12024,23 +13142,27 @@ function signedGithubWebhookRequest({ event, secret, payload, + deliveryId = "test-delivery", }: { event: string; secret: string; payload: unknown; + deliveryId?: string; }) { const body = JSON.stringify(payload); - return signedGithubWebhookBodyRequest({ event, secret, body }); + return signedGithubWebhookBodyRequest({ event, secret, body, deliveryId }); } function signedGithubWebhookBodyRequest({ event, secret, body, + deliveryId = "test-delivery", }: { event: string; secret: string; body: string; + deliveryId?: string; }) { const signature = `sha256=${createHmac("sha256", secret).update(body).digest("hex")}`; return new Request("https://clawsweeper.openclaw.ai/github/webhook", { @@ -12048,7 +13170,7 @@ function signedGithubWebhookBodyRequest({ headers: { "content-type": "application/json", "x-github-event": event, - "x-github-delivery": "test-delivery", + "x-github-delivery": deliveryId, "x-hub-signature-256": signature, }, body, diff --git a/test/exact-review-publication-batches.test.ts b/test/exact-review-publication-batches.test.ts index 6754884c9e..b37ea16332 100644 --- a/test/exact-review-publication-batches.test.ts +++ b/test/exact-review-publication-batches.test.ts @@ -71,6 +71,15 @@ class TestStorage { this.values.delete(key); } + async list(options?: { prefix?: string }) { + const prefix = options?.prefix || ""; + return new Map( + [...this.values.entries()] + .filter(([key]) => key.startsWith(prefix)) + .sort(([left], [right]) => left.localeCompare(right)), + ); + } + async getAlarm() { return this.alarmAt; } diff --git a/test/repair/comment-router-core.test.ts b/test/repair/comment-router-core.test.ts index 255c6872ff..6999f8b22a 100644 --- a/test/repair/comment-router-core.test.ts +++ b/test/repair/comment-router-core.test.ts @@ -70,6 +70,7 @@ import { staleAutomergeActivationReason, staleClosedItemCommandReason, shouldClearMaintainerCommandReaction, + supersededReviewStartStatusLeases, trustedAutomationPredatesReviewStartLease, trustedExactHeadReviewCompletionSince, trustedCloseBlockReason, @@ -1650,6 +1651,66 @@ test("expired review start leases select only provably lapsed dedicated lease co ); }); +test("superseded review start leases select only trusted dedicated comments for older heads", () => { + const itemNumber = 24; + const currentHead = "b".repeat(40); + const oldHead = "a".repeat(40); + const leaseComment = (id, headSha, overrides = {}) => ({ + id, + user: { login: overrides.login ?? "clawsweeper[bot]" }, + body: [ + "ClawSweeper status: review started.", + ``, + overrides.identityMarker ?? ``, + ].join("\n"), + }); + + assert.deepEqual( + supersededReviewStartStatusLeases({ + comments: [ + leaseComment(101, oldHead), + leaseComment(102, currentHead), + leaseComment(103, oldHead, { login: "contributor" }), + leaseComment(104, oldHead, { version: "2" }), + leaseComment(105, oldHead, { + identityMarker: ``, + }), + ], + itemNumber, + headSha: currentHead, + authoritativeHeadSha: currentHead, + trustedAuthors: new Set(["clawsweeper[bot]"]), + }), + [{ commentId: 101, headSha: oldHead }], + ); +}); + +test("stale review A cannot classify newer-head lease B as superseded", () => { + const itemNumber = 24; + const staleHead = "a".repeat(40); + const authoritativeHead = "b".repeat(40); + const newerLease = { + id: 202, + user: { login: "clawsweeper[bot]" }, + body: [ + "ClawSweeper status: review started.", + ``, + ``, + ].join("\n"), + }; + + assert.deepEqual( + supersededReviewStartStatusLeases({ + comments: [newerLease], + itemNumber, + headSha: staleHead, + authoritativeHeadSha: authoritativeHead, + trustedAuthors: new Set(["clawsweeper[bot]"]), + }), + [], + ); +}); + test("first server-created same-head review lease suppresses verdicts without its exact identity", () => { const itemNumber = 103109; const headSha = "0123456789abcdef0123456789abcdef01234567"; diff --git a/test/repair/comment-webhook.test.ts b/test/repair/comment-webhook.test.ts index e919bfaa6d..e4aaaa5bdd 100644 --- a/test/repair/comment-webhook.test.ts +++ b/test/repair/comment-webhook.test.ts @@ -346,7 +346,7 @@ test("webhook accepts eligible pull request events for generic steipete reposito fork: false, has_issues: true, }, - pull_request: { number: 42 }, + pull_request: { number: 42, head: { sha: "a".repeat(40) } }, installation: { id: 456 }, }, }); @@ -361,6 +361,7 @@ test("webhook accepts eligible pull request events for generic steipete reposito installationId: 456, sourceEvent: "pull_request", sourceAction: "synchronize", + sourceHeadSha: "a".repeat(40), supersedesInProgress: true, codexTimeoutMs: 600_000, mediaProofTimeoutMs: 0, @@ -458,6 +459,7 @@ test("pull request webhooks dispatch adaptive Codex timeout payload", async () = }, pull_request: { number: 91093, + head: { sha: "b".repeat(40) }, changed_files: 71, additions: 4176, deletions: 0, @@ -484,6 +486,10 @@ test("pull request webhooks dispatch adaptive Codex timeout payload", async () = (dispatchedBody?.client_payload as Record)?.media_proof_timeout_ms, 240_000, ); + assert.equal( + (dispatchedBody?.client_payload as Record)?.source_head_sha, + "b".repeat(40), + ); } finally { globalThis.fetch = previousFetch; restoreEnv("CLAWSWEEPER_APP_ID", previousAppId); diff --git a/test/review-comment-rendering.test.ts b/test/review-comment-rendering.test.ts index cbdd417708..9d9f142d13 100644 --- a/test/review-comment-rendering.test.ts +++ b/test/review-comment-rendering.test.ts @@ -298,6 +298,7 @@ test("spoofed durable markers cannot suppress a bot-owned start lease", () => { ); assert.match(postStart, /issueReviewCommentState\(options\.item\.number\)/); assert.match(postStart, /freshDedicatedReviewStartLeases\(\{/); + assert.match(postStart, /reapSupersededDedicatedReviewStartLeases\(/); assert.match(postStart, /heldReviewStartStatusCommentResult\(initialLease\.expiresAt, false\)/); assert.match(postStart, /heldReviewStartStatusCommentResult\(winner\.expiresAt, true\)/); assert.match(postStart, /issues\/\$\{options\.item\.number\}\/comments/); @@ -600,6 +601,7 @@ Reason: Maintainers should review the tests after the targeted lane is green. comment, /\*\*Summary\*\*\nAdds regression coverage for session-scoped model overrides\./, ); + assert.match(comment, /Reviewed head: `abc123def456`/); assert.doesNotMatch(comment, /\*\*Workflow note:\*\*/); assert.match(comment, /How this review workflow works<\/summary>/); assert.match( diff --git a/test/sweep-workflow.test.ts b/test/sweep-workflow.test.ts index 9d5bd8ef15..9e11357ae0 100644 --- a/test/sweep-workflow.test.ts +++ b/test/sweep-workflow.test.ts @@ -390,6 +390,19 @@ test("exact event review hands immutable artifacts to the queue-bounded publishe assert.equal(reserveLease.env?.GH_TOKEN, "${{ steps.target-write-token.outputs.token }}"); assert.match(reserveLease.run ?? "", /pnpm run --silent reserve-review-lease/); assert.match(reserveLease.run ?? "", /review-timeout-ms/); + assert.match(source, /Review exact item \{0\} rev \{1\} head \{2\}/); + assert.equal( + reserveLease.env?.EXACT_REVIEW_ITEM_KEY, + "${{ steps.claim-exact-review-queue.outputs.item_key }}", + ); + assert.equal( + reserveLease.env?.EXACT_REVIEW_CLAIM_GENERATION, + "${{ steps.claim-exact-review-queue.outputs.claim_generation }}", + ); + assert.equal( + reserveLease.env?.EXACT_REVIEW_SOURCE_HEAD_SHA, + "${{ fromJSON(steps.claim-exact-review-queue.outputs.decision).sourceHeadSha || '' }}", + ); const resolvePayload = step(reviewer, "Resolve event payload"); assert.match(resolvePayload.run ?? "", /maxExactReviewCodexTimeoutMs = 2_700_000/); assert.match( @@ -406,6 +419,14 @@ test("exact event review hands immutable artifacts to the queue-bounded publishe ); assert.match(step(reviewer, "Review exact event item").run ?? "", /--review-lease-owner/); assert.match(step(reviewer, "Review exact event item").run ?? "", /--review-lease-comment-id/); + assert.match(step(reviewer, "Review exact event item").run ?? "", /claim_generation/); + assert.match(step(reviewer, "Review exact event item").run ?? "", /run_attempt/); + assert.match(step(reviewer, "Review exact event item").run ?? "", /source_head_sha/); + assert.match( + step(reviewer, "Review exact event item").run ?? "", + /kill -TERM -- "-\$review_pgid"/, + ); + assert.match(step(reviewer, "Review exact event item").run ?? "", /sleep 60/); const create = step(reviewer, "Create exact review artifact bundle"); const upload = step(reviewer, "Upload exact review artifact bundle"); @@ -747,11 +768,16 @@ test("exact event review heartbeats its queue lease while Codex runs", () => { assert.doesNotMatch(review.run ?? "", /x-clawsweeper-exact-review-signature/); assert.doesNotMatch(review.run ?? "", /CLAWSWEEPER_WEBHOOK_SECRET/); assert.match(review.run ?? "", /internal\/exact-review\/heartbeat/); - assert.match(review.run ?? "", /sleep 300/); + assert.match(review.run ?? "", /^\s*sleep 60\s*$/m); assert.match(review.run ?? "", /heartbeat_payload=.*\|\| return 0/s); assert.doesNotMatch(review.run ?? "", /test -n "\$CLAWSWEEPER_WEBHOOK_SECRET"/); assert.match(review.run ?? "", /trap cleanup_heartbeat EXIT/); assert.match(review.run ?? "", /kill "\$heartbeat_pid" 2>\/dev\/null \|\| true/); + assert.match(review.run ?? "", /setsid timeout --kill-after=30s/); + assert.match(review.run ?? "", /review_pgid=\$review_pid/); + assert.match(review.run ?? "", /kill -TERM -- "-\$review_pgid"/); + assert.match(review.run ?? "", /kill -KILL -- "-\$review_pgid"/); + assert.match(review.run ?? "", /wait_for_review_group/); const publisher = workflow.jobs["event-review-publish"]!; assert.equal(