From 92a4625327e40c93b84c5d82263335bad9638559 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Wed, 15 Jul 2026 10:37:02 +0100 Subject: [PATCH 1/6] fix(flue-review): recover interrupted workflows --- infra/flue-review/.flue/app.ts | 24 +- infra/flue-review/.flue/cloudflare.ts | 219 ++++++++++++-- infra/flue-review/.flue/lib/github.ts | 86 +++--- .../flue-review/.flue/lib/review-watchdog.ts | 31 +- .../.flue/lib/workflow-admission.ts | 39 +++ infra/flue-review/.flue/workflows/review.ts | 9 +- infra/flue-review/test/github-checks.test.ts | 89 ++++++ .../test/review-watchdog-do.test.ts | 271 +++++++++++++++++- .../flue-review/test/review-watchdog.test.ts | 12 +- infra/flue-review/wrangler.jsonc | 8 + 10 files changed, 701 insertions(+), 87 deletions(-) create mode 100644 infra/flue-review/.flue/lib/workflow-admission.ts diff --git a/infra/flue-review/.flue/app.ts b/infra/flue-review/.flue/app.ts index 74eacb099..d99dd10d3 100644 --- a/infra/flue-review/.flue/app.ts +++ b/infra/flue-review/.flue/app.ts @@ -9,7 +9,6 @@ // inside the workflow's Durable Object, which is not bound by that budget. import { getRun, listRuns } from "@flue/runtime"; -import { flue } from "@flue/runtime/routing"; import { Hono } from "hono"; import { @@ -25,8 +24,7 @@ import { gatePullRequestEvent, getWebhookDeliveryId, } from "./lib/webhook.js"; - -const flueApp = flue(); +import { admitReviewWorkflow } from "./lib/workflow-admission.js"; // Extract a short, displayable message from an unknown run error without // risking an "[object Object]" stringification. @@ -150,6 +148,7 @@ app.post("/webhook/github", async (c) => { repo: decision.pr.repo, prNumber: decision.pr.prNumber, headSha: decision.pr.headSha, + workflowInput: decision.pr, stage: "admitted", lastProgressAt: Date.now(), }; @@ -243,17 +242,8 @@ app.post("/webhook/github", async (c) => { // block the webhook on the (minutes-long) review. let admit: Response; try { - admit = await flueApp.fetch( - new Request("https://flue.internal/workflows/review", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - ...decision.pr, - attemptId, - deliveryId, - checkRunId, - }), - }), + admit = await admitReviewWorkflow( + { ...decision.pr, attemptId, expectedRunId: attemptId, deliveryId, checkRunId }, c.env, c.executionCtx, ); @@ -270,7 +260,7 @@ app.post("/webhook/github", async (c) => { }), ); await watchdog - .finish(attemptId, { + .finish(attemptId, attemptId, { conclusion: "failure", summary: "The review could not be admitted. Reapply the `bot:review` label to retry.", }) @@ -292,7 +282,7 @@ app.post("/webhook/github", async (c) => { }), ); await watchdog - .finish(attemptId, { + .finish(attemptId, attemptId, { conclusion: "failure", summary: "The review could not be admitted. Reapply the `bot:review` label to retry.", }) @@ -327,7 +317,7 @@ app.post("/webhook/github", async (c) => { }), ); } else { - await watchdog.identify(attemptId, runId).catch((error) => { + await watchdog.identify(attemptId, attemptId, runId).catch((error) => { console.error( JSON.stringify({ message: "review watchdog correlation failed", diff --git a/infra/flue-review/.flue/cloudflare.ts b/infra/flue-review/.flue/cloudflare.ts index 1e78dff42..e53344235 100644 --- a/infra/flue-review/.flue/cloudflare.ts +++ b/infra/flue-review/.flue/cloudflare.ts @@ -1,20 +1,25 @@ +import { getRun } from "@flue/runtime"; import { DurableObject } from "cloudflare:workers"; import { completeReviewCheck, mintInstallationToken, readAppCreds } from "./lib/github.js"; import { isReviewAttemptStale, + reviewStaleAfter, REVIEW_SETUP_LEASE_MS, - REVIEW_STALE_AFTER_MS, type ReviewAttempt, type ReviewStage, type ReviewTerminal, } from "./lib/review-watchdog.js"; +import { admitReviewWorkflow } from "./lib/workflow-admission.js"; const ATTEMPT_KEY = "attempt"; const TERMINAL_RETRY_BASE_MS = 60_000; const TERMINAL_RETRY_MAX_MS = 60 * 60_000; const TERMINAL_RETRY_LIMIT = 10; const TERMINAL_RETENTION_MS = 7 * 24 * 60 * 60_000; +const WORKFLOW_RETRY_LIMIT = 2; +const WORKFLOW_STATUS_RETRY_MS = 60_000; +const WORKFLOW_ACTIVE_STALE_LIMIT_MS = 30 * 60_000; class TerminalConfigurationError extends Error {} @@ -51,7 +56,7 @@ export class ReviewWatchdog extends DurableObject { setupLeaseExpiresAt: Date.now() + REVIEW_SETUP_LEASE_MS, }; await this.ctx.storage.put(ATTEMPT_KEY, reserved); - await this.ctx.storage.setAlarm(attempt.lastProgressAt + REVIEW_STALE_AFTER_MS); + await this.ctx.storage.setAlarm(attempt.lastProgressAt + reviewStaleAfter(attempt.stage)); return { status: "acquired", attempt: reserved }; } @@ -66,13 +71,27 @@ export class ReviewWatchdog extends DurableObject { throw new Error("Review attempt was not reserved"); } await this.ctx.storage.put(ATTEMPT_KEY, { ...reserved, ...attempt }); - await this.ctx.storage.setAlarm(attempt.lastProgressAt + REVIEW_STALE_AFTER_MS); + await this.ctx.storage.setAlarm(attempt.lastProgressAt + reviewStaleAfter(attempt.stage)); } - async identify(attemptId: string, runId: string): Promise { + async identify(attemptId: string, expectedRunId: string, runId: string): Promise { const attempt = await this.ctx.storage.get(ATTEMPT_KEY); - if (!attempt || attempt.attemptId !== attemptId || attempt.terminal) return false; - await this.ctx.storage.put(ATTEMPT_KEY, { ...attempt, runId }); + if (attempt?.attemptId === attemptId && !attempt.terminal && attempt.runId === runId) { + return true; + } + if ( + !attempt || + attempt.attemptId !== attemptId || + attempt.runId !== expectedRunId || + attempt.terminal + ) { + return false; + } + await this.ctx.storage.put(ATTEMPT_KEY, { + ...attempt, + runId, + workflowActiveStaleSince: undefined, + }); return true; } @@ -91,12 +110,24 @@ export class ReviewWatchdog extends DurableObject { return true; } - async heartbeat(attemptId: string, stage: ReviewStage): Promise { + async heartbeat(attemptId: string, runId: string, stage: ReviewStage): Promise { const attempt = await this.ctx.storage.get(ATTEMPT_KEY); - if (!attempt || attempt.attemptId !== attemptId || attempt.terminal) return false; + if ( + !attempt || + attempt.attemptId !== attemptId || + attempt.runId !== runId || + attempt.terminal + ) { + return false; + } const lastProgressAt = Date.now(); - await this.ctx.storage.put(ATTEMPT_KEY, { ...attempt, stage, lastProgressAt }); - await this.ctx.storage.setAlarm(lastProgressAt + REVIEW_STALE_AFTER_MS); + await this.ctx.storage.put(ATTEMPT_KEY, { + ...attempt, + stage, + lastProgressAt, + workflowActiveStaleSince: undefined, + }); + await this.ctx.storage.setAlarm(lastProgressAt + reviewStaleAfter(stage)); return true; } @@ -107,9 +138,16 @@ export class ReviewWatchdog extends DurableObject { await this.ctx.storage.deleteAlarm(); } - async finish(attemptId: string, terminal: ReviewTerminal): Promise { + async finish(attemptId: string, runId: string, terminal: ReviewTerminal): Promise { const attempt = await this.ctx.storage.get(ATTEMPT_KEY); - if (!attempt || attempt.attemptId !== attemptId || attempt.terminal) return false; + if ( + !attempt || + attempt.attemptId !== attemptId || + attempt.runId !== runId || + attempt.terminal + ) { + return false; + } const terminalAttempt = { ...attempt, terminal }; await this.ctx.storage.put(ATTEMPT_KEY, terminalAttempt); try { @@ -120,6 +158,101 @@ export class ReviewWatchdog extends DurableObject { return true; } + private async retryWorkflow( + attempt: ReviewAttempt, + ): Promise<"admitted" | "pending" | "exhausted"> { + const workflowRetryCount = (attempt.workflowRetryCount ?? 0) + 1; + if ( + !attempt.workflowInput || + attempt.checkRunId === undefined || + workflowRetryCount > WORKFLOW_RETRY_LIMIT + ) { + return "exhausted"; + } + + const lastProgressAt = Date.now(); + const retryRunId = `${attempt.attemptId}:retry:${workflowRetryCount}`; + const retrying = { + ...attempt, + runId: retryRunId, + stage: "admitted" as const, + lastProgressAt, + workflowRetryCount, + workflowActiveStaleSince: undefined, + }; + await this.ctx.storage.put(ATTEMPT_KEY, retrying); + await this.ctx.storage.setAlarm(lastProgressAt + reviewStaleAfter("admitted")); + + try { + // Flue needs a fetch-style context to durably admit the replacement run. + // DurableObjectState supplies the waitUntil primitive used by this path. + const executionCtx = { + waitUntil: (promise: Promise) => this.ctx.waitUntil(promise), + passThroughOnException: () => undefined, + }; + const response = await admitReviewWorkflow( + { + ...attempt.workflowInput, + attemptId: attempt.attemptId, + expectedRunId: retryRunId, + deliveryId: attempt.deliveryId, + checkRunId: attempt.checkRunId, + }, + this.env, + executionCtx, + ); + if (!response.ok) throw new Error(`workflow admission returned ${response.status}`); + const admission: { runId?: string } = await response + .json<{ runId?: string }>() + .catch(() => ({})); + if (admission.runId) { + await this.identify(attempt.attemptId, retryRunId, admission.runId); + } + console.log( + JSON.stringify({ + message: "stale review workflow re-admitted", + attemptId: attempt.attemptId, + previousRunId: attempt.runId, + runId: admission.runId, + workflowRetryCount, + }), + ); + return "admitted"; + } catch (error) { + console.error( + JSON.stringify({ + message: "stale review workflow re-admission failed", + error: error instanceof Error ? error.message : String(error), + attemptId: attempt.attemptId, + previousRunId: attempt.runId, + workflowRetryCount, + }), + ); + return "pending"; + } + } + + private async workflowStatus( + attempt: ReviewAttempt, + ): Promise<"active" | "completed" | "recoverable" | "unavailable"> { + try { + const run = await getRun(attempt.runId); + if (!run || run.status === "errored") return "recoverable"; + return run.status; + } catch (error) { + console.error( + JSON.stringify({ + message: "review workflow status inspection failed", + error: error instanceof Error ? error.message : String(error), + attemptId: attempt.attemptId, + runId: attempt.runId, + }), + ); + await this.ctx.storage.setAlarm(Date.now() + WORKFLOW_STATUS_RETRY_MS); + return "unavailable"; + } + } + private async flushTerminal( attempt: ReviewAttempt & { terminal: ReviewTerminal }, ): Promise { @@ -206,8 +339,8 @@ export class ReviewWatchdog extends DurableObject { } return; } - if (!isReviewAttemptStale(attempt.lastProgressAt)) { - await this.ctx.storage.setAlarm(attempt.lastProgressAt + REVIEW_STALE_AFTER_MS); + if (!isReviewAttemptStale(attempt.lastProgressAt, Date.now(), attempt.stage)) { + await this.ctx.storage.setAlarm(attempt.lastProgressAt + reviewStaleAfter(attempt.stage)); return; } if (attempt.checkRunId === undefined) { @@ -215,21 +348,65 @@ export class ReviewWatchdog extends DurableObject { await this.ctx.storage.deleteAlarm(); return; } + const workflowStatus = await this.workflowStatus(attempt); + if (workflowStatus === "unavailable") return; + const currentAttempt = await this.ctx.storage.get(ATTEMPT_KEY); + if ( + !currentAttempt || + currentAttempt.terminal || + currentAttempt.runId !== attempt.runId || + currentAttempt.stage !== attempt.stage || + currentAttempt.lastProgressAt !== attempt.lastProgressAt + ) { + return; + } + if (workflowStatus === "active") { + const now = Date.now(); + const workflowActiveStaleSince = currentAttempt.workflowActiveStaleSince ?? now; + if (now - workflowActiveStaleSince < WORKFLOW_ACTIVE_STALE_LIMIT_MS) { + await this.ctx.storage.put(ATTEMPT_KEY, { + ...currentAttempt, + workflowActiveStaleSince, + }); + await this.ctx.storage.setAlarm(now + WORKFLOW_STATUS_RETRY_MS); + return; + } + } + if (workflowStatus === "completed") { + const terminalAttempt = { + ...currentAttempt, + terminal: { + conclusion: "success", + summary: "The automated review completed successfully.", + } satisfies ReviewTerminal, + }; + await this.ctx.storage.put(ATTEMPT_KEY, terminalAttempt); + try { + await this.flushTerminal(terminalAttempt); + } catch (error) { + await this.scheduleTerminalRetry(terminalAttempt, error); + } + return; + } + if (workflowStatus === "recoverable") { + const retryResult = await this.retryWorkflow(currentAttempt); + if (retryResult !== "exhausted") return; + } const terminal: ReviewTerminal = { conclusion: "timed_out", - summary: `The review stopped reporting progress while in the \`${attempt.stage}\` stage. Reapply the \`bot:review\` label to retry.`, + summary: `The review stopped reporting progress while in the \`${currentAttempt.stage}\` stage. Reapply the \`bot:review\` label to retry.`, }; - const terminalAttempt = { ...attempt, terminal }; + const terminalAttempt = { ...currentAttempt, terminal }; await this.ctx.storage.put(ATTEMPT_KEY, terminalAttempt); console.error( JSON.stringify({ message: "review watchdog timed out stale attempt", - attemptId: attempt.attemptId, - runId: attempt.runId, - deliveryId: attempt.deliveryId, - prNumber: attempt.prNumber, - stage: attempt.stage, + attemptId: currentAttempt.attemptId, + runId: currentAttempt.runId, + deliveryId: currentAttempt.deliveryId, + prNumber: currentAttempt.prNumber, + stage: currentAttempt.stage, }), ); try { diff --git a/infra/flue-review/.flue/lib/github.ts b/infra/flue-review/.flue/lib/github.ts index 16053a75a..598c38f35 100644 --- a/infra/flue-review/.flue/lib/github.ts +++ b/infra/flue-review/.flue/lib/github.ts @@ -431,6 +431,11 @@ function renderFindingsMarkdown(findings: ReviewResult["findings"]): string { return `\n\n---\n\n### Findings\n\n${lines.join("\n\n")}`; } +type ReviewLookup = + | { status: "present" } + | { status: "absent" } + | { status: "unavailable"; error: Error }; + async function reviewWasPosted( token: string, owner: string, @@ -438,19 +443,35 @@ async function reviewWasPosted( prNumber: number, commitId: string, marker: string, -): Promise { +): Promise { try { - const res = await githubFetch( - `${GITHUB_API}/repos/${owner}/${repo}/pulls/${prNumber}/reviews?per_page=100`, - { headers: installationHeaders(token) }, - ); - if (!res.ok) return false; - const reviews = await res.json>(); - return reviews.some( - (review) => review.commit_id === commitId && review.body?.includes(marker) === true, - ); - } catch { - return false; + for (let page = 1; ; page++) { + const suffix = page === 1 ? "" : `&page=${page}`; + const res = await githubFetch( + `${GITHUB_API}/repos/${owner}/${repo}/pulls/${prNumber}/reviews?per_page=100${suffix}`, + { headers: installationHeaders(token) }, + ); + if (!res.ok) { + return { + status: "unavailable", + error: new Error(`review marker inspection failed: ${res.status}`), + }; + } + const reviews = await res.json>(); + if ( + reviews.some( + (review) => review.commit_id === commitId && review.body?.includes(marker) === true, + ) + ) { + return { status: "present" }; + } + if (reviews.length < 100) return { status: "absent" }; + } + } catch (error) { + return { + status: "unavailable", + error: new Error("review marker inspection failed", { cause: error }), + }; } } @@ -473,6 +494,11 @@ export async function postReview( const event = verdictToEvent(result.verdict); const marker = attemptId ? `` : undefined; const summary = `${result.summary.trim() || FALLBACK_SUMMARY}${marker ? `\n\n${marker}` : ""}`; + if (commitId && marker) { + const lookup = await reviewWasPosted(token, owner, repo, prNumber, commitId, marker); + if (lookup.status === "present") return; + if (lookup.status === "unavailable") throw lookup.error; + } const headers = { authorization: `Bearer ${token}`, accept: "application/vnd.github+json", @@ -491,12 +517,9 @@ export async function postReview( try { res = await githubFetch(url, { method: "POST", headers, body: JSON.stringify(withComments) }); } catch (error) { - if ( - commitId && - marker && - (await reviewWasPosted(token, owner, repo, prNumber, commitId, marker)) - ) { - return; + if (commitId && marker) { + const lookup = await reviewWasPosted(token, owner, repo, prNumber, commitId, marker); + if (lookup.status === "present") return; } throw error; } @@ -507,12 +530,9 @@ export async function postReview( // carries the summary AND the findings inline, so the review still lands. const firstError = await res.text(); if (res.status !== 422) { - if ( - commitId && - marker && - (await reviewWasPosted(token, owner, repo, prNumber, commitId, marker)) - ) { - return; + if (commitId && marker) { + const lookup = await reviewWasPosted(token, owner, repo, prNumber, commitId, marker); + if (lookup.status === "present") return; } throw new Error(`postReview failed: ${res.status} ${firstError}`); } @@ -524,22 +544,16 @@ export async function postReview( try { res = await githubFetch(url, { method: "POST", headers, body: JSON.stringify(bodyOnly) }); } catch (error) { - if ( - commitId && - marker && - (await reviewWasPosted(token, owner, repo, prNumber, commitId, marker)) - ) { - return; + if (commitId && marker) { + const lookup = await reviewWasPosted(token, owner, repo, prNumber, commitId, marker); + if (lookup.status === "present") return; } throw error; } if (!res.ok) { - if ( - commitId && - marker && - (await reviewWasPosted(token, owner, repo, prNumber, commitId, marker)) - ) { - return; + if (commitId && marker) { + const lookup = await reviewWasPosted(token, owner, repo, prNumber, commitId, marker); + if (lookup.status === "present") return; } throw new Error( `postReview failed (with comments: ${firstError}); body-only retry: ${res.status} ${await res.text()}`, diff --git a/infra/flue-review/.flue/lib/review-watchdog.ts b/infra/flue-review/.flue/lib/review-watchdog.ts index 22cc87950..802376fe8 100644 --- a/infra/flue-review/.flue/lib/review-watchdog.ts +++ b/infra/flue-review/.flue/lib/review-watchdog.ts @@ -1,6 +1,16 @@ +import type { GatedPr } from "./webhook.js"; + export const REVIEW_STALE_AFTER_MS = 35 * 60_000; export const REVIEW_SETUP_LEASE_MS = 5 * 60_000; +const STAGE_STALE_AFTER_MS: Record = { + admitted: 5 * 60_000, + hydrating: 3 * 60_000, + fetching_diff: 3 * 60_000, + model_review: REVIEW_STALE_AFTER_MS, + posting_review: 5 * 60_000, +}; + export type ReviewStage = | "admitted" | "hydrating" @@ -16,6 +26,7 @@ export interface ReviewAttempt { repo: string; prNumber: number; headSha: string; + workflowInput?: GatedPr; checkRunId?: number; stage: ReviewStage; lastProgressAt: number; @@ -26,6 +37,8 @@ export interface ReviewAttempt { terminalReportedAt?: number; terminalAbandonedAt?: number; terminalRetryCount?: number; + workflowRetryCount?: number; + workflowActiveStaleSince?: number; } export interface ReviewTerminal { @@ -42,10 +55,10 @@ interface ReviewWatchdogRpc { >; arm(attempt: ReviewAttempt, setupLease: string): Promise; beginAdmission(attemptId: string, setupLease: string): Promise; - identify(attemptId: string, runId: string): Promise; - heartbeat(attemptId: string, stage: ReviewStage): Promise; + identify(attemptId: string, expectedRunId: string, runId: string): Promise; + heartbeat(attemptId: string, runId: string, stage: ReviewStage): Promise; complete(attemptId: string): Promise; - finish(attemptId: string, terminal: ReviewTerminal): Promise; + finish(attemptId: string, runId: string, terminal: ReviewTerminal): Promise; } export function getReviewWatchdog(env: Env, attemptId: string): ReviewWatchdogRpc { @@ -55,6 +68,14 @@ export function getReviewWatchdog(env: Env, attemptId: string): ReviewWatchdogRp return watchdog as ReviewWatchdogRpc; } -export function isReviewAttemptStale(lastProgressAt: number, now = Date.now()): boolean { - return now - lastProgressAt >= REVIEW_STALE_AFTER_MS; +export function reviewStaleAfter(stage: ReviewStage): number { + return STAGE_STALE_AFTER_MS[stage]; +} + +export function isReviewAttemptStale( + lastProgressAt: number, + now = Date.now(), + stage: ReviewStage = "model_review", +): boolean { + return now - lastProgressAt >= reviewStaleAfter(stage); } diff --git a/infra/flue-review/.flue/lib/workflow-admission.ts b/infra/flue-review/.flue/lib/workflow-admission.ts new file mode 100644 index 000000000..ba43986ec --- /dev/null +++ b/infra/flue-review/.flue/lib/workflow-admission.ts @@ -0,0 +1,39 @@ +import { flue } from "@flue/runtime/routing"; + +import type { GatedPr } from "./webhook.js"; + +const flueApp = flue(); + +interface AdmissionExecutionContext { + waitUntil(promise: Promise): void; + passThroughOnException(): void; +} + +export interface ReviewWorkflowInput extends GatedPr { + attemptId: string; + expectedRunId: string; + deliveryId: string; + checkRunId: number; +} + +export function admitReviewWorkflow( + input: ReviewWorkflowInput, + env: Env, + executionCtx: AdmissionExecutionContext, +): Promise { + // Hono and Workers expose slightly different ExecutionContext declarations, + // but Flue only needs their shared fetch-lifecycle methods here. + // oxlint-disable-next-line typescript/no-unsafe-type-assertion + const flueExecutionCtx = executionCtx as Parameters[2]; + return Promise.resolve( + flueApp.fetch( + new Request("https://flue.internal/workflows/review", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(input), + }), + env, + flueExecutionCtx, + ), + ); +} diff --git a/infra/flue-review/.flue/workflows/review.ts b/infra/flue-review/.flue/workflows/review.ts index e163c5dcb..1722ea0f5 100644 --- a/infra/flue-review/.flue/workflows/review.ts +++ b/infra/flue-review/.flue/workflows/review.ts @@ -60,6 +60,7 @@ const reviewPayloadSchema = v.object({ owner: v.string(), repo: v.string(), attemptId: v.optional(v.string()), + expectedRunId: v.optional(v.string()), deliveryId: v.optional(v.string()), checkRunId: v.optional(v.number()), }); @@ -222,6 +223,7 @@ async function reportStage( const active = await getReviewWatchdog(env, payload.attemptId).heartbeat( payload.attemptId, + runId, stage, ); if (!active) return false; @@ -253,6 +255,7 @@ async function finishReviewCheck( try { const finished = await getReviewWatchdog(env, payload.attemptId).finish( payload.attemptId, + runId, terminal, ); if (!finished) { @@ -290,7 +293,11 @@ async function run(context: ActionContext): Promise< if ( payload.attemptId && payload.checkRunId !== undefined && - !(await getReviewWatchdog(env, payload.attemptId).identify(payload.attemptId, runId)) + !(await getReviewWatchdog(env, payload.attemptId).identify( + payload.attemptId, + payload.expectedRunId ?? payload.attemptId, + runId, + )) ) { throw new Error("Review attempt is no longer active"); } diff --git a/infra/flue-review/test/github-checks.test.ts b/infra/flue-review/test/github-checks.test.ts index dd894f6c9..d3f359058 100644 --- a/infra/flue-review/test/github-checks.test.ts +++ b/infra/flue-review/test/github-checks.test.ts @@ -196,6 +196,7 @@ describe("GitHub review checks", () => { it("reconciles an ambiguously successful review POST by attempt marker", async () => { const fetchMock = vi .fn() + .mockResolvedValueOnce(Response.json([])) .mockResolvedValueOnce(new Response("server error", { status: 503 })) .mockResolvedValueOnce( Response.json([ @@ -207,6 +208,90 @@ describe("GitHub review checks", () => { ); vi.stubGlobal("fetch", fetchMock); + await expect( + postReview( + TOKEN, + "emdash-cms", + "emdash", + 42, + { verdict: "approve", summary: "Looks good", findings: [] }, + "head-sha", + "attempt-1", + ), + ).resolves.toBeUndefined(); + expect(fetchMock).toHaveBeenCalledTimes(3); + }); + + it("fails closed when review marker inspection is unavailable", async () => { + const fetchMock = vi + .fn() + .mockResolvedValue(new Response("unavailable", { status: 503 })); + vi.stubGlobal("fetch", fetchMock); + + await expect( + postReview( + TOKEN, + "emdash-cms", + "emdash", + 42, + { verdict: "approve", summary: "Looks good", findings: [] }, + "head-sha", + "attempt-1", + ), + ).rejects.toThrow("review marker inspection failed: 503"); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("does not post a review whose attempt marker already exists", async () => { + const fetchMock = vi.fn().mockResolvedValue( + Response.json([ + { + body: "Looks good\n\n", + commit_id: "head-sha", + }, + ]), + ); + vi.stubGlobal("fetch", fetchMock); + + await expect( + postReview( + TOKEN, + "emdash-cms", + "emdash", + 42, + { verdict: "approve", summary: "Looks good", findings: [] }, + "head-sha", + "attempt-1", + ), + ).resolves.toBeUndefined(); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledWith( + "https://api.github.com/repos/emdash-cms/emdash/pulls/42/reviews?per_page=100", + expect.objectContaining({ headers: expect.any(Object) }), + ); + }); + + it("finds an existing attempt marker on a later reviews page", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + Response.json( + Array.from({ length: 100 }, (_, index) => ({ + body: `Review ${index}`, + commit_id: "head-sha", + })), + ), + ) + .mockResolvedValueOnce( + Response.json([ + { + body: "Looks good\n\n", + commit_id: "head-sha", + }, + ]), + ); + vi.stubGlobal("fetch", fetchMock); + await expect( postReview( TOKEN, @@ -219,5 +304,9 @@ describe("GitHub review checks", () => { ), ).resolves.toBeUndefined(); expect(fetchMock).toHaveBeenCalledTimes(2); + expect(fetchMock).toHaveBeenLastCalledWith( + "https://api.github.com/repos/emdash-cms/emdash/pulls/42/reviews?per_page=100&page=2", + expect.any(Object), + ); }); }); diff --git a/infra/flue-review/test/review-watchdog-do.test.ts b/infra/flue-review/test/review-watchdog-do.test.ts index 62fcc294d..8ab1c74b4 100644 --- a/infra/flue-review/test/review-watchdog-do.test.ts +++ b/infra/flue-review/test/review-watchdog-do.test.ts @@ -4,6 +4,12 @@ const github = vi.hoisted(() => ({ completeReviewCheck: vi.fn(), readAppCreds: vi.fn(), })); +const workflow = vi.hoisted(() => ({ + admitReviewWorkflow: vi.fn(), +})); +const flue = vi.hoisted(() => ({ + getRun: vi.fn(), +})); vi.mock("cloudflare:workers", () => ({ DurableObject: class { @@ -23,6 +29,14 @@ vi.mock("../.flue/lib/github.js", () => ({ readAppCreds: github.readAppCreds, })); +vi.mock("../.flue/lib/workflow-admission.js", () => ({ + admitReviewWorkflow: workflow.admitReviewWorkflow, +})); + +vi.mock("@flue/runtime", () => ({ + getRun: flue.getRun, +})); + import { ReviewWatchdog } from "../.flue/cloudflare.js"; import type { ReviewAttempt } from "../.flue/lib/review-watchdog.js"; @@ -53,7 +67,7 @@ class MemoryStorage { function setup() { const storage = new MemoryStorage(); - const ctx = { storage }; + const ctx = { storage, waitUntil: vi.fn() }; const watchdog = new ReviewWatchdog(ctx as unknown as DurableObjectState, {} as unknown as Env); const attempt: ReviewAttempt = { attemptId: "attempt-1", @@ -77,6 +91,10 @@ beforeEach(() => { installationId: "2", privateKey: "key", }); + workflow.admitReviewWorkflow + .mockReset() + .mockResolvedValue(Response.json({ runId: "run-2" }, { status: 202 })); + flue.getRun.mockReset().mockResolvedValue({ status: "errored" }); }); describe("ReviewWatchdog terminal arbitration", () => { @@ -96,18 +114,20 @@ describe("ReviewWatchdog terminal arbitration", () => { expect(await watchdog.reserve(attempt, "lease-1")).toMatchObject({ status: "acquired" }); expect( - await watchdog.finish(attempt.attemptId, { + await watchdog.finish(attempt.attemptId, attempt.runId, { conclusion: "timed_out", summary: "timed out", }), ).toBe(true); expect( - await watchdog.finish(attempt.attemptId, { + await watchdog.finish(attempt.attemptId, attempt.runId, { conclusion: "success", summary: "late success", }), ).toBe(false); - expect(await watchdog.heartbeat(attempt.attemptId, "posting_review")).toBe(false); + expect(await watchdog.heartbeat(attempt.attemptId, attempt.runId, "posting_review")).toBe( + false, + ); expect(github.completeReviewCheck).toHaveBeenCalledTimes(1); expect(storage.values.get("attempt")).toMatchObject({ terminal: { conclusion: "timed_out" }, @@ -121,7 +141,10 @@ describe("ReviewWatchdog terminal arbitration", () => { github.completeReviewCheck.mockRejectedValue(new Error("GitHub unavailable")); await expect( - watchdog.finish(attempt.attemptId, { conclusion: "failure", summary: "failed" }), + watchdog.finish(attempt.attemptId, attempt.runId, { + conclusion: "failure", + summary: "failed", + }), ).resolves.toBe(true); const firstAlarm = storage.alarm; expect(firstAlarm).toBeTypeOf("number"); @@ -144,7 +167,10 @@ describe("ReviewWatchdog terminal arbitration", () => { github.readAppCreds.mockReturnValue(null); await expect( - watchdog.finish(attempt.attemptId, { conclusion: "failure", summary: "failed" }), + watchdog.finish(attempt.attemptId, attempt.runId, { + conclusion: "failure", + summary: "failed", + }), ).resolves.toBe(true); expect(storage.values.get("attempt")).toMatchObject({ terminalAbandonedAt: expect.any(Number), @@ -171,6 +197,239 @@ describe("ReviewWatchdog terminal arbitration", () => { }); }); + it("re-admits a stale workflow and fences the interrupted run", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.stage = "hydrating"; + attempt.lastProgressAt = 0; + attempt.workflowInput = { + prNumber: attempt.prNumber, + prTitle: "Review retry", + prBody: "", + headRef: "fix/retry", + headSha: attempt.headSha, + baseRef: "main", + baseSha: "b".repeat(40), + owner: attempt.owner, + repo: attempt.repo, + }; + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).toHaveBeenCalledWith( + expect.objectContaining({ attemptId: attempt.attemptId, checkRunId: 123 }), + expect.anything(), + expect.anything(), + ); + expect(github.completeReviewCheck).not.toHaveBeenCalled(); + expect(storage.values.get("attempt")).toMatchObject({ + runId: "run-2", + stage: "admitted", + workflowRetryCount: 1, + }); + expect(await watchdog.heartbeat(attempt.attemptId, attempt.runId, "model_review")).toBe(false); + expect(await watchdog.heartbeat(attempt.attemptId, "run-2", "hydrating")).toBe(true); + }); + + it("does not replace a stale workflow while its Flue run is active", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + attempt.workflowInput = { + prNumber: attempt.prNumber, + prTitle: "Active review", + prBody: "", + headRef: "fix/active", + headSha: attempt.headSha, + baseRef: "main", + baseSha: "b".repeat(40), + owner: attempt.owner, + repo: attempt.repo, + }; + flue.getRun.mockResolvedValue({ status: "active" }); + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).not.toHaveBeenCalled(); + expect(github.completeReviewCheck).not.toHaveBeenCalled(); + expect(storage.alarm).toBeGreaterThan(Date.now()); + }); + + it("reconciles a completed Flue run without repeating the review", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + flue.getRun.mockResolvedValue({ status: "completed" }); + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).not.toHaveBeenCalled(); + expect(github.completeReviewCheck).toHaveBeenCalledWith( + "token", + attempt.owner, + attempt.repo, + attempt.checkRunId, + expect.objectContaining({ conclusion: "success" }), + ); + expect(storage.values.get("attempt")).toMatchObject({ + terminal: { conclusion: "success" }, + }); + }); + + it("rejects a delayed ownership claim from a superseded admission", async () => { + const { attempt, storage, watchdog } = setup(); + await watchdog.reserve(attempt, "lease-1"); + + expect(await watchdog.identify(attempt.attemptId, attempt.runId, "run-2")).toBe(true); + expect(await watchdog.identify(attempt.attemptId, attempt.runId, "run-delayed")).toBe(false); + expect(storage.values.get("attempt")).toMatchObject({ runId: "run-2" }); + }); + + it("accepts repeated ownership claims for the same admitted run", async () => { + const { attempt, watchdog } = setup(); + await watchdog.reserve(attempt, "lease-1"); + + expect(await watchdog.identify(attempt.attemptId, attempt.runId, "run-2")).toBe(true); + expect(await watchdog.identify(attempt.attemptId, attempt.runId, "run-2")).toBe(true); + }); + + it("preserves the replacement budget after a transient admission failure", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + attempt.workflowInput = { + prNumber: attempt.prNumber, + prTitle: "Retry admission", + prBody: "", + headRef: "fix/retry", + headSha: attempt.headSha, + baseRef: "main", + baseSha: "b".repeat(40), + owner: attempt.owner, + repo: attempt.repo, + }; + workflow.admitReviewWorkflow + .mockResolvedValueOnce(new Response("unavailable", { status: 503 })) + .mockResolvedValueOnce(Response.json({ runId: "run-3" }, { status: 202 })); + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + + expect(github.completeReviewCheck).not.toHaveBeenCalled(); + expect(storage.values.get("attempt")).toMatchObject({ + runId: `${attempt.attemptId}:retry:1`, + workflowRetryCount: 1, + }); + const retrying = storage.values.get("attempt") as ReviewAttempt; + await storage.put("attempt", { ...retrying, lastProgressAt: 0 }); + flue.getRun.mockResolvedValue(null); + + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).toHaveBeenCalledTimes(2); + expect(storage.values.get("attempt")).toMatchObject({ + runId: "run-3", + workflowRetryCount: 2, + }); + }); + + it("times out after the bounded replacement budget is exhausted", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + attempt.workflowInput = { + prNumber: attempt.prNumber, + prTitle: "Exhaust admission", + prBody: "", + headRef: "fix/exhaust", + headSha: attempt.headSha, + baseRef: "main", + baseSha: "b".repeat(40), + owner: attempt.owner, + repo: attempt.repo, + }; + workflow.admitReviewWorkflow.mockResolvedValue(new Response("unavailable", { status: 503 })); + flue.getRun.mockResolvedValue(null); + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + let retrying = storage.values.get("attempt") as ReviewAttempt; + await storage.put("attempt", { ...retrying, lastProgressAt: 0 }); + await watchdog.alarm(); + retrying = storage.values.get("attempt") as ReviewAttempt; + await storage.put("attempt", { ...retrying, lastProgressAt: 0 }); + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).toHaveBeenCalledTimes(2); + expect(storage.values.get("attempt")).toMatchObject({ + workflowRetryCount: 2, + terminal: { conclusion: "timed_out" }, + }); + }); + + it("defers recovery when Flue run inspection is unavailable", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + flue.getRun.mockRejectedValue(new Error("registry unavailable")); + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).not.toHaveBeenCalled(); + expect(github.completeReviewCheck).not.toHaveBeenCalled(); + expect(storage.alarm).toBeGreaterThan(Date.now()); + }); + + it("does not overwrite a terminal state recorded during run inspection", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + attempt.workflowInput = { + prNumber: attempt.prNumber, + prTitle: "Concurrent finish", + prBody: "", + headRef: "fix/concurrent", + headSha: attempt.headSha, + baseRef: "main", + baseSha: "b".repeat(40), + owner: attempt.owner, + repo: attempt.repo, + }; + let resolveRun: (run: { status: "errored" }) => void = () => undefined; + flue.getRun.mockReturnValue( + new Promise((resolve) => { + resolveRun = resolve; + }), + ); + await watchdog.reserve(attempt, "lease-1"); + + const alarm = watchdog.alarm(); + await vi.waitFor(() => expect(flue.getRun).toHaveBeenCalled()); + await watchdog.finish(attempt.attemptId, attempt.runId, { + conclusion: "failure", + summary: "workflow failed", + }); + resolveRun({ status: "errored" }); + await alarm; + + expect(workflow.admitReviewWorkflow).not.toHaveBeenCalled(); + expect(storage.values.get("attempt")).toMatchObject({ + terminal: { conclusion: "failure" }, + }); + }); + + it("terminalizes a Flue run that remains active beyond the hard ceiling", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + attempt.workflowActiveStaleSince = Date.now() - 31 * 60_000; + flue.getRun.mockResolvedValue({ status: "active" }); + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).not.toHaveBeenCalled(); + expect(storage.values.get("attempt")).toMatchObject({ + terminal: { conclusion: "timed_out" }, + }); + }); + it("clears a pending alarm when setup is completed", async () => { const { attempt, storage, watchdog } = setup(); await watchdog.reserve(attempt, "lease-1"); diff --git a/infra/flue-review/test/review-watchdog.test.ts b/infra/flue-review/test/review-watchdog.test.ts index 89891df01..9171d165e 100644 --- a/infra/flue-review/test/review-watchdog.test.ts +++ b/infra/flue-review/test/review-watchdog.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from "vitest"; -import { isReviewAttemptStale, REVIEW_STALE_AFTER_MS } from "../.flue/lib/review-watchdog.js"; +import { + isReviewAttemptStale, + reviewStaleAfter, + REVIEW_STALE_AFTER_MS, +} from "../.flue/lib/review-watchdog.js"; describe("review watchdog", () => { it("does not mark an attempt stale before its deadline", () => { @@ -10,4 +14,10 @@ describe("review watchdog", () => { it("marks an attempt stale at its deadline", () => { expect(isReviewAttemptStale(1_000, 1_000 + REVIEW_STALE_AFTER_MS)).toBe(true); }); + + it("uses a shorter deadline while hydrating", () => { + const deadline = reviewStaleAfter("hydrating"); + expect(deadline).toBe(3 * 60_000); + expect(isReviewAttemptStale(1_000, 1_000 + deadline, "hydrating")).toBe(true); + }); }); diff --git a/infra/flue-review/wrangler.jsonc b/infra/flue-review/wrangler.jsonc index c6f73ecc3..c4eeec394 100644 --- a/infra/flue-review/wrangler.jsonc +++ b/infra/flue-review/wrangler.jsonc @@ -87,6 +87,14 @@ "tag": "v6-review-watchdog", "new_sqlite_classes": ["ReviewWatchdog"], }, + { + "tag": "v7-reset-registry-drop-v2", + "deleted_classes": ["FlueRegistry"], + }, + { + "tag": "v8-reset-registry-new-v2", + "new_sqlite_classes": ["FlueRegistry"], + }, ], "vars": { "NODE_OPTIONS": "--max-old-space-size=6144", From 13f834ffcb234ab35b8145247d949f024b48c9a1 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Wed, 15 Jul 2026 10:59:58 +0100 Subject: [PATCH 2/6] fix(flue-review): improve check progress details --- infra/flue-review/.flue/lib/github.ts | 66 +++++++++++++++++++- infra/flue-review/test/github-checks.test.ts | 25 +++++++- 2 files changed, 85 insertions(+), 6 deletions(-) diff --git a/infra/flue-review/.flue/lib/github.ts b/infra/flue-review/.flue/lib/github.ts index 598c38f35..144521e66 100644 --- a/infra/flue-review/.flue/lib/github.ts +++ b/infra/flue-review/.flue/lib/github.ts @@ -120,6 +120,10 @@ function installationHeaders(token: string): Record { return { ...GITHUB_HEADERS, authorization: `Bearer ${token}` }; } +function pullRequestUrl(owner: string, repo: string, prNumber: number, files = false): string { + return `https://github.com/${owner}/${repo}/pull/${prNumber}${files ? "/files" : ""}`; +} + async function requireGitHubResponse(res: Response, operation: string): Promise { if (!res.ok) throw new Error(`${operation} failed: ${res.status} ${await res.text()}`); } @@ -137,6 +141,7 @@ export async function createReviewCheck( name: "EmDash review", head_sha: input.headSha, status: "in_progress", + details_url: pullRequestUrl(owner, repo, input.prNumber, true), external_id: input.attemptId, started_at: new Date().toISOString(), output: { @@ -174,6 +179,55 @@ export async function findReviewCheck( return body.check_runs?.find((check) => check.external_id === attemptId)?.id; } +const REVIEW_PROGRESS = [ + { stage: "hydrating", label: "Prepare the workspace" }, + { stage: "fetching_diff", label: "Load the pull request diff" }, + { stage: "model_review", label: "Analyze the changes" }, + { stage: "posting_review", label: "Publish the review" }, +] as const; + +const REVIEW_STAGE_COPY: Record = { + hydrating: { + title: "Preparing", + guidance: "Next, EmDash will load the pull request diff.", + }, + fetching_diff: { + title: "Loading changes for", + guidance: "Next, the model will analyze the changed code.", + }, + model_review: { + title: "Analyzing", + guidance: + "This is usually the longest step and can take several minutes. Next, EmDash will publish the review to GitHub.", + }, + posting_review: { + title: "Publishing review for", + guidance: "The analysis is complete and the review should appear shortly.", + }, +}; + +function renderReviewProgress(stage: string, runId: string): string { + const currentIndex = REVIEW_PROGRESS.findIndex((step) => step.stage === stage); + const progress = REVIEW_PROGRESS.map((step, index) => { + if (index < currentIndex) return `- [x] ${step.label}`; + if (index === currentIndex) return `- [ ] **${step.label} (in progress)**`; + return `- [ ] ${step.label}`; + }); + return [ + "### Progress", + "", + ...progress, + "", + "
", + "Diagnostics", + "", + `Run ID: \`${runId}\``, + "", + `Stage: \`${stage}\``, + "
", + ].join("\n"); +} + export async function updateReviewCheck( token: string, owner: string, @@ -187,12 +241,17 @@ export async function updateReviewCheck( }, ): Promise { const correlation = input.runId ?? "pending admission"; + const stageCopy = REVIEW_STAGE_COPY[input.stage] ?? { + title: "Reviewing", + guidance: "EmDash will update this check when the next step begins.", + }; const body: Record = { status: "in_progress", + details_url: pullRequestUrl(owner, repo, input.prNumber, true), output: { - title: `Reviewing PR #${input.prNumber}`, - summary: input.detail, - text: `Run: \`${correlation}\`\n\nStage: \`${input.stage}\``, + title: `${stageCopy.title} PR #${input.prNumber}`, + summary: `${input.detail} ${stageCopy.guidance}`, + text: renderReviewProgress(input.stage, correlation), }, }; if (input.runId) body.external_id = input.runId; @@ -229,6 +288,7 @@ export async function completeReviewCheck( status: "completed", conclusion: input.conclusion, completed_at: new Date().toISOString(), + details_url: pullRequestUrl(owner, repo, input.prNumber), external_id: input.runId, output: { title, summary: input.summary, text: `Run: \`${input.runId}\`` }, }), diff --git a/infra/flue-review/test/github-checks.test.ts b/infra/flue-review/test/github-checks.test.ts index d3f359058..3f4dabb11 100644 --- a/infra/flue-review/test/github-checks.test.ts +++ b/infra/flue-review/test/github-checks.test.ts @@ -46,6 +46,7 @@ describe("GitHub review checks", () => { name: "EmDash review", head_sha: "abc123", status: "in_progress", + details_url: "https://github.com/emdash-cms/emdash/pull/42/files", external_id: "attempt-1", started_at: expect.any(String), output: { @@ -73,10 +74,27 @@ describe("GitHub review checks", () => { expect(requestBody(fetchMock)).toEqual({ status: "in_progress", external_id: "run_123", + details_url: "https://github.com/emdash-cms/emdash/pull/42/files", output: { - title: "Reviewing PR #42", - summary: "The model is reviewing the diff.", - text: "Run: `run_123`\n\nStage: `model_review`", + title: "Analyzing PR #42", + summary: + "The model is reviewing the diff. This is usually the longest step and can take several minutes. Next, EmDash will publish the review to GitHub.", + text: [ + "### Progress", + "", + "- [x] Prepare the workspace", + "- [x] Load the pull request diff", + "- [ ] **Analyze the changes (in progress)**", + "- [ ] Publish the review", + "", + "
", + "Diagnostics", + "", + "Run ID: `run_123`", + "", + "Stage: `model_review`", + "
", + ].join("\n"), }, }); }); @@ -99,6 +117,7 @@ describe("GitHub review checks", () => { expect(requestBody(fetchMock)).toMatchObject({ status: "completed", conclusion: "failure", + details_url: "https://github.com/emdash-cms/emdash/pull/42", completed_at: expect.any(String), external_id: "run_123", output: { From 16ccf3e1d76c5635765515cc95eec2c1917e85de Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Wed, 15 Jul 2026 11:02:02 +0100 Subject: [PATCH 3/6] chore: add empty changeset --- .changeset/loud-jobs-trade.md | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 .changeset/loud-jobs-trade.md diff --git a/.changeset/loud-jobs-trade.md b/.changeset/loud-jobs-trade.md new file mode 100644 index 000000000..a845151cc --- /dev/null +++ b/.changeset/loud-jobs-trade.md @@ -0,0 +1,2 @@ +--- +--- From c17578838ea553192153f58b40a8498d78d61c6f Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Wed, 15 Jul 2026 11:33:10 +0100 Subject: [PATCH 4/6] fix(flue-review): use EmDashBot name in checks --- infra/flue-review/.flue/lib/github.ts | 10 +++++----- infra/flue-review/test/github-checks.test.ts | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/infra/flue-review/.flue/lib/github.ts b/infra/flue-review/.flue/lib/github.ts index 144521e66..0d9510d9a 100644 --- a/infra/flue-review/.flue/lib/github.ts +++ b/infra/flue-review/.flue/lib/github.ts @@ -138,7 +138,7 @@ export async function createReviewCheck( method: "POST", headers: installationHeaders(token), body: JSON.stringify({ - name: "EmDash review", + name: "EmDashBot review", head_sha: input.headSha, status: "in_progress", details_url: pullRequestUrl(owner, repo, input.prNumber, true), @@ -166,7 +166,7 @@ export async function findReviewCheck( attemptId: string, ): Promise { const query = new URLSearchParams({ - check_name: "EmDash review", + check_name: "EmDashBot review", filter: "all", per_page: "100", }); @@ -189,7 +189,7 @@ const REVIEW_PROGRESS = [ const REVIEW_STAGE_COPY: Record = { hydrating: { title: "Preparing", - guidance: "Next, EmDash will load the pull request diff.", + guidance: "Next, EmDashBot will load the pull request diff.", }, fetching_diff: { title: "Loading changes for", @@ -198,7 +198,7 @@ const REVIEW_STAGE_COPY: Record = { model_review: { title: "Analyzing", guidance: - "This is usually the longest step and can take several minutes. Next, EmDash will publish the review to GitHub.", + "This is usually the longest step and can take several minutes. Next, EmDashBot will publish the review to GitHub.", }, posting_review: { title: "Publishing review for", @@ -243,7 +243,7 @@ export async function updateReviewCheck( const correlation = input.runId ?? "pending admission"; const stageCopy = REVIEW_STAGE_COPY[input.stage] ?? { title: "Reviewing", - guidance: "EmDash will update this check when the next step begins.", + guidance: "EmDashBot will update this check when the next step begins.", }; const body: Record = { status: "in_progress", diff --git a/infra/flue-review/test/github-checks.test.ts b/infra/flue-review/test/github-checks.test.ts index 3f4dabb11..4aef04b26 100644 --- a/infra/flue-review/test/github-checks.test.ts +++ b/infra/flue-review/test/github-checks.test.ts @@ -43,7 +43,7 @@ describe("GitHub review checks", () => { }), ); expect(requestBody(fetchMock)).toMatchObject({ - name: "EmDash review", + name: "EmDashBot review", head_sha: "abc123", status: "in_progress", details_url: "https://github.com/emdash-cms/emdash/pull/42/files", @@ -78,7 +78,7 @@ describe("GitHub review checks", () => { output: { title: "Analyzing PR #42", summary: - "The model is reviewing the diff. This is usually the longest step and can take several minutes. Next, EmDash will publish the review to GitHub.", + "The model is reviewing the diff. This is usually the longest step and can take several minutes. Next, EmDashBot will publish the review to GitHub.", text: [ "### Progress", "", From 61962d62b2b750b07190edc9cdd41e1c6856489a Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Wed, 15 Jul 2026 12:24:35 +0100 Subject: [PATCH 5/6] fix(flue-review): tighten stalled review recovery --- infra/flue-review/.flue/cloudflare.ts | 34 ++++++++++++-- infra/flue-review/.flue/lib/github.ts | 8 +++- .../flue-review/.flue/lib/review-watchdog.ts | 2 +- infra/flue-review/test/github-checks.test.ts | 25 +++++++++++ .../test/review-watchdog-do.test.ts | 44 ++++++++++++++++++- .../flue-review/test/review-watchdog.test.ts | 4 ++ 6 files changed, 111 insertions(+), 6 deletions(-) diff --git a/infra/flue-review/.flue/cloudflare.ts b/infra/flue-review/.flue/cloudflare.ts index e53344235..274476e4b 100644 --- a/infra/flue-review/.flue/cloudflare.ts +++ b/infra/flue-review/.flue/cloudflare.ts @@ -1,7 +1,12 @@ import { getRun } from "@flue/runtime"; import { DurableObject } from "cloudflare:workers"; -import { completeReviewCheck, mintInstallationToken, readAppCreds } from "./lib/github.js"; +import { + completeReviewCheck, + mintInstallationToken, + readAppCreds, + updateReviewCheck, +} from "./lib/github.js"; import { isReviewAttemptStale, reviewStaleAfter, @@ -19,7 +24,7 @@ const TERMINAL_RETRY_LIMIT = 10; const TERMINAL_RETENTION_MS = 7 * 24 * 60 * 60_000; const WORKFLOW_RETRY_LIMIT = 2; const WORKFLOW_STATUS_RETRY_MS = 60_000; -const WORKFLOW_ACTIVE_STALE_LIMIT_MS = 30 * 60_000; +const WORKFLOW_ACTIVE_STALE_LIMIT_MS = 5 * 60_000; class TerminalConfigurationError extends Error {} @@ -182,6 +187,29 @@ export class ReviewWatchdog extends DurableObject { }; await this.ctx.storage.put(ATTEMPT_KEY, retrying); await this.ctx.storage.setAlarm(lastProgressAt + reviewStaleAfter("admitted")); + try { + const creds = readAppCreds(this.env); + if (creds) { + const token = await mintInstallationToken(creds); + await updateReviewCheck(token, attempt.owner, attempt.repo, attempt.checkRunId, { + prNumber: attempt.prNumber, + runId: retryRunId, + stage: "hydrating", + detail: + "The previous review stopped reporting progress. EmDashBot is starting a replacement run.", + }); + } + } catch (error) { + console.error( + JSON.stringify({ + message: "review recovery check update failed", + error: error instanceof Error ? error.message : String(error), + attemptId: attempt.attemptId, + previousRunId: attempt.runId, + workflowRetryCount, + }), + ); + } try { // Flue needs a fetch-style context to durably admit the replacement run. @@ -237,7 +265,7 @@ export class ReviewWatchdog extends DurableObject { ): Promise<"active" | "completed" | "recoverable" | "unavailable"> { try { const run = await getRun(attempt.runId); - if (!run || run.status === "errored") return "recoverable"; + if (!run || run.status === "errored" || run.isError) return "recoverable"; return run.status; } catch (error) { console.error( diff --git a/infra/flue-review/.flue/lib/github.ts b/infra/flue-review/.flue/lib/github.ts index 0d9510d9a..6e1c07b54 100644 --- a/infra/flue-review/.flue/lib/github.ts +++ b/infra/flue-review/.flue/lib/github.ts @@ -496,6 +496,8 @@ type ReviewLookup = | { status: "absent" } | { status: "unavailable"; error: Error }; +const REVIEW_LOOKUP_MAX_PAGES = 10; + async function reviewWasPosted( token: string, owner: string, @@ -505,7 +507,7 @@ async function reviewWasPosted( marker: string, ): Promise { try { - for (let page = 1; ; page++) { + for (let page = 1; page <= REVIEW_LOOKUP_MAX_PAGES; page++) { const suffix = page === 1 ? "" : `&page=${page}`; const res = await githubFetch( `${GITHUB_API}/repos/${owner}/${repo}/pulls/${prNumber}/reviews?per_page=100${suffix}`, @@ -527,6 +529,10 @@ async function reviewWasPosted( } if (reviews.length < 100) return { status: "absent" }; } + return { + status: "unavailable", + error: new Error(`review marker inspection exceeded ${REVIEW_LOOKUP_MAX_PAGES} pages`), + }; } catch (error) { return { status: "unavailable", diff --git a/infra/flue-review/.flue/lib/review-watchdog.ts b/infra/flue-review/.flue/lib/review-watchdog.ts index 802376fe8..b81ed232b 100644 --- a/infra/flue-review/.flue/lib/review-watchdog.ts +++ b/infra/flue-review/.flue/lib/review-watchdog.ts @@ -1,6 +1,6 @@ import type { GatedPr } from "./webhook.js"; -export const REVIEW_STALE_AFTER_MS = 35 * 60_000; +export const REVIEW_STALE_AFTER_MS = 15 * 60_000; export const REVIEW_SETUP_LEASE_MS = 5 * 60_000; const STAGE_STALE_AFTER_MS: Record = { diff --git a/infra/flue-review/test/github-checks.test.ts b/infra/flue-review/test/github-checks.test.ts index 4aef04b26..d1fad7aa2 100644 --- a/infra/flue-review/test/github-checks.test.ts +++ b/infra/flue-review/test/github-checks.test.ts @@ -328,4 +328,29 @@ describe("GitHub review checks", () => { expect.any(Object), ); }); + + it("fails closed after inspecting 10 full review pages", async () => { + const reviews = Array.from({ length: 100 }, (_, index) => ({ + body: `Review ${index}`, + commit_id: "head-sha", + })); + const fetchMock = vi.fn(); + for (let page = 0; page < 10; page++) { + fetchMock.mockResolvedValueOnce(Response.json(reviews)); + } + vi.stubGlobal("fetch", fetchMock); + + await expect( + postReview( + TOKEN, + "emdash-cms", + "emdash", + 42, + { verdict: "approve", summary: "Looks good", findings: [] }, + "head-sha", + "attempt-1", + ), + ).rejects.toThrow("review marker inspection exceeded 10 pages"); + expect(fetchMock).toHaveBeenCalledTimes(10); + }); }); diff --git a/infra/flue-review/test/review-watchdog-do.test.ts b/infra/flue-review/test/review-watchdog-do.test.ts index 8ab1c74b4..9e7ff3f71 100644 --- a/infra/flue-review/test/review-watchdog-do.test.ts +++ b/infra/flue-review/test/review-watchdog-do.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const github = vi.hoisted(() => ({ completeReviewCheck: vi.fn(), readAppCreds: vi.fn(), + updateReviewCheck: vi.fn(), })); const workflow = vi.hoisted(() => ({ admitReviewWorkflow: vi.fn(), @@ -27,6 +28,7 @@ vi.mock("../.flue/lib/github.js", () => ({ completeReviewCheck: github.completeReviewCheck, mintInstallationToken: vi.fn().mockResolvedValue("token"), readAppCreds: github.readAppCreds, + updateReviewCheck: github.updateReviewCheck, })); vi.mock("../.flue/lib/workflow-admission.js", () => ({ @@ -86,6 +88,7 @@ function setup() { beforeEach(() => { github.completeReviewCheck.mockReset().mockResolvedValue(undefined); + github.updateReviewCheck.mockReset().mockResolvedValue(undefined); github.readAppCreds.mockReset().mockReturnValue({ appId: "1", installationId: "2", @@ -222,6 +225,18 @@ describe("ReviewWatchdog terminal arbitration", () => { expect.anything(), ); expect(github.completeReviewCheck).not.toHaveBeenCalled(); + expect(github.updateReviewCheck).toHaveBeenCalledWith( + "token", + attempt.owner, + attempt.repo, + attempt.checkRunId, + expect.objectContaining({ + prNumber: attempt.prNumber, + stage: "hydrating", + detail: + "The previous review stopped reporting progress. EmDashBot is starting a replacement run.", + }), + ); expect(storage.values.get("attempt")).toMatchObject({ runId: "run-2", stage: "admitted", @@ -276,6 +291,33 @@ describe("ReviewWatchdog terminal arbitration", () => { }); }); + it("re-admits a completed Flue run marked as an error", async () => { + const { attempt, storage, watchdog } = setup(); + attempt.lastProgressAt = 0; + attempt.workflowInput = { + prNumber: attempt.prNumber, + prTitle: "Errored review", + prBody: "", + headRef: "fix/errored", + headSha: attempt.headSha, + baseRef: "main", + baseSha: "b".repeat(40), + owner: attempt.owner, + repo: attempt.repo, + }; + flue.getRun.mockResolvedValue({ status: "completed", isError: true }); + await watchdog.reserve(attempt, "lease-1"); + + await watchdog.alarm(); + + expect(workflow.admitReviewWorkflow).toHaveBeenCalledTimes(1); + expect(github.completeReviewCheck).not.toHaveBeenCalled(); + expect(storage.values.get("attempt")).toMatchObject({ + runId: "run-2", + workflowRetryCount: 1, + }); + }); + it("rejects a delayed ownership claim from a superseded admission", async () => { const { attempt, storage, watchdog } = setup(); await watchdog.reserve(attempt, "lease-1"); @@ -418,7 +460,7 @@ describe("ReviewWatchdog terminal arbitration", () => { it("terminalizes a Flue run that remains active beyond the hard ceiling", async () => { const { attempt, storage, watchdog } = setup(); attempt.lastProgressAt = 0; - attempt.workflowActiveStaleSince = Date.now() - 31 * 60_000; + attempt.workflowActiveStaleSince = Date.now() - 6 * 60_000; flue.getRun.mockResolvedValue({ status: "active" }); await watchdog.reserve(attempt, "lease-1"); diff --git a/infra/flue-review/test/review-watchdog.test.ts b/infra/flue-review/test/review-watchdog.test.ts index 9171d165e..c5d8284e8 100644 --- a/infra/flue-review/test/review-watchdog.test.ts +++ b/infra/flue-review/test/review-watchdog.test.ts @@ -7,6 +7,10 @@ import { } from "../.flue/lib/review-watchdog.js"; describe("review watchdog", () => { + it("allows model review 15 minutes", () => { + expect(REVIEW_STALE_AFTER_MS).toBe(15 * 60_000); + }); + it("does not mark an attempt stale before its deadline", () => { expect(isReviewAttemptStale(1_000, 1_000 + REVIEW_STALE_AFTER_MS - 1)).toBe(false); }); From 4f9a1d77e9860aa33ba5ad5165613b1a344873f7 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Wed, 15 Jul 2026 12:28:53 +0100 Subject: [PATCH 6/6] fix(flue-review): clear trigger label on completion --- infra/flue-review/.flue/cloudflare.ts | 9 +++++++++ infra/flue-review/.flue/lib/github.ts | 17 +++++++++++++++++ infra/flue-review/test/github-checks.test.ts | 13 +++++++++++++ .../flue-review/test/review-watchdog-do.test.ts | 10 ++++++++++ 4 files changed, 49 insertions(+) diff --git a/infra/flue-review/.flue/cloudflare.ts b/infra/flue-review/.flue/cloudflare.ts index 274476e4b..9cdc11883 100644 --- a/infra/flue-review/.flue/cloudflare.ts +++ b/infra/flue-review/.flue/cloudflare.ts @@ -5,6 +5,7 @@ import { completeReviewCheck, mintInstallationToken, readAppCreds, + removePullRequestLabel, updateReviewCheck, } from "./lib/github.js"; import { @@ -25,6 +26,7 @@ const TERMINAL_RETENTION_MS = 7 * 24 * 60 * 60_000; const WORKFLOW_RETRY_LIMIT = 2; const WORKFLOW_STATUS_RETRY_MS = 60_000; const WORKFLOW_ACTIVE_STALE_LIMIT_MS = 5 * 60_000; +const MANUAL_REVIEW_LABEL = "bot:review"; class TerminalConfigurationError extends Error {} @@ -295,6 +297,13 @@ export class ReviewWatchdog extends DurableObject { prNumber: attempt.prNumber, runId: attempt.runId, }); + await removePullRequestLabel( + token, + attempt.owner, + attempt.repo, + attempt.prNumber, + MANUAL_REVIEW_LABEL, + ); await this.ctx.storage.put(ATTEMPT_KEY, { ...attempt, terminalReportedAt: Date.now(), diff --git a/infra/flue-review/.flue/lib/github.ts b/infra/flue-review/.flue/lib/github.ts index 6e1c07b54..41952f76b 100644 --- a/infra/flue-review/.flue/lib/github.ts +++ b/infra/flue-review/.flue/lib/github.ts @@ -296,6 +296,23 @@ export async function completeReviewCheck( await requireGitHubResponse(res, "complete review check"); } +export async function removePullRequestLabel( + token: string, + owner: string, + repo: string, + prNumber: number, + label: string, +): Promise { + const res = await githubFetch( + `${GITHUB_API}/repos/${owner}/${repo}/issues/${prNumber}/labels/${encodeURIComponent(label)}`, + { + method: "DELETE", + headers: installationHeaders(token), + }, + ); + if (res.status !== 404) await requireGitHubResponse(res, "remove pull request label"); +} + /** * Fetch the PR's unified diff (the canonical base...head diff, 3-dot). Used to * stage the exact changed lines into the agent's workspace, since the cf-shell diff --git a/infra/flue-review/test/github-checks.test.ts b/infra/flue-review/test/github-checks.test.ts index d1fad7aa2..07e3a84fb 100644 --- a/infra/flue-review/test/github-checks.test.ts +++ b/infra/flue-review/test/github-checks.test.ts @@ -6,6 +6,7 @@ import { findReviewCheck, fetchUnifiedDiff, postReview, + removePullRequestLabel, updateReviewCheck, } from "../.flue/lib/github.js"; @@ -22,6 +23,18 @@ afterEach(() => { }); describe("GitHub review checks", () => { + it("removes the manual review label", async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 200 })); + vi.stubGlobal("fetch", fetchMock); + + await removePullRequestLabel(TOKEN, "emdash-cms", "emdash", 42, "bot:review"); + + expect(fetchMock).toHaveBeenCalledWith( + "https://api.github.com/repos/emdash-cms/emdash/issues/42/labels/bot%3Areview", + expect.objectContaining({ method: "DELETE" }), + ); + }); + it("creates an in-progress check for the admitted head commit", async () => { const fetchMock = vi .fn() diff --git a/infra/flue-review/test/review-watchdog-do.test.ts b/infra/flue-review/test/review-watchdog-do.test.ts index 9e7ff3f71..fc2ea2489 100644 --- a/infra/flue-review/test/review-watchdog-do.test.ts +++ b/infra/flue-review/test/review-watchdog-do.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const github = vi.hoisted(() => ({ completeReviewCheck: vi.fn(), readAppCreds: vi.fn(), + removePullRequestLabel: vi.fn(), updateReviewCheck: vi.fn(), })); const workflow = vi.hoisted(() => ({ @@ -28,6 +29,7 @@ vi.mock("../.flue/lib/github.js", () => ({ completeReviewCheck: github.completeReviewCheck, mintInstallationToken: vi.fn().mockResolvedValue("token"), readAppCreds: github.readAppCreds, + removePullRequestLabel: github.removePullRequestLabel, updateReviewCheck: github.updateReviewCheck, })); @@ -88,6 +90,7 @@ function setup() { beforeEach(() => { github.completeReviewCheck.mockReset().mockResolvedValue(undefined); + github.removePullRequestLabel.mockReset().mockResolvedValue(undefined); github.updateReviewCheck.mockReset().mockResolvedValue(undefined); github.readAppCreds.mockReset().mockReturnValue({ appId: "1", @@ -132,6 +135,13 @@ describe("ReviewWatchdog terminal arbitration", () => { false, ); expect(github.completeReviewCheck).toHaveBeenCalledTimes(1); + expect(github.removePullRequestLabel).toHaveBeenCalledWith( + "token", + attempt.owner, + attempt.repo, + attempt.prNumber, + "bot:review", + ); expect(storage.values.get("attempt")).toMatchObject({ terminal: { conclusion: "timed_out" }, terminalReportedAt: expect.any(Number),