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