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) => {