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