From 80e917b596b59b9309d0aeaee3d20691d10eb3a8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 23:03:40 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m0/kix/runner-hardening.md Quest-Claim: aefbabad75ebea92 From c027ac0cb00d4312537e2417fa5fe29f4a216921 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 23:17:33 -0700 Subject: [PATCH 2/2] fix(runner): hold one runner per home, and page review threads in context A runner records itself in /running before looking for others, so a second runner on the same home refuses; a crashed holder's record is ignored and removed. A replaced runner now settles only once its agents exit, so it never lets go of the home while they still stop. The context read follows reviewThreads cursors until every thread is examined, so a busy pull request's newer unresolved threads are kept. Finishes quest/m0/kix/runner-hardening.md. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- cloudflare/src/github/service.ts | 49 ++++++++++++++++++------ cloudflare/test/publish.test.ts | 61 ++++++++++++++++++----------- quest/m0/kix/README.md | 1 - quest/m0/kix/runner-hardening.md | 34 ---------------- src/runner/cli.ts | 2 + src/runner/home.ts | 34 +++++++++++++++- src/runner/runner.ts | 10 +++-- test/runner.test.ts | 66 ++++++++++++++++++++++++++++++-- 9 files changed, 181 insertions(+), 78 deletions(-) delete mode 100644 quest/m0/kix/runner-hardening.md diff --git a/README.md b/README.md index 65f9531c..1eb4c1e8 100644 --- a/README.md +++ b/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 diff --git a/cloudflare/src/github/service.ts b/cloudflare/src/github/service.ts index dcb5d42c..0e71dd2b 100644 --- a/cloudflare/src/github/service.ts +++ b/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,7 +913,7 @@ 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 }; @@ -918,15 +921,37 @@ export class GitHubService { }; }; 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, diff --git a/cloudflare/test/publish.test.ts b/cloudflare/test/publish.test.ts index 4b73a52d..1ad12e4e 100644 --- a/cloudflare/test/publish.test.ts +++ b/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), + /** 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 = { "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; + 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); diff --git a/quest/m0/kix/README.md b/quest/m0/kix/README.md index 8da0e5e8..fb7e2983 100644 --- a/quest/m0/kix/README.md +++ b/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 diff --git a/quest/m0/kix/runner-hardening.md b/quest/m0/kix/runner-hardening.md deleted file mode 100644 index 322dd0be..00000000 --- a/quest/m0/kix/runner-hardening.md +++ /dev/null @@ -1,34 +0,0 @@ -# [S] Fix the runner's post-merge review findings - -## Goal - -Two `quest runner` processes never work in the same runner home at once, and -the `context` tool returns every unresolved review thread of a pull request, -however many threads it has. - -## Plan - -Decided 2026-10-22 (user) from Codex's post-merge review of -[kixelated/quest#188](https://github.com/kixelated/quest/pull/188) -([review](https://github.com/kixelated/quest/pull/188#pullrequestreview-5618747276)), -which covered rework that merged without a review: - -- **One runner per home (user).** `quest runner run` takes an exclusive lock - on its runner home and refuses to start while another process holds it. - The per-repository lock in `src/runner/workspace.ts` serializes only within - one process, so a second runner on the same home could reuse and modify a - job worktree while the first one's agents are still stopping after the hub - reassigns their jobs - ([P1](https://github.com/kixelated/quest/pull/188#discussion_r4351848499)). - Rejected: inter-process locks per repository and worktree, which allow - several runners per home at more complexity. The lock must release when its - holder dies, on Linux and macOS. -- **Page review threads.** The context query in - `cloudflare/src/github/service.ts` reads only the first 100 review threads - before filtering unresolved ones, so a newer unresolved thread on a busy PR - is silently omitted - ([P2](https://github.com/kixelated/quest/pull/188#discussion_r4351848505)). - Follow `pageInfo` cursors until every thread is examined. -- Ranked first in the epic, since both fix merged code. Test a second runner - refused on a held home, a lock released after a crash, and a pull request - with more than 100 review threads. diff --git a/src/runner/cli.ts b/src/runner/cli.ts index f983245d..ceae1b5c 100644 --- a/src/runner/cli.ts +++ b/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(); } } diff --git a/src/runner/home.ts b/src/runner/home.ts index 08eb39b8..704d712f 100644 --- a/src/runner/home.ts +++ b/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(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); diff --git a/src/runner/runner.ts b/src/runner/runner.ts index eba10e71..a2cdcb0f 100644 --- a/src/runner/runner.ts +++ b/src/runner/runner.ts @@ -39,7 +39,7 @@ export class Runner { #closing = false; #wake: ReturnType | 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, }, diff --git a/test/runner.test.ts b/test/runner.test.ts index 7f7f544f..e1de0517 100644 --- a/test/runner.test.ts +++ b/test/runner.test.ts @@ -1,6 +1,7 @@ -import { execFileSync } from "node:child_process"; -import { chmodSync, existsSync, mkdirSync, readFileSync, symlinkSync, writeFileSync } from "node:fs"; -import { join } from "node:path"; +import { execFileSync, spawn } from "node:child_process"; +import { once } from "node:events"; +import { chmodSync, existsSync, mkdirSync, readFileSync, readdirSync, symlinkSync, writeFileSync } from "node:fs"; +import { join, resolve } from "node:path"; import { describe, expect, test, vi } from "vitest"; import { Claude, streamReader } from "../src/runner/claude"; @@ -476,6 +477,52 @@ describe("login", () => { }); }); +describe("runner home", () => { + /** Another process holding `dir` until it is killed. */ + async function holder(dir: string) { + const root = resolve(import.meta.dirname, ".."); + const script = join(tempDir(), "hold.ts"); + writeFileSync( + script, + `import { RunnerHome } from ${JSON.stringify(join(root, "src/runner/home.ts"))};\n` + + `new RunnerHome(${JSON.stringify(dir)}).hold();\nconsole.log("held");\nsetInterval(() => {}, 1000);\n`, + ); + const child = spawn("bun", [`--config=${join(root, "bunfig.toml")}`, script], { + stdio: ["ignore", "pipe", "inherit"], + }); + const [line] = await once(child.stdout, "data"); + expect(String(line)).toBe("held\n"); + return child; + } + + posix("refuses a second runner while one holds the home", async () => { + const dir = tempDir(); + const first = new RunnerHome(dir).hold(); + expect(() => new RunnerHome(dir).hold()).toThrow("another `quest runner` is running in"); + first.release(); + new RunnerHome(dir).hold().release(); + + const child = await holder(dir); + try { + expect(() => new RunnerHome(dir).hold()).toThrow(`(process ${child.pid} `); + } finally { + child.kill("SIGKILL"); + } + }); + + posix("takes the home once its holder crashed", async () => { + const dir = tempDir(); + const child = await holder(dir); + child.kill("SIGKILL"); + await once(child, "exit"); + const held = new RunnerHome(dir).hold(); + // The crashed holder's record is gone, so only this one remains. + expect(readdirSync(join(dir, "running"))).toHaveLength(1); + held.release(); + expect(readdirSync(join(dir, "running"))).toEqual([]); + }); +}); + /** A fake `claude` that runs the steps the test wrote for a new or resumed session. */ const FAKE_CLAUDE = ` const fs = require("node:fs"); @@ -878,6 +925,19 @@ describe("quest runner", () => { expect(r.hub.of("status").at(-1)).toMatchObject({ attempts: [], slots: 0 }); }); + posix("a replaced runner ends only once its agents exit", async () => { + const r = rig(); + r.steps({ start: [ready, { wait: true }] }); + r.assign(job(r.upstream.commit)); + await waiting(r); + r.hub.socket.close(4000, "Replaced by a new connection"); + expect(await r.runner.done).toBe("replaced"); + // Its agent stopped before the runner let go of the home holding its checkout. + expect(r.runner.running).toEqual([]); + expect(r.calls().some((c) => c.terminated)).toBe(true); + expect(r.hub.of("result")).toEqual([]); + }); + posix("continues from the head kix reports after a publication's reply was lost", async () => { let landed = ""; const r = rig((body, source) => {