diff --git a/README.md b/README.md index c932a96c..2c941f63 100644 --- a/README.md +++ b/README.md @@ -738,10 +738,12 @@ the unreported required `quest land` check must be verified on the live host. On a base that lands through GitHub's merge queue, `quest land watch` runs on a maintainer's host, one process per repository, and tells each PR the queue -drops unmerged why. It polls; the host needs no public endpoint. +drops unmerged why. With `--fixer`, it also runs the land fixer on each PR the +queue dropped for a failure or a conflict, and puts it back at the back of the +queue. It polls; the host needs no public endpoint. ```sh -QUEST_LAND_KEY=/run/credentials/quest-land/key quest land watch --app-id 1039466 +QUEST_LAND_KEY=/run/credentials/quest-land/key quest land watch --app-id 1039466 --fixer /path/to/fixer-wrapper ``` Once a minute it reads the base's open PRs and their merge queue removals since @@ -752,18 +754,62 @@ quotes each failing job, its failed step, and its log's tail; at build concurrency 1 a group holds only that PR, so the failure is its own. A merge, and the App's own removals, get no comment. -Each pick starts with an enqueue by a writer; the App re-enqueues only within -one. A failed build or a conflict is recorded for the land fixer in -`removal-.json`, with the full log beside it. A push since the removal, a -person's removal, or a reason the watcher does not know ends the pick, and the -comment says the PR waits for a maintainer to enqueue it again. The state -directory remembers each removal handled, so a restart neither misses nor -repeats one; a crash between commenting and recording finds its own comment's -marker and posts nothing. The first pass reads the ten minutes before it. - -It shares `quest land serve`'s App key, state directory, and locks, so the two -never run for the same repository at once. Its token asks for `contents: read`, -`pull_requests: write`, and `actions: read`, which the App needs for the logs. +Each pick starts with an enqueue, by a writer or by the App, as when a +workflow enqueues Dependabot's PRs with the App's token. An App enqueue of a +head the watcher itself put back in the queue continues the watcher's pick +instead, until a handback ends it: the watcher's re-enqueue, GitHub's +auto-merge of it, or such a workflow picking the fixer's push. A failed build +or a conflict is recorded for the land fixer in `removal-.json`, with the +full log beside it. A push since the removal, a person's removal, or a reason +the watcher does not know ends the pick, and the comment says the PR waits for +a maintainer to enqueue it again. The state directory remembers each removal +handled, so a restart neither misses nor repeats one; a crash between +commenting and recording finds its own comment's marker and posts nothing. The +first pass reads the ten minutes before it. + +With `--fixer`, the watcher takes each recorded removal, oldest first, while +its PR still waits as the queue dropped it: + +- A conflict is first merged with the current base by the quest merge driver, + in the daemon's private repository, as the host gate merges a base. A clean + merge needs no fixer: the watcher lease-pushes it to the PR branch, records + it in the PR body's Review, and re-enqueues it. +- Otherwise the fixer runs in the background, as for the host gate: the same + credential-free wrapper, given the failure log and the PR's number, with the + dropped head on `pr-`, starting from the base the `merge_group` build + was built on, or from the current base for a conflict, with the merge's + conflicts as its log. At most `--fixer-jobs` (2) run at once; the rest wait + for a slot. `--fixer-effort` (medium) sets the fixer's + `model_reasoning_effort`, passed to the wrapper in `QUEST_LAND_FIXER_EFFORT`, + which `quest land fix --effort` reads by default. + +Each finished round is recorded in its pick's history, as the host gate +records its rounds: `fixed` for a fixer commit, `merged` for a clean merge, +`flaky` for a failed build the fixer requeued unchanged, and `unchanged` for a +conflict it requeued unchanged. Its failure signature is the test or step the +fixer names, falling back to the failing job and step, or, for a conflict, the +base it conflicted with. The pick keeps going while each round makes progress: +a round with no new commit for a failure that isn't flaky, the failure a fix +addressed recurring right after it, or three flaky rounds in a row hand it +back. A fix is validated, lease-pushed, and recorded in Review as for the host +gate, and a push GitHub refuses is handed back with its resolution for +`quest land apply`. The watcher then puts the fixed, merged, or flaky head back +in the queue with `expectedHeadOid` bound to it: it enqueues a head whose +required checks have passed, and arms auto-merge on one whose checks still +run, so GitHub enqueues it once they pass. Either joins the back of the queue. +It comments on the PR with what it did, after any note the fixer wrote. A +handback, from a stop rule, the fixer, a refusal, a crash, or a timeout, is +posted with its reason and a recommendation, and the PR leaves the watcher's +care until a maintainer enqueues it again. A round whose pick ended while it +ran, by a writer's push or enqueue, is not applied, and its comment says so. + +It shares `quest land serve`'s App key, state directory, private repository, +and locks, so the two never run for the same repository at once. Without a +fixer its token asks for `contents: read`, `pull_requests: write`, and +`actions: read`, which the App needs for the logs. With one it also asks for +`merge_queues: write`, which `enqueuePullRequest` needs, and `contents: write`, +for arming auto-merge and pushing fixes; a push that carries a workflow change +asks for `workflows: write`. ## Gate expensive commands diff --git a/quest/m0/land-fixer/README.md b/quest/m0/land-fixer/README.md index 5217f79e..b539ce2e 100644 --- a/quest/m0/land-fixer/README.md +++ b/quest/m0/land-fixer/README.md @@ -140,8 +140,8 @@ Decision prompts, selected options marked ✅: pick GitHub refuses fails `quest merge` instead), and flaky tests (now [Flaky failures](/quest/m0/queue-watcher/flaky.md), without the rerun). Deleted: the batch quest check, pre-checks, and throughput tuning, which - only served batches (the fixer cap and reasoning effort moved to - [Fix and re-enqueue](/quest/m0/queue-watcher/fix.md)); and queue status in + only served batches (the fixer cap and reasoning effort moved to the + watcher, as `quest land watch --fixer-jobs` and `--fixer-effort`); and queue status in the required check, since GitHub's merge queue shows each PR's position itself. diff --git a/quest/m0/land-fixer/pr-signal.md b/quest/m0/land-fixer/pr-signal.md index e411509c..b05aadec 100644 --- a/quest/m0/land-fixer/pr-signal.md +++ b/quest/m0/land-fixer/pr-signal.md @@ -36,8 +36,6 @@ refuses such a head, `quest merge` fails when GitHub refuses it, and either way the PR needs `/quest-iterate`. Retargeted 2026-11-01 to the [queue watcher](/quest/m0/queue-watcher/README.md), -which replaces the batched host gate. - -## Related - -- [Fix and re-enqueue](/quest/m0/queue-watcher/fix.md) - where the fixer uses the signal +which replaces the batched host gate. The watcher starts the fixer on each PR +the queue drops in `take` (`src/cli/land-watch-fix.ts`), with the removal's +`merge_group` log; the PR's own checks go beside it. diff --git a/quest/m0/queue-watcher/README.md b/quest/m0/queue-watcher/README.md index fdb580fc..e9788c01 100644 --- a/quest/m0/queue-watcher/README.md +++ b/quest/m0/queue-watcher/README.md @@ -130,7 +130,6 @@ Shared shape, for the children to follow: ## Required -- [Fix and re-enqueue](/quest/m0/queue-watcher/fix.md) - the fixer runs on each dropped PR in the background, and the watcher re-enqueues it at the back - [Bad picks](/quest/m0/queue-watcher/bad-picks.md) - a pick whose auto-merge is armed on a ready PR GitHub never queued is enqueued at the head it was armed on - [Flaky failures](/quest/m0/queue-watcher/flaky.md) - a flaky failure is re-enqueued unchanged, and a test seen flaking twice is disabled in a separate PR with a follow-up quest - [Watcher on the host](/quest/m0/queue-watcher/host.md) - kixelated/nixos runs the watcher for moq-dev/moq beside its fixer container diff --git a/quest/m0/queue-watcher/bad-picks.md b/quest/m0/queue-watcher/bad-picks.md index ea6e3f09..e485a007 100644 --- a/quest/m0/queue-watcher/bad-picks.md +++ b/quest/m0/queue-watcher/bad-picks.md @@ -39,3 +39,7 @@ decision 8). - An armed pick whose own required checks fail, or that turns conflicting, is left alone: it needs `/quest-iterate` and a new pick, and `quest status` shows its checks. +- The watcher's own re-enqueues are among these: it arms auto-merge on a head + the fixer pushed, or merged with the quest driver, while that head's + required checks run (`requeue` in `src/cli/land-watch-fix.ts`), and GitHub + may leave it out of the queue as it does any armed pick. diff --git a/quest/m0/queue-watcher/fix.md b/quest/m0/queue-watcher/fix.md deleted file mode 100644 index 4e4fc2e8..00000000 --- a/quest/m0/queue-watcher/fix.md +++ /dev/null @@ -1,52 +0,0 @@ -# [M] The fixer runs on each dropped PR and the watcher re-enqueues it - -## Goal - -When the merge queue drops a picked PR for a failure or a conflict, the -watcher starts the Codex land fixer on it in the background, with the failing -build's log or the conflicting base. When the fixer fixes the PR, or judges -the failure flaky, the watcher re-enqueues the resulting head at the back of -the queue. When the fixer hands back, the watcher posts its comment and the -PR leaves the watcher's care. The queue keeps landing other PRs meanwhile. - -## Plan - -Decided 2026-11-01 (maintainer, /quest-plan; see the -[epic](/quest/m0/queue-watcher/README.md)): re-enqueue at the back, and keep -going while each round makes progress. - -- Start from what `quest land watch` (`src/cli/land-watch.ts`) records for - each removal in its state directory, `removal-.json` (`readRemoval`): - a `fix` verdict carries `failure`, a `Failure` with the PR as its one entry - (keyed by the pick's enqueue time), the dropped head, the base the - `merge_group` build was built on, and the full log, ready for - `startFixer`; `jobs` names each failing job and step. Its comment says - nothing yet of the fixer; say there that it runs. -- Reuse the host gate's fixer: `startFixer`, the detached supervisor, and - `collectFixers` in `src/cli/land-fixer.ts`, the `qland-fix` container and - its launcher, `validateFix`, the ready-commit handback for - `quest land apply`, and the daemon posting the comments in the fixer's - result. The fixer's input is one PR and the dropped head, so the batch - blame step is unused. -- Re-enqueue with `enqueuePullRequest` (no `jump`) and `expectedHeadOid` set - to the head the fixer pushed or the unchanged head, as moq-bot. A fix the - fixer pushed carries its `## Review` record as a mechanical fix, as today. -- A conflict removal starts the fixer on a merge of the current base with the - quest merge driver, as `prepareHead` (`src/cli/land-head.ts`) does. -- Carried over from the dropped - [throughput tuning](https://github.com/moq-dev/quest/pull/376) plan, since - they apply to any fixer: cap concurrent fixer jobs and run them below the - queue's builds in `quest gate`, and pass an explicit - `model_reasoning_effort` (default `medium`, with a daemon option), since - fixer logs show `reasoning effort: none`. -- Apply the stop rules the host gate already applies - ([moq-dev/quest#384](https://github.com/moq-dev/quest/pull/384)): record - each round in the pick's history with `saveRound` and ask `stopRule` - (`src/cli/land-fixer.ts`) whether it hands back, as `collectJob` does. A - round is `fixed` for a fixer commit, `merged` for a clean quest-driver merge - of a conflict, `flaky` for the fixer's flaky verdict, and `unchanged` for - any other round with no new commit. Its failure signature is the test or - step the fixer names in its result's `failure`, falling back to the failing - job and step of the `merge_group` run; a conflict's is `conflictFailure` - of the base it conflicted with. A rule's handback posts `stopReason`. - Flaky handling is [Flaky failures](/quest/m0/queue-watcher/flaky.md). diff --git a/quest/m0/queue-watcher/flaky.md b/quest/m0/queue-watcher/flaky.md index 2c480836..f481c64a 100644 --- a/quest/m0/queue-watcher/flaky.md +++ b/quest/m0/queue-watcher/flaky.md @@ -51,7 +51,3 @@ naming the follow-up quest: `#[ignore = "flaky: /quest/m0/flaky/.md"]`, fix the test or delete it, and links the `merge_group` runs that showed the flake. A disabled test that is a quest's acceptance test still counts as its quest's work. - -## Required - -- [Fix and re-enqueue](/quest/m0/queue-watcher/fix.md) - the fixer's flaky verdict and the re-enqueue diff --git a/quest/m0/queue-watcher/host.md b/quest/m0/queue-watcher/host.md index 2dc2a185..75d4a932 100644 --- a/quest/m0/queue-watcher/host.md +++ b/quest/m0/queue-watcher/host.md @@ -26,9 +26,29 @@ Decided 2026-11-01 (maintainer, /quest-plan; see the - The runner's own setup and the `merge_group` workflow are moq's switch, not this quest. -## Required - -- [Fix and re-enqueue](/quest/m0/queue-watcher/fix.md) - the watcher command this deploys +The watcher command, from +[moq-dev/quest#389](https://github.com/moq-dev/quest/pull/389), run as the +same user and with the same state directory as moq's `quest land serve` +instance, and with `quest` and `flock` on PATH as for it, since the watcher +merges conflicts with the quest merge driver in the same private repository: + +```sh +quest land watch --app-id 1039466 --fixer +``` + +The fixer launcher client is the one `serve --fixer` uses, reaching +`/run/quest-land-fixer.sock` with the failure log and the PR's number. The +watcher needs no check sandbox: the `merge_group` builds run on GitHub's side. +`--fixer-jobs` (default 2) caps the watcher's own fixer jobs below the +launcher's shared slots, and `--fixer-effort` (default `medium`) sets +`QUEST_LAND_FIXER_EFFORT` in the wrapper's environment. The launcher passes +only the log and the PR numbers into the container, where `quest land fix` +defaults to `medium`, so a non-default effort needs the launcher to forward +that variable as `quest land fix --effort`. With a fixer, the watcher's +installation token asks for `merge_queues: write` (re-enqueueing, granted to +moq-bot on 2026-11-02), `contents: write` (arming auto-merge and pushing +fixes), `pull_requests: write`, and `actions: read`, and a push that carries a +workflow change asks for `workflows: write`. ## Related diff --git a/src/cli/hold.ts b/src/cli/hold.ts index 1da2915e..b453aa8d 100644 --- a/src/cli/hold.ts +++ b/src/cli/hold.ts @@ -301,41 +301,55 @@ export async function land( const PENDING = ["BLOCKED", "UNKNOWN"]; /** GraphQL error types that fault the token or the request, not the PR GitHub would enqueue. */ -const UNREFUSED = ["FORBIDDEN", "NOT_FOUND", "INSUFFICIENT_SCOPES", "INTERNAL"]; +export const UNREFUSED = ["FORBIDDEN", "NOT_FOUND", "INSUFFICIENT_SCOPES", "INTERNAL"]; /** - * Pick `snapshot`'s head to land through the base's merge queue, as {@link land} checked it: - * enqueue that head, or arm auto-merge bound to it while its required checks may be pending. - * The queue builds each PR on the current base, so a head behind it is enqueued as is. - * GitHub's queue events and auto-merge request record the pick, naming who made it. - * A pick GitHub refuses, such as one that conflicts or fails its required checks, fails: - * the PR needs a fix before it can land. + * Put `head` in the base's merge queue: enqueue it, or, while GitHub may still be running its + * required checks (`mergeState` pending), arm auto-merge bound to it, which enqueues it once they + * pass. Either joins the back of the queue. A refusal throws GitHub's error. */ -async function pick(labels: Labels, snapshot: Snapshot, method: Method): Promise { +export async function queueHead( + github: GitHub, + id: string, + head: string, + mergeState: string, + method: Method = "merge", +): Promise { const enqueuePull = () => - labels.github.graphql( + github.graphql( `mutation($id: ID!, $sha: GitObjectID!) { enqueuePullRequest(input: { pullRequestId: $id, expectedHeadOid: $sha }) { clientMutationId } }`, - { id: snapshot.id, sha: snapshot.head }, + { id, sha: head }, ); + if (!PENDING.includes(mergeState)) { + await enqueuePull(); + return; + } try { - if (!PENDING.includes(snapshot.mergeState)) { - await enqueuePull(); - return; - } - try { - await labels.github.graphql( - `mutation($id: ID!, $sha: GitObjectID!, $method: PullRequestMergeMethod!) { - enablePullRequestAutoMerge(input: { pullRequestId: $id, expectedHeadOid: $sha, mergeMethod: $method }) { clientMutationId } - }`, - { id: snapshot.id, sha: snapshot.head, method: method.toUpperCase() }, - ); - } catch (error) { - // An UNKNOWN state GitHub finished computing as ready: nothing is left to wait for. - if (!cleanAutoMergeRefusal(error instanceof Error ? error.message : null)) throw error; - await enqueuePull(); - } + await github.graphql( + `mutation($id: ID!, $sha: GitObjectID!, $method: PullRequestMergeMethod!) { + enablePullRequestAutoMerge(input: { pullRequestId: $id, expectedHeadOid: $sha, mergeMethod: $method }) { clientMutationId } + }`, + { id, sha: head, method: method.toUpperCase() }, + ); + } catch (error) { + // An UNKNOWN state GitHub finished computing as ready: nothing is left to wait for. + if (!cleanAutoMergeRefusal(error instanceof Error ? error.message : null)) throw error; + await enqueuePull(); + } +} + +/** + * Pick `snapshot`'s head to land through the base's merge queue, as {@link land} checked it, with + * {@link queueHead}. The queue builds each PR on the current base, so a head behind it is enqueued + * as is. GitHub's queue events and auto-merge request record the pick, naming who made it. + * A pick GitHub refuses, such as one that conflicts or fails its required checks, fails: + * the PR needs a fix before it can land. + */ +async function pick(labels: Labels, snapshot: Snapshot, method: Method): Promise { + try { + await queueHead(labels.github, snapshot.id, snapshot.head, snapshot.mergeState, method); } catch (error) { if (!(error instanceof GitHubGraphQLError) || error.types.some((type) => UNREFUSED.includes(type ?? ""))) throw error; diff --git a/src/cli/land-fix.ts b/src/cli/land-fix.ts index ed278397..9883c034 100644 --- a/src/cli/land-fix.ts +++ b/src/cli/land-fix.ts @@ -152,6 +152,14 @@ export function parseBaseFix(text: string): BaseFix { return value as BaseFix; } +/** Codex's reasoning efforts, for `model_reasoning_effort`. */ +export const EFFORTS = ["minimal", "low", "medium", "high", "xhigh"]; +/** + * The variable a daemon sets for its fixer wrapper to pass on as `quest land fix --effort`, since the + * wrapper's arguments are only the failure log and the PR numbers. + */ +export const EFFORT_VARIABLE = "QUEST_LAND_FIXER_EFFORT"; + export async function fix( root: string, log: string, @@ -159,7 +167,9 @@ export async function fix( agent = "codex", args?: string[], model?: string, + effort = "medium", ): Promise { + if (!EFFORTS.includes(effort)) throw new Error(`fix effort must be one of ${EFFORTS.join(", ")}`); // No PR numbers fixes the base alone, which fails its check without any PR. if ( numbers.some((number) => !Number.isSafeInteger(number) || number <= 0) || @@ -216,7 +226,7 @@ export async function fix( 'model_providers.quest-fixer.wire_api="responses"', // Without this, Codex runs the proxy's model with no reasoning at all ("reasoning effort: none"). "-c", - 'model_reasoning_effort="medium"', + `model_reasoning_effort=${JSON.stringify(effort)}`, ...(model ? ["--model", model] : []), "-", ]; diff --git a/src/cli/land-fixer.ts b/src/cli/land-fixer.ts index 3df4f405..bd017faa 100644 --- a/src/cli/land-fixer.ts +++ b/src/cli/land-fixer.ts @@ -37,11 +37,22 @@ import { BASE_JOB, type Halt, haltNotice, readHalt, saveHalt } from "./land-halt import { isParked, parked } from "./land-park"; import { excerpt, type Failure } from "./land-serve"; -class FixerRefusal extends Error {} +/** + * Why a fixer's result cannot be published, which hands its PR back. A push GitHub refused carries the + * validated resolution, for `quest land apply` to rebuild. + */ +export class FixerRefusal extends Error { + constructor( + message: string, + readonly resolution?: Resolution, + ) { + super(message); + } +} class BaseCheckError extends Error {} /** The failure log's lines the fixer cited, first to last; the comment bounds them. */ -function cited(log: string, [first, last]: [number, number]): string { +export function cited(log: string, [first, last]: [number, number]): string { if (!existsSync(log)) return ""; return readFileSync(log, "utf8") .split("\n") @@ -66,7 +77,7 @@ function readLog(path: string, window: "head" | "tail"): string { } /** Read a bounded tail, retaining the complete per-job log for host diagnosis. */ -function fixerLog(path: string): string { +export function fixerLog(path: string): string { return readLog(path, "tail").trimEnd().split("\n").slice(-100).join("\n"); } @@ -96,7 +107,7 @@ function deadline(path: string, job: Process, now: number, timeout: number): num } /** A fixer process the daemon tracks across restarts, and how it ended. */ -interface Process { +export interface Process { checkout: string; started: number; /** When the wrapper reported the launcher admitted it; unset while queued or for a wrapper that never reports. */ @@ -172,7 +183,7 @@ function save(path: string, data: unknown): void { writeFileSync(`${path}.new`, JSON.stringify(data), { mode: 0o600 }); renameSync(`${path}.new`, path); } -const historyPath = (state: string, number: number) => join(state, `fix-attempt-${number}.json`); +export const historyPath = (state: string, number: number) => join(state, `fix-attempt-${number}.json`); export function history(state: string, number: number): History | undefined { const path = historyPath(state, number); if (!existsSync(path)) return undefined; @@ -234,7 +245,7 @@ export function stopReason(rule: StopRule, failure: string): string { export function conflictFailure(base: string): string { return `merge conflict with ${base}`; } -function identity(pid: number): string | undefined { +export function identity(pid: number): string | undefined { try { return `${readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim()}:${readFileSync(`/proc/${pid}/stat`, "utf8").split(") ")[1].split(" ")[19]}`; } catch { @@ -265,9 +276,9 @@ export async function post( const extra = section?.(COMMENT_LIMIT - content.length - tail.length) ?? ""; await github.request(path, { body: { body: `${content}${extra}${tail}` } }); } -async function comment( +export async function comment( github: GitHub, - plan: Plan, + plan: Pick, entry: Pick, text: string, suffix: string, @@ -342,7 +353,7 @@ export async function startFixer( * A standalone repository in `checkout` holding `refs` fetched from the daemon's, detached at * `start`. Returns the packs the fetch wrote, whose objects the daemon already has. */ -function prepare(root: string, checkout: string, refs: string[], start: string): string[] { +export function prepare(root: string, checkout: string, refs: string[], start: string): string[] { git(checkout, ["init", "--quiet"]); // No background gc or maintenance may rewrite objects while the daemon imports the fixer's result. git(checkout, ["config", "gc.auto", "0"]); @@ -503,7 +514,14 @@ async function proposeFix( * and atomically records the wrapper's exit status. Its state path is never passed to the wrapper * or mounted in its container. */ -async function launch(path: string, job: Process, fixer: string, log: string, numbers: number[]): Promise { +export async function launch( + path: string, + job: Process, + fixer: string, + log: string, + numbers: number[], + env: NodeJS.ProcessEnv = {}, +): Promise { save(path, job); const output = openSync( `${path}.log`, @@ -523,7 +541,12 @@ async function launch(path: string, job: Process, fixer: string, log: string, nu log, ...numbers.map(String), ], - { cwd: job.checkout, detached: true, env: commandEnvironment(), stdio: ["ignore", output, output] }, + { + cwd: job.checkout, + detached: true, + env: { ...commandEnvironment(), ...env }, + stdio: ["ignore", output, output], + }, ); } finally { closeSync(output); @@ -543,7 +566,13 @@ async function launch(path: string, job: Process, fixer: string, log: string, nu } /** Whether the fixer process ended, recording its error or, on a clean exit, reading its result with `read`. */ -function settle(path: string, job: Process, deps: GitHubDependencies, timeout: number, read: () => void): boolean { +export function settle( + path: string, + job: Process, + deps: GitHubDependencies, + timeout: number, + read: () => void, +): boolean { // A restart stops the daemon's whole unit, fixers included, unless the host keeps them running. const daemon = identity(process.pid); const restarted = job.daemon && daemon && job.daemon !== daemon ? "; the daemon restarted while it ran" : ""; @@ -651,7 +680,7 @@ export function fixerText(text: string): string { .replace(/[\r\n\u0000-\u001f\u007f]/g, " "); return `\`${value}\``; } -function author(result?: Pick): string { +export function author(result?: Pick): string { const agent = result?.agent ?? "Codex"; const model = result?.model; return ( @@ -672,6 +701,94 @@ function review(body: string, head: string, reason: string, attribution: string) return `${body.slice(0, start)}\n\n${line}${body.slice(start)}`; } +/** + * Check a fixer's `fixed` head before publishing it on `entry`'s PR: a fork must allow maintainer + * edits, the head's objects are imported from the fixer's `checkout`, and it must descend from the + * pinned head and keep the protected paths, judged against the base the fixer was given rather than + * the current one, since other PRs may land protected changes while the result waits. + */ +export async function importFix( + root: string, + state: string, + plan: Pick, + entry: Pick, + checkout: string, + packs: string[] | undefined, + head: string, + read: RemoteRead, +): Promise { + if (entry.repository !== plan.repository && !entry.maintainerCanModify) + throw new FixerRefusal("fork does not allow maintainer edits"); + try { + importFixer(root, state, checkout, `pr-${entry.number}`, head, packs); + } catch (error) { + throw new FixerRefusal(String(error)); + } + await read(root, ["fetch", "--quiet", "--no-tags", baseRemote(plan), plan.base.sha]); + try { + validateFix(root, entry, head, plan.base.sha); + } catch (error) { + throw new FixerRefusal(String(error)); + } +} + +/** + * Publish `head` on `entry`'s PR in place of `standing`, the head it stands at: lease-push it, unless + * the PR is already at it (`current`), wait for GitHub to show it, and record it as a mechanical fix + * in the PR body's Review. A refused push throws a `FixerRefusal` carrying the resolution when `head` + * merges `plan`'s base on the pinned head, for `quest land apply`. + */ +export async function publishHead( + github: GitHub, + root: string, + plan: Pick, + entry: Pick, + standing: string, + current: string, + head: string, + deps: GitHubDependencies, + token: (workflows: boolean) => Promise, + record: { reason: string; attribution: string }, +): Promise { + const pullPath = `/repos/${plan.repository}/pulls/${entry.number}`; + if (current !== head) { + const pushed = await push( + root, + `https://github.com/${entry.repository}.git`, + entry.head, + head, + `refs/heads/${entry.branch}`, + token, + standing, + ); + if (pushed.status !== 0) { + // GitHub refuses an App's push of a workflow change it lacks the permission for, and a fork's + // branch rules may refuse any. The maintainer's own push may still pass, but `quest land apply` + // rebuilds only on the head the fixer worked from. + let ready: Resolution | undefined; + try { + ready = standing === entry.head ? resolution(root, entry.head, plan.base.sha, head) : undefined; + } catch (error) { + deps.warn(`quest land fixer: cannot hand back #${entry.number}'s resolution: ${String(error)}`); + } + throw new FixerRefusal(`fixer push refused: ${pushed.stderr.trim()}`, ready); + } + } + // GitHub shows a push a moment later. Wait for our own head; only a third head is a real move. + let latest = object(await (await github.request(pullPath)).json()); + for (let wait = 0; object(latest.head).sha === standing && wait < SYNC_PAUSES; wait++) { + await github.pause(1000); + latest = object(await (await github.request(pullPath)).json()); + } + if (object(latest.head).sha !== head) throw new FixerRefusal("head moved before Review update"); + await github.request(pullPath, { + method: "PATCH", + body: { + body: review(typeof latest.body === "string" ? latest.body : "", head, record.reason, record.attribution), + }, + }); +} + export async function collectFixers( root: string, state: string, @@ -852,76 +969,26 @@ async function collectJob( save(historyPath(state, number), { ...previous, requeued: queuedAs }); } else if (!error && result?.action !== "handback") { try { - if (result?.action === "fixed") { - if (entry.repository !== job.plan.repository && !entry.maintainerCanModify) - throw new FixerRefusal("fork does not allow maintainer edits"); - try { - importFixer(root, state, job.checkout, `pr-${number}`, head, job.packs); - } catch (error) { - throw new FixerRefusal(String(error)); - } - // Validate against the base the fixer was given, not the current one: other PRs may - // land protected changes while the result waits, and the requeue re-merges anyway. - await read(root, ["fetch", "--quiet", "--no-tags", baseRemote(job.plan), job.plan.base.sha]); - try { - validateFix(root, entry, head, job.plan.base.sha); - } catch (error) { - throw new FixerRefusal(String(error)); - } - } + if (result?.action === "fixed") + await importFix(root, state, job.plan, entry, job.checkout, job.packs, head, read); if (!job.published.includes(number)) { job.published.push(number); save(path, job); } - if (result?.action === "fixed" && current !== head) { - // The fix descends from the pin, not from a gate merge the PR stands at: it replaces that - // merge, as the gate replaces its own, since building on it would revert the newer base it - // merged. The lease still refuses any writer's commit. - const pushed = await push( - root, - `https://github.com/${entry.repository}.git`, - entry.head, - head, - `refs/heads/${entry.branch}`, - token, - standing, - ); - if (pushed.status !== 0) { - // GitHub refuses an App's push of a workflow change it lacks the permission for, and a - // fork's branch rules may refuse any. The maintainer's own push may still pass, but - // `quest land apply` rebuilds only on the head the fixer worked from. - try { - const ready = - standing === entry.head - ? resolution(root, entry.head, job.plan.base.sha, head) - : undefined; - if (ready) (job.resolutions ??= {})[number] = ready; - } catch (error) { - deps.warn(`quest land fixer: cannot hand back #${number}'s resolution: ${String(error)}`); - } - throw new FixerRefusal(`fixer push refused: ${pushed.stderr.trim()}`); - } - } - if (result?.action === "fixed") { - // GitHub shows a push a moment later. Wait for our own head; only a third head is a real move. - let latest = object(await (await github.request(pullPath)).json()); - for (let wait = 0; object(latest.head).sha === standing && wait < SYNC_PAUSES; wait++) { - await github.pause(1000); - latest = object(await (await github.request(pullPath)).json()); + try { + // The fix descends from the pin, not from a gate merge the PR stands at: it replaces that + // merge, as the gate replaces its own, since building on it would revert the newer base it + // merged. The lease still refuses any writer's commit. + await publishHead(github, root, job.plan, entry, standing, current, head, deps, token, { + reason: result.reason, + attribution: author(result), + }); + } catch (failure) { + if (failure instanceof FixerRefusal && failure.resolution) + (job.resolutions ??= {})[number] = failure.resolution; + throw failure; } - if (object(latest.head).sha !== head) throw new FixerRefusal("head moved before Review update"); - await github.request(pullPath, { - method: "PATCH", - body: { - body: review( - typeof latest.body === "string" ? latest.body : "", - head, - result.reason, - author(result), - ), - }, - }); } // A parked PR requeued unchanged is still queued as it was; anything else is pinned anew. diff --git a/src/cli/land-watch-fix.ts b/src/cli/land-watch-fix.ts new file mode 100644 index 00000000..ea28f0a5 --- /dev/null +++ b/src/cli/land-watch-fix.ts @@ -0,0 +1,583 @@ +// The merge queue watcher's fixer rounds: each PR the queue dropped for a failure or a conflict is fixed in +// the background and re-enqueued at the back of the queue, or handed back to a person. +import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { basename, join } from "node:path"; +import { GitHub, type GitHubDependencies, GitHubGraphQLError, object, type ObjectValue, string } from "./github"; +import { queueHead, UNREFUSED } from "./hold"; +import { baseRemote, type Entry, mergeBase, type RemoteRead } from "./land"; +import { resolutionSection, type Resolution } from "./land-apply"; +import { EFFORT_VARIABLE, type FixResult, parseFixResult, readFixResult } from "./land-fix"; +import { + author, + cited, + comment, + conflictFailure, + FixerRefusal, + fixerLog, + fixerMarkdown, + fixerText, + history, + type History, + historyPath, + identity, + importFix, + launch, + type Outcome, + post, + prepare, + type Process, + publishHead, + type Round, + saveRound, + settle, + stopReason, + stopRule, +} from "./land-fixer"; +import { save } from "./land-serve"; +import type { Removal } from "./land-watch"; + +/** What the watcher needs to run the fixer on the PRs the queue drops. */ +export interface FixContext { + /** The daemon's private repository, which no PR's checkout controls. */ + root: string; + /** The credential-free fixer wrapper, called with the failure log and the PR's number. */ + fixer: string; + /** The most fixer jobs that run at once; other dropped PRs wait for a slot, oldest first. */ + jobs: number; + /** The fixer's `model_reasoning_effort`, passed to its wrapper in `QUEST_LAND_FIXER_EFFORT`. */ + effort: string; + /** Minutes a fixer may run from its admission. */ + timeout: number; + token: (workflows: boolean) => Promise; + read: RemoteRead; +} + +/** One round of the fixer on one dropped PR, in `watch-job-.json`, from its start until it is collected. */ +interface WatchJob extends Process { + repository: string; + /** The base the round works on: the `merge_group` build's, or the current one for a conflict. */ + base: { ref: string; sha: string }; + /** The PR at the head the queue dropped; `enabledAt` is its pick's enqueue. */ + entry: Entry; + /** When the queue dropped it. */ + dropped: string; + reason: "failed_checks" | "merge_conflict"; + /** The failure log the fixer reads, which a handback's comment quotes. */ + log: string; + /** The failure's signature when the fixer names none: the failing job and step, or the conflict's base. */ + failure: string; + /** A clean merge of the base with the quest merge driver, which needs no fixer; `result` holds its head. */ + merged?: boolean; + result?: FixResult; + /** The head this round published on the PR, so a restart neither validates nor pushes it again. */ + published?: string; + /** The head being put back in the queue, recorded before asking GitHub, so a restart finds its own request. */ + requeuing?: string; + /** How GitHub took the head back, once it did: the round succeeded, and only its comments may remain. */ + requeued?: Requeue; + /** Why the round ended in a handback, recorded before posting it, so a restart keeps the reason. */ + handback?: string; + /** A validated fix whose push GitHub refused, handed back for `quest land apply`. */ + resolution?: Resolution; +} + +const JOB = /^watch-job-.*\.json$/; +const REMOVAL = /^removal-\d+\.json$/; +/** The merge driver's own result, not an agent's: it is recorded in the PR's Review under this name. */ +const MERGER = "quest land watch"; + +const PULL = `query($id: ID!) { node(id: $id) { ... on PullRequest { + state baseRefName headRefOid mergeStateStatus mergeQueueEntry { enqueuedAt } autoMergeRequest { enabledAt } +} } }`; + +async function readPull(github: GitHub, id: string): Promise { + return object((await github.graphql(PULL, { id })).node); +} + +/** + * Whether the PR is in the merge queue, or armed since `since` to join it once its required checks + * pass. An older request is not a pick of the head the queue dropped. + */ +const queued = (pull: ObjectValue, since = "") => + pull.mergeQueueEntry != null || + (pull.autoMergeRequest != null && string(object(pull.autoMergeRequest).enabledAt) > since); + +const removalPath = (state: string, number: number) => join(state, `removal-${number}.json`); + +/** The latest removal the watcher recorded for PR `number`, if any. */ +export function readRemoval(state: string, number: number): Removal | undefined { + const path = removalPath(state, number); + return existsSync(path) ? (JSON.parse(readFileSync(path, "utf8")) as Removal) : undefined; +} + +/** + * The heads the watcher put back in the queue for the pick `history` keeps, while that pick lasts: a + * handback ends it. An enqueue of one of them by the App continues the pick, whether the watcher made + * it, GitHub's auto-merge did, or a workflow that picks every push to its PRs, as moq's Dependabot one does. + */ +export function requeuedHeads(state: string, number: number): { pick: string; heads: string[] } | undefined { + const record = history(state, number); + if (!record || record.rounds.at(-1)?.outcome === "handback") return undefined; + return { pick: record.pick, heads: record.rounds.flatMap((round) => (round.after ? [round.after] : [])) }; +} + +/** + * End the pick `pick` of PR `number`, as a writer's push or a person's removal does: its history goes, + * so no later App enqueue continues it, and a round still running for it is not applied. + */ +export function endPick(state: string, number: number, pick: string): void { + if (history(state, number)?.pick === pick) rmSync(historyPath(state, number), { force: true }); +} + +/** The failing job and step the watcher found, as a failure's signature when the fixer names none. */ +function jobFailure(removal: Removal): string { + const job = removal.jobs[0]; + if (!job) return removal.reason; + return `${job.workflow} / ${job.name}${job.step ? `: ${job.step}` : ""}`; +} + +/** The running jobs, by file name. Unreadable ones are reported, and wait for a person. */ +function jobs(state: string, deps: Pick): Map { + const found = new Map(); + for (const name of readdirSync(state).filter((name) => JOB.test(name))) { + try { + found.set(name, JSON.parse(readFileSync(join(state, name), "utf8")) as WatchJob); + } catch (error) { + deps.warn(`quest land watch: unreadable fixer job ${name}: ${String(error)}`); + } + } + return found; +} + +/** + * Start the fixer on the dropped PRs waiting for it, oldest removal first, then collect the jobs that + * finished. A conflict is first merged with the quest merge driver, which needs no fixer slot. + */ +export async function advanceFixes( + repository: string, + base: { ref: string; sha: string }, + state: string, + context: FixContext, + deps: GitHubDependencies, +): Promise { + const github = new GitHub(deps); + // Collect first, so a finished job's slot is free, and a clean merge once it is made. + const output = await collectAll(github, state, context, deps); + const waiting = readdirSync(state) + .filter((name) => REMOVAL.test(name)) + .map((name) => JSON.parse(readFileSync(join(state, name), "utf8")) as Removal) + .filter((removal) => removal.failure && removal.taken === undefined) + .sort((a, b) => a.at.localeCompare(b.at) || a.number - b.number); + let running = [...jobs(state, deps).values()].filter((job) => !job.merged).length; + for (const removal of waiting) { + const line = await take(github, repository, base, state, context, deps, removal, running < context.jobs); + if (line?.startsWith("fixing ")) running++; + if (line) output.push(line); + } + output.push(...(await collectAll(github, state, context, deps, true))); + return output; +} + +/** + * Collect every finished job, or only the clean merges with `merges`. One that fails to collect is + * handed back, or retried on the next pass. + */ +async function collectAll( + github: GitHub, + state: string, + context: FixContext, + deps: GitHubDependencies, + merges = false, +): Promise { + const output: string[] = []; + for (const [name, job] of jobs(state, deps)) { + if (merges && !job.merged) continue; + const path = join(state, name); + try { + output.push(...(await collect(github, state, path, job, context, deps))); + } catch (error) { + // Once a round has asked GitHub to take its head back, it may already be in the queue: it never + // becomes a handback on an error, and the next pass retries it, which re-enqueues only if needed. + if (job.requeuing !== undefined) { + deps.warn(`quest land watch: fixer round ${name} pending: ${String(error)}`); + continue; + } + // Keep the first reason: a GitHub error while handing back a crash must not replace the crash. + job.handback ??= `fixer collection failed: ${String(error)}`; + try { + save(path, job); + output.push(...(await collect(github, state, path, job, context, deps))); + } catch (failure) { + deps.warn(`quest land watch: handback pending for ${name}: ${String(failure)}`); + } + } + } + return output; +} + +/** + * Take a removal waiting for the fixer, if its PR still waits as the queue dropped it: merge a + * conflict's base with the quest merge driver, or start the fixer when `slot` allows. Returns what it + * did, if anything. + */ +async function take( + github: GitHub, + repository: string, + current: { ref: string; sha: string }, + state: string, + context: FixContext, + deps: GitHubDependencies, + removal: Removal, + slot: boolean, +): Promise { + const failure = removal.failure!; + const entry = failure.plan.queue[0]; + const number = removal.number; + const mark = (taken: string) => save(removalPath(state, number), { ...removal, taken } satisfies Removal); + const pull = await readPull(github, entry.id); + // The pick ended while the PR waited: closed, retargeted, a writer's push, or a writer's new enqueue. + const why = + pull.state !== "OPEN" + ? "closed" + : string(pull.baseRefName) !== failure.plan.base.ref + ? "retargeted" + : string(pull.headRefOid) !== removal.head + ? "pushed" + : queued(pull, removal.at) + ? "picked" + : undefined; + if (why) { + mark(why); + return `dropped ${number} ${why}`; + } + // A job already started on it, as before a crash that lost this record, takes it. + const started = [...jobs(state, deps).entries()].find( + ([, job]) => job.entry.number === number && job.dropped === removal.at, + ); + if (started) { + mark(`${started[1].merged ? "merged" : "fixer"} ${started[0]}`); + return undefined; + } + const conflict = removal.reason === "merge_conflict"; + // A conflict is merged with the current base; a failed build is fixed on the base it was built on. + const base = conflict ? current : failure.plan.base; + // Without a fixer slot, only a conflict's merge with a base it was not yet tried on can go ahead. + if (!slot && !(conflict && removal.unmerged !== base.sha)) return undefined; + // The round begins its pick's history, unless it continues that pick. + if (history(state, number)?.pick !== entry.enabledAt) + save(historyPath(state, number), { pick: entry.enabledAt, rounds: [] } satisfies History); + await context.read(context.root, [ + "fetch", + "--quiet", + "--no-tags", + baseRemote({ repository }), + base.sha, + removal.head, + ]); + let log = removal.log; + if (conflict) { + const name = `job-${Date.now().toString(36)}-${number}`; + const path = join(state, `conflict-${number}-${base.sha.slice(0, 12)}.log`); + if (removal.unmerged !== base.sha) { + const merged = mergeBase(context.root, removal.head, base.sha); + if ("head" in merged) { + const job: WatchJob = { + checkout: "", + started: deps.now(), + repository, + base, + entry, + dropped: removal.at, + reason: "merge_conflict", + log: removal.log, + failure: conflictFailure(base.sha), + merged: true, + result: { + number, + action: "fixed", + head: merged.head, + reason: `merged ${base.ref} at ${base.sha} with the quest merge driver`, + agent: MERGER, + }, + }; + save(join(state, `watch-${name}.json`), job); + mark(`merged watch-${name}.json`); + return `merged ${number} ${merged.head}`; + } + writeFileSync( + path, + `The merge queue removed ${removal.url} at ${removal.head}: it conflicts with ${base.ref} at ${base.sha}, even with the quest merge driver.\n\n${merged.detail}\n`, + { mode: 0o600 }, + ); + // The conflict stands until the base moves: retry the merge only then. + removal = { ...removal, unmerged: base.sha }; + save(removalPath(state, number), removal); + } + if (existsSync(path)) log = path; + } + if (!slot) return undefined; + + const checkout = mkdtempSync(join(state, "fix-")); + try { + const packs = prepare( + context.root, + checkout, + [ + `${removal.head}:refs/heads/batch`, + `${base.sha}:refs/heads/base`, + `${removal.head}:refs/heads/pr-${number}`, + ], + // Start at the base: the agent loads its instructions and tooling only from code already landed. + base.sha, + ); + const job: WatchJob = { + checkout, + started: deps.now(), + daemon: identity(process.pid), + packs, + repository, + base, + entry, + dropped: removal.at, + reason: conflict ? "merge_conflict" : "failed_checks", + log, + failure: conflict ? conflictFailure(base.sha) : jobFailure(removal), + }; + const name = `job-${basename(checkout)}`; + await launch(join(state, `watch-${name}.json`), job, context.fixer, log, [number], { + [EFFORT_VARIABLE]: context.effort, + }); + mark(`fixer watch-${name}.json`); + return `fixing ${number}`; + } catch (error) { + rmSync(checkout, { recursive: true, force: true }); + throw error; + } +} + +/** End a job: its checkout and its records go, and its output log stays for diagnosis on the host. */ +function finish(path: string, job: WatchJob): void { + if (job.checkout) rmSync(job.checkout, { recursive: true, force: true }); + rmSync(`${path}.exit`, { force: true }); + rmSync(path); +} + +/** + * Collect a finished round: record it in its pick's history, and either put the PR back in the queue, + * at a fixed head or unchanged, or hand it back, as a stop rule, the fixer, or a refusal says. A + * round whose pick ended while it ran is dropped, and the PR says so. Returns what it did. + */ +async function collect( + github: GitHub, + state: string, + path: string, + job: WatchJob, + context: FixContext, + deps: GitHubDependencies, +): Promise { + const { entry } = job; + const number = entry.number; + if ( + !job.result && + !job.error && + !job.handback && + !settle(path, job, deps, context.timeout, () => { + job.result = parseFixResult(readFixResult(job.checkout), [number])[0]; + }) + ) + return []; + const name = basename(path); + const result = job.result; + const head = result?.action === "fixed" ? result.head : entry.head; + const attribution = author(result); + const plan = { repository: job.repository }; + const marked = { number, enabledAt: entry.enabledAt }; + // The fixer's own note goes first, so the watcher's outcome comment follows it. + const note = async () => { + if (result?.comment) + await post( + github, + plan, + marked, + `${attribution} land fixer note:\n\n${fixerMarkdown(result.comment)}`, + `note ${name}`, + attribution, + ); + }; + // Say how the round went back to the queue, once GitHub took it. + const announce = async (requeued: Requeue, result: FixResult): Promise => { + await note(); + const what = job.merged + ? `with ${job.base.ref} at ${job.base.sha} merged in by the quest merge driver` + : result.action === "fixed" + ? `with the land fixer's fix (${fixerText(attribution)}): ${fixerText(result.reason)}` + : `unchanged, since the land fixer judged the failure flaky: ${fixerText(result.failure ?? ("reason" in result ? result.reason : ""))}`; + const how = + requeued === "armed" + ? `armed auto-merge on this pull request at ${head}, so it joins the back of the merge queue once its required checks pass` + : `re-enqueued this pull request at ${head}, at the back of the merge queue`; + await post(github, plan, marked, `quest land watch: ${how}, ${what}.`, `requeued ${name}`, MERGER); + finish(path, job); + return [`requeued ${number} ${head}`]; + }; + if (job.requeued && result) return announce(job.requeued, result); + const pull = await readPull(github, entry.id); + const current = string(pull.headRefOid); + const record = history(state, number); + + if (pull.state !== "OPEN") { + finish(path, job); + return [`dropped ${number} closed`]; + } + // The queue dropped the PR again since. A round that already asked GitHub to re-enqueue its head is + // done, as before a restart: that removal is handled on its own, never undone by enqueuing it again. + const latest = readRemoval(state, number); + const newer = latest !== undefined && latest.at > job.dropped; + if (newer && job.requeuing !== undefined) { + finish(path, job); + return [`requeued ${number} ${job.requeuing}`]; + } + // The pick ended while the round ran: a writer's push, a writer's new enqueue, or the queue dropping a + // newer enqueue of it. Its result is not applied; a head or request this round made is still its own. + const ours = job.published !== undefined && current === job.published; + const retargeted = string(pull.baseRefName) !== job.base.ref; + const ended = retargeted + ? `it was retargeted to ${fixerText(string(pull.baseRefName))} while the fixer ran, so it no longer lands on ${fixerText(job.base.ref)}` + : current !== entry.head && !ours + ? `its head moved to ${current} while the fixer ran, and a push by anyone but the land fixer ends the pick` + : record?.pick !== entry.enabledAt || newer + ? "the pick it was fixing ended: the pull request was enqueued, or removed, again while the fixer ran" + : queued(pull, job.dropped) && !ours && job.requeuing === undefined + ? "a maintainer enqueued it again while the fixer ran, and that pick stands" + : undefined; + if (ended) { + await note(); + await post( + github, + plan, + marked, + `quest land watch: the land fixer's result for ${entry.head} was not applied: ${ended}.`, + `ended ${name}`, + MERGER, + ); + finish(path, job); + return [`dropped ${number} ended`]; + } + + let handback = job.handback ?? job.error; + let round: Round | undefined; + const conflict = job.reason === "merge_conflict"; + const failure = conflict || !result ? job.failure : (result.failure ?? job.failure); + if (!handback && result) { + // Requeued unchanged, a failed build is flaky by the fixer's judgment; a conflict still conflicts. + const outcome: Outcome = job.merged + ? "merged" + : result.action === "fixed" + ? "fixed" + : result.action === "requeue" + ? conflict + ? "unchanged" + : "flaky" + : "handback"; + const rule = stopRule( + record!.rounds.filter((item) => item.job !== name), + { outcome, failure }, + ); + if (rule) handback = stopReason(rule, failure); + round = rule + ? { job: name, outcome: "handback", before: entry.head, failure, rule } + : { job: name, outcome, before: entry.head, failure, ...(outcome === "handback" ? {} : { after: head }) }; + saveRound(state, number, record!, round); + } + + if (!handback && result && result.action !== "handback") { + try { + if (result.action === "fixed") { + if (job.published === undefined) { + if (job.merged) { + if (entry.repository !== job.repository && !entry.maintainerCanModify) + throw new FixerRefusal("fork does not allow maintainer edits"); + } else + await importFix(context.root, state, job, entry, job.checkout, job.packs, head, context.read); + job.published = head; + save(path, job); + } + await publishHead(github, context.root, job, entry, entry.head, current, head, deps, context.token, { + reason: result.reason, + attribution, + }); + } + if (job.requeuing === undefined) { + job.requeuing = head; + save(path, job); + } + job.requeued = await requeue(github, entry.id, head, job.base.ref, job.dropped); + save(path, job); + } catch (failure) { + if (!(failure instanceof FixerRefusal)) throw failure; + handback = failure.message; + if (failure.resolution) job.resolution = failure.resolution; + } + } + + if (job.requeued && result) return announce(job.requeued, result); + // Anything else is a handback. + await note(); + // Every handback ends the pick with a round of its own, a crash's or a timeout's too, so the App's + // next enqueue of a head this pick put back starts a new pick. + if (round?.outcome !== "handback") { + const { after: _, ...rest } = round ?? { job: name, before: entry.head, failure }; + saveRound(state, number, history(state, number)!, { ...rest, outcome: "handback" }); + } + if (handback) { + job.handback = handback; + save(path, job); + } + // The marker `quest status` reads for why the PR stopped landing. + await comment( + github, + plan, + marked, + handback + ? `${handback.slice(0, 300)}. Recommendation: run /quest-iterate before picking it again.` + : result?.action === "handback" + ? `${result.diagnosis.slice(0, 200)}. Recommendation: ${result.recommendation.slice(0, 200)}` + : "missing result; run /quest-iterate", + `handback ${name}`, + attribution, + !handback && result?.action === "handback" && result.log ? cited(job.log, result.log) : fixerLog(path), + job.reason, + job.resolution ? (room) => resolutionSection(number, job.resolution!, room) : undefined, + ); + finish(path, job); + return [`handback ${number}`]; +} + +/** How `head` went back: enqueued, or armed to join the queue once its required checks pass. */ +type Requeue = "enqueued" | "armed"; + +/** + * Put `head` back in `base`'s merge queue, at the back. A PR queued or armed at it since the queue + * dropped it, at `dropped`, needs nothing more: a restart finds its own request, and a workflow may pick + * the fixer's push first. A refusal, or a PR retargeted from `base`, hands the PR back. + */ +async function requeue(github: GitHub, id: string, head: string, base: string, dropped: string): Promise { + const how = (pull: ObjectValue): Requeue => (pull.mergeQueueEntry != null ? "enqueued" : "armed"); + let pull = await readPull(github, id); + if (string(pull.baseRefName) !== base) throw new FixerRefusal(`retargeted from ${base} before it was re-enqueued`); + if (string(pull.headRefOid) !== head) throw new FixerRefusal("head moved before it was re-enqueued"); + if (!queued(pull, dropped)) { + try { + await queueHead(github, id, head, string(pull.mergeStateStatus)); + } catch (error) { + if (!(error instanceof GitHubGraphQLError) || error.types.some((type) => UNREFUSED.includes(type ?? ""))) + throw error; + pull = await readPull(github, id); + if (string(pull.headRefOid) !== head || !queued(pull, dropped)) + throw new FixerRefusal(`GitHub refused to re-enqueue ${head}: ${error.message}`); + return how(pull); + } + pull = await readPull(github, id); + } + return how(pull); +} diff --git a/src/cli/land-watch.ts b/src/cli/land-watch.ts index c830da4e..49585479 100644 --- a/src/cli/land-watch.ts +++ b/src/cli/land-watch.ts @@ -1,6 +1,6 @@ // The merge queue watcher: notice each PR GitHub's merge queue drops unmerged, record why, and say so on it. -import { existsSync, readFileSync, writeFileSync } from "node:fs"; -import { join } from "node:path"; +import { accessSync, constants, existsSync, readFileSync, writeFileSync } from "node:fs"; +import { join, resolve } from "node:path"; import { setTimeout as sleep } from "node:timers/promises"; import { array, @@ -13,6 +13,7 @@ import { string, } from "./github"; import type { Entry } from "./land"; +import { authenticatedRead } from "./land-git"; import { fixerText, post, redactTokens } from "./land-fixer"; import { apiDependencies, @@ -21,15 +22,25 @@ import { excerpt, type Failure, installationToken, + privateRepository, save, startDaemon, } from "./land-serve"; +import { advanceFixes, endPick, type FixContext, readRemoval, requeuedHeads } from "./land-watch-fix"; + +export { readRemoval } from "./land-watch-fix"; export interface WatchOptions { base?: string; /** Empty for the default, `defaultState`, which `quest land serve` shares. */ state: string; appId: string; + /** The credential-free fixer wrapper; without one, removals are only recorded. */ + fixer?: string; + /** The most fixer jobs that run at once. */ + fixerJobs: number; + /** The fixer's `model_reasoning_effort`. */ + fixerEffort: string; } /** @@ -77,6 +88,10 @@ export interface Removal { log: string; /** For a `fix` verdict: the fixer's input, the dropped head as the pick's one entry on the base it was built on. */ failure?: Failure; + /** What the fixer step made of a `fix` verdict; unset while it waits for one. */ + taken?: string; + /** For a conflict: the base the quest merge driver could not merge, retried only once the base moves. */ + unmerged?: string; } /** The watcher's progress, in `watch.json`. */ @@ -155,28 +170,41 @@ async function removalsOf(github: GitHub, pull: ObjectValue, since: string): Pro } /** - * The pick a removal belongs to: the latest enqueue at or before it by anyone but the App, which only - * enqueues within a pick, and the latest enqueue of any kind, from which its builds are looked for. + * The pick a removal belongs to, and the latest enqueue of any kind, from which its builds are looked + * for. A pick is the latest enqueue at or before the removal by anyone, the App included, as when a + * workflow enqueues Dependabot's PRs with its token, except one continuing the watcher's own pick: an + * App enqueue since that pick began, of a head the watcher put back in the queue for it + * (`requeuedHeads`), which is the watcher's re-enqueue, GitHub's auto-merge of it, or such a workflow + * picking the fixer's push. */ async function pickOf( github: GitHub, id: string, removal: ObjectValue, + head: string, app: string, + own: { pick: string; heads: string[] } | undefined, ): Promise<{ pick: QueuePick; queued?: string }> { const at = string(removal.createdAt); const data = await github.graphql(ENQUEUES, { id }); const enqueues = array(object(object(data.node).timelineItems).nodes).filter( (node) => typeof node.createdAt === "string" && node.createdAt <= at, ); - const writer = enqueues.filter((node) => account(login(node.actor)) !== app).at(-1); + const continues = own !== undefined && own.heads.includes(head); + const picks = enqueues.filter( + (node) => !(continues && account(login(node.actor)) === app && string(node.createdAt) > own.pick), + ); + const writer = picks.at(-1); const queued = enqueues.at(-1)?.createdAt as string | undefined; if (writer) return { pick: { by: login(writer.enqueuer) || login(writer.actor), at: string(writer.createdAt) }, ...(queued ? { queued } : {}), }; - return { pick: { by: login(removal.enqueuer), at: queued ?? at }, ...(queued ? { queued } : {}) }; + return { + pick: { by: login(removal.enqueuer), at: continues ? own.pick : (queued ?? at) }, + ...(queued ? { queued } : {}), + }; } /** A job log as GitHub serves it, without its byte-order mark or per-line timestamps, and with no token. */ @@ -286,9 +314,12 @@ export function jobsSection(blocks: string[], room: number): string { return text; } -/** What the watcher does about it, if anything it needs to say. */ -const AFTER: Record string> = { - fix: () => "", +/** What the watcher does about it, if anything it needs to say; `fixing` when a land fixer is configured. */ +const AFTER: Record string> = { + fix: (_, fixing) => + fixing + ? "\n\nThe land fixer works on it in the background, and the watcher re-enqueues it at the back of the queue once it is fixed or judged flaky, or hands it back here with what it needs." + : "", pushed: (current) => `\n\nIts head has moved to ${current} since, and a push by anyone but the land fixer ends the pick: the watcher leaves this pull request alone until a maintainer enqueues it again.`, removed: () => @@ -308,6 +339,7 @@ async function handle( base: { ref: string; sha: string }, state: string, app: string, + fixing: boolean, pull: ObjectValue, event: ObjectValue, ): Promise { @@ -318,7 +350,7 @@ async function handle( const current = string(pull.headRefOid); const head = event.beforeCommit == null ? current : string(object(event.beforeCommit).oid); const at = string(event.createdAt); - const { pick, queued } = await pickOf(github, string(pull.id), event, app); + const { pick, queued } = await pickOf(github, string(pull.id), event, head, app, requeuedHeads(state, number)); const verdict: Verdict = head !== current ? "pushed" @@ -369,7 +401,7 @@ async function handle( github, { repository }, { number, enabledAt: pick.at }, - `quest land watch: the merge queue removed this pull request at ${head} because ${because(reason, actor, base.ref)} (\`${reason.replace(/[^a-z_]/g, "")}\`).${AFTER[verdict](current)}${missing}`, + `quest land watch: the merge queue removed this pull request at ${head} because ${because(reason, actor, base.ref)} (\`${reason.replace(/[^a-z_]/g, "")}\`).${AFTER[verdict](current, fixing)}${missing}`, `removed ${string(event.id)}`, "quest land watch", undefined, @@ -416,25 +448,25 @@ async function handle( : {}), }; // A removal GitHub showed late never replaces the record of a newer one. - if ((readRemoval(state, number)?.at ?? "") <= at) save(join(state, `removal-${number}.json`), removal); + if ((readRemoval(state, number)?.at ?? "") <= at) { + save(join(state, `removal-${number}.json`), removal); + // Any other verdict ends the pick: its fixer history goes, so a later enqueue starts afresh. + if (verdict !== "fix") endPick(state, number, pick.at); + } return `removed ${number} ${reason} ${verdict}`; } -/** The latest removal the watcher recorded for PR `number`, if any. */ -export function readRemoval(state: string, number: number): Removal | undefined { - const path = join(state, `removal-${number}.json`); - return existsSync(path) ? (JSON.parse(readFileSync(path, "utf8")) as Removal) : undefined; -} - /** * One pass over the base's open PRs: handle each removal from the merge queue since the last pass, * oldest first, and remember each one handled, so a restart neither misses nor repeats one. The * first pass starts `SETTLE` before it. A PR closed before a pass reads its removal is not read. + * With a fixer, `fix` then starts it on the PRs the queue dropped and collects its rounds. */ export async function watchPass( repository: Repository, options: Pick, deps: GitHubDependencies, + fix?: FixContext, ): Promise { const github = new GitHub(deps); const name = `${repository.owner}/${repository.repo}`; @@ -493,7 +525,16 @@ export async function watchPass( const id = string(event.id); const handled = (progress.handled[number] ??= {}); if (id in handled) continue; - const line = await handle(github, name, { ref, sha: sha! }, options.state, app, pull, event); + const line = await handle( + github, + name, + { ref, sha: sha! }, + options.state, + app, + fix !== undefined, + pull, + event, + ); if (line) output.push(line); handled[id] = at; save(path, progress); @@ -507,30 +548,69 @@ export async function watchPass( } progress.since = started; save(path, progress); + if (fix) output.push(...(await advanceFixes(name, { ref, sha: sha! }, options.state, fix, deps))); return output; } -/** Watch the merge queue of `repository`'s base, one pass a minute, until the process stops. */ +type Permissions = Record; + +/** + * The installation token permissions the watcher asks for. `api` reads the queue's events, comments, + * and the `merge_group` logs (`actions: read`); with a fixer it also re-enqueues, which needs + * `merge_queues: write` for `enqueuePullRequest` and `contents: write` for arming auto-merge, and it + * edits the PR body's Review. `push` lease-pushes a fix or merge, and `pushWorkflows` one that carries + * a workflow change, as `quest land serve` does; `read` fetches. + */ +export function watchPermissions(fixing: boolean): { + api: Permissions; + push?: Permissions; + pushWorkflows?: Permissions; + read?: Permissions; +} { + if (!fixing) return { api: { contents: "read", pull_requests: "write", actions: "read" } }; + const push: Permissions = { contents: "write", pull_requests: "write" }; + return { + api: { contents: "write", pull_requests: "write", actions: "read", merge_queues: "write" }, + push, + pushWorkflows: { ...push, workflows: "write" }, + read: { contents: "read" }, + }; +} + +/** + * Watch the merge queue of `repository`'s base, one pass a minute, until the process stops. With a + * fixer, the PRs it drops are fixed from the daemon's private repository in the state directory. + */ export async function watch( root: string, repository: Repository, options: WatchOptions, warn: (text: string) => void, ): Promise { + // The wrapper runs in the fixer's checkout, where a relative path would name the PR's own file. + if (options.fixer) { + options.fixer = resolve(options.fixer); + accessSync(options.fixer, constants.X_OK); + } const unlock = await startDaemon(root, repository, options); try { - const deps = apiDependencies( - () => - installationToken( - repository, - options, - { contents: "read", pull_requests: "write", actions: "read" }, - { fetch, now: Date.now }, - ), - { fetch, now: Date.now, sleep, warn }, - ); + const now = { fetch, now: Date.now }; + const permissions = watchPermissions(options.fixer !== undefined); + const mint = (wanted: Permissions) => installationToken(repository, options, wanted, now); + const deps = apiDependencies(() => mint(permissions.api), { ...now, sleep, warn }); + const fix: FixContext | undefined = options.fixer + ? { + root: privateRepository(options.state), + fixer: options.fixer, + jobs: options.fixerJobs, + effort: options.fixerEffort, + timeout: 60, + token: (workflows) => mint(workflows ? permissions.pushWorkflows! : permissions.push!), + read: authenticatedRead(() => mint(permissions.read!)), + } + : undefined; await daemonLoop(async () => { - for (const line of await watchPass(repository, options, deps)) warn(line); + for (const line of await watchPass(repository, options, deps, fix)) warn(line); return false; }, deps); } finally { diff --git a/src/cli/main.ts b/src/cli/main.ts index 74c22a81..fc62d125 100644 --- a/src/cli/main.ts +++ b/src/cli/main.ts @@ -44,7 +44,7 @@ import { gitHubDependencies, GitHubInputError, mergedHead, pullRequest, readRemo import { type Hold, finish, land, release, take } from "./hold"; import { LAND_CHECK } from "./land"; import { apply, pushCommand } from "./land-apply"; -import { fix, FixerAgentError } from "./land-fix"; +import { EFFORT_VARIABLE, EFFORTS, fix, FixerAgentError } from "./land-fix"; import { serve, type ServeOptions } from "./land-serve"; import { watch, type WatchOptions } from "./land-watch"; import { landed } from "./landed"; @@ -525,19 +525,32 @@ export function main(argv: string[], out: Output): number | Promise { (value: string, previous: string[]) => [...previous, value], [] as string[], ) - .action((log: string, prs: string[], options: { agent: string; agentArg: string[]; model?: string }) => { - settle( - fix( - root(), - log, - prs.map(Number), - options.agent, - options.agentArg.length ? options.agentArg : undefined, - options.model, - ), - () => print("fixer result written"), - ); - }); + .addOption( + new Option( + "--effort ", + `Default Codex's model_reasoning_effort (default: $${EFFORT_VARIABLE}, or medium)`, + ).choices(EFFORTS), + ) + .action( + ( + log: string, + prs: string[], + options: { agent: string; agentArg: string[]; model?: string; effort?: string }, + ) => { + settle( + fix( + root(), + log, + prs.map(Number), + options.agent, + options.agentArg.length ? options.agentArg : undefined, + options.model, + options.effort ?? (process.env[EFFORT_VARIABLE] || "medium"), + ), + () => print("fixer result written"), + ); + }, + ); landCommand .command("apply") .description( @@ -588,11 +601,24 @@ export function main(argv: string[], out: Output): number | Promise { landCommand .command("watch") .description( - "Watch the base's GitHub merge queue as a daemon, one per repository: once a minute, read the base's open PRs and the merge queue's removals since the last pass, and comment once on each PR the queue dropped unmerged, naming GitHub's reason and, for a failed `merge_group` build, the failing jobs and their logs' tails. A failed build or a conflict is recorded for the land fixer; a push since the removal, a person's removal, or an unknown reason ends the pick. A pick is the latest enqueue by anyone but the App, which re-enqueues only within one. State keeps each removal handled, so a restart neither misses nor repeats one, and each removal's full log. It shares `quest land serve`'s App key, state directory, and locks, so the two never run beside each other. The App needs `actions: read` for the logs.", + "Watch the base's GitHub merge queue as a daemon, one per repository: once a minute, read the base's open PRs and the merge queue's removals since the last pass, and comment once on each PR the queue dropped unmerged, naming GitHub's reason and, for a failed `merge_group` build, the failing jobs and their logs' tails. A failed build or a conflict is recorded for the land fixer; a push since the removal, a person's removal, or an unknown reason ends the pick. A pick is the latest enqueue by anyone, the App included (as for Dependabot's PRs), except the App's enqueues of heads the watcher itself put back in the queue for the pick. With --fixer, a conflict is first merged with the base by the quest merge driver; otherwise the fixer runs in the background on the failure, at most --fixer-jobs at once, oldest removal first. A fixed or merged head is lease-pushed and recorded in Review, and each fixed, merged, or flaky head is re-enqueued at the back of the queue (or armed with auto-merge while its checks run). The pick keeps going while each round makes progress: a round with no new commit for a failure that isn't flaky, the failure a fix addressed recurring right after it, or three flaky rounds in a row hand it back, as does the fixer, with a comment. State keeps each removal handled, so a restart neither misses nor repeats one, each removal's full log, the fixer's jobs, and each pick's rounds. It shares `quest land serve`'s App key, state directory, private repository, and locks, so the two never run beside each other. The App needs `actions: read` for the logs, and with a fixer `merge_queues: write` to re-enqueue, `contents: write`, and `workflows: write` to push fixes that touch workflows.", ) .option("--base ", "Base branch (default: repository default)") .option("--state ", "Private daemon state directory (default: $STATE_DIRECTORY or the XDG state home)") .requiredOption("--app-id ", "GitHub App id") + .option( + "--fixer ", + "Credential-free fixer wrapper executable, called with the failure log and the PR number (60 minutes from its admission line, 120 in total)", + ) + .option("--fixer-jobs ", "Most fixer jobs to run at once", positiveInteger("fixer-jobs"), 2) + .addOption( + new Option( + "--fixer-effort ", + `The fixer's model_reasoning_effort, passed to the wrapper in $${EFFORT_VARIABLE}`, + ) + .choices(EFFORTS) + .default("medium"), + ) .action((options: WatchOptions) => { const origin = readRemote(root(), "origin"); const repository = origin && remoteRepository(origin).repository; diff --git a/test/land-watch.test.ts b/test/land-watch.test.ts index de41e9af..2e24e1d2 100644 --- a/test/land-watch.test.ts +++ b/test/land-watch.test.ts @@ -1,13 +1,23 @@ // `quest land watch` against a fake of the GitHub records it reads: merge queue events and `merge_group` builds. -import { readFileSync, writeFileSync } from "node:fs"; +import { execFileSync } from "node:child_process"; +import { chmodSync, existsSync, mkdirSync, readdirSync, readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; +import { setTimeout as pause } from "node:timers/promises"; +import { fileURLToPath } from "node:url"; -import { expect, test } from "vitest"; +import { beforeEach, describe, expect, test, vi } from "vitest"; import type { GitHubDependencies } from "../src/cli/github"; -import { jobsSection, readRemoval, SETTLE, watchPass } from "../src/cli/land-watch"; -import { tempDir } from "./fixture"; +import { readRemote } from "../src/cli/land"; +import { FLAKY_LIMIT, history } from "../src/cli/land-fixer"; +import { privateRepository } from "../src/cli/land-serve"; +import { jobsSection, readRemoval, SETTLE, watchPass, watchPermissions } from "../src/cli/land-watch"; +import type { FixContext } from "../src/cli/land-watch-fix"; +import { configure, GIT_SETTINGS, initRepository, tempDir } from "./fixture"; + +// The fixer tests merge with the quest driver and run fixer wrappers, which take longer than vitest's default. +vi.setConfig({ testTimeout: 30_000 }); const REPOSITORY = { owner: "upstream", repo: "project" }; const BASE = "b".repeat(40); @@ -31,9 +41,20 @@ interface Added { } interface Pull { number: number; + /** Its head, unless the world has an upstream repository, where its branch holds it. */ head: string; removals: Removed[]; enqueues: Added[]; + state?: string; + /** `mergeStateStatus`, CLEAN by default. */ + mergeState?: string; + /** Its base branch, main by default. */ + base?: string; + /** When it joined the merge queue, while it is in it. */ + queued?: string; + /** When auto-merge was armed, while it is. */ + armed?: string; + body?: string; } interface Job { id: number; @@ -58,13 +79,30 @@ interface World { comments: Record; /** Requests the fake answered, as `METHOD path`. */ requests: string[]; + /** Each mutation, as `name number head`. */ + mutations: string[]; now: number; + /** The upstream repository that holds each PR's branch, `branch-`. */ + origin?: string; + /** A comment GitHub refuses to post, once. */ + refuse?: (body: string) => boolean; + /** A read of a PR's queue state that fails, once. */ + unreadable?: (pull: Pull) => boolean; } const at = (minutes: number) => new Date(Date.UTC(2026, 10, 1, 12, minutes)).toISOString().replace(".000", ""); function world(pulls: Pull[], runs: Run[] = []): World { - return { pulls, runs, comments: {}, requests: [], now: Date.parse(at(0)) }; + return { pulls, runs, comments: {}, requests: [], mutations: [], now: Date.parse(at(0)) }; +} + +/** A PR's head: its branch in the upstream repository, when the world has one. */ +function headOf(world: World, pull: Pull): string { + return world.origin ? git(world.origin, "rev-parse", `refs/heads/branch-${pull.number}`) : pull.head; +} + +function git(cwd: string, ...args: string[]): string { + return execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); } function removalNode(event: Removed) { @@ -104,6 +142,40 @@ function deps(world: World): GitHubDependencies { if (query.includes("viewer")) return Response.json({ data: { viewer: { login: `${APP}[bot]` } } }); if (query.includes("defaultBranchRef")) return Response.json({ data: { repository: { defaultBranchRef: { name: "main" } } } }); + const pullOf = () => world.pulls.find((item) => `PR_${item.number}` === variables.id)!; + const mutation = /^\s*mutation[^{]*\{\s*(\w+)/.exec(query)?.[1]; + if (mutation === "enqueuePullRequest" || mutation === "enablePullRequestAutoMerge") { + const pull = pullOf(); + world.mutations.push(`${mutation} ${pull.number} ${variables.sha}`); + if (variables.sha !== headOf(world, pull)) + return Response.json({ errors: [{ type: "UNPROCESSABLE", message: "head moved" }] }); + const now = new Date(world.now).toISOString().replace(".000", ""); + if (mutation === "enablePullRequestAutoMerge") pull.armed = now; + else { + pull.queued = now; + pull.enqueues.push({ createdAt: now, actor: APP }); + } + return Response.json({ data: { [mutation]: { clientMutationId: null } } }); + } + if (query.includes("mergeStateStatus")) { + const pull = pullOf(); + if (world.unreadable?.(pull)) { + world.unreadable = undefined; + return Response.json({ errors: [{ type: "INTERNAL", message: "unavailable" }] }); + } + return Response.json({ + data: { + node: { + state: pull.state ?? "OPEN", + baseRefName: pull.base ?? "main", + headRefOid: headOf(world, pull), + mergeStateStatus: pull.mergeState ?? "CLEAN", + mergeQueueEntry: pull.queued ? { enqueuedAt: pull.queued } : null, + autoMergeRequest: pull.armed ? { enabledAt: pull.armed } : null, + }, + }, + }); + } if (query.includes("ADDED_TO_MERGE_QUEUE_EVENT")) { const pull = world.pulls.find((item) => `PR_${item.number}` === variables.id)!; const nodes = pull.enqueues.map((event) => ({ @@ -124,14 +196,18 @@ function deps(world: World): GitHubDependencies { return Response.json({ data: { repository: { - ref: { target: { oid: BASE } }, + ref: { + target: { + oid: world.origin ? git(world.origin, "rev-parse", "refs/heads/main") : BASE, + }, + }, pullRequests: { nodes: world.pulls.map((pull) => ({ id: `PR_${pull.number}`, number: pull.number, url: `https://github.com/upstream/project/pull/${pull.number}`, title: `PR ${pull.number}`, - headRefOid: pull.head, + headRefOid: headOf(world, pull), headRefName: `branch-${pull.number}`, maintainerCanModify: false, headRepository: { nameWithOwner: "upstream/project" }, @@ -175,11 +251,25 @@ function deps(world: World): GitHubDependencies { const job = world.runs.flatMap((run) => run.jobs).find((item) => item.id === Number(logs[1])); return job?.log === undefined ? new Response("missing", { status: 404 }) : new Response(job.log); } + const pullPath = /^\/repos\/upstream\/project\/pulls\/(\d+)$/.exec(path); + if (pullPath) { + const pull = world.pulls.find((item) => item.number === Number(pullPath[1]))!; + if (init?.method === "PATCH") pull.body = JSON.parse(String(init.body)).body; + return Response.json({ + number: pull.number, + head: { sha: headOf(world, pull) }, + body: pull.body ?? "", + }); + } const comments = /^\/repos\/upstream\/project\/issues\/(\d+)\/comments$/.exec(path); if (comments) { const list = (world.comments[Number(comments[1])] ??= []); if (init?.method === "POST") { const comment = { user: { login: `${APP}[bot]` }, body: JSON.parse(String(init.body)).body }; + if (world.refuse?.(comment.body)) { + world.refuse = undefined; + return Response.json({ message: "refused" }, { status: 422 }); + } list.push(comment); return Response.json(comment, { status: 201 }); } @@ -190,10 +280,10 @@ function deps(world: World): GitHubDependencies { }; } -const failedRun = (number: number, created: string): Run => ({ +const failedRun = (number: number, created: string, built = BUILT): Run => ({ id: 7, name: "Check", - head_branch: `gh-readonly-queue/main/pr-${number}-${BUILT}`, + head_branch: `gh-readonly-queue/main/pr-${number}-${built}`, head_sha: "e".repeat(40), created_at: created, conclusion: "failure", @@ -217,6 +307,14 @@ const failedRun = (number: number, created: string): Run => ({ test("a failed merge_group build is commented with its failing job's log and recorded for the fixer", async () => { const state = tempDir(); + // The watcher put HEAD back in the queue in an earlier round of the maintainer's pick. + writeFileSync( + join(state, "fix-attempt-5.json"), + JSON.stringify({ + pick: at(1), + rounds: [{ job: "watch-job-1.json", outcome: "flaky", before: HEAD, after: HEAD, failure: "just test" }], + }), + ); const github = world( [ { @@ -224,7 +322,7 @@ test("a failed merge_group build is commented with its failing job's log and rec head: HEAD, enqueues: [ { createdAt: at(1), actor: "maintainer" }, - // The App re-enqueuing a fix continues the maintainer's pick. + // The App re-enqueuing that head continues the maintainer's pick. { createdAt: at(2), actor: APP }, ], removals: [{ id: "R1", createdAt: at(5), reason: "failed_checks", actor: "github-merge-queue" }], @@ -454,9 +552,602 @@ test("the first pass reads removals from SETTLE before it, and forgets closed PR expect(JSON.parse(readFileSync(join(state, "watch.json"), "utf8")).handled).toEqual({}); }); +test("the watcher's tokens ask for writes only with a fixer, and pushes for workflows only when they carry one", () => { + expect(watchPermissions(false)).toEqual({ api: { contents: "read", pull_requests: "write", actions: "read" } }); + expect(watchPermissions(true)).toEqual({ + api: { contents: "write", pull_requests: "write", actions: "read", merge_queues: "write" }, + push: { contents: "write", pull_requests: "write" }, + pushWorkflows: { contents: "write", pull_requests: "write", workflows: "write" }, + read: { contents: "read" }, + }); +}); + test("failing jobs that do not fit the comment are counted instead", () => { expect(jobsSection(["a".repeat(50), "b".repeat(50)], 1000)).toBe("a".repeat(50) + "b".repeat(50)); const cut = jobsSection(["a".repeat(500), "b".repeat(500), "c"], 800); expect(cut).toBe(`${"a".repeat(500)}\n\n2 more failing jobs do not fit in this comment.`); expect(cut.length).toBeLessThanOrEqual(800); }); + +const bin = fileURLToPath(new URL("../src/cli/bin.ts", import.meta.url)); +const bunfig = fileURLToPath(new URL("../bunfig.toml", import.meta.url)); +const PATH = process.env.PATH; + +const EPIC = (entries: string[]) => + `# Epic\n\n## Goal\n\nAn epic.\n\n## Required\n\n${entries.map((entry) => `- [${entry}](/quest/m0/${entry}.md)`).join("\n")}\n`; + +/** github.com's `upstream/project` on disk: main, and each PR's branch `branch-`, also at `refs/pull//head`. */ +function upstream() { + const dir = tempDir(); + const hub = join(dir, "github"); + const origin = initRepository(join(hub, "upstream", "project.git"), { bare: true }); + const work = initRepository(join(dir, "work")); + const commit = (files: Record, message: string) => { + for (const [path, content] of Object.entries(files)) { + mkdirSync(join(work, path, ".."), { recursive: true }); + writeFileSync(join(work, path), content); + } + git(work, "add", "."); + git(work, "commit", "--quiet", "-m", message); + return git(work, "rev-parse", "HEAD"); + }; + commit( + { + ".gitattributes": "quest/** merge=quest\n", + "quest/README.md": "# Quests\n\n## Goal\n\nRoot.\n\n## Required\n\n- [M0](/quest/m0/README.md)\n", + "quest/m0/README.md": EPIC(["a"]), + "quest/m0/a.md": "# [S] A\n\n## Goal\n\nA.\n", + "code.txt": "one\ntwo\nthree\n", + }, + "base", + ); + git(work, "push", "--quiet", origin, "HEAD:refs/heads/main"); + const main = () => git(origin, "rev-parse", "refs/heads/main"); + return { + hub, + origin, + main, + /** Open PR `number` on main as it is now. Returns its head. */ + pull(number: number, files: Record): string { + git(work, "checkout", "--quiet", "-B", `branch-${number}`, main()); + const head = commit(files, `PR ${number}`); + git( + work, + "push", + "--quiet", + origin, + `${head}:refs/heads/branch-${number}`, + `${head}:refs/pull/${number}/head`, + ); + return head; + }, + /** A writer's push to PR `number`. */ + push(number: number, files: Record): string { + git( + work, + "checkout", + "--quiet", + "-B", + `branch-${number}`, + git(origin, "rev-parse", `refs/heads/branch-${number}`), + ); + const head = commit(files, `more of PR ${number}`); + git(work, "push", "--quiet", origin, `${head}:refs/heads/branch-${number}`); + return head; + }, + /** Land something else on main. Returns the new base. */ + advance(files: Record): string { + git(work, "checkout", "--quiet", "-B", "main", main()); + const head = commit(files, "advance main"); + git(work, "push", "--quiet", origin, "HEAD:refs/heads/main"); + return head; + }, + }; +} + +/** + * A fixer wrapper that runs `/round-`, the test's script for the PR's next round, in the + * fixer's checkout, recording each call's arguments and effort in `/calls`. + */ +function fixer(): { + wrapper: string; + tools: string; + round: (number: number, script: string) => void; + calls: () => string[]; +} { + const tools = tempDir(); + const wrapper = join(tools, "wrapper"); + writeFileSync( + wrapper, + `#!/bin/sh +set -eu +[ "$#" = 2 ] +printf '%s %s %s\\n' "$QUEST_LAND_FIXER_EFFORT" "$1" "$2" >> '${tools}/calls' +. '${tools}/round-'"$2" +`, + { mode: 0o755 }, + ); + return { + wrapper, + tools, + round: (number, script) => writeFileSync(join(tools, `round-${number}`), script), + calls: () => + existsSync(join(tools, "calls")) ? readFileSync(join(tools, "calls"), "utf8").trim().split("\n") : [], + }; +} + +/** A fixer round that commits `file` on the PR's branch and reports it fixed, naming `failure`. */ +const fixScript = (file: string, failure = "just test: queue.test.ts") => `git checkout -q "pr-$2" +printf 'fixed\\n' > '${file}' +git add '${file}' +git commit -qm 'mechanical fix' +printf '[{"number":%s,"action":"fixed","head":"%s","reason":"add the missing file","failure":"${failure}"}]' "$2" "$(git rev-parse HEAD)" > quest-land-result.json +`; +const flakyScript = `printf '[{"number":%s,"action":"requeue","reason":"healthy; flaky test","failure":"just test: render_timeout","comment":"render_timeout timed out under load."}]' "$2" > quest-land-result.json +`; + +/** Wait for `count` fixer jobs to have exited. */ +async function exited(state: string, count = 1): Promise { + for (let tries = 0; tries < 3000; tries++) { + if (readdirSync(state).filter((name) => /^watch-job-.*\.json\.exit$/.test(name)).length >= count) return; + await pause(10); + } + throw new Error("the fixer never exited"); +} + +// The fixer, like the daemon, needs Linux process identities. +describe.skipIf(process.platform === "win32")("the watcher's land fixer", () => { + beforeEach(() => { + // The merge driver runs `quest resolve` from PATH, as on the host. + const tools = tempDir(); + writeFileSync(join(tools, "quest"), `#!/bin/sh\nexec bun --config='${bunfig}' '${bin}' "$@"\n`); + chmodSync(join(tools, "quest"), 0o755); + process.env.PATH = `${tools}:${PATH}`; + }); + + function setup(jobs = 2) { + const repo = upstream(); + const state = tempDir(); + const root = privateRepository(state); + configure(root, { [`url.${repo.hub}/.insteadOf`]: "https://github.com/", ...GIT_SETTINGS }); + const fix = fixer(); + const context: FixContext = { + root, + fixer: fix.wrapper, + jobs, + effort: "high", + timeout: 60, + token: async () => "token", + read: readRemote, + }; + return { repo, state, fix, context }; + } + + test("a failed build is fixed and re-enqueued, and the same failure right after the fix is handed back", async () => { + const { repo, state, fix, context } = setup(); + const base = repo.main(); + const head = repo.pull(5, { "feature.txt": "new\n" }); + const pull: Pull = { + number: 5, + head, + enqueues: [{ createdAt: at(1), actor: "maintainer" }], + removals: [ + { id: "R1", createdAt: at(5), reason: "failed_checks", actor: "github-merge-queue", before: head }, + ], + body: "A feature.\n\n## Review\n\nreviewed by Codex (head x)\n", + }; + const github = { ...world([pull], [failedRun(5, at(3), base)]), origin: repo.origin }; + github.now = Date.parse(at(6)); + fix.round(5, fixScript("fixed.txt")); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + "removed 5 failed_checks fix", + "fixing 5", + ]); + expect(github.comments[5][0].body).toContain("The land fixer works on it in the background"); + await exited(state); + // The wrapper got the removal's full log, the PR, and the effort. + expect(fix.calls()).toEqual([`high ${readRemoval(state, 5)!.log} 5`]); + + github.now = Date.parse(at(10)); + // GitHub takes the head back, then fails the read that follows: the round may be in the queue, so it is + // retried rather than handed back, and finds its own enqueue. + github.unreadable = (pull) => pull.queued !== undefined; + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([]); + // Then GitHub refuses the outcome comment once: the round stays requeued, and only its comment is retried. + github.refuse = (body) => body.includes("re-enqueued this pull request"); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([]); + const [line] = await watchPass(REPOSITORY, { state }, deps(github), context); + const fixed = git(repo.origin, "rev-parse", "refs/heads/branch-5"); + expect(line).toBe(`requeued 5 ${fixed}`); + expect(git(repo.origin, "rev-parse", `${fixed}^`)).toBe(head); + expect(github.mutations).toEqual([`enqueuePullRequest 5 ${fixed}`]); + expect(pull.body).toContain(`mechanical fix by Codex: \`add the missing file\` (head ${fixed})`); + expect(github.comments[5].at(-1)!.body).toContain( + `re-enqueued this pull request at ${fixed}, at the back of the merge queue`, + ); + expect(history(state, 5)).toMatchObject({ + pick: at(1), + rounds: [{ outcome: "fixed", before: head, after: fixed, failure: "just test: queue.test.ts" }], + }); + // The job and its checkout are gone; the fixer's output log stays for diagnosis. + expect(readdirSync(state).filter((name) => /^watch-job-.*\.json$|^fix-(?!attempt-)/.test(name))).toEqual([]); + expect(readdirSync(state).filter((name) => name.endsWith(".json.log"))).toHaveLength(1); + + // The fixed head fails the same way: the App's re-enqueue continues the maintainer's pick, and the + // fixer's second fix of the same failure is a repeat, handed back unpublished. + pull.queued = undefined; + pull.removals.push({ + id: "R2", + createdAt: at(20), + reason: "failed_checks", + actor: "github-merge-queue", + before: fixed, + }); + github.runs.push({ ...failedRun(5, at(15), base), id: 8, head_sha: "f".repeat(40) }); + github.now = Date.parse(at(21)); + fix.round(5, fixScript("again.txt")); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + "removed 5 failed_checks fix", + "fixing 5", + ]); + expect(readRemoval(state, 5)!.pick).toEqual({ by: "maintainer", at: at(1) }); + await exited(state); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual(["handback 5"]); + expect(git(repo.origin, "rev-parse", "refs/heads/branch-5")).toBe(fixed); + expect(github.mutations).toHaveLength(1); + const handback = github.comments[5].at(-1)!.body; + expect(handback).toContain("stop rule (repeated failure)"); + expect(handback).toContain(""); + expect(history(state, 5)!.rounds.map((round) => round.outcome)).toEqual(["fixed", "handback"]); + expect(history(state, 5)!.rounds[1].rule).toBe("repeated"); + }); + + test("a quest-file conflict is merged with the quest driver, and a code conflict goes to the fixer", async () => { + const { repo, state, fix, context } = setup(1); + const quests = repo.pull(3, { + "quest/m0/README.md": EPIC(["a", "c"]), + "quest/m0/c.md": "# [S] C\n\n## Goal\n\nC.\n", + }); + const code = repo.pull(4, { "code.txt": "one\nII\nthree\n" }); + const failing = repo.pull(6, { "feature.txt": "new\n" }); + const built = repo.main(); + const base = repo.advance({ + "quest/m0/README.md": EPIC(["a", "b"]), + "quest/m0/b.md": "# [S] B\n\n## Goal\n\nB.\n", + "code.txt": "one\n2\nthree\n", + }); + const conflict = (number: number, head: string, minute: number): Pull => ({ + number, + head, + enqueues: [{ createdAt: at(1), actor: "maintainer" }], + removals: [ + { + id: `R${number}`, + createdAt: at(minute), + reason: "merge_conflict", + actor: "github-merge-queue", + before: head, + }, + ], + // Its checks run again on the merge: auto-merge waits for them. + mergeState: "BLOCKED", + }); + const github = { + ...world( + [ + conflict(3, quests, 4), + conflict(4, code, 5), + { + number: 6, + head: failing, + enqueues: [{ createdAt: at(1), actor: "maintainer" }], + removals: [ + { + id: "R6", + createdAt: at(6), + reason: "failed_checks", + actor: "github-merge-queue", + before: failing, + }, + ], + }, + ], + [failedRun(6, at(3), built)], + ), + origin: repo.origin, + }; + github.now = Date.parse(at(7)); + fix.round( + 4, + `git checkout -q "pr-$2" +git merge -q base || true +printf 'one\\nII and 2\\nthree\\n' > code.txt +git add code.txt +git commit -qm 'preserve both sides' +printf '[{"number":%s,"action":"fixed","head":"%s","reason":"preserve both conflict sides"}]' "$2" "$(git rev-parse HEAD)" > quest-land-result.json +`, + ); + fix.round(6, flakyScript); + const merged = () => git(repo.origin, "rev-parse", "refs/heads/branch-3"); + const pull3Body = () => github.pulls.find((pull) => pull.number === 3)!.body; + // One fixer slot: the quest-file conflict needs none, the code conflict takes it, and the failed build waits. + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + "removed 3 merge_conflict fix", + "removed 4 merge_conflict fix", + "removed 6 failed_checks fix", + `merged 3 ${merged()}`, + "fixing 4", + `requeued 3 ${merged()}`, + ]); + expect(git(repo.origin, "show", `${merged()}:quest/m0/README.md`)).toBe(EPIC(["a", "b", "c"]).trim()); + expect(git(repo.origin, "rev-parse", `${merged()}^1`, `${merged()}^2`).split("\n")).toEqual([quests, base]); + expect(github.mutations).toEqual([`enablePullRequestAutoMerge 3 ${merged()}`]); + expect(github.comments[3].at(-1)!.body).toContain( + `armed auto-merge on this pull request at ${merged()}, so it joins the back of the merge queue once its required checks pass, with main at ${base} merged in by the quest merge driver`, + ); + expect(pull3Body()).toContain( + `mechanical fix by quest land watch: \`merged main at ${base} with the quest merge driver\``, + ); + expect(history(state, 3)!.rounds).toMatchObject([ + { outcome: "merged", failure: `merge conflict with ${base}` }, + ]); + await exited(state); + // The code conflict's fixer got the current base and the merge's conflict, never the build's base. + const [call] = fix.calls(); + expect(call).toMatch(new RegExp(`^high .*/conflict-4-${base.slice(0, 12)}\\.log 4$`)); + expect(readFileSync(call.split(" ")[1], "utf8")).toContain("CONFLICT (content): Merge conflict in code.txt"); + + github.now = Date.parse(at(8)); + const resolved = () => git(repo.origin, "rev-parse", "refs/heads/branch-4"); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + `requeued 4 ${resolved()}`, + "fixing 6", + ]); + expect(git(repo.origin, "show", `${resolved()}:code.txt`)).toBe("one\nII and 2\nthree"); + await exited(state); + expect(fix.calls()[1]).toBe(`high ${readRemoval(state, 6)!.log} 6`); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([`requeued 6 ${failing}`]); + expect(github.mutations.slice(1)).toEqual([ + `enablePullRequestAutoMerge 4 ${resolved()}`, + `enqueuePullRequest 6 ${failing}`, + ]); + const [note, requeued] = github.comments[6].slice(-2).map((comment) => comment.body); + expect(note).toContain("land fixer note:\n\n> render_timeout timed out under load."); + expect(requeued).toContain( + "unchanged, since the land fixer judged the failure flaky: `just test: render_timeout`", + ); + }); + + test("a Dependabot pick by the App is fixed like any other, until it is flaky three times in a row", async () => { + const { repo, state, fix, context } = setup(); + const base = repo.main(); + const head = repo.pull(7, { "Cargo.lock": "bumped\n" }); + // Its workflow enqueues it with the App's token. + const pull: Pull = { number: 7, head, enqueues: [{ createdAt: at(1), actor: APP }], removals: [] }; + const github = { ...world([pull], []), origin: repo.origin }; + fix.round(7, flakyScript); + for (let round = 1; round <= FLAKY_LIMIT; round++) { + const minute = round * 10; + pull.queued = undefined; + pull.removals.push({ + id: `R${round}`, + createdAt: at(minute), + reason: "failed_checks", + actor: "github-merge-queue", + before: head, + }); + github.runs.push({ ...failedRun(7, at(minute - 2), base), id: round }); + github.now = Date.parse(at(minute + 1)); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + "removed 7 failed_checks fix", + "fixing 7", + ]); + expect(readRemoval(state, 7)!.pick).toEqual({ by: `${APP}[bot]`, at: at(1) }); + await exited(state); + github.now = Date.parse(at(minute + 2)); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + round < FLAKY_LIMIT ? `requeued 7 ${head}` : "handback 7", + ]); + } + expect(github.mutations).toEqual([`enqueuePullRequest 7 ${head}`, `enqueuePullRequest 7 ${head}`]); + expect(github.comments[7].at(-1)!.body).toContain("stop rule (flaky limit)"); + expect(history(state, 7)!.rounds.map((round) => round.outcome)).toEqual(["flaky", "flaky", "handback"]); + + // The handback ended the pick: the workflow's next enqueue, even of the same head, is a new one. + pull.enqueues.push({ createdAt: at(50), actor: APP }); + pull.removals.push({ + id: "R9", + createdAt: at(55), + reason: "failed_checks", + actor: "github-merge-queue", + before: head, + }); + github.now = Date.parse(at(56)); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + "removed 7 failed_checks fix", + "fixing 7", + ]); + expect(readRemoval(state, 7)!.pick).toEqual({ by: `${APP}[bot]`, at: at(50) }); + expect(history(state, 7)).toEqual({ pick: at(50), rounds: [] }); + await exited(state); + }); + + test("a pick that ends while the fixer runs is left alone, and every handback ends the pick", async () => { + const { repo, state, fix, context } = setup(5); + const base = repo.main(); + const moved = repo.pull(8, { "feature.txt": "new\n" }); + const stuck = repo.pull(9, { "other.txt": "new\n" }); + const again = repo.pull(10, { "again.txt": "new\n" }); + const crashed = repo.pull(11, { "crash.txt": "new\n" }); + const retargeted = repo.pull(13, { "retarget.txt": "new\n" }); + const waiting = repo.pull(14, { "waiting.txt": "new\n" }); + const removed = (number: number, head: string): Pull => ({ + number, + head, + enqueues: [{ createdAt: at(1), actor: "maintainer" }], + removals: [ + { + id: `R${number}`, + createdAt: at(5), + reason: "failed_checks", + actor: "github-merge-queue", + before: head, + }, + ], + }); + const pulls = [ + removed(8, moved), + removed(9, stuck), + removed(10, again), + removed(11, crashed), + removed(13, retargeted), + // Retargeted before its fixer could start. + { ...removed(14, waiting), base: "release" }, + ]; + const github = { + ...world( + pulls, + pulls.map((pull) => ({ ...failedRun(pull.number, at(3), base), id: pull.number })), + ), + origin: repo.origin, + }; + github.now = Date.parse(at(6)); + fix.round(8, fixScript("fixed.txt")); + fix.round( + 9, + `printf '[{"number":%s,"action":"handback","diagnosis":"the test asserts the old behavior","recommendation":"decide which behavior is right","log":[1,1]}]' "$2" > quest-land-result.json\n`, + ); + fix.round(10, fixScript("fixed.txt")); + fix.round(11, "echo 'agent crashed' >&2\nexit 3\n"); + fix.round(13, fixScript("fixed.txt")); + expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([ + "removed 8 failed_checks fix", + "removed 9 failed_checks fix", + "removed 10 failed_checks fix", + "removed 11 failed_checks fix", + "removed 13 failed_checks fix", + "removed 14 failed_checks fix", + "fixing 8", + "fixing 9", + "fixing 10", + "fixing 11", + "fixing 13", + "dropped 14 retargeted", + ]); + await exited(state, 5); + // 13 was retargeted while its fixer ran. + pulls[4].base = "release"; + // While the fixers ran, a writer pushed to 8, and a maintainer enqueued 10 again, which the queue + // dropped again before the watcher collected the first round. + const pushed = repo.push(8, { "feature.txt": "newer\n" }); + pulls[2].enqueues.push({ createdAt: at(7), actor: "maintainer" }); + pulls[2].removals.push({ + id: "R10b", + createdAt: at(8), + reason: "failed_checks", + actor: "github-merge-queue", + before: again, + }); + github.runs.push({ ...failedRun(10, at(7), base), id: 100 }); + github.now = Date.parse(at(9)); + fix.round(10, flakyScript); + expect((await watchPass(REPOSITORY, { state }, deps(github), context)).sort()).toEqual([ + "dropped 10 ended", + "dropped 13 ended", + "dropped 8 ended", + "fixing 10", + "handback 11", + "handback 9", + "removed 10 failed_checks fix", + ]); + expect(git(repo.origin, "rev-parse", "refs/heads/branch-8")).toBe(pushed); + expect(git(repo.origin, "rev-parse", "refs/heads/branch-10")).toBe(again); + expect(github.mutations).toEqual([]); + expect(github.comments[8].at(-1)!.body).toContain(`its head moved to ${pushed} while the fixer ran`); + expect(github.comments[13].at(-1)!.body).toContain("it was retargeted to `release` while the fixer ran"); + expect(git(repo.origin, "rev-parse", "refs/heads/branch-13")).toBe(retargeted); + expect(fix.calls().some((call) => call.endsWith(" 14"))).toBe(false); + expect(github.comments[10].at(-1)!.body).toContain("the pick it was fixing ended"); + const handback = github.comments[9].at(-1)!.body; + expect(handback).toContain( + "Codex land fixer: `the test asserts the old behavior. Recommendation: decide which behavior is right`", + ); + expect(handback).toContain("
The merge queue removed");
+		expect(history(state, 9)!.rounds).toMatchObject([{ outcome: "handback" }]);
+		// A crash ends the pick too, with a round of its own.
+		expect(github.comments[11].at(-1)!.body).toContain("fixer crashed (exit 3)");
+		expect(history(state, 11)!.rounds).toMatchObject([
+			{ outcome: "handback", before: crashed, failure: "Check / test: just test" },
+		]);
+		// The newer pick of 10 starts its own history and round; 8, 9 and 11 are not taken again.
+		expect(history(state, 10)).toEqual({ pick: at(7), rounds: [] });
+		await exited(state);
+		github.now = Date.parse(at(10));
+		expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([`requeued 10 ${again}`]);
+		expect(history(state, 10)!.rounds).toMatchObject([{ outcome: "flaky" }]);
+		expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([]);
+	});
+
+	test("a person's removal ends the pick, even of a round re-enqueued just before a restart", async () => {
+		const { repo, state, fix, context } = setup();
+		const base = repo.main();
+		const head = repo.pull(12, { "Cargo.lock": "bumped\n" });
+		const pull: Pull = {
+			number: 12,
+			head,
+			enqueues: [{ createdAt: at(1), actor: APP }],
+			removals: [
+				{ id: "R1", createdAt: at(5), reason: "failed_checks", actor: "github-merge-queue", before: head },
+			],
+		};
+		const github = { ...world([pull], [failedRun(12, at(3), base)]), origin: repo.origin };
+		github.now = Date.parse(at(6));
+		fix.round(12, flakyScript);
+		await watchPass(REPOSITORY, { state }, deps(github), context);
+		await exited(state);
+		// The daemon re-enqueued the round's head, then stopped before it finished the job; a maintainer
+		// removed the PR from the queue before it restarted.
+		const [job] = readdirSync(state).filter((name) => /^watch-job-.*\.json$/.test(name));
+		const record = JSON.parse(readFileSync(join(state, job), "utf8"));
+		const result = JSON.parse(readFileSync(join(record.checkout, "quest-land-result.json"), "utf8"))[0];
+		writeFileSync(join(state, job), JSON.stringify({ ...record, result, requeuing: head }));
+		writeFileSync(
+			join(state, "fix-attempt-12.json"),
+			JSON.stringify({
+				pick: at(1),
+				rounds: [{ job, outcome: "flaky", before: head, after: head, failure: "just test: render_timeout" }],
+			}),
+		);
+		pull.enqueues.push({ createdAt: at(7), actor: APP });
+		pull.removals.push({ id: "R2", createdAt: at(8), reason: "manual", actor: "maintainer", before: head });
+		github.now = Date.parse(at(9));
+		expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([
+			"removed 12 manual removed",
+			`requeued 12 ${head}`,
+		]);
+		// Nothing enqueued it again, and the pick's history is gone with it.
+		expect(github.mutations).toEqual([]);
+		expect(history(state, 12)).toBeUndefined();
+
+		// The workflow's next enqueue of the same head is a new pick, with no flaky rounds behind it.
+		pull.enqueues.push({ createdAt: at(20), actor: APP });
+		pull.removals.push({
+			id: "R3",
+			createdAt: at(25),
+			reason: "failed_checks",
+			actor: "github-merge-queue",
+			before: head,
+		});
+		github.runs.push({ ...failedRun(12, at(23), base), id: 13 });
+		github.now = Date.parse(at(26));
+		expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([
+			"removed 12 failed_checks fix",
+			"fixing 12",
+		]);
+		expect(readRemoval(state, 12)!.pick).toEqual({ by: `${APP}[bot]`, at: at(20) });
+		expect(history(state, 12)).toEqual({ pick: at(20), rounds: [] });
+		await exited(state);
+		github.now = Date.parse(at(27));
+		expect(await watchPass(REPOSITORY, { state }, deps(github), context)).toEqual([`requeued 12 ${head}`]);
+	});
+});
diff --git a/test/land.test.ts b/test/land.test.ts
index a2739cf1..bb16d034 100644
--- a/test/land.test.ts
+++ b/test/land.test.ts
@@ -3135,6 +3135,12 @@ ${requeueScript}
 		);
 		await fix(dir, log, [1], agent);
 		expect(readFileSync(join(dir, "argv.txt"), "utf8").split("\n")).toContain('model_reasoning_effort="medium"');
+		// A daemon's --fixer-effort reaches it as --effort.
+		await fix(dir, log, [1], agent, undefined, undefined, "high");
+		expect(readFileSync(join(dir, "argv.txt"), "utf8").split("\n")).toContain('model_reasoning_effort="high"');
+		await expect(fix(dir, log, [1], agent, undefined, undefined, 'high"\nx')).rejects.toThrow(
+			"fix effort must be one of",
+		);
 	});
 	test("refuses to start an agent without a supplied PR's branch", async () => {
 		const dir = fixRepository();