feat(kix): reconcile projects after each webhook

cloudflare/GITHUB.md

@@ -313,8 +313,8 @@ The jobs/funding quests own live eligibility checks and contribution execution.
 ## Project state and activity
 
 An enrolled project's setup page links to `/projects/github/<owner>/<repo>/state`.
-A maintainer explicitly refreshes observations with its same-origin form. The
-service rereads the default branch at an exact commit using the preview's shared
+Webhooks reread observations in the background (see below), and a maintainer
+can refresh them explicitly with its same-origin form. Each read rereads the default branch at an exact commit using the preview's shared
 snapshot reader and validates the Quest tree with the pure core. The authenticated
 board and nested quest pages use the existing `Project` seam. Source links retain
 the exact commit; live project state never replaces GitHub as authority.
@@ -333,17 +333,42 @@ 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.
 
+Each webhook that names an issue or PR (a comment on a PR names the PR by its
+number), or that invalidates candidates (below), asks for a background read of its enabled project once its delivery commits.
+The read runs after the response (`waitUntil`) with the App's read-only,
+repository-scoped installation token, and is the same read as a refresh: one
+page of each pass, plus every entity a webhook named since it was last read,
+fetched by number in the same GraphQL query. A new issue or PR therefore shows
+its title and state after one read however far the history pass is from it,
+and a closed one stops showing as open. A lease in `github_project_state` lets
+one read run at a time per project: it answers every request made before it
+started, requests made during it get one more read, and duplicate deliveries
+never reach it, so a burst costs one or two reads rather than one each. A
+per-entity webhook count lets a read clear only the updates it saw. One
+invocation reads at most three rounds, to stay within a Worker's subrequest
+budget; a request left after them waits for the next webhook or a refresh.
+A read overtaken by another read or by a webhook's revision bump reads again.
+Paused and disconnected projects are not read. A failed or rate-limited read is
+recorded and shown next to the refresh fallback, and is not retried: only a
+request made during it gets another round, and otherwise the next webhook or a
+refresh reads again. Checks, statuses and other pushes name no entity and
+change no candidate, so they cost no read.
+
+The inbox also treats a PR whose row still says open as closed once a
+complete, current occupancy pass no longer lists it, so a missed close webhook
+corrects itself on the next read.
+
 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.
 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
+and restarts the occupancy pass; new work waits for the next read, 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,
+reviews and other pushes do not invalidate candidates, 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
@@ -365,7 +390,10 @@ Claimed on the board. Runner/run associations belong to job coordination.
 
 Every cached read and refresh proves current maintainer authority, live repository
 identity/public visibility, installation access, and enrollment for that
-installation. Refresh rechecks those boundaries after provider reads. Paused
+installation. Refresh rechecks those boundaries after provider reads. A
+background read has no user, so it checks the same boundaries except maintainer
+authority; it publishes observations only, and grants or caches no authority.
+Every page and action still rechecks the maintainer acting on it. Paused
 projects remain readable but have no candidates. Disconnect cascades cached
 snapshots and activity; explicitly enrolling a replacement installation clears
 prior observations. No positive authority cache is created. Candidate data is an

cloudflare/README.md

@@ -172,9 +172,11 @@ progress, and their projects. Each item is one job target (`issues/1`,
 `pulls/2`, a quest path), so an issue, its planning job, and that job's
 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 availability
-is pending. Every action is a plain form post to the job routes, which
+requests, new issues, and ready quests. Each webhook rereads its project in the
+background, so a new issue or PR arrives with its title, ready to act on. An
+item offers only actions that would work: none on a paused project, and only a
+refresh while availability is pending or a background read failed, with the
+failure's reason. Every action is a plain form post to the job routes, which
 recheck authority and move on to the next active decision (or the empty
 state) with a short notice such as "Answer recorded"; the item acted on stays
 reachable from its list with its earlier answers. A refused action says why

cloudflare/dev/github.ts

@@ -101,7 +101,17 @@ function graphql(repo: Repo, query: string) {
 			})),
 		);
 	}
