diff --git a/.github/workflows/sweep.yml b/.github/workflows/sweep.yml index 8c3f2f95e9..081b069974 100644 --- a/.github/workflows/sweep.yml +++ b/.github/workflows/sweep.yml @@ -271,8 +271,19 @@ jobs: payload="$(TARGET_REPO="$target_repo" TARGET_BRANCH="$target_branch" node <<'NODE' const payload = JSON.parse(process.env.CLIENT_PAYLOAD || "{}"); const itemKind = payload.item_kind === "pull_request" ? "pull_request" : "issue"; - const sourceEvent = payload.source_event === "pull_request" ? "pull_request" : "issues"; + const sourceEvent = + payload.source_event === "pull_request" || payload.source_event === "pull_request_target" + ? "pull_request" + : "issues"; const dispatchKey = String(payload.dispatch_key || "").trim(); + const ingressFingerprint = String(payload.ingress_fingerprint || "").trim().toLowerCase(); + const ingress = + payload.ingress_route === "target_dispatcher" && + itemKind === "pull_request" && + sourceEvent === "pull_request" && + /^[0-9a-f]{64}$/.test(ingressFingerprint) + ? { route: "target_dispatcher", fingerprint: ingressFingerprint } + : undefined; process.stdout.write( JSON.stringify({ delivery_id: dispatchKey @@ -305,6 +316,7 @@ jobs: ? { additionalPrompt: payload.additional_prompt } : {}), }, + ...(ingress ? { ingress } : {}), }), ); NODE diff --git a/dashboard/exact-review-queue.ts b/dashboard/exact-review-queue.ts index 64e1efad8d..e91953e8b7 100644 --- a/dashboard/exact-review-queue.ts +++ b/dashboard/exact-review-queue.ts @@ -93,9 +93,14 @@ export type ExactReviewPublication = { export type ExactReviewDecision = ExactReviewBaseDecision & { publication?: ExactReviewPublication; }; +export type ExactReviewIngress = { + route: "direct_webhook" | "target_dispatcher"; + fingerprint: string; +}; export type ExactReviewQueueItem = { key: string; decision: ExactReviewDecision; + ingressFingerprint?: string; leaseDecision?: ExactReviewDecision; state: "pending" | "dispatching" | "leased" | "parked"; revision: number; @@ -275,6 +280,7 @@ type ExactReviewQueueMetricTotals = { type ExactReviewSourceAuthorityReservation = { deliveryId: string; decision: ExactReviewDecision; + ingress?: ExactReviewIngress; installationId: number; sourceAuthoritySeq: number; attempts: number; @@ -400,6 +406,7 @@ 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_INGRESS_TABLE = "exact_review_queue_ingress"; 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"; @@ -455,6 +462,7 @@ const DEFAULT_STATE_WRITER_COORDINATOR_QUEUED_STALE_MS = 2 * 60_000; // hung runner into a permanent queue owner. The Git fence blocks any in-flight // push that outlives this absolute coordinator horizon. const DEFAULT_STATE_WRITER_COORDINATOR_MAX_LEASE_AGE_MS = 30 * 60_000; +const EXACT_REVIEW_INGRESS_FINGERPRINT_PATTERN = /^[0-9a-f]{64}$/; export class ExactReviewQueue { private storage; @@ -489,6 +497,7 @@ export class ExactReviewQueue { const body = objectValue(await request.json().catch(() => null)); const deliveryId = String(body.delivery_id || "").trim(); const decision = exactReviewDecisionFrom(body.decision); + const ingress = body.ingress === undefined ? undefined : exactReviewIngressFrom(body.ingress); const installationId = Number(body.installation_id); if ( !deliveryId || @@ -496,6 +505,7 @@ export class ExactReviewQueue { !decision || decision.itemKind !== "pull_request" || decision.publication || + (body.ingress !== undefined && (!ingress || ingress.route !== "direct_webhook")) || !Number.isInteger(installationId) || installationId <= 0 ) { @@ -521,7 +531,8 @@ export class ExactReviewQueue { existing.deliveryId !== deliveryId || existing.installationId !== installationId || stableJson(exactReviewDecisionWithoutSourceAuthority(existing.decision)) !== - stableJson(decision) + stableJson(decision) || + stableJson(existing.ingress || null) !== stableJson(ingress || null) ) { throw new Error("conflicting exact-review source authority reservation"); } @@ -537,6 +548,7 @@ export class ExactReviewQueue { const created: ExactReviewSourceAuthorityReservation = { deliveryId, decision: { ...decision, sourceAuthoritySeq: next }, + ...(ingress ? { ingress } : {}), installationId, sourceAuthoritySeq: next, attempts: 0, @@ -894,11 +906,21 @@ export class ExactReviewQueue { const body = objectValue(await request.json().catch(() => null)); const deliveryId = String(body.delivery_id || "").trim(); const decision = exactReviewDecisionFrom(body.decision); + const ingress = body.ingress === undefined ? undefined : exactReviewIngressFrom(body.ingress); if (!deliveryId) return json({ error: "missing_delivery_id" }, 400); if (deliveryId.startsWith(EXACT_REVIEW_QUEUE_LEGACY_GENERATION_PREFIX)) { return json({ error: "reserved_delivery_id" }, 400); } if (!decision) return json({ error: "invalid_exact_review_item" }, 400); + if (body.ingress !== undefined && !ingress) { + return json({ error: "invalid_exact_review_ingress" }, 400); + } + if ( + ingress && + (decision.itemKind !== "pull_request" || decision.sourceEvent !== "pull_request") + ) { + return json({ error: "invalid_exact_review_ingress" }, 400); + } if (!isExactReviewQueueTargetEnabled(decision, this.env)) { return json({ ok: true, accepted: false, reason: "target not enabled" }, 202); } @@ -910,6 +932,7 @@ export class ExactReviewQueue { : new Set(); const accepted = this.storage.transactionSync(() => { this.pruneDeliveryReceiptsSync(now); + this.pruneIngressReceiptsSync(now); this.storage.sql.exec( `DELETE FROM ${EXACT_REVIEW_QUEUE_DELIVERY_TABLE} WHERE delivery_id = ? AND received_at <= ?`, @@ -929,6 +952,9 @@ export class ExactReviewQueue { this.syncLegacyCompatibilitySync(this.readStateSync()); return { deduped: true as const }; } + const counterpartIngress = ingress + ? this.recordIngressSync(ingress, decision.targetBranch, now) + : null; const state = this.readStateSync(); // A delayed or lost alarm must not let an expired one-shot recovery @@ -939,6 +965,22 @@ export class ExactReviewQueue { exactReviewPublicationDispatchLeaseMs(this.env), exactReviewHeartbeatGraceMs(this.env), ); + const key = exactReviewItemKey(decision); + const currentIngressItem = state.items[key]; + // A counterpart receipt is conclusive only after its own route reached + // queue state. Preserve the one live fallback-first item for a verified + // direct promotion; otherwise, suppress the old event instead of + // replacing a newer queue revision. + if ( + counterpartIngress?.admitted && + !( + currentIngressItem?.ingressFingerprint === ingress?.fingerprint && + exactReviewIngressCanPromoteFallback(ingress, decision) + ) + ) { + this.writeStateSync(state); + return { deduped: true as const, crossRoute: true as const, key, state }; + } let supersededPublications = 0; if (incomingPublicationRevision) { const incomingLineage = exactReviewPublicationLineage(decision); @@ -974,6 +1016,7 @@ export class ExactReviewQueue { return { deduped: true as const, superseded: true as const, + ...(counterpartIngress ? { crossRoute: true as const } : {}), publicationRevision: incomingPublicationRevision.sourceRevision, supersededByRevision: newestSourceRevision, state, @@ -1081,6 +1124,7 @@ export class ExactReviewQueue { return { deduped: true as const, semantic: true as const, + ...(counterpartIngress ? { crossRoute: true as const } : {}), key: retained.item.key, semanticDuplicatesRemoved, state, @@ -1108,10 +1152,10 @@ export class ExactReviewQueue { supersededPublications += 1; } } - const key = exactReviewItemKey(decision); const current = state.items[key]; let supersededRunId: string | null = null; let supersessionAudit: ExactReviewSupersessionAudit | null = null; + let ingressAdmitted = false; if (current) { const ignoredRecovery = isLowPriorityExactReviewDecision(decision); // A recovery is only a one-shot repair of a failed shard. It may create a queue item, @@ -1148,6 +1192,7 @@ export class ExactReviewQueue { return { deduped: true as const, staleSource: true as const, + ...(counterpartIngress ? { crossRoute: true as const } : {}), key, state, }; @@ -1205,6 +1250,7 @@ export class ExactReviewQueue { current.firstFailureAt = undefined; current.lastFailureReason = undefined; } + ingressAdmitted = true; } } else { if ( @@ -1220,6 +1266,7 @@ export class ExactReviewQueue { state.items[key] = { key, decision, + ...(ingress ? { ingressFingerprint: ingress.fingerprint } : {}), state: "pending", revision: this.nextExactReviewItemRevisionSync(key), createdAt: now, @@ -1227,6 +1274,22 @@ export class ExactReviewQueue { nextAttemptAt: exactReviewQueueDebouncedAttemptAt(state, decision, now, now, this.env), attempts: 0, }; + ingressAdmitted = true; + } + if ( + ingressAdmitted && + decision.itemKind === "pull_request" && + decision.sourceEvent === "pull_request" + ) { + if (ingress) { + state.items[key].ingressFingerprint = ingress.fingerprint; + this.markIngressAdmittedSync(ingress, now); + } else { + // A later legacy-only update must not retain the older event's + // cross-route identity: a delayed verified direct delivery for + // that older event is not a safe promotion of this revision. + delete state.items[key].ingressFingerprint; + } } this.writeStateSync(state); if (supersededPublications) { @@ -1271,6 +1334,9 @@ export class ExactReviewQueue { superseded_by_revision: accepted.supersededByRevision, } : {}), + ...("crossRoute" in accepted && accepted.crossRoute + ? { dedupe_scope: "cross_route" } + : {}), }, 202, ); @@ -5035,6 +5101,38 @@ export class ExactReviewQueue { `CREATE INDEX IF NOT EXISTS state_append_dead_letters_status ON ${STATE_APPEND_DEAD_LETTER_TABLE} (status, last_failed_at, dead_letter_id)`, ); + this.storage.sql.exec( + `CREATE TABLE IF NOT EXISTS ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} ( + fingerprint TEXT NOT NULL, + route TEXT NOT NULL CHECK (route IN ('direct_webhook', 'target_dispatcher')), + target_branch TEXT NOT NULL, + received_at INTEGER NOT NULL, + admitted_at INTEGER, + PRIMARY KEY (fingerprint, route) + ) STRICT`, + ); + const hasIngressAdmission = Array.from( + this.storage.sql.exec( + `SELECT name FROM pragma_table_info('${EXACT_REVIEW_QUEUE_INGRESS_TABLE}') + WHERE name = 'admitted_at'`, + ), + ).length; + if (!hasIngressAdmission) { + this.storage.sql.exec( + `ALTER TABLE ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} ADD COLUMN admitted_at INTEGER`, + ); + } + // Before admission state was explicit, direct-route receipts were the + // canonical durable ingress record. Preserve that completed-delivery + // meaning on the rolling schema upgrade and on a later re-upgrade after a + // rollback. Current code records an unadmitted direct receipt as zero, so + // NULL remains exclusive to the pre-admission-schema worker; legacy + // fallback receipts stay NULL until this queue admits them. + this.storage.sql.exec( + `UPDATE ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} + SET admitted_at = received_at + WHERE route = 'direct_webhook' AND admitted_at IS NULL`, + ); // Flow telemetry is independent of queue rollback compatibility. A // separate singleton keeps cumulative lane counters monotonic without // changing the normalized queue schema or its legacy shadow contract. @@ -5149,6 +5247,10 @@ export class ExactReviewQueue { `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 INDEX IF NOT EXISTS exact_review_queue_ingress_received_at + ON ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} (received_at, fingerprint)`, + ); this.storage.sql.exec( `CREATE TABLE IF NOT EXISTS ${EXACT_REVIEW_QUEUE_METRIC_BUCKET_TABLE} ( bucket_start INTEGER PRIMARY KEY, @@ -6604,6 +6706,66 @@ export class ExactReviewQueue { } } + private pruneIngressReceiptsSync(now: number) { + const cutoff = now - EXACT_REVIEW_QUEUE_DELIVERY_TTL_MS; + for (let batch = 0; batch < EXACT_REVIEW_QUEUE_DELIVERY_PRUNE_MAX_BATCHES; batch += 1) { + const deleted = Array.from( + this.storage.sql.exec( + `DELETE FROM ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} + WHERE (fingerprint, route) IN ( + SELECT fingerprint, route + FROM ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} + WHERE received_at <= ? + ORDER BY received_at, fingerprint, route + LIMIT ${EXACT_REVIEW_QUEUE_DELIVERY_PRUNE_BATCH} + ) + RETURNING fingerprint`, + cutoff, + ), + ); + if (deleted.length < EXACT_REVIEW_QUEUE_DELIVERY_PRUNE_BATCH) break; + } + } + + private recordIngressSync(ingress: ExactReviewIngress, targetBranch: string, now: number) { + const counterpart = ingress.route === "direct_webhook" ? "target_dispatcher" : "direct_webhook"; + const matched = Array.from( + this.storage.sql.exec( + `SELECT admitted_at FROM ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} + WHERE fingerprint = ? AND route = ? AND target_branch = ? + LIMIT 1`, + ingress.fingerprint, + counterpart, + targetBranch, + ), + )[0] as { admitted_at?: number | null } | undefined; + this.storage.sql.exec( + `INSERT INTO ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} + (fingerprint, route, target_branch, received_at, admitted_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(fingerprint, route) DO UPDATE SET + target_branch = excluded.target_branch, + received_at = excluded.received_at`, + ingress.fingerprint, + ingress.route, + targetBranch, + now, + ingress.route === "direct_webhook" ? 0 : null, + ); + return matched ? { admitted: Number(matched.admitted_at) > 0 } : null; + } + + private markIngressAdmittedSync(ingress: ExactReviewIngress, now: number) { + this.storage.sql.exec( + `UPDATE ${EXACT_REVIEW_QUEUE_INGRESS_TABLE} + SET admitted_at = CASE WHEN admitted_at IS NULL OR admitted_at = 0 THEN ? ELSE admitted_at END + WHERE fingerprint = ? AND route = ?`, + now, + ingress.fingerprint, + ingress.route, + ); + } + private deliveryReceiptCountSync() { const row = Array.from( this.storage.sql.exec( @@ -6733,6 +6895,7 @@ export class ExactReviewQueue { headers: { "content-type": "application/json" }, body: JSON.stringify({ delivery_id: reservation.deliveryId, + ...(reservation.ingress ? { ingress: reservation.ingress } : {}), decision: { ...reservation.decision, sourceHeadVerified: true, @@ -6932,6 +7095,8 @@ function exactReviewSourceAuthorityReservationFrom( const reservation = objectValue(value); const deliveryId = String(reservation.deliveryId || "").trim(); const decision = exactReviewDecisionFrom(reservation.decision); + const ingress = + reservation.ingress === undefined ? undefined : exactReviewIngressFrom(reservation.ingress); const installationId = Number(reservation.installationId); const sourceAuthoritySeq = Number(reservation.sourceAuthoritySeq); const attempts = Number(reservation.attempts); @@ -6942,6 +7107,7 @@ function exactReviewSourceAuthorityReservationFrom( !decision || decision.itemKind !== "pull_request" || decision.publication || + (reservation.ingress !== undefined && (!ingress || ingress.route !== "direct_webhook")) || !Number.isInteger(installationId) || installationId <= 0 || !Number.isSafeInteger(sourceAuthoritySeq) || @@ -6958,6 +7124,7 @@ function exactReviewSourceAuthorityReservationFrom( return { deliveryId, decision, + ...(ingress ? { ingress } : {}), installationId, sourceAuthoritySeq, attempts, @@ -6965,6 +7132,31 @@ function exactReviewSourceAuthorityReservationFrom( }; } +function exactReviewIngressFrom(value): ExactReviewIngress | null { + const ingress = objectValue(value); + const route = String(ingress.route || ""); + const fingerprint = String(ingress.fingerprint || "") + .trim() + .toLowerCase(); + if (route !== "direct_webhook" && route !== "target_dispatcher") return null; + if (!EXACT_REVIEW_INGRESS_FINGERPRINT_PATTERN.test(fingerprint)) return null; + return { route, fingerprint }; +} + +function exactReviewIngressCanPromoteFallback( + ingress: ExactReviewIngress | undefined, + decision: ExactReviewDecision, +) { + const sourceAuthoritySeq = Number(decision.sourceAuthoritySeq || 0); + return ( + ingress?.route === "direct_webhook" && + decision.sourceHeadVerified === true && + /^[0-9a-f]{40}$/.test(String(decision.sourceHeadSha || "").toLowerCase()) && + Number.isSafeInteger(sourceAuthoritySeq) && + sourceAuthoritySeq > 0 + ); +} + function exactReviewBaseDecisionFrom(value): ExactReviewBaseDecision | null { const decision = objectValue(value); const targetRepo = String(decision.targetRepo || "").trim(); diff --git a/dashboard/worker.ts b/dashboard/worker.ts index 82e5a24b3d..590fab3f37 100644 --- a/dashboard/worker.ts +++ b/dashboard/worker.ts @@ -29,6 +29,7 @@ import { type ExactReviewClaimedRun, type ExactReviewCompletionOutcome, type ExactReviewDecision, + type ExactReviewIngress, } from "./exact-review-queue.ts"; import { AUTOMERGE_METRICS_EVENT_TYPE, @@ -1071,11 +1072,17 @@ 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 ingress = await exactReviewPullRequestIngress({ + event, + payload, + decision: itemDecision, + }); const sourceAuthority = itemDecision.itemKind === "pull_request" ? await reserveExactReviewSourceAuthority(env, { deliveryId, decision: itemDecision, + ingress, }) : null; if (itemDecision.itemKind === "pull_request" && sourceAuthority === null) { @@ -1123,6 +1130,7 @@ async function githubWebhook(request, env, ctx) { env, deliveryId, decision: exactReviewDecision, + ingress, }); if (!queued) return json({ error: "exact_review_queue_not_configured" }, 503); if (sourceAuthoritySeq !== null) { @@ -1461,6 +1469,31 @@ async function bindLivePullRequestHeadAuthority({ : null; } +async function exactReviewPullRequestIngress({ event, payload, decision }) { + if (event !== "pull_request" || decision.itemKind !== "pull_request") return undefined; + const pullRequest = objectValue(payload.pull_request); + const headSha = String(objectValue(pullRequest.head).sha || "") + .trim() + .toLowerCase(); + const updatedAt = String(pullRequest.updated_at || "").trim(); + if (!/^[0-9a-f]{40}$/.test(headSha) || !updatedAt) return undefined; + return { + route: "direct_webhook" as const, + fingerprint: await sha256Text( + JSON.stringify({ + version: 1, + target_repo: decision.targetRepo.toLowerCase(), + item_number: decision.itemNumber, + action: decision.sourceAction, + head_sha: headSha, + updated_at: updatedAt, + body: typeof pullRequest.body === "string" ? pullRequest.body : "", + label: String(objectValue(payload.label).name || ""), + }), + ), + } satisfies ExactReviewIngress; +} + function isCloseGuardLabel(value) { const label = String(objectValue(value).name || "") .trim() @@ -1517,9 +1550,11 @@ async function reserveExactReviewSourceAuthority( { deliveryId, decision, + ingress, }: { deliveryId: string; decision: ExactReviewDecision & { installationId?: number }; + ingress?: ExactReviewIngress; }, ): Promise<{ deduped: true } | { sourceAuthoritySeq: number } | null> { const queue = exactReviewQueueStub(env); @@ -1531,6 +1566,7 @@ async function reserveExactReviewSourceAuthority( body: JSON.stringify({ delivery_id: deliveryId, decision, + ...(ingress ? { ingress } : {}), installation_id: decision.installationId, }), }), @@ -1880,10 +1916,12 @@ async function authenticatedExactReviewReconcile(request, env) { async function enqueueExactReview({ deliveryId, decision, + ingress, env, }: { deliveryId: string; decision: ExactReviewDecision; + ingress?: ExactReviewIngress; env: DashboardEnv; }) { const queue = exactReviewQueueStub(env); @@ -1892,7 +1930,7 @@ async function enqueueExactReview({ new Request("https://clawsweeper-exact-review-queue/enqueue", { method: "POST", headers: { "content-type": "application/json" }, - body: JSON.stringify({ delivery_id: deliveryId, decision }), + body: JSON.stringify({ delivery_id: deliveryId, decision, ...(ingress ? { ingress } : {}) }), }), ); const body = objectValue(await response.json().catch(() => null)); diff --git a/docs/target-dispatcher.md b/docs/target-dispatcher.md index f3ce84a92b..12698f194f 100644 --- a/docs/target-dispatcher.md +++ b/docs/target-dispatcher.md @@ -130,14 +130,51 @@ jobs: echo "::notice::Skipping ClawSweeper dispatch because no dispatch credential is configured." exit 0 fi + ingress_fingerprint="$(node <<'NODE' + const crypto = require("node:crypto"); + const fs = require("node:fs"); + const event = JSON.parse(fs.readFileSync(process.env.GITHUB_EVENT_PATH, "utf8")); + const pullRequest = event.pull_request && typeof event.pull_request === "object" + ? event.pull_request + : {}; + const headSha = String(pullRequest.head?.sha || "").trim().toLowerCase(); + const updatedAt = String(pullRequest.updated_at || "").trim(); + if ( + process.env.ITEM_KIND !== "pull_request" || + !/^[0-9a-f]{40}$/.test(headSha) || + !updatedAt + ) { + process.stdout.write(""); + } else { + process.stdout.write( + crypto + .createHash("sha256") + .update( + JSON.stringify({ + version: 1, + target_repo: String(process.env.TARGET_REPO || "").toLowerCase(), + item_number: Number(process.env.ITEM_NUMBER), + action: String(process.env.SOURCE_ACTION || ""), + head_sha: headSha, + updated_at: updatedAt, + body: typeof pullRequest.body === "string" ? pullRequest.body : "", + label: String(event.label?.name || ""), + }), + ) + .digest("hex"), + ); + } + NODE + )" payload="$(jq -nc \ --arg target_repo "$TARGET_REPO" \ --argjson item_number "$ITEM_NUMBER" \ --arg item_kind "$ITEM_KIND" \ --arg source_event "$SOURCE_EVENT" \ --arg source_action "$SOURCE_ACTION" \ + --arg ingress_fingerprint "$ingress_fingerprint" \ --argjson supersedes_in_progress "$SUPERSEDES_IN_PROGRESS" \ - '{event_type:"clawsweeper_item",client_payload:{target_repo:$target_repo,item_number:$item_number,item_kind:$item_kind,source_event:$source_event,source_action:$source_action,supersedes_in_progress:$supersedes_in_progress}}')" + '{event_type:"clawsweeper_item",client_payload:({target_repo:$target_repo,item_number:$item_number,item_kind:$item_kind,source_event:$source_event,source_action:$source_action,supersedes_in_progress:$supersedes_in_progress} + (if $ingress_fingerprint != "" then {ingress_route:"target_dispatcher",ingress_fingerprint:$ingress_fingerprint} else {} end))}')" gh api repos/openclaw/clawsweeper/dispatches \ --method POST \ --input - <<< "$payload" @@ -251,6 +288,31 @@ still handles the broader backlog, with `stale_insufficient_info` and `mostly_implemented_on_main` blocked until the item is at least 60 days old; stale-insufficient-info issues also require 60 days without a non-bot comment. +## Cross-route exact-review identity + +The direct GitHub App webhook and this compatibility dispatcher are independent +reliability routes. Do not disable either one. For pull requests, the template +above emits an opaque SHA-256 fingerprint of the immutable event snapshot +(repository, number, action, head SHA, update timestamp, body, and label). The +ClawSweeper durable queue coalesces only the matching fingerprint and resolved +target branch when it has seen it from the other route. A target default-branch +change therefore does not cross-route-coalesce. It also cannot let an unverified +fallback replace an already verified direct source decision; that fallback is +recorded as stale source. If the fallback arrives first, the later verified +direct decision promotes that same queue item instead of creating another one. +A route with no valid fingerprint remains admissible when no verified direct +decision exists, so a legacy-only delivery stays a safe fallback. The durable +receipt remains after a completed review, so a delayed matching counterpart is +also suppressed rather than recreating the completed review. A fallback that +the queue rejects as stale is not an admission receipt, so it cannot suppress +the later verified direct event. + +Before enabling this protocol for a target repository, roll out the dispatcher +and verify direct-only, legacy-only, cross-route duplicate, later body/revision, +and maintainer-command cases. The hash is an opaque queue receipt, not a +head-SHA dedupe key; body and metadata updates therefore produce a new event +identity. + `openclaw/clawhub` dispatches are intentionally skipped while the receiver variable `CLAWSWEEPER_ENABLE_CLAWHUB` is not `1`. Enable it only after the ClawSweeper GitHub App is installed on `openclaw/clawhub`; otherwise the diff --git a/test/clawsweeper.test.ts b/test/clawsweeper.test.ts index 82c833639f..ed74a54a7c 100644 --- a/test/clawsweeper.test.ts +++ b/test/clawsweeper.test.ts @@ -2203,6 +2203,9 @@ test("sweep workflow executes only durable queue leases without runner-side admi assert.match(legacyIntakeBlock, /gh api "repos\/\$target_repo" --jq \.default_branch/); assert.match(legacyIntakeBlock, /targetBranch: process\.env\.TARGET_BRANCH/); assert.doesNotMatch(legacyIntakeBlock, /targetBranch: payload\.target_branch \|\| "main"/); + assert.match(legacyIntakeBlock, /payload\.source_event === "pull_request_target"/); + assert.match(legacyIntakeBlock, /payload\.ingress_route === "target_dispatcher"/); + assert.match(legacyIntakeBlock, /fingerprint: ingressFingerprint/); assert.match(legacyIntakeBlock, /commandStatusMarker: payload\.command_status_marker/); assert.match(legacyIntakeBlock, /statusCommentId: payload\.status_comment_id/); assert.match(legacyIntakeBlock, /additionalPrompt: payload\.additional_prompt/); @@ -2279,6 +2282,21 @@ test("sweep workflow executes only durable queue leases without runner-side admi assert.doesNotMatch(exactReviewStep, /--codex-timeout-ms 600000/); }); +test("target dispatcher documents opt-in cross-route identity", () => { + const dispatcher = readText("docs/target-dispatcher.md"); + + assert.match(dispatcher, /## Cross-route exact-review identity/); + assert.doesNotMatch(dispatcher, /maintainer decision required/i); + assert.match(dispatcher, /ingress_route:"target_dispatcher"/); + assert.match(dispatcher, /ingress_fingerprint/); + assert.match(dispatcher, /recorded as stale source/); + assert.match(dispatcher, /later verified\s+direct decision promotes that same queue item/); + assert.match(dispatcher, /legacy-only delivery stays a safe\s+fallback/); + assert.match(dispatcher, /delayed matching counterpart is\s+also suppressed/); + assert.match(dispatcher, /rejects as stale is not an admission receipt/); + assert.match(dispatcher, /not a\s+head-SHA dedupe key/); +}); + test("sweep workflow gives high-context Codex reviews twenty minutes by default", () => { const workflow = readText(".github/workflows/sweep.yml"); diff --git a/test/dashboard-worker.test.ts b/test/dashboard-worker.test.ts index 3be5252b2a..14ff32c310 100644 --- a/test/dashboard-worker.test.ts +++ b/test/dashboard-worker.test.ts @@ -15094,7 +15094,7 @@ test("hosted synchronize webhook binds and enqueues only the live pull request h } }); -test("hosted pull request verification survives a transient failure and queue restart", async () => { +test("hosted pull request verification preserves ingress through a transient failure and queue restart", async () => { const originalFetch = globalThis.fetch; const originalNow = Date.now; let now = 3_500_000; @@ -15167,6 +15167,32 @@ test("hosted pull request verification survives a transient failure and queue re storage.rawHas("exact-review-source-authority-reservation:v1:test-delivery"), true, ); + const fallbackFingerprint = createHash("sha256") + .update( + JSON.stringify({ + version: 1, + target_repo: "openclaw/gogcli", + item_number: 595, + action: "synchronize", + head_sha: sourceHeadSha, + updated_at: "2026-07-23T13:00:02Z", + body: "", + label: "", + }), + ) + .digest("hex"); + const fallback = await queue.fetch( + buildExactReviewQueueRequest( + "legacy-after-deferred-direct", + 595, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: fallbackFingerprint } }, + ), + ); + assert.equal((await fallback.json()).queued, true); const restarted = new ExactReviewQueue({ storage }, queueEnv); await restarted.alarm(); @@ -15185,10 +15211,22 @@ test("hosted pull request verification survives a transient failure and queue re now = deferred.nextAttemptAt; await restarted.alarm(); const stored = (await storage.get("exact-review-queue")) as { - items: Record; + items: Record< + string, + { + revision: number; + decision: { + sourceHeadSha?: string; + sourceHeadVerified?: boolean; + sourceAuthoritySeq?: number; + }; + } + >; }; + assert.equal(stored.items["openclaw/gogcli#595"].revision, 2); assert.equal(stored.items["openclaw/gogcli#595"].decision.sourceHeadSha, sourceHeadSha); assert.equal(stored.items["openclaw/gogcli#595"].decision.sourceHeadVerified, true); + assert.equal(stored.items["openclaw/gogcli#595"].decision.sourceAuthoritySeq, 1); assert.equal( storage.rawHas("exact-review-source-authority-reservation:v1:test-delivery"), false, @@ -15261,6 +15299,737 @@ test("hosted reopened webhook advances to its verified current head", async () = } }); +test("exact-review queue drops a delayed matching ingress after the first review completes", async () => { + const sourceHeadSha = "f".repeat(40); + const fingerprint = "e".repeat(64); + const harness = createExactReviewAdmissionHarness(() => + jsonResponse({ state: "open", head: { sha: sourceHeadSha } }), + ); + + try { + const direct = await harness.queue.fetch( + buildExactReviewQueueRequest( + "completed-direct-ingress", + 601, + "synchronize", + "pull_request", + "openclaw/gogcli", + { + targetBranch: "trunk", + sourceHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + { ingress: { route: "direct_webhook", fingerprint } }, + ), + ); + assert.equal((await direct.json()).queued, true); + + await harness.queue.alarm(); + const dispatched = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + const item = dispatched.items["openclaw/gogcli#601"]; + assert.ok(item); + const claim = await harness.queue.fetch( + new Request("https://clawsweeper-exact-review-queue/claim", { + method: "POST", + body: JSON.stringify({ + lease_id: item.leaseId, + item_key: "openclaw/gogcli#601", + lease_revision: item.leaseRevision, + run_id: "6010", + run_attempt: 1, + }), + }), + ); + assert.equal(claim.status, 200); + const claimed = (await claim.json()) as { claim_generation: number }; + const completed = await harness.queue.fetch( + new Request("https://clawsweeper-exact-review-queue/complete", { + method: "POST", + body: JSON.stringify({ + lease_id: item.leaseId, + item_key: "openclaw/gogcli#601", + lease_revision: item.leaseRevision, + claim_generation: claimed.claim_generation, + run_id: "6010", + run_attempt: 1, + outcome: "success", + }), + }), + ); + assert.deepEqual(await completed.json(), { ok: true, requeued: false }); + + const fallback = await harness.queue.fetch( + buildExactReviewQueueRequest( + "delayed-fallback-ingress", + 601, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint } }, + ), + ); + assert.deepEqual(await fallback.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#601", + dedupe_scope: "cross_route", + }); + const afterFallback = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(afterFallback.items["openclaw/gogcli#601"], undefined); + } finally { + harness.restore(); + } +}); + +test("exact-review queue upgrades ingress receipts with admission tracking", async () => { + const storage = new MemoryDurableStorage(); + const receivedAt = Date.now(); + storage.sql.exec(`CREATE TABLE exact_review_queue_ingress ( + fingerprint TEXT NOT NULL, + route TEXT NOT NULL, + target_branch TEXT NOT NULL, + received_at INTEGER NOT NULL, + PRIMARY KEY (fingerprint, route) + ) STRICT`); + storage.sql.exec( + `INSERT INTO exact_review_queue_ingress (fingerprint, route, target_branch, received_at) + VALUES (?, ?, ?, ?)`, + "f".repeat(64), + "direct_webhook", + "trunk", + receivedAt, + ); + const queue = new ExactReviewQueue({ storage }, {}); + + await queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")); + const columns = Array.from( + storage.sql.exec(`SELECT name FROM pragma_table_info('exact_review_queue_ingress')`), + ) as Array<{ name: string }>; + assert.ok(columns.some((column) => column.name === "admitted_at")); + const migrated = Array.from( + storage.sql.exec( + `SELECT admitted_at FROM exact_review_queue_ingress + WHERE fingerprint = ? AND route = 'direct_webhook'`, + "f".repeat(64), + ), + )[0] as { admitted_at: number }; + assert.equal(migrated.admitted_at, receivedAt); + const delayedFallback = await queue.fetch( + buildExactReviewQueueRequest( + "legacy-after-upgrade", + 601, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: "f".repeat(64) } }, + ), + ); + assert.deepEqual(await delayedFallback.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#601", + dedupe_scope: "cross_route", + }); +}); + +test("exact-review queue re-upgrade admits a direct receipt written by a rollback", async () => { + const storage = new MemoryDurableStorage(); + const firstUpgrade = new ExactReviewQueue({ storage }, {}); + await firstUpgrade.fetch(new Request("https://clawsweeper-exact-review-queue/stats")); + const receivedAt = Date.now(); + storage.sql.exec( + `INSERT INTO exact_review_queue_ingress + (fingerprint, route, target_branch, received_at, admitted_at) + VALUES (?, ?, ?, ?, NULL)`, + "9".repeat(64), + "direct_webhook", + "trunk", + receivedAt, + ); + + const reupgraded = new ExactReviewQueue({ storage }, {}); + const migrated = Array.from( + storage.sql.exec( + `SELECT admitted_at FROM exact_review_queue_ingress + WHERE fingerprint = ? AND route = 'direct_webhook'`, + "9".repeat(64), + ), + )[0] as { admitted_at: number }; + assert.equal(migrated.admitted_at, receivedAt); + const delayedFallback = await reupgraded.fetch( + buildExactReviewQueueRequest( + "fallback-after-reupgrade", + 605, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: "9".repeat(64) } }, + ), + ); + assert.deepEqual(await delayedFallback.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#605", + dedupe_scope: "cross_route", + }); +}); + +test("unadmitted fallback receipts do not suppress a later verified direct event", async () => { + const firstHeadSha = "a".repeat(40); + const secondHeadSha = "b".repeat(40); + const firstFingerprint = "a".repeat(64); + const secondFingerprint = "b".repeat(64); + const harness = createExactReviewAdmissionHarness(() => + jsonResponse({ state: "open", head: { sha: firstHeadSha } }), + ); + + try { + await harness.queue.fetch( + buildExactReviewQueueRequest( + "verified-first", + 602, + "synchronize", + "pull_request", + "openclaw/gogcli", + { + targetBranch: "trunk", + sourceHeadSha: firstHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + { ingress: { route: "direct_webhook", fingerprint: firstFingerprint } }, + ), + ); + const staleFallback = await harness.queue.fetch( + buildExactReviewQueueRequest( + "stale-fallback-second", + 602, + "edited", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: secondFingerprint } }, + ), + ); + assert.deepEqual(await staleFallback.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#602", + stale_source: true, + }); + + await harness.queue.alarm(); + const dispatched = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + const item = dispatched.items["openclaw/gogcli#602"]; + const claim = await harness.queue.fetch( + new Request("https://clawsweeper-exact-review-queue/claim", { + method: "POST", + body: JSON.stringify({ + lease_id: item.leaseId, + item_key: "openclaw/gogcli#602", + lease_revision: item.leaseRevision, + run_id: "6020", + run_attempt: 1, + }), + }), + ); + const claimed = (await claim.json()) as { claim_generation: number }; + const completed = await harness.queue.fetch( + new Request("https://clawsweeper-exact-review-queue/complete", { + method: "POST", + body: JSON.stringify({ + lease_id: item.leaseId, + item_key: "openclaw/gogcli#602", + lease_revision: item.leaseRevision, + claim_generation: claimed.claim_generation, + run_id: "6020", + run_attempt: 1, + outcome: "success", + }), + }), + ); + assert.deepEqual(await completed.json(), { ok: true, requeued: false }); + + const verifiedSecond = await harness.queue.fetch( + buildExactReviewQueueRequest( + "verified-second", + 602, + "edited", + "pull_request", + "openclaw/gogcli", + { + targetBranch: "trunk", + sourceHeadSha: secondHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + sourceUpdatedAt: "2026-07-23T13:01:02Z", + }, + { ingress: { route: "direct_webhook", fingerprint: secondFingerprint } }, + ), + ); + assert.equal((await verifiedSecond.json()).queued, true); + const afterDirect = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(afterDirect.items["openclaw/gogcli#602"].decision.sourceAction, "edited"); + assert.equal(afterDirect.items["openclaw/gogcli#602"].decision.sourceHeadSha, secondHeadSha); + } finally { + harness.restore(); + } +}); + +test("a delayed counterpart cannot replace a newer admitted fallback", async () => { + const firstFingerprint = "c".repeat(64); + const secondFingerprint = "d".repeat(64); + const firstHeadSha = "c".repeat(40); + const harness = createExactReviewAdmissionHarness(() => jsonResponse({ state: "open" })); + + try { + await harness.queue.fetch( + buildExactReviewQueueRequest( + "fallback-first-complete", + 603, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: firstFingerprint } }, + ), + ); + await harness.queue.alarm(); + const dispatched = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + const item = dispatched.items["openclaw/gogcli#603"]; + const claim = await harness.queue.fetch( + new Request("https://clawsweeper-exact-review-queue/claim", { + method: "POST", + body: JSON.stringify({ + lease_id: item.leaseId, + item_key: "openclaw/gogcli#603", + lease_revision: item.leaseRevision, + run_id: "6030", + run_attempt: 1, + }), + }), + ); + const claimed = (await claim.json()) as { claim_generation: number }; + await harness.queue.fetch( + new Request("https://clawsweeper-exact-review-queue/complete", { + method: "POST", + body: JSON.stringify({ + lease_id: item.leaseId, + item_key: "openclaw/gogcli#603", + lease_revision: item.leaseRevision, + claim_generation: claimed.claim_generation, + run_id: "6030", + run_attempt: 1, + outcome: "success", + }), + }), + ); + const newerFallback = await harness.queue.fetch( + buildExactReviewQueueRequest( + "fallback-second-pending", + 603, + "edited", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: secondFingerprint } }, + ), + ); + assert.equal((await newerFallback.json()).queued, true); + + const delayedDirect = await harness.queue.fetch( + buildExactReviewQueueRequest( + "delayed-direct-first", + 603, + "synchronize", + "pull_request", + "openclaw/gogcli", + { + targetBranch: "trunk", + sourceHeadSha: firstHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 1, + sourceUpdatedAt: "2026-07-23T13:00:02Z", + }, + { ingress: { route: "direct_webhook", fingerprint: firstFingerprint } }, + ), + ); + assert.deepEqual(await delayedDirect.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#603", + dedupe_scope: "cross_route", + }); + const afterDelayed = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(afterDelayed.items["openclaw/gogcli#603"].decision.sourceAction, "edited"); + } finally { + harness.restore(); + } +}); + +test("a delayed direct ingress cannot promote across a newer legacy-only update", async () => { + const firstFingerprint = "e".repeat(64); + const firstHeadSha = "e".repeat(40); + const secondHeadSha = "f".repeat(40); + const harness = createExactReviewAdmissionHarness(() => jsonResponse({ state: "open" })); + + try { + const fallback = await harness.queue.fetch( + buildExactReviewQueueRequest( + "fallback-before-legacy-update", + 604, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: firstFingerprint } }, + ), + ); + assert.equal((await fallback.json()).queued, true); + + const legacyUpdate = await harness.queue.fetch( + buildExactReviewQueueRequest( + "legacy-only-newer-update", + 604, + "edited", + "pull_request", + "openclaw/gogcli", + { + targetBranch: "trunk", + sourceHeadSha: secondHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 2, + }, + ), + ); + assert.equal((await legacyUpdate.json()).queued, true); + + const delayedDirect = await harness.queue.fetch( + buildExactReviewQueueRequest( + "delayed-direct-before-legacy-update", + 604, + "synchronize", + "pull_request", + "openclaw/gogcli", + { + targetBranch: "trunk", + sourceHeadSha: firstHeadSha, + sourceHeadVerified: true, + sourceAuthoritySeq: 3, + }, + { ingress: { route: "direct_webhook", fingerprint: firstFingerprint } }, + ), + ); + assert.deepEqual(await delayedDirect.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#604", + dedupe_scope: "cross_route", + }); + const afterDelayed = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(afterDelayed.items["openclaw/gogcli#604"].decision.sourceAction, "edited"); + assert.equal(afterDelayed.items["openclaw/gogcli#604"].ingressFingerprint, undefined); + } finally { + harness.restore(); + } +}); + +test("exact-review queue coalesces matching ingress and promotes verified direct authority", async () => { + const storage = new MemoryDurableStorage(); + const queue = new ExactReviewQueue({ storage }, {}); + const env = { + CLAWSWEEPER_WEBHOOK_SECRET: "test-secret", + GITHUB_TOKEN: "test-token", + EXACT_REVIEW_QUEUE: new MemoryDurableNamespace(queue), + }; + const repository = { + full_name: "openclaw/gogcli", + default_branch: "trunk", + private: false, + archived: false, + fork: false, + has_issues: true, + }; + const pullRequest = { + number: 597, + head: { sha: "a".repeat(40) }, + updated_at: "2026-07-19T10:19:00Z", + body: "Add durable proof.", + }; + const originalFetch = globalThis.fetch; + const liveHeads = new Map([ + [597, "a".repeat(40)], + [599, "c".repeat(40)], + [600, "d".repeat(40)], + ]); + globalThis.fetch = async (input) => { + const url = new URL(input instanceof Request ? input.url : String(input)); + const match = url.pathname.match(/^\/repos\/openclaw\/gogcli\/pulls\/(\d+)$/); + assert.ok(match); + const head = liveHeads.get(Number(match[1])); + assert.ok(head); + return jsonResponse({ head: { sha: head } }); + }; + + try { + const direct = await worker.fetch( + signedGithubWebhookRequest({ + event: "pull_request", + secret: "test-secret", + deliveryId: "direct-pr-delivery", + payload: { + action: "synchronize", + repository, + pull_request: pullRequest, + installation: { id: 123 }, + }, + }), + env, + ); + assert.deepEqual(await direct.json(), { + ok: true, + queued: true, + item_key: "openclaw/gogcli#597", + superseded_publications: 0, + }); + const fingerprint = createHash("sha256") + .update( + JSON.stringify({ + version: 1, + target_repo: "openclaw/gogcli", + item_number: 597, + action: "synchronize", + head_sha: "a".repeat(40), + updated_at: "2026-07-19T10:19:00Z", + body: "Add durable proof.", + label: "", + }), + ) + .digest("hex"); + const fallback = await queue.fetch( + buildExactReviewQueueRequest( + "legacy-pr-delivery", + 597, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint } }, + ), + ); + assert.deepEqual(await fallback.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#597", + dedupe_scope: "cross_route", + }); + + const fallbackFirstFingerprint = createHash("sha256") + .update( + JSON.stringify({ + version: 1, + target_repo: "openclaw/gogcli", + item_number: 599, + action: "synchronize", + head_sha: "c".repeat(40), + updated_at: "2026-07-19T10:21:00Z", + body: "Fallback arrived first.", + label: "", + }), + ) + .digest("hex"); + const fallbackFirst = await queue.fetch( + buildExactReviewQueueRequest( + "legacy-first-delivery", + 599, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: fallbackFirstFingerprint } }, + ), + ); + assert.deepEqual(await fallbackFirst.json(), { + ok: true, + queued: true, + item_key: "openclaw/gogcli#599", + superseded_publications: 0, + }); + const directSecond = await worker.fetch( + signedGithubWebhookRequest({ + event: "pull_request", + secret: "test-secret", + deliveryId: "direct-after-legacy-delivery", + payload: { + action: "synchronize", + repository, + pull_request: { + number: 599, + head: { sha: "c".repeat(40) }, + updated_at: "2026-07-19T10:21:00Z", + body: "Fallback arrived first.", + }, + installation: { id: 123 }, + }, + }), + env, + ); + assert.deepEqual(await directSecond.json(), { + ok: true, + queued: true, + item_key: "openclaw/gogcli#599", + superseded_publications: 0, + }); + + const branchChangeFingerprint = createHash("sha256") + .update( + JSON.stringify({ + version: 1, + target_repo: "openclaw/gogcli", + item_number: 600, + action: "synchronize", + head_sha: "d".repeat(40), + updated_at: "2026-07-19T10:22:00Z", + body: "The default branch changed.", + label: "", + }), + ) + .digest("hex"); + const directBeforeBranchChange = await worker.fetch( + signedGithubWebhookRequest({ + event: "pull_request", + secret: "test-secret", + deliveryId: "direct-old-default-branch-delivery", + payload: { + action: "synchronize", + repository: { ...repository, default_branch: "old-default" }, + pull_request: { + number: 600, + head: { sha: "d".repeat(40) }, + updated_at: "2026-07-19T10:22:00Z", + body: "The default branch changed.", + }, + installation: { id: 123 }, + }, + }), + env, + ); + assert.deepEqual(await directBeforeBranchChange.json(), { + ok: true, + queued: true, + item_key: "openclaw/gogcli#600", + superseded_publications: 0, + }); + const fallbackAfterBranchChange = await queue.fetch( + buildExactReviewQueueRequest( + "legacy-new-default-branch-delivery", + 600, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "new-default" }, + { ingress: { route: "target_dispatcher", fingerprint: branchChangeFingerprint } }, + ), + ); + assert.deepEqual(await fallbackAfterBranchChange.json(), { + ok: true, + deduped: true, + item_key: "openclaw/gogcli#600", + stale_source: true, + }); + + const legacyOnly = await queue.fetch( + buildExactReviewQueueRequest( + "legacy-only-delivery", + 598, + "synchronize", + "pull_request", + "openclaw/gogcli", + { targetBranch: "trunk" }, + { ingress: { route: "target_dispatcher", fingerprint: "b".repeat(64) } }, + ), + ); + assert.deepEqual(await legacyOnly.json(), { + ok: true, + queued: true, + item_key: "openclaw/gogcli#598", + superseded_publications: 0, + }); + + const bodyUpdate = await worker.fetch( + signedGithubWebhookRequest({ + event: "pull_request", + secret: "test-secret", + deliveryId: "direct-pr-body-update", + payload: { + action: "edited", + repository, + pull_request: { + ...pullRequest, + updated_at: "2026-07-19T10:20:00Z", + body: "Add revised durable proof.", + }, + installation: { id: 123 }, + }, + }), + env, + ); + assert.deepEqual(await bodyUpdate.json(), { + ok: true, + queued: true, + item_key: "openclaw/gogcli#597", + superseded_publications: 0, + }); + const state = (await storage.get("exact-review-queue")) as { + items: Record< + string, + { + revision: number; + decision: { + sourceAction: string; + targetBranch: string; + sourceHeadSha?: string; + sourceHeadVerified?: boolean; + sourceAuthoritySeq?: number; + }; + } + >; + }; + assert.equal(state.items["openclaw/gogcli#597"].revision, 2); + assert.equal(state.items["openclaw/gogcli#597"].decision.sourceAction, "edited"); + assert.equal(state.items["openclaw/gogcli#599"].revision, 2); + assert.equal(state.items["openclaw/gogcli#599"].decision.sourceHeadSha, "c".repeat(40)); + assert.equal(state.items["openclaw/gogcli#599"].decision.sourceHeadVerified, true); + assert.ok(state.items["openclaw/gogcli#599"].decision.sourceAuthoritySeq); + // A compatibility fallback may be legacy-only, but it cannot replace a + // source-head-verified direct decision merely because its branch resolves differently. + assert.equal(state.items["openclaw/gogcli#600"].revision, 1); + assert.equal(state.items["openclaw/gogcli#600"].decision.targetBranch, "old-default"); + } 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) } }); @@ -16184,6 +16953,7 @@ function buildExactReviewQueueRequest( itemKind: "issue" | "pull_request" = "issue", targetRepo = "openclaw/gogcli", decisionOverrides: Record = {}, + envelopeOverrides: Record = {}, ) { const sourceEvent = itemKind === "issue" ? "issues" : "pull_request"; return new Request("https://clawsweeper-exact-review-queue/enqueue", { @@ -16200,6 +16970,7 @@ function buildExactReviewQueueRequest( supersedesInProgress: sourceAction === "edited" || sourceAction === "synchronize", ...decisionOverrides, }, + ...envelopeOverrides, }), }); }