flagon-io/g1t

public

2,306 lines93,510 bytesCode Blame

1	//! The work service: issues, 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 authored;
8	mod capture;
9	mod checks;
10	mod compute;
11	mod confidence;
12	mod guardrails;
13	mod inbox;
14	mod lifecycle;
15	mod memory;
16	mod mentions;
17	mod mergeability;
18	mod plans;
19	mod messages;
20	mod prefetch;
21	mod queue;
22	mod retired;
23	mod reviews;
24	mod rows;
25	mod runs;
26	mod settings;
27	mod statuses;
28
29	use g1t_contracts::events::{
30	CommentCreated, Event, IssueEvent, NewEvent, Publish, PullEvent, SessionAppended,
31	};
32	use g1t_contracts::identity::UsernameArgs;
33	use g1t_contracts::repos::{
34	ForkArgs, GetArgs, HeadArgs, LandArgs, Landed, NeedsAgentReason, PullBranchUpdate, ReadableArgs, Repo, RepoPath,
35	UpdatePullBranchArgs,
36	};
37	use g1t_contracts::access::{self, Capability, Denied};
38	use g1t_contracts::time::rfc3339;
39	use g1t_contracts::work::*;
40	use futures_util::future::{try_join, try_join3, try_join_all};
41	use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
42	use g1t_kit::{args, now_ms, reply, rpc_method};
43	use serde::Serialize;
44	use worker::wasm_bindgen::JsValue;
45	use worker::{
46	Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, event,
47	};
48
49	use retired::writable;
50	use rows::{CommentRow, IssueRow, MovedRow, NumberRow, PULL_COLUMNS, PullRow, SessionRow, Snapshot, ValueRow};
51
52	const SOURCE: &str = "work";
53	const MAX_ENTRY_BATCH: usize = 200;
54	const MAX_ENTRY_CHARS: usize = 64_000;
55	const MAX_TITLE_CHARS: usize = 200;
56	const SESSION_PAGE: u32 = 500;
57	const LIST_PAGE: u32 = 100;
58	const MAX_ASSIGNEES: usize = 10;
59	const UNVERIFIED: &str = "Confirm your email address first. Check your inbox, or resend the link from the banner on g1t.sh.";
60
61	const ISSUE_COLUMNS: &str = "issues.*,
62	(SELECT count(*) FROM pulls WHERE pulls.issue_id = issues.id) AS pull_count,
63	(SELECT agent FROM pulls
64	WHERE pulls.issue_id = issues.id AND pulls.status IN ('draft', 'open')
65	AND pulls.fork_repo_id IS NOT NULL
66	ORDER BY pulls.number DESC LIMIT 1) AS agent,
67	(SELECT count(*) FROM comments
68	WHERE comments.repo_id = issues.repo_id AND comments.number = issues.number
69	AND comments.kind = 'comment') AS comment_count";
70
71	fn no_issue<T>() -> Outcome<T> {
72	Outcome::fail(FailureCode::NotFound, "Issue not found.")
73	}
74
75	fn no_pull<T>() -> Outcome<T> {
76	Outcome::fail(FailureCode::NotFound, "Pull request not found.")
77	}
78
79	/// Whether a pull request was behind when its mergeability was last
80	/// worked out, and for which pair of commits (mergeability.rs).
81	#[derive(serde::Deserialize)]
82	struct StoredBehind {
83	#[serde(default)]
84	behind: Option<u8>,
85	#[serde(default)]
86	mergeable_key: Option<String>,
87	}
88
89	impl StoredBehind {
90	/// The stored answer, if it was worked out for `head`.
91	fn for_head(&self, head: Option<&str>) -> Option<bool> {
92	let (worked_for, _) = self.mergeable_key.as_deref()?.split_once("..")?;
93	(Some(worked_for) == head).then_some(self.behind? != 0)
94	}
95	}
96
97	/// Refuses `actor` unless their role on `repo` has `capability`: not found
98	/// when they cannot read it, forbidden with the role it needs otherwise.
99	pub(crate) fn allowed(actor: Option<&User>, repo: &Repo, capability: Capability) -> Outcome<()> {
100	match access::check(actor, repo, capability) {
101	Ok(()) => Outcome::Ok(()),
102	Err(Denied::NotFound) => Outcome::fail(FailureCode::NotFound, "Repository not found."),
103	Err(Denied::Forbidden) => Outcome::fail(
104	FailureCode::Forbidden,
105	access::needs(capability, &format!("{}/{}", repo.namespace, repo.name)),
106	),
107	}
108	}
109
110	fn optional(value: &Option<String>) -> JsValue {
111	value.as_deref().map_or(JsValue::NULL, JsValue::from)
112	}
113
114	fn optional_number(value: Option<u32>) -> JsValue {
115	value.map_or(JsValue::NULL, JsValue::from)
116	}
117
118	/// The lowercase name a `State` is stored and sent as.
119	fn state_name(state: Option<State>) -> Option<&'static str> {
120	state.map(\|state\| match state {
121	State::Open => "open",
122	State::Closed => "closed",
123	})
124	}
125
126	/// A trimmed title, or why it cannot be used.
127	fn valid_title(title: &str) -> std::result::Result<&str, &'static str> {
128	let title = title.trim();
129	if title.is_empty() {
130	Err("A title is required.")
131	} else if title.chars().count() > MAX_TITLE_CHARS {
132	Err("That title is too long.")
133	} else {
134	Ok(title)
135	}
136	}
137
138	/// Unwraps an `Outcome`, returning its failure from the enclosing method.
139	macro_rules! check {
140	($outcome:expr) => {
141	match $outcome {
142	Outcome::Ok(value) => value,
143	Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
144	}
145	};
146	}
147
148	struct Work {
149	db: D1Database,
150	identity: Fetcher,
151	repos: Fetcher,
152	events: Fetcher,
153	/// GitHub Actions: runs a merge queue's `merge_group` workflows.
154	actions: Fetcher,
155	/// Where this request's time went, for its `Server-Timing`.
156	timing: g1t_kit::d1::Timing,
157	/// A pull request's rows read in one batch for this request
158	/// (prefetch.rs), which the helpers below read instead of the database.
159	prefetched: std::cell::RefCell<Option<std::rc::Rc<prefetch::Prefetched>>>,
160	}
161
162	impl Work {
163	async fn publish<T: Serialize>(
164	&self,
165	kind: &'static str,
166	repo_id: &str,
167	actor: &User,
168	data: T,
169	) -> Result<()> {
170	self.publish_as(kind, repo_id, Some(actor.id.clone()), data)
171	.await
172	}
173
174	/// A pull request's owner (whoever asked g1t for it, or its author) as
175	/// a viewer who can read its repository and source. Stored people carry
176	/// no memberships, so a private repository would otherwise look missing
177	/// to them. The membership given reads and nothing more: it is for
178	/// looking, never for acting.
179	pub(crate) async fn owner_viewer(&self, pull: &Pull) -> Result<Viewer> {
180	let path: Option<RepoPath> = g1t_kit::call(
181	&self.repos,
182	"path_by_id",
183	&g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
184	)
185	.await?;
186	let mut owner = pull.owner().clone();
187	if let Some(path) = path
188	&& !owner.is_member(&path.namespace.to_lowercase())
189	{
190	owner.workspaces.push(g1t_contracts::Membership {
191	base_permission: Some(access::BasePermission::Read),
192	..g1t_contracts::Membership::member(path.namespace.to_lowercase())
193	});
194	}
195	Ok(Some(owner))
196	}
197
198	/// Publishes an event caused by `actor`, or by g1t itself.
199	async fn publish_as<T: Serialize>(
200	&self,
201	kind: &'static str,
202	repo_id: &str,
203	actor: Option<String>,
204	data: T,
205	) -> Result<()> {
206	let event = NewEvent {
207	kind,
208	source: SOURCE,
209	repo_id: Some(repo_id.to_owned()),
210	actor,
211	data,
212	};
213	g1t_kit::call(
214	&self.events,
215	"publish",
216	&Publish {
217	events: vec![event],
218	},
219	)
220	.await
221	}
222
223	/// The repository, if the viewer may see it. Whether they may is
224	/// decided by the repos service.
225	async fn repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
226	self.timing
227	.rpc(g1t_kit::call(
228	&self.repos,
229	"get",
230	&GetArgs {
231	path: path.clone(),
232	viewer: viewer.clone(),
233	},
234	))
235	.await
236	}
237
238	/// The next number in the repository's sequence. Taking it is one
239	/// statement, so concurrent opens cannot be given the same number.
240	async fn next_number(&self, repo_id: &str) -> Result<u32> {
241	let row = self
242	.db
243	.prepare(
244	"INSERT INTO counters (repo_id, last) VALUES (?, 1)
245	ON CONFLICT (repo_id) DO UPDATE SET last = last + 1
246	RETURNING last AS n",
247	)
248	.bind(&[repo_id.into()])?
249	.first::<NumberRow>(None)
250	.await?;
251	row.map(\|row\| row.n)
252	.ok_or_else(\|\| worker::Error::RustError("no number was assigned".into()))
253	}
254
255	async fn issue(&self, repo_id: &str, number: u32) -> Result<Option<Issue>> {
256	Ok(self
257	.db
258	.prepare(format!(
259	"SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = ? AND number = ?"
260	))
261	.bind(&[repo_id.into(), number.into()])?
262	.first::<IssueRow>(None)
263	.await?
264	.map(Issue::from))
265	}
266
267	async fn pull(&self, repo_id: &str, number: u32) -> Result<Option<Pull>> {
268	Ok(self
269	.db
270	.prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? AND number = ?"))
271	.bind(&[repo_id.into(), number.into()])?
272	.first::<PullRow>(None)
273	.await?
274	.map(Pull::from))
275	}
276
277	/// The repository and one of its issues, as seen by `viewer`.
278	async fn issue_at(
279	&self,
280	path: &RepoPath,
281	number: u32,
282	viewer: &Viewer,
283	) -> Result<Outcome<(Repo, Issue)>> {
284	let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
285	return Ok(no_issue());
286	};
287	Ok(match self.issue(&repo.id, number).await? {
288	Some(issue) => Outcome::Ok((repo, issue)),
289	None => no_issue(),
290	})
291	}
292
293	/// The repository and one of its pull requests, as seen by `viewer`.
294	async fn pull_at(
295	&self,
296	path: &RepoPath,
297	number: u32,
298	viewer: &Viewer,
299	) -> Result<Outcome<(Repo, Pull)>> {
300	let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
301	return Ok(no_pull());
302	};
303	Ok(match self.pull(&repo.id, number).await? {
304	Some(pull) => Outcome::Ok((repo, pull)),
305	None => no_pull(),
306	})
307	}
308
309	/// Records something that happened to an issue or a pull request, so
310	/// that it shows in the conversation where it happened. `text` is what
311	/// `author` did, as the rest of a sentence starting with their name.
312	pub(crate) async fn note(
313	&self,
314	repo_id: &str,
315	number: u32,
316	author: (&str, &str),
317	text: &str,
318	) -> Result<()> {
319	let now = now_ms();
320	self.db
321	.prepare(
322	"INSERT INTO comments
323	(id, repo_id, number, author_id, author_name, body, kind, created_at)
324	VALUES (?, ?, ?, ?, ?, ?, 'event', ?)",
325	)
326	.bind(&[
327	new_id("cmt", now).into(),
328	repo_id.into(),
329	number.into(),
330	author.0.into(),
331	author.1.into(),
332	text.into(),
333	rfc3339(now).into(),
334	])?
335	.run()
336	.await?;
337	Ok(())
338	}
339
340	/// Notes who was added to and removed from a list of people, such as
341	/// "assigned ana" or "requested a review from g1t".
342	async fn note_changes(
343	&self,
344	repo_id: &str,
345	number: u32,
346	actor: &User,
347	before: &[String],
348	after: &[String],
349	(added, removed): (&str, &str),
350	) -> Result<()> {
351	let joined = \|names: Vec<&String>\| {
352	names
353	.into_iter()
354	.map(String::as_str)
355	.collect::<Vec<_>>()
356	.join(", ")
357	};
358	let new: Vec<&String> = after.iter().filter(\|name\| !before.contains(name)).collect();
359	let gone: Vec<&String> = before.iter().filter(\|name\| !after.contains(name)).collect();
360	let who = (actor.id.as_str(), actor.username.as_str());
361	if !new.is_empty() {
362	// Taking something on oneself reads better said that way.
363	let text = if added == "assigned" && new == [&actor.username] {
364	"self-assigned this".to_owned()
365	} else {
366	format!("{added} {}", joined(new))
367	};
368	self.note(repo_id, number, who, &text).await?;
369	}
370	if !gone.is_empty() {
371	self.note(repo_id, number, who, &format!("{removed} {}", joined(gone)))
372	.await?;
373	}
374	Ok(())
375	}
376
377	fn issue_event(issue: &Issue) -> IssueEvent {
378	IssueEvent {
379	issue_id: issue.id.clone(),
380	repo_id: issue.repo_id.clone(),
381	number: issue.number,
382	author: Some((&issue.author).into()),
383	requested_by: issue.requested_by.as_ref().map(Into::into),
384	..IssueEvent::default()
385	}
386	}
387
388	/// The commit a pull request's change is at in git right now: its
389	/// fork's default branch, or its branch.
390	async fn live_head(&self, pull: &Pull) -> Result<Option<String>> {
391	g1t_kit::call(
392	&self.repos,
393	"head",
394	&HeadArgs {
395	repo_id: pull.fork_repo_id.clone().unwrap_or_else(\|\| pull.repo_id.clone()),
396	branch: pull.branch.clone().unwrap_or_default(),
397	},
398	)
399	.await
400	}
401
402	fn pull_event(pull: &Pull) -> PullEvent {
403	PullEvent {
404	pull_id: pull.id.clone(),
405	repo_id: pull.repo_id.clone(),
406	number: pull.number,
407	author: Some((&pull.author).into()),
408	requested_by: pull.requested_by.as_ref().map(Into::into),
409	issue: pull.issue,
410	confidence: pull.confidence.clone(),
411	..PullEvent::default()
412	}
413	}
414
415	// --- Issues ------------------------------------------------------------
416
417	/// Opens an issue for g1t to take at once: refused before
418	/// anything is opened unless the actor may put agents to work here. The
419	/// runner's `delegate` starts the agent on it.
420	async fn delegate_issue(&self, a: DelegateIssueArgs) -> Result<Outcome<Issue>> {
421	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
422	check!(writable(&repo));
423	check!(allowed(Some(&a.actor), &repo, Capability::Run));
424	self.open_issue(OpenIssueArgs {
425	actor: a.actor,
426	repo: a.repo,
427	title: a.title,
428	body: a.body,
429	labels: a.labels,
430	checks: a.checks,
431	})
432	.await
433	}
434
435	async fn open_issue(&self, a: OpenIssueArgs) -> Result<Outcome<Issue>> {
436	if !a.actor.verified {
437	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
438	}
439	let title = match valid_title(&a.title) {
440	Ok(title) => title,
441	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
442	};
443	let Some(labels) = normalize_labels(&a.labels) else {
444	return Ok(Outcome::fail(
445	FailureCode::Invalid,
446	"An issue can have up to 10 labels of up to 40 characters each.",
447	));
448	};
449	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
450	check!(writable(&repo));
451	// Commands given the old way are words for the agent now: added to
452	// the body under "Definition of done". What has to pass to merge is
453	// the branch's required checks.
454	let body = with_definition_of_done(&a.body, &commands_pass(&a.checks));
455
456	let now = now_ms();
457	let id = new_id("iss", now);
458	let number = self.next_number(&repo.id).await?;
459	let timestamp = rfc3339(now);
460	// What g1t's agent files at work is g1t's, for the person it works for.
461	let (author, requested_by) = authorship(&a.actor, false);
462	self.db
463	.prepare(
464	"INSERT INTO issues
465	(id, repo_id, number, title, body, labels, checks, author_id, author_name,
466	requested_by_id, requested_by_name, created_at, updated_at)
467	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
468	)
469	.bind(&[
470	id.as_str().into(),
471	repo.id.as_str().into(),
472	number.into(),
473	title.into(),
474	body.into(),
475	serde_json::to_string(&labels)?.into(),
476	"[]".into(),
477	author.id.as_str().into(),
478	author.username.as_str().into(),
479	optional(&requested_by.as_ref().map(\|user\| user.id.clone())),
480	optional(&requested_by.as_ref().map(\|user\| user.username.clone())),
481	timestamp.as_str().into(),
482	timestamp.as_str().into(),
483	])?
484	.run()
485	.await?;
486	let Some(issue) = self.issue(&repo.id, number).await? else {
487	return Ok(no_issue());
488	};
489	self.apply_label_rule(&a.actor, &issue, &[]).await?;
490	self.publish(
491	"issue.opened",
492	&repo.id,
493	&a.actor,
494	IssueEvent {
495	title: Some(issue.title.clone()),
496	..Self::issue_event(&issue)
497	},
498	)
499	.await?;
500	Ok(Outcome::Ok(issue))
501	}
502
503	async fn list_issues(&self, a: ListIssuesArgs) -> Result<Outcome<Vec<Issue>>> {
504	let state = state_name(a.state);
505	let label = a
506	.label
507	.map(\|label\| label.trim().to_lowercase())
508	.filter(\|label\| !label.is_empty());
509	let list = \|repo_id: String\| {
510	let label = label.clone();
511	async move {
512	let state = state.map_or(JsValue::NULL, JsValue::from);
513	let query = self
514	.db
515	.prepare(format!(
516	"SELECT {ISSUE_COLUMNS} FROM issues
517	WHERE repo_id = ? AND (? IS NULL OR state = ?)
518	AND (? IS NULL OR EXISTS
519	(SELECT 1 FROM json_each(issues.labels) WHERE json_each.value = ?))
520	ORDER BY number DESC LIMIT ?"
521	))
522	.bind(&[
523	repo_id.into(),
524	state.clone(),
525	state,
526	optional(&label),
527	optional(&label),
528	LIST_PAGE.into(),
529	])?;
530	self.timing.db(1, query.all()).await?.results::<IssueRow>()
531	}
532	};
533	let (_, rows) = check!(self.repo_then(&a.repo, &a.viewer, list).await?);
534	Ok(Outcome::Ok(rows.into_iter().map(Issue::from).collect()))
535	}
536
537	/// An issue, the pull requests for it and its comments: one batch,
538	/// started beside the access check (prefetch.rs).
539	async fn get_issue(&self, a: ViewArgs) -> Result<Outcome<IssueDetail>> {
540	let number = a.number;
541	let read = \|repo_id: String\| async move {
542	let key = [JsValue::from(repo_id.as_str()), JsValue::from(number)];
543	let statements = vec![
544	self.db
545	.prepare(format!("SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = ?1 AND number = ?2"))
546	.bind(&key)?,
547	self.db
548	.prepare(format!(
549	"SELECT {PULL_COLUMNS} FROM pulls
550	WHERE issue_id = (SELECT id FROM issues WHERE repo_id = ?1 AND number = ?2)
551	ORDER BY number"
552	))
553	.bind(&key)?,
554	self.db
555	.prepare("SELECT * FROM comments WHERE repo_id = ?1 AND number = ?2 ORDER BY id LIMIT 500")
556	.bind(&key)?,
557	];
558	let results = self.timing.db(3, self.db.batch(statements)).await?;
559	let rows = \|index: usize\| results.get(index).ok_or_else(\|\| worker::Error::RustError("short batch".into()));
560	Ok::<_, worker::Error>((
561	rows(0)?.results::<IssueRow>()?.into_iter().next().map(Issue::from),
562	rows(1)?.results::<PullRow>()?.into_iter().map(Pull::from).collect::<Vec<_>>(),
563	rows(2)?.results::<CommentRow>()?.into_iter().map(Comment::from).collect::<Vec<_>>(),
564	))
565	};
566	let Outcome::Ok((_, (Some(issue), pulls, comments))) = self.repo_then(&a.repo, &a.viewer, read).await? else {
567	return Ok(no_issue());
568	};
569	Ok(Outcome::Ok(IssueDetail { comments, pulls, issue }))
570	}
571
572	/// The issue, if `actor` wrote it or may triage the repository's issues.
573	async fn manageable_issue(
574	&self,
575	actor: &User,
576	path: &RepoPath,
577	number: u32,
578	) -> Result<Outcome<Issue>> {
579	let (repo, issue) = check!(self.issue_at(path, number, &Some(actor.clone())).await?);
580	check!(writable(&repo));
581	if issue.owner().id != actor.id {
582	check!(allowed(Some(actor), &repo, Capability::Triage));
583	}
584	Ok(Outcome::Ok(issue))
585	}
586
587	async fn update_issue(&self, a: UpdateIssueArgs) -> Result<Outcome<Issue>> {
588	let issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
589	let title = match a.title.as_deref().map(valid_title) {
590	Some(Err(message)) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
591	Some(Ok(title)) => Some(title.to_owned()),
592	None => None,
593	};
594	let labels = match a.labels.as_deref().map(normalize_labels) {
595	Some(None) => {
596	return Ok(Outcome::fail(
597	FailureCode::Invalid,
598	"An issue can have up to 10 labels of up to 40 characters each.",
599	));
600	}
601	Some(Some(labels)) => Some(serde_json::to_string(&labels)?),
602	None => None,
603	};
604	let assignees = match a.assignees {
605	Some(names) => Some(check!(self.valid_assignees(names).await?)),
606	None => None,
607	};
608	let assigned = assignees.as_ref().map(serde_json::to_string).transpose()?;
609	let body = a.body.map(\|body\| body.trim().to_owned());
610	self.db
611	.prepare(
612	"UPDATE issues
613	SET title = COALESCE(?, title), body = COALESCE(?, body),
614	labels = COALESCE(?, labels), assignees = COALESCE(?, assignees),
615	updated_at = ?
616	WHERE id = ?",
617	)
618	.bind(&[
619	optional(&title),
620	optional(&body),
621	optional(&labels),
622	optional(&assigned),
623	rfc3339(now_ms()).into(),
624	issue.id.as_str().into(),
625	])?
626	.run()
627	.await?;
628	let before = issue.assignees.clone();
629	let labels_before = issue.labels.clone();
630	let Some(issue) = self.issue(&issue.repo_id, issue.number).await? else {
631	return Ok(no_issue());
632	};
633	self.apply_label_rule(&a.actor, &issue, &labels_before).await?;
634	self.publish(
635	"issue.updated",
636	&issue.repo_id,
637	&a.actor,
638	Self::issue_event(&issue),
639	)
640	.await?;
641	if let Some(assignees) = assignees {
642	self.note_changes(
643	&issue.repo_id,
644	issue.number,
645	&a.actor,
646	&before,
647	&assignees,
648	("assigned", "unassigned"),
649	)
650	.await?;
651	self.publish(
652	"issue.assigned",
653	&issue.repo_id,
654	&a.actor,
655	IssueEvent {
656	assignees: Some(assignees),
657	..Self::issue_event(&issue)
658	},
659	)
660	.await?;
661	}
662	Ok(Outcome::Ok(issue))
663	}
664
665	/// Usernames as given, tidied, if each names an account.
666	async fn valid_assignees(&self, names: Vec<String>) -> Result<Outcome<Vec<String>>> {
667	let mut assignees: Vec<String> = Vec::new();
668	for name in names {
669	let name = name.trim().trim_start_matches('@').to_lowercase();
670	if name.is_empty() \|\| assignees.contains(&name) {
671	continue;
672	}
673	if assignees.len() == MAX_ASSIGNEES {
674	return Ok(Outcome::fail(
675	FailureCode::Invalid,
676	format!("An issue can be assigned to at most {MAX_ASSIGNEES} people."),
677	));
678	}
679	let account: Viewer = g1t_kit::call(
680	&self.identity,
681	"user_by_username",
682	&UsernameArgs {
683	username: name.clone(),
684	},
685	)
686	.await?;
687	if account.is_none() {
688	return Ok(Outcome::fail(
689	FailureCode::Invalid,
690	format!("There is no account named {name}."),
691	));
692	}
693	assignees.push(name);
694	}
695	Ok(Outcome::Ok(assignees))
696	}
697
698	/// Open issues assigned to the viewer, in every repository. Callers
699	/// show only those in repositories the viewer can still see.
700	async fn list_assigned_issues(&self, a: ViewerArgs) -> Result<Vec<Issue>> {
701	let Some(viewer) = a.viewer else {
702	return Ok(Vec::new());
703	};
704	let rows = self
705	.db
706	.prepare(format!(
707	"SELECT {ISSUE_COLUMNS} FROM issues
708	WHERE state = 'open' AND EXISTS (
709	SELECT 1 FROM json_each(issues.assignees) WHERE json_each.value = ?)
710	ORDER BY updated_at DESC LIMIT 50"
711	))
712	.bind(&[viewer.username.into()])?
713	.all()
714	.await?
715	.results::<IssueRow>()?;
716	Ok(rows.into_iter().map(Issue::from).collect())
717	}
718
719	async fn close_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
720	let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
721	if issue.state == State::Closed {
722	return Ok(Outcome::fail(
723	FailureCode::Conflict,
724	"This issue is already closed.",
725	));
726	}
727	let reason = a.reason.unwrap_or(IssueReason::Completed);
728	let now = rfc3339(now_ms());
729	self.db
730	.prepare(
731	"UPDATE issues SET state = 'closed', reason = ?, closed_at = ?, updated_at = ?
732	WHERE id = ?",
733	)
734	.bind(&[
735	reason.as_str().into(),
736	now.as_str().into(),
737	now.as_str().into(),
738	issue.id.as_str().into(),
739	])?
740	.run()
741	.await?;
742	self.publish(
743	"issue.closed",
744	&issue.repo_id,
745	&a.actor,
746	IssueEvent {
747	reason: Some(reason.as_str()),
748	..Self::issue_event(&issue)
749	},
750	)
751	.await?;
752	self.note(
753	&issue.repo_id,
754	issue.number,
755	(&a.actor.id, &a.actor.username),
756	match reason {
757	IssueReason::Completed => "closed this as completed",
758	IssueReason::NotPlanned => "closed this as not planned",
759	},
760	)
761	.await?;
762	issue.state = State::Closed;
763	issue.reason = Some(reason);
764	issue.closed_at = Some(now.clone());
765	issue.updated_at = now;
766	Ok(Outcome::Ok(issue))
767	}
768
769	async fn reopen_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
770	let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
771	if issue.state == State::Open {
772	return Ok(Outcome::fail(
773	FailureCode::Conflict,
774	"This issue is already open.",
775	));
776	}
777	let now = rfc3339(now_ms());
778	self.db
779	.prepare(
780	"UPDATE issues
781	SET state = 'open', reason = NULL, resolved_by = NULL, closed_at = NULL,
782	updated_at = ?
783	WHERE id = ?",
784	)
785	.bind(&[now.as_str().into(), issue.id.as_str().into()])?
786	.run()
787	.await?;
788	self.publish(
789	"issue.reopened",
790	&issue.repo_id,
791	&a.actor,
792	Self::issue_event(&issue),
793	)
794	.await?;
795	self.note(
796	&issue.repo_id,
797	issue.number,
798	(&a.actor.id, &a.actor.username),
799	"reopened this",
800	)
801	.await?;
802	issue.state = State::Open;
803	issue.reason = None;
804	issue.resolved_by = None;
805	issue.closed_at = None;
806	issue.updated_at = now;
807	Ok(Outcome::Ok(issue))
808	}
809
810	/// The default labels, then every other label in use on the repository.
811	async fn list_labels(&self, a: ViewArgs) -> Result<Outcome<Vec<String>>> {
812	let read = \|repo_id: String\| async move {
813	let query = self
814	.db
815	.prepare(
816	"SELECT DISTINCT json_each.value AS value
817	FROM issues, json_each(issues.labels)
818	WHERE issues.repo_id = ? ORDER BY 1 LIMIT 200",
819	)
820	.bind(&[repo_id.into()])?;
821	self.timing.db(1, query.all()).await?.results::<ValueRow>()
822	};
823	let (_, used) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
824	let mut labels: Vec<String> = DEFAULT_LABELS.iter().map(\|label\| (*label).into()).collect();
825	for row in used {
826	if !labels.contains(&row.value) {
827	labels.push(row.value);
828	}
829	}
830	Ok(Outcome::Ok(labels))
831	}
832
833	async fn counts(&self, a: ViewArgs) -> Result<Outcome<Counts>> {
834	let read = \|repo_id: String\| async move {
835	let query = self
836	.db
837	.prepare(
838	"SELECT
839	(SELECT count(*) FROM issues WHERE repo_id = ?1 AND state = 'open') AS issues,
840	(SELECT count(*) FROM pulls
841	WHERE repo_id = ?1 AND status IN ('draft', 'open')) AS pulls",
842	)
843	.bind(&[repo_id.into()])?;
844	self.timing.db(1, query.first::<Counts>(None)).await
845	};
846	let (_, counts) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
847	Ok(Outcome::Ok(counts.unwrap_or(Counts {
848	issues: 0,
849	pulls: 0,
850	})))
851	}
852
853	// --- Comments ----------------------------------------------------------
854
855	async fn add_comment(&self, a: AddCommentArgs) -> Result<Outcome<Comment>> {
856	if !a.actor.verified {
857	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
858	}
859	let body = a.body.trim();
860	// An approval speaks for itself; anything else has to say something.
861	if body.is_empty() && a.verdict != Some(Verdict::Approve) {
862	return Ok(Outcome::fail(
863	FailureCode::Invalid,
864	"A comment cannot be empty.",
865	));
866	}
867	let path = a
868	.path
869	.as_deref()
870	.map(str::trim)
871	.filter(\|path\| !path.is_empty());
872	let line = a.line.filter(\|line\| *line > 0 && path.is_some());
873	if body.chars().count() > MAX_ENTRY_CHARS {
874	return Ok(Outcome::fail(
875	FailureCode::Invalid,
876	"That comment is too long.",
877	));
878	}
879	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
880	check!(writable(&repo));
881	// The number names an issue or a pull request, never both.
882	let mut pull_id = None;
883	let table = if self.issue(&repo.id, a.number).await?.is_some() {
884	if path.is_some() \|\| a.verdict.is_some() {
885	return Ok(Outcome::fail(
886	FailureCode::Invalid,
887	"Only a pull request can be reviewed or commented on by line.",
888	));
889	}
890	"issues"
891	} else if let Some(pull) = self.pull(&repo.id, a.number).await? {
892	if a.verdict.is_some() && pull.is_owned_by(&a.actor.id) {
893	return Ok(Outcome::fail(
894	FailureCode::Forbidden,
895	"You cannot approve or request changes on your own pull request.",
896	));
897	}
898	pull_id = Some(pull.id.clone());
899	"pulls"
900	} else {
901	return Ok(Outcome::fail(
902	FailureCode::NotFound,
903	"No issue or pull request has that number.",
904	));
905	};
906
907	let now = now_ms();
908	let comment = Comment {
909	kind: CommentKind::Comment,
910	id: new_id("cmt", now),
911	author: a.actor.clone(),
912	body: body.to_owned(),
913	path: path.map(str::to_owned),
914	line,
915	verdict: a.verdict,
916	created_at: rfc3339(now),
917	};
918	self.db
919	.batch(vec![
920	self.db
921	.prepare(
922	"INSERT INTO comments
923	(id, repo_id, number, author_id, author_name, body, path, line,
924	verdict, created_at)
925	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
926	)
927	.bind(&[
928	comment.id.as_str().into(),
929	repo.id.as_str().into(),
930	a.number.into(),
931	a.actor.id.as_str().into(),
932	a.actor.username.as_str().into(),
933	body.into(),
934	optional(&comment.path),
935	optional_number(line),
936	a.verdict
937	.map_or(JsValue::NULL, \|verdict\| verdict.as_str().into()),
938	comment.created_at.as_str().into(),
939	])?,
940	self.db
941	.prepare(format!(
942	"UPDATE {table} SET updated_at = ? WHERE repo_id = ? AND number = ?"
943	))
944	.bind(&[
945	comment.created_at.as_str().into(),
946	repo.id.as_str().into(),
947	a.number.into(),
948	])?,
949	])
950	.await?;
951	self.note_mention(&a.actor, &repo, a.number, &comment, pull_id.as_deref()).await?;
952	self.publish(
953	"comment.created",
954	&repo.id,
955	&a.actor,
956	CommentCreated {
957	comment_id: comment.id.clone(),
958	repo_id: repo.id.clone(),
959	number: a.number,
960	pull_id,
961	verdict: a.verdict,
962	},
963	)
964	.await?;
965	Ok(Outcome::Ok(comment))
966	}
967
968	// --- Pull requests -----------------------------------------------------
969
970	async fn open_pull(&self, a: OpenPullArgs) -> Result<Outcome<Pull>> {
971	if !a.actor.verified {
972	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
973	}
974	let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
975	check!(writable(&repo));
976	// g1t's own agent at work spends the workspace's compute; a pull
977	// request anyone else's agent makes is like any other.
978	if matches!(a.runtime, Runtime::Hosted) {
979	check!(allowed(Some(&a.actor), &repo, Capability::Run));
980	}
981	let issue = match a.issue {
982	Some(number) => match self.issue(&repo.id, number).await? {
983	Some(issue) if issue.state == State::Open => Some(issue),
984	Some(_) => {
985	return Ok(Outcome::fail(
986	FailureCode::Conflict,
987	"This issue is closed.",
988	));
989	}
990	None => return Ok(no_issue()),
991	},
992	None => None,
993	};
994	// A pull request for an issue takes the issue's title unless given one.
995	let title = match (a.title.trim(), &issue) {
996	("", Some(issue)) => issue.title.clone(),
997	(title, _) => match valid_title(title) {
998	Ok(title) => title.to_owned(),
999	Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
1000	},
1001	};
1002	let agent = match a.agent.trim() {
1003	"" => "agent",
1004	agent => agent,
1005	};
1006	let runtime = match a.runtime {
1007	Runtime::Hosted => "hosted",
1008	Runtime::External => "external",
1009	};
1010
1011	let now = now_ms();
1012	let id = new_id("pr", now);
1013	let branch = a
1014	.branch
1015	.as_deref()
1016	.map(str::trim)
1017	.filter(\|branch\| !branch.is_empty());
1018	// The change is on a branch already pushed to the repository, or
1019	// will be made in a fork created for this pull request.
1020	let (fork, head) = match branch {
1021	Some(branch) => {
1022	if branch == repo.default_branch {
1023	return Ok(Outcome::fail(
1024	FailureCode::Invalid,
1025	format!("Choose a branch other than {branch}."),
1026	));
1027	}
1028	let head: Option<String> = g1t_kit::call(
1029	&self.repos,
1030	"head",
1031	&HeadArgs {
1032	repo_id: repo.id.clone(),
1033	branch: branch.to_owned(),
1034	},
1035	)
1036	.await?;
1037	let Some(head) = head else {
1038	return Ok(Outcome::fail(
1039	FailureCode::NotFound,
1040	format!("There is no branch named {branch}. Push it first."),
1041	));
1042	};
1043	let existing = self
1044	.db
1045	.prepare(
1046	"SELECT number AS n FROM pulls
1047	WHERE repo_id = ? AND source_branch = ? AND status IN ('draft', 'open')",
1048	)
1049	.bind(&[repo.id.as_str().into(), branch.into()])?
1050	.first::<NumberRow>(None)
1051	.await?;
1052	if let Some(existing) = existing {
1053	return Ok(Outcome::fail(
1054	FailureCode::Conflict,
1055	format!("Pull request #{} is already open for {branch}.", existing.n),
1056	));
1057	}
1058	(None, Some(head))
1059	}
1060	None => {
1061	let fork: Outcome<Repo> = g1t_kit::call(
1062	&self.repos,
1063	"fork_for_pull",
1064	&ForkArgs {
1065	source_id: repo.id.clone(),
1066	pull_id: id.clone(),
1067	actor: a.actor.clone(),
1068	},
1069	)
1070	.await?;
1071	(Some(check!(fork)), None)
1072	}
1073	};
1074	// A branch already holds the work, so its pull request is ready for
1075	// review from the start; one with a fork starts as a draft.
1076	let status = if branch.is_some() { "open" } else { "draft" };
1077	let body = Some(a.body.trim().to_owned()).filter(\|body\| !body.is_empty());
1078
1079	let number = self.next_number(&repo.id).await?;
1080	let timestamp = rfc3339(now);
1081	// A change g1t makes is g1t's, for whoever asked for it. This is
1082	// what lifecycle::made_by_g1t reads back.
1083	let by_g1t = matches!(a.runtime, Runtime::Hosted) && agent == reviews::AGENT_NAME && fork.is_some();
1084	let (author, requested_by) = authorship(&a.actor, by_g1t);
1085	self.db
1086	.prepare(
1087	"INSERT INTO pulls
1088	(id, repo_id, number, issue_id, issue_number, title, body, agent, runtime,
1089	status, fork_repo_id, fork_namespace, fork_name, source_branch, head_commit,
1090	author_id, author_name, requested_by_id, requested_by_name, created_at, updated_at)
1091	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
1092	)
1093	.bind(&[
1094	id.as_str().into(),
1095	repo.id.as_str().into(),
1096	number.into(),
1097	optional(&issue.as_ref().map(\|issue\| issue.id.clone())),
1098	optional_number(issue.as_ref().map(\|issue\| issue.number)),
1099	title.into(),
1100	optional(&body),
1101	agent.into(),
1102	runtime.into(),
1103	status.into(),
1104	optional(&fork.as_ref().map(\|fork\| fork.id.clone())),
1105	optional(&fork.as_ref().map(\|fork\| fork.namespace.clone())),
1106	optional(&fork.as_ref().map(\|fork\| fork.name.clone())),
1107	optional(&branch.map(str::to_owned)),
1108	optional(&head),
1109	author.id.as_str().into(),
1110	author.username.as_str().into(),
1111	optional(&requested_by.as_ref().map(\|user\| user.id.clone())),
1112	optional(&requested_by.as_ref().map(\|user\| user.username.clone())),
1113	timestamp.as_str().into(),
1114	timestamp.as_str().into(),
1115	])?
1116	.run()
1117	.await?;
1118	let Some(pull) = self.pull(&repo.id, number).await? else {
1119	return Ok(no_pull());
1120	};
1121	self.manage(&pull).await?;
1122	// Someone is on it now, so it is no longer waiting for an agent.
1123	if let Some(issue) = pull.issue {
1124	self.db
1125	.prepare("UPDATE issues SET queued_by = NULL WHERE repo_id = ? AND number = ?")
1126	.bind(&[repo.id.as_str().into(), issue.into()])?
1127	.run()
1128	.await?;
1129	}
1130	if let Some(issue) = pull.issue {
1131	let text = if lifecycle::made_by_g1t(&pull) {
1132	format!("assigned this to g1t, which opened #{}", pull.number)
1133	} else {
1134	format!("opened #{} for this", pull.number)
1135	};
1136	self.note(&repo.id, issue, (&a.actor.id, &a.actor.username), &text)
1137	.await?;
1138	}
1139	self.publish(
1140	"pull.opened",
1141	&repo.id,
1142	&a.actor,
1143	PullEvent {
1144	agent: Some(pull.agent.clone()),
1145	..Self::pull_event(&pull)
1146	},
1147	)
1148	.await?;
1149	Ok(Outcome::Ok(pull))
1150	}
1151
1152	async fn list_pulls(&self, a: ListPullsArgs) -> Result<Outcome<Vec<Pull>>> {
1153	let filter = match a.state {
1154	Some(State::Open) => "AND status IN ('draft', 'open')",
1155	Some(State::Closed) => "AND status IN ('merged', 'closed')",
1156	None => "",
1157	};
1158	let read = \|repo_id: String\| async move {
1159	let query = self
1160	.db
1161	.prepare(format!(
1162	"SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id = ? {filter} ORDER BY number DESC LIMIT ?"
1163	))
1164	.bind(&[repo_id.into(), LIST_PAGE.into()])?;
1165	self.timing.db(1, query.all()).await?.results::<PullRow>()
1166	};
1167	let (_, rows) = check!(self.repo_then(&a.repo, &a.viewer, read).await?);
1168	Ok(Outcome::Ok(rows.into_iter().map(Pull::from).collect()))
1169	}
1170
1171	/// `pulls_for_repos`: what `list_pulls` gives, open and closed, for many
1172	/// repositories at once: one access check with repos for all of them
1173	/// and one query, instead of two of each per repository.
1174	async fn pulls_for_repos(&self, a: PullsForReposArgs) -> Result<Vec<RepoPulls>> {
1175	let ids: Vec<String> = a.repo_ids.into_iter().take(MAX_PULLS_FOR_REPOS).collect();
1176	if ids.is_empty() {
1177	return Ok(Vec::new());
1178	}
1179	let limit = a.limit.clamp(1, LIST_PAGE);
1180	// The rows are read beside the access check, for every id asked
1181	// about; those of repositories the viewer cannot read are dropped.
1182	let asked = serde_json::to_string(&ids)?;
1183	let check = ReadableArgs { ids, viewer: a.viewer };
1184	let readable = self.timing.rpc(g1t_kit::call::<_, Vec<Repo>>(&self.repos, "readable", &check));
1185	let (readable, rows) = try_join(readable, self.timing.db(1, self.newest_pulls(asked, limit))).await?;
1186	if readable.is_empty() {
1187	return Ok(Vec::new());
1188	}
1189	let mut answer: Vec<RepoPulls> = readable
1190	.iter()
1191	.map(\|repo\| RepoPulls { repo_id: repo.id.clone(), open: Vec::new(), closed: Vec::new() })
1192	.collect();
1193	for pull in rows.into_iter().map(Pull::from) {
1194	let Some(entry) = answer.iter_mut().find(\|entry\| entry.repo_id == pull.repo_id) else {
1195	continue;
1196	};
1197	match pull.status {
1198	PullStatus::Draft \| PullStatus::Open => entry.open.push(pull),
1199	PullStatus::Merged \| PullStatus::Closed => entry.closed.push(pull),
1200	}
1201	}
1202	for entry in &mut answer {
1203	entry.open.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1204	entry.closed.sort_by_key(\|pull\| std::cmp::Reverse(pull.number));
1205	}
1206	Ok(answer)
1207	}
1208
1209	/// The newest `limit` of each repository's open (draft or open) and
1210	/// closed (merged or closed) pull requests, for the ids in `ids` (JSON).
1211	async fn newest_pulls(&self, ids: String, limit: u32) -> Result<Vec<PullRow>> {
1212	self.db
1213	.prepare(format!(
1214	"SELECT * FROM (
1215	SELECT {PULL_COLUMNS}, ROW_NUMBER() OVER (
1216	PARTITION BY pulls.repo_id, pulls.status IN ('draft', 'open') ORDER BY pulls.number DESC
1217	) AS place
1218	FROM pulls WHERE pulls.repo_id IN (SELECT value FROM json_each(?1))
1219	) WHERE place <= ?2"
1220	))
1221	.bind(&[ids.into(), limit.into()])?
1222	.all()
1223	.await?
1224	.results::<PullRow>()
1225	}
1226
1227	async fn get_pull(&self, a: ViewArgs) -> Result<Outcome<PullDetail>> {
1228	let number = a.number;
1229	// Every row the page and the lifecycle read, in one batch started
1230	// beside the access check; the helpers below read from it.
1231	let read = \|repo_id: String\| self.prefetch_pull(repo_id, number);
1232	let Outcome::Ok((repo, Some(found))) = self.repo_then(&a.repo, &a.viewer, read).await? else {
1233	return Ok(no_pull());
1234	};
1235	let Some(row) = found.first::<PullRow>(prefetch::Slot::Pull)? else {
1236	return Ok(no_pull());
1237	};
1238	let stored = found.first::<StoredBehind>(prefetch::Slot::Pull)?;
1239	let issue = found.first::<IssueRow>(prefetch::Slot::Issue)?.map(Issue::from);
1240	let comments: Vec<Comment> =
1241	found.rows::<CommentRow>(prefetch::Slot::Comments)?.into_iter().map(Comment::from).collect();
1242	self.keep_prefetched(Some(found));
1243	let detail = self.pull_detail(repo, Pull::from(row), issue, comments, stored).await;
1244	self.keep_prefetched(None);
1245	detail
1246	}
1247
1248	async fn pull_detail(
1249	&self,
1250	repo: Repo,
1251	mut pull: Pull,
1252	issue: Option<Issue>,
1253	comments: Vec<Comment>,
1254	stored: Option<StoredBehind>,
1255	) -> Result<Outcome<PullDetail>> {
1256	// Whether it is behind, as worked out with its mergeability on the
1257	// last push to either side (mergeability.rs), when that was for
1258	// its head as it is now; otherwise asked of the repos service.
1259	let known_behind = stored.and_then(\|stored\| stored.for_head(pull.head_commit.as_deref()));
1260	// Worked out on each push; this covers a pull request from before
1261	// that was recorded.
1262	if pull.files.is_empty() && pull.head_commit.is_some() {
1263	pull.files = self.refresh_files(&pull).await?;
1264	}
1265	// Everything else at once: none of it depends on the rest, and each
1266	// is a round trip of its own.
1267	let standing = async {
1268	// Mergeability first: where g1t sees a pull request through, a
1269	// conflict decides its next step.
1270	let behind = async {
1271	match known_behind {
1272	Some(behind) => Ok(behind),
1273	None => {
1274	let behind = self.is_behind(&repo.id, &pull).await?;
1275	// Kept for the next view when the mergeability on
1276	// record is for this head: a pull request from
1277	// before `behind` was kept asks once.
1278	if let Some(head) = pull.head_commit.as_deref()
1279	&& pull.status.is_active()
1280	{
1281	self.db
1282	.prepare(
1283	"UPDATE pulls SET behind = ?1
1284	WHERE id = ?2 AND behind IS NULL AND mergeable_key LIKE ?3 \|\| '..%'",
1285	)
1286	.bind(&[u32::from(behind).into(), pull.id.as_str().into(), head.into()])?
1287	.run()
1288	.await?;
1289	}
1290	Ok(behind)
1291	}
1292	}
1293	};
1294	let (merge, behind) = try_join(self.mergeability(&pull), behind).await?;
1295	let assessed = self.assess_with_confidence(&pull, &issue, behind).await?;
1296	let confidence = assessed.as_ref().and_then(\|(_, _, confidence)\| confidence.clone());
1297	let lifecycle = assessed.map(\|(lifecycle, _, _)\| lifecycle);
1298	Ok::<_, worker::Error>((behind, (lifecycle, confidence), merge))
1299	};
1300	let (((behind, (lifecycle, confidence), (mergeable, conflicts)), (landing, stalled), comments), (checks, overlaps, review_pending)) =
1301	try_join(
1302	try_join3(standing, self.landing_state(&pull.id), async { Ok(comments) }),
1303	try_join3(
1304	self.latest_checks(&pull.id),
1305	self.overlaps(&pull),
1306	self.review_pending(&pull.id),
1307	),
1308	)
1309	.await?;
1310	// As just worked out, rather than as it was read.
1311	if confidence.is_some() {
1312	pull.confidence = confidence;
1313	}
1314	let (statuses, settings) =
1315	try_join(self.statuses(&repo.id, pull.head_commit.as_deref()), self.settings(&repo.id)).await?;
1316	Ok(Outcome::Ok(PullDetail {
1317	required_checks: required_checks(&settings.required_checks, &statuses),
1318	comments,
1319	checks,
1320	overlaps,
1321	behind,
1322	review_pending,
1323	lifecycle,
1324	landing,
1325	stalled,
1326	messages: self.messages(&pull.id).await?,
1327	statuses,
1328	mergeable,
1329	conflicts,
1330	earlier_checks: self.earlier_checks(&pull.id).await?,
1331	issue,
1332	pull,
1333	}))
1334	}
1335
1336	/// The pull request, if it is still active and `actor` opened it or
1337	/// may triage the repository's pull requests.
1338	async fn manageable_pull(
1339	&self,
1340	actor: &User,
1341	path: &RepoPath,
1342	number: u32,
1343	) -> Result<Outcome<Pull>> {
1344	let (repo, pull) = check!(self.pull_at(path, number, &Some(actor.clone())).await?);
1345	check!(writable(&repo));
1346	if !pull.is_owned_by(&actor.id) {
1347	check!(allowed(Some(actor), &repo, Capability::Triage));
1348	}
1349	if !pull.status.is_active() {
1350	return Ok(Outcome::fail(
1351	FailureCode::Conflict,
1352	format!("This pull request is already {}.", pull.status.as_str()),
1353	));
1354	}
1355	Ok(Outcome::Ok(pull))
1356	}
1357
1358	/// Brings a pull request up to date with the default branch without a
1359	/// sandbox, where the repos service can do that safely. Whoever could
1360	/// have pushed the merge themselves may ask: whoever opened it (or asked
1361	/// g1t for it), for a fork; anyone who may push, for a branch of the
1362	/// repository. When it needs a
1363	/// real merge, says so, naming the conflicting files if a probe found
1364	/// them, and pushes nothing.
1365	async fn catch_up_pull(&self, a: PullActionArgs) -> Result<Outcome<PullBranchUpdate>> {
1366	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &Some(a.actor.clone())).await?);
1367	check!(writable(&repo));
1368	if !pull.status.is_active() {
1369	return Ok(Outcome::fail(
1370	FailureCode::Conflict,
1371	format!("This pull request is already {}.", pull.status.as_str()),
1372	));
1373	}
1374	if pull.fork_repo_id.is_some() {
1375	if !pull.is_owned_by(&a.actor.id) {
1376	return Ok(Outcome::fail(
1377	FailureCode::Forbidden,
1378	"Only whoever opened this pull request, or asked g1t for it, can update it.",
1379	));
1380	}
1381	} else {
1382	check!(allowed(Some(&a.actor), &repo, Capability::Push));
1383	}
1384	let updated: Outcome<PullBranchUpdate> = g1t_kit::call(
1385	&self.repos,
1386	"update_pull_branch",
1387	&UpdatePullBranchArgs {
1388	source_id: pull.fork_repo_id.clone().unwrap_or_else(\|\| repo.id.clone()),
1389	branch: pull.branch.clone(),
1390	number: pull.number,
1391	actor: a.actor,
1392	},
1393	)
1394	.await?;
1395	// A probe that found conflicts says more than "both changed it".
1396	if let Outcome::Ok(PullBranchUpdate::NeedsAgent { .. }) = &updated
1397	&& let Some(files) = self.conflicting_files(&pull).await?
1398	&& !files.is_empty()
1399	{
1400	return Ok(Outcome::Ok(PullBranchUpdate::NeedsAgent {
1401	reason: NeedsAgentReason::Conflicting,
1402	detail: "Merging it conflicts.".to_owned(),
1403	paths: files,
1404	}));
1405	}
1406	Ok(updated)
1407	}
1408
1409	async fn update_pull(&self, a: UpdatePullArgs) -> Result<Outcome<Pull>> {
1410	let pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1411	let assignees = match a.assignees {
1412	Some(names) => Some(check!(self.valid_assignees(names).await?)),
1413	None => None,
1414	};
1415	let reviewers = match a.reviewers {
1416	Some(names) => {
1417	// g1t is not an account; everyone else has to be.
1418	let agent = names
1419	.iter()
1420	.any(\|name\| name.trim().eq_ignore_ascii_case(reviews::AGENT_NAME));
1421	let people = names
1422	.into_iter()
1423	.filter(\|name\| !name.trim().eq_ignore_ascii_case(reviews::AGENT_NAME))
1424	.collect();
1425	let mut reviewers = check!(self.valid_assignees(people).await?);
1426	// Nobody is asked to review their own, nor what they had g1t make.
1427	reviewers.retain(\|name\| *name != pull.owner().username);
1428	if agent {
1429	reviewers.insert(0, reviews::AGENT_NAME.to_owned());
1430	}
1431	Some(reviewers)
1432	}
1433	None => None,
1434	};
1435	self.db
1436	.prepare(
1437	"UPDATE pulls
1438	SET assignees = COALESCE(?, assignees), reviewers = COALESCE(?, reviewers),
1439	updated_at = ?
1440	WHERE id = ?",
1441	)
1442	.bind(&[
1443	optional(&assignees.as_ref().map(serde_json::to_string).transpose()?),
1444	optional(&reviewers.as_ref().map(serde_json::to_string).transpose()?),
1445	rfc3339(now_ms()).into(),
1446	pull.id.as_str().into(),
1447	])?
1448	.run()
1449	.await?;
1450	if let Some(assignees) = &assignees {
1451	self.note_changes(
1452	&pull.repo_id,
1453	pull.number,
1454	&a.actor,
1455	&pull.assignees,
1456	assignees,
1457	("assigned", "unassigned"),
1458	)
1459	.await?;
1460	}
1461	if let Some(reviewers) = &reviewers {
1462	self.note_changes(
1463	&pull.repo_id,
1464	pull.number,
1465	&a.actor,
1466	&pull.reviewers,
1467	reviewers,
1468	(
1469	"requested a review from",
1470	"withdrew the request for a review from",
1471	),
1472	)
1473	.await?;
1474	}
1475	Ok(match self.pull(&pull.repo_id, pull.number).await? {
1476	Some(pull) => Outcome::Ok(pull),
1477	None => no_pull(),
1478	})
1479	}
1480
1481	/// Marks a draft ready for review, or updates the description of one
1482	/// that already is.
1483	async fn ready_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1484	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1485	let summary = Some(a.summary.trim().to_owned()).filter(\|summary\| !summary.is_empty());
1486	let now = rfc3339(now_ms());
1487	self.db
1488	.prepare(
1489	"UPDATE pulls SET status = 'open', body = COALESCE(?, body), updated_at = ?
1490	WHERE id = ?",
1491	)
1492	.bind(&[
1493	optional(&summary),
1494	now.as_str().into(),
1495	pull.id.as_str().into(),
1496	])?
1497	.run()
1498	.await?;
1499	if pull.status == PullStatus::Draft {
1500	// The head as it is now: the push that came just before may not
1501	// have reached `head_commit` yet, and workflows run on it.
1502	let commit = self.live_head(&pull).await?.or_else(\|\| pull.head_commit.clone());
1503	self.publish(
1504	"pull.ready",
1505	&pull.repo_id,
1506	&a.actor,
1507	PullEvent {
1508	commit,
1509	..Self::pull_event(&pull)
1510	},
1511	)
1512	.await?;
1513	}
1514	if pull.status == PullStatus::Draft {
1515	self.note(
1516	&pull.repo_id,
1517	pull.number,
1518	(&a.actor.id, &a.actor.username),
1519	"marked this ready for review",
1520	)
1521	.await?;
1522	}
1523	pull.status = PullStatus::Open;
1524	pull.body = summary.or(pull.body);
1525	pull.updated_at = now;
1526	Ok(Outcome::Ok(pull))
1527	}
1528
1529	async fn close_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1530	let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1531	let now = rfc3339(now_ms());
1532	self.db
1533	.prepare("UPDATE pulls SET status = 'closed', updated_at = ? WHERE id = ?")
1534	.bind(&[now.as_str().into(), pull.id.as_str().into()])?
1535	.run()
1536	.await?;
1537	self.publish(
1538	"pull.closed",
1539	&pull.repo_id,
1540	&a.actor,
1541	Self::pull_event(&pull),
1542	)
1543	.await?;
1544	self.note(
1545	&pull.repo_id,
1546	pull.number,
1547	(&a.actor.id, &a.actor.username),
1548	"closed this",
1549	)
1550	.await?;
1551	// A closed pull request leaves the merge queue.
1552	if self
1553	.leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was closed."))
1554	.await?
1555	{
1556	self.publish_as(
1557	"queue.changed",
1558	&pull.repo_id,
1559	None,
1560	g1t_contracts::events::QueueChanged {
1561	repo_id: pull.repo_id.clone(),
1562	},
1563	)
1564	.await?;
1565	}
1566	pull.status = PullStatus::Closed;
1567	pull.updated_at = now;
1568	Ok(Outcome::Ok(pull))
1569	}
1570
1571	/// Lands the pull request on the repository's default branch. Unless
1572	/// told to keep it open, that resolves the issue it was for: the issue
1573	/// closes naming this pull request, and the others still in progress
1574	/// for it close as superseded.
1575	async fn merge_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1576	let viewer = Some(a.actor.clone());
1577	let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
1578	check!(writable(&repo));
1579	match pull.status {
1580	PullStatus::Open => {}
1581	PullStatus::Draft => {
1582	return Ok(Outcome::fail(
1583	FailureCode::Conflict,
1584	"This pull request is still a draft. Mark it ready for review first.",
1585	));
1586	}
1587	status => {
1588	return Ok(Outcome::fail(
1589	FailureCode::Conflict,
1590	format!("This pull request is already {}.", status.as_str()),
1591	));
1592	}
1593	}
1594	let settings = self.settings(&repo.id).await?;
1595	// The default branch's protection: its required checks must pass on
1596	// the head, for a person's pull request and an agent's alike. Where
1597	// the repository does not allow bypassing them, asking to bypass
1598	// them changes nothing.
1599	if !a.ignore_checks \|\| !settings.allow_ignoring_checks {
1600	let queue = (pull.check_status == Some(CheckStatus::Failed))
1601	.then(\|\| "It failed in the merge queue; push a fix to try again.".to_owned());
1602	let required = statuses::WorkflowFacts::of(
1603	&self.statuses(&repo.id, pull.head_commit.as_deref()).await?,
1604	&settings.required_checks,
1605	)
1606	.refusal();
1607	if let Some(reason) = queue.or(required) {
1608	let remedy = if settings.allow_ignoring_checks {
1609	"Wait or fix them, or bypass the required checks as you merge."
1610	} else {
1611	"This repository only merges pull requests whose required checks pass."
1612	};
1613	return Ok(Outcome::fail(
1614	FailureCode::Conflict,
1615	format!("{reason} {remedy}"),
1616	));
1617	}
1618	}
1619	if let Some(missing) = self.approvals_gap(&settings, &pull).await? {
1620	return Ok(Outcome::fail(FailureCode::Conflict, missing));
1621	}
1622	// Known ahead of time to conflict: neither a merge nor the queue
1623	// would get through, so say what has to be resolved now.
1624	if let Some(files) = self.conflicting_files(&pull).await? {
1625	let named = if files.is_empty() {
1626	String::new()
1627	} else {
1628	format!(" in {}", files.join(", "))
1629	};
1630	return Ok(Outcome::fail(
1631	FailureCode::Conflict,
1632	format!(
1633	"This branch has conflicts with {}{named} that must be resolved first. Have g1t resolve them, or merge {0} into it, fix them and push.",
1634	repo.default_branch
1635	),
1636	));
1637	}
1638
1639	// A repository that merges through a queue: it joins the queue, and
1640	// lands once its state together with everything ahead has passed.
1641	if settings.merge_queue {
1642	if !a.actor.verified {
1643	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1644	}
1645	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
1646	return self.enqueue(&repo, &pull, &a.actor, a.keep_issue_open).await;
1647	}
1648
1649	// The default branch has moved under it. Unless the repository
1650	// insists on that being dealt with first, bring it up to date and
1651	// land it when that is done.
1652	if self.is_behind(&repo.id, &pull).await? {
1653	if settings.require_up_to_date {
1654	return Ok(Outcome::fail(
1655	FailureCode::Conflict,
1656	format!(
1657	"{} has moved since this pull request was made, and this repository requires pull requests to be up to date before they merge. Catch up with {0} first.",
1658	repo.default_branch
1659	),
1660	));
1661	}
1662	if !a.actor.verified {
1663	return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1664	}
1665	check!(allowed(Some(&a.actor), &repo, Capability::Merge));
1666	self.request_landing(&pull, &a.actor, a.keep_issue_open)
1667	.await?;
1668	return Ok(Outcome::Ok(pull));
1669	}
1670
1671	// Whether the actor may write to the repository is decided by repos.
1672	let landed: Outcome<Landed> = g1t_kit::call(
1673	&self.repos,
1674	"land",
1675	&LandArgs {
1676	// A pull request from a branch lands from the repository itself.
1677	source_id: pull.fork_repo_id.clone().unwrap_or_else(\|\| repo.id.clone()),
1678	branch: pull.branch.clone(),
1679	actor: a.actor.clone(),
1680	},
1681	)
1682	.await?;
1683	let landed = check!(landed);
1684	Ok(Outcome::Ok(
1685	self.record_merge(&repo, pull, &a.actor, a.keep_issue_open, landed)
1686	.await?,
1687	))
1688	}
1689
1690	/// Records a pull request as merged once the default branch holds it:
1691	/// closes its issue, supersedes the others for it, and says so.
1692	pub(crate) async fn record_merge(
1693	&self,
1694	repo: &Repo,
1695	mut pull: Pull,
1696	actor: &User,
1697	keep_issue_open: bool,
1698	landed: Landed,
1699	) -> Result<Pull> {
1700	let issue = match pull.issue {
1701	Some(number) if !keep_issue_open => self
1702	.issue(&repo.id, number)
1703	.await?
1704	.filter(\|issue\| issue.state == State::Open),
1705	_ => None,
1706	};
1707	let now = rfc3339(now_ms());
1708	let mut statements = vec![
1709	self.db
1710	.prepare(
1711	"UPDATE pulls
1712	SET status = 'merged', head_commit = ?, merge_base = ?, merged_by = ?,
1713	merged_at = ?, updated_at = ?
1714	WHERE id = ?",
1715	)
1716	.bind(&[
1717	landed.commit.as_str().into(),
1718	optional(&landed.previous),
1719	actor.username.as_str().into(),
1720	now.as_str().into(),
1721	now.as_str().into(),
1722	pull.id.as_str().into(),
1723	])?,
1724	];
1725	if let Some(issue) = &issue {
1726	statements.push(
1727	self.db
1728	.prepare(
1729	"UPDATE issues
1730	SET state = 'closed', reason = 'completed', resolved_by = ?,
1731	closed_at = ?, updated_at = ?
1732	WHERE id = ?",
1733	)
1734	.bind(&[
1735	pull.number.into(),
1736	now.as_str().into(),
1737	now.as_str().into(),
1738	issue.id.as_str().into(),
1739	])?,
1740	);
1741	statements.push(
1742	self.db
1743	.prepare(
1744	"UPDATE pulls SET status = 'closed', superseded_by = ?, updated_at = ?
1745	WHERE issue_id = ? AND id != ? AND status IN ('draft', 'open')",
1746	)
1747	.bind(&[
1748	pull.number.into(),
1749	now.as_str().into(),
1750	issue.id.as_str().into(),
1751	pull.id.as_str().into(),
1752	])?,
1753	);
1754	}
1755	self.db.batch(statements).await?;
1756
1757	self.publish(
1758	"pull.merged",
1759	&repo.id,
1760	actor,
1761	PullEvent {
1762	commit: Some(landed.commit.clone()),
1763	..Self::pull_event(&pull)
1764	},
1765	)
1766	.await?;
1767	if let Some(issue) = &issue {
1768	self.publish(
1769	"issue.closed",
1770	&repo.id,
1771	actor,
1772	IssueEvent {
1773	reason: Some(IssueReason::Completed.as_str()),
1774	resolved_by: Some(pull.number),
1775	..Self::issue_event(issue)
1776	},
1777	)
1778	.await?;
1779	}
1780
1781	let who = (actor.id.as_str(), actor.username.as_str());
1782	self.note(&repo.id, pull.number, who, "merged this").await?;
1783	if let Some(issue) = &issue {
1784	self.note(
1785	&repo.id,
1786	issue.number,
1787	who,
1788	&format!("closed this by merging #{}", pull.number),
1789	)
1790	.await?;
1791	}
1792	pull.status = PullStatus::Merged;
1793	pull.head_commit = Some(landed.commit.clone());
1794	pull.merge_base = landed.previous;
1795	pull.merged_by = Some(actor.username.clone());
1796	pull.merged_at = Some(now.clone());
1797	pull.updated_at = now;
1798	Ok(pull)
1799	}
1800
1801	async fn list_active_pulls(&self, a: ViewerArgs) -> Result<Vec<ActivePull>> {
1802	let Some(viewer) = a.viewer else {
1803	return Ok(Vec::new());
1804	};
1805	// The pull requests and their issues, in one round trip: their own,
1806	// and those g1t made for them (Pull::owner).
1807	let author = [JsValue::from(viewer.id.as_str())];
1808	let found = self
1809	.timing
1810	.db(
1811	2,
1812	self.db.batch(vec![
1813	self.db
1814	.prepare(format!(
1815	"SELECT {PULL_COLUMNS} FROM pulls
1816	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open')
1817	ORDER BY updated_at DESC LIMIT 50"
1818	))
1819	.bind(&author)?,
1820	self.db
1821	.prepare(format!(
1822	"SELECT {ISSUE_COLUMNS} FROM issues WHERE issues.id IN (
1823	SELECT issue_id FROM pulls
1824	WHERE COALESCE(requested_by_id, author_id) = ?1 AND status IN ('draft', 'open') AND issue_id IS NOT NULL
1825	ORDER BY updated_at DESC LIMIT 50)"
1826	))
1827	.bind(&author)?,
1828	]),
1829	)
1830	.await?;
1831	let (Some(found), Some(issues)) = (found.first(), found.get(1)) else {
1832	return Ok(Vec::new());
1833	};
1834	let snapshots = found.results::<Snapshot>()?;
1835	let pulls: Vec<Pull> = found.results::<PullRow>()?.into_iter().map(Pull::from).collect();
1836	let issues: Vec<Issue> = issues.results::<IssueRow>()?.into_iter().map(Issue::from).collect();
1837	let issues = &issues;
1838	// Where each stands: the remembered assessment when there is one,
1839	// and worked out otherwise.
1840	try_join_all(pulls.into_iter().zip(snapshots).map(\|(pull, snapshot)\| async move {
1841	let issue = pull.issue.and_then(\|number\| {
1842	issues
1843	.iter()
1844	.find(\|issue\| issue.repo_id == pull.repo_id && issue.number == number)
1845	.cloned()
1846	});
1847	// Only a pull request g1t is seeing through has a lifecycle.
1848	let lifecycle = if !lifecycle::made_by_g1t(&pull) \|\| snapshot.managed == 0 {
1849	None
1850	} else if let (Some(stage), Some(detail)) = (snapshot.stage, snapshot.stage_detail) {
1851	Some(Lifecycle {
1852	stage,
1853	detail,
1854	revisions: snapshot.revisions,
1855	})
1856	} else {
1857	let behind = self.is_behind(&pull.repo_id, &pull).await?;
1858	self.assess(&pull, &issue, behind)
1859	.await?
1860	.map(\|(lifecycle, _)\| lifecycle)
1861	};
1862	Ok::<_, worker::Error>(ActivePull {
1863	pull,
1864	issue,
1865	lifecycle,
1866	})
1867	}))
1868	.await
1869	}
1870
1871	// --- Sessions ----------------------------------------------------------
1872
1873	async fn append_session(&self, a: AppendSessionArgs) -> Result<Outcome<Appended>> {
1874	if a.entries.is_empty() {
1875	return Ok(Outcome::Ok(Appended { count: 0 }));
1876	}
1877	if a.entries.len() > MAX_ENTRY_BATCH {
1878	return Ok(Outcome::fail(
1879	FailureCode::Invalid,
1880	format!("Send at most {MAX_ENTRY_BATCH} entries at a time."),
1881	));
1882	}
1883	let viewer = Some(a.actor.clone());
1884	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
1885	if !pull.is_owned_by(&a.actor.id) {
1886	return Ok(Outcome::fail(
1887	FailureCode::Forbidden,
1888	"Only whoever opened a pull request, or asked g1t for it, can record its session.",
1889	));
1890	}
1891
1892	let now = rfc3339(now_ms());
1893	let count = a.entries.len() as u32;
1894	let mut statements = Vec::with_capacity(a.entries.len() + 1);
1895	for entry in a.entries {
1896	let kind = serde_json::to_value(entry.kind)?;
1897	let text: String = entry.text.chars().take(MAX_ENTRY_CHARS).collect();
1898	// Each insert takes the next sequence number itself, so two
1899	// writers appending at once cannot collide.
1900	statements.push(
1901	self.db
1902	.prepare(
1903	"INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
1904	SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
1905	FROM session_entries WHERE pull_id = ?",
1906	)
1907	.bind(&[
1908	pull.id.as_str().into(),
1909	kind.as_str().unwrap_or("note").into(),
1910	text.into(),
1911	optional(&entry.tool),
1912	optional(&entry.commit.or_else(\|\| pull.head_commit.clone())),
1913	now.as_str().into(),
1914	pull.id.as_str().into(),
1915	])?,
1916	);
1917	}
1918	statements.push(
1919	self.db
1920	.prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
1921	.bind(&[now.as_str().into(), pull.id.as_str().into()])?,
1922	);
1923	self.db.batch(statements).await?;
1924	self.publish(
1925	"session.appended",
1926	&pull.repo_id,
1927	&a.actor,
1928	SessionAppended {
1929	pull_id: pull.id.clone(),
1930	repo_id: pull.repo_id.clone(),
1931	number: pull.number,
1932	count,
1933	},
1934	)
1935	.await?;
1936	Ok(Outcome::Ok(Appended { count }))
1937	}
1938
1939	/// Adds entries to a pull request's session, each taking the next
1940	/// sequence number, without announcing it.
1941	pub(crate) async fn append_entries(&self, pull: &Pull, entries: &[NewSessionEntry]) -> Result<()> {
1942	let now = rfc3339(now_ms());
1943	let mut statements = Vec::with_capacity(entries.len());
1944	for entry in entries {
1945	let kind = serde_json::to_value(entry.kind)?;
1946	let text: String = entry.text.chars().take(MAX_ENTRY_CHARS).collect();
1947	statements.push(
1948	self.db
1949	.prepare(
1950	"INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
1951	SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
1952	FROM session_entries WHERE pull_id = ?",
1953	)
1954	.bind(&[
1955	pull.id.as_str().into(),
1956	kind.as_str().unwrap_or("note").into(),
1957	text.into(),
1958	optional(&entry.tool),
1959	optional(&entry.commit.clone().or_else(\|\| pull.head_commit.clone())),
1960	now.as_str().into(),
1961	pull.id.as_str().into(),
1962	])?,
1963	);
1964	}
1965	self.db.batch(statements).await?;
1966	Ok(())
1967	}
1968
1969	async fn read_session(&self, a: ViewArgs) -> Result<Outcome<Vec<SessionEntry>>> {
1970	let (_, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
1971	let rows = self
1972	.db
1973	.prepare(
1974	"SELECT seq, kind, text, tool, \"commit\", at FROM session_entries
1975	WHERE pull_id = ? AND seq > ? ORDER BY seq LIMIT ?",
1976	)
1977	.bind(&[pull.id.into(), a.after_seq.into(), SESSION_PAGE.into()])?
1978	.all()
1979	.await?
1980	.results::<SessionRow>()?;
1981	Ok(Outcome::Ok(
1982	rows.into_iter().map(SessionEntry::from).collect(),
1983	))
1984	}
1985
1986	/// A push moves the head of the pull request it concerns: the one whose
1987	/// fork was pushed to, or the one opened from the branch that moved.
1988	async fn on_event(&self, event: &Event) -> Result<()> {
1989	if event.kind != "git.push" {
1990	return Ok(());
1991	}
1992	let (Some(repo_id), Some(after), Some(git_ref)) = (
1993	event.repo_id.as_deref(),
1994	event.data["after"].as_str(),
1995	event.data["ref"].as_str(),
1996	) else {
1997	return Ok(());
1998	};
1999	let now = rfc3339(now_ms());
2000	// The head moved, so whatever the checks said no longer applies, and
2001	// whatever step g1t was waiting on has been taken.
2002	let moved = "UPDATE pulls
2003	SET head_commit = ?, updated_at = ?, check_status = NULL, check_run_id = NULL,
2004	working_on = NULL, working_until = NULL, stalled = NULL";
2005	let active = "status IN ('draft', 'open') AND head_commit IS NOT ?";
2006	let returning =
2007	"RETURNING id, repo_id, number, issue_number, status, author_id, author_name, requested_by_id, requested_by_name";
2008	let mut pulls: Vec<MovedRow> = Vec::new();
2009	// A fork carries its pull request on its default branch.
2010	if event.data["defaultBranch"].as_bool() == Some(true) {
2011	pulls.extend(
2012	self.db
2013	.prepare(format!(
2014	"{moved} WHERE fork_repo_id = ? AND {active} {returning}"
2015	))
2016	.bind(&[
2017	after.into(),
2018	now.as_str().into(),
2019	repo_id.into(),
2020	after.into(),
2021	])?
2022	.all()
2023	.await?
2024	.results::<MovedRow>()?,
2025	);
2026	}
2027	if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
2028	pulls.extend(
2029	self.db
2030	.prepare(format!(
2031	"{moved} WHERE repo_id = ? AND source_branch = ? AND {active} {returning}"
2032	))
2033	.bind(&[
2034	after.into(),
2035	now.as_str().into(),
2036	repo_id.into(),
2037	branch.into(),
2038	after.into(),
2039	])?
2040	.all()
2041	.await?
2042	.results::<MovedRow>()?,
2043	);
2044	}
2045	// What each now changes, so overlaps show while the work is under way.
2046	for moved in &pulls {
2047	if let Some(pull) = self.pull_by_id(&moved.id).await? {
2048	self.refresh_files(&pull).await?;
2049	}
2050	}
2051	// A merge that was waiting for this push to bring it up to date.
2052	for moved in &pulls {
2053	self.land_if_requested(&moved.id).await?;
2054	}
2055	// Whether each still merges cleanly, and, when a default branch
2056	// moved, every open pull request into it.
2057	let moved_ids: Vec<String> = pulls.iter().map(\|pull\| pull.id.clone()).collect();
2058	self.after_push(repo_id, event.data["defaultBranch"].as_bool() == Some(true), &moved_ids)
2059	.await;
2060	// A draft is announced when it is marked ready instead.
2061	for pull in pulls
2062	.into_iter()
2063	.filter(\|pull\| pull.status == PullStatus::Open)
2064	{
2065	self.publish_as(
2066	"pull.updated",
2067	&pull.repo_id,
2068	event.actor.clone(),
2069	PullEvent {
2070	author: Some(g1t_contracts::credentials::Principal { id: pull.author_id, username: pull.author_name }),
2071	requested_by: pull
2072	.requested_by_id
2073	.zip(pull.requested_by_name)
2074	.map(\|(id, username)\| g1t_contracts::credentials::Principal { id, username }),
2075	pull_id: pull.id,
2076	repo_id: pull.repo_id.clone(),
2077	number: pull.number,
2078	issue: pull.issue_number,
2079	commit: Some(after.to_owned()),
2080	..PullEvent::default()
2081	},
2082	)
2083	.await?;
2084	}
2085	Ok(())
2086	}
2087	}
2088
2089	fn service(env: &Env) -> Result<Work> {
2090	Ok(Work {
2091	db: env.d1("DB")?,
2092	identity: env.service("IDENTITY")?,
2093	repos: env.service("REPOS")?,
2094	events: env.service("EVENTS")?,
2095	actions: env.service("ACTIONS")?,
2096	timing: g1t_kit::d1::Timing::default(),
2097	prefetched: std::cell::RefCell::new(None),
2098	})
2099	}
2100
2101	#[event(fetch)]
2102	async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
2103	let Some(method) = rpc_method(&request) else {
2104	return Response::error("Not found", 404);
2105	};
2106	// A replica near the caller when it asks for one (crates/kit/src/d1.rs).
2107	let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
2108	let body: serde_json::Value = request.json().await?;
2109	let mut work = service(&env)?;
2110	work.db = db;
2111
2112	let answered = match method.as_str() {
2113	"open_issue" => reply(&work.open_issue(args(body)?).await?),
2114	"delegate_issue" => reply(&work.delegate_issue(args(body)?).await?),
2115	"report_confidence" => reply(&work.report_confidence(args(body)?).await?),
2116	"list_issues" => reply(&work.list_issues(args(body)?).await?),
2117	"get_issue" => reply(&work.get_issue(args(body)?).await?),
2118	"update_issue" => reply(&work.update_issue(args(body)?).await?),
2119	"close_issue" => reply(&work.close_issue(args(body)?).await?),
2120	"reopen_issue" => reply(&work.reopen_issue(args(body)?).await?),
2121	"list_labels" => reply(&work.list_labels(args(body)?).await?),
2122	"counts" => reply(&work.counts(args(body)?).await?),
2123	"add_comment" => reply(&work.add_comment(args(body)?).await?),
2124	"start_checks" => reply(&work.start_checks(args(body)?).await?),
2125	"seen_checks" => reply(&work.seen_checks(args(body)?).await?),
2126	"report_checks" => reply(&work.report_checks(args(body)?).await?),
2127	"set_commit_status" => reply(&work.set_commit_status(args(body)?).await?),
2128	"start_review" => reply(&work.start_review(args(body)?).await?),
2129	"advance" => reply(&work.advance(args(body)?).await?),
2130	"stall" => reply(&work.stall(args(body)?).await?),
2131	"managed_pulls" => reply(&work.managed_pulls(args(body)?).await?),
2132	"queue" => reply(&work.queue(args(body)?).await?),
2133	"queue_build" => reply(&work.queue_build(args(body)?).await?),
2134	"report_queue" => reply(&work.report_queue(args(body)?).await?),
2135	"remove_from_queue" => reply(&work.remove_from_queue(args(body)?).await?),
2136	"message_agent" => reply(&work.message_agent(args(body)?).await?),
2137	"locate_pull" => reply(&work.locate_pull(args(body)?).await?),
2138	"inbox_subject" => reply(&work.inbox_subject(args(body)?).await?),
2139	"answer_message" => reply(&work.answer_message(args(body)?).await?),
2140	"take_messages" => reply(&work.take_messages(args(body)?).await?),
2141	"wake_for_messages" => reply(&work.wake_for_messages(args(body)?).await?),
2142	"catch_up_job" => reply(&work.catch_up_job(args(body)?).await?),
2143	"get_settings" => reply(&work.get_settings(args(body)?).await?),
2144	"update_settings" => reply(&work.update_settings(args(body)?).await?),
2145	"report_review" => reply(&work.report_review(args(body)?).await?),
2146	"open_pull" => reply(&work.open_pull(args(body)?).await?),
2147	"list_pulls" => reply(&work.list_pulls(args(body)?).await?),
2148	"pulls_for_repos" => reply(&work.pulls_for_repos(args(body)?).await?),
2149	"get_pull" => reply(&work.get_pull(args(body)?).await?),
2150	"update_pull" => reply(&work.update_pull(args(body)?).await?),
2151	"catch_up_pull" => reply(&work.catch_up_pull(args(body)?).await?),
2152	"ready_pull" => reply(&work.ready_pull(args(body)?).await?),
2153	"close_pull" => reply(&work.close_pull(args(body)?).await?),
2154	"merge_pull" => reply(&work.merge_pull(args(body)?).await?),
2155	"list_active_pulls" => reply(&work.list_active_pulls(args(body)?).await?),
2156	"by_author" => reply(&work.by_author(args(body)?).await?),
2157	"start_plan" => reply(&work.start_plan(args(body)?).await?),
2158	"report_plan" => reply(&work.report_plan(args(body)?).await?),
2159	"get_plan" => reply(&work.get_plan(args(body)?).await?),
2160	"list_plans" => reply(&work.list_plans(args(body)?).await?),
2161	"apply_plan" => reply(&work.apply_plan(args(body)?).await?),
2162	"queue_issue" => reply(&work.queue_issue(args(body)?).await?),
2163	"ready_issues" => reply(&work.ready_issues(args(body)?).await?),
2164	"list_assigned_issues" => reply(&work.list_assigned_issues(args(body)?).await?),
2165	"append_session" => reply(&work.append_session(args(body)?).await?),
2166	"read_session" => reply(&work.read_session(args(body)?).await?),
2167	// Agents at work, their sessions, and memory (runs.rs, memory.rs).
2168	"open_run" => reply(&work.open_run(args(body)?).await?),
2169	"report_run" => reply(&work.report_run(args(body)?).await?),
2170	"stop_run" => reply(&work.stop_run(args(body)?).await?),
2171	"list_runs" => reply(&work.list_runs(args(body)?).await?),
2172	"get_run" => reply(&work.get_run(args(body)?).await?),
2173	"list_sessions" => reply(&work.list_sessions(args(body)?).await?),
2174	"get_session" => reply(&work.get_session(args(body)?).await?),
2175	"list_memories" => reply(&work.list_memories(args(body)?).await?),
2176	"add_memory" => reply(&work.add_memory(args(body)?).await?),
2177	"update_memory" => reply(&work.update_memory(args(body)?).await?),
2178	"delete_memory" => reply(&work.delete_memory(args(body)?).await?),
2179	"recall" => reply(&work.recall(args(body)?).await?),
2180	"memory_context" => reply(&work.memory_context(args(body)?).await?),
2181	// What agents may do in a sandbox (guardrails.rs).
2182	"get_guardrails" => reply(&work.get_guardrails(args(body)?).await?),
2183	"update_guardrails" => reply(&work.update_guardrails(args(body)?).await?),
2184	"run_guardrails" => reply(&work.run_guardrails(args(body)?).await?),
2185	// Plan caps the runner applies (compute.rs).
2186	"active_agents" => reply(&work.active_agents(args(body)?).await?),
2187	"issue_spend" => reply(&work.issue_spend(args(body)?).await?),
2188	"wait_for_slot" => reply(&work.wait_for_slot(args(body)?).await?),
2189	"agent_comment" => reply(&work.agent_comment(args(body)?).await?),
2190	"add_wait" => reply(&work.add_wait(args(body)?).await?),
2191	"waiting_workspaces" => reply(&work.waiting_workspaces(args(body)?).await?),
2192	"take_wait" => reply(&work.take_wait(args(body)?).await?),
2193	// The runs whose sandboxes stop with their repository (retired.rs).
2194	"runs_in_repo" => reply(&work.runs_in_repo(args(body)?).await?),
2195	"run_cost" => reply(&work.run_cost(args(body)?).await?),
2196	"start_mergecheck" => reply(&work.start_mergecheck(args(body)?).await?),
2197	"report_mergecheck" => reply(&work.report_mergecheck(args(body)?).await?),
2198	// Memory that fills itself, and its review queue (capture.rs).
2199	method if capture::METHODS.contains(&method) => capture::dispatch(&work, method, body).await,
2200	// @g1t in comments, and the label rule (mentions.rs).
2201	"take_mention" => reply(&work.take_mention(args(body)?).await?),
2202	"mention_revision" => reply(&work.mention_revision(args(body)?).await?),
2203	"reply_mention" => reply(&work.reply_mention(args(body)?).await?),
2204	"get_agent_rules" => reply(&work.get_agent_rules(args(body)?).await?),
2205	"set_agent_rules" => reply(&work.set_agent_rules(args(body)?).await?),
2206	_ => Response::error("Unknown method", 404),
2207	};
2208	served.finish_timed(answered, &work.timing)
2209	}
2210
2211	/// Events from the bus, delivered on this service's own queue.
2212	#[event(queue)]
2213	async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
2214	let work = service(&env)?;
2215	for message in batch.messages()? {
2216	// A workspace renamed: its agent runs and memory move to the slug it has now.
2217	if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), &[memory::RENAMED, guardrails::RENAMED].concat()).await? {
2218	message.ack();
2219	continue;
2220	}
2221	// A repository renamed or transferred: its runs, memory, guardrails
2222	// and runs waiting for a slot follow.
2223	if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), &[memory::TRANSFERRED, guardrails::TRANSFERRED, retired::WAITS_MOVED].concat()).await? {
2224	message.ack();
2225	continue;
2226	}
2227	// A workspace deleted: what it kept for itself goes.
2228	if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), memory::DELETED).await? {
2229	message.ack();
2230	continue;
2231	}
2232	// A repository deleted, archived or purged, or a branch renamed (retired.rs).
2233	if work.on_retired(message.body()).await? {
2234	message.ack();
2235	continue;
2236	}
2237	capture::on_event(&work, message.body()).await;
2238	work.on_event(message.body()).await?;
2239	message.ack();
2240	}
2241	Ok(())
2242	}
2243
2244	/// The rules that once read a pull request's author read its owner now:
2245	/// whoever asked g1t for it, or its author. For each, the person who asked
2246	/// is held to what an author was, and g1t's agent (a token it works with)
2247	/// gains nothing by being the author.
2248	#[cfg(test)]
2249	mod owner_rules {
2250	use super::*;
2251	use crate::rows::stored::{ASKER, G1T, pull};
2252	use g1t_contracts::identity::AGENT_ID;
2253
2254	const SOMEONE: &str = "usr_2";
2255
2256	#[test]
2257	fn no_self_approval() {
2258	// add_comment refuses a verdict on one that is theirs.
2259	let made = pull(G1T, Some(ASKER));
2260	assert!(made.is_owned_by(ASKER.0), "the person who asked cannot approve it");
2261	assert!(!made.is_owned_by(AGENT_ID), "g1t's review agent still gives its verdict");
2262	assert!(!made.is_owned_by(SOMEONE));
2263	}
2264
2265	#[test]
2266	fn what_an_author_could_do_without_a_role() {
2267	// manageable_pull (update, ready, close), catch_up_pull on a fork,
2268	// append_session, and steering with message_agent: theirs to do.
2269	let made = pull(G1T, Some(ASKER));
2270	assert!(made.is_owned_by(ASKER.0));
2271	assert!(!made.is_owned_by(SOMEONE), "anyone else still needs the role");
2272	assert!(!made.is_owned_by(AGENT_ID), "being its author gives g1t's tokens nothing more");
2273	}
2274
2275	#[test]
2276	fn nobody_is_asked_to_review_what_they_asked_for() {
2277	// update_pull drops the owner from the reviewers asked.
2278	let made = pull(G1T, Some(ASKER));
2279	let mut reviewers = vec!["syntaqx".to_owned(), "ana".to_owned()];
2280	reviewers.retain(\|name\| *name != made.owner().username);
2281	assert_eq!(reviewers, ["ana"]);
2282	}
2283
2284	#[test]
2285	fn sandboxes_act_as_whoever_asked() {
2286	// LifecycleJob, ReviewJob, MergecheckJob and the merge queue's job
2287	// carry who the sandbox's credential acts for: a real account.
2288	let made = pull(G1T, Some(ASKER));
2289	assert_eq!(made.owner().id, ASKER.0);
2290	let acts_as = made.requested_by.unwrap_or(made.author);
2291	assert_eq!(acts_as.id, ASKER.0);
2292	// g1t's own work, which nobody asked for, acts as g1t, as before.
2293	let own = pull(("g1t", "g1t"), None);
2294	assert_eq!(own.requested_by.unwrap_or(own.author).id, "g1t");
2295	}
2296
2297	#[test]
2298	fn events_name_g1t_and_whoever_asked() {
2299	let made = pull(G1T, Some(ASKER));
2300	let event = serde_json::to_value(Work::pull_event(&made)).unwrap();
2301	assert_eq!(event["author"], serde_json::json!({ "id": AGENT_ID, "username": "g1t" }));
2302	assert_eq!(event["requestedBy"], serde_json::json!({ "id": "usr_1", "username": "syntaqx" }));
2303	let own = serde_json::to_value(Work::pull_event(&pull(ASKER, None))).unwrap();
2304	assert!(own.get("requestedBy").is_none());
2305	}
2306	}