flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

git agents cloudflare rust forge mcp

2,861 linesCode BlameRaw

1	//! The work service: work items, pull requests, comments and sessions.
2	//!
3	//! Other services reach it over `POST /rpc/<method>`; see
4	//! `g1t_contracts::work` for the methods and their arguments. It also
5	//! consumes its queue of events from the bus.
6
7	mod attachments;
8	mod authored;
9	mod automerge;
10	mod checks;
11	mod closing;
12	mod ghost;
13	mod codeowners;
14	mod commit_checks;
15	mod commit_comments;
16	mod contributions;
17	mod coordination;
18	mod decisions;
19	mod guardrails;
20	mod head_repo;
21	mod inbox;
22	mod insights;
23	mod integration;
24	mod items;
25	mod lifecycle;
26	#[cfg(test)]
27	mod migrated;
28	mod merge_message;
29	mod mergeability;
30	mod merging;
31	mod prefetch;
32	mod pushed;
33	mod queue;
34	mod reactions;
35	mod retired;
36	mod reviews;
37	mod rows;
38	mod rulesets;
39	mod runs;
40	mod settings;
41	mod statuses;
42	mod team_reviews;
43
44	use g1t_contracts::events::{
45	ChangedFrom, CommentChanges, CommentCreated, CommentDeleted, CommentEdited, CommentOn, DeletedComment, Event, NewEvent,
46	Publish, PullEvent, SessionAppended,
47	};
48	use g1t_contracts::identity::UsernameArgs;
49	use g1t_contracts::repos::{
50	ForkArgs, GetArgs, HeadArgs, LandArgs, Landed, NeedsAgentReason, PullBranchUpdate, ReadableArgs, Repo, RepoPath,
51	UpdateMethod, UpdatePullBranchArgs,
52	};
53	use g1t_contracts::access::{self, Capability, Denied};
54	use g1t_contracts::time::rfc3339;
55	use g1t_contracts::work::*;
56	use futures_util::future::{try_join, try_join3, try_join_all};
57	use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
58	use g1t_kit::{args, now_ms, reply, rpc_method};
59	use serde::Serialize;
60	use worker::wasm_bindgen::JsValue;
61	use worker::{
62	Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, ScheduleContext, ScheduledEvent,
63	event,
64	};
65
66	use items::Held;
67	use retired::writable;
68	use rows::{CommentRow, MovedRow, NumberRow, PULL_COLUMNS, PullRow, SessionRow, Snapshot};
69
70	pub(crate) const SOURCE: &str = "work";
71	const MAX_ENTRY_BATCH: usize = 200;
72	pub(crate) const MAX_ENTRY_CHARS: usize = 64_000;
73	const MAX_TITLE_CHARS: usize = 200;
74	const SESSION_PAGE: u32 = 500;
75	pub(crate) const LIST_PAGE: u32 = 100;
76	const MAX_ASSIGNEES: usize = 10;
77	pub(crate) const UNVERIFIED: &str = "Confirm your email address first. Check your inbox, or resend the link from the banner on g1t.sh.";
78
79	/// Where a note or a comment goes: a pull request of a repository, by its
80	/// number, or a work item, by its id.
81	#[derive(Clone, Copy, Debug, PartialEq, Eq)]
82	pub(crate) enum On<'a> {
83	Pull(&'a str, u32),
84	Item(&'a str),
85	}
86
87	/// What a comment, a reaction or a set of labels is on: a pull request of
88	/// a repository, or a work item. See `AddCommentArgs`.
89	#[derive(Clone, Debug)]
90	#[allow(clippy::large_enum_variant)]
91	pub(crate) enum Thread {
92	Pull(Repo, Pull),
93	Item(Held),
94	}
95
96	impl Thread {
97	/// Where its comments and notes are kept.
98	pub(crate) fn on(&self) -> On<'_> {
99	match self {
100	Thread::Pull(_, pull) => On::Pull(&pull.repo_id, pull.number),
101	Thread::Item(held) => On::Item(&held.item.id),
102	}
103	}
104
105	/// The id its description's reactions are kept under.
106	pub(crate) fn subject(&self) -> &str {
107	match self {
108	Thread::Pull(_, pull) => &pull.id,
109	Thread::Item(held) => &held.item.id,
110	}
111	}
112
113	/// Where it is, as the `comment.*` events say.
114	pub(crate) fn comment_on(&self) -> CommentOn {
115	match self {
116	Thread::Pull(_, pull) => CommentOn {
117	repo_id: Some(pull.repo_id.clone()),
118	number: pull.number,
119	pull_id: Some(pull.id.clone()),
120	..CommentOn::default()
121	},
122	Thread::Item(held) => CommentOn {
123	repo_id: held.event_repo().map(str::to_owned),
124	number: held.item.number,
125	item_id: Some(held.item.id.clone()),
126	workspace: Some(held.item.workspace.clone()),
127	repo_ids: held.repo_ids(),
128	..CommentOn::default()
129	},
130	}
131	}
132
133	/// The repository its events are filed under.
134	pub(crate) fn event_repo(&self) -> Option<&str> {
135	match self {
136	Thread::Pull(repo, _) => Some(&repo.id),
137	Thread::Item(held) => held.event_repo(),
138	}
139	}
140
141	/// Whether `actor` may do `capability` on it.
142	pub(crate) fn may(&self, actor: &User, capability: Capability) -> bool {
143	match self {
144	Thread::Pull(repo, _) => access::can(Some(actor), repo, capability),
145	Thread::Item(held) => held.may(Some(actor), capability),
146	}
147	}
148	}
149
150	fn no_pull<T>() -> Outcome<T> {
151	Outcome::fail(FailureCode::NotFound, "Pull request not found.")
152	}
153
154	/// Whether a pull request was behind when its mergeability was last
155	/// worked out, and for which pair of commits (mergeability.rs).
156	#[derive(serde::Deserialize)]
157	struct StoredBehind {
158	#[serde(default)]
159	behind: Option<u8>,
160	#[serde(default)]
161	mergeable_key: Option<String>,
162	}
163
164	impl StoredBehind {
165	/// The stored answer, if it was worked out for `head`.
166	fn for_head(&self, head: Option<&str>) -> Option<bool> {
167	let (worked_for, _) = self.mergeable_key.as_deref()?.split_once("..")?;
168	(Some(worked_for) == head).then_some(self.behind? != 0)
169	}
170	}
171
172	/// A pull request's work item, by id, read beside its row.
173	#[derive(serde::Deserialize)]
174	struct PullItemRow {
175	#[serde(default)]
176	item_id: Option<String>,
177	}
178
179	/// Refuses `actor` unless their role on `repo` has `capability`: not found
180	/// when they cannot read it, forbidden with the role it needs otherwise.
181	pub(crate) fn allowed(actor: Option<&User>, repo: &Repo, capability: Capability) -> Outcome<()> {
182	match access::check(actor, repo, capability) {
183	Ok(()) => Outcome::Ok(()),
184	Err(Denied::NotFound) => Outcome::fail(FailureCode::NotFound, "Repository not found."),
185	Err(Denied::Forbidden) => Outcome::fail(
186	FailureCode::Forbidden,
187	access::needs(capability, &format!("{}/{}", repo.namespace, repo.name)),
188	),
189	}
190	}
191
192	pub(crate) fn optional(value: &Option<String>) -> JsValue {
193	value.as_deref().map_or(JsValue::NULL, JsValue::from)
194	}
195
196	pub(crate) fn optional_number(value: Option<u32>) -> JsValue {
197	value.map_or(JsValue::NULL, JsValue::from)
198	}
199
200	/// Names the branch a pull request merges into when it is the default
201	/// branch, which is stored as none so that it follows a change of default.
202	pub(crate) fn fill_base(pull: &mut Pull, repo: &Repo) {
203	if pull.base.as_deref().is_none_or(str::is_empty) {
204	pull.base = Some(repo.default_branch.clone());
205	}
206	}
207
208	/// The branch a pull request is stored as merging into: none for the
209	/// default branch.
210	fn stored_base(base: &str, repo: &Repo) -> Option<String> {
211	let base = base.trim();
212	(!base.is_empty() && base != repo.default_branch).then(\|\| base.to_owned())
213	}
214
215	/// The lowercase name a `State` is stored and sent as.
216	pub(crate) fn state_name(state: Option<State>) -> Option<&'static str> {
217	state.map(\|state\| match state {
218	State::Open => "open",
219	State::Closed => "closed",
220	})
221	}
222
223	/// A trimmed title, or why it cannot be used.
224	pub(crate) fn valid_title(title: &str) -> std::result::Result<&str, &'static str> {
225	let title = title.trim();
226	if title.is_empty() {
227	Err("A title is required.")
228	} else if title.chars().count() > MAX_TITLE_CHARS {
229	Err("That title is too long.")
230	} else {
231	Ok(title)
232	}
233	}
234
235	/// What a closed pull request was when it was closed.
236	#[derive(serde::Deserialize)]
237	struct ClosedFrom {
238	#[serde(default)]
239	closed_from: Option<String>,
240	}
241
242	/// Why a pull request in `status` cannot be reopened, if it cannot: only a
243	/// closed one can, never a merged one.
244	fn reopen_refusal(status: PullStatus) -> Option<&'static str> {
245	match status {
246	PullStatus::Closed => None,
247	PullStatus::Merged => Some("This pull request was merged; it cannot be reopened."),
248	PullStatus::Draft \| PullStatus::Open => Some("This pull request is already open."),
249	}
250	}
251
252	/// What a closed pull request is reopened as: the draft it was, when it
253	/// was closed as one, else ready for review.
254	fn reopened_status(closed_from: Option<&str>) -> PullStatus {
255	if closed_from == Some("draft") { PullStatus::Draft } else { PullStatus::Open }
256	}
257
258	/// Why a pull request in `status` cannot be turned into a draft, if it
259	/// cannot: only one that is open, ready for review, can.
260	fn draft_refusal(status: PullStatus) -> Option<&'static str> {
261	match status {
262	PullStatus::Open => None,
263	PullStatus::Draft => Some("This pull request is already a draft."),
264	PullStatus::Merged => Some("This pull request is already merged."),
265	PullStatus::Closed => Some("This pull request is closed. Reopen it first."),
266	}
267	}
268
269	/// Unwraps an `Outcome`, returning its failure from the enclosing method.
270	macro_rules! check {
271	($outcome:expr) => {
272	match $outcome {
273	Outcome::Ok(value) => value,
274	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
275	}
276	};
277	}
278
279	struct Work {
280	db: D1Database,
281	identity: Fetcher,
282	repos: Fetcher,
283	events: Fetcher,
284	/// GitHub Actions: runs a merge queue's `merge_group` workflows.
285	actions: Fetcher,
286	/// Where this request's time went, for its `Server-Timing`.
287	timing: g1t_kit::d1::Timing,
288	/// A pull request's rows read in one batch for this request
289	/// (prefetch.rs), which the helpers below read instead of the database.
290	prefetched: std::cell::RefCell<Option<std::rc::Rc<prefetch::Prefetched>>>,
291	/// Repositories read in this request, by id, for rules that need one
292	/// where only a pull request is at hand (rulesets.rs `repo_for`).
293	known_repos: std::cell::RefCell<std::collections::HashMap<String, Repo>>,
294	}
295
296	impl Work {
297	async fn publish<T: Serialize>(
298	&self,
299	kind: &'static str,
300	repo_id: &str,
301	actor: &User,
302	data: T,
303	) -> Result<()> {
304	// What a workflow job's token did is marked, so it starts no
305	// workflows (`g1t_contracts::events::CAUSED_BY_JOB`).
306	self.publish_as(kind, repo_id, Some(actor.id.clone()), g1t_contracts::events::marked(data, Some(actor)))
307	.await
308	}
309
310	/// A pull request's owner (whoever asked g1t for it, or its author) as
311	/// a viewer who can read its repository and source. Stored people carry
312	/// no memberships, so a private repository would otherwise look missing
313	/// to them. The membership given reads and nothing more: it is for
314	/// looking, never for acting.
315	pub(crate) async fn owner_viewer(&self, pull: &Pull) -> Result<Viewer> {
316	let path: Option<RepoPath> = g1t_kit::call(
317	&self.repos,
318	"path_by_id",
319	&g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
320	)
321	.await?;
322	let mut owner = pull.owner().clone();
323	if let Some(path) = path
324	&& !owner.is_member(&path.namespace.to_lowercase())
325	{
326	owner.workspaces.push(g1t_contracts::Membership {
327	base_permission: Some(access::BasePermission::Read),
328	..g1t_contracts::Membership::member(path.namespace.to_lowercase())
329	});
330	}
331	Ok(Some(owner))
332	}
333
334	/// Publishes an event caused by `actor`, or by g1t itself.
335	async fn publish_as<T: Serialize>(
336	&self,
337	kind: &'static str,
338	repo_id: &str,
339	actor: Option<String>,
340	data: T,
341	) -> Result<()> {
342	let data = serde_json::to_value(data)?;
343	// Store agent updates before external event delivery. A retry may repeat
344	// an event, but an external delivery failure cannot erase its inbox copy.
345	// Coordination changes write their updates with themselves instead
346	// (coordination::updates) and call `publish_value`.
347	self.deliver_update(kind, repo_id, &data).await?;
348	self.publish_value(kind, repo_id, actor, data).await
349	}
350
351	/// Publishes an event whose session updates are already written.
352	pub(crate) async fn publish_value(
353	&self,
354	kind: &'static str,
355	repo_id: &str,
356	actor: Option<String>,
357	data: serde_json::Value,
358	) -> Result<()> {
359	self.publish_on(kind, Some(repo_id), actor, data).await
360	}
361
362	/// Publishes an event about a repository, or about none (a work item
363	/// filed in its workspace alone).
364	pub(crate) async fn publish_on(
365	&self,
366	kind: &'static str,
367	repo_id: Option<&str>,
368	actor: Option<String>,
369	data: serde_json::Value,
370	) -> Result<()> {
371	let event = NewEvent {
372	kind,
373	source: SOURCE,
374	repo_id: repo_id.map(str::to_owned),
375	actor,
376	data,
377	};
378	g1t_kit::call(
379	&self.events,
380	"publish",
381	&Publish {
382	events: vec![event],
383	},
384	)
385	.await
386	}
387
388	/// The repository, if the viewer may see it. Whether they may is
389	/// decided by the repos service.
390	async fn repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
391	let found: Outcome<Repo> = self
392	.timing
393	.rpc(g1t_kit::call(
394	&self.repos,
395	"get",
396	&GetArgs {
397	path: path.clone(),
398	viewer: viewer.clone(),
399	},
400	))
401	.await?;
402	if let Outcome::Ok(repo) = &found {
403	self.known_repos.borrow_mut().insert(repo.id.clone(), repo.clone());
404	}
405	Ok(found)
406	}
407
408	async fn pull(&self, repo_id: &str, number: u32) -> Result<Option<Pull>> {
409	Ok(self
410	.db
411	.prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? AND number = ?"))
412	.bind(&[repo_id.into(), number.into()])?
413	.first::<PullRow>(None)
414	.await?
415	.map(Pull::from))
416	}
417
418	/// The thread a pair of `repo` (a pull request of it) or `workspace`
419	/// (a work item of it) and `number` names, if `viewer` may read it.
420	pub(crate) async fn thread(&self, viewer: &Viewer, repo: Option<&RepoPath>, workspace: &str, number: u32) -> Result<Outcome<Thread>> {
421	Ok(match repo {
422	Some(path) => match self.pull_at(path, number, viewer).await? {
423	Outcome::Ok((repo, pull)) => Outcome::Ok(Thread::Pull(repo, pull)),
424	Outcome::Fail(failure) => Outcome::Fail(failure),
425	},
426	None => match items::for_viewer(self, viewer, workspace, number, Capability::Read).await? {
427	Outcome::Ok(held) => Outcome::Ok(Thread::Item(held)),
428	Outcome::Fail(failure) => Outcome::Fail(failure),
429	},
430	})
431	}
432
433	/// The repository and one of its pull requests, as seen by `viewer`.
434	async fn pull_at(
435	&self,
436	path: &RepoPath,
437	number: u32,
438	viewer: &Viewer,
439	) -> Result<Outcome<(Repo, Pull)>> {
440	let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
441	return Ok(no_pull());
442	};
443	Ok(match self.pull(&repo.id, number).await? {
444	Some(pull) => Outcome::Ok((repo, pull)),
445	None => no_pull(),
446	})
447	}
448
449	/// Records something that happened to a pull request, so that it shows
450	/// in the conversation where it happened. `text` is what `author` did,
451	/// as the rest of a sentence starting with their name: an id and a
452	/// name, or someone acting, as [`rows::By`] records them.
453	pub(crate) async fn note<'a>(
454	&self,
455	repo_id: &str,
456	number: u32,
457	author: impl Into<rows::By<'a>>,
458	text: &str,
459	) -> Result<()> {
460	self.note_on(On::Pull(repo_id, number), author.into(), text).await
461	}
462
463	/// Records something that happened, on a pull request or a work item.
464	pub(crate) async fn note_on(&self, on: On<'_>, author: rows::By<'_>, text: &str) -> Result<()> {
465	let now = now_ms();
466	let (repo_id, number, item_id) = match on {
467	On::Pull(repo_id, number) => (JsValue::from(repo_id), JsValue::from(number), JsValue::NULL),
468	On::Item(item_id) => (JsValue::NULL, JsValue::NULL, JsValue::from(item_id)),
469	};
470	self.db
471	.prepare(
472	"INSERT INTO comments
473	(id, repo_id, number, item_id, author_id, author_name, body, kind, created_at,
474	acting_for_id, acting_for_name)
475	VALUES (?, ?, ?, ?, ?, ?, ?, 'event', ?, ?, ?)",
476	)
477	.bind(&[
478	new_id("cmt", now).into(),
479	repo_id,
480	number,
481	item_id,
482	author.id.into(),
483	author.name.into(),
484	text.into(),
485	rfc3339(now).into(),
486	author.acting_for.map_or(JsValue::NULL, \|(id, _)\| id.into()),
487	author.acting_for.map_or(JsValue::NULL, \|(_, name)\| name.into()),
488	])?
489	.run()
490	.await?;
491	Ok(())
492	}
493
494	/// Notes who was added to and removed from a list of people, such as
495	/// "assigned ana" or "requested a review from g1t".
496	pub(crate) async fn note_changes(
497	&self,
498	on: On<'_>,
499	actor: &User,
500	before: &[String],
501	after: &[String],
502	(added, removed): (&str, &str),
503	) -> Result<()> {
504	let joined = \|names: Vec<&String>\| {
505	names
506	.into_iter()
507	.map(String::as_str)
508	.collect::<Vec<_>>()
509	.join(", ")
510	};
511	let new: Vec<&String> = after.iter().filter(\|name\| !before.contains(name)).collect();
512	let gone: Vec<&String> = before.iter().filter(\|name\| !after.contains(name)).collect();
513	let who = actor;
514	if !new.is_empty() {
515	// Taking something on oneself reads better said that way (not
516	// when g1t, acting for them, did it).
517	let text = if added == "assigned" && new == [&actor.username] && actor.acting.is_none() {
518	"self-assigned this".to_owned()
519	} else {
520	format!("{added} {}", joined(new))
521	};
522	self.note_on(on, who.into(), &text).await?;
523	}
524	if !gone.is_empty() {
525	self.note_on(on, who.into(), &format!("{removed} {}", joined(gone))).await?;
526	}
527	Ok(())
528	}
529
530	/// The commit a pull request's change is at in git right now: its
531	/// working copy's default branch, or its branch, of the repository or
532	/// of the fork it is in.
533	async fn live_head(&self, pull: &Pull) -> Result<Option<String>> {
534	g1t_kit::call(
535	&self.repos,
536	"head",
537	&HeadArgs {
538	repo_id: pull.source_id().to_owned(),
539	branch: pull.branch.clone().unwrap_or_default(),
540	},
541	)
542	.await
543	}
544
545	fn pull_event(pull: &Pull) -> PullEvent {
546	PullEvent {
547	pull_id: pull.id.clone(),
548	repo_id: pull.repo_id.clone(),
549	number: pull.number,
550	author: Some((&pull.author).into()),
551	requested_by: pull.requested_by.as_ref().map(Into::into),
552	item: pull.item,
553	..PullEvent::default()
554	}
555	}
556
557	/// Usernames as given, tidied, if each names an account.
558	async fn valid_assignees(&self, names: Vec<String>) -> Result<Outcome<Vec<String>>> {
559	let mut assignees: Vec<String> = Vec::new();
560	for name in names {
561	let name = name.trim().trim_start_matches('@').to_lowercase();
562	if name.is_empty() \|\| assignees.contains(&name) {
563	continue;
564	}
565	if assignees.len() == MAX_ASSIGNEES {
566	return Ok(Outcome::fail(
567	FailureCode::Invalid,
568	format!("A work item or pull request can be assigned to at most {MAX_ASSIGNEES} people."),
569	));
570	}
571	// g1t has no account, and takes the work on: a session on the
572	// workspace's own Claude (integrations' summon.rs).
573	if name == g1t_contracts::system::USERNAME {
574	assignees.push(name);
575	continue;
576	}
577	let account: Viewer = g1t_kit::call(
578	&self.identity,
579	"user_by_username",
580	&UsernameArgs {
581	username: name.clone(),
582	},
583	)
584	.await?;
585	if account.is_none() {
586	return Ok(Outcome::fail(
587	FailureCode::Invalid,
588	format!("There is no account named {name}."),
589	));
590	}
591	assignees.push(name);
592	}
593	Ok(Outcome::Ok(assignees))
594	}
595
596	/// How many work items linked to a repository are open, and how many
597	/// of its pull requests.
598	async fn counts(&self, a: ViewArgs) -> Result<Outcome<Counts>> {
599	let read = \|repo_id: String\| async move {
600	let query = self
601	.db
602	.prepare(concat!(
603	"SELECT
604	",
605	items::sql::open_items_linked_to!("?1"),
606	" AS items,
607	(SELECT count(*) FROM pulls
608	WHERE repo_id = ?1 AND status IN ('draft', 'open')) AS pulls",
609	))
610	.bind(&[repo_id.into()])?;
611	self.timing.db(1, query.first::<Counts>(None)).await
612	};
613	let (_, counts) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
614	Ok(Outcome::Ok(counts.unwrap_or(Counts { items: 0, pulls: 0 })))
615	}
616
617	// --- Comments ----------------------------------------------------------
618
619	async fn add_comment(&self, a: AddCommentArgs) -> Result<Outcome<Comment>> {
620	if !a.actor.verified {
621	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
622	}
623	let body = a.body.trim();
624	// An approval speaks for itself; anything else has to say something.
625	if body.is_empty() && a.verdict != Some(Verdict::Approve) {
626	return Ok(Outcome::fail(
627	FailureCode::Invalid,
628	"A comment cannot be empty.",
629	));
630	}
631	let path = a
632	.path
633	.as_deref()
634	.map(str::trim)
635	.filter(\|path\| !path.is_empty());
636	let line = a.line.filter(\|line\| *line > 0 && path.is_some());
637	if body.chars().count() > MAX_ENTRY_CHARS {
638	return Ok(Outcome::fail(
639	FailureCode::Invalid,
640	"That comment is too long.",
641	));
642	}
643	let thread = check!(self.thread(&Some(a.actor.clone()), a.repo.as_ref(), &a.workspace, a.number).await?);
644	match &thread {
645	Thread::Pull(repo, pull) => {
646	check!(writable(repo));
647	if a.verdict.is_some() && pull.is_owned_by(&a.actor.id) {
648	return Ok(Outcome::fail(
649	FailureCode::Forbidden,
650	"You cannot approve or request changes on your own pull request.",
651	));
652	}
653	}
654	Thread::Item(held) => {
655	check!(held.allowed(Some(&a.actor), Capability::Participate));
656	if path.is_some() \|\| a.verdict.is_some() {
657	return Ok(Outcome::fail(
658	FailureCode::Invalid,
659	"Only a pull request can be reviewed or commented on by line.",
660	));
661	}
662	}
663	}
664
665	let now = now_ms();
666	// A session g1t started for someone writes as g1t, for them.
667	let by = rows::By::from(&a.actor);
668	let comment = Comment {
669	kind: CommentKind::Comment,
670	id: new_id("cmt", now),
671	author: if by.id == a.actor.id { a.actor.clone() } else { rows::user(by.id.to_owned(), by.name.to_owned()) },
672	body: body.to_owned(),
673	path: path.map(str::to_owned),
674	line,
675	verdict: a.verdict,
676	created_at: rfc3339(now),
677	edited_at: None,
678	agent: None,
679	acting_for: by.acting_for.map(\|(id, name)\| rows::user(id.to_owned(), name.to_owned())),
680	advisory: false,
681	reactions: Vec::new(),
682	};
683	let (repo_id, number, item_id) = match thread.on() {
684	On::Pull(repo_id, number) => (JsValue::from(repo_id), JsValue::from(number), JsValue::NULL),
685	On::Item(item_id) => (JsValue::NULL, JsValue::NULL, JsValue::from(item_id)),
686	};
687	let touched = match &thread {
688	Thread::Pull(_, pull) => self
689	.db
690	.prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
691	.bind(&[comment.created_at.as_str().into(), pull.id.as_str().into()])?,
692	Thread::Item(held) => items::stamp(self, &held.item.id, &comment.created_at)?,
693	};
694	self.db
695	.batch(vec![
696	self.db
697	.prepare(
698	"INSERT INTO comments
699	(id, repo_id, number, item_id, author_id, author_name, body, path, line,
700	verdict, created_at, acting_for_id, acting_for_name)
701	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
702	)
703	.bind(&[
704	comment.id.as_str().into(),
705	repo_id,
706	number,
707	item_id,
708	by.id.into(),
709	by.name.into(),
710	body.into(),
711	optional(&comment.path),
712	optional_number(line),
713	a.verdict
714	.map_or(JsValue::NULL, \|verdict\| verdict.as_str().into()),
715	comment.created_at.as_str().into(),
716	by.acting_for.map_or(JsValue::NULL, \|(id, _)\| id.into()),
717	by.acting_for.map_or(JsValue::NULL, \|(_, name)\| name.into()),
718	])?,
719	touched,
720	])
721	.await?;
722	self.publish_thread(
723	"comment.created",
724	&thread,
725	&a.actor,
726	CommentCreated {
727	comment_id: comment.id.clone(),
728	on: thread.comment_on(),
729	verdict: a.verdict,
730	..CommentCreated::default()
731	},
732	)
733	.await?;
734	// An approval is a decision: a person's is approved.
735	if let Thread::Pull(repo, pull) = &thread
736	&& a.verdict == Some(Verdict::Approve)
737	{
738	let said = if body.is_empty() { String::new() } else { format!(": {}", body.chars().take(300).collect::<String>()) };
739	self.keep_decision_quietly(
740	&repo.id,
741	decisions::NewDecision {
742	item: self.pull_item_id(&pull.id).await?,
743	pull: Some(a.number),
744	session_id: None,
745	actor: Some(&a.actor),
746	kind: g1t_contracts::decisions::DecisionKind::Approval,
747	confidence: decisions::approval_confidence(&a.actor),
748	summary: format!("{} approved #{}{said}", a.actor.username, a.number),
749	evidence: serde_json::json!({ "comment": comment.id }),
750	},
751	)
752	.await;
753	}
754	Ok(Outcome::Ok(comment))
755	}
756
757	/// Publishes an event about a thread, filed under its repository.
758	pub(crate) async fn publish_thread<T: Serialize>(&self, kind: &'static str, thread: &Thread, actor: &User, data: T) -> Result<()> {
759	match thread.event_repo() {
760	Some(repo_id) => self.publish(kind, repo_id, actor, data).await,
761	None => {
762	let data = g1t_contracts::events::marked(data, Some(actor));
763	self.publish_on(kind, None, Some(actor.id.clone()), data).await
764	}
765	}
766	}
767
768	/// A comment, with the thread it is on, when `actor` may edit it, or
769	/// with `deleting`, delete it (`may_change_comment`).
770	async fn changeable_comment(&self, a: &CommentActionArgs, deleting: bool) -> Result<Outcome<(Thread, CommentRow)>> {
771	let actor = Some(a.actor.clone());
772	let id = a.comment_id.trim();
773	let not_found = \|\| Ok(Outcome::fail(FailureCode::NotFound, "Comment not found."));
774	let (thread, row) = match &a.repo {
775	Some(path) => {
776	let repo = check!(self.repo(path, &actor).await?);
777	check!(writable(&repo));
778	let found = self
779	.db
780	.prepare("SELECT * FROM comments WHERE id = ? AND repo_id = ?")
781	.bind(&[id.into(), repo.id.as_str().into()])?
782	.first::<CommentRow>(None)
783	.await?;
784	let Some(row) = found else { return not_found() };
785	let Some(pull) = self.pull(&repo.id, row.number.unwrap_or_default()).await? else { return not_found() };
786	(Thread::Pull(repo, pull), row)
787	}
788	None => {
789	let found = self
790	.db
791	.prepare("SELECT * FROM comments WHERE id = ? AND item_id IS NOT NULL")
792	.bind(&[id.into()])?
793	.first::<CommentRow>(None)
794	.await?;
795	let Some(row) = found else { return not_found() };
796	let held = match row.item_id.as_deref() {
797	Some(item_id) => items::by_id(self, item_id).await?,
798	None => None,
799	};
800	let Some(held) = held.filter(\|held\| held.item.workspace == items::workspace_of(&a.workspace, None)) else {
801	return not_found();
802	};
803	if !held.may(actor.as_ref(), Capability::Read) {
804	return not_found();
805	}
806	(Thread::Item(held), row)
807	}
808	};
809	let maintains = thread.may(&a.actor, Capability::ManageSettings);
810	// What an agent wrote as itself is answered for by whoever it acted for.
811	let author_id = row.answerable_id().to_owned();
812	if let Err(refusal) = may_change_comment(
813	row.kind,
814	row.has_verdict(),
815	&author_id,
816	&a.actor.id,
817	maintains,
818	deleting,
819	) {
820	// Someone who may change it otherwise was refused for what it is.
821	let code = if author_id == a.actor.id \|\| maintains { FailureCode::Conflict } else { FailureCode::Forbidden };
822	return Ok(Outcome::fail(code, refusal));
823	}
824	Ok(Outcome::Ok((thread, row)))
825	}
826
827	/// Changes the text of a comment: its author's to do, or a
828	/// maintainer's. Publishes `comment.edited` with what it said before.
829	async fn edit_comment(&self, a: CommentActionArgs) -> Result<Outcome<Comment>> {
830	let body = a.body.trim().to_owned();
831	if body.chars().count() > MAX_ENTRY_CHARS {
832	return Ok(Outcome::fail(FailureCode::Invalid, "That comment is too long."));
833	}
834	let (thread, row) = check!(self.changeable_comment(&a, false).await?);
835	// An approval speaks for itself; anything else has to say something.
836	if body.is_empty() && row.verdict != Some(Verdict::Approve) && row.agent_verdict.as_deref() != Some("approve") {
837	return Ok(Outcome::fail(FailureCode::Invalid, "A comment cannot be empty."));
838	}
839	let mut comment = Comment::from(row);
840	if comment.body == body {
841	return Ok(Outcome::Ok(comment));
842	}
843	let now = rfc3339(now_ms());
844	self.db
845	.prepare("UPDATE comments SET body = ?, edited_at = ? WHERE id = ?")
846	.bind(&[body.as_str().into(), now.as_str().into(), comment.id.as_str().into()])?
847	.run()
848	.await?;
849	let before = std::mem::replace(&mut comment.body, body);
850	comment.edited_at = Some(now);
851	self.publish_thread(
852	"comment.edited",
853	&thread,
854	&a.actor,
855	CommentEdited {
856	comment_id: comment.id.clone(),
857	on: thread.comment_on(),
858	changes: CommentChanges { body: ChangedFrom { from: before } },
859	},
860	)
861	.await?;
862	Ok(Outcome::Ok(comment))
863	}
864
865	/// Deletes a comment: its author's to do, or a maintainer's. A review
866	/// that gave a verdict stays. Publishes `comment.deleted` with the
867	/// comment as it was.
868	async fn delete_comment(&self, a: CommentActionArgs) -> Result<Outcome<bool>> {
869	let (thread, row) = check!(self.changeable_comment(&a, true).await?);
870	self.db
871	.prepare("DELETE FROM comments WHERE id = ?")
872	.bind(&[row.id.as_str().into()])?
873	.run()
874	.await?;
875	self.db.prepare("DELETE FROM reactions WHERE subject = ?").bind(&[row.id.as_str().into()])?.run().await?;
876	let comment = Comment::from(row);
877	self.publish_thread(
878	"comment.deleted",
879	&thread,
880	&a.actor,
881	CommentDeleted {
882	comment_id: comment.id.clone(),
883	on: thread.comment_on(),
884	comment: DeletedComment {
885	id: comment.id,
886	body: comment.body,
887	author: (&comment.author).into(),
888	created_at: comment.created_at,
889	path: comment.path,
890	line: comment.line,
891	},
892	},
893	)
894	.await?;
895	Ok(Outcome::Ok(true))
896	}
897
898	// --- Pull requests -----------------------------------------------------
899
900	async fn open_pull(&self, a: OpenPullArgs) -> Result<Outcome<Pull>> {
901	if !a.actor.verified {
902	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
903	}
904	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
905	check!(writable(&repo));
906	let workspace = repo.namespace.to_lowercase();
907	// A work item of the repository's workspace, which the actor can see.
908	let item = match a.item {
909	Some(number) => match items::for_viewer(self, &Some(a.actor.clone()), &workspace, number, Capability::Read).await? {
910	Outcome::Ok(held) if held.item.state == State::Open => Some(held),
911	Outcome::Ok(_) => {
912	return Ok(Outcome::fail(FailureCode::Conflict, "This work item is closed."));
913	}
914	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
915	},
916	None => None,
917	};
918	// A pull request for a work item takes its title unless given one.
919	let title = match (a.title.trim(), &item) {
920	("", Some(held)) => held.item.title.clone(),
921	(title, _) => match valid_title(title) {
922	Ok(title) => title.to_owned(),
923	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
924	},
925	};
926	// Unnamed, the change is its author's, unless an agent opened it.
927	let agent = match a.agent.trim() {
928	"" if g1t_contracts::rules::is_agent(&a.actor) => "agent",
929	"" => a.actor.username.as_str(),
930	agent => agent,
931	};
932	let runtime = match a.runtime {
933	Runtime::Hosted => "hosted",
934	Runtime::External => "external",
935	};
936
937	let now = now_ms();
938	let id = new_id("pr", now);
939	let branch = a
940	.branch
941	.as_deref()
942	.map(str::trim)
943	.filter(\|branch\| !branch.is_empty());
944	// A branch of someone's fork of the repository, rather than of the
945	// repository itself (head_repo.rs).
946	let head_repo = match &a.head_repo {
947	Some(path) => {
948	if branch.is_none() {
949	return Ok(Outcome::fail(
950	FailureCode::Invalid,
951	format!("Say which branch of {}/{} holds the change.", path.namespace, path.name),
952	));
953	}
954	check!(self.head_repo_for(path, &repo, &a.actor).await?)
955	}
956	None => None,
957	};
958	let head_path = head_repo.as_ref().map(\|fork\| RepoPath { namespace: fork.namespace.clone(), name: fork.name.clone() });
959	// The branch it merges into: the default branch unless another is
960	// asked for, which has to exist.
961	let base = a.base.as_deref().and_then(\|base\| stored_base(base, &repo));
962	if let Some(base) = &base {
963	if head_repo.is_none() && branch == Some(base.as_str()) {
964	return Ok(Outcome::fail(
965	FailureCode::Invalid,
966	format!("A pull request cannot merge {base} into itself. Choose another base."),
967	));
968	}
969	let exists: Option<String> = g1t_kit::call(
970	&self.repos,
971	"head",
972	&HeadArgs { repo_id: repo.id.clone(), branch: base.clone() },
973	)
974	.await?;
975	if exists.is_none() {
976	return Ok(Outcome::fail(FailureCode::NotFound, format!("There is no branch named {base} to merge into.")));
977	}
978	}
979	let base_name = base.clone().unwrap_or_else(\|\| repo.default_branch.clone());
980	// The change is on a branch already pushed to the repository or to
981	// a fork of it, or will be made in a fork created for this pull
982	// request.
983	let head_repo_id = head_repo.as_ref().map(\|fork\| fork.id.clone());
984	let (fork, head) = match branch {
985	Some(branch) => {
986	let source = head_repo::source_label(head_path.as_ref(), branch);
987	// A fork's branch may share the base's name: it is another
988	// repository's.
989	if head_repo.is_none() && branch == base_name {
990	return Ok(Outcome::fail(
991	FailureCode::Invalid,
992	format!("Choose a branch other than {branch}."),
993	));
994	}
995	let head: Option<String> = g1t_kit::call(
996	&self.repos,
997	"head",
998	&HeadArgs {
999	repo_id: head_repo_id.clone().unwrap_or_else(\|\| repo.id.clone()),
1000	branch: branch.to_owned(),
1001	},
1002	)
1003	.await?;
1004	let Some(head) = head else {
1005	return Ok(Outcome::fail(
1006	FailureCode::NotFound,
1007	format!("There is no branch named {source}. Push it first."),
1008	));
1009	};
1010	// One open pull request for each branch (of the repository
1011	// or of one fork) and base.
1012	let existing = self
1013	.db
1014	.prepare(head_repo::OPEN_FROM_BRANCH)
1015	.bind(&[repo.id.as_str().into(), branch.into(), optional(&head_repo_id), optional(&base)])?
1016	.first::<NumberRow>(None)
1017	.await?;
1018	if let Some(existing) = existing {
1019	return Ok(Outcome::fail(
1020	FailureCode::Conflict,
1021	format!("Pull request #{} is already open from {source} into {base_name}.", existing.n),
1022	));
1023	}
1024	(None, Some(head))
1025	}
1026	None => {
1027	let fork: Outcome<Repo> = g1t_kit::call(
1028	&self.repos,
1029	"fork_for_pull",
1030	&ForkArgs {
1031	source_id: repo.id.clone(),
1032	pull_id: id.clone(),
1033	actor: a.actor.clone(),
1034	},
1035	)
1036	.await?;
1037	(Some(check!(fork)), None)
1038	}
1039	};
1040	// A branch already holds the work, so its pull request is ready for
1041	// review from the start; one with a fork starts as a draft.
1042	// Asked for as a draft, it waits until it is marked ready.
1043	let status = if branch.is_some() && !a.draft { "open" } else { "draft" };
1044	let body = Some(a.body.trim().to_owned()).filter(\|body\| !body.is_empty());
1045
1046	// Numbered in the workspace, never as one of this repository's are.
1047	items::remember_repo(self, &repo).await?;
1048	let number = items::next_number(self, &workspace, Some(&repo.id)).await?;
1049	let timestamp = rfc3339(now);
1050	let (author, requested_by) = authorship(&a.actor);
1051	// The work items it says it closes and is part of (closing.rs).
1052	let item_id = item.as_ref().map(\|held\| held.item.id.clone());
1053	let named = items::named(self, &repo, item_id.as_deref(), &format!("{title}\n{}", body.as_deref().unwrap_or_default())).await?;
1054	self.db
1055	.prepare(
1056	"INSERT INTO pulls
1057	(id, repo_id, number, item_id, title, body, agent, runtime,
1058	status, fork_repo_id, fork_namespace, fork_name, source_branch, head_commit,
1059	author_id, author_name, requested_by_id, requested_by_name, created_at, updated_at,
1060	base_branch, head_repo_id, head_namespace, head_name, closes_items, parts)
1061	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
1062	)
1063	.bind(&[
1064	id.as_str().into(),
1065	repo.id.as_str().into(),
1066	number.into(),
1067	optional(&item_id),
1068	title.into(),
1069	optional(&body),
1070	agent.into(),
1071	runtime.into(),
1072	status.into(),
1073	optional(&fork.as_ref().map(\|fork\| fork.id.clone())),
1074	optional(&fork.as_ref().map(\|fork\| fork.namespace.clone())),
1075	optional(&fork.as_ref().map(\|fork\| fork.name.clone())),
1076	optional(&branch.map(str::to_owned)),
1077	optional(&head),
1078	author.id.as_str().into(),
1079	author.username.as_str().into(),
1080	optional(&requested_by.as_ref().map(\|user\| user.id.clone())),
1081	optional(&requested_by.as_ref().map(\|user\| user.username.clone())),
1082	timestamp.as_str().into(),
1083	timestamp.as_str().into(),
1084	optional(&base),
1085	optional(&head_repo_id),
1086	optional(&head_path.as_ref().map(\|path\| path.namespace.clone())),
1087	optional(&head_path.as_ref().map(\|path\| path.name.clone())),
1088	serde_json::to_string(&named.close_ids())?.into(),
1089	serde_json::to_string(&named.part_ids())?.into(),
1090	])?
1091	.run()
1092	.await?;
1093	let Some(mut pull) = self.pull(&repo.id, number).await? else {
1094	return Ok(no_pull());
1095	};
1096	fill_base(&mut pull, &repo);
1097	// The work item hears of it, and links the repository it is in.
1098	if let Some(held) = &item {
1099	let text = format!("opened {}#{} for this", items::path_of(&repo), pull.number);
1100	items::note(self, &held.item.id, &a.actor, &text).await?;
1101	items::touch(self, held, &repo, &a.actor).await?;
1102	}
1103	// Its head and merge refs in the repository, before its events go out.
1104	self.keep_pull_refs(&pull).await;
1105	self.publish(
1106	"pull.opened",
1107	&repo.id,
1108	&a.actor,
1109	PullEvent {
1110	agent: Some(pull.agent.clone()),
1111	base: pull.base.clone(),
1112	..Self::pull_event(&pull)
1113	},
1114	)
1115	.await?;
1116	// Its code owners asked to review (codeowners.rs).
1117	self.refresh_code_owners(&pull).await;
1118	let pull = self.pull(&repo.id, number).await?.unwrap_or(pull);
1119	// A branch pushed before may overlap others from the start.
1120	self.announce_overlaps(&pull, Some(a.actor.id.clone())).await;
1121	Ok(Outcome::Ok(pull))
1122	}
1123
1124	async fn list_pulls(&self, a: ListPullsArgs) -> Result<Outcome<Vec<Pull>>> {
1125	let filter = match a.state {
1126	Some(State::Open) => "AND status IN ('draft', 'open')",
1127	Some(State::Closed) => "AND status IN ('merged', 'closed')",
1128	None => "",
1129	};
1130	let label = a
1131	.label
1132	.map(\|label\| label.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase())
1133	.filter(\|label\| !label.is_empty());
1134	let milestone = a.milestone;
1135	let read = \|repo_id: String\| {
1136	let label = label.clone();
1137	async move {
1138	let query = self
1139	.db
1140	.prepare(format!(
1141	"SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? {filter}
1142	AND (? IS NULL OR EXISTS
1143	(SELECT 1 FROM json_each(pulls.labels) WHERE json_each.value = ?))
1144	AND (? IS NULL OR milestone = ?)
1145	ORDER BY number DESC LIMIT ?"
1146	))
1147	.bind(&[
1148	repo_id.into(),
1149	optional(&label),
1150	optional(&label),
1151	optional_number(milestone),
1152	optional_number(milestone),
1153	LIST_PAGE.into(),
1154	])?;
1155	self.timing.db(1, query.all()).await?.results::<PullRow>()
1156	}
1157	};
1158	let (repo, rows) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
1159	let base = a.base.map(\|base\| base.trim().to_owned()).filter(\|base\| !base.is_empty());
1160	Ok(Outcome::Ok(
1161	rows.into_iter()
1162	.map(\|row\| {
1163	let mut pull = Pull::from(row);
1164	fill_base(&mut pull, &repo);
1165	pull
1166	})
1167	.filter(\|pull\| base.as_deref().is_none_or(\|base\| pull.base.as_deref() == Some(base)))
1168	.collect(),
1169	))
1170	}
1171
1172	/// `pulls_for_repos`: what `list_pulls` gives, open and closed, for many
1173	/// repositories at once: one access check with repos for all of them
1174	/// and one query, instead of two of each per repository.
1175	async fn pulls_for_repos(&self, a: PullsForReposArgs) -> Result<Vec<RepoPulls>> {
1176	let ids: Vec<String> = a.repo_ids.into_iter().take(MAX_PULLS_FOR_REPOS).collect();
1177	if ids.is_empty() {
1178	return Ok(Vec::new());
1179	}
1180	let limit = a.limit.clamp(1, LIST_PAGE);
1181	// The rows are read beside the access check, for every id asked
1182	// about; those of repositories the viewer cannot read are dropped.
1183	let asked = serde_json::to_string(&ids)?;
1184	let check = ReadableArgs { ids, viewer: a.viewer };
1185	let readable = self.timing.rpc(g1t_kit::call::<_, Vec<Repo>>(&self.repos, "readable", &check));
1186	let (readable, rows) = try_join(readable, self.timing.db(1, self.newest_pulls(asked, limit))).await?;
1187	if readable.is_empty() {
1188	return Ok(Vec::new());
1189	}
1190	let mut answer: Vec<RepoPulls> = readable
1191	.iter()
1192	.map(\|repo\| RepoPulls { repo_id: repo.id.clone(), open: Vec::new(), closed: Vec::new() })
1193	.collect();
1194	for pull in rows.into_iter().map(Pull::from) {
1195	let Some(entry) = answer.iter_mut().find(\|entry\| entry.repo_id == pull.repo_id) else {
1196	continue;
1197	};
1198	match pull.status {
1199	PullStatus::Draft \| PullStatus::Open => entry.open.push(pull),
1200	PullStatus::Merged \| PullStatus::Closed => entry.closed.push(pull),
1201	}
1202	}
1203	for entry in &mut answer {
1204	entry.open.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1205	entry.closed.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1206	}
1207	Ok(answer)
1208	}
1209
1210	/// The newest `limit` of each repository's open (draft or open) and
1211	/// closed (merged or closed) pull requests, for the ids in `ids` (JSON).
1212	async fn newest_pulls(&self, ids: String, limit: u32) -> Result<Vec<PullRow>> {
1213	self.db
1214	.prepare(format!(
1215	"SELECT * FROM (
1216	SELECT {PULL_COLUMNS}, ROW_NUMBER() OVER (
1217	PARTITION BY pulls.repo_id, pulls.status IN ('draft', 'open') ORDER BY pulls.number DESC
1218	) AS place
1219	FROM pulls WHERE pulls.repo_id IN (SELECT value FROM json_each(?1))
1220	) WHERE place <= ?2"
1221	))
1222	.bind(&[ids.into(), limit.into()])?
1223	.all()
1224	.await?
1225	.results::<PullRow>()
1226	}
1227
1228	async fn get_pull(&self, a: ViewArgs) -> Result<Outcome<PullDetail>> {
1229	let number = a.number;
1230	// Every row the page and the lifecycle read, in one batch started
1231	// beside the access check; the helpers below read from it.
1232	let namespace = a.repo.namespace.clone();
1233	let read = \|repo_id: String\| self.prefetch_pull(repo_id, namespace.clone(), number);
1234	let Outcome::Ok((repo, Some(found))) = self.repo_then(&a.repo, &a.viewer, read).await? else {
1235	return Ok(no_pull());
1236	};
1237	let Some(row) = found.first::<PullRow>(prefetch::Slot::Pull)? else {
1238	return Ok(no_pull());
1239	};
1240	let stored = found.first::<StoredBehind>(prefetch::Slot::Pull)?;
1241	let item = found.held(prefetch::Slot::Item)?.map(\|held\| held.item);
1242	let mut comments: Vec<Comment> =
1243	found.rows::<CommentRow>(prefetch::Slot::Comments)?.into_iter().map(Comment::from).collect();
1244	let reactions = reactions::attach(&found.rows::<reactions::ReactionRow>(prefetch::Slot::Reactions)?, &row.id, &mut comments);
1245	self.keep_prefetched(Some(found));
1246	let default_branch = repo.default_branch.clone();
1247	let detail = self.pull_detail(repo, Pull::from(row), item, comments, stored, &a.viewer).await;
1248	self.keep_prefetched(None);
1249	// The branch it merges into, named, for whoever reads it.
1250	Ok(match detail? {
1251	Outcome::Ok(mut detail) => {
1252	if detail.pull.base.as_deref().is_none_or(str::is_empty) {
1253	detail.pull.base = Some(default_branch);
1254	}
1255	detail.reactions = reactions;
1256	Outcome::Ok(detail)
1257	}
1258	failed => failed,
1259	})
1260	}
1261
1262	async fn pull_detail(
1263	&self,
1264	repo: Repo,
1265	mut pull: Pull,
1266	item: Option<WorkItem>,
1267	comments: Vec<Comment>,
1268	stored: Option<StoredBehind>,
1269	viewer: &Viewer,
1270	) -> Result<Outcome<PullDetail>> {
1271	// Whether it is behind, as worked out with its mergeability on the
1272	// last push to either side (mergeability.rs), when that was for
1273	// its head as it is now; otherwise asked of the repos service.
1274	let known_behind = stored.and_then(\|stored\| stored.for_head(pull.head_commit.as_deref()));
1275	// Worked out on each push; this covers a pull request from before
1276	// that was recorded.
1277	if pull.files.is_empty() && pull.head_commit.is_some() {
1278	pull.files = self.refresh_files(&pull).await?;
1279	}
1280	// Everything else at once: none of it depends on the rest, and each
1281	// is a round trip of its own.
1282	let standing = async {
1283	// Mergeability first: where g1t sees a pull request through, a
1284	// conflict decides its next step.
1285	let behind = async {
1286	match known_behind {
1287	Some(behind) => Ok(behind),
1288	None => {
1289	let behind = self.is_behind(&pull).await?;
1290	// Kept for the next view when the mergeability on
1291	// record is for this head: a pull request from
1292	// before `behind` was kept asks once.
1293	if let Some(head) = pull.head_commit.as_deref()
1294	&& pull.status.is_active()
1295	{
1296	self.db
1297	.prepare(
1298	"UPDATE pulls SET behind = ?1
1299	WHERE id = ?2 AND behind IS NULL AND mergeable_key LIKE ?3 \|\| '..%'",
1300	)
1301	.bind(&[u32::from(behind).into(), pull.id.as_str().into(), head.into()])?
1302	.run()
1303	.await?;
1304	}
1305	Ok(behind)
1306	}
1307	}
1308	};
1309	let (merge, behind) = try_join(self.mergeability(&pull), behind).await?;
1310	let lifecycle = self.assess(&pull, &item, behind).await?.map(\|(lifecycle, _)\| lifecycle);
1311	Ok::<_, worker::Error>((behind, lifecycle, merge))
1312	};
1313	let (((behind, lifecycle, (mergeable, conflicts)), (landing, stalled), comments), (checks, overlaps)) =
1314	try_join(
1315	try_join3(standing, self.landing_state(&pull.id), async { Ok(comments) }),
1316	try_join(self.latest_checks(&pull.id), self.overlaps(&pull)),
1317	)
1318	.await?;
1319	// The rules of the branch it merges into, as they stack, and which
1320	// of them it does not meet yet, for whoever is looking.
1321	let (statuses, settings, gate) = try_join3(
1322	self.statuses(&repo.id, pull.head_commit.as_deref()),
1323	self.settings_on(&repo, &pull),
1324	async {
1325	if pull.status.is_active() {
1326	self.merge_gate_seen(&repo, &pull, viewer.as_ref(), behind).await.map(Some)
1327	} else {
1328	Ok(None)
1329	}
1330	},
1331	)
1332	.await?;
1333	let code_owners = self.pull_code_owners(&pull, &comments, &settings).await?;
1334	// The work items it closes, as its title and description say now;
1335	// none where the repository does not close them that way.
1336	if pull.status.is_active() {
1337	let named = self.items_to_close(&pull, &repo, &settings, false).await?;
1338	pull.closes = named.close_numbers();
1339	pull.parts = named.part_numbers();
1340	}
1341	let required = if settings.required_checks.is_empty() {
1342	Vec::new()
1343	} else {
1344	let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses.clone()).await?;
1345	required_checks(&settings.required_checks, &checked)
1346	};
1347	Ok(Outcome::Ok(PullDetail {
1348	required_checks: required,
1349	rules: gate.map(\|gate\| rulesets::merge_rules(&gate.judged, &gate.requirements, pull.base_branch(&repo.default_branch) == repo.default_branch)),
1350	code_owners,
1351	comments,
1352	checks,
1353	overlaps,
1354	behind,
1355	lifecycle,
1356	landing,
1357	stalled,
1358	statuses,
1359	mergeable,
1360	conflicts,
1361	earlier_checks: self.earlier_checks(&pull.id).await?,
1362	reactions: Vec::new(),
1363	item,
1364	pull,
1365	}))
1366	}
1367
1368	/// The pull request, whatever its status, if its repository is not
1369	/// archived and `actor` opened it or may triage its pull requests.
1370	async fn managed_pull(
1371	&self,
1372	actor: &User,
1373	path: &RepoPath,
1374	number: u32,
1375	) -> Result<Outcome<Pull>> {
1376	let (repo, pull) = check!(self.pull_at(path, number, &Some(actor.clone())).await?);
1377	check!(writable(&repo));
1378	if !pull.is_owned_by(&actor.id) {
1379	check!(allowed(Some(actor), &repo, Capability::Triage));
1380	}
1381	Ok(Outcome::Ok(pull))
1382	}
1383
1384	/// The pull request, if it is still active and `actor` opened it or
1385	/// may triage the repository's pull requests.
1386	async fn manageable_pull(
1387	&self,
1388	actor: &User,
1389	path: &RepoPath,
1390	number: u32,
1391	) -> Result<Outcome<Pull>> {
1392	let pull = check!(self.managed_pull(actor, path, number).await?);
1393	if !pull.status.is_active() {
1394	return Ok(Outcome::fail(
1395	FailureCode::Conflict,
1396	format!("This pull request is already {}.", pull.status.as_str()),
1397	));
1398	}
1399	Ok(Outcome::Ok(pull))
1400	}
1401
1402	/// Brings a pull request up to date with the default branch without a
1403	/// sandbox, where the repos service can do that safely. Whoever could
1404	/// have pushed the merge themselves may ask: whoever opened it (or asked
1405	/// g1t for it), for a fork; anyone who may push, for a branch of the
1406	/// repository. When it needs a
1407	/// real merge, says so, naming the conflicting files if a probe found
1408	/// them, and pushes nothing.
1409	async fn catch_up_pull(&self, a: PullActionArgs) -> Result<Outcome<PullBranchUpdate>> {
1410	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &Some(a.actor.clone())).await?);
1411	check!(writable(&repo));
1412	if !pull.status.is_active() {
1413	return Ok(Outcome::fail(
1414	FailureCode::Conflict,
1415	format!("This pull request is already {}.", pull.status.as_str()),
1416	));
1417	}
1418	if pull.fork_repo_id.is_some() {
1419	if !pull.is_owned_by(&a.actor.id) {
1420	return Ok(Outcome::fail(
1421	FailureCode::Forbidden,
1422	"Only whoever opened this pull request, or asked g1t for it, can update it.",
1423	));
1424	}
1425	} else if !pull.from_head_repo() {
1426	check!(allowed(Some(&a.actor), &repo, Capability::Push));
1427	}
1428	// A fork's branch is updated by those who may push to the fork,
1429	// which repos decides.
1430	let updated: Outcome<PullBranchUpdate> = g1t_kit::call(
1431	&self.repos,
1432	"update_pull_branch",
1433	&UpdatePullBranchArgs {
1434	source_id: pull.source_id().to_owned(),
1435	branch: pull.branch.clone(),
1436	number: pull.number,
1437	actor: a.actor,
1438	target_branch: pull.base.clone(),
1439	target_id: pull.target_id(),
1440	method: a.update_method.unwrap_or_default(),
1441	expected_head: a.expected_head.clone().filter(\|head\| !head.trim().is_empty()),
1442	},
1443	)
1444	.await?;
1445	// A probe that found conflicts says more than "both changed it".
1446	if let Outcome::Ok(PullBranchUpdate::NeedsAgent { .. }) = &updated
1447	&& let Some(files) = self.conflicting_files(&pull).await?
1448	&& !files.is_empty()
1449	{
1450	return Ok(Outcome::Ok(PullBranchUpdate::NeedsAgent {
1451	reason: NeedsAgentReason::Conflicting,
1452	detail: "Merging it conflicts.".to_owned(),
1453	paths: files,
1454	}));
1455	}
1456	Ok(updated)
1457	}
1458
1459	async fn update_pull(&self, a: UpdatePullArgs) -> Result<Outcome<Pull>> {
1460	let pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1461	// The branch it merges into, its milestone and its labels first:
1462	// each can be refused, and then nothing else changes.
1463	if a.base.is_some() \|\| a.milestone.is_some() \|\| a.labels.is_some() {
1464	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1465	if let Some(base) = &a.base {
1466	check!(self.change_base(&a.actor, &repo, &pull, base).await?);
1467	}
1468	if let Some(number) = a.milestone {
1469	check!(items::set_milestone(self, &a.actor, &items::Labeled::Pull(repo.clone(), pull.clone()), number).await?);
1470	}
1471	if let Some(labels) = &a.labels {
1472	check!(items::relabel(self, &a.actor, &items::Labeled::Pull(repo.clone(), pull.clone()), labels).await?);
1473	}
1474	}
1475	let assignees = match a.assignees {
1476	Some(names) => Some(check!(self.valid_assignees(names).await?)),
1477	None => None,
1478	};
1479	// Teams, named `workspace/team`, apart from the people.
1480	let (team_names, a_reviewers) = match a.reviewers {
1481	Some(names) => {
1482	let (teams, people): (Vec<String>, Vec<String>) =
1483	names.into_iter().partition(\|name\| team_reviews::team_name(name).is_some());
1484	(Some(teams), Some(people))
1485	}
1486	None => (None, None),
1487	};
1488	let teams = match team_names {
1489	Some(names) => Some(check!(self.valid_team_reviewers(&a.actor, &a.repo, &pull, names).await?)),
1490	None => None,
1491	};
1492	let reviewers = match a_reviewers {
1493	Some(names) => {
1494	let mut reviewers = check!(self.valid_assignees(names).await?);
1495	// Nobody is asked to review their own, nor what they had g1t make.
1496	reviewers.retain(\|name\| *name != pull.owner().username);
1497	Some(reviewers)
1498	}
1499	None => None,
1500	};
1501	self.db
1502	.prepare(
1503	"UPDATE pulls
1504	SET assignees = COALESCE(?, assignees), updated_at = ?
1505	WHERE id = ?",
1506	)
1507	.bind(&[
1508	optional(&assignees.as_ref().map(serde_json::to_string).transpose()?),
1509	rfc3339(now_ms()).into(),
1510	pull.id.as_str().into(),
1511	])?
1512	.run()
1513	.await?;
1514	if let Some(assignees) = &assignees {
1515	self.note_changes(
1516	On::Pull(&pull.repo_id, pull.number),
1517	&a.actor,
1518	&pull.assignees,
1519	assignees,
1520	("assigned", "unassigned"),
1521	)
1522	.await?;
1523	}
1524	// Who was newly assigned or asked to review, and whose request was
1525	// withdrawn: the inbox tells them, and webhooks say so.
1526	let newly = \|after: &[String], before: &[String]\| -> Vec<String> {
1527	after.iter().filter(\|name\| !before.contains(name)).cloned().collect()
1528	};
1529	if let Some(assignees) = &assignees {
1530	let added = newly(assignees, &pull.assignees);
1531	if !added.is_empty() {
1532	self.publish(
1533	"pull.assigned",
1534	&pull.repo_id,
1535	&a.actor,
1536	PullEvent {
1537	assignees: Some(assignees.clone()),
1538	added: Some(added),
1539	..Self::pull_event(&pull)
1540	},
1541	)
1542	.await?;
1543	}
1544	}
1545	if reviewers.is_some() \|\| teams.is_some() {
1546	let people = reviewers.unwrap_or_else(\|\| pull.reviewers.clone());
1547	let teams = teams.unwrap_or_else(\|\| pull.team_reviewers.clone());
1548	self.set_reviewers(&pull, people, teams, Some(&a.actor), false).await?;
1549	}
1550	Ok(match self.pull(&pull.repo_id, pull.number).await? {
1551	Some(pull) => Outcome::Ok(pull),
1552	None => no_pull(),
1553	})
1554	}
1555
1556	/// Points an open pull request at another branch to merge into. Needs
1557	/// the Write role. What it would merge, whether it is behind, its
1558	/// mergeability and its checks are all worked out against the new base.
1559	async fn change_base(&self, actor: &User, repo: &Repo, pull: &Pull, base: &str) -> Result<Outcome<()>> {
1560	check!(allowed(Some(actor), repo, Capability::Push));
1561	let base = base.trim();
1562	if base.is_empty() {
1563	return Ok(Outcome::fail(FailureCode::Invalid, "Name the branch it should merge into."));
1564	}
1565	let before = pull.base_branch(&repo.default_branch).to_owned();
1566	if base == before {
1567	return Ok(Outcome::Ok(()));
1568	}
1569	if !pull.from_other_repo() && pull.branch.as_deref() == Some(base) {
1570	return Ok(Outcome::fail(
1571	FailureCode::Invalid,
1572	format!("A pull request cannot merge {base} into itself. Choose another base."),
1573	));
1574	}
1575	let exists: Option<String> = g1t_kit::call(
1576	&self.repos,
1577	"head",
1578	&HeadArgs { repo_id: repo.id.clone(), branch: base.to_owned() },
1579	)
1580	.await?;
1581	if exists.is_none() {
1582	return Ok(Outcome::fail(FailureCode::NotFound, format!("There is no branch named {base} to merge into.")));
1583	}
1584	// In the merge queue it was headed for the default branch; it
1585	// leaves the queue for another base.
1586	let left_queue = self
1587	.leave(&pull.repo_id, pull, QueueState::Removed, Some("Its base branch changed."))
1588	.await?;
1589	// Its tested commit is being published to the default branch: a new
1590	// base would describe a merge that is not the one taking place.
1591	if !left_queue && self.publishing(&pull.id).await? {
1592	return Ok(Outcome::fail(FailureCode::Conflict, "This pull request is being published. Wait for landing to finish."));
1593	}
1594	let stored = stored_base(base, repo);
1595	self.db
1596	.prepare(
1597	"UPDATE pulls
1598	SET base_branch = ?, updated_at = ?, land_requested = NULL, land_requested_at = NULL, behind = NULL,
1599	mergeable = NULL, mergeable_key = NULL, conflicts = NULL
1600	WHERE id = ?",
1601	)
1602	.bind(&[optional(&stored), rfc3339(now_ms()).into(), pull.id.as_str().into()])?
1603	.run()
1604	.await?;
1605	self.note(
1606	&pull.repo_id,
1607	pull.number,
1608	actor,
1609	&format!("changed the base branch from `{before}` to `{base}`"),
1610	)
1611	.await?;
1612	self.publish(
1613	"pull.base_changed",
1614	&pull.repo_id,
1615	actor,
1616	PullEvent { base: Some(base.to_owned()), ..Self::pull_event(pull) },
1617	)
1618	.await?;
1619	if left_queue {
1620	self.publish_as(
1621	"queue.changed",
1622	&pull.repo_id,
1623	None,
1624	g1t_contracts::events::QueueChanged { repo_id: pull.repo_id.clone() },
1625	)
1626	.await?;
1627	}
1628	// Its merge ref, and whether it merges cleanly, against the new base.
1629	if let Some(moved) = self.pull_by_id(&pull.id).await? {
1630	self.keep_pull_refs(&moved).await;
1631	if let Err(error) = self.assess_mergeability(&moved).await {
1632	worker::console_warn!("mergeability of {}: {error}", pull.id);
1633	}
1634	}
1635	Ok(Outcome::Ok(()))
1636	}
1637
1638	/// Marks a draft ready for review, or updates the description of one
1639	/// that already is.
1640	async fn ready_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1641	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1642	let summary = Some(a.summary.trim().to_owned()).filter(\|summary\| !summary.is_empty());
1643	let now = rfc3339(now_ms());
1644	// The work items it closes and is part of, as its description says
1645	// now (closing.rs).
1646	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1647	let own = self.pull_item_id(&pull.id).await?;
1648	let text = format!("{}\n{}", pull.title, summary.as_deref().or(pull.body.as_deref()).unwrap_or_default());
1649	let named = items::named(self, &repo, own.as_deref(), &text).await?;
1650	self.db
1651	.prepare(
1652	"UPDATE pulls SET status = 'open', body = COALESCE(?, body), updated_at = ?, closes_items = ?, parts = ?
1653	WHERE id = ?",
1654	)
1655	.bind(&[
1656	optional(&summary),
1657	now.as_str().into(),
1658	serde_json::to_string(&named.close_ids())?.into(),
1659	serde_json::to_string(&named.part_ids())?.into(),
1660	pull.id.as_str().into(),
1661	])?
1662	.run()
1663	.await?;
1664	if pull.status == PullStatus::Draft {
1665	// The head as it is now: the push that came just before may not
1666	// have reached `head_commit` yet, and workflows run on it.
1667	let commit = self.live_head(&pull).await?.or_else(\|\| pull.head_commit.clone());
1668	self.publish(
1669	"pull.ready",
1670	&pull.repo_id,
1671	&a.actor,
1672	PullEvent {
1673	commit,
1674	..Self::pull_event(&pull)
1675	},
1676	)
1677	.await?;
1678	}
1679	if pull.status == PullStatus::Draft {
1680	self.note(
1681	&pull.repo_id,
1682	pull.number,
1683	&a.actor,
1684	"marked this ready for review",
1685	)
1686	.await?;
1687	}
1688	pull.status = PullStatus::Open;
1689	pull.body = summary.or(pull.body);
1690	pull.closes = named.close_numbers();
1691	pull.parts = named.part_numbers();
1692	pull.updated_at = now;
1693	// A draft's code owners are asked once it is ready.
1694	self.refresh_code_owners(&pull).await;
1695	if let Some(fresh) = self.pull(&pull.repo_id, pull.number).await? {
1696	pull.reviewers = fresh.reviewers;
1697	pull.team_reviewers = fresh.team_reviewers;
1698	}
1699	Ok(Outcome::Ok(pull))
1700	}
1701
1702	async fn close_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1703	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1704	let now = rfc3339(now_ms());
1705	let closed = self.db
1706	// What it was closed as, so that reopening brings that back.
1707	.prepare("UPDATE pulls SET status = 'closed', closed_from = ?, updated_at = ?, auto_merge = NULL WHERE id = ? AND status <> 'merged'
1708	AND NOT EXISTS (SELECT 1 FROM queue_entries q WHERE q.pull_id = pulls.id AND q.state = 'passed' AND q.landing_commit IS NOT NULL)
1709	RETURNING id AS value")
1710	.bind(&[pull.status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1711	.first::<rows::ValueRow>(None)
1712	.await?;
1713	if closed.is_none() {
1714	return Ok(Outcome::fail(FailureCode::Conflict, "This pull request is being published or has already merged."));
1715	}
1716	self.publish(
1717	"pull.closed",
1718	&pull.repo_id,
1719	&a.actor,
1720	Self::pull_event(&pull),
1721	)
1722	.await?;
1723	self.note(
1724	&pull.repo_id,
1725	pull.number,
1726	&a.actor,
1727	"closed this",
1728	)
1729	.await?;
1730	// A closed pull request leaves the merge queue.
1731	if self
1732	.leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was closed."))
1733	.await?
1734	{
1735	self.publish_as(
1736	"queue.changed",
1737	&pull.repo_id,
1738	None,
1739	g1t_contracts::events::QueueChanged {
1740	repo_id: pull.repo_id.clone(),
1741	},
1742	)
1743	.await?;
1744	}
1745	pull.status = PullStatus::Closed;
1746	pull.auto_merge = None;
1747	pull.updated_at = now;
1748	Ok(Outcome::Ok(pull))
1749	}
1750
1751	/// Opens a closed pull request again: as the draft it was, if it was
1752	/// closed as one, else ready for review. A merged one stays merged.
1753	async fn reopen_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1754	let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
1755	if let Some(refusal) = reopen_refusal(pull.status) {
1756	return Ok(Outcome::fail(FailureCode::Conflict, refusal));
1757	}
1758	// The head as it is now, which workflows run on. A branch of the
1759	// repository that was deleted since leaves nothing to reopen; a
1760	// fork removed after the close is made again (repos, `pull.reopened`).
1761	let head = self.live_head(&pull).await?;
1762	if head.is_none() && pull.fork_repo_id.is_none() {
1763	return Ok(Outcome::fail(
1764	FailureCode::Conflict,
1765	format!(
1766	"The branch {} no longer exists. Push it again to reopen this pull request.",
1767	head_repo::source_label(pull.head_repo.as_ref(), pull.branch.as_deref().unwrap_or_default())
1768	),
1769	));
1770	}
1771	let closed_from: Option<ClosedFrom> = self
1772	.db
1773	.prepare("SELECT closed_from FROM pulls WHERE id = ?")
1774	.bind(&[pull.id.as_str().into()])?
1775	.first(None)
1776	.await?;
1777	let status = reopened_status(closed_from.and_then(\|row\| row.closed_from).as_deref());
1778	let now = rfc3339(now_ms());
1779	self.db
1780	.prepare("UPDATE pulls SET status = ?, closed_from = NULL, superseded_by = NULL, updated_at = ? WHERE id = ?")
1781	.bind(&[status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1782	.run()
1783	.await?;
1784	self.publish(
1785	"pull.reopened",
1786	&pull.repo_id,
1787	&a.actor,
1788	PullEvent {
1789	commit: head.or_else(\|\| pull.head_commit.clone()),
1790	..Self::pull_event(&pull)
1791	},
1792	)
1793	.await?;
1794	self.note(
1795	&pull.repo_id,
1796	pull.number,
1797	&a.actor,
1798	"reopened this",
1799	)
1800	.await?;
1801	pull.status = status;
1802	pull.superseded_by = None;
1803	pull.updated_at = now;
1804	// Its refs in the repository, and whether it still merges cleanly,
1805	// now that it is open again.
1806	self.keep_pull_refs(&pull).await;
1807	if let Err(error) = self.assess_mergeability(&pull).await {
1808	worker::console_warn!("mergeability of {}: {error}", pull.id);
1809	}
1810	Ok(Outcome::Ok(pull))
1811	}
1812
1813	/// Turns a pull request that is ready for review back into a draft: it
1814	/// cannot be merged until it is marked ready again, and it leaves the
1815	/// merge queue and any merge that was waiting for it to catch up.
1816	async fn convert_pull_to_draft(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1817	let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
1818	if let Some(refusal) = draft_refusal(pull.status) {
1819	return Ok(Outcome::fail(FailureCode::Conflict, refusal));
1820	}
1821	let now = rfc3339(now_ms());
1822	let drafted = self.db
1823	.prepare(
1824	"UPDATE pulls SET status = 'draft', land_requested = NULL, land_requested_at = NULL, updated_at = ?
1825	WHERE id = ? AND status = 'open'
1826	AND NOT EXISTS (SELECT 1 FROM queue_entries q WHERE q.pull_id = pulls.id AND q.state = 'passed' AND q.landing_commit IS NOT NULL)
1827	RETURNING id AS value",
1828	)
1829	.bind(&[now.as_str().into(), pull.id.as_str().into()])?
1830	.first::<rows::ValueRow>(None)
1831	.await?;
1832	if drafted.is_none() {
1833	return Ok(Outcome::fail(FailureCode::Conflict, "This pull request changed or is already being published."));
1834	}
1835	self.publish(
1836	"pull.converted_to_draft",
1837	&pull.repo_id,
1838	&a.actor,
1839	Self::pull_event(&pull),
1840	)
1841	.await?;
1842	self.note(
1843	&pull.repo_id,
1844	pull.number,
1845	&a.actor,
1846	"marked this as a draft",
1847	)
1848	.await?;
1849	if self
1850	.leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was marked as a draft."))
1851	.await?
1852	{
1853	self.publish_as(
1854	"queue.changed",
1855	&pull.repo_id,
1856	None,
1857	g1t_contracts::events::QueueChanged {
1858	repo_id: pull.repo_id.clone(),
1859	},
1860	)
1861	.await?;
1862	}
1863	pull.status = PullStatus::Draft;
1864	pull.updated_at = now;
1865	Ok(Outcome::Ok(pull))
1866	}
1867
1868	/// Lands the pull request on the branch it merges into, by the method
1869	/// asked for or the repository's default (merging.rs). Unless told to
1870	/// keep them open, a merge into the default branch lands the work items
1871	/// it closes in this repository: its own and those it names
1872	/// (closing.rs), each closing once nothing else is left to land, and
1873	/// the others still in progress here for its own close as superseded.
1874	async fn merge_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1875	let viewer = Some(a.actor.clone());
1876	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
1877	check!(writable(&repo));
1878	match pull.status {
1879	PullStatus::Open => {}
1880	PullStatus::Draft => {
1881	return Ok(Outcome::fail(
1882	FailureCode::Conflict,
1883	"This pull request is still a draft. Mark it ready for review first.",
1884	));
1885	}
1886	status => {
1887	return Ok(Outcome::fail(
1888	FailureCode::Conflict,
1889	format!("This pull request is already {}.", status.as_str()),
1890	));
1891	}
1892	}
1893	// Whoever looked at one head merges that one.
1894	if let Some(expected) = a.expected_head.as_deref().map(str::trim).filter(\|head\| !head.is_empty())
1895	&& pull.head_commit.as_deref() != Some(expected)
1896	{
1897	return Ok(Outcome::fail(
1898	FailureCode::Conflict,
1899	"This pull request has new commits since you looked. Review them, then merge again.",
1900	));
1901	}
1902	let base = pull.base_branch(&repo.default_branch).to_owned();
1903	// How it merges: as asked, or the repository's default, where the
1904	// repository and the rules for its base allow it.
1905	let (choosing, requirements) = self.settings_and_requirements(&repo, &pull).await?;
1906	let method = match merging::choose_method(a.merge_method, &choosing, &requirements, &base) {
1907	Ok(method) => method,
1908	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1909	};
1910	// The rules of the branch it merges into, as they stack: the one
1911	// gate for the merge button, the API, MCP, auto-merge and the queue,
1912	// for a person's pull request and an agent's alike. A bypass counts
1913	// when the merger asks for it, or for g1t when a ruleset lists it.
1914	let gate = self.merge_gate_by(&repo, &pull, Some(&a.actor), a.ignore_checks, true, method).await?;
1915	let bypassable = gate.bypassable();
1916	let gate = if a.bypass_rules \|\| a.actor.is_system() { gate } else { gate.without_bypass() };
1917	let settings = rulesets::overlay(self.settings(&repo.id).await?, &gate.requirements, base == repo.default_branch);
1918	if pull.check_status == Some(CheckStatus::Failed) && !(a.ignore_checks && settings.allow_ignoring_checks) {
1919	return Ok(Outcome::fail(
1920	FailureCode::Conflict,
1921	"It failed in the merge queue; push a fix to try again.",
1922	));
1923	}
1924	if let Some(refusal) = gate.refusal() {
1925	if access::can(Some(&a.actor), &repo, Capability::Merge) {
1926	self.record_merge_evaluations(&repo, &pull, &gate).await;
1927	}
1928	let offer = if bypassable && !a.bypass_rules {
1929	" You may bypass these rules: merge again and ask to bypass them (bypass_rules)."
1930	} else {
1931	""
1932	};
1933	return Ok(Outcome::fail(FailureCode::Conflict, format!("{refusal}{offer}")));
1934	}
1935	// Known ahead of time to conflict: neither a merge nor the queue
1936	// would get through, so say what has to be resolved now.
1937	if let Some(files) = self.conflicting_files(&pull).await? {
1938	let named = if files.is_empty() {
1939	String::new()
1940	} else {
1941	format!(" in {}", files.join(", "))
1942	};
1943	return Ok(Outcome::fail(
1944	FailureCode::Conflict,
1945	format!(
1946	"This branch has conflicts with {base}{named} that must be resolved first. Merge {base} into it, fix them and push."
1947	),
1948	));
1949	}
1950
1951	// A repository that merges through a queue: it joins the queue, and
1952	// lands once its state together with everything ahead has passed.
1953	// Its state is built by the method asked for, else the merge queue
1954	// rule's, and lands exactly as tested (queue.rs).
1955	if settings.merge_queue {
1956	if !a.actor.verified {
1957	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1958	}
1959	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
1960	let queued = match a.merge_method {
1961	Some(_) => method,
1962	None => {
1963	let rule = self.queue_rule(&repo.id).await?;
1964	match merging::choose_method(Some(rule.merge_method), &choosing, &requirements, &base) {
1965	Ok(method) => method,
1966	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1967	}
1968	}
1969	};
1970	self.record_merge_evaluations(&repo, &pull, &gate).await;
1971	// Landing honours the same bypass request against the rules as
1972	// they are then (queue.rs `settle`).
1973	let message = (a.commit_title.as_deref(), a.commit_message.as_deref());
1974	return self.enqueue(&repo, &pull, &a.actor, a.keep_item_open, a.bypass_rules, queued, message).await;
1975	}
1976
1977	// The commit it makes, for a merge or a squash: what the merger
1978	// wrote, over the repository's default.
1979	let message = match self
1980	.merge_message_for(&repo, &pull, &settings, method, a.commit_title.as_deref(), a.commit_message.as_deref())
1981	.await?
1982	{
1983	Ok(message) => message,
1984	Err(problem) => return Ok(Outcome::fail(FailureCode::Invalid, problem)),
1985	};
1986
1987	// The branch it merges into has moved under it. A merge, squash or
1988	// rebase makes new commits on its head anyway; a fast-forward needs
1989	// it brought up to date first, and lands when that is done. Unless
1990	// the repository insists on that being dealt with first.
1991	let behind = self.is_behind(&pull).await?;
1992	match merging::when_behind(method, behind, settings.require_up_to_date, &base) {
1993	merging::WhenBehind::Land => {}
1994	merging::WhenBehind::Refuse(refusal) => return Ok(Outcome::fail(FailureCode::Conflict, refusal)),
1995	merging::WhenBehind::CatchUpFirst => {
1996	if !a.actor.verified {
1997	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1998	}
1999	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
2000	self.record_merge_evaluations(&repo, &pull, &gate).await;
2001	// Brought up to date here, without a sandbox: the merge is
2002	// recorded first, and lands once the push that catches it up
2003	// arrives and the required checks pass on that commit
2004	// (lifecycle.rs `land_if_requested`, `land_when_checked`).
2005	// Merging past failing checks, where that is allowed, carries
2006	// over: it then lands without waiting for them.
2007	self.request_landing(
2008	&pull,
2009	&a.actor,
2010	a.keep_item_open,
2011	a.delete_branch,
2012	a.ignore_checks && settings.allow_ignoring_checks,
2013	)
2014	.await?;
2015	// Bounded: a repository that does not answer gets a clear
2016	// failure back to whoever clicked, not a request that hangs
2017	// until the site gives up on it.
2018	let updated: Option<Outcome<PullBranchUpdate>> = g1t_kit::call_within(
2019	&self.repos,
2020	"update_pull_branch",
2021	&UpdatePullBranchArgs {
2022	source_id: pull.source_id().to_owned(),
2023	branch: pull.branch.clone(),
2024	number: pull.number,
2025	actor: a.actor.clone(),
2026	target_branch: pull.base.clone(),
2027	target_id: pull.target_id(),
2028	// Linear history takes no merge commit on the way.
2029	method: if requirements.linear_history { UpdateMethod::Rebase } else { UpdateMethod::Merge },
2030	expected_head: pull.head_commit.clone(),
2031	},
2032	lifecycle::REPOS_WRITE_WITHIN,
2033	)
2034	.await?;
2035	// A push that still arrives later only brings it up to date.
2036	let updated = updated.unwrap_or_else(lifecycle::repository_unreachable);
2037	match updated {
2038	Outcome::Ok(PullBranchUpdate::Updated { .. }) => return Ok(Outcome::Ok(pull)),
2039	// Moved again in between, or already caught up: land it now.
2040	Outcome::Ok(PullBranchUpdate::UpToDate { .. }) => self.forget_landing(&pull.id).await?,
2041	Outcome::Ok(PullBranchUpdate::NeedsAgent { paths, .. }) => {
2042	self.forget_landing(&pull.id).await?;
2043	let named = if paths.is_empty() { String::new() } else { format!(" ({})", paths.join(", ")) };
2044	return Ok(Outcome::fail(
2045	FailureCode::Conflict,
2046	format!(
2047	"{base} has moved since this pull request was made, and both changed the same files{named}. Merge {base} into it, resolve any conflicts, push, and merge again."
2048	),
2049	));
2050	}
2051	// Only those who can push where its branch is can catch
2052	// it up; anyone may merge it another way.
2053	Outcome::Fail(failure) if failure.code == FailureCode::Forbidden => {
2054	self.forget_landing(&pull.id).await?;
2055	let source = pull.source_path(&a.repo);
2056	return Ok(Outcome::fail(
2057	FailureCode::Conflict,
2058	merging::cannot_fast_forward(&base, &format!("{}/{}", source.namespace, source.name)),
2059	));
2060	}
2061	Outcome::Fail(failure) => {
2062	self.forget_landing(&pull.id).await?;
2063	return Ok(Outcome::Fail(failure));
2064	}
2065	}
2066	}
2067	}
2068
2069	// Whether the actor may write to the repository is decided by repos.
2070	let head = pull.head_commit.clone();
2071	// Bounded like the catch-up above. Landing the same checked commit
2072	// again is recognised as done (repos `land`), so trying again after
2073	// a land that finished late records it.
2074	let landed: Option<Outcome<Landed>> = g1t_kit::call_within(
2075	&self.repos,
2076	"land",
2077	&LandArgs {
2078	// A pull request from a branch lands from the repository
2079	// itself, or from the fork the branch is in.
2080	source_id: pull.source_id().to_owned(),
2081	branch: pull.branch.clone(),
2082	actor: a.actor.clone(),
2083	target_branch: Some(base.clone()),
2084	expected_commit: head.clone(),
2085	target_id: pull.target_id(),
2086	method: Some(method),
2087	message,
2088	author: merging::squash_author(&pull, method),
2089	sign_off: settings.require_web_commit_sign_off,
2090	pull: Some(pull.number),
2091	},
2092	lifecycle::REPOS_WRITE_WITHIN,
2093	)
2094	.await?;
2095	let landed = check!(landed.unwrap_or_else(lifecycle::repository_unreachable));
2096	self.record_merge_evaluations(&repo, &pull, &gate).await;
2097	let merged = self.record_merge(&repo, pull, &a.actor, a.keep_item_open, landed, None).await?;
2098	// Its branch, where the merger or the repository says it goes.
2099	if merged.status == PullStatus::Merged {
2100	self.delete_branch_after_merge(&repo, &merged, &a.actor, a.delete_branch, head).await?;
2101	}
2102	Ok(Outcome::Ok(self.pull_by_id(&merged.id).await?.unwrap_or(merged)))
2103	}
2104
2105	/// Records a pull request as merged once the default branch holds it:
2106	/// lands the work items it closes (closing.rs), links those it is part
2107	/// of, supersedes the others for its own work item in this repository,
2108	/// and says so. `pushed` names the branch when a push brought its head
2109	/// in, rather than a merge from the site or the queue (pushed.rs).
2110	pub(crate) async fn record_merge(
2111	&self,
2112	repo: &Repo,
2113	mut pull: Pull,
2114	actor: &User,
2115	keep_item_open: bool,
2116	landed: Landed,
2117	pushed: Option<&str>,
2118	) -> Result<Pull> {
2119	// Only a merge into the default branch lands work: into another
2120	// branch, it has not landed yet. Its own work item and those it
2121	// names land, where the repository closes them that way.
2122	let settings = self.settings(&repo.id).await?;
2123	let named = self.items_to_close(&pull, repo, &settings, keep_item_open).await?;
2124	let own = self.pull_item_id(&pull.id).await?;
2125	// A fast-forward leaves the base at what was landed, which is its
2126	// head; the other methods leave its head as it was.
2127	let head = if landed.method == g1t_contracts::rules::MergeMethod::FastForward {
2128	Some(landed.commit.clone())
2129	} else {
2130	pull.head_commit.clone().or_else(\|\| Some(landed.commit.clone()))
2131	};
2132	let now = rfc3339(now_ms());
2133	let mut statements = vec![self
2134	.db
2135	.prepare(
2136	"UPDATE pulls
2137	SET status = 'merged', head_commit = ?, merge_base = ?, merged_by = ?,
2138	merged_at = ?, updated_at = ?, merge_method = ?, merge_commit = ?,
2139	auto_merge = NULL, closes_items = ?, parts = ?
2140	WHERE id = ? AND status <> 'merged' RETURNING id AS value",
2141	)
2142	.bind(&[
2143	optional(&head),
2144	optional(&landed.previous),
2145	actor.username.as_str().into(),
2146	now.as_str().into(),
2147	now.as_str().into(),
2148	landed.method.as_str().into(),
2149	landed.commit.as_str().into(),
2150	serde_json::to_string(&named.close_ids())?.into(),
2151	serde_json::to_string(&named.part_ids())?.into(),
2152	pull.id.as_str().into(),
2153	])?];
2154	// The others for its own work item in this repository were other
2155	// ways to do the same thing.
2156	if let Some(own) = own.as_deref().filter(\|own\| named.close_ids().iter().any(\|id\| id == own)) {
2157	statements.push(
2158	self.db
2159	.prepare(
2160	"UPDATE pulls SET status = 'closed', superseded_by = ?, updated_at = ?
2161	WHERE item_id = ? AND repo_id = ? AND id != ? AND status IN ('draft', 'open')",
2162	)
2163	.bind(&[pull.number.into(), now.as_str().into(), own.into(), repo.id.as_str().into(), pull.id.as_str().into()])?,
2164	);
2165	}
2166	let recorded = self.db.batch(statements).await?;
2167	if recorded.first().map(\|result\| result.results::<rows::ValueRow>()).transpose()?.is_none_or(\|rows\| rows.is_empty()) {
2168	return Ok(self.pull_by_id(&pull.id).await?.unwrap_or(pull));
2169	}
2170
2171	self.publish(
2172	"pull.merged",
2173	&repo.id,
2174	actor,
2175	PullEvent {
2176	commit: Some(landed.commit.clone()),
2177	..Self::pull_event(&pull)
2178	},
2179	)
2180	.await?;
2181	let who = (actor.id.as_str(), actor.username.as_str());
2182	let note = match pushed {
2183	Some(branch) => format!("merged this by pushing {} to {branch}", landed.commit),
2184	None => merging::merged_note(landed.method).to_owned(),
2185	};
2186	self.note(&repo.id, pull.number, who, &note).await?;
2187	let landing = items::Landing { pull: Some(pull.number), commit: None, pushed_by: None };
2188	items::land(self, repo, &landing, actor, &named.close_ids(), &named.part_ids(), None, false).await?;
2189	let through_queue = self.landed_through_queue(&pull.id).await?;
2190	self.keep_decision_quietly(
2191	&repo.id,
2192	decisions::NewDecision {
2193	item: own.clone(),
2194	pull: Some(pull.number),
2195	session_id: None,
2196	actor: (!decisions::is_g1t(actor)).then_some(actor),
2197	kind: g1t_contracts::decisions::DecisionKind::Merge,
2198	confidence: decisions::merge_confidence(through_queue, actor),
2199	summary: match pushed {
2200	_ if through_queue => format!("#{} landed through the merge queue.", pull.number),
2201	Some(branch) => format!("{} merged #{} by pushing to {branch}.", actor.username, pull.number),
2202	None => format!("{} merged #{}.", actor.username, pull.number),
2203	},
2204	evidence: serde_json::json!({
2205	"commit": landed.commit,
2206	"previous": landed.previous,
2207	"method": landed.method,
2208	"merge_queue": through_queue,
2209	"pushed": pushed.is_some(),
2210	}),
2211	},
2212	)
2213	.await;
2214	pull.status = PullStatus::Merged;
2215	pull.head_commit = head;
2216	pull.merge_method = Some(landed.method);
2217	pull.merge_commit = Some(landed.commit.clone());
2218	pull.auto_merge = None;
2219	pull.closes = named.close_numbers();
2220	pull.parts = named.part_numbers();
2221	pull.merge_base = landed.previous;
2222	pull.merged_by = Some(actor.username.clone());
2223	pull.merged_at = Some(now.clone());
2224	pull.updated_at = now;
2225	Ok(pull)
2226	}
2227
2228	async fn list_active_pulls(&self, a: ViewerArgs) -> Result<Vec<ActivePull>> {
2229	let Some(viewer) = a.viewer else {
2230	return Ok(Vec::new());
2231	};
2232	// The pull requests and their work items, in one round trip: their
2233	// own, and those g1t made for them (Pull::owner).
2234	let author = [JsValue::from(viewer.id.as_str())];
2235	let found = self
2236	.timing
2237	.db(
2238	2,
2239	self.db.batch(vec![
2240	self.db
2241	.prepare(format!(
2242	"SELECT {PULL_COLUMNS} FROM pulls
2243	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open')
2244	ORDER BY updated_at DESC LIMIT 50"
2245	))
2246	.bind(&author)?,
2247	self.db
2248	.prepare(items::select(
2249	"work_items.id IN (
2250	SELECT item_id FROM pulls
2251	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open') AND item_id IS NOT NULL
2252	ORDER BY updated_at DESC LIMIT 50)",
2253	))
2254	.bind(&author)?,
2255	]),
2256	)
2257	.await?;
2258	let (Some(found), Some(item_rows)) = (found.first(), found.get(1)) else {
2259	return Ok(Vec::new());
2260	};
2261	let snapshots = found.results::<Snapshot>()?;
2262	let item_ids: Vec<Option<String>> = found.results::<PullItemRow>()?.into_iter().map(\|row\| row.item_id).collect();
2263	let pulls: Vec<Pull> = found.results::<PullRow>()?.into_iter().map(Pull::from).collect();
2264	let items: Vec<Held> = items::held_rows(item_rows)?;
2265	let items = &items;
2266	// Where each stands: the remembered assessment when there is one,
2267	// and worked out otherwise.
2268	try_join_all(pulls.into_iter().zip(snapshots).zip(item_ids).map(\|((pull, snapshot), item_id)\| async move {
2269	let item = item_id.and_then(\|id\| items.iter().find(\|held\| held.item.id == id).map(\|held\| held.item.clone()));
2270	// Every open pull request has a lifecycle (lifecycle.rs), kept
2271	// fresh by what moves it (`advance`, on each event that can);
2272	// anyone's draft but g1t's has none.
2273	let managed = lifecycle::made_by_g1t(&pull) && snapshot.managed != 0;
2274	let lifecycle = if pull.status == PullStatus::Draft && !managed {
2275	None
2276	} else if let (Some(stage), Some(detail)) = (snapshot.stage, snapshot.stage_detail) {
2277	Some(Lifecycle {
2278	stage,
2279	detail,
2280	revisions: snapshot.revisions,
2281	})
2282	} else {
2283	// Never worked out yet: once, and remembered.
2284	let behind = self.behind_now(&pull).await?;
2285	self.assess(&pull, &item, behind)
2286	.await?
2287	.map(\|(lifecycle, _)\| lifecycle)
2288	};
2289	Ok::<_, worker::Error>(ActivePull {
2290	pull,
2291	item,
2292	lifecycle,
2293	})
2294	}))
2295	.await
2296	}
2297
2298	// --- Sessions ----------------------------------------------------------
2299
2300	async fn append_session(&self, a: AppendSessionArgs) -> Result<Outcome<Appended>> {
2301	if a.entries.is_empty() {
2302	return Ok(Outcome::Ok(Appended { count: 0 }));
2303	}
2304	if a.entries.len() > MAX_ENTRY_BATCH {
2305	return Ok(Outcome::fail(
2306	FailureCode::Invalid,
2307	format!("Send at most {MAX_ENTRY_BATCH} entries at a time."),
2308	));
2309	}
2310	let viewer = Some(a.actor.clone());
2311	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
2312	if !pull.is_owned_by(&a.actor.id) {
2313	return Ok(Outcome::fail(
2314	FailureCode::Forbidden,
2315	"Only whoever opened a pull request, or asked g1t for it, can record its session.",
2316	));
2317	}
2318
2319	let now = rfc3339(now_ms());
2320	let count = a.entries.len() as u32;
2321	let mut statements = Vec::with_capacity(a.entries.len() + 1);
2322	for entry in a.entries {
2323	let kind = serde_json::to_value(entry.kind)?;
2324	let text = coordination::shown::clean(&entry.text.chars().take(MAX_ENTRY_CHARS).collect::<String>());
2325	// Each insert takes the next sequence number itself, so two
2326	// writers appending at once cannot collide.
2327	statements.push(
2328	self.db
2329	.prepare(
2330	"INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
2331	SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
2332	FROM session_entries WHERE pull_id = ?",
2333	)
2334	.bind(&[
2335	pull.id.as_str().into(),
2336	kind.as_str().unwrap_or("note").into(),
2337	text.into(),
2338	optional(&entry.tool),
2339	optional(&entry.commit.or_else(\|\| pull.head_commit.clone())),
2340	now.as_str().into(),
2341	pull.id.as_str().into(),
2342	])?,
2343	);
2344	}
2345	statements.push(
2346	self.db
2347	.prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
2348	.bind(&[now.as_str().into(), pull.id.as_str().into()])?,
2349	);
2350	self.db.batch(statements).await?;
2351	self.publish(
2352	"session.appended",
2353	&pull.repo_id,
2354	&a.actor,
2355	SessionAppended {
2356	pull_id: pull.id.clone(),
2357	repo_id: pull.repo_id.clone(),
2358	number: pull.number,
2359	count,
2360	},
2361	)
2362	.await?;
2363	Ok(Outcome::Ok(Appended { count }))
2364	}
2365
2366	async fn read_session(&self, a: ViewArgs) -> Result<Outcome<Vec<SessionEntry>>> {
2367	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
2368	let rows = self
2369	.db
2370	.prepare(
2371	"SELECT seq, kind, text, tool, \"commit\", at FROM session_entries
2372	WHERE pull_id = ? AND seq > ? ORDER BY seq LIMIT ?",
2373	)
2374	.bind(&[pull.id.into(), a.after_seq.into(), SESSION_PAGE.into()])?
2375	.all()
2376	.await?
2377	.results::<SessionRow>()?;
2378	Ok(Outcome::Ok(
2379	rows.into_iter().map(SessionEntry::from).collect(),
2380	))
2381	}
2382
2383	/// A push moves the head of the pull request it concerns: the one whose
2384	/// fork was pushed to, or the one opened from the branch that moved.
2385	async fn on_event(&self, event: &Event) -> Result<()> {
2386	// A new repository: where it is, for its workspace's work items and
2387	// numbers; a workspace's first starts it with the default labels.
2388	if event.kind == "repo.created" {
2389	return items::repo_created(self, &event.data).await;
2390	}
2391	if event.kind == "repo.visibility_changed" \|\| event.kind == "repo.updated" {
2392	return items::repo_visibility(self, &event.data).await;
2393	}
2394	// Pull requests into the branch that became the default merge into
2395	// the default branch, which is stored as none.
2396	if event.kind == "repo.default_branch_changed"
2397	&& let (Some(repo_id), Some(to)) = (event.repo_id.as_deref(), event.data["to"].as_str())
2398	{
2399	self.db
2400	.prepare(
2401	"UPDATE pulls SET base_branch = NULL
2402	WHERE repo_id = ? AND base_branch = ? AND status IN ('draft', 'open')",
2403	)
2404	.bind(&[repo_id.into(), to.into()])?
2405	.run()
2406	.await?;
2407	return Ok(());
2408	}
2409	if event.kind != "git.push" {
2410	return Ok(());
2411	}
2412	let (Some(repo_id), Some(after), Some(git_ref)) = (
2413	event.repo_id.as_deref(),
2414	event.data["after"].as_str(),
2415	event.data["ref"].as_str(),
2416	) else {
2417	return Ok(());
2418	};
2419	let now = rfc3339(now_ms());
2420	// Who moved it, for rules about the most recent push.
2421	let pusher: JsValue = event.actor.as_deref().map_or(JsValue::NULL, Into::into);
2422	// The head moved, so whatever the checks said no longer applies, and
2423	// whatever step g1t was waiting on has been taken.
2424	let moved = "UPDATE pulls
2425	SET head_commit = ?, updated_at = ?, head_pushed_by = ?, head_pushed_at = ?, check_status = NULL, check_run_id = NULL,
2426	working_on = NULL, working_until = NULL, stalled = NULL";
2427	let active = "status IN ('draft', 'open') AND head_commit IS NOT ?";
2428	let returning = "RETURNING id, repo_id, number, status, author_id, author_name, requested_by_id, requested_by_name";
2429	let mut pulls: Vec<MovedRow> = Vec::new();
2430	// A fork carries its pull request on its default branch.
2431	if event.data["defaultBranch"].as_bool() == Some(true) {
2432	pulls.extend(
2433	self.db
2434	.prepare(format!(
2435	"{moved} WHERE fork_repo_id = ? AND {active} {returning}"
2436	))
2437	.bind(&[
2438	after.into(),
2439	now.as_str().into(),
2440	pusher.clone(),
2441	now.as_str().into(),
2442	repo_id.into(),
2443	after.into(),
2444	])?
2445	.all()
2446	.await?
2447	.results::<MovedRow>()?,
2448	);
2449	}
2450	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2451	// A branch of the repository, and a branch of a fork pull
2452	// requests into other repositories were opened from.
2453	pulls.extend(
2454	self.db
2455	.prepare(format!(
2456	"{moved} WHERE {} AND {active} {returning}",
2457	head_repo::ON_BRANCH
2458	))
2459	.bind(&[
2460	after.into(),
2461	now.as_str().into(),
2462	pusher.clone(),
2463	now.as_str().into(),
2464	repo_id.into(),
2465	repo_id.into(),
2466	branch.into(),
2467	after.into(),
2468	])?
2469	.all()
2470	.await?
2471	.results::<MovedRow>()?,
2472	);
2473	}
2474	// What each now changes, so overlaps show while the work is under way.
2475	let mut moved_pulls = Vec::new();
2476	for moved in &pulls {
2477	if let Some(mut pull) = self.pull_by_id(&moved.id).await? {
2478	pull.files = self.refresh_files(&pull).await?;
2479	// Owners of files it now changes are asked too.
2480	self.refresh_code_owners(&pull).await;
2481	// A new overlap with another pull request is announced.
2482	self.announce_overlaps(&pull, event.actor.clone()).await;
2483	// Auto-merge someone else turned on stops (automerge.rs).
2484	self.auto_merge_after_push(&pull, event.actor.as_deref()).await?;
2485	moved_pulls.push(pull);
2486	}
2487	}
2488	// The harness sessions that pushed the branch, or that its new
2489	// commits name in trailers, are tied to its pull requests.
2490	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2491	let commits = event.data["commits"].as_u64().and_then(\|n\| u32::try_from(n).ok());
2492	self.link_push(repo_id, branch, after, commits, &moved_pulls).await;
2493	}
2494	// A merge that was waiting for this push to bring it up to date.
2495	for moved in &pulls {
2496	self.land_if_requested(&moved.id).await?;
2497	}
2498	// The pull requests into the branch whose heads it brought in are
2499	// merged, and the work items its commits name landed (pushed.rs).
2500	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2501	self.after_branch_push(event, repo_id, branch, after).await;
2502	}
2503	// Whether each still merges cleanly, and, when a default branch
2504	// moved, every open pull request into it.
2505	let moved_ids: Vec<String> = pulls.iter().map(\|pull\| pull.id.clone()).collect();
2506	let into = match event.data["defaultBranch"].as_bool() {
2507	Some(true) => mergeability::Moved::DefaultBranch,
2508	_ => match git_ref.strip_prefix("refs/heads/") {
2509	Some(branch) => mergeability::Moved::Branch(branch),
2510	None => mergeability::Moved::Nothing,
2511	},
2512	};
2513	self.after_push(repo_id, into, &moved_ids).await;
2514	// A draft is announced when it is marked ready instead.
2515	for pull in pulls
2516	.into_iter()
2517	.filter(\|pull\| pull.status == PullStatus::Open)
2518	{
2519	let item = moved_pulls.iter().find(\|moved\| moved.id == pull.id).and_then(\|moved\| moved.item);
2520	self.publish_as(
2521	"pull.updated",
2522	&pull.repo_id,
2523	event.actor.clone(),
2524	PullEvent {
2525	author: Some(g1t_contracts::credentials::Principal { id: pull.author_id, username: pull.author_name }),
2526	requested_by: pull
2527	.requested_by_id
2528	.zip(pull.requested_by_name)
2529	.map(\|(id, username)\| g1t_contracts::credentials::Principal { id, username }),
2530	pull_id: pull.id,
2531	repo_id: pull.repo_id.clone(),
2532	number: pull.number,
2533	item,
2534	commit: Some(after.to_owned()),
2535	..PullEvent::default()
2536	}
2537	.carrying(&event.data),
2538	)
2539	.await?;
2540	}
2541	Ok(())
2542	}
2543	}
2544
2545	/// An event's data that carries on what caused the event it follows from.
2546	trait Carrying: serde::Serialize + Sized {
2547	/// As JSON, marked as a workflow job's doing when `cause` was.
2548	fn carrying(self, cause: &serde_json::Value) -> serde_json::Value {
2549	g1t_contracts::events::carried(self, cause)
2550	}
2551	}
2552
2553	impl Carrying for PullEvent {}
2554
2555	fn service(env: &Env) -> Result<Work> {
2556	Ok(Work {
2557	db: env.d1("DB")?,
2558	identity: env.service("IDENTITY")?,
2559	repos: env.service("REPOS")?,
2560	events: env.service("EVENTS")?,
2561	actions: env.service("ACTIONS")?,
2562	timing: g1t_kit::d1::Timing::default(),
2563	prefetched: std::cell::RefCell::new(None),
2564	known_repos: std::cell::RefCell::new(std::collections::HashMap::new()),
2565	})
2566	}
2567
2568	#[event(fetch)]
2569	async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
2570	let Some(method) = rpc_method(&request) else {
2571	return Response::error("Not found", 404);
2572	};
2573	// A replica near the caller when it asks for one (crates/kit/src/d1.rs).
2574	let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
2575	let body: serde_json::Value = request.json().await?;
2576	let mut work = service(&env)?;
2577	work.db = db;
2578
2579	let answered = match method.as_str() {
2580	// Work items (items/) and moving issues into them (items/migration.rs).
2581	"open_work_item" => reply(&work.open_work_item(args(body)?).await?),
2582	"list_work_items" => reply(&work.list_work_items(args(body)?).await?),
2583	"get_work_item" => reply(&work.get_work_item(args(body)?).await?),
2584	"update_work_item" => reply(&work.update_work_item(args(body)?).await?),
2585	"close_work_item" => reply(&work.close_work_item(args(body)?).await?),
2586	"reopen_work_item" => reply(&work.reopen_work_item(args(body)?).await?),
2587	"link_work_item" => reply(&work.link_work_item(args(body)?).await?),
2588	"unlink_work_item" => reply(&work.unlink_work_item(args(body)?).await?),
2589	"resolve_ref" => reply(&work.resolve_ref(args(body)?).await?),
2590	"legacy_item_map" => reply(&work.legacy_item_map(args(body)?).await?),
2591	"migrate_work_items" => {
2592	let _: MigrateWorkItemsArgs = args(body)?;
2593	reply(&work.migrate_work_items().await?)
2594	}
2595	"list_labels" => reply(&work.list_labels(args(body)?).await?),
2596	"save_label" => reply(&work.save_label(args(body)?).await?),
2597	"delete_label" => reply(&work.delete_label(args(body)?).await?),
2598	"add_default_labels" => reply(&work.add_default_labels(args(body)?).await?),
2599	"set_labels" => reply(&work.set_labels(args(body)?).await?),
2600	"list_milestones" => reply(&work.list_milestones(args(body)?).await?),
2601	"get_milestone" => reply(&work.get_milestone(args(body)?).await?),
2602	"save_milestone" => reply(&work.save_milestone(args(body)?).await?),
2603	"delete_milestone" => reply(&work.delete_milestone(args(body)?).await?),
2604	"counts" => reply(&work.counts(args(body)?).await?),
2605	"add_comment" => reply(&work.add_comment(args(body)?).await?),
2606	"upload_attachment" => reply(&work.upload_attachment(&env, args(body)?).await?),
2607	"edit_comment" => reply(&work.edit_comment(args(body)?).await?),
2608	"delete_comment" => reply(&work.delete_comment(args(body)?).await?),
2609	"react" => reply(&work.react(args(body)?).await?),
2610	"commit_comments" => reply(&work.commit_comments(args(body)?).await?),
2611	"closed_by_commit" => reply(&work.closed_by_commit(args(body)?).await?),
2612	"add_commit_comment" => reply(&work.add_commit_comment(args(body)?).await?),
2613	"edit_commit_comment" => reply(&work.edit_commit_comment(args(body)?).await?),
2614	"delete_commit_comment" => reply(&work.delete_commit_comment(args(body)?).await?),
2615	"seen_checks" => reply(&work.seen_checks(args(body)?).await?),
2616	"report_checks" => reply(&work.report_checks(args(body)?).await?),
2617	"set_commit_status" => reply(&work.set_commit_status(args(body)?).await?),
2618	"create_commit_status" => reply(&work.create_commit_status(args(body)?).await?),
2619	"commit_statuses" => reply(&work.commit_statuses(args(body)?).await?),
2620	"combined_status" => reply(&work.combined_status(args(body)?).await?),
2621	"create_check_run" => reply(&work.create_check_run(args(body)?).await?),
2622	"update_check_run" => reply(&work.update_check_run(args(body)?).await?),
2623	"get_check_run" => reply(&work.get_check_run(args(body)?).await?),
2624	"check_run_annotations" => reply(&work.check_run_annotations(args(body)?).await?),
2625	"ref_check_runs" => reply(&work.ref_check_runs(args(body)?).await?),
2626	"ref_check_suites" => reply(&work.ref_check_suites(args(body)?).await?),
2627	"get_check_suite" => reply(&work.get_check_suite(args(body)?).await?),
2628	"rerequest_check_run" => reply(&work.rerequest_check_run(args(body)?).await?),
2629	"rerequest_check_suite" => reply(&work.rerequest_check_suite(args(body)?).await?),
2630	"request_check_action" => reply(&work.request_check_action(args(body)?).await?),
2631	"commit_checks" => reply(&work.commit_checks(args(body)?).await?),
2632	"advance" => reply(&work.advance(args(body)?).await?),
2633	"managed_pulls" => reply(&work.managed_pulls(args(body)?).await?),
2634	"queue" => reply(&work.queue(args(body)?).await?),
2635	"queue_build" => reply(&work.queue_build(args(body)?).await?),
2636	"report_queue" => reply(&work.report_queue(args(body)?).await?),
2637	"remove_from_queue" => reply(&work.remove_from_queue(args(body)?).await?),
2638	"locate_pull" => reply(&work.locate_pull(args(body)?).await?),
2639	"inbox_subject" => reply(&work.inbox_subject(args(body)?).await?),
2640	"get_settings" => reply(&work.get_settings(args(body)?).await?),
2641	"codeowners_errors" => reply(&work.codeowners_errors(args(body)?).await?),
2642	"update_settings" => reply(&work.update_settings(args(body)?).await?),
2643	"open_pull" => reply(&work.open_pull(args(body)?).await?),
2644	"list_pulls" => reply(&work.list_pulls(args(body)?).await?),
2645	"delivery_insights" => reply(&work.delivery_insights(args(body)?).await?),
2646	"pulls_for_repos" => reply(&work.pulls_for_repos(args(body)?).await?),
2647	"get_pull" => reply(&work.get_pull(args(body)?).await?),
2648	"update_pull" => reply(&work.update_pull(args(body)?).await?),
2649	"catch_up_pull" => reply(&work.catch_up_pull(args(body)?).await?),
2650	"ready_pull" => reply(&work.ready_pull(args(body)?).await?),
2651	"close_pull" => reply(&work.close_pull(args(body)?).await?),
2652	"reopen_pull" => reply(&work.reopen_pull(args(body)?).await?),
2653	"convert_pull_to_draft" => reply(&work.convert_pull_to_draft(args(body)?).await?),
2654	"merge_pull" => reply(&work.merge_pull(args(body)?).await?),
2655	// How it may merge, auto-merge and its branch (merging.rs, automerge.rs).
2656	"merge_choices" => reply(&work.merge_choices(args(body)?).await?),
2657	"set_auto_merge" => reply(&work.set_auto_merge(args(body)?).await?),
2658	"delete_head_branch" => reply(&work.delete_head_branch(args(body)?).await?),
2659	"restore_head_branch" => reply(&work.restore_head_branch(args(body)?).await?),
2660	"list_active_pulls" => reply(&work.list_active_pulls(args(body)?).await?),
2661	"by_author" => reply(&work.by_author(args(body)?).await?),
2662	"contributions" => reply(&work.contributions(args(body)?).await?),
2663	"append_session" => reply(&work.append_session(args(body)?).await?),
2664	"read_session" => reply(&work.read_session(args(body)?).await?),
2665	// Decision records (decisions.rs).
2666	"record_decision" => reply(&work.record_decision(args(body)?).await?),
2667	"list_decisions" => reply(&work.list_decisions(args(body)?).await?),
2668	// Coordinating work: harness sessions, claims, dependencies and a
2669	// work item's whole context (coordination/).
2670	"start_session" => reply(&work.start_session(args(body)?).await?),
2671	"update_session" => reply(&work.update_session(args(body)?).await?),
2672	"end_session" => reply(&work.end_session(args(body)?).await?),
2673	"get_work_session" => reply(&work.get_work_session(args(body)?).await?),
2674	"list_work_sessions" => reply(&work.list_work_sessions(args(body)?).await?),
2675	"pull_sessions" => reply(&work.pull_sessions(args(body)?).await?),
2676	"record_session_entries" => reply(&work.record_session_entries(args(body)?).await?),
2677	"claim_work_item" => reply(&work.claim_work_item(args(body)?).await?),
2678	"release_claim" => reply(&work.release_claim(args(body)?).await?),
2679	"hand_off_work_item" => reply(&work.hand_off_work_item(args(body)?).await?),
2680	"list_claims" => reply(&work.list_claims(args(body)?).await?),
2681	"get_work" => reply(&work.get_work(args(body)?).await?),
2682	// Runs of sandboxes (runs.rs).
2683	"open_run" => reply(&work.open_run(args(body)?).await?),
2684	"report_run" => reply(&work.report_run(args(body)?).await?),
2685	"list_runs" => reply(&work.list_runs(args(body)?).await?),
2686	"get_run" => reply(&work.get_run(args(body)?).await?),
2687	"list_sessions" => reply(&work.list_sessions(args(body)?).await?),
2688	"get_session" => reply(&work.get_session(args(body)?).await?),
2689	// What a sandbox may reach (guardrails.rs).
2690	"get_guardrails" => reply(&work.get_guardrails(args(body)?).await?),
2691	"update_guardrails" => reply(&work.update_guardrails(args(body)?).await?),
2692	"run_guardrails" => reply(&work.run_guardrails(args(body)?).await?),
2693	// The runs whose sandboxes stop with their repository (retired.rs).
2694	"runs_in_repo" => reply(&work.runs_in_repo(args(body)?).await?),
2695	"start_mergecheck" => reply(&work.start_mergecheck(args(body)?).await?),
2696	"report_mergecheck" => reply(&work.report_mergecheck(args(body)?).await?),
2697	// Rulesets (rulesets.rs): kept here, enforced here on merge and by
2698	// repos on push.
2699	"list_rulesets" => reply(&work.list_rulesets(args(body)?).await?),
2700	"get_ruleset" => reply(&work.get_ruleset(args(body)?).await?),
2701	"save_ruleset" => reply(&work.save_ruleset(args(body)?).await?),
2702	"delete_ruleset" => reply(&work.delete_ruleset(args(body)?).await?),
2703	"effective_rules" => reply(&work.effective_rules(args(body)?).await?),
2704	"rule_evaluations" => reply(&work.rule_evaluations(args(body)?).await?),
2705	"ref_rules" => reply(&work.ref_rules(args(body)?).await?),
2706	"record_evaluations" => reply(&work.record_evaluations(args(body)?).await?),
2707	"set_requires_pull_request" => reply(&work.set_requires_pull_request(args(body)?).await?),
2708	_ => Response::error("Unknown method", 404),
2709	};
2710	served.finish_timed(answered, &work.timing)
2711	}
2712
2713	/// The cron trigger: moves on the job that moves issues into work items
2714	/// (items/migration.rs), until it is done.
2715	#[event(scheduled)]
2716	async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
2717	let moved = async {
2718	let work = service(&env)?;
2719	work.migrate_work_items().await
2720	};
2721	if let Err(error) = moved.await {
2722	worker::console_error!("moving issues into work items: {error}");
2723	}
2724	}
2725
2726	/// Events from the bus, delivered on this service's own queue.
2727	#[event(queue)]
2728	async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
2729	let work = service(&env)?;
2730	for message in batch.messages()? {
2731	// A workspace renamed: its runs, work items, labels, milestones and
2732	// numbers move to the slug it has now.
2733	let renamed = [runs::RENAMED, guardrails::RENAMED, rulesets::RENAMED, items::RENAMED, attachments::RENAMED].concat();
2734	if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), &renamed).await? {
2735	message.ack();
2736	continue;
2737	}
2738	// A repository renamed or transferred: its runs and guardrails
2739	// follow, and where it is is recorded for work items and numbers.
2740	let transferred = [runs::TRANSFERRED, guardrails::TRANSFERRED, head_repo::TRANSFERRED, items::TRANSFERRED].concat();
2741	if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), &transferred).await? {
2742	message.ack();
2743	continue;
2744	}
2745	// A workspace deleted: what it kept for itself goes, its work items'
2746	// attachments' bytes first.
2747	if message.body().kind == "workspace.deleted"
2748	&& let Some(slug) = message.body().data["slug"].as_str()
2749	{
2750	attachments::purge_workspace(&env, &work.db, slug).await?;
2751	}
2752	let deleted = [guardrails::DELETED, items::DELETED, attachments::DELETED].concat();
2753	if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), &deleted).await? {
2754	message.ack();
2755	continue;
2756	}
2757	// An account purged: what it wrote shows as ghost (ghost.rs).
2758	let ghost = ghost::statements();
2759	let ghost: Vec<&str> = ghost.iter().map(String::as_str).collect();
2760	if g1t_kit::user_deleted::on_event(&env.d1("DB")?, message.body(), &ghost).await? {
2761	message.ack();
2762	continue;
2763	}
2764	// A repository purged: its attachments' bytes go before their rows do.
2765	if let Some(g1t_kit::lifecycle::Lifecycle::Purged(purged)) = g1t_kit::lifecycle::read(message.body()) {
2766	attachments::purge(&env, &work.db, &purged.repo_id).await?;
2767	}
2768	// A repository deleted, archived or purged, or a branch renamed (retired.rs).
2769	if work.on_retired(message.body()).await? {
2770	message.ack();
2771	continue;
2772	}
2773	work.on_event(message.body()).await?;
2774	message.ack();
2775	}
2776	Ok(())
2777	}
2778
2779	/// The rules that once read a pull request's author read its owner now:
2780	/// whoever asked g1t for it, or its author. For each, the person who asked
2781	/// is held to what an author was, and g1t's agent (a token it works with)
2782	/// gains nothing by being the author.
2783	#[cfg(test)]
2784	mod owner_rules {
2785	use super::*;
2786	use crate::rows::stored::{ASKER, G1T, pull};
2787	use g1t_contracts::identity::AGENT_ID;
2788
2789	const SOMEONE: &str = "usr_2";
2790
2791	#[test]
2792	fn no_self_approval() {
2793	// add_comment refuses a verdict on one that is theirs.
2794	let made = pull(G1T, Some(ASKER));
2795	assert!(made.is_owned_by(ASKER.0), "the person who asked cannot approve it");
2796	assert!(!made.is_owned_by(AGENT_ID), "g1t's review agent still gives its verdict");
2797	assert!(!made.is_owned_by(SOMEONE));
2798	}
2799
2800	#[test]
2801	fn what_an_author_could_do_without_a_role() {
2802	// manageable_pull (update, ready, close), catch_up_pull on a fork,
2803	// and append_session: theirs to do.
2804	let made = pull(G1T, Some(ASKER));
2805	assert!(made.is_owned_by(ASKER.0));
2806	assert!(!made.is_owned_by(SOMEONE), "anyone else still needs the role");
2807	assert!(!made.is_owned_by(AGENT_ID), "being its author gives g1t's tokens nothing more");
2808	}
2809
2810	#[test]
2811	fn nobody_is_asked_to_review_what_they_asked_for() {
2812	// update_pull drops the owner from the reviewers asked.
2813	let made = pull(G1T, Some(ASKER));
2814	let mut reviewers = vec!["syntaqx".to_owned(), "ana".to_owned()];
2815	reviewers.retain(\|name\| *name != made.owner().username);
2816	assert_eq!(reviewers, ["ana"]);
2817	}
2818
2819	#[test]
2820	fn sandboxes_act_as_whoever_asked() {
2821	// MergecheckJob and the merge queue's job
2822	// carry who the sandbox's credential acts for: a real account.
2823	let made = pull(G1T, Some(ASKER));
2824	assert_eq!(made.owner().id, ASKER.0);
2825	let acts_as = made.requested_by.unwrap_or(made.author);
2826	assert_eq!(acts_as.id, ASKER.0);
2827	// g1t's own work, which nobody asked for, acts as g1t, as before.
2828	let own = pull(("g1t", "g1t"), None);
2829	assert_eq!(own.requested_by.unwrap_or(own.author).id, "g1t");
2830	}
2831
2832	#[test]
2833	fn events_name_g1t_and_whoever_asked() {
2834	let made = pull(G1T, Some(ASKER));
2835	let event = serde_json::to_value(Work::pull_event(&made)).unwrap();
2836	assert_eq!(event["author"], serde_json::json!({ "id": AGENT_ID, "username": "g1t" }));
2837	assert_eq!(event["requestedBy"], serde_json::json!({ "id": "usr_1", "username": "syntaqx" }));
2838	let own = serde_json::to_value(Work::pull_event(&pull(ASKER, None))).unwrap();
2839	assert!(own.get("requestedBy").is_none());
2840	}
2841
2842	#[test]
2843	fn only_a_closed_pull_request_reopens_and_only_an_open_one_turns_draft() {
2844	assert_eq!(reopen_refusal(PullStatus::Closed), None);
2845	assert!(reopen_refusal(PullStatus::Merged).is_some(), "a merge cannot be undone");
2846	assert!(reopen_refusal(PullStatus::Open).is_some());
2847	assert!(reopen_refusal(PullStatus::Draft).is_some());
2848	assert_eq!(draft_refusal(PullStatus::Open), None);
2849	assert!(draft_refusal(PullStatus::Draft).is_some());
2850	assert!(draft_refusal(PullStatus::Closed).is_some());
2851	assert!(draft_refusal(PullStatus::Merged).is_some());
2852	}
2853
2854	#[test]
2855	fn a_pull_request_reopens_as_what_it_was_closed_as() {
2856	assert_eq!(reopened_status(Some("draft")), PullStatus::Draft);
2857	assert_eq!(reopened_status(Some("open")), PullStatus::Open);
2858	// Closed before this was recorded, or by a merge of another.
2859	assert_eq!(reopened_status(None), PullStatus::Open);
2860	}
2861	}