fix(github): keep project availability live on busy repositories

cloudflare/GITHUB.md

@@ -314,24 +314,32 @@ snapshot reader and validates the Quest tree with the pure core. The authenticat
 board and nested quest pages use the existing `Project` seam. Source links retain
 the exact commit; live project state never replaces GitHub as authority.
 
-Each refresh reconciles one page of native issues, **all** PRs (including human
-contributions and closed/merged PRs), and `quest/` branches. Independent opaque
-cursors persist across requests. Continue reconciliation until all connections
-finish; candidates remain empty while a scan is incomplete or a webhook has
-invalidated it. The next completed scan also detects deleted entities and removed
-claims. Refreshing a completed scan starts again from the authoritative entity
-connections, so missed webhooks and later changes to old entities are recovered.
+Each refresh advances two passes with independent opaque cursors. The history
+pass reads one page of native issues and **all** PRs (including human
+contributions and closed/merged PRs); continue reconciliation until it finishes.
+A completed history pass detects deleted entities, and refreshing after one starts
+again, so missed webhooks and later changes to old entities are recovered. The
+occupancy pass reads open PRs and `quest/` branches, the only activity that
+decides candidates, and starts again on every refresh after it completes. Its
+pages are small, so candidates are normally current after a single refresh no
+matter how long the history is. Candidates remain pending while the occupancy
+pass is incomplete or a webhook has invalidated it.
 This deliberately avoids GitHub's bounded, delayed activity feed; GitHub does
 [not automatically redeliver failed webhooks](https://docs.github.com/en/webhooks/using-webhooks/handling-failed-webhook-deliveries).
 There is no scheduled background poll in this stage.
 
-A single GraphQL activity query reads up to 50 issues, 50 PRs and 100 branch
-claims, keeping external request count independent of entity/file counts. Bulk
+A single GraphQL activity query reads up to 50 issues, 50 PRs, 100 open PRs and
+100 branch claims, keeping external request count independent of entity/file counts. Bulk
 D1 writes use `json_each`, avoiding a query per entity or SQLite's parameter
 limit. A revision comparison commits each page and cursor atomically; concurrent
 refreshes or an intervening webhook cannot publish over a newer observation.
-A webhook restarts an unfinished scan on the next refresh. Installation tokens
-remain read-only, repository-scoped, and revoked after use.
+Only a PR opening, closing, reopening or receiving new commits, a `quest/` branch
+being created or deleted, or a push to the default branch invalidates candidates
+and restarts the occupancy pass; new work waits for the next refresh, so a job
+binds the PR head and tree the maintainer saw. Checks, statuses, comments,
+reviews and other pushes wait for the history pass, so a busy repository neither
+withholds candidates nor races a refresh. Installation tokens remain read-only,
+repository-scoped, and revoked after use.
 
 Durable activity items use the GitHub repository ID, entity kind and full database
 ID. GitHub's GraphQL `fullDatabaseId` strings and webhook numeric IDs map to the

cloudflare/README.md

@@ -76,7 +76,9 @@ A job is one maintainer-authorized workflow (`plan`, `start`, `iterate`, or
 `bootstrap`) on one target, pinned to a revision (`src/jobs/`). Maintainers
 create, answer, and cancel jobs at `/projects/github/<owner>/<repo>/jobs`, and
 every request rechecks current GitHub authority. `start` takes only quests that
-observations show ready and unclaimed. `iterate` jobs are untrusted and route
+observations show ready and unclaimed. A job on an issue or pull request
+rereads it from GitHub, refusing one that closed or whose head moved since
+observations read it. `iterate` jobs are untrusted and route
 only to Linux runners that declare VM builds. Without a `flake.nix`, the route
 offers a `bootstrap` job instead. A paused project gets no new jobs and none are
 claimed; resuming it dispatches the jobs that waited. Disconnecting a project
@@ -169,8 +171,8 @@ is in progress, and their projects. Each item is one job target (`issues/1`,
 questions stay one item, and webhook updates and reconnects never add another
 prompt. Agents waiting on an answer come first, then failed jobs, new pull
 requests, new issues, and ready quests. An item offers only actions that
-would work: none on a paused project, and only a refresh while observations
-are incomplete. Every action is a plain form post to the job routes, which
+would work: none on a paused project, and only a refresh while availability
+is pending. Every action is a plain form post to the job routes, which
 recheck authority and return to the inbox; prompts render as escaped text and
 never preselect the recommendation. A `publication` decision lists the refused
 paths or why the branch is blocked, and takes the maintainer's reply as text.

cloudflare/migrations/0011_project_scan.sql

@@ -0,0 +1,3 @@
+-- Scans split into history and occupancy passes. Observations are a cache, so
+-- drop scans in the old shape; the next refresh starts both passes.
+UPDATE github_project_state SET scan=NULL;

cloudflare/src/github/service.ts

@@ -454,7 +454,7 @@ export class GitHubService {
 		const state = await this.#state(repo, installation);
 		await this.#gate(installation);
 		if (state === null) throw new GitHubError("Project enrollment changed");
-		if (result && state !== "enabled") result.available = [];
+		if (result && state !== "enabled") result.project.available = [];
 		return result;
 	}
 
@@ -480,17 +480,18 @@ export class GitHubService {
 					.bind(repo.id)
 					.first<{ revision: number; scan: string | null; invalidated: number; epoch: string }>();
 				const previous: Scan | null = row?.scan ? JSON.parse(row.scan) : null;
-				const scan: Scan =
-					previous && !previous.complete && row?.invalidated !== 1
-						? previous
-						: {
-								id: crypto.randomUUID(),
-								issues: null,
-								pulls: null,
-								refs: null,
-								claims: [],
-								complete: false,
-							};
+				// History resumes through any webhook; only occupancy, which is
+				// small, restarts to observe a possibly opened, closed, or claimed quest.
+				const scan: Scan = {
+					history:
+						previous && !previous.history.complete
+							? previous.history
+							: { id: crypto.randomUUID(), issues: null, pulls: null, complete: false },
+					occupancy:
+						previous && !previous.occupancy.complete && row!.invalidated !== 1
+							? previous.occupancy
+							: { open: null, refs: null, pulls: [], claims: [], complete: false },
+				};
 				const snapshot = await this.#snapshot(token, repo, current);
 				const response = await this.#api<{ data?: { repository: ActivityPage | null }; errors?: unknown[] }>(
 					token,
@@ -501,9 +502,10 @@ export class GitHubService {
 						variables: {
 							owner: repo.owner,
 							name: repo.name,
-							issues: scan.issues || null,
-							pulls: scan.pulls || null,
-							refs: scan.refs || null,
+							issues: scan.history.issues || null,
+							pulls: scan.history.pulls || null,
+							open: scan.occupancy.open || null,
+							refs: scan.occupancy.refs || null,
 						},
 					},
 				);
@@ -535,23 +537,38 @@ export class GitHubService {
 	}
 
 	/**
-	 * What a new job may start from: `resolve` picks it from complete
-	 * observations, then this checks whether its revision has a flake for the
+	 * What a new job may start from: `resolve` picks it from current
+	 * observations, then this checks that its issue or pull request is still
+	 * open at the observed head, and whether its revision has a flake for the
 	 * runner's devShell.
 	 */
-	jobSource<T extends { revision: string }>(
+	jobSource<T extends { target: string; revision: string }>(
 		repo: GitHubRepository,
 		userId: string,
 		resolve: (state: ProjectState) => T,
 	) {
-		return this.#authorized(repo, userId, { contents: "read" }, async (token, current, installation) => {
+		const permissions = { contents: "read", issues: "read", pull_requests: "read" } as const;
+		return this.#authorized(repo, userId, permissions, async (token, current, installation) => {
 			this.#public(current);
 			const enrolled = await this.#state(repo, installation);
 			if (enrolled === null) throw new GitHubError("Project is not enrolled for this installation");
 			const state = await this.#observations(repo, installation);
-			if (!state?.complete) throw new GitHubError("Refresh project activity before starting work");
+			// Pending availability means a webhook or an unfinished pass left open PRs and claims unread.
+			if (!state || state.project.available === null)
+				throw new GitHubError("Refresh project activity before starting work");
 			// Check the revision the job will run, such as a pull request's head.
 			const job = resolve(state);
+			// Only history rereads an issue or PR, so its activity row can still say
+			// open after it closed; read the target itself before binding a job to it.
+			const [, kind, target] = job.target.match(/^(issues|pulls)\/(\d+)$/) ?? [];
+			if (kind) {
+				const live = await this.#api<{ state: string; head?: { sha: string } }>(
+					token,
+					`${repositoryPath(repo)}/${kind}/${number(Number(target))}`,
+				);
+				if (live.state !== "open" || (kind === "pulls" && live.head?.sha !== job.revision))
+					throw new GitHubError("This changed on GitHub since it was read; refresh and review it again");
+			}
 			let flake = true;
 			try {
 				await this.#api(token, `${repositoryPath(repo)}/contents/flake.nix?ref=${sha(job.revision)}`);

cloudflare/src/github/state.ts

@@ -3,13 +3,12 @@ import type { Project, Snapshot } from "../board/project";
 import { GitHubError, type GitHubRepository } from "./service";
 import { positiveId } from "./webhook";
 
+type Cursor = string | null | false;
 export type Scan = {
-	id: string;
-	issues: string | null | false;
-	pulls: string | null | false;
-	refs: string | null | false;
-	claims: string[];
-	complete: boolean;
+	/** Pages every issue and PR, so edits to old entities and deletions are found. */
+	history: { id: string; issues: Cursor; pulls: Cursor; complete: boolean };
+	/** Pages only what availability depends on: open PRs and quest branches. */
+	occupancy: { open: Cursor; refs: Cursor; pulls: string[]; claims: string[]; complete: boolean };
 };
 type Connection<T> = { nodes: T[]; pageInfo: { hasNextPage: boolean; endCursor: string | null } };
 type Entity = {
@@ -36,6 +35,7 @@ export type Pull = Entity & {
 export type ActivityPage = {
 	issues?: Connection<Entity>;
 	pullRequests?: Connection<Pull>;
+	open?: Connection<Pull>;
 	refs?: Connection<{ name: string; prefix: string }>;
 };
 export type Activity = {
@@ -51,31 +51,37 @@ export type ProjectState = {
 	project: Project;
 	activity: Activity[];
 	claims: string[];
-	available: string[];
 	complete: boolean;
 	observedAt: number;
 	revision: number;
 };
 
 const entityFields = "databaseId:fullDatabaseId number title state updatedAt author { login } comments { totalCount }";
+const pullFields = `${entityFields} headRefName headRefOid isDraft reviewDecision reviews { totalCount } latestReviews(first:20) { nodes { state author {login} commit {oid} } pageInfo {hasNextPage} } commits(last:1) { nodes { commit { oid statusCheckRollup { state } } } }`;
 // Cursor variables are declared only for unfinished connections (GraphQL rejects
 // unused variables). Fixed page sizes bound subrequests, independent of tree size.
-export function activityQuery(scan: Scan) {
+export function activityQuery({ history, occupancy }: Scan) {
 	const fields: string[] = [];
 	const variables = ["$owner: String!", "$name: String!"];
-	if (scan.issues !== false) {
+	if (history.issues !== false) {
 		variables.push("$issues: String");
 		fields.push(
 			`issues(first:50,after:$issues,orderBy:{field:CREATED_AT,direction:ASC}) { nodes { ${entityFields} } pageInfo { hasNextPage endCursor } }`,
 		);
 	}
-	if (scan.pulls !== false) {
+	if (history.pulls !== false) {
 		variables.push("$pulls: String");
 		fields.push(
-			`pullRequests(first:50,after:$pulls,orderBy:{field:CREATED_AT,direction:ASC}) { nodes { ${entityFields} headRefName headRefOid isDraft reviewDecision reviews { totalCount } latestReviews(first:20) { nodes { state author {login} commit {oid} } pageInfo {hasNextPage} } commits(last:1) { nodes { commit { oid statusCheckRollup { state } } } } } pageInfo { hasNextPage endCursor } }`,
+			`pullRequests(first:50,after:$pulls,orderBy:{field:CREATED_AT,direction:ASC}) { nodes { ${pullFields} } pageInfo { hasNextPage endCursor } }`,
+		);
+	}
+	if (occupancy.open !== false) {
+		variables.push("$open: String");
+		fields.push(
+			`open:pullRequests(states:OPEN,first:100,after:$open,orderBy:{field:CREATED_AT,direction:ASC}) { nodes { ${pullFields} } pageInfo { hasNextPage endCursor } }`,
 		);
 	}
-	if (scan.refs !== false) {
+	if (occupancy.refs !== false) {
 		variables.push("$refs: String");
 		fields.push(
 			'refs(refPrefix:"refs/heads/quest/",first:100,after:$refs) { nodes { name prefix } pageInfo { hasNextPage endCursor } }',
@@ -84,7 +90,7 @@ export function activityQuery(scan: Scan) {
 	return `query(${variables.join(",")}) { repository(owner:$owner,name:$name) { ${fields.join("\n")} } }`;
 }
 
-function next<T>(connection: Connection<T> | undefined, cursor: Scan["issues"]) {
+function next<T>(connection: Connection<T> | undefined, cursor: Cursor) {
 	if (cursor === false) return false;
 	if (!connection || !Array.isArray(connection.nodes) || !connection.pageInfo)
 		throw new GitHubError("Incomplete project activity");
@@ -111,29 +117,39 @@ export async function reconcile(
 	epoch: string,
 ) {
 	const now = Date.now();
+	const { history, occupancy } = scan;
 	const updated: Scan = {
-		id: scan.id,
-		issues: next(page.issues, scan.issues),
-		pulls: next(page.pullRequests, scan.pulls),
-		refs: next(page.refs, scan.refs),
-		claims: [...scan.claims],
-		complete: false,
+		history: {
+			id: history.id,
+			issues: next(page.issues, history.issues),
+			pulls: next(page.pullRequests, history.pulls),
+			complete: false,
+		},
+		occupancy: {
+			open: next(page.open, occupancy.open),
+			refs: next(page.refs, occupancy.refs),
+			pulls: [...occupancy.pulls],
+			claims: [...occupancy.claims],
+			complete: false,
+		},
 	};
-	if (scan.refs !== false)
+	if (occupancy.refs !== false)
 		for (const ref of page.refs!.nodes) {
 			if (typeof ref.name !== "string" || ref.prefix !== "refs/heads/quest/")
 				throw new GitHubError("Invalid branch observation");
 			// Ref.name is relative to the requested prefix.
-			updated.claims.push(`quest/${ref.name}`);
+			updated.occupancy.claims.push(`quest/${ref.name}`);
 		}
-	updated.claims = [...new Set(updated.claims)];
-	updated.complete = updated.issues === false && updated.pulls === false && updated.refs === false;
-	const entities: { kind: string; item: Entity }[] = [];
-	for (const [kind, connection] of [
-		["issue", page.issues],
-		["pull", page.pullRequests],
+	updated.history.complete = updated.history.issues === false && updated.history.pulls === false;
+	updated.occupancy.complete = updated.occupancy.open === false && updated.occupancy.refs === false;
+	// Keyed by identity: an open PR can arrive from both PR connections at once.
+	const entities = new Map<string, { kind: string; item: Entity }>();
+	for (const [kind, connection, cursor, open] of [
+		["issue", page.issues, history.issues, false],
+		["pull", page.pullRequests, history.pulls, false],
+		["pull", page.open, occupancy.open, true],
 	] as const) {
-		if (scan[kind === "issue" ? "issues" : "pulls"] === false) continue;
+		if (cursor === false) continue;
 		for (const item of connection!.nodes) {
 			if (
 				typeof item.databaseId !== "string" ||
@@ -151,9 +167,12 @@ export async function reconcile(
 				)
 					throw new GitHubError("Pull request head changed; refresh again");
 			}
-			entities.push({ kind, item });
+			entities.set(`${kind}/${item.databaseId}`, { kind, item });
+			if (open) updated.occupancy.pulls.push(item.databaseId);
 		}
 	}
+	updated.occupancy.claims = [...new Set(updated.occupancy.claims)];
+	updated.occupancy.pulls = [...new Set(updated.occupancy.pulls)];
 	// json_each keeps the transaction to three statements and a few bindings,
 	// within D1's free query budget and SQLite's bound-parameter limit.
 	const statements = [
@@ -164,12 +183,12 @@ export async function reconcile(
  FROM json_each(?) WHERE EXISTS(SELECT 1 FROM github_project_state WHERE repository=? AND revision=? AND epoch=?)
  ON CONFLICT(repository,kind,id) DO UPDATE SET number=excluded.number,changedAt=CASE WHEN github_project_activity.data IS NOT excluded.data THEN excluded.changedAt ELSE github_project_activity.changedAt END,data=excluded.data,seenIn=excluded.seenIn,missing=0`,
 			)
-			.bind(repo.id, now, now, scan.id, JSON.stringify(entities), repo.id, revision, epoch),
+			.bind(repo.id, now, now, history.id, JSON.stringify([...entities.values()]), repo.id, revision, epoch),
 		db
 			.prepare(
 				"UPDATE github_project_activity SET missing=1 WHERE repository=? AND (seenIn IS NULL OR seenIn<>?) AND ?=1 AND EXISTS(SELECT 1 FROM github_project_state WHERE repository=? AND revision=? AND epoch=?)",
 			)
-			.bind(repo.id, scan.id, updated.complete ? 1 : 0, repo.id, revision, epoch),
+			.bind(repo.id, history.id, updated.history.complete ? 1 : 0, repo.id, revision, epoch),
 		db
 			.prepare(
 				"UPDATE github_project_state SET snapshot=?,scan=?,observedAt=?,invalidated=?,revision=revision+1 WHERE repository=? AND revision=? AND epoch=?",
@@ -202,8 +221,11 @@ export async function readState(db: D1Database, repo: GitHubRepository): Promise
 	const rows = records[1]!.results as (Omit<Activity, "data"> & { data: string | null })[];
 	const activity: Activity[] = rows.map((item) => ({ ...item, data: item.data ? JSON.parse(item.data) : null }));
 	const paths = new Set(snapshot.documents.map((d) => d.path));
+	// A PR's activity row can still say OPEN until history rereads it, so only
+	// PRs in the latest occupancy pass count as open.
+	const open = new Set(scan.occupancy.pulls);
 	const changes = activity
-		.filter((item) => item.kind === "pull" && !item.missing && item.data?.state === "OPEN")
+		.filter((item) => item.kind === "pull" && open.has(item.id) && !item.missing && item.data?.state === "OPEN")
 		.flatMap((item) => {
 			const pull = item.data as Pull;
 			const quest = associatedQuest(pull.headRefName, paths);
@@ -217,12 +239,16 @@ export async function readState(db: D1Database, repo: GitHubRepository): Promise
 					]
 				: [];
 		});
-	const claims = scan.claims.map((ref) => associatedQuest(ref, paths)).filter((q): q is string => q !== null);
-	const complete = scan.complete && row.invalidated === 0;
+	const claims = scan.occupancy.claims
+		.map((ref) => associatedQuest(ref, paths))
+		.filter((q): q is string => q !== null);
 	const occupied = new Set([...claims, ...changes.map((c) => c.quest)]);
-	const available = complete
-		? ready(snapshot.documents.map((d) => parse(d.path, d.content))).filter((path) => !occupied.has(path))
-		: [];
+	// Unknown until every open PR and quest branch is read since the last
+	// webhook that could have opened, closed, or claimed a quest.
+	const available =
+		scan.occupancy.complete && row.invalidated === 0
+			? ready(snapshot.documents.map((d) => parse(d.path, d.content))).filter((path) => !occupied.has(path))
+			: null;
 	const project: Project = {
 		...snapshot,
 		name: `github/${repo.owner}/${repo.name}`,
@@ -234,14 +260,13 @@ export async function readState(db: D1Database, repo: GitHubRepository): Promise
 		issues: activity
 			.filter((item) => item.kind === "issue" && !item.missing && item.data?.state === "OPEN")
 			.map((item) => ({ path: `issues/${item.number}`, title: item.data!.title })),
-		available: complete ? available : null,
+		available,
 	};
 	return {
 		project,
 		activity,
 		claims,
-		available,
-		complete,
+		complete: scan.history.complete,
 		observedAt: row.observedAt,
 		revision: row.revision,
 	};