-	return { data: { repository: result } };
+	// Entities a webhook named, read by number as GitHub does: an issue number
+	// that is a pull request, or a missing one, is NOT_FOUND at its alias.
+	const errors: unknown[] = [];
+	for (const [, alias, field, number] of query.matchAll(/(s\d+):(issue|pullRequest)\(number:(\d+)\)/g)) {
+		const pull = repo.pulls.find((p) => p.number === Number(number));
+		const issue = repo.issues.find((i) => i.number === Number(number));
+		const found = field === "pullRequest" ? pull && pullNode(repo, pull) : issue && entity(repo, issue);
+		result[alias!] = found ?? null;
+		if (!found) errors.push({ type: "NOT_FOUND", path: ["repository", alias] });
+	}
+	return { data: { repository: result }, ...(errors.length ? { errors } : {}) };
 }
 
 function tree(repo: Repo) {

cloudflare/dev/seed.ts

@@ -213,7 +213,7 @@ export async function seed(options: { fetch: Fetch; db: D1Database; world: World
 		"pause archive",
 	);
 
-	// An issue opened since the refresh: its webhook leaves a pointer with no data yet.
+	// An issue opened since the refresh: its webhook rereads the project in the background.
 	const pointer = { number: 9, title: "Sync stops after the laptop sleeps", author: "reporter" };
 	rocket.issues.push(pointer);
 	const body = JSON.stringify({

cloudflare/migrations/0013_reconcile.sql

@@ -0,0 +1,11 @@
+-- A webhook asks for a background read of its project. `requested` stays set
+-- until a read starts, and `lease` (an expiry in ms) lets only one read run at
+-- a time, so a burst of deliveries coalesces into one read. `failure` keeps the
+-- last background read's error until a read or refresh succeeds.
+ALTER TABLE github_project_state ADD COLUMN requested INTEGER NOT NULL DEFAULT 0 CHECK(requested IN (0,1));
+ALTER TABLE github_project_state ADD COLUMN lease INTEGER;
+ALTER TABLE github_project_state ADD COLUMN failure TEXT;
+-- Counts the webhooks that named an entity since it was last read by number, so
+-- a read clears only what it saw.
+ALTER TABLE github_project_activity ADD COLUMN stale INTEGER NOT NULL DEFAULT 0;
+UPDATE github_project_activity SET stale=1 WHERE data IS NULL AND missing=0;

cloudflare/src/github/service.ts

@@ -2,7 +2,15 @@ import { createHash, createPrivateKey, sign } from "node:crypto";
 import { check, isQuest, parse, ready } from "quest/core";
 import { osiApproved } from "../licenses";
 import { positiveId } from "./webhook";
-import { activityQuery, type ActivityPage, type ProjectState, type Scan, reconcile, readState } from "./state";
+import {
+	activityQuery,
+	type ActivityPage,
+	type ProjectState,
+	type Scan,
+	reconcile,
+	readState,
+	staleEntities,
+} from "./state";
 
 export interface GitHubRepository {
 	owner: string;
@@ -11,6 +19,17 @@ export interface GitHubRepository {
 }
 type Permission = "metadata" | "contents" | "issues" | "pull_requests" | "checks" | "statuses";
 type Permissions = Partial<Record<Permission, "read" | "write">>;
+/** What reading a project's observations needs. */
+const observing: Permissions = {
+	contents: "read",
+	issues: "read",
+	pull_requests: "read",
+	checks: "read",
+	statuses: "read",
+};
+// Longer than a Worker keeps background work after its response, so a read
+// that died with its Worker frees the project for the next webhook.
+const observeLease = 60_000;
 interface Installation {
 	id: number;
 	app_id: number;
@@ -54,6 +73,13 @@ export class GitHubError extends Error {
 		super(message);
 	}
 }
+
+/** Another read published, or a webhook invalidated, while this one read, so it must read again. */
+export class ProjectChanged extends GitHubError {
+	constructor() {
+		super("Project changed during refresh; refresh again");
+	}
+}
 /** A publication the bot cannot make, which needs a maintainer's decision rather than a retry. */
 export class PublicationBlocked extends GitHubError {}
 /** A maintainer check that was answered with a denial, rather than one that could not complete. */
@@ -521,80 +547,150 @@ export class GitHubService {
 
 	/** Observations only. Neither a refresh nor an arriving entity launches work. */
 	projectState(repo: GitHubRepository, userId: string, refresh = false): Promise<ProjectState | null> {
-		return this.#authorized(
-			repo,
-			userId,
-			refresh
-				? { contents: "read", issues: "read", pull_requests: "read", checks: "read", statuses: "read" }
-				: {},
-			async (token, current, installation) => {
-				this.#public(current);
-				if ((await this.#state(repo, installation)) === null)
-					throw new GitHubError("Project is not enrolled for this installation");
-				if (!refresh) return this.#observations(repo, installation);
-				await this.#env.DB.prepare("INSERT OR IGNORE INTO github_project_state(repository,epoch) VALUES(?,?)")
-					.bind(repo.id, crypto.randomUUID())
-					.run();
-				const row = await this.#env.DB.prepare(
-					"SELECT revision,scan,invalidated,epoch FROM github_project_state WHERE repository=?",
-				)
-					.bind(repo.id)
-					.first<{ revision: number; scan: string | null; invalidated: number; epoch: string }>();
-				const previous: Scan | null = row?.scan ? JSON.parse(row.scan) : null;
-				// 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,
-					"/graphql",
-					"POST",
-					{
-						query: activityQuery(scan),
-						variables: {
-							owner: repo.owner,
-							name: repo.name,
-							issues: scan.history.issues || null,
-							pulls: scan.history.pulls || null,
-							open: scan.occupancy.open || null,
-							refs: scan.occupancy.refs || null,
-						},
-					},
-				);
-				if (response.errors?.length || !response.data?.repository)
-					throw new GitHubError("Project activity unavailable");
-				// A rename, transfer, visibility change, revocation, or lost
-				// maintainer role during the scan must not publish fresh observations.
-				const latest = await this.#api<Repository>(token, repositoryPath(repo));
-				this.#repository(repo, latest);
-				this.#public(latest);
-				if (latest.default_branch !== current.default_branch)
-					throw new GitHubError("Default branch changed; refresh again");
-				await this.#maintainer(token, repo, userId);
-				await this.#gate(installation);
-				if ((await this.#state(repo, installation)) === null)
-					throw new GitHubError("Project enrollment changed");
-				await reconcile(
-					this.#env.DB,
-					repo,
-					snapshot,
-					scan,
-					response.data.repository,
-					row!.revision,
-					row!.epoch,
-				);
-				return this.#observations(repo, installation);
+		return this.#authorized(repo, userId, refresh ? observing : {}, async (token, current, installation) => {
+			this.#public(current);
+			if ((await this.#state(repo, installation)) === null)
+				throw new GitHubError("Project is not enrolled for this installation");
+			// A lost maintainer role during the read must not publish fresh observations.
+			if (refresh)
+				await this.#read(token, repo, current, installation, () => this.#maintainer(token, repo, userId));
+			return this.#observations(repo, installation);
+		});
+	}
+
+	/**
+	 * Read one page of a project's observations from GitHub and publish them,
+	 * unless `recheck` fails or the repository, installation, or enrollment
+	 * changed during the read.
+	 */
+	async #read(
+		token: string,
+		repo: GitHubRepository,
+		current: Repository,
+		installation: number,
+		recheck: () => Promise<void>,
+	) {
+		await this.#env.DB.prepare("INSERT OR IGNORE INTO github_project_state(repository,epoch) VALUES(?,?)")
+			.bind(repo.id, crypto.randomUUID())
+			.run();
+		const row = await this.#env.DB.prepare(
+			"SELECT revision,scan,invalidated,epoch FROM github_project_state WHERE repository=?",
+		)
+			.bind(repo.id)
+			.first<{ revision: number; scan: string | null; invalidated: number; epoch: string }>();
+		const previous: Scan | null = row?.scan ? JSON.parse(row.scan) : null;
+		// 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 },
+		};
+		// What webhooks named is read by number, so it is current after one read
+		// however far the history pass is from it.
+		const stale = await staleEntities(this.#env.DB, repo.id);
+		const snapshot = await this.#snapshot(token, repo, current);
+		const response = await this.#api<{
+			data?: { repository: ActivityPage | null };
+			errors?: { path?: unknown[] }[];
+		}>(token, "/graphql", "POST", {
+			query: activityQuery(scan, stale),
+			variables: {
+				owner: repo.owner,
+				name: repo.name,
+				issues: scan.history.issues || null,
+				pulls: scan.history.pulls || null,
+				open: scan.occupancy.open || null,
+				refs: scan.occupancy.refs || null,
 			},
+		});
+		// An entity read by number may be gone, such as a deleted issue (GitHub
+		// reports NOT_FOUND at its alias). Any other error is a partial result.
+		// Accepting any error at the alias keeps one unreadable entity from
+		// blocking every later read.
+		const partial = response.errors?.some(
+			(error) =>
+				error.path?.length !== 2 || error.path[0] !== "repository" || !/^s\d+$/.test(String(error.path[1])),
 		);
+		if (partial || !response.data?.repository) throw new GitHubError("Project activity unavailable");
+		// A rename, transfer, visibility change, or revocation during the read
+		// must not publish fresh observations.
+		const latest = await this.#api<Repository>(token, repositoryPath(repo));
+		this.#repository(repo, latest);
+		this.#public(latest);
+		if (latest.default_branch !== current.default_branch)
+			throw new GitHubError("Default branch changed; refresh again");
+		await recheck();
+		await this.#gate(installation);
+		if ((await this.#state(repo, installation)) === null) throw new GitHubError("Project enrollment changed");
+		await reconcile(this.#env.DB, repo, snapshot, scan, response.data.repository, stale, row!.revision, row!.epoch);
+	}
+
+	/**
+	 * Reread an enabled project after a webhook asked for it, with the App's
+	 * own token. One read runs at a time: it answers every request made before
+	 * it started, and requests made during it get one more read, so a burst of
+	 * deliveries costs one or two reads rather than one each. This publishes
+	 * observations only. It checks no maintainer, so it neither grants nor
+	 * caches authority; every page and action still rechecks whoever acts. A
+	 * read that another read or a webhook overtook reads again. A failure is
+	 * recorded for the inbox and not retried: only a request made during the
+	 * failed read gets another round. Each invocation reads at most three
+	 * rounds to stay within one Worker invocation's subrequest budget; a
+	 * request left after them waits for the next webhook or a refresh.
+	 */
+	async observe(id: number) {
+		const db = this.#env.DB;
+		for (let round = 0; round < 3; round++) {
+			const now = Date.now();
+			const lease = now + observeLease;
+			const claimed = await db
+				.prepare(
+					`UPDATE github_project_state SET requested=0,lease=? WHERE repository=? AND requested=1 AND (lease IS NULL OR lease<?)
+ AND EXISTS(SELECT 1 FROM github_projects WHERE id=? AND state='enabled')`,
+				)
+				.bind(lease, id, now, id)
+				.run();
+			if (claimed.meta.changes !== 1) return;
+			let failure: string | null = null;
+			let overtaken = false;
+			try {
+				const repo = await db
+					.prepare("SELECT id,owner,name FROM github_projects WHERE id=?")
+					.bind(id)
+					.first<GitHubRepository>();
+				if (repo)
+					await this.#withRepository(repo, observing, async (token, current, installation) => {
+						this.#public(current);
+						if ((await this.#state(repo, installation)) !== "enabled")
+							throw new GitHubError("Project enrollment changed");
+						await this.#read(token, repo, current, installation, async () => {});
+					});
+			} catch (error) {
+				// A manual refresh that started before the webhook, or a webhook's
+				// revision bump, can publish first and leave this read's entities stale.
+				if (error instanceof ProjectChanged) overtaken = true;
+				else {
+					// Provider errors carry no response bodies or secrets; anything else is a bug.
+					failure = error instanceof GitHubError ? error.message : "Unexpected error";
+					if (!(error instanceof GitHubError))
+						console.error({ event: "github_observe_failed", name: (error as Error)?.name });
+				}
+			}
+			// An overtaken read, or more named entities than one read covers, asks for another round.
+			const more = overtaken || (!failure && (await staleEntities(db, id)).length > 0);
+			await db
+				.prepare(
+					"UPDATE github_project_state SET lease=NULL,failure=COALESCE(?,failure),requested=MAX(requested,?) WHERE repository=? AND lease=?",
+				)
+				.bind(failure, more ? 1 : 0, id, lease)
+				.run();
+		}
 	}
 
 	/**