Fix the runner's post-merge review findings

README.md

@@ -528,7 +528,7 @@ another maintainer's answer only once you approve it in the inbox.
 The flake's `runner` package is a `quest` whose Claude Code is pinned by this
 repository's lock (`nix run github:kixelated/quest#runner -- runner`);
 otherwise `claude` comes from `QUEST_RUNNER_CLAUDE` or `PATH`. State lives in `$QUEST_RUNNER_HOME`, by default
-`$XDG_DATA_HOME/quest/runner`. To keep a runner up, run `quest runner` as a user
+`$XDG_DATA_HOME/quest/runner`; one `quest runner` runs in a home at a time. To keep a runner up, run `quest runner` as a user
 service, such as a systemd user unit with `Restart=on-failure`.
 
 ## Prunable local checkouts

cloudflare/src/github/service.ts

@@ -872,12 +872,15 @@ export class GitHubService {
 		const permissions: Permissions =
 			kind === "issues" ? { issues: "read" } : { pull_requests: "read", checks: "read", statuses: "read" };
 		const comments = "comments(last: 100) { nodes { author { login } body createdAt } }";
+		// One page of review threads; a pull request's later pages are read `after` the one before.
+		const threads =
+			"reviewThreads(first: 100, after: $threads) { nodes { isResolved isOutdated path line comments(first: 50) { nodes { author { login } body createdAt } } } pageInfo { hasNextPage endCursor } }";
 		const query =
 			kind === "issues"
 				? `issue(number: $number) { number title state body author { login } ${comments} }`
 				: `pullRequest(number: $number) { number title state body isDraft author { login } baseRefName headRefName headRefOid ${comments}
  reviews(last: 50) { nodes { author { login } state body submittedAt } }
- reviewThreads(first: 100) { nodes { isResolved isOutdated path line comments(first: 50) { nodes { author { login } body createdAt } } } }
+ ${threads}
  commits(last: 1) { nodes { commit { oid statusCheckRollup { state contexts(first: 100) { nodes { __typename ... on CheckRun { name status conclusion detailsUrl } ... on StatusContext { context state description targetUrl } } } } } } } }`;
 		type Comment = { author: { login: string } | null; body: string; createdAt?: string; submittedAt?: string };
 		type Thread = {
@@ -910,23 +913,45 @@ export class GitHubService {
 			headRefOid?: string;
 			comments: { nodes: Comment[] };
 			reviews?: { nodes: (Comment & { state: string })[] };
-			reviewThreads?: { nodes: Thread[] };
+			reviewThreads?: { nodes: Thread[]; pageInfo: { hasNextPage: boolean; endCursor: string | null } };
 			commits?: {
 				nodes: {
 					commit: { oid: string; statusCheckRollup: { state: string; contexts: { nodes: Check[] } } | null };
 				}[];
 			};
 		};
 		return this.#authorized(repo, userId, permissions, async (token) => {
-			const result = await this.#api<{
-				data?: { repository: { issue?: Item | null; pullRequest?: Item | null } | null };
-				errors?: unknown[];
-			}>(token, "/graphql", "POST", {
-				query: `query($owner: String!, $name: String!, $number: Int!) { repository(owner: $owner, name: $name) { ${query} } }`,
-				variables: { owner: repo.owner, name: repo.name, number: number(Number(value)) },
-			});
-			const item = result.data?.repository?.[kind === "issues" ? "issue" : "pullRequest"];
-			if (result.errors?.length || !item) throw new GitHubError("This issue or pull request is unavailable");
+			const read = async (fields: string, after: string | null = null) => {
+				const pull = kind === "pulls";
+				const result = await this.#api<{
+					data?: { repository: { issue?: Item | null; pullRequest?: Item | null } | null };
+					errors?: unknown[];
+				}>(token, "/graphql", "POST", {
+					query: `query($owner: String!, $name: String!, $number: Int!${pull ? ", $threads: String" : ""}) { repository(owner: $owner, name: $name) { ${fields} } }`,
+					variables: {
+						owner: repo.owner,
+						name: repo.name,
+						number: number(Number(value)),
+						...(pull && { threads: after }),
+					},
+				});
+				const item = result.data?.repository?.[pull ? "pullRequest" : "issue"];
+				if (result.errors?.length || !item) throw new GitHubError("This issue or pull request is unavailable");
+				return item;
+			};
+			const item = await read(query);
+			// Every thread is read before the unresolved ones are kept: a busy pull request has more than a page.
+			const all: Thread[] = [];
+			let after: string | null = null;
+			for (let page = item.reviewThreads; page;) {
+				all.push(...page.nodes);
+				if (!page.pageInfo.hasNextPage) break;
+				if (!page.pageInfo.endCursor || page.pageInfo.endCursor === after)
+					throw new GitHubError("This pull request's review threads are unavailable");
+				after = page.pageInfo.endCursor;
+				page = (await read(`pullRequest(number: $number) { ${threads} }`, after)).reviewThreads;
+				if (!page) throw new GitHubError("This pull request's review threads are unavailable");
+			}
 			const comment = (c: Comment) => ({
 				author: c.author?.login ?? null,
 				body: c.body,
@@ -950,7 +975,7 @@ export class GitHubService {
 				head: { ref: item.headRefName ?? null, oid: item.headRefOid ?? null },
 				reviews: (item.reviews?.nodes ?? []).map((r) => ({ ...comment(r), state: r.state })),
 				// Resolved threads need nothing more; outdated ones may still need an answer.
-				threads: (item.reviewThreads?.nodes ?? [])
+				threads: all
 					.filter((t) => !t.isResolved)
 					.map((t) => ({
 						path: t.path,

cloudflare/test/publish.test.ts

@@ -18,6 +18,16 @@ const upstream = { owner: "upstream", name: "project", id: 10 };
 type Request = { method: string; path: string; body: any; token: string | null };
 type Commit = { parents: string[]; tree: string; message: string };
 
+function thread(path: string, line: number, body: string, isResolved: boolean) {
+	return {
+		isResolved,
+		isOutdated: false,
+		path,
+		line,
+		comments: { nodes: [{ author: { login: "bot" }, body, createdAt: "t" }] },
+	};
+}
+
 /** A stateful stand-in for the parts of GitHub's API that publication uses. */
 function fakeGitHub() {
 	let next = 1;
@@ -46,7 +56,15 @@ function fakeGitHub() {
 		now: 0,
 		/** Runs while a branch is created, as if the job changed meanwhile. */
 		onCreate: null as null | (() => Promise<unknown>),
+		/** The iterated pull request's review threads, read 100 at a time. */
+		threads: [thread("a.ts", 1, "Done already", true), thread("b.ts", 7, "Handle null", false)],
 	};
+	function threadPage(after: string | null) {
+		const start = after ? Number(after) : 0;
+		const nodes = fake.threads.slice(start, start + 100);
+		const end = start + nodes.length;
+		return { nodes, pageInfo: { hasNextPage: end < fake.threads.length, endCursor: String(end) } };
+	}
 	const repos: Record<string, any> = {
 		"upstream/project": { id: 10, fork: false, default_branch: "main" },
 		"quest-bot-forks/project": { id: 20, fork: true, default_branch: "main" },
@@ -120,28 +138,7 @@ function fakeGitHub() {
 							reviews: {
 								nodes: [{ author: { login: "bot" }, state: "COMMENTED", body: "", submittedAt: "t" }],
 							},
-							reviewThreads: {
-								nodes: [
-									{
-										isResolved: true,
-										isOutdated: false,
-										path: "a.ts",
-										line: 1,
-										comments: {
-											nodes: [{ author: { login: "bot" }, body: "Done already", createdAt: "t" }],
-										},
-									},
-									{
-										isResolved: false,
-										isOutdated: false,
-										path: "b.ts",
-										line: 7,
-										comments: {
-											nodes: [{ author: { login: "bot" }, body: "Handle null", createdAt: "t" }],
-										},
-									},
-								],
-							},
+							reviewThreads: threadPage(body.variables.threads),
 							commits: {
 								nodes: [
 									{
@@ -1253,6 +1250,26 @@ describe("job context", () => {
 		});
 	});
 
+	it("reads every review thread of a busy pull request", async () => {
+		github.threads = Array.from({ length: 250 }, (_, i) =>
+			thread(`f${i}.ts`, i, "Fix this", ![0, 150, 249].includes(i)),
+		);
+		const id = await running({
+			workflow: "iterate",
+			target: "pulls/2",
+			revision: prHead,
+			branch: "contributor/fix",
+			trusted: false,
+		});
+		const body = (await (await read(id, 1)).json()) as Record<string, any>;
+		expect(body.threads.map((t: { path: string }) => t.path)).toEqual(["f0.ts", "f150.ts", "f249.ts"]);
+		expect(github.requests.filter((r) => r.path === "/graphql").map((r) => r.body.variables.threads)).toEqual([
+			null,
+			"100",
+			"200",
+		]);
+	});
+
 	it("has nothing to read for a quest's job", async () => {
 		const id = await running();
 		const response = await read(id, 1);

quest/m0/kix/README.md

@@ -110,7 +110,6 @@ fixes what it finds:
 
 ## Required
 
-- [Runner hardening](/quest/m0/kix/runner-hardening.md) - one runner per home, and `context` reads every review thread
 - [Deploy kix.quest](/quest/m0/kix/deploy.md) - the operator registers the App and deploys the Worker with GitHub sign-in at kix.quest
 - [Operator setup](/quest/m0/kix/operator-setup.md) - fork namespace, pilot installations, and a verified test PR
 - [Per-project forks](/quest/m0/kix/forks.md) - each project publishes through its own fork, recorded by repository ID

src/runner/cli.ts

@@ -195,6 +195,7 @@ export async function runRunner(io: RunnerIO, options: { slots: number }): Promi
 		fetch: fetcher,
 		log,
 	});
+	const held = state.hold();
 	const signals = ["SIGINT", "SIGTERM", "SIGHUP"] as const;
 	const stop = () => {
 		log("stopping: publishing running work so it moves to another runner");
@@ -212,5 +213,6 @@ export async function runRunner(io: RunnerIO, options: { slots: number }): Promi
 		return 0;
 	} finally {
 		for (const signal of signals) process.off(signal, stop);
+		held.release();
 	}
 }

src/runner/home.ts

@@ -1,12 +1,13 @@
 // A runner's state on disk: its kix sign-in, its agent accounts, the
 // repositories it fetched, and its jobs' checkouts. Everything lives in one
 // user-only directory, so two runners on one machine just use two homes.
-import { randomBytes } from "node:crypto";
-import { readFileSync, rmSync } from "node:fs";
+import { randomBytes, randomUUID } from "node:crypto";
+import { readFileSync, readdirSync, rmSync } from "node:fs";
 import { homedir } from "node:os";
 import { join } from "node:path";
 
 import { privateDirectory } from "../cli/agent";
+import { type Owner, alive, currentOwner } from "../cli/supervise";
 import { save } from "../cli/worktree";
 
 /** The runner's kix sign-in, from `quest runner login`. */
@@ -91,6 +92,35 @@ export class RunnerHome {
 		return found;
 	}
 
+	/**
+	 * Take this home for one `quest runner` process until `release`, so no two
+	 * runners reuse one job checkout. The process records itself before looking
+	 * for others, so of two that start at once at least one sees the other and
+	 * refuses; neither ever works alongside the other. A record whose process
+	 * has exited, such as one that crashed, is removed instead.
+	 */
+	hold(): { release(): void } {
+		const dir = privateDirectory(join(this.dir, "running"));
+		const record = join(dir, `${randomUUID()}.json`);
+		save(record, currentOwner());
+		const release = () => rmSync(record, { force: true });
+		for (const name of readdirSync(dir)) {
+			const other = join(dir, name);
+			if (other === record || !name.endsWith(".json")) continue;
+			const owner = load<Owner>(other);
+			if (owner === undefined) continue;
+			if (!alive(owner)) {
+				rmSync(other, { force: true });
+				continue;
+			}
+			release();
+			throw new Error(
+				`another \`quest runner\` is running in ${this.dir} (process ${owner.pid} on ${owner.host}); stop it first`,
+			);
+		}
+		return { release };
+	}
+
 	/** Where fetched repositories and job checkouts live. */
 	path(...parts: string[]): string {
 		return join(this.dir, ...parts);

src/runner/runner.ts

@@ -39,7 +39,7 @@ export class Runner {
 	#closing = false;
 	#wake: ReturnType<typeof setTimeout> | undefined;
 	#stopped: () => void = () => {};
-	/** Settles once the runner shut down or kix replaced its connection. */
+	/** Settles once the runner shut down or kix replaced its connection, and its agents exited. */
 	readonly done: Promise<"stopped" | "replaced">;
 
 	constructor(options: RunnerOptions) {
@@ -56,10 +56,14 @@ export class Runner {
 				push: (push) => this.#push(push),
 				replaced: () => {
 					this.#closing = true;
-					for (const attempt of this.#live) attempt.abandon();
+					const attempts = [...this.#live];
+					for (const attempt of attempts) attempt.abandon();
 					clearTimeout(this.#wake);
 					this.connection.close();
-					resolve("replaced");
+					// The runner keeps its home until its agents exit, so none outlives its hold on their checkouts.
+					void Promise.allSettled(attempts.map((attempt) => attempt.finished)).then(() =>
+						resolve("replaced"),
+					);
 				},
 				log: options.log,
 			},