diff --git a/dashboard/exact-review-queue.ts b/dashboard/exact-review-queue.ts index 2cbcae2a12..14dd95ba28 100644 --- a/dashboard/exact-review-queue.ts +++ b/dashboard/exact-review-queue.ts @@ -202,6 +202,7 @@ export type ExactReviewQueueState = { dispatchFailureAt?: number; dispatchFailureFingerprint?: string; dispatchConsecutiveFailures?: number; + reviewAdmissionNextAt?: number; publicationBatchDispatchedAt?: number; publicationBatchDispatchSucceeded?: boolean; publicationBatchDispatchPendingUntil?: number; @@ -345,6 +346,9 @@ const EXACT_REVIEW_RETRY_LIMIT = 8; const EXACT_REVIEW_RECONCILE_RUN_LIMIT = 128; const EXACT_REVIEW_RECONCILE_CLAIM_MATCH_LIMIT = EXACT_REVIEW_RECONCILE_RUN_LIMIT * 2; export const EXACT_REVIEW_RECONCILE_CONCURRENCY = 8; +const EXACT_REVIEW_ADMISSION_LIVE_CHECK_CONCURRENCY = 4; +const EXACT_REVIEW_ADMISSION_LIVE_CHECK_MAX_ITEMS = 4; +const EXACT_REVIEW_ADMISSION_INTERVAL_MS = 5_000; const EXACT_REVIEW_RECONCILE_LIST_PAGE_LIMIT = 3; const EXACT_REVIEW_PUBLICATION_ENQUEUE_SUPERSEDE_LIMIT = 100; const EXACT_REVIEW_PUBLICATION_RECONCILE_LIMIT = 100; @@ -2091,6 +2095,16 @@ export class ExactReviewQueue { await this.scheduleNext(current, Date.now()); return; } + const snapshotReviewAdmissionNextAt = Number(snapshot.dispatcher?.reviewAdmissionNextAt || 0); + if ( + snapshotReviewAdmissionNextAt > startedAt && + snapshotAdmission.some((item) => !exactReviewQueueIsPublication(item)) && + !snapshotAdmission.some(exactReviewQueueIsPublication) + ) { + if (snapshotChanged && !batchDispatchAttempted) await this.writeState(snapshot); + await this.scheduleNext(batchDispatchAttempted ? this.readStateSync() : snapshot, startedAt); + return; + } let preflight: { ok: true; token: string; workflowState: string } | { ok: false } = { ok: false, @@ -2107,23 +2121,29 @@ export class ExactReviewQueue { const now = Date.now(); const state = this.readStateSync(); // Do not rely on reading back the marker written before preflight. Carry the - // current alarm's dispatch result through every later dispatcher write. - const persistedBatchDispatcherFields = exactReviewBatchDispatcherFields(state.dispatcher); - const batchDispatcherFields = batchDispatchRecordedAt - ? { - ...persistedBatchDispatcherFields, - publicationBatchDispatchedAt: batchDispatchRecordedAt, - publicationBatchDispatchSucceeded: batchDispatchSucceeded, - } - : persistedBatchDispatcherFields; + // current alarm's dispatch result only while it still owns that marker: a + // batch claim may clear its pending reservation while later external reads + // release the Durable Object input gate. + const batchDispatcherFieldsFor = (dispatcher: ExactReviewQueueState["dispatcher"]) => { + const persisted = exactReviewBatchDispatcherFields(dispatcher); + return batchDispatchRecordedAt !== undefined && + dispatcher?.publicationBatchDispatchedAt === batchDispatchRecordedAt + ? { + ...persisted, + publicationBatchDispatchedAt: batchDispatchRecordedAt, + publicationBatchDispatchSucceeded: batchDispatchSucceeded, + } + : persisted; + }; + const batchDispatcherFields = batchDispatcherFieldsFor(state.dispatcher); const batchOwnership = this.batchStore.activeLeaseSnapshot(now); - reclaimExpiredExactReviewLeases( + const reclaimed = reclaimExpiredExactReviewLeases( state, now, exactReviewPublicationDispatchLeaseMs(this.env), exactReviewHeartbeatGraceMs(this.env), ); - expireExactReviewPublicationItems(state, now, this.env); + const expired = expireExactReviewPublicationItems(state, now, this.env); // The preflight fetch releases the input gate, so publication demand may // have crossed a scale boundary while the workflow state was checked. const publicationControl = this.refreshPublicationControlSync(state, now); @@ -2147,6 +2167,9 @@ export class ExactReviewQueue { this.freshPublicationItemKeysSync(state, now), exactReviewPublicationFreshLaneMaxItems(this.env), ); + const reviewAdmissionNextAt = Number(state.dispatcher?.reviewAdmissionNextAt || 0); + const admission = + reviewAdmissionNextAt > now ? admitted.filter(exactReviewQueueIsPublication) : admitted; if (!preflight.ok) { const retryAt = now + exactReviewWorkflowPausedRetryMs(this.env); state.dispatcher = { @@ -2175,35 +2198,184 @@ export class ExactReviewQueue { return; } + // Keep any local lease reclamation or publication recovery before the + // live lookup: its awaits release the input gate, so the re-read below + // must start from this alarm's housekeeping result. + if (reclaimed || expired) await this.writeState(state); + + // A queued review can become terminal before it has a worker. Probe only + // the bounded admission set, then revalidate the exact pending revision + // after the external reads release the Durable Object input gate. + // Keep a short durable admission interval as well as a bounded pass. This + // prevents a ready backlog from turning the one-second alarm wake-up into + // repeated App-token, item-read, and workflow-dispatch bursts. + const reviewCandidates = admission + .filter((item) => !exactReviewQueueIsPublication(item)) + .map((item) => ({ key: item.key, revision: item.revision, decision: item.decision })); + const liveCandidates = reviewCandidates.slice(0, EXACT_REVIEW_ADMISSION_LIVE_CHECK_MAX_ITEMS); + const targetTokens = new Map>(); + const targetTokenFor = (targetRepo: string) => { + let token = targetTokens.get(targetRepo); + if (!token) { + token = exactReviewTargetReadToken(this.env, targetRepo); + targetTokens.set(targetRepo, token); + } + return token; + }; + const liveStates = await mapWithConcurrency( + liveCandidates, + EXACT_REVIEW_ADMISSION_LIVE_CHECK_CONCURRENCY, + async (candidate) => { + try { + const token = await targetTokenFor(candidate.decision.targetRepo); + return { + ...candidate, + state: await exactReviewTargetItemState(token, candidate.decision), + }; + } catch (error) { + const failure = exactReviewAdmissionFailure(error); + console.warn( + `exact-review admission target check failed for ${candidate.key}`, + error instanceof Error ? error.message : String(error), + ); + return { ...candidate, state: "unavailable" as const, failure }; + } + }, + ); + + const checkedAt = Date.now(); + const liveStateByCandidate = new Map(liveStates.map((candidate) => [candidate.key, candidate])); + const checkedState = this.readStateSync(); + const checkedBatchDispatcherFields = batchDispatcherFieldsFor(checkedState.dispatcher); const priorDispatchConsecutiveFailures = Number( - state.dispatcher?.dispatchConsecutiveFailures || 0, + checkedState.dispatcher?.dispatchConsecutiveFailures || 0, + ); + let globalAdmissionFailure: ExactReviewDispatchFailure | null = null; + for (const candidate of liveStates) { + if (candidate.state !== "unavailable" || candidate.failure.scope !== "global") continue; + globalAdmissionFailure = candidate.failure; + break; + } + let terminalCompleted = 0; + for (const candidate of liveStates) { + const item = checkedState.items[candidate.key]; + if ( + !item || + item.revision !== candidate.revision || + item.state !== "pending" || + exactReviewQueueIsPublication(item) + ) { + continue; + } + if (candidate.state === "terminal" && !exactReviewQueueHasCommandContext(item)) { + delete checkedState.items[item.key]; + terminalCompleted += 1; + continue; + } + if (candidate.state === "unavailable") { + // A shared GitHub or credential failure must not consume each item's + // retry budget. The dispatcher backoff below holds the whole admission + // pass until that dependency recovers. + if (globalAdmissionFailure) continue; + item.attempts += 1; + const failureAttempts = Number(item.reviewFailureAttempts || 0) + 1; + item.reviewFailureAttempts = failureAttempts; + if (failureAttempts >= EXACT_REVIEW_RETRY_LIMIT) { + item.state = "parked"; + item.parkedReason = "review_retry_exhausted"; + item.updatedAt = checkedAt; + continue; + } + item.nextAttemptAt = Math.max( + exactReviewQueueEnqueueAttemptAt(checkedState, checkedAt), + checkedAt + exactReviewRetryDelayMs(item.attempts), + ); + item.updatedAt = checkedAt; + } + } + if (globalAdmissionFailure) { + const consecutiveFailures = priorDispatchConsecutiveFailures + 1; + const retryAt = + checkedAt + + exactReviewDispatchGlobalRetryDelayMs(consecutiveFailures, globalAdmissionFailure); + checkedState.dispatcher = { + state: "blocked", + reason: exactReviewDispatchDispatcherReason(globalAdmissionFailure.failureClass), + workflowState: preflight.workflowState, + checkedAt, + retryAt, + dispatchFailureStatus: globalAdmissionFailure.status, + dispatchFailureClass: globalAdmissionFailure.failureClass, + dispatchFailureAt: checkedAt, + dispatchFailureFingerprint: globalAdmissionFailure.fingerprint, + dispatchConsecutiveFailures: consecutiveFailures, + ...checkedBatchDispatcherFields, + }; + await this.writeState( + checkedState, + terminalCompleted ? { reviewCompleted: terminalCompleted } : undefined, + ); + await this.scheduleNext(checkedState, checkedAt); + return; + } + const dispatchable = admission.flatMap((candidate) => { + const item = checkedState.items[candidate.key]; + if (!item || item.revision !== candidate.revision || item.state !== "pending") return []; + if (exactReviewQueueIsPublication(item)) return [item]; + const live = liveStateByCandidate.get(candidate.key); + // A command acknowledgement needs the workflow's terminal completion + // path even when the target is already closed. Unprobed reviews wait for + // a later bounded admission pass instead of bypassing the live check. + return live?.state === "open" || + (live?.state === "terminal" && exactReviewQueueHasCommandContext(item)) + ? [item] + : []; + }); + + const hasReadyPendingReview = Object.values(checkedState.items).some( + (item) => + !exactReviewQueueIsPublication(item) && + item.state === "pending" && + item.nextAttemptAt <= checkedAt, ); - state.dispatcher = { + const shouldThrottleReviewAdmission = + liveCandidates.length === EXACT_REVIEW_ADMISSION_LIVE_CHECK_MAX_ITEMS || + (terminalCompleted > 0 && hasReadyPendingReview); + const nextReviewAdmissionAt = shouldThrottleReviewAdmission + ? checkedAt + EXACT_REVIEW_ADMISSION_INTERVAL_MS + : Number(checkedState.dispatcher?.reviewAdmissionNextAt || 0); + checkedState.dispatcher = { state: "active", workflowState: preflight.workflowState, - checkedAt: now, - ...batchDispatcherFields, + checkedAt, + ...(nextReviewAdmissionAt > checkedAt + ? { reviewAdmissionNextAt: nextReviewAdmissionAt } + : {}), + ...checkedBatchDispatcherFields, }; - for (const item of admitted) { + for (const item of dispatchable) { item.state = "dispatching"; item.leaseId = crypto.randomUUID(); item.leaseRevision = item.revision; item.leaseDecision = { ...item.decision }; item.leaseExpiresAt = - now + + checkedAt + (item.decision.sourceAction === EXACT_REVIEW_ARTIFACT_PUBLISH_SOURCE_ACTION ? exactReviewPublicationDispatchLeaseMs(this.env) : exactReviewDispatchLeaseMs(this.env)); item.claimedRunId = undefined; item.claimedRunAttempt = undefined; item.claimGeneration = undefined; - item.dispatchedAt = now; + item.dispatchedAt = checkedAt; item.claimedAt = undefined; - item.updatedAt = now; + item.updatedAt = checkedAt; } - await this.writeState(state); - if (!admitted.length) { - await this.scheduleNext(state, now); + await this.writeState( + checkedState, + terminalCompleted ? { reviewCompleted: terminalCompleted } : undefined, + ); + if (!dispatchable.length) { + await this.scheduleNext(checkedState, checkedAt); return; } @@ -2214,7 +2386,7 @@ export class ExactReviewQueue { attempted: boolean; }> = []; let globalFailure: ExactReviewDispatchFailure | null = null; - for (const item of admitted) { + for (const item of dispatchable) { if (globalFailure) { failures.push({ key: item.key, @@ -2278,6 +2450,7 @@ export class ExactReviewQueue { currentChanged = true; } if (globalFailure) { + const currentBatchDispatcherFields = batchDispatcherFieldsFor(current.dispatcher); const consecutiveFailures = priorDispatchConsecutiveFailures + 1; const retryAt = completedAt + exactReviewDispatchGlobalRetryDelayMs(consecutiveFailures, globalFailure); @@ -2292,7 +2465,7 @@ export class ExactReviewQueue { dispatchFailureAt: completedAt, dispatchFailureFingerprint: globalFailure.fingerprint, dispatchConsecutiveFailures: consecutiveFailures, - ...batchDispatcherFields, + ...currentBatchDispatcherFields, }; currentChanged = true; } @@ -5743,6 +5916,9 @@ export class ExactReviewQueue { exactReviewHeartbeatGraceMs(this.env), legacyExcludedItemKeys, batchOwnership.nextLeaseExpiresAt, + Number(state.dispatcher?.reviewAdmissionNextAt || 0) > now + ? Number(state.dispatcher?.reviewAdmissionNextAt) + : null, ); const reviewNext = this.nextReviewReconcileAtSync(now); const batchDeparture = exactReviewPublicationBatchDeparture( @@ -6920,6 +7096,10 @@ function exactReviewQueueIsPublication(item: Pick) { + return Boolean(item.decision.commandStatusMarker || item.decision.statusCommentId); +} + function exactReviewQueueLane(item: ExactReviewQueueItem) { return exactReviewQueueIsPublication(item) ? "publication" : "review"; } @@ -7271,6 +7451,9 @@ function exactReviewQueueStats( heartbeatGraceMs, excludedItemKeys, publicationBlockedUntil, + Number(state.dispatcher?.reviewAdmissionNextAt || 0) > now + ? Number(state.dispatcher?.reviewAdmissionNextAt) + : null, ); const lanes = { review: exactReviewQueueLaneStats( @@ -7451,6 +7634,7 @@ export function exactReviewQueueNextWakeAt( heartbeatGraceMs = DEFAULT_EXACT_REVIEW_HEARTBEAT_GRACE_MS, excludedItemKeys: ReadonlySet = new Set(), publicationBlockedUntil: number | null = null, + reviewAdmissionBlockedUntil: number | null = null, ) { const items = Object.values(state.items); if (!items.length) return null; @@ -7525,7 +7709,7 @@ export function exactReviewQueueNextWakeAt( return [Math.max(item.nextAttemptAt, blockedUntil)]; } const target = item.decision.targetRepo; - const blockedUntil = [ + const capacityBlockedUntil = [ ...(activeReviews.length >= capacity && activeReviewWakeAt.length ? [Math.min(...activeReviewWakeAt)] : []), @@ -7537,7 +7721,8 @@ export function exactReviewQueueNextWakeAt( return [ Math.max( item.nextAttemptAt, - blockedUntil.length ? Math.min(...blockedUntil) : item.nextAttemptAt, + reviewAdmissionBlockedUntil ?? item.nextAttemptAt, + capacityBlockedUntil.length ? Math.min(...capacityBlockedUntil) : item.nextAttemptAt, ), ]; } @@ -8386,6 +8571,51 @@ async function exactReviewSourceAuthorityLiveHead( .toLowerCase(); } +async function exactReviewTargetReadToken(env, targetRepo: string) { + const credentials = githubAppCredentials(env); + if (!credentials) throw new Error("github app is not configured"); + const appJwt = await signGithubAppJwt(credentials.issuer, credentials.privateKey); + const installationId = await githubAppInstallationId(appJwt, targetRepo); + return createGithubAppTokenFor({ + appJwt, + installationId, + label: targetRepo, + repositories: [repoName(targetRepo)], + permissions: { issues: "read", pull_requests: "read" }, + }); +} + +async function exactReviewTargetItemState(token: string, decision: ExactReviewDecision) { + try { + const item = await githubTokenJson({ + token, + path: `/repos/${decision.targetRepo}/issues/${decision.itemNumber}`, + method: "GET", + body: undefined, + errorLabel: "live review item state", + }); + const state = String(item.state || "").trim(); + if (state === "open") return "open" as const; + if (state === "closed") return "terminal" as const; + throw new Error("live review item state response missing state"); + } catch (error) { + if (error instanceof GitHubRequestError && error.status === 410) return "terminal" as const; + if (error instanceof GitHubRequestError && error.status === 404) { + // GitHub masks inaccessible private repositories as 404. Treat the item + // as missing only when this token can still read its repository. + await githubTokenJson({ + token, + path: `/repos/${decision.targetRepo}`, + method: "GET", + body: undefined, + errorLabel: "live review target repository", + }); + return "terminal" as const; + } + throw error; + } +} + export async function exactReviewActionsReadToken(env) { return exactReviewRepositoryToken(env, { actions: "read" }); } @@ -8610,12 +8840,14 @@ type ExactReviewDispatchFailure = { class GitHubRequestError extends Error { readonly status?: number; readonly timedOut: boolean; + readonly rateLimited: boolean; - constructor(message: string, status?: number, timedOut = false) { + constructor(message: string, status?: number, timedOut = false, rateLimited = false) { super(message); this.name = "GitHubRequestError"; this.status = status; this.timedOut = timedOut; + this.rateLimited = rateLimited; } } @@ -8626,10 +8858,10 @@ function exactReviewDispatchFailure(error: unknown): ExactReviewDispatchFailure ? "timeout" : status === 400 || status === 404 || status === 422 ? "permanent_rejection" - : status === 401 || status === 403 - ? "authentication" - : status === 429 - ? "rate_limit" + : requestError?.rateLimited || status === 429 + ? "rate_limit" + : status === 401 || status === 403 + ? "authentication" : status !== undefined && status >= 500 ? "github_outage" : "network"; @@ -8641,6 +8873,17 @@ function exactReviewDispatchFailure(error: unknown): ExactReviewDispatchFailure }; } +function exactReviewAdmissionFailure(error: unknown): ExactReviewDispatchFailure { + const failure = exactReviewDispatchFailure(error); + // This error arose while checking one specific target. A target installation + // may lack issue-read access even though other target installations are + // healthy, so a 403 must not hold the whole queue. + if (error instanceof GitHubRequestError && error.status === 403 && !error.rateLimited) { + return { ...failure, scope: "item" }; + } + return failure; +} + function exactReviewDispatchFailureFingerprint( failureClass: ExactReviewDispatchFailureClass, status?: number, @@ -8719,6 +8962,8 @@ async function githubTokenJson({ token, path, method = "GET", body, errorLabel } throw new GitHubRequestError( `${errorLabel || "GitHub"} ${response.status}${text ? `: ${text.slice(0, 240)}` : ""}`, response.status, + false, + githubResponseRateLimited(response, text), ); } if (response.status === 204) return {}; @@ -8890,21 +9135,52 @@ async function githubAppInstallationId(appJwt, repo) { async function githubAppJson(path, appJwt, options: GithubAppJsonOptions = {}) { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort("timeout"), GITHUB_TIMEOUT_MS); - const response = await fetch(`https://api.github.com${path}`, { - method: options.method || "GET", - signal: controller.signal, - headers: { - Accept: "application/vnd.github+json", - "Content-Type": "application/json", - "User-Agent": "openclaw-clawsweeper-status", - Authorization: `Bearer ${appJwt}`, - }, - body: options.body, - }).finally(() => clearTimeout(timeout)); - if (!response.ok) throw new Error(`${options.errorLabel || "GitHub App"} ${response.status}`); + let response: Response; + try { + response = await fetch(`https://api.github.com${path}`, { + method: options.method || "GET", + signal: controller.signal, + headers: { + Accept: "application/vnd.github+json", + "Content-Type": "application/json", + "User-Agent": "openclaw-clawsweeper-status", + Authorization: `Bearer ${appJwt}`, + }, + body: options.body, + }); + } catch (error) { + const timedOut = + controller.signal.aborted || + (error instanceof Error && (error.name === "AbortError" || error.message === "timeout")); + throw new GitHubRequestError( + `${options.errorLabel || "GitHub App"} ${timedOut ? "timed out" : "network failure"}`, + undefined, + timedOut, + ); + } finally { + clearTimeout(timeout); + } + if (!response.ok) { + const text = await response.text().catch(() => ""); + throw new GitHubRequestError( + `${options.errorLabel || "GitHub App"} ${response.status}${text ? `: ${text.slice(0, 240)}` : ""}`, + response.status, + false, + githubResponseRateLimited(response, text), + ); + } return response.json(); } +function githubResponseRateLimited(response: Response, text: string) { + return ( + response.status === 429 || + response.headers.has("retry-after") || + response.headers.get("x-ratelimit-remaining") === "0" || + /(?:secondary )?rate limit|api rate limit|abuse detection/i.test(text) + ); +} + async function signGithubAppJwt(issuer, privateKey) { const now = Math.floor(Date.now() / 1000); const header = base64UrlEncode(JSON.stringify({ alg: "RS256", typ: "JWT" })); diff --git a/src/repair/publish-event-result.ts b/src/repair/publish-event-result.ts index 964d3e3633..088901a542 100644 --- a/src/repair/publish-event-result.ts +++ b/src/repair/publish-event-result.ts @@ -115,6 +115,7 @@ try { } catch (error) { const retryableFailure = error instanceof GitCommandTimeoutError || error instanceof StatePublishContentionError; + const completionKind = retryableFailure ? "retryable_failure" : "permanent_failure"; const reasonCode = error instanceof GitCommandTimeoutError ? "github_transient" @@ -125,11 +126,15 @@ try { : error instanceof RecordTupleError ? "tuple_protocol_invalid" : "unknown_failure"; - writePublicationCompletionOutputs( - retryableFailure ? "retryable_failure" : "permanent_failure", - reasonCode, - errorFingerprint(error), - ); + const fingerprint = errorFingerprint(error); + if (options.batchMutationOutput) { + writeBatchMutationResult(options.batchMutationOutput, { + kind: completionKind, + reasonCode, + errorFingerprint: fingerprint, + }); + } + writePublicationCompletionOutputs(completionKind, reasonCode, fingerprint); throw error; } diff --git a/test/dashboard-worker.test.ts b/test/dashboard-worker.test.ts index fc6f334545..4cdd383e97 100644 --- a/test/dashboard-worker.test.ts +++ b/test/dashboard-worker.test.ts @@ -3924,9 +3924,12 @@ test("exact-review queue coalesces deliveries, dispatches a bound rollout snapsh await workflowCheckRelease; return jsonResponse({ state: workflowState }); } - if (url.pathname === "/repos/openclaw/clawsweeper/installation") { + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) { return jsonResponse({ id: 999 }); } + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) { + return jsonResponse({ state: "open" }); + } if (url.pathname === "/app/installations/999/access_tokens") { return jsonResponse({ token: "dispatch-token" }); } @@ -4095,6 +4098,625 @@ test("exact-review queue coalesces deliveries, dispatches a bound rollout snapsh } }); +test("exact-review queue resolves a closed item before dispatch", async () => { + const harness = createExactReviewAdmissionHarness(() => jsonResponse({ state: "closed" })); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("terminal-item", 597, "opened"))) + .status, + 202, + ); + + await harness.queue.alarm(); + + const stats = await ( + await harness.queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(harness.dispatched.length, 0); + assert.equal(stats.pending, 0); + assert.equal(stats.dispatching, 0); + assert.equal(stats.lanes.review.completed_total, 1); + const state = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items["openclaw/gogcli#597"], undefined); + } finally { + harness.restore(); + } +}); + +test("exact-review queue dispatches a closed command item to complete its acknowledgement", async () => { + const harness = createExactReviewAdmissionHarness(() => jsonResponse({ state: "closed" })); + const commandStatusMarker = + ""; + try { + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest("terminal-command-item", 597, "opened", "issue", undefined, { + commandStatusMarker, + statusCommentId: 9001, + }), + ) + ).status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 1); + const state = (await harness.storage.get("exact-review-queue")) as { + items: Record< + string, + { + state: string; + leaseDecision?: { commandStatusMarker?: string; statusCommentId?: number }; + } + >; + }; + assert.equal(state.items["openclaw/gogcli#597"]?.state, "dispatching"); + assert.equal( + state.items["openclaw/gogcli#597"]?.leaseDecision?.commandStatusMarker, + commandStatusMarker, + ); + assert.equal(state.items["openclaw/gogcli#597"]?.leaseDecision?.statusCommentId, 9001); + } finally { + harness.restore(); + } +}); + +test("exact-review queue limits live admission probes to one bounded pass", async () => { + const originalNow = Date.now; + let now = Date.parse("2026-07-24T23:45:00.000Z"); + Date.now = () => now; + let liveChecks = 0; + const harness = createExactReviewAdmissionHarness( + () => { + liveChecks += 1; + return jsonResponse({ state: "closed" }); + }, + { maxConcurrent: "16" }, + ); + try { + for (let index = 0; index < 10; index += 1) { + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest( + `bounded-admission-${index}`, + 700 + index, + "opened", + "issue", + ), + ) + ).status, + 202, + ); + } + + await harness.queue.alarm(); + + assert.equal(liveChecks, 4); + assert.equal(harness.dispatched.length, 0); + let stats = await ( + await harness.queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.pending, 6); + assert.equal(stats.lanes.review.completed_total, 4); + assert.equal(await harness.storage.getAlarm(), now + 5_000); + + await harness.queue.alarm(); + + assert.equal(liveChecks, 4); + now += 5_000; + await harness.queue.alarm(); + + assert.equal(liveChecks, 8); + stats = await ( + await harness.queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.pending, 2); + assert.equal(stats.lanes.review.completed_total, 8); + + now += 5_000; + await harness.queue.alarm(); + + assert.equal(liveChecks, 10); + stats = await ( + await harness.queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.pending, 0); + assert.equal(stats.lanes.review.completed_total, 10); + } finally { + Date.now = originalNow; + harness.restore(); + } +}); + +test("exact-review queue throttles partial terminal admission passes", async () => { + const originalNow = Date.now; + let now = Date.parse("2026-07-24T23:50:00.000Z"); + Date.now = () => now; + let liveChecks = 0; + const harness = createExactReviewAdmissionHarness( + () => { + liveChecks += 1; + return jsonResponse({ state: "closed" }); + }, + { maxConcurrent: "1" }, + ); + try { + for (let index = 0; index < 2; index += 1) { + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest(`partial-admission-${index}`, 800 + index, "opened"), + ) + ).status, + 202, + ); + } + + await harness.queue.alarm(); + + assert.equal(liveChecks, 1); + assert.equal(await harness.storage.getAlarm(), now + 5_000); + await harness.queue.alarm(); + assert.equal(liveChecks, 1); + + now += 5_000; + await harness.queue.alarm(); + assert.equal(liveChecks, 2); + } finally { + Date.now = originalNow; + harness.restore(); + } +}); + +test("exact-review queue resolves missing target responses before dispatch", async () => { + for (const status of [404, 410]) { + const harness = createExactReviewAdmissionHarness(() => new Response(null, { status })); + try { + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest(`missing-item-${status}`, 597, "opened"), + ) + ).status, + 202, + ); + + await harness.queue.alarm(); + + const stats = await ( + await harness.queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(harness.dispatched.length, 0); + assert.equal(stats.pending, 0); + assert.equal(stats.dispatching, 0); + } finally { + harness.restore(); + } + } +}); + +test("exact-review queue retains a 404 item when the target repository is inaccessible", async () => { + const harness = createExactReviewAdmissionHarness(() => new Response(null, { status: 404 }), { + targetRepository: () => new Response(null, { status: 404 }), + }); + try { + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest("inaccessible-target", 597, "opened"), + ) + ).status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 0); + const state = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items["openclaw/gogcli#597"]?.state, "pending"); + assert.equal(state.items["openclaw/gogcli#597"]?.attempts, 1); + assert.equal(state.items["openclaw/gogcli#597"]?.reviewFailureAttempts, 1); + } finally { + harness.restore(); + } +}); + +test("exact-review queue dispatches an item that remains open", async () => { + const harness = createExactReviewAdmissionHarness(() => jsonResponse({ state: "open" })); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("open-item", 597, "opened"))).status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 1); + const stats = await ( + await harness.queue.fetch(new Request("https://clawsweeper-exact-review-queue/stats")) + ).json(); + assert.equal(stats.dispatching, 1); + } finally { + harness.restore(); + } +}); + +test("exact-review queue bounds item-specific terminal-state check failures", async () => { + const harness = createExactReviewAdmissionHarness( + () => new Response(JSON.stringify({ message: "unprocessable" }), { status: 422 }), + ); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("unavailable-item", 597, "opened"))) + .status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 0); + const state = (await harness.storage.get("exact-review-queue")) as { + items: Record< + string, + { state: string; attempts: number; reviewFailureAttempts?: number; nextAttemptAt: number } + >; + }; + const item = state.items["openclaw/gogcli#597"]; + assert.equal(item?.state, "pending"); + assert.equal(item?.attempts, 1); + assert.equal(item?.reviewFailureAttempts, 1); + assert.ok((item?.nextAttemptAt || 0) > Date.now()); + } finally { + harness.restore(); + } +}); + +test("exact-review queue parks an item after repeated item-specific target-state failures", async () => { + const harness = createExactReviewAdmissionHarness( + () => new Response(JSON.stringify({ message: "unprocessable" }), { status: 422 }), + ); + try { + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest("repeated-unavailable", 597, "opened"), + ) + ).status, + 202, + ); + + for (let attempt = 1; attempt <= 8; attempt += 1) { + await harness.queue.alarm(); + const state = (await harness.storage.get("exact-review-queue")) as { + items: Record< + string, + { + state: string; + attempts: number; + reviewFailureAttempts?: number; + nextAttemptAt: number; + parkedReason?: string; + } + >; + }; + const item = state.items["openclaw/gogcli#597"]; + assert.equal(item?.attempts, attempt); + assert.equal(item?.reviewFailureAttempts, attempt); + if (attempt < 8) { + assert.equal(item?.state, "pending"); + item.nextAttemptAt = Date.now() - 1; + await harness.storage.put("exact-review-queue", state); + } else { + assert.equal(item?.state, "parked"); + assert.equal(item?.parkedReason, "review_retry_exhausted"); + } + } + assert.equal(harness.dispatched.length, 0); + } finally { + harness.restore(); + } +}); + +test("exact-review queue globally backs off admission GitHub outages without charging item attempts", async () => { + const harness = createExactReviewAdmissionHarness( + () => new Response(JSON.stringify({ message: "unavailable" }), { status: 503 }), + ); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("admission-outage", 597, "opened"))) + .status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 0); + const state = (await harness.storage.get("exact-review-queue")) as { + dispatcher: { + state: string; + reason: string; + dispatchFailureStatus: number; + dispatchConsecutiveFailures: number; + }; + items: Record; + }; + assert.equal(state.dispatcher.state, "blocked"); + assert.equal(state.dispatcher.reason, "dispatch_github_outage"); + assert.equal(state.dispatcher.dispatchFailureStatus, 503); + assert.equal(state.dispatcher.dispatchConsecutiveFailures, 1); + assert.equal(state.items["openclaw/gogcli#597"]?.state, "pending"); + assert.equal(state.items["openclaw/gogcli#597"]?.attempts, 0); + assert.equal(state.items["openclaw/gogcli#597"]?.reviewFailureAttempts, undefined); + } finally { + harness.restore(); + } +}); + +test("exact-review queue globally backs off admission 403 rate limits without charging item attempts", async () => { + const harness = createExactReviewAdmissionHarness( + () => + new Response(JSON.stringify({ message: "You have exceeded a secondary rate limit." }), { + status: 403, + headers: { "x-ratelimit-remaining": "0" }, + }), + ); + try { + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest("admission-rate-limit", 597, "opened"), + ) + ).status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 0); + const state = (await harness.storage.get("exact-review-queue")) as { + dispatcher: { + state: string; + reason: string; + dispatchFailureStatus: number; + dispatchFailureClass: string; + }; + items: Record; + }; + assert.equal(state.dispatcher.state, "blocked"); + assert.equal(state.dispatcher.reason, "dispatch_rate_limit"); + assert.equal(state.dispatcher.dispatchFailureStatus, 403); + assert.equal(state.dispatcher.dispatchFailureClass, "rate_limit"); + assert.equal(state.items["openclaw/gogcli#597"]?.state, "pending"); + assert.equal(state.items["openclaw/gogcli#597"]?.attempts, 0); + assert.equal(state.items["openclaw/gogcli#597"]?.reviewFailureAttempts, undefined); + } finally { + harness.restore(); + } +}); + +test("exact-review queue keeps healthy targets moving when one target App access fails", async () => { + const harness = createExactReviewAdmissionHarness(() => jsonResponse({ state: "open" }), { + maxConcurrent: "2", + targetInstallation: (targetRepo) => + targetRepo === "openclaw/gogcli" + ? new Response(JSON.stringify({ message: "not installed" }), { status: 404 }) + : jsonResponse({ id: 999 }), + }); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("target-app-failure", 597, "opened"))) + .status, + 202, + ); + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest( + "healthy-target", + 598, + "opened", + "issue", + "openclaw/openclaw", + ), + ) + ).status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 1); + assert.equal(harness.dispatched[0]?.client_payload?.target_repo, "openclaw/openclaw"); + const state = (await harness.storage.get("exact-review-queue")) as { + dispatcher: { state: string }; + items: Record; + }; + assert.equal(state.dispatcher.state, "active"); + assert.equal(state.items["openclaw/gogcli#597"]?.state, "pending"); + assert.equal(state.items["openclaw/gogcli#597"]?.attempts, 1); + assert.equal(state.items["openclaw/gogcli#597"]?.reviewFailureAttempts, 1); + assert.equal(state.items["openclaw/openclaw#598"]?.state, "dispatching"); + } finally { + harness.restore(); + } +}); + +test("exact-review queue keeps healthy targets moving when one target item read is forbidden", async () => { + const harness = createExactReviewAdmissionHarness(() => jsonResponse({ state: "open" }), { + maxConcurrent: "2", + targetItem: (targetRepo) => + targetRepo === "openclaw/gogcli" + ? new Response(JSON.stringify({ message: "forbidden" }), { status: 403 }) + : jsonResponse({ state: "open" }), + }); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("forbidden-target", 597, "opened"))) + .status, + 202, + ); + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest( + "healthy-target-after-forbidden", + 598, + "opened", + "issue", + "openclaw/openclaw", + ), + ) + ).status, + 202, + ); + + await harness.queue.alarm(); + + assert.equal(harness.dispatched.length, 1); + assert.equal(harness.dispatched[0]?.client_payload?.target_repo, "openclaw/openclaw"); + const state = (await harness.storage.get("exact-review-queue")) as { + dispatcher: { state: string }; + items: Record; + }; + assert.equal(state.dispatcher.state, "active"); + assert.equal(state.items["openclaw/gogcli#597"]?.state, "pending"); + assert.equal(state.items["openclaw/gogcli#597"]?.attempts, 1); + assert.equal(state.items["openclaw/gogcli#597"]?.reviewFailureAttempts, 1); + assert.equal(state.items["openclaw/openclaw#598"]?.state, "dispatching"); + } finally { + harness.restore(); + } +}); + +test("exact-review admission does not restore a publication batch claim reservation", async () => { + let releaseLookup!: () => void; + let signalLookupStarted!: () => void; + const lookupStarted = new Promise((resolve) => { + signalLookupStarted = resolve; + }); + const lookupRelease = new Promise((resolve) => { + releaseLookup = resolve; + }); + const harness = createExactReviewAdmissionHarness( + async () => { + signalLookupStarted(); + await lookupRelease; + return jsonResponse({ state: "open" }); + }, + { + publicationBatching: true, + dispatch: () => new Response(JSON.stringify({ message: "unavailable" }), { status: 503 }), + }, + ); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("batch-claim-race", 597, "opened"))) + .status, + 202, + ); + assert.equal( + ( + await harness.queue.fetch( + buildExactReviewQueueRequest( + "batch-claim-publication", + 598, + "exact_review_artifact_publish", + "issue", + "openclaw/openclaw", + exactReviewPublicationOverrides(598, "5980", "opened", 1, "openclaw/openclaw"), + ), + ) + ).status, + 202, + ); + const reserved = (await harness.storage.get("exact-review-queue")) as { + dispatcher?: Record; + }; + reserved.dispatcher = { + state: "active", + checkedAt: Date.now(), + publicationBatchDispatchPendingUntil: Date.now() + 5 * 60_000, + }; + await harness.storage.put("exact-review-queue", reserved); + + const alarm = harness.queue.alarm(); + await lookupStarted; + const claim = await harness.queue.fetch( + new Request("https://clawsweeper-exact-review-queue/publication-batches/claim", { + method: "POST", + body: JSON.stringify({ + claim_id: "admission-race-batch", + lease_owner: "admission-race-owner", + max_items: 1, + }), + }), + ); + assert.equal((await claim.json()).claimed, true); + const afterClaim = (await harness.storage.get("exact-review-queue")) as { + dispatcher?: { publicationBatchDispatchPendingUntil?: number }; + }; + assert.equal(afterClaim.dispatcher?.publicationBatchDispatchPendingUntil, undefined); + + releaseLookup(); + await alarm; + + const afterAlarm = (await harness.storage.get("exact-review-queue")) as { + dispatcher?: { publicationBatchDispatchPendingUntil?: number }; + }; + assert.equal(afterAlarm.dispatcher?.publicationBatchDispatchPendingUntil, undefined); + } finally { + harness.restore(); + } +}); + +test("exact-review terminal admission does not remove a newer queue revision", async () => { + let releaseLookup!: () => void; + let signalLookupStarted!: () => void; + const lookupStarted = new Promise((resolve) => { + signalLookupStarted = resolve; + }); + const lookupRelease = new Promise((resolve) => { + releaseLookup = resolve; + }); + const harness = createExactReviewAdmissionHarness(async () => { + signalLookupStarted(); + await lookupRelease; + return jsonResponse({ state: "closed" }); + }); + try { + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("stale-terminal", 597, "opened"))) + .status, + 202, + ); + const alarm = harness.queue.alarm(); + await lookupStarted; + assert.equal( + (await harness.queue.fetch(buildExactReviewQueueRequest("newer-revision", 597, "edited"))) + .status, + 202, + ); + releaseLookup(); + await alarm; + + assert.equal(harness.dispatched.length, 0); + const state = (await harness.storage.get("exact-review-queue")) as { + items: Record; + }; + assert.equal(state.items["openclaw/gogcli#597"]?.state, "pending"); + assert.equal(state.items["openclaw/gogcli#597"]?.revision, 2); + } finally { + harness.restore(); + } +}); + test("exact-review queue upgrades flow metrics without losing publication completions", async () => { const storage = new MemoryDurableStorage(); storage.sql.exec( @@ -4990,9 +5612,12 @@ test("exact-review queue admits at most one active item per target repository", if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") { return jsonResponse({ state: "active" }); } - if (url.pathname === "/repos/openclaw/clawsweeper/installation") { + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) { return jsonResponse({ id: 999 }); } + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) { + return jsonResponse({ state: "open" }); + } if (url.pathname === "/app/installations/999/access_tokens") { return jsonResponse({ token: "dispatch-token" }); } @@ -5063,8 +5688,10 @@ test("exact-review review retries stop at the attempt ceiling and park the item" if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") { return jsonResponse({ state: "active" }); } - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") { return jsonResponse({ token: "dispatch-token" }); } @@ -5302,8 +5929,10 @@ test("exact-review queue can use the global capacity for one target", async () = const url = new URL(String(input)); if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") return jsonResponse({ state: "active" }); - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse({ token: "dispatch-token" }); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") { @@ -5359,8 +5988,10 @@ test("exact-review queue keeps publication artifacts durable outside review capa const url = new URL(String(input)); if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") return jsonResponse({ state: "active" }); - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse(Object.fromEntries([["token", "t"]])); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") { @@ -5634,8 +6265,10 @@ test("exact-review queue wakes while target capacity remains", async () => { const url = new URL(String(input)); if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") return jsonResponse({ state: "active" }); - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse({ token: "dispatch-token" }); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") { @@ -5683,8 +6316,10 @@ test("exact-review queue defers retained backlog until a paused dispatcher retry const url = new URL(String(input)); if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") return jsonResponse({ state: "active" }); - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse({ token: "dispatch-token" }); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") { @@ -6184,8 +6819,10 @@ test("exact-review queue retries dispatch failures and reclaims an unclaimed lea } return jsonResponse({ state: "active" }); } - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse({ token: "dispatch-token" }); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") { @@ -6311,8 +6948,10 @@ test("exact-review queue parks permanent dispatch rejection and explicit command const url = new URL(String(input)); if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") return jsonResponse({ state: "active" }); - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse({ token: "dispatch-token" }); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") @@ -6404,8 +7043,10 @@ test("exact-review queue globally backs off GitHub outages without charging item const url = new URL(String(input)); if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") return jsonResponse({ state: "active" }); - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse({ token: "dispatch-token" }); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") @@ -6465,8 +7106,10 @@ test("exact-review queue preserves a claimed lease after an ambiguous dispatch f const url = new URL(String(input)); if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") return jsonResponse({ state: "active" }); - if (url.pathname === "/repos/openclaw/clawsweeper/installation") + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/installation$/.test(url.pathname)) return jsonResponse({ id: 999 }); + if (/^\/repos\/openclaw\/(?:clawsweeper|gogcli|openclaw)\/issues\/\d+$/.test(url.pathname)) + return jsonResponse({ state: "open" }); if (url.pathname === "/app/installations/999/access_tokens") return jsonResponse({ token: "t" }); if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") { @@ -13868,6 +14511,81 @@ function signedStateAppendRequest(path: string, payload: unknown, secret: string }); } +function createExactReviewAdmissionHarness( + liveItem: () => Response | Promise, + options: { + maxConcurrent?: string; + publicationBatching?: boolean; + targetInstallation?: (targetRepo: string) => Response | Promise; + targetRepository?: (targetRepo: string) => Response | Promise; + targetItem?: (targetRepo: string) => Response | Promise; + dispatch?: () => Response | Promise; + } = {}, +) { + const originalFetch = globalThis.fetch; + const storage = new MemoryDurableStorage(); + const dispatched: Record[] = []; + const { privateKey } = generateKeyPairSync("rsa", { + modulusLength: 2048, + privateKeyEncoding: { type: "pkcs8", format: "pem" }, + publicKeyEncoding: { type: "spki", format: "pem" }, + }); + globalThis.fetch = async (input, init) => { + const url = new URL(String(input)); + if (url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/sweep.yml") { + return jsonResponse({ state: "active" }); + } + const installation = url.pathname.match(/^\/repos\/(openclaw\/[^/]+)\/installation$/); + if (installation) { + return options.targetInstallation?.(installation[1]) ?? jsonResponse({ id: 999 }); + } + const repository = url.pathname.match(/^\/repos\/(openclaw\/[^/]+)$/); + if (repository) { + return ( + options.targetRepository?.(repository[1]) ?? jsonResponse({ full_name: repository[1] }) + ); + } + if (url.pathname === "/app/installations/999/access_tokens") { + return jsonResponse({ token: "queue-token" }); + } + const targetItem = url.pathname.match( + /^\/repos\/(openclaw\/(?:gogcli|openclaw))\/issues\/\d+$/, + ); + if (targetItem) { + return options.targetItem?.(targetItem[1]) ?? liveItem(); + } + if (url.pathname === "/repos/openclaw/clawsweeper/dispatches") { + dispatched.push(JSON.parse(String(init?.body))); + return options.dispatch?.() ?? new Response(null, { status: 204 }); + } + throw new Error(`unexpected fetch ${url}`); + }; + const queue = new ExactReviewQueue( + { storage }, + { + CLAWSWEEPER_APP_CLIENT_ID: "Iv23test", + CLAWSWEEPER_APP_PRIVATE_KEY: privateKey, + EXACT_REVIEW_DISPATCH_DEBOUNCE_MS: "0", + EXACT_REVIEW_QUEUE_MAX_CONCURRENT: options.maxConcurrent ?? "1", + ...(options.publicationBatching + ? { + EXACT_REVIEW_PUBLICATION_BATCHING_ENABLED: "1", + EXACT_REVIEW_PUBLICATION_BATCH_SIZE: "1", + EXACT_REVIEW_PUBLICATION_BATCH_WAIT_MS: "300000", + } + : {}), + }, + ); + return { + queue, + storage, + dispatched, + restore: () => { + globalThis.fetch = originalFetch; + }, + }; +} + function buildExactReviewQueueRequest( deliveryId: string, itemNumber: number, diff --git a/test/exact-review-publication-batches.test.ts b/test/exact-review-publication-batches.test.ts index 1958e8f96e..686f226afc 100644 --- a/test/exact-review-publication-batches.test.ts +++ b/test/exact-review-publication-batches.test.ts @@ -1249,6 +1249,11 @@ test("rollout dispatches one full batch workflow without admitting legacy publis headers: { "content-type": "application/json" }, }); } + if (url.pathname === "/repos/openclaw/openclaw/installation") { + return new Response(JSON.stringify({ id: 1000 }), { + headers: { "content-type": "application/json" }, + }); + } if (url.pathname === "/app/installations/999/access_tokens") { const body = JSON.parse(String(init?.body)); assert.deepEqual(body.permissions, { actions: "write", contents: "write" }); @@ -1256,6 +1261,22 @@ test("rollout dispatches one full batch workflow without admitting legacy publis headers: { "content-type": "application/json" }, }); } + if (url.pathname === "/app/installations/1000/access_tokens") { + const body = JSON.parse(String(init?.body)); + assert.deepEqual(body, { + repository_names: ["openclaw"], + permissions: { issues: "read", pull_requests: "read" }, + }); + return new Response(JSON.stringify({ token: "target-token" }), { + headers: { "content-type": "application/json" }, + }); + } + if (url.pathname === "/repos/openclaw/openclaw/issues/114") { + assert.equal(new Headers(init?.headers).get("authorization"), "Bearer target-token"); + return new Response(JSON.stringify({ state: "open" }), { + headers: { "content-type": "application/json" }, + }); + } if ( url.pathname === "/repos/openclaw/clawsweeper/actions/workflows/exact-review-batch-publish.yml/dispatches" diff --git a/test/repair/exact-review-batch-workflow.test.ts b/test/repair/exact-review-batch-workflow.test.ts index 526f582c51..4838924f4a 100644 --- a/test/repair/exact-review-batch-workflow.test.ts +++ b/test/repair/exact-review-batch-workflow.test.ts @@ -54,6 +54,10 @@ test("batch workflow signs queue ownership, isolates item failures, and commits assert.match(prepareSource, /"retryable_failure", "artifact_unavailable"/); assert.match(prepareSource, /"permanent_failure", "tuple_protocol_invalid"/); assert.match(prepareSource, /EXACT_REVIEW_BATCH_MUTATION_OUTPUT/); + assert.match( + publisherSource, + /if \(options\.batchMutationOutput\)[\s\S]*?writeBatchMutationResult\(options\.batchMutationOutput, \{[\s\S]*?kind: completionKind,[\s\S]*?reasonCode,/, + ); // Keep the fixture from looking like an embedded credential while still // proving that artifact downloads use the owner-scoped repository token. const ghToken = ["GH", "TOKEN"].join("_"); diff --git a/test/sweep-workflow.test.ts b/test/sweep-workflow.test.ts index 421eff97cb..0703e12280 100644 --- a/test/sweep-workflow.test.ts +++ b/test/sweep-workflow.test.ts @@ -668,7 +668,11 @@ test("exact event review hands immutable artifacts to the queue-bounded publishe assert.match(publisherSource, /error instanceof GitCommandTimeoutError/); assert.match( publisherSource, - /writePublicationCompletionOutputs\(\s*retryableFailure \? "retryable_failure" : "permanent_failure",\s*reasonCode,\s*errorFingerprint\(error\),\s*\);/, + /const completionKind = retryableFailure \? "retryable_failure" : "permanent_failure"/, + ); + assert.match( + publisherSource, + /writePublicationCompletionOutputs\(completionKind, reasonCode, fingerprint\);/, ); assert.doesNotMatch(publisherSource, /retryableFailure \? "github_transient" : undefined/); assert.match(publishComplete.run ?? "", /"state_contention"/);