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,895 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).filter(\|id\| !id.is_empty()), 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 =
1057	items::named(self, &repo, item_id.as_deref(), &format!("{title}\n{}", body.as_deref().unwrap_or_default()), &[&a.actor]).await?;
1058	self.db
1059	.prepare(
1060	"INSERT INTO pulls
1061	(id, repo_id, number, item_id, title, body, agent, runtime,
1062	status, fork_repo_id, fork_namespace, fork_name, source_branch, head_commit,
1063	author_id, author_name, requested_by_id, requested_by_name, created_at, updated_at,
1064	base_branch, head_repo_id, head_namespace, head_name, closes_items, parts)
1065	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
1066	)
1067	.bind(&[
1068	id.as_str().into(),
1069	repo.id.as_str().into(),
1070	number.into(),
1071	optional(&item_id),
1072	title.into(),
1073	optional(&body),
1074	agent.into(),
1075	runtime.into(),
1076	status.into(),
1077	optional(&fork.as_ref().map(\|fork\| fork.id.clone())),
1078	optional(&fork.as_ref().map(\|fork\| fork.namespace.clone())),
1079	optional(&fork.as_ref().map(\|fork\| fork.name.clone())),
1080	optional(&branch.map(str::to_owned)),
1081	optional(&head),
1082	author.id.as_str().into(),
1083	author.username.as_str().into(),
1084	optional(&requested_by.as_ref().map(\|user\| user.id.clone())),
1085	optional(&requested_by.as_ref().map(\|user\| user.username.clone())),
1086	timestamp.as_str().into(),
1087	timestamp.as_str().into(),
1088	optional(&base),
1089	optional(&head_repo_id),
1090	optional(&head_path.as_ref().map(\|path\| path.namespace.clone())),
1091	optional(&head_path.as_ref().map(\|path\| path.name.clone())),
1092	serde_json::to_string(&named.close_ids())?.into(),
1093	serde_json::to_string(&named.part_ids())?.into(),
1094	])?
1095	.run()
1096	.await?;
1097	let Some(mut pull) = self.pull(&repo.id, number).await? else {
1098	return Ok(no_pull());
1099	};
1100	fill_base(&mut pull, &repo);
1101	// The work item hears of it, and links the repository it is in.
1102	if let Some(held) = &item {
1103	let text = format!("opened {}#{} for this", items::path_of(&repo), pull.number);
1104	items::note(self, &held.item.id, &a.actor, &text).await?;
1105	items::touch(self, held, &repo, &a.actor).await?;
1106	}
1107	// Its head and merge refs in the repository, before its events go out.
1108	self.keep_pull_refs(&pull).await;
1109	self.publish(
1110	"pull.opened",
1111	&repo.id,
1112	&a.actor,
1113	PullEvent {
1114	agent: Some(pull.agent.clone()),
1115	base: pull.base.clone(),
1116	..Self::pull_event(&pull)
1117	},
1118	)
1119	.await?;
1120	// Its code owners asked to review (codeowners.rs).
1121	self.refresh_code_owners(&pull).await;
1122	let pull = self.pull(&repo.id, number).await?.unwrap_or(pull);
1123	// A branch pushed before may overlap others from the start.
1124	self.announce_overlaps(&pull, Some(a.actor.id.clone())).await;
1125	Ok(Outcome::Ok(pull))
1126	}
1127
1128	async fn list_pulls(&self, a: ListPullsArgs) -> Result<Outcome<Vec<Pull>>> {
1129	let filter = match a.state {
1130	Some(State::Open) => "AND status IN ('draft', 'open')",
1131	Some(State::Closed) => "AND status IN ('merged', 'closed')",
1132	None => "",
1133	};
1134	let label = a
1135	.label
1136	.map(\|label\| label.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase())
1137	.filter(\|label\| !label.is_empty());
1138	let milestone = a.milestone;
1139	let read = \|repo_id: String\| {
1140	let label = label.clone();
1141	async move {
1142	let query = self
1143	.db
1144	.prepare(format!(
1145	"SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? {filter}
1146	AND (? IS NULL OR EXISTS
1147	(SELECT 1 FROM json_each(pulls.labels) WHERE json_each.value = ?))
1148	AND (? IS NULL OR milestone = ?)
1149	ORDER BY number DESC LIMIT ?"
1150	))
1151	.bind(&[
1152	repo_id.into(),
1153	optional(&label),
1154	optional(&label),
1155	optional_number(milestone),
1156	optional_number(milestone),
1157	LIST_PAGE.into(),
1158	])?;
1159	self.timing.db(1, query.all()).await?.results::<PullRow>()
1160	}
1161	};
1162	let (repo, rows) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
1163	let base = a.base.map(\|base\| base.trim().to_owned()).filter(\|base\| !base.is_empty());
1164	Ok(Outcome::Ok(
1165	rows.into_iter()
1166	.map(\|row\| {
1167	let mut pull = Pull::from(row);
1168	fill_base(&mut pull, &repo);
1169	pull
1170	})
1171	.filter(\|pull\| base.as_deref().is_none_or(\|base\| pull.base.as_deref() == Some(base)))
1172	.collect(),
1173	))
1174	}
1175
1176	/// `pulls_for_repos`: what `list_pulls` gives, open and closed, for many
1177	/// repositories at once: one access check with repos for all of them
1178	/// and one query, instead of two of each per repository.
1179	async fn pulls_for_repos(&self, a: PullsForReposArgs) -> Result<Vec<RepoPulls>> {
1180	let ids: Vec<String> = a.repo_ids.into_iter().take(MAX_PULLS_FOR_REPOS).collect();
1181	if ids.is_empty() {
1182	return Ok(Vec::new());
1183	}
1184	let limit = a.limit.clamp(1, LIST_PAGE);
1185	// The rows are read beside the access check, for every id asked
1186	// about; those of repositories the viewer cannot read are dropped.
1187	let asked = serde_json::to_string(&ids)?;
1188	let check = ReadableArgs { ids, viewer: a.viewer };
1189	let readable = self.timing.rpc(g1t_kit::call::<_, Vec<Repo>>(&self.repos, "readable", &check));
1190	let (readable, rows) = try_join(readable, self.timing.db(1, self.newest_pulls(asked, limit))).await?;
1191	if readable.is_empty() {
1192	return Ok(Vec::new());
1193	}
1194	let mut answer: Vec<RepoPulls> = readable
1195	.iter()
1196	.map(\|repo\| RepoPulls { repo_id: repo.id.clone(), open: Vec::new(), closed: Vec::new() })
1197	.collect();
1198	for pull in rows.into_iter().map(Pull::from) {
1199	let Some(entry) = answer.iter_mut().find(\|entry\| entry.repo_id == pull.repo_id) else {
1200	continue;
1201	};
1202	match pull.status {
1203	PullStatus::Draft \| PullStatus::Open => entry.open.push(pull),
1204	PullStatus::Merged \| PullStatus::Closed => entry.closed.push(pull),
1205	}
1206	}
1207	for entry in &mut answer {
1208	entry.open.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1209	entry.closed.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1210	}
1211	Ok(answer)
1212	}
1213
1214	/// The newest `limit` of each repository's open (draft or open) and
1215	/// closed (merged or closed) pull requests, for the ids in `ids` (JSON).
1216	async fn newest_pulls(&self, ids: String, limit: u32) -> Result<Vec<PullRow>> {
1217	self.db
1218	.prepare(format!(
1219	"SELECT * FROM (
1220	SELECT {PULL_COLUMNS}, ROW_NUMBER() OVER (
1221	PARTITION BY pulls.repo_id, pulls.status IN ('draft', 'open') ORDER BY pulls.number DESC
1222	) AS place
1223	FROM pulls WHERE pulls.repo_id IN (SELECT value FROM json_each(?1))
1224	) WHERE place <= ?2"
1225	))
1226	.bind(&[ids.into(), limit.into()])?
1227	.all()
1228	.await?
1229	.results::<PullRow>()
1230	}
1231
1232	async fn get_pull(&self, a: ViewArgs) -> Result<Outcome<PullDetail>> {
1233	let number = a.number;
1234	// Every row the page and the lifecycle read, in one batch started
1235	// beside the access check; the helpers below read from it.
1236	let namespace = a.repo.namespace.clone();
1237	let read = \|repo_id: String\| self.prefetch_pull(repo_id, namespace.clone(), number);
1238	let Outcome::Ok((repo, Some(found))) = self.repo_then(&a.repo, &a.viewer, read).await? else {
1239	return Ok(no_pull());
1240	};
1241	let Some(row) = found.first::<PullRow>(prefetch::Slot::Pull)? else {
1242	return Ok(no_pull());
1243	};
1244	let stored = found.first::<StoredBehind>(prefetch::Slot::Pull)?;
1245	// Its work item, as far as the viewer may see it.
1246	let item = found
1247	.held(prefetch::Slot::Item)?
1248	.filter(\|held\| held.may(a.viewer.as_ref(), Capability::Read))
1249	.map(\|held\| held.shown(a.viewer.as_ref()));
1250	let mut comments: Vec<Comment> =
1251	found.rows::<CommentRow>(prefetch::Slot::Comments)?.into_iter().map(Comment::from).collect();
1252	let reactions = reactions::attach(&found.rows::<reactions::ReactionRow>(prefetch::Slot::Reactions)?, &row.id, &mut comments);
1253	self.keep_prefetched(Some(found));
1254	let default_branch = repo.default_branch.clone();
1255	let detail = self.pull_detail(repo, Pull::from(row), item, comments, stored, &a.viewer).await;
1256	self.keep_prefetched(None);
1257	// The branch it merges into, named, for whoever reads it.
1258	Ok(match detail? {
1259	Outcome::Ok(mut detail) => {
1260	if detail.pull.base.as_deref().is_none_or(str::is_empty) {
1261	detail.pull.base = Some(default_branch);
1262	}
1263	detail.reactions = reactions;
1264	Outcome::Ok(detail)
1265	}
1266	failed => failed,
1267	})
1268	}
1269
1270	async fn pull_detail(
1271	&self,
1272	repo: Repo,
1273	mut pull: Pull,
1274	item: Option<WorkItem>,
1275	comments: Vec<Comment>,
1276	stored: Option<StoredBehind>,
1277	viewer: &Viewer,
1278	) -> Result<Outcome<PullDetail>> {
1279	// Whether it is behind, as worked out with its mergeability on the
1280	// last push to either side (mergeability.rs), when that was for
1281	// its head as it is now; otherwise asked of the repos service.
1282	let known_behind = stored.and_then(\|stored\| stored.for_head(pull.head_commit.as_deref()));
1283	// Worked out on each push; this covers a pull request from before
1284	// that was recorded.
1285	if pull.files.is_empty() && pull.head_commit.is_some() {
1286	pull.files = self.refresh_files(&pull).await?;
1287	}
1288	// Everything else at once: none of it depends on the rest, and each
1289	// is a round trip of its own.
1290	let standing = async {
1291	// Mergeability first: where g1t sees a pull request through, a
1292	// conflict decides its next step.
1293	let behind = async {
1294	match known_behind {
1295	Some(behind) => Ok(behind),
1296	None => {
1297	let behind = self.is_behind(&pull).await?;
1298	// Kept for the next view when the mergeability on
1299	// record is for this head: a pull request from
1300	// before `behind` was kept asks once.
1301	if let Some(head) = pull.head_commit.as_deref()
1302	&& pull.status.is_active()
1303	{
1304	self.db
1305	.prepare(
1306	"UPDATE pulls SET behind = ?1
1307	WHERE id = ?2 AND behind IS NULL AND mergeable_key LIKE ?3 \|\| '..%'",
1308	)
1309	.bind(&[u32::from(behind).into(), pull.id.as_str().into(), head.into()])?
1310	.run()
1311	.await?;
1312	}
1313	Ok(behind)
1314	}
1315	}
1316	};
1317	let (merge, behind) = try_join(self.mergeability(&pull), behind).await?;
1318	let lifecycle = self.assess(&pull, &item, behind).await?.map(\|(lifecycle, _)\| lifecycle);
1319	Ok::<_, worker::Error>((behind, lifecycle, merge))
1320	};
1321	let (((behind, lifecycle, (mergeable, conflicts)), (landing, stalled), comments), (checks, overlaps)) =
1322	try_join(
1323	try_join3(standing, self.landing_state(&pull.id), async { Ok(comments) }),
1324	try_join(self.latest_checks(&pull.id), self.overlaps(&pull)),
1325	)
1326	.await?;
1327	// The rules of the branch it merges into, as they stack, and which
1328	// of them it does not meet yet, for whoever is looking.
1329	let (statuses, settings, gate) = try_join3(
1330	self.statuses(&repo.id, pull.head_commit.as_deref()),
1331	self.settings_on(&repo, &pull),
1332	async {
1333	if pull.status.is_active() {
1334	self.merge_gate_seen(&repo, &pull, viewer.as_ref(), behind).await.map(Some)
1335	} else {
1336	Ok(None)
1337	}
1338	},
1339	)
1340	.await?;
1341	let code_owners = self.pull_code_owners(&pull, &comments, &settings).await?;
1342	// The work items it closes, as its title and description say now;
1343	// none where the repository does not close them that way.
1344	if pull.status.is_active() {
1345	let named = self.items_to_close(&pull, &repo, &settings, false, &[]).await?;
1346	pull.closes = named.close_numbers();
1347	pull.parts = named.part_numbers();
1348	}
1349	let required = if settings.required_checks.is_empty() {
1350	Vec::new()
1351	} else {
1352	let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses.clone()).await?;
1353	required_checks(&settings.required_checks, &checked)
1354	};
1355	Ok(Outcome::Ok(PullDetail {
1356	required_checks: required,
1357	rules: gate.map(\|gate\| rulesets::merge_rules(&gate.judged, &gate.requirements, pull.base_branch(&repo.default_branch) == repo.default_branch)),
1358	code_owners,
1359	comments,
1360	checks,
1361	overlaps,
1362	behind,
1363	lifecycle,
1364	landing,
1365	stalled,
1366	statuses,
1367	mergeable,
1368	conflicts,
1369	earlier_checks: self.earlier_checks(&pull.id).await?,
1370	reactions: Vec::new(),
1371	item,
1372	pull,
1373	}))
1374	}
1375
1376	/// The pull request, whatever its status, if its repository is not
1377	/// archived and `actor` opened it or may triage its pull requests.
1378	async fn managed_pull(
1379	&self,
1380	actor: &User,
1381	path: &RepoPath,
1382	number: u32,
1383	) -> Result<Outcome<Pull>> {
1384	let (repo, pull) = check!(self.pull_at(path, number, &Some(actor.clone())).await?);
1385	check!(writable(&repo));
1386	if !pull.is_owned_by(&actor.id) {
1387	check!(allowed(Some(actor), &repo, Capability::Triage));
1388	}
1389	Ok(Outcome::Ok(pull))
1390	}
1391
1392	/// The pull request, if it is still active and `actor` opened it or
1393	/// may triage the repository's pull requests.
1394	async fn manageable_pull(
1395	&self,
1396	actor: &User,
1397	path: &RepoPath,
1398	number: u32,
1399	) -> Result<Outcome<Pull>> {
1400	let pull = check!(self.managed_pull(actor, path, number).await?);
1401	if !pull.status.is_active() {
1402	return Ok(Outcome::fail(
1403	FailureCode::Conflict,
1404	format!("This pull request is already {}.", pull.status.as_str()),
1405	));
1406	}
1407	Ok(Outcome::Ok(pull))
1408	}
1409
1410	/// Brings a pull request up to date with the default branch without a
1411	/// sandbox, where the repos service can do that safely. Whoever could
1412	/// have pushed the merge themselves may ask: whoever opened it (or asked
1413	/// g1t for it), for a fork; anyone who may push, for a branch of the
1414	/// repository. When it needs a
1415	/// real merge, says so, naming the conflicting files if a probe found
1416	/// them, and pushes nothing.
1417	async fn catch_up_pull(&self, a: PullActionArgs) -> Result<Outcome<PullBranchUpdate>> {
1418	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &Some(a.actor.clone())).await?);
1419	check!(writable(&repo));
1420	if !pull.status.is_active() {
1421	return Ok(Outcome::fail(
1422	FailureCode::Conflict,
1423	format!("This pull request is already {}.", pull.status.as_str()),
1424	));
1425	}
1426	if pull.fork_repo_id.is_some() {
1427	if !pull.is_owned_by(&a.actor.id) {
1428	return Ok(Outcome::fail(
1429	FailureCode::Forbidden,
1430	"Only whoever opened this pull request, or asked g1t for it, can update it.",
1431	));
1432	}
1433	} else if !pull.from_head_repo() {
1434	check!(allowed(Some(&a.actor), &repo, Capability::Push));
1435	}
1436	// A fork's branch is updated by those who may push to the fork,
1437	// which repos decides.
1438	let updated: Outcome<PullBranchUpdate> = g1t_kit::call(
1439	&self.repos,
1440	"update_pull_branch",
1441	&UpdatePullBranchArgs {
1442	source_id: pull.source_id().to_owned(),
1443	branch: pull.branch.clone(),
1444	number: pull.number,
1445	actor: a.actor,
1446	target_branch: pull.base.clone(),
1447	target_id: pull.target_id(),
1448	method: a.update_method.unwrap_or_default(),
1449	expected_head: a.expected_head.clone().filter(\|head\| !head.trim().is_empty()),
1450	},
1451	)
1452	.await?;
1453	// A probe that found conflicts says more than "both changed it".
1454	if let Outcome::Ok(PullBranchUpdate::NeedsAgent { .. }) = &updated
1455	&& let Some(files) = self.conflicting_files(&pull).await?
1456	&& !files.is_empty()
1457	{
1458	return Ok(Outcome::Ok(PullBranchUpdate::NeedsAgent {
1459	reason: NeedsAgentReason::Conflicting,
1460	detail: "Merging it conflicts.".to_owned(),
1461	paths: files,
1462	}));
1463	}
1464	Ok(updated)
1465	}
1466
1467	async fn update_pull(&self, a: UpdatePullArgs) -> Result<Outcome<Pull>> {
1468	let pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1469	// The branch it merges into, its milestone and its labels first:
1470	// each can be refused, and then nothing else changes.
1471	if a.base.is_some() \|\| a.milestone.is_some() \|\| a.labels.is_some() {
1472	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1473	if let Some(base) = &a.base {
1474	check!(self.change_base(&a.actor, &repo, &pull, base).await?);
1475	}
1476	if let Some(number) = a.milestone {
1477	check!(items::set_milestone(self, &a.actor, &items::Labeled::Pull(repo.clone(), pull.clone()), number).await?);
1478	}
1479	if let Some(labels) = &a.labels {
1480	check!(items::relabel(self, &a.actor, &items::Labeled::Pull(repo.clone(), pull.clone()), labels).await?);
1481	}
1482	}
1483	let assignees = match a.assignees {
1484	Some(names) => Some(check!(self.valid_assignees(names).await?)),
1485	None => None,
1486	};
1487	// Teams, named `workspace/team`, apart from the people.
1488	let (team_names, a_reviewers) = match a.reviewers {
1489	Some(names) => {
1490	let (teams, people): (Vec<String>, Vec<String>) =
1491	names.into_iter().partition(\|name\| team_reviews::team_name(name).is_some());
1492	(Some(teams), Some(people))
1493	}
1494	None => (None, None),
1495	};
1496	let teams = match team_names {
1497	Some(names) => Some(check!(self.valid_team_reviewers(&a.actor, &a.repo, &pull, names).await?)),
1498	None => None,
1499	};
1500	let reviewers = match a_reviewers {
1501	Some(names) => {
1502	let mut reviewers = check!(self.valid_assignees(names).await?);
1503	// Nobody is asked to review their own, nor what they had g1t make.
1504	reviewers.retain(\|name\| *name != pull.owner().username);
1505	Some(reviewers)
1506	}
1507	None => None,
1508	};
1509	self.db
1510	.prepare(
1511	"UPDATE pulls
1512	SET assignees = COALESCE(?, assignees), updated_at = ?
1513	WHERE id = ?",
1514	)
1515	.bind(&[
1516	optional(&assignees.as_ref().map(serde_json::to_string).transpose()?),
1517	rfc3339(now_ms()).into(),
1518	pull.id.as_str().into(),
1519	])?
1520	.run()
1521	.await?;
1522	if let Some(assignees) = &assignees {
1523	self.note_changes(
1524	On::Pull(&pull.repo_id, pull.number),
1525	&a.actor,
1526	&pull.assignees,
1527	assignees,
1528	("assigned", "unassigned"),
1529	)
1530	.await?;
1531	}
1532	// Who was newly assigned or asked to review, and whose request was
1533	// withdrawn: the inbox tells them, and webhooks say so.
1534	let newly = \|after: &[String], before: &[String]\| -> Vec<String> {
1535	after.iter().filter(\|name\| !before.contains(name)).cloned().collect()
1536	};
1537	if let Some(assignees) = &assignees {
1538	let added = newly(assignees, &pull.assignees);
1539	if !added.is_empty() {
1540	self.publish(
1541	"pull.assigned",
1542	&pull.repo_id,
1543	&a.actor,
1544	PullEvent {
1545	assignees: Some(assignees.clone()),
1546	added: Some(added),
1547	..Self::pull_event(&pull)
1548	},
1549	)
1550	.await?;
1551	}
1552	}
1553	if reviewers.is_some() \|\| teams.is_some() {
1554	let people = reviewers.unwrap_or_else(\|\| pull.reviewers.clone());
1555	let teams = teams.unwrap_or_else(\|\| pull.team_reviewers.clone());
1556	self.set_reviewers(&pull, people, teams, Some(&a.actor), false).await?;
1557	}
1558	Ok(match self.pull(&pull.repo_id, pull.number).await? {
1559	Some(pull) => Outcome::Ok(pull),
1560	None => no_pull(),
1561	})
1562	}
1563
1564	/// Points an open pull request at another branch to merge into. Needs
1565	/// the Write role. What it would merge, whether it is behind, its
1566	/// mergeability and its checks are all worked out against the new base.
1567	async fn change_base(&self, actor: &User, repo: &Repo, pull: &Pull, base: &str) -> Result<Outcome<()>> {
1568	check!(allowed(Some(actor), repo, Capability::Push));
1569	let base = base.trim();
1570	if base.is_empty() {
1571	return Ok(Outcome::fail(FailureCode::Invalid, "Name the branch it should merge into."));
1572	}
1573	let before = pull.base_branch(&repo.default_branch).to_owned();
1574	if base == before {
1575	return Ok(Outcome::Ok(()));
1576	}
1577	if !pull.from_other_repo() && pull.branch.as_deref() == Some(base) {
1578	return Ok(Outcome::fail(
1579	FailureCode::Invalid,
1580	format!("A pull request cannot merge {base} into itself. Choose another base."),
1581	));
1582	}
1583	let exists: Option<String> = g1t_kit::call(
1584	&self.repos,
1585	"head",
1586	&HeadArgs { repo_id: repo.id.clone(), branch: base.to_owned() },
1587	)
1588	.await?;
1589	if exists.is_none() {
1590	return Ok(Outcome::fail(FailureCode::NotFound, format!("There is no branch named {base} to merge into.")));
1591	}
1592	// In the merge queue it was headed for the default branch; it
1593	// leaves the queue for another base.
1594	let left_queue = self
1595	.leave(&pull.repo_id, pull, QueueState::Removed, Some("Its base branch changed."))
1596	.await?;
1597	// Its tested commit is being published to the default branch: a new
1598	// base would describe a merge that is not the one taking place.
1599	if !left_queue && self.publishing(&pull.id).await? {
1600	return Ok(Outcome::fail(FailureCode::Conflict, "This pull request is being published. Wait for landing to finish."));
1601	}
1602	let stored = stored_base(base, repo);
1603	self.db
1604	.prepare(
1605	"UPDATE pulls
1606	SET base_branch = ?, updated_at = ?, land_requested = NULL, land_requested_at = NULL, behind = NULL,
1607	mergeable = NULL, mergeable_key = NULL, conflicts = NULL
1608	WHERE id = ?",
1609	)
1610	.bind(&[optional(&stored), rfc3339(now_ms()).into(), pull.id.as_str().into()])?
1611	.run()
1612	.await?;
1613	self.note(
1614	&pull.repo_id,
1615	pull.number,
1616	actor,
1617	&format!("changed the base branch from `{before}` to `{base}`"),
1618	)
1619	.await?;
1620	self.publish(
1621	"pull.base_changed",
1622	&pull.repo_id,
1623	actor,
1624	PullEvent { base: Some(base.to_owned()), ..Self::pull_event(pull) },
1625	)
1626	.await?;
1627	if left_queue {
1628	self.publish_as(
1629	"queue.changed",
1630	&pull.repo_id,
1631	None,
1632	g1t_contracts::events::QueueChanged { repo_id: pull.repo_id.clone() },
1633	)
1634	.await?;
1635	}
1636	// Its merge ref, and whether it merges cleanly, against the new base.
1637	if let Some(moved) = self.pull_by_id(&pull.id).await? {
1638	self.keep_pull_refs(&moved).await;
1639	if let Err(error) = self.assess_mergeability(&moved).await {
1640	worker::console_warn!("mergeability of {}: {error}", pull.id);
1641	}
1642	}
1643	Ok(Outcome::Ok(()))
1644	}
1645
1646	/// Marks a draft ready for review, or updates the description of one
1647	/// that already is.
1648	async fn ready_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1649	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1650	let summary = Some(a.summary.trim().to_owned()).filter(\|summary\| !summary.is_empty());
1651	let now = rfc3339(now_ms());
1652	// The work items it closes and is part of, as its description says
1653	// now (closing.rs).
1654	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
1655	let own = self.pull_item_id(&pull.id).await?;
1656	let text = format!("{}\n{}", pull.title, summary.as_deref().or(pull.body.as_deref()).unwrap_or_default());
1657	let named = items::named(self, &repo, own.as_deref(), &text, &[&a.actor]).await?;
1658	self.db
1659	.prepare(
1660	"UPDATE pulls SET status = 'open', body = COALESCE(?, body), updated_at = ?, closes_items = ?, parts = ?
1661	WHERE id = ?",
1662	)
1663	.bind(&[
1664	optional(&summary),
1665	now.as_str().into(),
1666	serde_json::to_string(&named.close_ids())?.into(),
1667	serde_json::to_string(&named.part_ids())?.into(),
1668	pull.id.as_str().into(),
1669	])?
1670	.run()
1671	.await?;
1672	if pull.status == PullStatus::Draft {
1673	// The head as it is now: the push that came just before may not
1674	// have reached `head_commit` yet, and workflows run on it.
1675	let commit = self.live_head(&pull).await?.or_else(\|\| pull.head_commit.clone());
1676	self.publish(
1677	"pull.ready",
1678	&pull.repo_id,
1679	&a.actor,
1680	PullEvent {
1681	commit,
1682	..Self::pull_event(&pull)
1683	},
1684	)
1685	.await?;
1686	}
1687	if pull.status == PullStatus::Draft {
1688	self.note(
1689	&pull.repo_id,
1690	pull.number,
1691	&a.actor,
1692	"marked this ready for review",
1693	)
1694	.await?;
1695	}
1696	pull.status = PullStatus::Open;
1697	pull.body = summary.or(pull.body);
1698	pull.closes = named.close_numbers();
1699	pull.parts = named.part_numbers();
1700	pull.updated_at = now;
1701	// A draft's code owners are asked once it is ready.
1702	self.refresh_code_owners(&pull).await;
1703	if let Some(fresh) = self.pull(&pull.repo_id, pull.number).await? {
1704	pull.reviewers = fresh.reviewers;
1705	pull.team_reviewers = fresh.team_reviewers;
1706	}
1707	Ok(Outcome::Ok(pull))
1708	}
1709
1710	async fn close_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1711	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1712	let now = rfc3339(now_ms());
1713	let closed = self.db
1714	// What it was closed as, so that reopening brings that back.
1715	.prepare("UPDATE pulls SET status = 'closed', closed_from = ?, updated_at = ?, auto_merge = NULL WHERE id = ? AND status <> 'merged'
1716	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)
1717	RETURNING id AS value")
1718	.bind(&[pull.status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1719	.first::<rows::ValueRow>(None)
1720	.await?;
1721	if closed.is_none() {
1722	return Ok(Outcome::fail(FailureCode::Conflict, "This pull request is being published or has already merged."));
1723	}
1724	self.publish(
1725	"pull.closed",
1726	&pull.repo_id,
1727	&a.actor,
1728	Self::pull_event(&pull),
1729	)
1730	.await?;
1731	self.note(
1732	&pull.repo_id,
1733	pull.number,
1734	&a.actor,
1735	"closed this",
1736	)
1737	.await?;
1738	// A closed pull request leaves the merge queue.
1739	if self
1740	.leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was closed."))
1741	.await?
1742	{
1743	self.publish_as(
1744	"queue.changed",
1745	&pull.repo_id,
1746	None,
1747	g1t_contracts::events::QueueChanged {
1748	repo_id: pull.repo_id.clone(),
1749	},
1750	)
1751	.await?;
1752	}
1753	pull.status = PullStatus::Closed;
1754	pull.auto_merge = None;
1755	pull.updated_at = now;
1756	Ok(Outcome::Ok(pull))
1757	}
1758
1759	/// Opens a closed pull request again: as the draft it was, if it was
1760	/// closed as one, else ready for review. A merged one stays merged.
1761	async fn reopen_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1762	let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
1763	if let Some(refusal) = reopen_refusal(pull.status) {
1764	return Ok(Outcome::fail(FailureCode::Conflict, refusal));
1765	}
1766	// The head as it is now, which workflows run on. A branch of the
1767	// repository that was deleted since leaves nothing to reopen; a
1768	// fork removed after the close is made again (repos, `pull.reopened`).
1769	let head = self.live_head(&pull).await?;
1770	if head.is_none() && pull.fork_repo_id.is_none() {
1771	return Ok(Outcome::fail(
1772	FailureCode::Conflict,
1773	format!(
1774	"The branch {} no longer exists. Push it again to reopen this pull request.",
1775	head_repo::source_label(pull.head_repo.as_ref(), pull.branch.as_deref().unwrap_or_default())
1776	),
1777	));
1778	}
1779	let closed_from: Option<ClosedFrom> = self
1780	.db
1781	.prepare("SELECT closed_from FROM pulls WHERE id = ?")
1782	.bind(&[pull.id.as_str().into()])?
1783	.first(None)
1784	.await?;
1785	let status = reopened_status(closed_from.and_then(\|row\| row.closed_from).as_deref());
1786	let now = rfc3339(now_ms());
1787	self.db
1788	.prepare("UPDATE pulls SET status = ?, closed_from = NULL, superseded_by = NULL, updated_at = ? WHERE id = ?")
1789	.bind(&[status.as_str().into(), now.as_str().into(), pull.id.as_str().into()])?
1790	.run()
1791	.await?;
1792	self.publish(
1793	"pull.reopened",
1794	&pull.repo_id,
1795	&a.actor,
1796	PullEvent {
1797	commit: head.or_else(\|\| pull.head_commit.clone()),
1798	..Self::pull_event(&pull)
1799	},
1800	)
1801	.await?;
1802	self.note(
1803	&pull.repo_id,
1804	pull.number,
1805	&a.actor,
1806	"reopened this",
1807	)
1808	.await?;
1809	pull.status = status;
1810	pull.superseded_by = None;
1811	pull.updated_at = now;
1812	// Its refs in the repository, and whether it still merges cleanly,
1813	// now that it is open again.
1814	self.keep_pull_refs(&pull).await;
1815	if let Err(error) = self.assess_mergeability(&pull).await {
1816	worker::console_warn!("mergeability of {}: {error}", pull.id);
1817	}
1818	Ok(Outcome::Ok(pull))
1819	}
1820
1821	/// Turns a pull request that is ready for review back into a draft: it
1822	/// cannot be merged until it is marked ready again, and it leaves the
1823	/// merge queue and any merge that was waiting for it to catch up.
1824	async fn convert_pull_to_draft(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1825	let mut pull = check!(self.managed_pull(&a.actor, &a.repo, a.number).await?);
1826	if let Some(refusal) = draft_refusal(pull.status) {
1827	return Ok(Outcome::fail(FailureCode::Conflict, refusal));
1828	}
1829	let now = rfc3339(now_ms());
1830	let drafted = self.db
1831	.prepare(
1832	"UPDATE pulls SET status = 'draft', land_requested = NULL, land_requested_at = NULL, updated_at = ?
1833	WHERE id = ? AND status = 'open'
1834	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)
1835	RETURNING id AS value",
1836	)
1837	.bind(&[now.as_str().into(), pull.id.as_str().into()])?
1838	.first::<rows::ValueRow>(None)
1839	.await?;
1840	if drafted.is_none() {
1841	return Ok(Outcome::fail(FailureCode::Conflict, "This pull request changed or is already being published."));
1842	}
1843	self.publish(
1844	"pull.converted_to_draft",
1845	&pull.repo_id,
1846	&a.actor,
1847	Self::pull_event(&pull),
1848	)
1849	.await?;
1850	self.note(
1851	&pull.repo_id,
1852	pull.number,
1853	&a.actor,
1854	"marked this as a draft",
1855	)
1856	.await?;
1857	if self
1858	.leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was marked as a draft."))
1859	.await?
1860	{
1861	self.publish_as(
1862	"queue.changed",
1863	&pull.repo_id,
1864	None,
1865	g1t_contracts::events::QueueChanged {
1866	repo_id: pull.repo_id.clone(),
1867	},
1868	)
1869	.await?;
1870	}
1871	pull.status = PullStatus::Draft;
1872	pull.updated_at = now;
1873	Ok(Outcome::Ok(pull))
1874	}
1875
1876	/// Lands the pull request on the branch it merges into, by the method
1877	/// asked for or the repository's default (merging.rs). Unless told to
1878	/// keep them open, a merge into the default branch lands the work items
1879	/// it closes in this repository: its own and those it names
1880	/// (closing.rs), each closing once nothing else is left to land, and
1881	/// the others still in progress here for its own close as superseded.
1882	async fn merge_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1883	let viewer = Some(a.actor.clone());
1884	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
1885	check!(writable(&repo));
1886	match pull.status {
1887	PullStatus::Open => {}
1888	PullStatus::Draft => {
1889	return Ok(Outcome::fail(
1890	FailureCode::Conflict,
1891	"This pull request is still a draft. Mark it ready for review first.",
1892	));
1893	}
1894	status => {
1895	return Ok(Outcome::fail(
1896	FailureCode::Conflict,
1897	format!("This pull request is already {}.", status.as_str()),
1898	));
1899	}
1900	}
1901	// Whoever looked at one head merges that one.
1902	if let Some(expected) = a.expected_head.as_deref().map(str::trim).filter(\|head\| !head.is_empty())
1903	&& pull.head_commit.as_deref() != Some(expected)
1904	{
1905	return Ok(Outcome::fail(
1906	FailureCode::Conflict,
1907	"This pull request has new commits since you looked. Review them, then merge again.",
1908	));
1909	}
1910	let base = pull.base_branch(&repo.default_branch).to_owned();
1911	// How it merges: as asked, or the repository's default, where the
1912	// repository and the rules for its base allow it.
1913	let (choosing, requirements) = self.settings_and_requirements(&repo, &pull).await?;
1914	let method = match merging::choose_method(a.merge_method, &choosing, &requirements, &base) {
1915	Ok(method) => method,
1916	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1917	};
1918	// The rules of the branch it merges into, as they stack: the one
1919	// gate for the merge button, the API, MCP, auto-merge and the queue,
1920	// for a person's pull request and an agent's alike. A bypass counts
1921	// when the merger asks for it, or for g1t when a ruleset lists it.
1922	let gate = self.merge_gate_by(&repo, &pull, Some(&a.actor), a.ignore_checks, true, method).await?;
1923	let bypassable = gate.bypassable();
1924	let gate = if a.bypass_rules \|\| a.actor.is_system() { gate } else { gate.without_bypass() };
1925	let settings = rulesets::overlay(self.settings(&repo.id).await?, &gate.requirements, base == repo.default_branch);
1926	if pull.check_status == Some(CheckStatus::Failed) && !(a.ignore_checks && settings.allow_ignoring_checks) {
1927	return Ok(Outcome::fail(
1928	FailureCode::Conflict,
1929	"It failed in the merge queue; push a fix to try again.",
1930	));
1931	}
1932	if let Some(refusal) = gate.refusal() {
1933	if access::can(Some(&a.actor), &repo, Capability::Merge) {
1934	self.record_merge_evaluations(&repo, &pull, &gate).await;
1935	}
1936	let offer = if bypassable && !a.bypass_rules {
1937	" You may bypass these rules: merge again and ask to bypass them (bypass_rules)."
1938	} else {
1939	""
1940	};
1941	return Ok(Outcome::fail(FailureCode::Conflict, format!("{refusal}{offer}")));
1942	}
1943	// Known ahead of time to conflict: neither a merge nor the queue
1944	// would get through, so say what has to be resolved now.
1945	if let Some(files) = self.conflicting_files(&pull).await? {
1946	let named = if files.is_empty() {
1947	String::new()
1948	} else {
1949	format!(" in {}", files.join(", "))
1950	};
1951	return Ok(Outcome::fail(
1952	FailureCode::Conflict,
1953	format!(
1954	"This branch has conflicts with {base}{named} that must be resolved first. Merge {base} into it, fix them and push."
1955	),
1956	));
1957	}
1958
1959	// A repository that merges through a queue: it joins the queue, and
1960	// lands once its state together with everything ahead has passed.
1961	// Its state is built by the method asked for, else the merge queue
1962	// rule's, and lands exactly as tested (queue.rs).
1963	if settings.merge_queue {
1964	if !a.actor.verified {
1965	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1966	}
1967	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
1968	let queued = match a.merge_method {
1969	Some(_) => method,
1970	None => {
1971	let rule = self.queue_rule(&repo.id).await?;
1972	match merging::choose_method(Some(rule.merge_method), &choosing, &requirements, &base) {
1973	Ok(method) => method,
1974	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1975	}
1976	}
1977	};
1978	self.record_merge_evaluations(&repo, &pull, &gate).await;
1979	// Landing honours the same bypass request against the rules as
1980	// they are then (queue.rs `settle`).
1981	let message = (a.commit_title.as_deref(), a.commit_message.as_deref());
1982	return self.enqueue(&repo, &pull, &a.actor, a.keep_item_open, a.bypass_rules, queued, message).await;
1983	}
1984
1985	// The commit it makes, for a merge or a squash: what the merger
1986	// wrote, over the repository's default.
1987	let message = match self
1988	.merge_message_for(&repo, &pull, &settings, method, a.commit_title.as_deref(), a.commit_message.as_deref())
1989	.await?
1990	{
1991	Ok(message) => message,
1992	Err(problem) => return Ok(Outcome::fail(FailureCode::Invalid, problem)),
1993	};
1994
1995	// The branch it merges into has moved under it. A merge, squash or
1996	// rebase makes new commits on its head anyway; a fast-forward needs
1997	// it brought up to date first, and lands when that is done. Unless
1998	// the repository insists on that being dealt with first.
1999	let behind = self.is_behind(&pull).await?;
2000	match merging::when_behind(method, behind, settings.require_up_to_date, &base) {
2001	merging::WhenBehind::Land => {}
2002	merging::WhenBehind::Refuse(refusal) => return Ok(Outcome::fail(FailureCode::Conflict, refusal)),
2003	merging::WhenBehind::CatchUpFirst => {
2004	if !a.actor.verified {
2005	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
2006	}
2007	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
2008	self.record_merge_evaluations(&repo, &pull, &gate).await;
2009	// Brought up to date here, without a sandbox: the merge is
2010	// recorded first, and lands once the push that catches it up
2011	// arrives and the required checks pass on that commit
2012	// (lifecycle.rs `land_if_requested`, `land_when_checked`).
2013	// Merging past failing checks, where that is allowed, carries
2014	// over: it then lands without waiting for them.
2015	self.request_landing(
2016	&pull,
2017	&a.actor,
2018	a.keep_item_open,
2019	a.delete_branch,
2020	a.ignore_checks && settings.allow_ignoring_checks,
2021	)
2022	.await?;
2023	// Bounded: a repository that does not answer gets a clear
2024	// failure back to whoever clicked, not a request that hangs
2025	// until the site gives up on it.
2026	let updated: Option<Outcome<PullBranchUpdate>> = g1t_kit::call_within(
2027	&self.repos,
2028	"update_pull_branch",
2029	&UpdatePullBranchArgs {
2030	source_id: pull.source_id().to_owned(),
2031	branch: pull.branch.clone(),
2032	number: pull.number,
2033	actor: a.actor.clone(),
2034	target_branch: pull.base.clone(),
2035	target_id: pull.target_id(),
2036	// Linear history takes no merge commit on the way.
2037	method: if requirements.linear_history { UpdateMethod::Rebase } else { UpdateMethod::Merge },
2038	expected_head: pull.head_commit.clone(),
2039	},
2040	lifecycle::REPOS_WRITE_WITHIN,
2041	)
2042	.await?;
2043	// A push that still arrives later only brings it up to date.
2044	let updated = updated.unwrap_or_else(lifecycle::repository_unreachable);
2045	match updated {
2046	Outcome::Ok(PullBranchUpdate::Updated { .. }) => return Ok(Outcome::Ok(pull)),
2047	// Moved again in between, or already caught up: land it now.
2048	Outcome::Ok(PullBranchUpdate::UpToDate { .. }) => self.forget_landing(&pull.id).await?,
2049	Outcome::Ok(PullBranchUpdate::NeedsAgent { paths, .. }) => {
2050	self.forget_landing(&pull.id).await?;
2051	let named = if paths.is_empty() { String::new() } else { format!(" ({})", paths.join(", ")) };
2052	return Ok(Outcome::fail(
2053	FailureCode::Conflict,
2054	format!(
2055	"{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."
2056	),
2057	));
2058	}
2059	// Only those who can push where its branch is can catch
2060	// it up; anyone may merge it another way.
2061	Outcome::Fail(failure) if failure.code == FailureCode::Forbidden => {
2062	self.forget_landing(&pull.id).await?;
2063	let source = pull.source_path(&a.repo);
2064	return Ok(Outcome::fail(
2065	FailureCode::Conflict,
2066	merging::cannot_fast_forward(&base, &format!("{}/{}", source.namespace, source.name)),
2067	));
2068	}
2069	Outcome::Fail(failure) => {
2070	self.forget_landing(&pull.id).await?;
2071	return Ok(Outcome::Fail(failure));
2072	}
2073	}
2074	}
2075	}
2076
2077	// Whether the actor may write to the repository is decided by repos.
2078	let head = pull.head_commit.clone();
2079	// Bounded like the catch-up above. Landing the same checked commit
2080	// again is recognised as done (repos `land`), so trying again after
2081	// a land that finished late records it.
2082	let landed: Option<Outcome<Landed>> = g1t_kit::call_within(
2083	&self.repos,
2084	"land",
2085	&LandArgs {
2086	// A pull request from a branch lands from the repository
2087	// itself, or from the fork the branch is in.
2088	source_id: pull.source_id().to_owned(),
2089	branch: pull.branch.clone(),
2090	actor: a.actor.clone(),
2091	target_branch: Some(base.clone()),
2092	expected_commit: head.clone(),
2093	target_id: pull.target_id(),
2094	method: Some(method),
2095	message,
2096	author: merging::squash_author(&pull, method),
2097	sign_off: settings.require_web_commit_sign_off,
2098	pull: Some(pull.number),
2099	},
2100	lifecycle::REPOS_WRITE_WITHIN,
2101	)
2102	.await?;
2103	let landed = check!(landed.unwrap_or_else(lifecycle::repository_unreachable));
2104	self.record_merge_evaluations(&repo, &pull, &gate).await;
2105	let merged = self.record_merge(&repo, pull, &a.actor, a.keep_item_open, landed, None).await?;
2106	// Its branch, where the merger or the repository says it goes.
2107	if merged.status == PullStatus::Merged {
2108	self.delete_branch_after_merge(&repo, &merged, &a.actor, a.delete_branch, head).await?;
2109	}
2110	Ok(Outcome::Ok(self.pull_by_id(&merged.id).await?.unwrap_or(merged)))
2111	}
2112
2113	/// Records a pull request as merged once the default branch holds it:
2114	/// lands the work items it closes (closing.rs), links those it is part
2115	/// of, supersedes the others for its own work item in this repository,
2116	/// and says so. `pushed` names the branch when a push brought its head
2117	/// in, rather than a merge from the site or the queue (pushed.rs).
2118	pub(crate) async fn record_merge(
2119	&self,
2120	repo: &Repo,
2121	mut pull: Pull,
2122	actor: &User,
2123	keep_item_open: bool,
2124	landed: Landed,
2125	pushed: Option<&str>,
2126	) -> Result<Pull> {
2127	// Only a merge into the default branch lands work: into another
2128	// branch, it has not landed yet. Its own work item and those it
2129	// names land, where the repository closes them that way.
2130	let settings = self.settings(&repo.id).await?;
2131	// Only what whoever merges it, and whoever it is for, may close: an
2132	// outside collaborator's "Fixes #N" closes no work item they cannot see.
2133	let owner = self.pull_owner(&pull).await?;
2134	let mut who: Vec<&User> = vec![actor];
2135	who.extend(owner.as_ref().filter(\|owner\| owner.id != actor.id));
2136	let named = self.items_to_close(&pull, repo, &settings, keep_item_open, &who).await?;
2137	let own = self.pull_item_id(&pull.id).await?;
2138	// A fast-forward leaves the base at what was landed, which is its
2139	// head; the other methods leave its head as it was.
2140	let head = if landed.method == g1t_contracts::rules::MergeMethod::FastForward {
2141	Some(landed.commit.clone())
2142	} else {
2143	pull.head_commit.clone().or_else(\|\| Some(landed.commit.clone()))
2144	};
2145	let now = rfc3339(now_ms());
2146	let mut statements = vec![self
2147	.db
2148	.prepare(
2149	"UPDATE pulls
2150	SET status = 'merged', head_commit = ?, merge_base = ?, merged_by = ?,
2151	merged_at = ?, updated_at = ?, merge_method = ?, merge_commit = ?,
2152	auto_merge = NULL, closes_items = ?, parts = ?
2153	WHERE id = ? AND status <> 'merged' RETURNING id AS value",
2154	)
2155	.bind(&[
2156	optional(&head),
2157	optional(&landed.previous),
2158	actor.username.as_str().into(),
2159	now.as_str().into(),
2160	now.as_str().into(),
2161	landed.method.as_str().into(),
2162	landed.commit.as_str().into(),
2163	serde_json::to_string(&named.close_ids())?.into(),
2164	serde_json::to_string(&named.part_ids())?.into(),
2165	pull.id.as_str().into(),
2166	])?];
2167	// The others for its own work item in this repository were other
2168	// ways to do the same thing.
2169	if let Some(own) = own.as_deref().filter(\|own\| named.close_ids().iter().any(\|id\| id == own)) {
2170	statements.push(
2171	self.db
2172	.prepare(
2173	"UPDATE pulls SET status = 'closed', superseded_by = ?, updated_at = ?
2174	WHERE item_id = ? AND repo_id = ? AND id != ? AND status IN ('draft', 'open')",
2175	)
2176	.bind(&[pull.number.into(), now.as_str().into(), own.into(), repo.id.as_str().into(), pull.id.as_str().into()])?,
2177	);
2178	}
2179	let recorded = self.db.batch(statements).await?;
2180	if recorded.first().map(\|result\| result.results::<rows::ValueRow>()).transpose()?.is_none_or(\|rows\| rows.is_empty()) {
2181	return Ok(self.pull_by_id(&pull.id).await?.unwrap_or(pull));
2182	}
2183
2184	self.publish(
2185	"pull.merged",
2186	&repo.id,
2187	actor,
2188	PullEvent {
2189	commit: Some(landed.commit.clone()),
2190	..Self::pull_event(&pull)
2191	},
2192	)
2193	.await?;
2194	let who = (actor.id.as_str(), actor.username.as_str());
2195	let note = match pushed {
2196	Some(branch) => format!("merged this by pushing {} to {branch}", landed.commit),
2197	None => merging::merged_note(landed.method).to_owned(),
2198	};
2199	self.note(&repo.id, pull.number, who, &note).await?;
2200	let landing = items::Landing { pull: Some(pull.number), commit: None, pushed_by: None };
2201	items::land(self, repo, &landing, actor, &named.close_ids(), &named.part_ids(), None, false).await?;
2202	let through_queue = self.landed_through_queue(&pull.id).await?;
2203	self.keep_decision_quietly(
2204	&repo.id,
2205	decisions::NewDecision {
2206	item: own.clone(),
2207	pull: Some(pull.number),
2208	session_id: None,
2209	actor: (!decisions::is_g1t(actor)).then_some(actor),
2210	kind: g1t_contracts::decisions::DecisionKind::Merge,
2211	confidence: decisions::merge_confidence(through_queue, actor),
2212	summary: match pushed {
2213	_ if through_queue => format!("#{} landed through the merge queue.", pull.number),
2214	Some(branch) => format!("{} merged #{} by pushing to {branch}.", actor.username, pull.number),
2215	None => format!("{} merged #{}.", actor.username, pull.number),
2216	},
2217	evidence: serde_json::json!({
2218	"commit": landed.commit,
2219	"previous": landed.previous,
2220	"method": landed.method,
2221	"merge_queue": through_queue,
2222	"pushed": pushed.is_some(),
2223	}),
2224	},
2225	)
2226	.await;
2227	pull.status = PullStatus::Merged;
2228	pull.head_commit = head;
2229	pull.merge_method = Some(landed.method);
2230	pull.merge_commit = Some(landed.commit.clone());
2231	pull.auto_merge = None;
2232	pull.closes = named.close_numbers();
2233	pull.parts = named.part_numbers();
2234	pull.merge_base = landed.previous;
2235	pull.merged_by = Some(actor.username.clone());
2236	pull.merged_at = Some(now.clone());
2237	pull.updated_at = now;
2238	Ok(pull)
2239	}
2240
2241	async fn list_active_pulls(&self, a: ViewerArgs) -> Result<Vec<ActivePull>> {
2242	let Some(viewer) = a.viewer else {
2243	return Ok(Vec::new());
2244	};
2245	// The pull requests and their work items, in one round trip: their
2246	// own, and those g1t made for them (Pull::owner).
2247	let author = [JsValue::from(viewer.id.as_str())];
2248	let found = self
2249	.timing
2250	.db(
2251	2,
2252	self.db.batch(vec![
2253	self.db
2254	.prepare(format!(
2255	"SELECT {PULL_COLUMNS} FROM pulls
2256	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open')
2257	ORDER BY updated_at DESC LIMIT 50"
2258	))
2259	.bind(&author)?,
2260	self.db
2261	.prepare(items::select(
2262	"work_items.id IN (
2263	SELECT item_id FROM pulls
2264	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open') AND item_id IS NOT NULL
2265	ORDER BY updated_at DESC LIMIT 50)",
2266	))
2267	.bind(&author)?,
2268	]),
2269	)
2270	.await?;
2271	let (Some(found), Some(item_rows)) = (found.first(), found.get(1)) else {
2272	return Ok(Vec::new());
2273	};
2274	let snapshots = found.results::<Snapshot>()?;
2275	let item_ids: Vec<Option<String>> = found.results::<PullItemRow>()?.into_iter().map(\|row\| row.item_id).collect();
2276	let pulls: Vec<Pull> = found.results::<PullRow>()?.into_iter().map(Pull::from).collect();
2277	let items: Vec<Held> = items::held_rows(item_rows)?;
2278	let items = &items;
2279	let viewer = &viewer;
2280	// Where each stands: the remembered assessment when there is one,
2281	// and worked out otherwise.
2282	try_join_all(pulls.into_iter().zip(snapshots).zip(item_ids).map(\|((pull, snapshot), item_id)\| async move {
2283	let item = item_id.and_then(\|id\| {
2284	items
2285	.iter()
2286	.find(\|held\| held.item.id == id && held.may(Some(viewer), Capability::Read))
2287	.map(\|held\| held.clone().shown(Some(viewer)))
2288	});
2289	// Every open pull request has a lifecycle (lifecycle.rs), kept
2290	// fresh by what moves it (`advance`, on each event that can);
2291	// anyone's draft but g1t's has none.
2292	let managed = lifecycle::made_by_g1t(&pull) && snapshot.managed != 0;
2293	let lifecycle = if pull.status == PullStatus::Draft && !managed {
2294	None
2295	} else if let (Some(stage), Some(detail)) = (snapshot.stage, snapshot.stage_detail) {
2296	Some(Lifecycle {
2297	stage,
2298	detail,
2299	revisions: snapshot.revisions,
2300	})
2301	} else {
2302	// Never worked out yet: once, and remembered.
2303	let behind = self.behind_now(&pull).await?;
2304	self.assess(&pull, &item, behind)
2305	.await?
2306	.map(\|(lifecycle, _)\| lifecycle)
2307	};
2308	Ok::<_, worker::Error>(ActivePull {
2309	pull,
2310	item,
2311	lifecycle,
2312	})
2313	}))
2314	.await
2315	}
2316
2317	// --- Sessions ----------------------------------------------------------
2318
2319	async fn append_session(&self, a: AppendSessionArgs) -> Result<Outcome<Appended>> {
2320	if a.entries.is_empty() {
2321	return Ok(Outcome::Ok(Appended { count: 0 }));
2322	}
2323	if a.entries.len() > MAX_ENTRY_BATCH {
2324	return Ok(Outcome::fail(
2325	FailureCode::Invalid,
2326	format!("Send at most {MAX_ENTRY_BATCH} entries at a time."),
2327	));
2328	}
2329	let viewer = Some(a.actor.clone());
2330	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
2331	if !pull.is_owned_by(&a.actor.id) {
2332	return Ok(Outcome::fail(
2333	FailureCode::Forbidden,
2334	"Only whoever opened a pull request, or asked g1t for it, can record its session.",
2335	));
2336	}
2337
2338	let now = rfc3339(now_ms());
2339	let count = a.entries.len() as u32;
2340	let mut statements = Vec::with_capacity(a.entries.len() + 1);
2341	for entry in a.entries {
2342	let kind = serde_json::to_value(entry.kind)?;
2343	let text = coordination::shown::clean(&entry.text.chars().take(MAX_ENTRY_CHARS).collect::<String>());
2344	// Each insert takes the next sequence number itself, so two
2345	// writers appending at once cannot collide.
2346	statements.push(
2347	self.db
2348	.prepare(
2349	"INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
2350	SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
2351	FROM session_entries WHERE pull_id = ?",
2352	)
2353	.bind(&[
2354	pull.id.as_str().into(),
2355	kind.as_str().unwrap_or("note").into(),
2356	text.into(),
2357	optional(&entry.tool),
2358	optional(&entry.commit.or_else(\|\| pull.head_commit.clone())),
2359	now.as_str().into(),
2360	pull.id.as_str().into(),
2361	])?,
2362	);
2363	}
2364	statements.push(
2365	self.db
2366	.prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
2367	.bind(&[now.as_str().into(), pull.id.as_str().into()])?,
2368	);
2369	self.db.batch(statements).await?;
2370	self.publish(
2371	"session.appended",
2372	&pull.repo_id,
2373	&a.actor,
2374	SessionAppended {
2375	pull_id: pull.id.clone(),
2376	repo_id: pull.repo_id.clone(),
2377	number: pull.number,
2378	count,
2379	},
2380	)
2381	.await?;
2382	Ok(Outcome::Ok(Appended { count }))
2383	}
2384
2385	async fn read_session(&self, a: ViewArgs) -> Result<Outcome<Vec<SessionEntry>>> {
2386	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
2387	let rows = self
2388	.db
2389	.prepare(
2390	"SELECT seq, kind, text, tool, \"commit\", at FROM session_entries
2391	WHERE pull_id = ? AND seq > ? ORDER BY seq LIMIT ?",
2392	)
2393	.bind(&[pull.id.into(), a.after_seq.into(), SESSION_PAGE.into()])?
2394	.all()
2395	.await?
2396	.results::<SessionRow>()?;
2397	Ok(Outcome::Ok(
2398	rows.into_iter().map(SessionEntry::from).collect(),
2399	))
2400	}
2401
2402	/// A push moves the head of the pull request it concerns: the one whose
2403	/// fork was pushed to, or the one opened from the branch that moved.
2404	async fn on_event(&self, event: &Event) -> Result<()> {
2405	// A new repository: where it is, for its workspace's work items and
2406	// numbers; a workspace's first starts it with the default labels.
2407	if event.kind == "repo.created" {
2408	return items::repo_created(self, &event.data).await;
2409	}
2410	if event.kind == "repo.restored" {
2411	return items::repo_restored(self, &event.data).await;
2412	}
2413	if event.kind == "repo.visibility_changed" \|\| event.kind == "repo.updated" {
2414	return items::repo_visibility(self, &event.data).await;
2415	}
2416	// Pull requests into the branch that became the default merge into
2417	// the default branch, which is stored as none.
2418	if event.kind == "repo.default_branch_changed"
2419	&& let (Some(repo_id), Some(to)) = (event.repo_id.as_deref(), event.data["to"].as_str())
2420	{
2421	self.db
2422	.prepare(
2423	"UPDATE pulls SET base_branch = NULL
2424	WHERE repo_id = ? AND base_branch = ? AND status IN ('draft', 'open')",
2425	)
2426	.bind(&[repo_id.into(), to.into()])?
2427	.run()
2428	.await?;
2429	return Ok(());
2430	}
2431	if event.kind != "git.push" {
2432	return Ok(());
2433	}
2434	let (Some(repo_id), Some(after), Some(git_ref)) = (
2435	event.repo_id.as_deref(),
2436	event.data["after"].as_str(),
2437	event.data["ref"].as_str(),
2438	) else {
2439	return Ok(());
2440	};
2441	let now = rfc3339(now_ms());
2442	// Who moved it, for rules about the most recent push.
2443	let pusher: JsValue = event.actor.as_deref().map_or(JsValue::NULL, Into::into);
2444	// The head moved, so whatever the checks said no longer applies, and
2445	// whatever step g1t was waiting on has been taken.
2446	let moved = "UPDATE pulls
2447	SET head_commit = ?, updated_at = ?, head_pushed_by = ?, head_pushed_at = ?, check_status = NULL, check_run_id = NULL,
2448	working_on = NULL, working_until = NULL, stalled = NULL";
2449	let active = "status IN ('draft', 'open') AND head_commit IS NOT ?";
2450	let returning = "RETURNING id, repo_id, number, status, author_id, author_name, requested_by_id, requested_by_name";
2451	let mut pulls: Vec<MovedRow> = Vec::new();
2452	// A fork carries its pull request on its default branch.
2453	if event.data["defaultBranch"].as_bool() == Some(true) {
2454	pulls.extend(
2455	self.db
2456	.prepare(format!(
2457	"{moved} WHERE fork_repo_id = ? AND {active} {returning}"
2458	))
2459	.bind(&[
2460	after.into(),
2461	now.as_str().into(),
2462	pusher.clone(),
2463	now.as_str().into(),
2464	repo_id.into(),
2465	after.into(),
2466	])?
2467	.all()
2468	.await?
2469	.results::<MovedRow>()?,
2470	);
2471	}
2472	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2473	// A branch of the repository, and a branch of a fork pull
2474	// requests into other repositories were opened from.
2475	pulls.extend(
2476	self.db
2477	.prepare(format!(
2478	"{moved} WHERE {} AND {active} {returning}",
2479	head_repo::ON_BRANCH
2480	))
2481	.bind(&[
2482	after.into(),
2483	now.as_str().into(),
2484	pusher.clone(),
2485	now.as_str().into(),
2486	repo_id.into(),
2487	repo_id.into(),
2488	branch.into(),
2489	after.into(),
2490	])?
2491	.all()
2492	.await?
2493	.results::<MovedRow>()?,
2494	);
2495	}
2496	// What each now changes, so overlaps show while the work is under way.
2497	let mut moved_pulls = Vec::new();
2498	for moved in &pulls {
2499	if let Some(mut pull) = self.pull_by_id(&moved.id).await? {
2500	pull.files = self.refresh_files(&pull).await?;
2501	// Owners of files it now changes are asked too.
2502	self.refresh_code_owners(&pull).await;
2503	// A new overlap with another pull request is announced.
2504	self.announce_overlaps(&pull, event.actor.clone()).await;
2505	// Auto-merge someone else turned on stops (automerge.rs).
2506	self.auto_merge_after_push(&pull, event.actor.as_deref()).await?;
2507	moved_pulls.push(pull);
2508	}
2509	}
2510	// The harness sessions that pushed the branch, or that its new
2511	// commits name in trailers, are tied to its pull requests.
2512	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2513	let commits = event.data["commits"].as_u64().and_then(\|n\| u32::try_from(n).ok());
2514	self.link_push(repo_id, branch, after, commits, &moved_pulls).await;
2515	}
2516	// A merge that was waiting for this push to bring it up to date.
2517	for moved in &pulls {
2518	self.land_if_requested(&moved.id).await?;
2519	}
2520	// The pull requests into the branch whose heads it brought in are
2521	// merged, and the work items its commits name landed (pushed.rs).
2522	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2523	self.after_branch_push(event, repo_id, branch, after).await;
2524	}
2525	// Whether each still merges cleanly, and, when a default branch
2526	// moved, every open pull request into it.
2527	let moved_ids: Vec<String> = pulls.iter().map(\|pull\| pull.id.clone()).collect();
2528	let into = match event.data["defaultBranch"].as_bool() {
2529	Some(true) => mergeability::Moved::DefaultBranch,
2530	_ => match git_ref.strip_prefix("refs/heads/") {
2531	Some(branch) => mergeability::Moved::Branch(branch),
2532	None => mergeability::Moved::Nothing,
2533	},
2534	};
2535	self.after_push(repo_id, into, &moved_ids).await;
2536	// A draft is announced when it is marked ready instead.
2537	for pull in pulls
2538	.into_iter()
2539	.filter(\|pull\| pull.status == PullStatus::Open)
2540	{
2541	let item = moved_pulls.iter().find(\|moved\| moved.id == pull.id).and_then(\|moved\| moved.item);
2542	self.publish_as(
2543	"pull.updated",
2544	&pull.repo_id,
2545	event.actor.clone(),
2546	PullEvent {
2547	author: Some(g1t_contracts::credentials::Principal { id: pull.author_id, username: pull.author_name }),
2548	requested_by: pull
2549	.requested_by_id
2550	.zip(pull.requested_by_name)
2551	.map(\|(id, username)\| g1t_contracts::credentials::Principal { id, username }),
2552	pull_id: pull.id,
2553	repo_id: pull.repo_id.clone(),
2554	number: pull.number,
2555	item,
2556	commit: Some(after.to_owned()),
2557	..PullEvent::default()
2558	}
2559	.carrying(&event.data),
2560	)
2561	.await?;
2562	}
2563	Ok(())
2564	}
2565	}
2566
2567	/// An event's data that carries on what caused the event it follows from.
2568	trait Carrying: serde::Serialize + Sized {
2569	/// As JSON, marked as a workflow job's doing when `cause` was.
2570	fn carrying(self, cause: &serde_json::Value) -> serde_json::Value {
2571	g1t_contracts::events::carried(self, cause)
2572	}
2573	}
2574
2575	impl Carrying for PullEvent {}
2576
2577	fn service(env: &Env) -> Result<Work> {
2578	Ok(Work {
2579	db: env.d1("DB")?,
2580	identity: env.service("IDENTITY")?,
2581	repos: env.service("REPOS")?,
2582	events: env.service("EVENTS")?,
2583	actions: env.service("ACTIONS")?,
2584	integrations: env.service("INTEGRATIONS").ok(),
2585	timing: g1t_kit::d1::Timing::default(),
2586	prefetched: std::cell::RefCell::new(None),
2587	known_repos: std::cell::RefCell::new(std::collections::HashMap::new()),
2588	})
2589	}
2590
2591	#[event(fetch)]
2592	async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
2593	let Some(method) = rpc_method(&request) else {
2594	return Response::error("Not found", 404);
2595	};
2596	// A replica near the caller when it asks for one (crates/kit/src/d1.rs).
2597	let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
2598	let body: serde_json::Value = request.json().await?;
2599	let mut work = service(&env)?;
2600	work.db = db;
2601
2602	let answered = match method.as_str() {
2603	// Work items (items/) and moving issues into them (items/migration.rs).
2604	"open_work_item" => reply(&work.open_work_item(args(body)?).await?),
2605	"list_work_items" => reply(&work.list_work_items(args(body)?).await?),
2606	"get_work_item" => reply(&work.get_work_item(args(body)?).await?),
2607	"update_work_item" => reply(&work.update_work_item(args(body)?).await?),
2608	"close_work_item" => reply(&work.close_work_item(args(body)?).await?),
2609	"reopen_work_item" => reply(&work.reopen_work_item(args(body)?).await?),
2610	"link_work_item" => reply(&work.link_work_item(args(body)?).await?),
2611	"unlink_work_item" => reply(&work.unlink_work_item(args(body)?).await?),
2612	"resolve_ref" => reply(&work.resolve_ref(args(body)?).await?),
2613	"legacy_item_map" => reply(&work.legacy_item_map(args(body)?).await?),
2614	"migrate_work_items" => {
2615	let _: MigrateWorkItemsArgs = args(body)?;
2616	reply(&work.migrate_work_items().await?)
2617	}
2618	"list_labels" => reply(&work.list_labels(args(body)?).await?),
2619	"save_label" => reply(&work.save_label(args(body)?).await?),
2620	"delete_label" => reply(&work.delete_label(args(body)?).await?),
2621	"add_default_labels" => reply(&work.add_default_labels(args(body)?).await?),
2622	"set_labels" => reply(&work.set_labels(args(body)?).await?),
2623	"list_milestones" => reply(&work.list_milestones(args(body)?).await?),
2624	"get_milestone" => reply(&work.get_milestone(args(body)?).await?),
2625	"save_milestone" => reply(&work.save_milestone(args(body)?).await?),
2626	"delete_milestone" => reply(&work.delete_milestone(args(body)?).await?),
2627	"counts" => reply(&work.counts(args(body)?).await?),
2628	"add_comment" => reply(&work.add_comment(args(body)?).await?),
2629	"upload_attachment" => reply(&work.upload_attachment(&env, args(body)?).await?),
2630	"edit_comment" => reply(&work.edit_comment(args(body)?).await?),
2631	"delete_comment" => reply(&work.delete_comment(args(body)?).await?),
2632	"react" => reply(&work.react(args(body)?).await?),
2633	"commit_comments" => reply(&work.commit_comments(args(body)?).await?),
2634	"closed_by_commit" => reply(&work.closed_by_commit(args(body)?).await?),
2635	"add_commit_comment" => reply(&work.add_commit_comment(args(body)?).await?),
2636	"edit_commit_comment" => reply(&work.edit_commit_comment(args(body)?).await?),
2637	"delete_commit_comment" => reply(&work.delete_commit_comment(args(body)?).await?),
2638	"seen_checks" => reply(&work.seen_checks(args(body)?).await?),
2639	"report_checks" => reply(&work.report_checks(args(body)?).await?),
2640	"set_commit_status" => reply(&work.set_commit_status(args(body)?).await?),
2641	"create_commit_status" => reply(&work.create_commit_status(args(body)?).await?),
2642	"commit_statuses" => reply(&work.commit_statuses(args(body)?).await?),
2643	"combined_status" => reply(&work.combined_status(args(body)?).await?),
2644	"create_check_run" => reply(&work.create_check_run(args(body)?).await?),
2645	"update_check_run" => reply(&work.update_check_run(args(body)?).await?),
2646	"get_check_run" => reply(&work.get_check_run(args(body)?).await?),
2647	"check_run_annotations" => reply(&work.check_run_annotations(args(body)?).await?),
2648	"ref_check_runs" => reply(&work.ref_check_runs(args(body)?).await?),
2649	"ref_check_suites" => reply(&work.ref_check_suites(args(body)?).await?),
2650	"get_check_suite" => reply(&work.get_check_suite(args(body)?).await?),
2651	"rerequest_check_run" => reply(&work.rerequest_check_run(args(body)?).await?),
2652	"rerequest_check_suite" => reply(&work.rerequest_check_suite(args(body)?).await?),
2653	"request_check_action" => reply(&work.request_check_action(args(body)?).await?),
2654	"commit_checks" => reply(&work.commit_checks(args(body)?).await?),
2655	"advance" => reply(&work.advance(args(body)?).await?),
2656	"managed_pulls" => reply(&work.managed_pulls(args(body)?).await?),
2657	"queue" => reply(&work.queue(args(body)?).await?),
2658	"queue_build" => reply(&work.queue_build(args(body)?).await?),
2659	"report_queue" => reply(&work.report_queue(args(body)?).await?),
2660	"remove_from_queue" => reply(&work.remove_from_queue(args(body)?).await?),
2661	"locate_pull" => reply(&work.locate_pull(args(body)?).await?),
2662	"inbox_subject" => reply(&work.inbox_subject(args(body)?).await?),
2663	"get_settings" => reply(&work.get_settings(args(body)?).await?),
2664	"codeowners_errors" => reply(&work.codeowners_errors(args(body)?).await?),
2665	"update_settings" => reply(&work.update_settings(args(body)?).await?),
2666	"open_pull" => reply(&work.open_pull(args(body)?).await?),
2667	"list_pulls" => reply(&work.list_pulls(args(body)?).await?),
2668	"delivery_insights" => reply(&work.delivery_insights(args(body)?).await?),
2669	"pulls_for_repos" => reply(&work.pulls_for_repos(args(body)?).await?),
2670	"get_pull" => reply(&work.get_pull(args(body)?).await?),
2671	"update_pull" => reply(&work.update_pull(args(body)?).await?),
2672	"catch_up_pull" => reply(&work.catch_up_pull(args(body)?).await?),
2673	"ready_pull" => reply(&work.ready_pull(args(body)?).await?),
2674	"close_pull" => reply(&work.close_pull(args(body)?).await?),
2675	"reopen_pull" => reply(&work.reopen_pull(args(body)?).await?),
2676	"convert_pull_to_draft" => reply(&work.convert_pull_to_draft(args(body)?).await?),
2677	"merge_pull" => reply(&work.merge_pull(args(body)?).await?),
2678	// How it may merge, auto-merge and its branch (merging.rs, automerge.rs).
2679	"merge_choices" => reply(&work.merge_choices(args(body)?).await?),
2680	"set_auto_merge" => reply(&work.set_auto_merge(args(body)?).await?),
2681	"delete_head_branch" => reply(&work.delete_head_branch(args(body)?).await?),
2682	"restore_head_branch" => reply(&work.restore_head_branch(args(body)?).await?),
2683	"list_active_pulls" => reply(&work.list_active_pulls(args(body)?).await?),
2684	"by_author" => reply(&work.by_author(args(body)?).await?),
2685	"contributions" => reply(&work.contributions(args(body)?).await?),
2686	"append_session" => reply(&work.append_session(args(body)?).await?),
2687	"read_session" => reply(&work.read_session(args(body)?).await?),
2688	// Decision records (decisions.rs).
2689	"record_decision" => reply(&work.record_decision(args(body)?).await?),
2690	"list_decisions" => reply(&work.list_decisions(args(body)?).await?),
2691	// Coordinating work: harness sessions, claims, dependencies and a
2692	// work item's whole context (coordination/).
2693	"start_session" => reply(&work.start_session(args(body)?).await?),
2694	"update_session" => reply(&work.update_session(args(body)?).await?),
2695	"end_session" => reply(&work.end_session(args(body)?).await?),
2696	"title_session" => reply(&work.title_session(args(body)?).await?),
2697	"get_work_session" => reply(&work.get_work_session(args(body)?).await?),
2698	"list_work_sessions" => reply(&work.list_work_sessions(args(body)?).await?),
2699	"list_workspace_sessions" => reply(&work.list_workspace_sessions(args(body)?).await?),
2700	"get_workspace_session" => reply(&work.get_workspace_session(args(body)?).await?),
2701	"pull_sessions" => reply(&work.pull_sessions(args(body)?).await?),
2702	"record_session_entries" => reply(&work.record_session_entries(args(body)?).await?),
2703	"claim_work_item" => reply(&work.claim_work_item(args(body)?).await?),
2704	"release_claim" => reply(&work.release_claim(args(body)?).await?),
2705	"hand_off_work_item" => reply(&work.hand_off_work_item(args(body)?).await?),
2706	"list_claims" => reply(&work.list_claims(args(body)?).await?),
2707	"get_work" => reply(&work.get_work(args(body)?).await?),
2708	// Runs of sandboxes (runs.rs).
2709	"open_run" => reply(&work.open_run(args(body)?).await?),
2710	"report_run" => reply(&work.report_run(args(body)?).await?),
2711	"list_runs" => reply(&work.list_runs(args(body)?).await?),
2712	"get_run" => reply(&work.get_run(args(body)?).await?),
2713	"list_sessions" => reply(&work.list_sessions(args(body)?).await?),
2714	"get_session" => reply(&work.get_session(args(body)?).await?),
2715	// What a sandbox may reach (guardrails.rs).
2716	"get_guardrails" => reply(&work.get_guardrails(args(body)?).await?),
2717	"update_guardrails" => reply(&work.update_guardrails(args(body)?).await?),
2718	"run_guardrails" => reply(&work.run_guardrails(args(body)?).await?),
2719	// The runs whose sandboxes stop with their repository (retired.rs).
2720	"runs_in_repo" => reply(&work.runs_in_repo(args(body)?).await?),
2721	"start_mergecheck" => reply(&work.start_mergecheck(args(body)?).await?),
2722	"report_mergecheck" => reply(&work.report_mergecheck(args(body)?).await?),
2723	// Rulesets (rulesets.rs): kept here, enforced here on merge and by
2724	// repos on push.
2725	"list_rulesets" => reply(&work.list_rulesets(args(body)?).await?),
2726	"get_ruleset" => reply(&work.get_ruleset(args(body)?).await?),
2727	"save_ruleset" => reply(&work.save_ruleset(args(body)?).await?),
2728	"delete_ruleset" => reply(&work.delete_ruleset(args(body)?).await?),
2729	"effective_rules" => reply(&work.effective_rules(args(body)?).await?),
2730	"rule_evaluations" => reply(&work.rule_evaluations(args(body)?).await?),
2731	"ref_rules" => reply(&work.ref_rules(args(body)?).await?),
2732	"record_evaluations" => reply(&work.record_evaluations(args(body)?).await?),
2733	"set_requires_pull_request" => reply(&work.set_requires_pull_request(args(body)?).await?),
2734	_ => Response::error("Unknown method", 404),
2735	};
2736	served.finish_timed(answered, &work.timing)
2737	}
2738
2739	/// The cron trigger: moves on the job that moves issues into work items
2740	/// (items/migration.rs), until it is done, and settles the sessions'
2741	/// titles that are due (coordination/titles.rs).
2742	#[event(scheduled)]
2743	async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
2744	let moved = async {
2745	let work = service(&env)?;
2746	work.migrate_work_items().await
2747	};
2748	if let Err(error) = moved.await {
2749	worker::console_error!("moving issues into work items: {error}");
2750	}
2751	let settled = async {
2752	let work = service(&env)?;
2753	work.settle_titles().await
2754	};
2755	if let Err(error) = settled.await {
2756	worker::console_error!("settling session titles: {error}");
2757	}
2758	}
2759
2760	/// Events from the bus, delivered on this service's own queue.
2761	#[event(queue)]
2762	async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
2763	let work = service(&env)?;
2764	for message in batch.messages()? {
2765	// A workspace renamed: its runs, work items, labels, milestones and
2766	// numbers move to the slug it has now.
2767	let renamed = [runs::RENAMED, guardrails::RENAMED, rulesets::RENAMED, items::RENAMED, attachments::RENAMED].concat();
2768	if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), &renamed).await? {
2769	message.ack();
2770	continue;
2771	}
2772	// A repository renamed or transferred: its runs and guardrails
2773	// follow, and where it is is recorded for work items and numbers.
2774	let transferred = [runs::TRANSFERRED, guardrails::TRANSFERRED, head_repo::TRANSFERRED, items::TRANSFERRED].concat();
2775	if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), &transferred).await? {
2776	message.ack();
2777	continue;
2778	}
2779	// A workspace deleted: what it kept for itself goes, its work items'
2780	// attachments' bytes first.
2781	if message.body().kind == "workspace.deleted"
2782	&& let Some(slug) = message.body().data["slug"].as_str()
2783	{
2784	attachments::purge_workspace(&env, &work.db, slug).await?;
2785	}
2786	let deleted = [guardrails::DELETED, items::DELETED, attachments::DELETED].concat();
2787	if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), &deleted).await? {
2788	message.ack();
2789	continue;
2790	}
2791	// An account purged: what it wrote shows as ghost (ghost.rs).
2792	let ghost = ghost::statements();
2793	let ghost: Vec<&str> = ghost.iter().map(String::as_str).collect();
2794	if g1t_kit::user_deleted::on_event(&env.d1("DB")?, message.body(), &ghost).await? {
2795	message.ack();
2796	continue;
2797	}
2798	// A repository purged: its attachments' bytes go before their rows do.
2799	if let Some(g1t_kit::lifecycle::Lifecycle::Purged(purged)) = g1t_kit::lifecycle::read(message.body()) {
2800	attachments::purge(&env, &work.db, &purged.repo_id).await?;
2801	}
2802	// A repository deleted, archived or purged, or a branch renamed (retired.rs).
2803	if work.on_retired(message.body()).await? {
2804	message.ack();
2805	continue;
2806	}
2807	work.on_event(message.body()).await?;
2808	message.ack();
2809	}
2810	Ok(())
2811	}
2812
2813	/// The rules that once read a pull request's author read its owner now:
2814	/// whoever asked g1t for it, or its author. For each, the person who asked
2815	/// is held to what an author was, and g1t's agent (a token it works with)
2816	/// gains nothing by being the author.
2817	#[cfg(test)]
2818	mod owner_rules {
2819	use super::*;
2820	use crate::rows::stored::{ASKER, G1T, pull};
2821	use g1t_contracts::identity::AGENT_ID;
2822
2823	const SOMEONE: &str = "usr_2";
2824
2825	#[test]
2826	fn no_self_approval() {
2827	// add_comment refuses a verdict on one that is theirs.
2828	let made = pull(G1T, Some(ASKER));
2829	assert!(made.is_owned_by(ASKER.0), "the person who asked cannot approve it");
2830	assert!(!made.is_owned_by(AGENT_ID), "g1t's review agent still gives its verdict");
2831	assert!(!made.is_owned_by(SOMEONE));
2832	}
2833
2834	#[test]
2835	fn what_an_author_could_do_without_a_role() {
2836	// manageable_pull (update, ready, close), catch_up_pull on a fork,
2837	// and append_session: theirs to do.
2838	let made = pull(G1T, Some(ASKER));
2839	assert!(made.is_owned_by(ASKER.0));
2840	assert!(!made.is_owned_by(SOMEONE), "anyone else still needs the role");
2841	assert!(!made.is_owned_by(AGENT_ID), "being its author gives g1t's tokens nothing more");
2842	}
2843
2844	#[test]
2845	fn nobody_is_asked_to_review_what_they_asked_for() {
2846	// update_pull drops the owner from the reviewers asked.
2847	let made = pull(G1T, Some(ASKER));
2848	let mut reviewers = vec!["syntaqx".to_owned(), "ana".to_owned()];
2849	reviewers.retain(\|name\| *name != made.owner().username);
2850	assert_eq!(reviewers, ["ana"]);
2851	}
2852
2853	#[test]
2854	fn sandboxes_act_as_whoever_asked() {
2855	// MergecheckJob and the merge queue's job
2856	// carry who the sandbox's credential acts for: a real account.
2857	let made = pull(G1T, Some(ASKER));
2858	assert_eq!(made.owner().id, ASKER.0);
2859	let acts_as = made.requested_by.unwrap_or(made.author);
2860	assert_eq!(acts_as.id, ASKER.0);
2861	// g1t's own work, which nobody asked for, acts as g1t, as before.
2862	let own = pull(("g1t", "g1t"), None);
2863	assert_eq!(own.requested_by.unwrap_or(own.author).id, "g1t");
2864	}
2865
2866	#[test]
2867	fn events_name_g1t_and_whoever_asked() {
2868	let made = pull(G1T, Some(ASKER));
2869	let event = serde_json::to_value(Work::pull_event(&made)).unwrap();
2870	assert_eq!(event["author"], serde_json::json!({ "id": AGENT_ID, "username": "g1t" }));
2871	assert_eq!(event["requestedBy"], serde_json::json!({ "id": "usr_1", "username": "syntaqx" }));
2872	let own = serde_json::to_value(Work::pull_event(&pull(ASKER, None))).unwrap();
2873	assert!(own.get("requestedBy").is_none());
2874	}
2875
2876	#[test]
2877	fn only_a_closed_pull_request_reopens_and_only_an_open_one_turns_draft() {
2878	assert_eq!(reopen_refusal(PullStatus::Closed), None);
2879	assert!(reopen_refusal(PullStatus::Merged).is_some(), "a merge cannot be undone");
2880	assert!(reopen_refusal(PullStatus::Open).is_some());
2881	assert!(reopen_refusal(PullStatus::Draft).is_some());
2882	assert_eq!(draft_refusal(PullStatus::Open), None);
2883	assert!(draft_refusal(PullStatus::Draft).is_some());
2884	assert!(draft_refusal(PullStatus::Closed).is_some());
2885	assert!(draft_refusal(PullStatus::Merged).is_some());
2886	}
2887
2888	#[test]
2889	fn a_pull_request_reopens_as_what_it_was_closed_as() {
2890	assert_eq!(reopened_status(Some("draft")), PullStatus::Draft);
2891	assert_eq!(reopened_status(Some("open")), PullStatus::Open);
2892	// Closed before this was recorded, or by a merge of another.
2893	assert_eq!(reopened_status(None), PullStatus::Open);
2894	}
2895	}