Batch inbox job hydration per project

cloudflare/src/jobs/store.ts

@@ -727,25 +727,57 @@ export async function readJob(db: D1Database, id: string) {
 	]);
 	const row = job!.results[0] as Job | undefined;
 	if (!row) return null;
+	return jobView(row, events!.results as Event[], questions!.results as Question[]);
+}
+
+function jobView(row: Job, events: Event[], questions: Question[]) {
 	const json = (value: string | null): unknown => (value === null ? null : JSON.parse(value));
 	return {
 		...row,
 		result: json(row.result),
-		events: (events!.results as Event[]).map((e) => ({ ...e, body: json(e.body as string) })),
-		questions: (questions!.results as Question[]).map((q) => ({
+		events: events.map((e) => ({ ...e, body: json(e.body as string) })),
+		questions: questions.map((q) => ({
 			...q,
 			prompt: json(q.prompt as string),
 			answer: json(q.answer as string | null),
 		})),
 	};
 }
 
+/** Hydrate a selection in three snapshot queries, without multiplying events by questions. */
+async function readJobs(db: D1Database, select: string, ...parameters: (string | number)[]) {
+	const [jobs, events, questions] = await db.batch([
+		db.prepare(select).bind(...parameters),
+		db
+			.prepare(
+				`WITH selected AS (${select}) SELECT e.job,e.attempt,e.kind,e.body,e.at
+				FROM job_events e JOIN selected j ON j.id=e.job ORDER BY e.rowid`,
+			)
+			.bind(...parameters),
+		db
+			.prepare(
+				`WITH selected AS (${select}) SELECT q.job,q.id,q.attempt,q.kind,q.prompt,q.answer,q.answeredBy,q.askedAt,q.answeredAt,q.approvedAt
+				FROM job_questions q JOIN selected j ON j.id=q.job ORDER BY q.askedAt,q.rowid`,
+			)
+			.bind(...parameters),
+	]);
+	const byJob = <T extends { job: string }>(rows: T[]) => {
+		const grouped = new Map<string, Omit<T, "job">[]>();
+		for (const { job, ...row } of rows) {
+			if (!grouped.has(job)) grouped.set(job, []);
+			grouped.get(job)!.push(row);
+		}
+		return grouped;
+	};
+	const eventRows = byJob(events!.results as (Event & { job: string })[]);
+	const questionRows = byJob(questions!.results as (Question & { job: string })[]);
+	return (jobs!.results as Job[]).map((row) =>
+		jobView(row, eventRows.get(row.id) ?? [], questionRows.get(row.id) ?? []),
+	);
+}
+
 export async function listJobs(db: D1Database, repository: number) {
-	const rows = await db
-		.prepare("SELECT id FROM jobs WHERE repository=? ORDER BY createdAt DESC,id LIMIT 50")
-		.bind(repository)
-		.all<{ id: string }>();
-	return (await Promise.all(rows.results.map((row) => readJob(db, row.id)))).filter((job) => job !== null);
+	return readJobs(db, "SELECT * FROM jobs WHERE repository=? ORDER BY createdAt DESC,id LIMIT 50", repository);
 }
 
 export type Published = Pick<Job, "branch" | "head" | "pending" | "pull">;
@@ -777,16 +809,15 @@ export async function active(db: D1Database, repository: number, target: string)
  * one of the `observed` issues, pull requests, or ready quests.
  */
 export async function listInboxJobs(db: D1Database, repository: number, observed: string[]) {
-	const rows = await db
-		.prepare(
-			`SELECT id FROM jobs j WHERE repository=?1
+	return readJobs(
+		db,
+		`SELECT * FROM jobs j WHERE repository=?1
 				AND NOT EXISTS(SELECT 1 FROM jobs n WHERE n.repository=?1 AND n.target=j.target AND (n.createdAt>j.createdAt OR (n.createdAt=j.createdAt AND n.id<j.id)))
 				AND (state IN ('queued','running','waiting','failed') OR target IN (SELECT value FROM json_each(?2)))
 			ORDER BY createdAt DESC,id`,
-		)
-		.bind(repository, JSON.stringify(observed))
-		.all<{ id: string }>();
-	return (await Promise.all(rows.results.map((row) => readJob(db, row.id)))).filter((job) => job !== null);
+		repository,
+		JSON.stringify(observed),
+	);
 }
 
 /** The most follow-ups a delivery suggests, and the longest one, in characters. */
@@ -848,17 +879,16 @@ export async function followupRequest(db: D1Database, repository: number, job: s
  * delivered is one of the `open` ones and no later job works on that pull request.
  */
 export async function listActivityJobs(db: D1Database, repository: number, open: number[]) {
-	const rows = await db
-		.prepare(
-			`SELECT id FROM jobs j WHERE repository=?1
+	return readJobs(
+		db,
+		`SELECT * FROM jobs j WHERE repository=?1
 				AND NOT EXISTS(SELECT 1 FROM jobs n WHERE n.repository=?1 AND n.target=j.target AND (n.createdAt>j.createdAt OR (n.createdAt=j.createdAt AND n.id<j.id)))
 				AND (state IN ('queued','running','waiting') OR (pull IN (SELECT value FROM json_each(?2))
 					AND NOT EXISTS(SELECT 1 FROM jobs n WHERE n.repository=?1 AND n.target='pulls/'||j.pull AND n.id<>j.id AND n.createdAt>=j.createdAt)))
 			ORDER BY createdAt DESC,id`,
-		)
-		.bind(repository, JSON.stringify(open))
-		.all<{ id: string }>();
-	return (await Promise.all(rows.results.map((row) => readJob(db, row.id)))).filter((job) => job !== null);
+		repository,
+		JSON.stringify(open),
+	);
 }
 
 type Pin = { runner: string; account: string };

cloudflare/test/inbox-reads.test.ts

@@ -0,0 +1,155 @@
+import { env } from "cloudflare:workers";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { load } from "../src/inbox/model";
+import * as store from "../src/jobs/store";
+import { commit, maintainerOf, project } from "./project-fixtures";
+
+const db = env.DB;
+let user = "";
+beforeEach(async () => {
+	({ user } = await maintainerOf());
+});
+afterEach(() => vi.restoreAllMocks());
+
+/** Count executed statements, including each statement inside a D1 batch. */
+function counted() {
+	let queries = 0;
+	const originals = new WeakMap<D1PreparedStatement, D1PreparedStatement>();
+	const statement = (inner: D1PreparedStatement): D1PreparedStatement => {
+		const proxy = new Proxy(inner, {
+			get(target, key) {
+				if (key === "bind") return (...args: unknown[]) => statement(target.bind(...args));
+				const value = Reflect.get(target, key, target);
+				if (typeof value !== "function") return value;
+				return (...args: unknown[]) => {
+					if (["all", "first", "run", "raw"].includes(String(key))) queries++;
+					return value.apply(target, args);
+				};
+			},
+		});
+		originals.set(proxy, inner);
+		return proxy;
+	};
+	return {
+		db: new Proxy(db, {
+			get(target, key) {
+				if (key === "prepare") return (sql: string) => statement(target.prepare(sql));
+				if (key === "batch")
+					return (statements: D1PreparedStatement[]) => {
+						queries += statements.length;
+						return target.batch(statements.map((s) => originals.get(s) ?? s));
+					};
+				const value = Reflect.get(target, key, target);
+				return typeof value === "function" ? value.bind(target) : value;
+			},
+		}),
+		queries: () => queries,
+	};
+}
+
+/** Many targets, with interleaved attempts, events, answered and unanswered questions. */
+async function seed(count: number, repository = 10) {
+	const ids: string[] = [];
+	const statements: D1PreparedStatement[] = [];
+	for (let i = 0; i < count; i++) {
+		const id = `job-${repository}-${String(i).padStart(3, "0")}`;
+		ids.push(id);
+		statements.push(
+			db
+				.prepare(
+					`INSERT INTO jobs(id,repository,workflow,target,revision,branch,actor,trusted,state,attempt,result,createdAt,updatedAt)
+					VALUES(?,?,'start',?,?,'quest/work',?,1,?,2,?,?,?)`,
+				)
+				.bind(
+					id,
+					repository,
+					`quest/job-${i}.md`,
+					commit,
+					user,
+					i % 2 ? "completed" : "waiting",
+					JSON.stringify({ followups: [`Follow-up ${i}`] }),
+					i,
+					100 + i,
+				),
+		);
+		if (i % 3 === 0) continue;
+		// Event order follows insertion, not timestamp or attempt.
+		for (const attempt of [2, 1])
+			statements.push(
+				db
+					.prepare("INSERT INTO job_events(job,attempt,kind,body,at) VALUES(?,?,'summary',?,?)")
+					.bind(id, attempt, JSON.stringify({ summary: `${id}-${attempt}` }), attempt),
+			);
+		// Questions order by askedAt, using rowid to break ties, not by attempt.
+		for (const attempt of [2, 1])
+			statements.push(
+				db
+					.prepare(
+						`INSERT INTO job_questions(id,job,attempt,kind,prompt,answer,answeredBy,askedAt,answeredAt,approvedAt)
+						VALUES(?,?,?,'agent',?,?,?,?,?,?)`,
+					)
+					.bind(
+						`${id}-${attempt}`,
+						id,
+						attempt,
+						JSON.stringify({ questions: [{ question: `Choice ${id}-${attempt}?`, options: [] }] }),
+						attempt === 1 ? JSON.stringify({ text: `Answer ${id}` }) : null,
+						attempt === 1 ? user : null,
+						attempt,
+						attempt === 1 ? 10 : null,
+						attempt === 1 ? 11 : null,
+					),
+			);
+	}
+	await db.batch(statements);
+	return ids;
+}
+
+describe("batched inbox reads", () => {
+	it("hydrates many jobs identically with a constant number of queries", async () => {
+		const observed = Array.from({ length: 120 }, (_, i) => `quest/job-${i}.md`);
+		const empty = counted();
+		expect(await store.listInboxJobs(empty.db, 10, observed)).toEqual([]);
+		expect(empty.queries()).toBe(3);
+		const ids = await seed(120);
+		await project(11, user);
+		await seed(2, 11);
+		const expected = await Promise.all([...ids].reverse().map((id) => store.readJob(db, id)));
+		const all = counted();
+		const result = await store.listInboxJobs(all.db, 10, observed);
+		expect(all.queries()).toBe(empty.queries());
+		expect(result).toEqual(expected);
+		expect(result.find((job) => job.id === ids[1])).toMatchObject({
+			attempt: 2,
+			result: { followups: ["Follow-up 1"] },
+			events: [{ attempt: 2 }, { attempt: 1 }],
+			questions: [
+				{ attempt: 1, answer: { text: `Answer ${ids[1]}` }, answeredBy: user, answeredAt: 10, approvedAt: 11 },
+				{ attempt: 2, answer: null, answeredBy: null, answeredAt: null, approvedAt: null },
+			],
+		});
+		expect(result.find((job) => job.id === ids[0])).toMatchObject({ events: [], questions: [] });
+		const active = counted();
+		expect(await store.listActivityJobs(active.db, 10, [])).toEqual(
+			expected.filter((job) => job!.state === "waiting"),
+		);
+		expect(active.queries()).toBe(3);
+		const recent = counted();
+		expect(await store.listJobs(recent.db, 10)).toEqual(expected.slice(0, 50));
+		expect(recent.queries()).toBe(3);
+	});
+
+	it("keeps the entire inbox load's query count fixed as jobs grow", async () => {
+		const read = async () => {
+			const queries = counted();
+			const result = await load({ ...env, DB: queries.db }, user);
+			return { queries: queries.queries(), result };
+		};
+		const empty = await read();
+		await seed(120);
+		const many = await read();
+		expect(many.queries).toBe(empty.queries);
+		expect(many.result.activity[0]).toHaveLength(60);
+		expect(many.result.items.filter((item) => item.job)).toHaveLength(60);
+	});
+});

quest/m1/README.md

@@ -19,6 +19,5 @@ additional login providers. This milestone does not gate the first release.
 - [Live inbox](/quest/m1/inbox-polish.md) - live agent cards, active quests on the log, runner choice for every job, expiring offers
 - [Worker dev setup](/quest/m1/worker-dev.md) - per-server cookies, seeded raw output, one fake GitHub, each test run alone
 - [Merge driver tuning](/quest/m1/resolve-tuning.md) - measure `quest resolve` on moq's history and match git's markers
-- [Batched inbox reads](/quest/m1/inbox-reads.md) - a fixed number of D1 queries per project on each inbox load
 - [Incremental history](/quest/m1/history-pass.md) - each history pass reads only what changed since the last
 - [Gate owns descendants](/quest/m1/gate-owns-descendants.md) - cancelling `quest gate` leaves none of its command's processes behind, on Linux and macOS

quest/m1/history-pass.md

@@ -16,7 +16,6 @@ Decided 2026-10-22 (user) as a follow-up of
 - Page by update time from the last pass's high-water mark, and read bodies
   only for open drafts, whose `## Decisions needed` section decides whether
   the board shows them as claimed or handed back.
-- m1, next to [batched inbox reads](/quest/m1/inbox-reads.md): m0 pilots are
-  small enough that full passes are affordable.
+- m1: m0 pilots are small enough that full passes are affordable.
 - Dropped (user): caching installation tokens across an inbox load, until
   pilots show slow inboxes.