Skip to content
1,941 linesCodeBlameRaw
1//! Seeing a pull request through. Once a g1t agent has made a change, g1t
2//! takes each remaining step itself: waiting for the checks the
3//! repository's workflows report on it, a review by another agent, sending
4//! the author back to address a failed check (with what its jobs printed)
5//! or the review, and catching up when the branch it would land on has
6//! moved. It stops when the pull request meets everything the default
7//! branch's protection requires, or when it has tried and a person has to
8//! decide.
9//!
10//! This service decides what the next step is and claims it. The runner
11//! service asks, on every event that could change the answer, and carries
12//! the step out in a sandbox.
13
14use g1t_contracts::repos::{GetByIdArgs, Repo, RepoPath};
15use g1t_contracts::time::rfc3339;
16use g1t_contracts::work::*;
17use g1t_contracts::{FailureCode, Membership, Outcome, User, Viewer};
18use g1t_kit::now_ms;
19use std::collections::HashMap;
20
21use serde::Deserialize;
22use worker::Result;
23use worker::wasm_bindgen::JsValue;
24
25use crate::Work;
26use crate::prefetch::Slot;
27use crate::reviews::{AGENT_ID, AGENT_NAME};
28use crate::statuses::{self, WorkflowFacts};
29use crate::rows::ValueRow;
30
31/// How long a claimed step is waited for before it may be taken again.
32const REVIEW_MINUTES: u64 = 20;
33const REVISION_MINUTES: u64 = 60;
34const CATCH_UP_MINUTES: u64 = 30;
35const MERGE_MINUTES: u64 = 2;
36/// Who a merge made by a repository's settings is attributed to. Not an
37/// account: `g1t` cannot be registered.
38pub(crate) const POLICY_ACTOR_ID: &str = "g1t_policy";
39pub(crate) const POLICY_ACTOR_NAME: &str = "g1t";
40/// How much of a failed check's output the author is shown.
41const MAX_CHECK_OUTPUT_CHARS: usize = 4_000;
42/// How many failed jobs' logs the author is shown, and how much of each.
43const MAX_FAILED_JOBS: usize = 3;
44const MAX_JOB_LOG_CHARS: usize = 3_000;
45/// The most pages of a job's log read to find its end.
46const MAX_LOG_PAGES: usize = 6;
47const MANAGED_PAGE: u32 = 200;
48
49/// The part of a pull request's row that tracks its lifecycle.
50#[derive(Deserialize)]
51struct Progress {
52 managed: u8,
53 revisions: u32,
54 revised_at: Option<String>,
55 working_on: Option<String>,
56 working_until: Option<String>,
57 stalled: Option<String>,
58}
59
60/// A merge asked for while the pull request was behind, as stored.
61#[derive(serde::Serialize, Deserialize)]
62#[serde(rename_all = "camelCase")]
63struct LandRequest {
64 actor: User,
65 keep_issue_open: bool,
66}
67
68#[derive(Deserialize)]
69struct LandRow {
70 land_requested: Option<String>,
71 land_requested_at: Option<String>,
72 stalled: Option<String>,
73}
74
75#[derive(Deserialize)]
76struct FinishedReview {
77 finished_at: String,
78 verdict: Option<Verdict>,
79}
80
81#[derive(Deserialize)]
82struct ReviewNote {
83 body: String,
84 path: Option<String>,
85 line: Option<u32>,
86}
87
88/// What the author is being sent back to address.
89pub(crate) enum Feedback {
90 /// The merge queue took it out: its combined state failed.
91 FailedChecks,
92 /// Checks that failed on its head.
93 FailedWorkflows,
94 /// The review that finished at this time.
95 Review(String),
96 /// What a person who asked for changes wrote since the last revision.
97 Person(PersonRequest),
98}
99
100/// A person's request for changes that still stands: their latest verdict,
101/// made after the agent last revised.
102#[derive(Clone, Debug, Deserialize)]
103pub(crate) struct PersonRequest {
104 author_id: String,
105 author_name: String,
106 created_at: String,
107}
108
109/// What should happen next, if it is g1t's turn.
110pub(crate) enum Next {
111 /// A step is under way, or it is a person's turn.
112 Wait,
113 Review,
114 Revise(Feedback),
115 CatchUp,
116 /// Land it, because the repository says ready pull requests land.
117 Merge,
118}
119
120/// Whether g1t made this pull request, and so sees it through.
121pub(crate) fn made_by_g1t(pull: &Pull) -> bool {
122 pull.runtime == Runtime::Hosted && pull.agent == AGENT_NAME && pull.fork.is_some()
123}
124
125/// Whether a verdict by `reviewer` is a person's other than the pull
126/// request's owner (whoever asked g1t for it, or its author): neither
127/// theirs nor g1t's agent's.
128pub(crate) fn from_someone_else(pull: &Pull, reviewer: &str) -> bool {
129 !pull.is_owned_by(reviewer) && reviewer != AGENT_ID
130}
131
132fn at(stage: Stage, detail: impl Into<String>, revisions: u32) -> Lifecycle {
133 Lifecycle {
134 stage,
135 detail: detail.into(),
136 revisions,
137 }
138}
139
140fn times(count: u32) -> String {
141 match count {
142 1 => "once".to_owned(),
143 2 => "twice".to_owned(),
144 count => format!("{count} times"),
145 }
146}
147
148/// Everything the next step depends on.
149struct Facts {
150 /// Still being made: not yet marked ready for review.
151 draft: bool,
152 /// Why g1t stopped, if it has.
153 stalled: Option<String>,
154 /// The step under way, if one was claimed and is still being waited for.
155 working_on: Option<String>,
156 /// `failed` when the merge queue took it out.
157 check_status: Option<CheckStatus>,
158 /// A review someone asked for is being written.
159 review_pending: bool,
160 revisions: u32,
161 /// The latest finished review of the change as it is now.
162 review: Option<FinishedReview>,
163 /// Whether the branch it would land on has moved without it.
164 behind: bool,
165 /// Whether it is known to conflict with the branch it would land on.
166 conflicting: bool,
167 /// Whether the repository lands a ready pull request by itself.
168 auto_merge: bool,
169 /// Whether the repository refuses to merge one that is behind.
170 require_up_to_date: bool,
171 /// Whether a second agent reviews it without being asked.
172 agent_review: bool,
173 /// How many times the author may be sent back.
174 max_revisions: u32,
175 /// What the repository's approval rule still wants, if anything.
176 approvals_missing: Option<String>,
177 /// A person asked for changes since the agent last revised.
178 person_request: Option<PersonRequest>,
179 /// Its place in the merge queue, and what is ahead of it there.
180 queued: Option<(QueueState, Vec<u32>)>,
181 /// What the checks on its head say, against the required ones.
182 workflows: WorkflowFacts,
183 /// Why the agent's change has low confidence, when the repository asks
184 /// a person before merging one and no person has approved it since.
185 low_confidence: Option<String>,
186}
187
188/// Where a pull request stands, and the step to take if it is g1t's turn.
189///
190/// The order is: nothing while a step is under way; a person asking for
191/// changes is answered first; its checks must finish, and the required
192/// ones pass; then a review must approve; then it must be up to date. A
193/// person's or an agent's request for changes, or a failed check, sends the
194/// author back, a limited number of times, after which a person is asked.
195/// A check the branch does not require stops holding it once the
196/// revisions run out.
197fn decide(facts: Facts) -> (Lifecycle, Next) {
198 let revisions = facts.revisions;
199 let wait = |stage, detail: &str| (at(stage, detail, revisions), Next::Wait);
200 let exhausted = revisions >= facts.max_revisions;
201
202 if facts.draft {
203 return wait(Stage::Working, "g1t is making the change.");
204 }
205 if let Some(reason) = &facts.stalled {
206 return wait(Stage::NeedsYou, reason);
207 }
208 if let Some((state, ahead)) = &facts.queued {
209 let named = ahead.iter().map(|n| format!("#{n}")).collect::<Vec<_>>().join(", ");
210 let detail = match (state, ahead.is_empty()) {
211 (QueueState::Testing, true) => "In the merge queue: being tested on the default branch as it is.".to_owned(),
212 (QueueState::Testing, false) => {
213 format!("In the merge queue: being tested together with {named}, ahead of it.")
214 }
215 (QueueState::Passed, true) => "Passed in the merge queue. Landing.".to_owned(),
216 (QueueState::Passed, false) => {
217 format!("Passed in the merge queue together with {named}. It lands once they have.")
218 }
219 _ => "In the merge queue, waiting for its turn to be tested.".to_owned(),
220 };
221 return wait(Stage::Queued, &detail);
222 }
223 match facts.working_on.as_deref() {
224 Some("revision") => {
225 return wait(
226 Stage::Revising,
227 "The agent is addressing what the checks or the review found.",
228 );
229 }
230 Some("catch_up") => {
231 return wait(
232 Stage::CatchingUp,
233 "The agent is merging in the branch this will land on, which has moved.",
234 );
235 }
236 Some("answer") => {
237 return wait(
238 Stage::Answering,
239 "The agent is answering what another agent asked it.",
240 );
241 }
242 Some("merge") => return wait(Stage::Ready, "Merging."),
243 Some(_) => return wait(Stage::Reviewing, "g1t is reviewing the change."),
244 None => {}
245 }
246 if facts.review_pending {
247 return wait(Stage::Reviewing, "g1t is reviewing the change.");
248 }
249 // A person asked for changes: the agent makes them, as it would for a
250 // review it asked for, before anything else.
251 if let Some(request) = &facts.person_request {
252 if exhausted {
253 return wait(
254 Stage::NeedsYou,
255 &format!(
256 "{} asked for changes, and the agent has already revised {}.",
257 request.author_name,
258 times(revisions)
259 ),
260 );
261 }
262 return (
263 at(
264 Stage::Revising,
265 format!(
266 "{} asked for changes. The agent is being sent back to make them.",
267 request.author_name
268 ),
269 revisions,
270 ),
271 Next::Revise(Feedback::Person(request.clone())),
272 );
273 }
274
275 // The merge queue took it out: its change failed together with what
276 // was ahead of it.
277 match facts.check_status {
278 Some(CheckStatus::Failed) if exhausted => {
279 return wait(
280 Stage::NeedsYou,
281 &format!("It failed in the merge queue after the agent revised {}.", times(revisions)),
282 );
283 }
284 Some(CheckStatus::Failed) => {
285 return (
286 at(
287 Stage::Revising,
288 "It failed in the merge queue. The agent is being sent back to fix it.",
289 revisions,
290 ),
291 Next::Revise(Feedback::FailedChecks),
292 );
293 }
294 Some(CheckStatus::Errored) => {
295 return wait(Stage::NeedsYou, "Its checks could not be run.");
296 }
297 _ => {}
298 }
299
300 // Its checks: the agent fixes any that failed. Once it is out of
301 // revisions, only the checks the branch requires still hold it.
302 if !facts.workflows.failed.is_empty() {
303 let failed = statuses::list(&facts.workflows.failed);
304 let required = facts.workflows.required_failed();
305 if !exhausted {
306 return (
307 at(
308 Stage::Revising,
309 format!("{failed} failed. The agent is being sent back to fix it."),
310 revisions,
311 ),
312 Next::Revise(Feedback::FailedWorkflows),
313 );
314 }
315 if !required.is_empty() {
316 return wait(
317 Stage::NeedsYou,
318 &format!(
319 "The required {} {} still {} after the agent revised {}.",
320 if required.len() == 1 { "check" } else { "checks" },
321 statuses::list(&required),
322 if required.len() == 1 { "fails" } else { "fail" },
323 times(revisions)
324 ),
325 );
326 }
327 }
328 if !facts.workflows.pending.is_empty() {
329 return wait(
330 Stage::Checking,
331 &format!("Waiting for {} to finish.", statuses::list(&facts.workflows.pending)),
332 );
333 }
334 let expected = facts.workflows.expected();
335 if !expected.is_empty() {
336 return wait(
337 Stage::Checking,
338 &format!(
339 "Waiting for the required {} {} to report on its latest commit.",
340 if expected.len() == 1 { "check" } else { "checks" },
341 statuses::list(&expected)
342 ),
343 );
344 }
345
346 // A second agent reviews it, unless the repository leaves review to people.
347 if facts.agent_review {
348 match facts.review {
349 None => {
350 return (
351 at(
352 Stage::Reviewing,
353 "g1t is about to review the change.",
354 revisions,
355 ),
356 Next::Review,
357 );
358 }
359 Some(FinishedReview { verdict: None, .. }) => {
360 return wait(Stage::NeedsYou, "The review could not be completed.");
361 }
362 Some(FinishedReview {
363 verdict: Some(Verdict::RequestChanges),
364 ..
365 }) if exhausted => {
366 return wait(
367 Stage::NeedsYou,
368 &format!(
369 "The review still asks for changes after the agent revised {}.",
370 times(revisions)
371 ),
372 );
373 }
374 Some(FinishedReview {
375 verdict: Some(Verdict::RequestChanges),
376 finished_at,
377 }) => {
378 return (
379 at(
380 Stage::Revising,
381 "The review asked for changes. The agent is being sent back to make them.",
382 revisions,
383 ),
384 Next::Revise(Feedback::Review(finished_at)),
385 );
386 }
387 Some(FinishedReview {
388 verdict: Some(Verdict::Approve),
389 ..
390 }) => {}
391 }
392 }
393
394 // A conflict found ahead of time is resolved before anything else that
395 // is left: it could not merge, by a person or by the queue, until then.
396 if facts.conflicting {
397 return (
398 at(
399 Stage::CatchingUp,
400 "It conflicts with the branch it will land on. The agent is merging that branch in and resolving the conflicts.",
401 revisions,
402 ),
403 Next::CatchUp,
404 );
405 }
406 // Only where the repository insists is catching up a step of its own,
407 // followed by its checks again. Elsewhere it happens as part of merging.
408 if facts.behind && facts.require_up_to_date {
409 return (
410 at(
411 Stage::CatchingUp,
412 "The branch it will land on has moved. The agent is catching up.",
413 revisions,
414 ),
415 Next::CatchUp,
416 );
417 }
418 // The repository wants approvals this does not have yet: people's turn,
419 // so it is shown as needing someone, not as g1t still working.
420 if let Some(missing) = &facts.approvals_missing {
421 return wait(Stage::NeedsYou, missing);
422 }
423 // Everything else is met, but g1t is not sure of the change: a person
424 // decides, rather than auto-merge or the queue.
425 if let Some(reasons) = &facts.low_confidence {
426 return wait(
427 Stage::NeedsYou,
428 &format!(
429 "The agent's confidence in this change is low ({reasons}). This repository asks a person before merging it: approve it to let it land, or ask for changes."
430 ),
431 );
432 }
433 if facts.auto_merge {
434 return (
435 at(
436 Stage::Ready,
437 "Everything this repository asks for is met. Merging, as its settings say.",
438 revisions,
439 ),
440 Next::Merge,
441 );
442 }
443 wait(
444 Stage::Ready,
445 if facts.behind {
446 "Ready to merge. Merging brings it up to date with the default branch first."
447 } else {
448 "Everything this repository asks for is met. Ready to merge."
449 },
450 )
451}
452
453impl Work {
454 /// Where a pull request stands and what g1t does next, remembered so
455 /// lists can show it without working it out again. `None` for one g1t
456 /// is not seeing through.
457 pub(crate) async fn assess(
458 &self,
459 pull: &Pull,
460 issue: &Option<Issue>,
461 behind: bool,
462 ) -> Result<Option<(Lifecycle, Next)>> {
463 Ok(self
464 .assess_with_confidence(pull, issue, behind)
465 .await?
466 .map(|(lifecycle, next, _)| (lifecycle, next)))
467 }
468
469 /// [`Self::assess`], with how sure g1t is of the change once the agent
470 /// has finished it.
471 pub(crate) async fn assess_with_confidence(
472 &self,
473 pull: &Pull,
474 issue: &Option<Issue>,
475 behind: bool,
476 ) -> Result<Option<(Lifecycle, Next, Option<Confidence>)>> {
477 let assessed = self.assess_now(pull, issue, behind).await?;
478 if let Some((lifecycle, _, _)) = &assessed
479 && !self.remembered_as(&pull.id, lifecycle)?
480 {
481 self.remember(&pull.id, lifecycle).await?;
482 }
483 Ok(assessed)
484 }
485
486 /// Whether the row read for this request already says `lifecycle`, so
487 /// that showing a pull request does not write it again unchanged.
488 fn remembered_as(&self, pull_id: &str, lifecycle: &Lifecycle) -> Result<bool> {
489 let Some(found) = self.prefetched_pull(pull_id) else {
490 return Ok(false);
491 };
492 Ok(found
493 .first::<crate::rows::Snapshot>(Slot::Pull)?
494 .is_some_and(|stored| {
495 stored.stage == Some(lifecycle.stage) && stored.stage_detail.as_deref() == Some(lifecycle.detail.as_str())
496 }))
497 }
498
499 /// Saves where a pull request stands, for [`Self::remembered`].
500 pub(crate) async fn remember(&self, pull_id: &str, lifecycle: &Lifecycle) -> Result<()> {
501 let stage = serde_json::to_value(lifecycle.stage)?;
502 self.db
503 .prepare("UPDATE pulls SET stage = ?, stage_detail = ? WHERE id = ?")
504 .bind(&[
505 stage.as_str().unwrap_or_default().into(),
506 lifecycle.detail.as_str().into(),
507 pull_id.into(),
508 ])?
509 .run()
510 .await?;
511 Ok(())
512 }
513
514 async fn assess_now(
515 &self,
516 pull: &Pull,
517 // What done means is in its body, for the agent; the merge waits on
518 // the branch's required checks, not on the issue.
519 _issue: &Option<Issue>,
520 behind: bool,
521 ) -> Result<Option<(Lifecycle, Next, Option<Confidence>)>> {
522 if !pull.status.is_active() {
523 return Ok(None);
524 }
525 // As read for this request (prefetch.rs), or now.
526 let prefetched = self.prefetched_pull(&pull.id);
527 let progress = match &prefetched {
528 Some(found) => found.first::<Progress>(Slot::Pull)?,
529 None => {
530 self.db
531 .prepare(
532 "SELECT managed, revisions, revised_at, working_on, working_until, stalled
533 FROM pulls WHERE id = ?",
534 )
535 .bind(&[pull.id.as_str().into()])?
536 .first::<Progress>(None)
537 .await?
538 }
539 };
540 let Some(progress) = progress.filter(|progress| progress.managed != 0) else {
541 return Ok(None);
542 };
543 let now = rfc3339(now_ms());
544 let working_on = progress
545 .working_until
546 .as_deref()
547 .is_some_and(|until| until > now.as_str())
548 .then(|| progress.working_on.clone().unwrap_or_default());
549 let review = match &prefetched {
550 Some(found) => found.first::<FinishedReview>(Slot::Review)?,
551 None => {
552 self.db
553 .prepare(
554 "SELECT finished_at, verdict FROM review_runs
555 WHERE pull_id = ? AND finished_at IS NOT NULL ORDER BY id DESC LIMIT 1",
556 )
557 .bind(&[pull.id.as_str().into()])?
558 .first::<FinishedReview>(None)
559 .await?
560 }
561 }
562 // A review of what the change was before its last revision says
563 // nothing about what it is now.
564 .filter(|review| {
565 progress
566 .revised_at
567 .as_deref()
568 .is_none_or(|revised| review.finished_at.as_str() >= revised)
569 });
570 let settings = self.settings_for(pull).await?;
571 let statuses = self.statuses(&pull.repo_id, pull.head_commit.as_deref()).await?;
572 let workflows = self.workflow_facts(&pull.repo_id, pull.head_commit.as_deref(), &statuses, &settings.required_checks).await?;
573 // Once the agent has finished the change: how sure g1t is of it, and
574 // whether that holds it for a person.
575 let (confidence, low_confidence) = if pull.status == PullStatus::Draft {
576 (None, None)
577 } else {
578 let review_comments = match &review {
579 Some(review) => self.review_comments(pull, &review.finished_at).await?,
580 None => 0,
581 };
582 let confidence = self
583 .assess_confidence(
584 pull,
585 crate::confidence::Signals {
586 required: crate::confidence::RequiredSignal::of(&workflows.required),
587 queue_failed: pull.check_status == Some(CheckStatus::Failed),
588 revisions: progress.revisions,
589 agent_review: settings.agent_review,
590 review: review.as_ref().and_then(|review| review.verdict),
591 review_comments,
592 ..Default::default()
593 },
594 )
595 .await?;
596 let held = settings.hold_low_confidence
597 && confidence.level == ConfidenceLevel::Low
598 && !self.person_approved(pull, progress.revised_at.as_deref()).await?;
599 let reasons = held.then(|| confidence.reasons.join(", "));
600 (Some(confidence), reasons)
601 };
602 // The rules of the branch it merges into, as g1t sees them when it
603 // merges by itself: what people must still do, whether an agent's
604 // change may land here unattended, and a cost cap that holds the
605 // agent until a person approves.
606 let gate = match self.repo_for(pull).await? {
607 Some(repo) => Some(self.merge_gate(&repo, pull, Some(&User::system(&repo.namespace)), false, true).await?),
608 None => None,
609 };
610 let level = confidence.as_ref().map(|confidence| confidence.level).or(pull.confidence.as_ref().map(|c| c.level));
611 let rules_allow_auto_merge = gate
612 .as_ref()
613 .is_none_or(|gate| g1t_rules::merge::auto_merge_refusal(&gate.requirements, level).is_none());
614 let held = gate.as_ref().and_then(|gate| {
615 g1t_rules::outcome::blocking(&gate.judged)
616 .into_iter()
617 .find(|violation| violation.rule == "cost_cap")
618 .map(|violation| format!("{} {}", violation.message, violation.remedy))
619 });
620 let (lifecycle, next) = decide(Facts {
621 draft: pull.status == PullStatus::Draft,
622 stalled: progress.stalled.or(held),
623 working_on,
624 check_status: pull.check_status,
625 review_pending: self.review_pending(&pull.id).await?,
626 revisions: progress.revisions,
627 review,
628 behind,
629 conflicting: self.conflicting_files(pull).await?.is_some(),
630 auto_merge: settings.auto_merge && rules_allow_auto_merge,
631 require_up_to_date: settings.require_up_to_date,
632 agent_review: settings.agent_review,
633 max_revisions: settings.max_revisions,
634 approvals_missing: gate.as_ref().and_then(|gate| gate.people_gap()),
635 person_request: self
636 .person_request(pull, progress.revised_at.as_deref())
637 .await?,
638 queued: self.queued_entry(&pull.id).await?,
639 workflows,
640 low_confidence,
641 });
642 Ok(Some((lifecycle, next, confidence)))
643 }
644
645 /// The latest request for changes by a person other than its owner
646 /// (whoever asked g1t for it, or its author), if it is that person's
647 /// latest verdict and came after the last revision.
648 async fn person_request(
649 &self,
650 pull: &Pull,
651 revised_at: Option<&str>,
652 ) -> Result<Option<PersonRequest>> {
653 #[derive(Deserialize)]
654 struct Verdicts {
655 author_id: String,
656 author_name: String,
657 verdict: Verdict,
658 created_at: String,
659 }
660 let rows = match self.prefetched_pull(&pull.id) {
661 Some(found) => found
662 .rows::<Verdicts>(Slot::Verdicts)?
663 .into_iter()
664 .filter(|row| from_someone_else(pull, &row.author_id))
665 .collect(),
666 None => self
667 .db
668 .prepare(
669 "SELECT author_id, author_name, verdict, created_at FROM comments
670 WHERE repo_id = ? AND number = ? AND verdict IS NOT NULL
671 AND author_id != ? AND author_id != ?
672 ORDER BY id",
673 )
674 .bind(&[
675 pull.repo_id.as_str().into(),
676 pull.number.into(),
677 pull.owner().id.as_str().into(),
678 AGENT_ID.into(),
679 ])?
680 .all()
681 .await?
682 .results::<Verdicts>()?,
683 };
684 // Each person's latest verdict is the one that stands.
685 let mut latest: HashMap<String, Verdicts> = HashMap::new();
686 for row in rows {
687 latest.insert(row.author_id.clone(), row);
688 }
689 Ok(latest
690 .into_values()
691 .filter(|row| row.verdict == Verdict::RequestChanges)
692 .filter(|row| revised_at.is_none_or(|revised| row.created_at.as_str() > revised))
693 .max_by(|a, b| a.created_at.cmp(&b.created_at))
694 .map(|row| PersonRequest {
695 author_id: row.author_id,
696 author_name: row.author_name,
697 created_at: row.created_at,
698 }))
699 }
700
701 /// Marks a pull request a g1t agent has just opened as one g1t sees
702 /// through.
703 pub(crate) async fn manage(&self, pull: &Pull) -> Result<()> {
704 if !made_by_g1t(pull) {
705 return Ok(());
706 }
707 self.db
708 .prepare("UPDATE pulls SET managed = 1 WHERE id = ?")
709 .bind(&[pull.id.as_str().into()])?
710 .run()
711 .await?;
712 Ok(())
713 }
714
715 /// Takes a step for a pull request, if nobody else has. One statement,
716 /// so that two callers cannot both take it.
717 pub(crate) async fn claim(&self, pull_id: &str, step: &str, minutes: u64, revising: bool) -> Result<bool> {
718 let now = now_ms();
719 let revision = if revising {
720 ", revisions = revisions + 1, revised_at = ?1"
721 } else {
722 ""
723 };
724 Ok(self
725 .db
726 .prepare(format!(
727 "UPDATE pulls SET working_on = ?2, working_until = ?3{revision}
728 WHERE id = ?4 AND status = 'open' AND stalled IS NULL
729 AND (working_until IS NULL OR working_until < ?1)
730 RETURNING id AS value"
731 ))
732 .bind(&[
733 rfc3339(now).into(),
734 step.into(),
735 rfc3339(now + minutes * 60 * 1000).into(),
736 pull_id.into(),
737 ])?
738 .first::<ValueRow>(None)
739 .await?
740 .is_some())
741 }
742
743 /// What the author is told when sent back: the checks that failed and
744 /// what their failing jobs printed, why the merge queue took it out, or
745 /// the review and its comments on lines.
746 async fn feedback(&self, pull: &Pull, feedback: &Feedback) -> Result<String> {
747 match feedback {
748 Feedback::FailedChecks => {
749 let run = self.latest_checks(&pull.id).await?;
750 // Why it failed, when that is more than a list of commands:
751 // the merge queue saying what broke in the combined state.
752 let why = run.as_ref().and_then(|run| run.error.clone()).map(|error| format!("{error}\n\n")).unwrap_or_default();
753 let failed: Vec<String> = run
754 .map(|run| run.results)
755 .unwrap_or_default()
756 .into_iter()
757 .filter(|result| !result.passed)
758 .map(|result| {
759 let length = result.output.chars().count();
760 let output: String = result
761 .output
762 .chars()
763 .skip(length.saturating_sub(MAX_CHECK_OUTPUT_CHARS))
764 .collect();
765 let exit = result
766 .exit_code
767 .map_or("it was stopped for taking too long".to_owned(), |code| {
768 format!("exit code {code}")
769 });
770 format!("`{}` failed ({exit}):\n\n{}", result.command, output.trim())
771 })
772 .collect();
773 if failed.is_empty() {
774 return Ok(format!(
775 "{why}Find the cause, fix it in your change, and push. Use get_workflow_run and get_job_logs for any workflow named above."
776 ));
777 }
778 Ok(format!(
779 "{why}These commands failed against your change.\n\n{}",
780 failed.join("\n\n")
781 ))
782 }
783 Feedback::FailedWorkflows => {
784 let (statuses, settings) = futures_util::future::try_join(
785 self.statuses(&pull.repo_id, pull.head_commit.as_deref()),
786 self.settings_for(pull),
787 )
788 .await?;
789 let required = |context: &str| {
790 let name = g1t_contracts::work::check_name(context).0;
791 settings.required_checks.iter().any(|wanted| wanted.eq_ignore_ascii_case(name))
792 };
793 let failing: Vec<&CommitStatus> =
794 statuses.iter().filter(|s| s.state == "failure" || s.state == "error").collect();
795 let failed: Vec<String> = failing
796 .iter()
797 .map(|s| {
798 let run = s.target_url.as_deref().and_then(|url| url.rsplit('/').next()).unwrap_or_default();
799 format!(
800 "- {}{} ({}): run `{run}`, {}",
801 s.context,
802 if required(&s.context) { ", required to merge" } else { "" },
803 s.description.as_deref().unwrap_or("failed"),
804 s.target_url.as_deref().unwrap_or_default()
805 )
806 })
807 .collect();
808 let logs = self.failing_logs(pull, &failing).await.unwrap_or_default();
809 // Named outright: the agent cannot guess it from its fork.
810 let repo = g1t_kit::call::<_, Option<RepoPath>>(
811 &self.repos,
812 "path_by_id",
813 &g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
814 )
815 .await?
816 .map(|path| format!("{}/{}", path.namespace, path.name))
817 .unwrap_or_default();
818 let logs = if logs.is_empty() {
819 String::new()
820 } else {
821 format!("\n\nThe end of what the failing jobs printed:\n\n{}", logs.join("\n\n"))
822 };
823 Ok(format!(
824 "These checks failed on your latest commit to {repo}. They are the repository's workflows, run on your pull request:\n\n{}{logs}\n\n\
825 For more, use the `get_workflow_run` tool (repo `{repo}` and the run's id), \
826 then `get_job_logs` for the job that failed. Fix the cause in the code, not the workflow, \
827 unless the workflow itself is wrong. Push, and the checks run again.",
828 failed.join("\n")
829 ))
830 }
831 Feedback::Review(finished_at) => {
832 // Everything a review says is recorded at the moment it finished.
833 let notes = self
834 .db
835 .prepare(
836 "SELECT body, path, line FROM comments
837 WHERE repo_id = ? AND number = ? AND author_id = ? AND created_at = ?
838 ORDER BY id",
839 )
840 .bind(&[
841 pull.repo_id.as_str().into(),
842 pull.number.into(),
843 AGENT_ID.into(),
844 finished_at.as_str().into(),
845 ])?
846 .all()
847 .await?
848 .results::<ReviewNote>()?;
849 let mut on_lines = Vec::new();
850 let mut summary = String::new();
851 for note in notes {
852 match (note.path, note.line) {
853 (Some(path), Some(line)) => {
854 on_lines.push(format!("- `{path}` line {line}: {}", note.body));
855 }
856 (Some(path), None) => on_lines.push(format!("- `{path}`: {}", note.body)),
857 (None, _) => summary = note.body,
858 }
859 }
860 let mut text = format!(
861 "Another agent reviewed your change and asked for changes.\n\n{summary}"
862 );
863 if !on_lines.is_empty() {
864 text.push_str("\n\nIts comments on lines:\n");
865 text.push_str(&on_lines.join("\n"));
866 }
867 Ok(text)
868 }
869 Feedback::Person(request) => {
870 // What they wrote since the agent last revised, which their
871 // request for changes closes.
872 let revised: Option<String> = self
873 .db
874 .prepare("SELECT revised_at AS value FROM pulls WHERE id = ?")
875 .bind(&[pull.id.as_str().into()])?
876 .first::<Option<String>>(Some("value"))
877 .await?
878 .flatten();
879 let notes = self
880 .db
881 .prepare(
882 "SELECT body, path, line FROM comments
883 WHERE repo_id = ? AND number = ? AND author_id = ?
884 AND created_at > ? AND created_at <= ?
885 ORDER BY id",
886 )
887 .bind(&[
888 pull.repo_id.as_str().into(),
889 pull.number.into(),
890 request.author_id.as_str().into(),
891 revised.unwrap_or_default().into(),
892 request.created_at.as_str().into(),
893 ])?
894 .all()
895 .await?
896 .results::<ReviewNote>()?;
897 let mut on_lines = Vec::new();
898 let mut said = Vec::new();
899 for note in notes {
900 match (note.path, note.line) {
901 (Some(path), Some(line)) => {
902 on_lines.push(format!("- `{path}` line {line}: {}", note.body));
903 }
904 (Some(path), None) => on_lines.push(format!("- `{path}`: {}", note.body)),
905 (None, _) => said.push(note.body),
906 }
907 }
908 let mut text = format!(
909 "{} reviewed your change and asked for changes.\n\n{}",
910 request.author_name,
911 said.join("\n\n")
912 );
913 if !on_lines.is_empty() {
914 text.push_str("\n\nTheir comments on lines:\n");
915 text.push_str(&on_lines.join("\n"));
916 }
917 Ok(text)
918 }
919 }
920 }
921
922 /// The end of what the failed jobs of failing workflow runs printed, a
923 /// few jobs at most, for an agent sent back to fix them. Empty where the
924 /// runs or their logs cannot be read.
925 async fn failing_logs(&self, pull: &Pull, failing: &[&CommitStatus]) -> Result<Vec<String>> {
926 use g1t_contracts::actions::{JobLog, LogsArgs, RunArgs, RunDetail};
927 let Some(repo) = g1t_kit::call::<_, Option<RepoPath>>(
928 &self.repos,
929 "path_by_id",
930 &g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
931 )
932 .await?
933 else {
934 return Ok(Vec::new());
935 };
936 let viewer = self.owner_viewer(pull).await?;
937 let mut out = Vec::new();
938 for status in failing {
939 let Some(run_id) = status
940 .target_url
941 .as_deref()
942 .and_then(|url| url.split("/actions/runs/").nth(1))
943 .map(|rest| rest.split(['/', '?', '#']).next().unwrap_or_default().to_owned())
944 .filter(|id| !id.is_empty())
945 else {
946 continue;
947 };
948 let detail: Outcome<RunDetail> = g1t_kit::call(
949 &self.actions,
950 "run",
951 &RunArgs { repo: repo.clone(), viewer: viewer.clone(), id: run_id, attempt: None },
952 )
953 .await?;
954 let Outcome::Ok(detail) = detail else { continue };
955 let failed_jobs = detail.jobs.into_iter().filter(|job| {
956 job.status == "completed" && matches!(job.conclusion.as_deref(), Some("failure" | "timed_out"))
957 });
958 for job in failed_jobs {
959 if out.len() >= MAX_FAILED_JOBS {
960 return Ok(out);
961 }
962 let mut text = String::new();
963 let mut after = 0;
964 for _ in 0..MAX_LOG_PAGES {
965 let page: Outcome<JobLog> = g1t_kit::call(
966 &self.actions,
967 "logs",
968 &LogsArgs { repo: repo.clone(), viewer: viewer.clone(), job: job.id.clone(), after },
969 )
970 .await?;
971 let Outcome::Ok(page) = page else { break };
972 let Some(last) = page.chunks.last().map(|chunk| chunk.seq) else { break };
973 for chunk in &page.chunks {
974 text.push_str(&chunk.text);
975 if !chunk.text.ends_with('\n') {
976 text.push('\n');
977 }
978 }
979 // Only the end is kept, so the start can go as it is read.
980 let length = text.chars().count();
981 if length > MAX_JOB_LOG_CHARS * 2 {
982 text = text.chars().skip(length - MAX_JOB_LOG_CHARS).collect();
983 }
984 after = last;
985 if page.chunks.len() < 500 {
986 break;
987 }
988 }
989 let length = text.chars().count();
990 let tail: String = text.chars().skip(length.saturating_sub(MAX_JOB_LOG_CHARS)).collect();
991 if !tail.trim().is_empty() {
992 out.push(format!("{} / {}:\n```\n{}\n```", status.context, job.name, tail.trim_end()));
993 }
994 }
995 }
996 Ok(out)
997 }
998
999 pub(crate) async fn advance(&self, a: AdvanceArgs) -> Result<Advance> {
1000 let Some(pull) = self.pull_by_id(&a.pull_id).await? else {
1001 return Ok(Advance::None);
1002 };
1003 if pull.status != PullStatus::Open {
1004 return Ok(Advance::None);
1005 }
1006 // As a member: a private repository would look missing otherwise,
1007 // and the pull request would never move.
1008 let viewer: Viewer = self.owner_viewer(&pull).await?;
1009 let repo: Outcome<Repo> = g1t_kit::call(
1010 &self.repos,
1011 "get_by_id",
1012 &GetByIdArgs {
1013 id: pull.repo_id.clone(),
1014 viewer,
1015 },
1016 )
1017 .await?;
1018 let (Outcome::Ok(repo), Some(source)) = (crate::retired::unless_archived(repo), pull.fork.clone()) else {
1019 return Ok(Advance::None);
1020 };
1021 // The branch it merges into: what it catches up with.
1022 let base = pull.base_branch(&repo.default_branch).to_owned();
1023 let issue = match pull.issue {
1024 Some(number) => self.issue(&pull.repo_id, number).await?,
1025 None => None,
1026 };
1027 let behind = self.is_behind(&repo.id, &pull).await?;
1028 let Some((lifecycle, next)) = self.assess(&pull, &issue, behind).await? else {
1029 return Ok(Advance::None);
1030 };
1031
1032 if matches!(next, Next::Merge) {
1033 self.merge_by_policy(&repo, &pull).await?;
1034 return Ok(Advance::None);
1035 }
1036 let (step, minutes) = match &next {
1037 Next::Wait | Next::Merge => return Ok(Advance::None),
1038 Next::Review => ("review", REVIEW_MINUTES),
1039 Next::Revise(_) => ("revision", REVISION_MINUTES),
1040 Next::CatchUp => ("catch_up", CATCH_UP_MINUTES),
1041 };
1042 let feedback = match &next {
1043 Next::Revise(feedback) => self.feedback(&pull, feedback).await?,
1044 Next::CatchUp => self.conflict_note(&pull, &base).await?,
1045 _ => String::new(),
1046 };
1047 if !self
1048 .claim(&pull.id, step, minutes, matches!(next, Next::Revise(_)))
1049 .await?
1050 {
1051 return Ok(Advance::None);
1052 }
1053 // Said in the conversation, so nobody has to wonder why a review or
1054 // a new commit appeared.
1055 let told = match &next {
1056 Next::Review => {
1057 self.db
1058 .prepare(
1059 "UPDATE pulls SET reviewers = json_insert(reviewers, '$[#]', ?1)
1060 WHERE id = ?2 AND NOT EXISTS (
1061 SELECT 1 FROM json_each(pulls.reviewers) WHERE json_each.value = ?1)",
1062 )
1063 .bind(&[AGENT_NAME.into(), pull.id.as_str().into()])?
1064 .run()
1065 .await?;
1066 "requested a review from g1t".to_owned()
1067 }
1068 Next::Revise(Feedback::FailedChecks) => {
1069 "sent g1t back to fix what failed in the merge queue".to_owned()
1070 }
1071 Next::Revise(Feedback::FailedWorkflows) => {
1072 "sent g1t back to fix the failed checks".to_owned()
1073 }
1074 Next::Revise(Feedback::Review(_)) => {
1075 "sent g1t back to address the review".to_owned()
1076 }
1077 _ => format!("asked g1t to bring this up to date with {base}"),
1078 };
1079 self.note(
1080 &pull.repo_id,
1081 pull.number,
1082 (POLICY_ACTOR_ID, POLICY_ACTOR_NAME),
1083 &told,
1084 )
1085 .await?;
1086 let job = LifecycleJob {
1087 pull_id: pull.id,
1088 repo: RepoPath {
1089 namespace: repo.namespace,
1090 name: repo.name,
1091 },
1092 number: pull.number,
1093 author: pull.requested_by.unwrap_or(pull.author),
1094 source,
1095 branch: None,
1096 default_branch: base,
1097 title: pull.title,
1098 description: pull.body.unwrap_or_default(),
1099 issue,
1100 feedback,
1101 round: lifecycle.revisions + 1,
1102 };
1103 Ok(match next {
1104 Next::Review => Advance::Review { job },
1105 Next::Revise(_) => Advance::Revise { job },
1106 Next::CatchUp => Advance::CatchUp { job },
1107 Next::Wait | Next::Merge => Advance::None,
1108 })
1109 }
1110
1111 /// Lands a pull request that is ready, on the authority of the
1112 /// repository's settings instead of a person's click.
1113 async fn merge_by_policy(&self, repo: &Repo, pull: &Pull) -> Result<()> {
1114 if !self.claim(&pull.id, "merge", MERGE_MINUTES, false).await? {
1115 return Ok(());
1116 }
1117 // g1t acts for the workspace whose members turned this on, with the
1118 // default base permission (Write), which merging needs.
1119 let actor = User {
1120 id: POLICY_ACTOR_ID.to_owned(),
1121 username: POLICY_ACTOR_NAME.to_owned(),
1122 verified: true,
1123 workspaces: vec![Membership::member(repo.namespace.to_lowercase())],
1124 ..User::default()
1125 };
1126 let merged = self
1127 .merge_pull(PullActionArgs {
1128 actor,
1129 repo: RepoPath {
1130 namespace: repo.namespace.clone(),
1131 name: repo.name.clone(),
1132 },
1133 number: pull.number,
1134 summary: String::new(),
1135 keep_issue_open: false,
1136 ignore_checks: false,
1137 bypass_rules: false,
1138 })
1139 .await?;
1140 match merged {
1141 Outcome::Ok(_) => Ok(()),
1142 // Most likely the branch moved in the moment between: let go, and
1143 // the next look at it will catch up and try again.
1144 Outcome::Fail(failure) if failure.code == FailureCode::Conflict => {
1145 self.db
1146 .prepare(
1147 "UPDATE pulls SET working_on = NULL, working_until = NULL
1148 WHERE id = ? AND working_on = 'merge'",
1149 )
1150 .bind(&[pull.id.as_str().into()])?
1151 .run()
1152 .await?;
1153 Ok(())
1154 }
1155 Outcome::Fail(failure) => {
1156 self.stall(StallArgs {
1157 pull_id: pull.id.clone(),
1158 reason: format!("g1t could not merge this: {}", failure.message),
1159 by: None,
1160 })
1161 .await?;
1162 Ok(())
1163 }
1164 }
1165 }
1166
1167 /// Records that a merge was asked for while the pull request was
1168 /// behind, and announces it so that the runner brings it up to date.
1169 pub(crate) async fn request_landing(
1170 &self,
1171 pull: &Pull,
1172 actor: &User,
1173 keep_issue_open: bool,
1174 ) -> Result<()> {
1175 let now = now_ms();
1176 let request = serde_json::to_string(&LandRequest {
1177 actor: actor.clone(),
1178 keep_issue_open,
1179 })?;
1180 self.db
1181 .prepare(
1182 "UPDATE pulls
1183 SET land_requested = ?, land_requested_at = ?, stalled = NULL,
1184 working_on = 'catch_up', working_until = ?
1185 WHERE id = ?",
1186 )
1187 .bind(&[
1188 request.into(),
1189 rfc3339(now).into(),
1190 rfc3339(now + CATCH_UP_MINUTES * 60 * 1000).into(),
1191 pull.id.as_str().into(),
1192 ])?
1193 .run()
1194 .await?;
1195 self.publish(
1196 "pull.merge_requested",
1197 &pull.repo_id,
1198 actor,
1199 Self::pull_event(pull),
1200 )
1201 .await
1202 }
1203
1204 /// The merge waiting on a pull request, if one was asked for recently
1205 /// enough to still stand.
1206 async fn land_request(&self, pull_id: &str) -> Result<Option<LandRequest>> {
1207 let row = self.land_row(pull_id).await?;
1208 let oldest = rfc3339(now_ms().saturating_sub(CATCH_UP_MINUTES * 60 * 1000));
1209 Ok(row
1210 .filter(|row| {
1211 row.land_requested_at
1212 .as_deref()
1213 .is_some_and(|at| at >= oldest.as_str())
1214 })
1215 .and_then(|row| row.land_requested)
1216 .and_then(|request| serde_json::from_str(&request).ok()))
1217 }
1218
1219 /// Whether a merge is waiting on a pull request, and why g1t stopped
1220 /// working on it if it did.
1221 pub(crate) async fn landing_state(&self, pull_id: &str) -> Result<(bool, Option<String>)> {
1222 let stalled = self.land_row(pull_id).await?.and_then(|row| row.stalled);
1223 Ok((self.land_request(pull_id).await?.is_some(), stalled))
1224 }
1225
1226 /// The merge waiting on a pull request and why g1t stopped, as read
1227 /// for this request (prefetch.rs) or now.
1228 async fn land_row(&self, pull_id: &str) -> Result<Option<LandRow>> {
1229 if let Some(found) = self.prefetched_pull(pull_id) {
1230 return found.first::<LandRow>(Slot::Pull);
1231 }
1232 self.db
1233 .prepare("SELECT land_requested, land_requested_at, stalled FROM pulls WHERE id = ?")
1234 .bind(&[pull_id.into()])?
1235 .first::<LandRow>(None)
1236 .await
1237 }
1238
1239 async fn forget_landing(&self, pull_id: &str) -> Result<()> {
1240 self.db
1241 .prepare(
1242 "UPDATE pulls SET land_requested = NULL, land_requested_at = NULL WHERE id = ?",
1243 )
1244 .bind(&[pull_id.into()])?
1245 .run()
1246 .await?;
1247 Ok(())
1248 }
1249
1250 /// Lands a pull request whose head has just moved, if a merge of it was
1251 /// waiting for exactly that. What it was caught up to was already
1252 /// checked and reviewed apart from the merge, so the checks are not
1253 /// waited for again; a repository that wants them rerun turns on
1254 /// "require up to date", and then nothing is landed this way.
1255 pub(crate) async fn land_if_requested(&self, pull_id: &str) -> Result<()> {
1256 let Some(request) = self.land_request(pull_id).await? else {
1257 return Ok(());
1258 };
1259 self.forget_landing(pull_id).await?;
1260 let Some(pull) = self.pull_by_id(pull_id).await? else {
1261 return Ok(());
1262 };
1263 let repo: Outcome<Repo> = g1t_kit::call(
1264 &self.repos,
1265 "get_by_id",
1266 &GetByIdArgs {
1267 id: pull.repo_id.clone(),
1268 viewer: Some(request.actor.clone()),
1269 },
1270 )
1271 .await?;
1272 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
1273 return Ok(());
1274 };
1275 let merged = self
1276 .merge_pull(PullActionArgs {
1277 actor: request.actor,
1278 repo: RepoPath {
1279 namespace: repo.namespace,
1280 name: repo.name,
1281 },
1282 number: pull.number,
1283 summary: String::new(),
1284 keep_issue_open: request.keep_issue_open,
1285 ignore_checks: true,
1286 bypass_rules: false,
1287 })
1288 .await?;
1289 if let Outcome::Fail(failure) = merged {
1290 self.stall(StallArgs {
1291 pull_id: pull.id,
1292 reason: format!(
1293 "It was brought up to date but could not be merged: {}",
1294 failure.message
1295 ),
1296 by: None,
1297 })
1298 .await?;
1299 }
1300 Ok(())
1301 }
1302
1303 /// What the runner needs to bring a pull request up to date for a merge
1304 /// that is waiting on it.
1305 pub(crate) async fn catch_up_job(&self, a: CatchUpJobArgs) -> Result<Option<LifecycleJob>> {
1306 if self.land_request(&a.pull_id).await?.is_none() {
1307 return Ok(None);
1308 }
1309 let Some(pull) = self.pull_by_id(&a.pull_id).await? else {
1310 return Ok(None);
1311 };
1312 // Its owner (whoever asked g1t for it, or its author) can read both
1313 // the repository and the pull request's source.
1314 let repo: Outcome<Repo> = g1t_kit::call(
1315 &self.repos,
1316 "get_by_id",
1317 &GetByIdArgs {
1318 id: pull.repo_id.clone(),
1319 viewer: self.owner_viewer(&pull).await?,
1320 },
1321 )
1322 .await?;
1323 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
1324 return Ok(None);
1325 };
1326 let path = RepoPath {
1327 namespace: repo.namespace,
1328 name: repo.name,
1329 };
1330 let issue = match pull.issue {
1331 Some(number) => self.issue(&pull.repo_id, number).await?,
1332 None => None,
1333 };
1334 let base = pull.base_branch(&repo.default_branch).to_owned();
1335 let feedback = self.conflict_note(&pull, &base).await?;
1336 Ok(Some(LifecycleJob {
1337 pull_id: pull.id,
1338 source: pull.fork.unwrap_or_else(|| path.clone()),
1339 repo: path,
1340 number: pull.number,
1341 author: pull.requested_by.unwrap_or(pull.author),
1342 branch: pull.branch,
1343 default_branch: base,
1344 title: pull.title,
1345 description: pull.body.unwrap_or_default(),
1346 issue,
1347 feedback,
1348 round: 0,
1349 }))
1350 }
1351
1352 /// For an agent catching up: the files g1t already knows conflict, so
1353 /// it reads them first. Empty when none are known.
1354 pub(crate) async fn conflict_note(&self, pull: &Pull, default_branch: &str) -> Result<String> {
1355 Ok(match self.conflicting_files(pull).await? {
1356 Some(files) if !files.is_empty() => format!(
1357 "g1t found ahead of time that merging {default_branch} into this pull request conflicts in these files: {}.",
1358 files.join(", ")
1359 ),
1360 _ => String::new(),
1361 })
1362 }
1363
1364 /// Stops seeing a pull request through until a person steps in. The
1365 /// first stop is published (`pull.stalled`), which tells its people
1366 /// that it needs them; a stop on one already stopped only says why.
1367 pub(crate) async fn stall(&self, a: StallArgs) -> Result<bool> {
1368 let before = self.pull_by_id(&a.pull_id).await?;
1369 let was_stalled = self.is_stalled(&a.pull_id).await?;
1370 self.db
1371 .prepare(
1372 "UPDATE pulls
1373 SET stalled = ?, working_on = NULL, working_until = NULL,
1374 land_requested = NULL, land_requested_at = NULL
1375 WHERE id = ? AND status = 'open'",
1376 )
1377 .bind(&[a.reason.trim().into(), a.pull_id.as_str().into()])?
1378 .run()
1379 .await?;
1380 self.db
1381 .prepare("UPDATE pulls SET stage = 'needs_you', stage_detail = ? WHERE id = ?")
1382 .bind(&[a.reason.trim().into(), a.pull_id.as_str().into()])?
1383 .run()
1384 .await?;
1385 if let Some(pull) = before.filter(|pull| !was_stalled && pull.status == PullStatus::Open) {
1386 self.publish_as(
1387 "pull.stalled",
1388 &pull.repo_id,
1389 a.by.clone(),
1390 g1t_contracts::events::PullEvent {
1391 detail: Some(a.reason.trim().to_owned()),
1392 ..Self::pull_event(&pull)
1393 },
1394 )
1395 .await?;
1396 }
1397 Ok(true)
1398 }
1399
1400 /// Whether g1t has stopped seeing the pull request through.
1401 pub(crate) async fn is_stalled(&self, pull_id: &str) -> Result<bool> {
1402 Ok(self
1403 .db
1404 .prepare("SELECT 1 AS value FROM pulls WHERE id = ? AND stalled IS NOT NULL")
1405 .bind(&[pull_id.into()])?
1406 .first::<u32>(Some("value"))
1407 .await?
1408 .is_some())
1409 }
1410
1411 /// Says that a pull request g1t had stopped on is going again, which
1412 /// closes what it was waiting on a person for.
1413 pub(crate) async fn announce_resumed(&self, pull_id: &str, actor: Option<String>) -> Result<()> {
1414 if let Some(pull) = self.pull_by_id(pull_id).await? {
1415 self.publish_as("pull.resumed", &pull.repo_id, actor, Self::pull_event(&pull)).await?;
1416 }
1417 Ok(())
1418 }
1419
1420 pub(crate) async fn managed_pulls(&self, a: ManagedPullsArgs) -> Result<Vec<String>> {
1421 let rows = self
1422 .db
1423 .prepare(
1424 "SELECT id AS value FROM pulls
1425 WHERE status = 'open' AND managed = 1 AND stalled IS NULL
1426 AND (?1 IS NULL OR repo_id = ?1)
1427 ORDER BY updated_at DESC LIMIT ?2",
1428 )
1429 .bind(&[
1430 a.repo_id.map_or(JsValue::NULL, JsValue::from),
1431 MANAGED_PAGE.into(),
1432 ])?
1433 .all()
1434 .await?
1435 .results::<ValueRow>()?;
1436 Ok(rows.into_iter().map(|row| row.value).collect())
1437 }
1438}
1439
1440#[cfg(test)]
1441mod tests {
1442 use super::*;
1443
1444 const MAX_REVISIONS: u32 = 2;
1445
1446 #[test]
1447 fn a_verdict_from_whoever_asked_for_g1t_s_change_is_not_someone_else_s() {
1448 use crate::rows::stored::{ASKER, G1T, pull};
1449 let made = pull(G1T, Some(ASKER));
1450 assert!(made_by_g1t(&made));
1451 // The person it was made for is held to what an author was: their
1452 // request for changes does not send g1t back as a reviewer's
1453 // would, and their approval does not lift a hold.
1454 assert!(!from_someone_else(&made, ASKER.0));
1455 assert!(!from_someone_else(&made, AGENT_ID));
1456 assert!(from_someone_else(&made, "usr_reviewer"));
1457 // Anyone's own pull request, the same.
1458 let own = pull(ASKER, None);
1459 assert!(!from_someone_else(&own, ASKER.0));
1460 assert!(from_someone_else(&own, "usr_reviewer"));
1461 }
1462
1463 #[test]
1464 fn approvals_the_repository_wants_are_waited_for() {
1465 let short = || Facts {
1466 review: reviewed(Some(Verdict::Approve)),
1467 approvals_missing: Some("This repository requires 1 approving review.".to_owned()),
1468 ..facts()
1469 };
1470 let (lifecycle, next) = decide(short());
1471 assert_eq!(lifecycle.stage, Stage::NeedsYou);
1472 assert_eq!(
1473 lifecycle.detail,
1474 "This repository requires 1 approving review."
1475 );
1476 assert!(matches!(next, Next::Wait));
1477 // Not even a repository that merges by itself merges without them.
1478 let automatic = Facts {
1479 auto_merge: true,
1480 ..short()
1481 };
1482 assert_eq!(outcome(automatic), (Stage::NeedsYou, "wait"));
1483 }
1484
1485 #[test]
1486 fn a_repository_can_leave_review_to_people() {
1487 let unreviewed = Facts {
1488 agent_review: false,
1489 ..facts()
1490 };
1491 assert_eq!(outcome(unreviewed), (Stage::Ready, "wait"));
1492 let failing = Facts {
1493 agent_review: false,
1494 check_status: Some(CheckStatus::Failed),
1495 ..facts()
1496 };
1497 assert_eq!(outcome(failing), (Stage::Revising, "revise for checks"));
1498 }
1499
1500 #[test]
1501 fn a_repository_sets_how_often_the_author_is_sent_back() {
1502 let never = Facts {
1503 max_revisions: 0,
1504 check_status: Some(CheckStatus::Failed),
1505 ..facts()
1506 };
1507 assert_eq!(outcome(never), (Stage::NeedsYou, "wait"));
1508 }
1509
1510 /// Checks on a head commit: `(context, state)` statuses, against the
1511 /// required check names.
1512 fn checks(statuses: &[(&str, &str)], required: &[&str]) -> WorkflowFacts {
1513 let statuses: Vec<CommitStatus> = statuses
1514 .iter()
1515 .map(|(context, state)| CommitStatus {
1516 context: (*context).to_owned(),
1517 state: (*state).to_owned(),
1518 description: None,
1519 target_url: None,
1520 updated_at: String::new(),
1521 source: None,
1522 check_run_id: None,
1523 })
1524 .collect();
1525 let required: Vec<String> = required.iter().map(|name| (*name).to_owned()).collect();
1526 WorkflowFacts::of(&statuses, &required)
1527 }
1528
1529 /// A pull request that is ready for review, whose required check
1530 /// passed, and nothing else yet.
1531 fn facts() -> Facts {
1532 Facts {
1533 draft: false,
1534 stalled: None,
1535 working_on: None,
1536 check_status: None,
1537 review_pending: false,
1538 revisions: 0,
1539 review: None,
1540 behind: false,
1541 conflicting: false,
1542 auto_merge: false,
1543 require_up_to_date: false,
1544 agent_review: true,
1545 max_revisions: MAX_REVISIONS,
1546 approvals_missing: None,
1547 person_request: None,
1548 queued: None,
1549 workflows: checks(&[("CI / pull_request", "success")], &["CI"]),
1550 low_confidence: None,
1551 }
1552 }
1553
1554 #[test]
1555 fn failed_checks_send_the_agent_back_and_running_ones_wait() {
1556 let failed = Facts {
1557 workflows: checks(&[("CI / pull_request", "failure")], &["CI"]),
1558 ..facts()
1559 };
1560 let (lifecycle, next) = decide(failed);
1561 assert!(matches!(next, Next::Revise(Feedback::FailedWorkflows)));
1562 assert!(lifecycle.detail.contains("CI / pull_request failed"));
1563 let running = Facts {
1564 workflows: checks(&[("CI / pull_request", "pending")], &["CI"]),
1565 ..facts()
1566 };
1567 let (lifecycle, next) = decide(running);
1568 assert!(matches!(next, Next::Wait));
1569 assert!(lifecycle.detail.contains("Waiting for CI / pull_request"));
1570 }
1571
1572 #[test]
1573 fn a_check_the_branch_does_not_require_is_fixed_but_does_not_hold_it_for_ever() {
1574 // Lint is not required: the agent is still sent back to fix it...
1575 let lint = checks(&[("CI / pull_request", "success"), ("Lint / pull_request", "failure")], &["CI"]);
1576 let failing = Facts { workflows: lint.clone(), ..facts() };
1577 assert_eq!(outcome(failing), (Stage::Revising, "revise for workflows"));
1578 // ...but once it is out of revisions, Lint no longer holds it.
1579 let exhausted = Facts {
1580 workflows: lint,
1581 revisions: MAX_REVISIONS,
1582 review: reviewed(Some(Verdict::Approve)),
1583 ..facts()
1584 };
1585 assert_eq!(outcome(exhausted), (Stage::Ready, "wait"));
1586 // A required check that still fails asks a person.
1587 let required = Facts {
1588 workflows: checks(&[("CI / pull_request", "failure")], &["CI"]),
1589 revisions: MAX_REVISIONS,
1590 ..facts()
1591 };
1592 let (lifecycle, next) = decide(required);
1593 assert_eq!(lifecycle.stage, Stage::NeedsYou);
1594 assert_eq!(lifecycle.detail, "The required check CI still fails after the agent revised twice.");
1595 assert!(matches!(next, Next::Wait));
1596 }
1597
1598 #[test]
1599 fn a_required_check_that_has_not_reported_is_waited_for() {
1600 let missing = Facts {
1601 workflows: checks(&[("CI / pull_request", "success")], &["CI", "Deploy"]),
1602 review: reviewed(Some(Verdict::Approve)),
1603 auto_merge: true,
1604 ..facts()
1605 };
1606 let (lifecycle, next) = decide(missing);
1607 assert_eq!(lifecycle.stage, Stage::Checking);
1608 assert_eq!(lifecycle.detail, "Waiting for the required check Deploy to report on its latest commit.");
1609 assert!(matches!(next, Next::Wait), "auto-merge must not land it");
1610 }
1611
1612 #[test]
1613 fn a_queued_pull_request_waits_in_the_queue() {
1614 let queued = Facts {
1615 review: reviewed(Some(Verdict::Approve)),
1616 queued: Some((QueueState::Testing, vec![12, 14])),
1617 auto_merge: true,
1618 ..facts()
1619 };
1620 let (lifecycle, next) = decide(queued);
1621 assert_eq!(lifecycle.stage, Stage::Queued);
1622 assert!(lifecycle.detail.contains("#12, #14"));
1623 assert!(matches!(next, Next::Wait));
1624 }
1625
1626 fn asked_by_a_person() -> Option<PersonRequest> {
1627 Some(PersonRequest {
1628 author_id: "usr_reviewer".to_owned(),
1629 author_name: "g1t-reviewer".to_owned(),
1630 created_at: "2026-10-02T11:00:00.000Z".to_owned(),
1631 })
1632 }
1633
1634 #[test]
1635 fn a_person_asking_for_changes_sends_the_agent_back() {
1636 let asked = Facts {
1637 review: reviewed(Some(Verdict::Approve)),
1638 approvals_missing: Some("A reviewer has asked for changes.".to_owned()),
1639 person_request: asked_by_a_person(),
1640 ..facts()
1641 };
1642 let (lifecycle, _) = decide(Facts {
1643 person_request: asked_by_a_person(),
1644 ..facts()
1645 });
1646 assert!(lifecycle.detail.starts_with("g1t-reviewer asked for changes"));
1647 assert_eq!(outcome(asked), (Stage::Revising, "revise for a person"));
1648 }
1649
1650 #[test]
1651 fn a_person_is_asked_once_the_revisions_run_out() {
1652 let exhausted = Facts {
1653 person_request: asked_by_a_person(),
1654 revisions: MAX_REVISIONS,
1655 ..facts()
1656 };
1657 assert_eq!(outcome(exhausted), (Stage::NeedsYou, "wait"));
1658 }
1659
1660 fn reviewed(verdict: Option<Verdict>) -> Option<FinishedReview> {
1661 Some(FinishedReview {
1662 finished_at: "2026-10-02T10:00:00.000Z".to_owned(),
1663 verdict,
1664 })
1665 }
1666
1667 /// The stage, and a word for the step to take.
1668 fn outcome(facts: Facts) -> (Stage, &'static str) {
1669 let (lifecycle, next) = decide(facts);
1670 let step = match next {
1671 Next::Wait => "wait",
1672 Next::Review => "review",
1673 Next::Revise(Feedback::FailedChecks) => "revise for checks",
1674 Next::Revise(Feedback::FailedWorkflows) => "revise for workflows",
1675 Next::Revise(Feedback::Review(_)) => "revise for review",
1676 Next::Revise(Feedback::Person(_)) => "revise for a person",
1677 Next::CatchUp => "catch up",
1678 Next::Merge => "merge",
1679 };
1680 (lifecycle.stage, step)
1681 }
1682
1683 #[test]
1684 fn nothing_is_started_while_the_agent_is_still_working() {
1685 let draft = Facts {
1686 draft: true,
1687 check_status: None,
1688 ..facts()
1689 };
1690 assert_eq!(outcome(draft), (Stage::Working, "wait"));
1691 }
1692
1693 #[test]
1694 fn checks_come_before_review() {
1695 let unchecked = Facts {
1696 workflows: checks(&[], &["CI"]),
1697 ..facts()
1698 };
1699 assert_eq!(outcome(unchecked), (Stage::Checking, "wait"));
1700 let running = Facts {
1701 workflows: checks(&[("CI / pull_request", "pending")], &["CI"]),
1702 ..facts()
1703 };
1704 assert_eq!(outcome(running), (Stage::Checking, "wait"));
1705 assert_eq!(outcome(facts()), (Stage::Reviewing, "review"));
1706 }
1707
1708 #[test]
1709 fn a_branch_that_requires_no_checks_goes_straight_to_review() {
1710 let unchecked = Facts {
1711 workflows: WorkflowFacts::default(),
1712 ..facts()
1713 };
1714 assert_eq!(outcome(unchecked), (Stage::Reviewing, "review"));
1715 }
1716
1717 #[test]
1718 fn failing_in_the_merge_queue_sends_the_author_back() {
1719 let failed = Facts {
1720 check_status: Some(CheckStatus::Failed),
1721 ..facts()
1722 };
1723 let (lifecycle, next) = decide(failed);
1724 assert_eq!(lifecycle.detail, "It failed in the merge queue. The agent is being sent back to fix it.");
1725 assert!(matches!(next, Next::Revise(Feedback::FailedChecks)));
1726 }
1727
1728 #[test]
1729 fn checks_that_could_not_run_are_a_persons_problem() {
1730 let errored = Facts {
1731 check_status: Some(CheckStatus::Errored),
1732 ..facts()
1733 };
1734 assert_eq!(outcome(errored), (Stage::NeedsYou, "wait"));
1735 }
1736
1737 #[test]
1738 fn a_review_asking_for_changes_sends_the_author_back() {
1739 let changes = Facts {
1740 review: reviewed(Some(Verdict::RequestChanges)),
1741 ..facts()
1742 };
1743 assert_eq!(outcome(changes), (Stage::Revising, "revise for review"));
1744 }
1745
1746 #[test]
1747 fn the_author_is_sent_back_only_so_many_times() {
1748 let failing = Facts {
1749 check_status: Some(CheckStatus::Failed),
1750 revisions: MAX_REVISIONS,
1751 ..facts()
1752 };
1753 assert_eq!(outcome(failing), (Stage::NeedsYou, "wait"));
1754 let unconvinced = Facts {
1755 review: reviewed(Some(Verdict::RequestChanges)),
1756 revisions: MAX_REVISIONS,
1757 ..facts()
1758 };
1759 assert_eq!(outcome(unconvinced), (Stage::NeedsYou, "wait"));
1760 // One short of the limit still gets another go.
1761 let once = Facts {
1762 check_status: Some(CheckStatus::Failed),
1763 revisions: MAX_REVISIONS - 1,
1764 ..facts()
1765 };
1766 assert_eq!(outcome(once), (Stage::Revising, "revise for checks"));
1767 }
1768
1769 #[test]
1770 fn a_review_that_could_not_be_written_is_not_retried() {
1771 let broken = Facts {
1772 review: reviewed(None),
1773 ..facts()
1774 };
1775 assert_eq!(outcome(broken), (Stage::NeedsYou, "wait"));
1776 }
1777
1778 #[test]
1779 fn being_behind_only_holds_a_change_up_where_the_repository_says_so() {
1780 let behind = || Facts {
1781 review: reviewed(Some(Verdict::Approve)),
1782 behind: true,
1783 ..facts()
1784 };
1785 // By default it is ready as it is; merging brings it up to date.
1786 assert_eq!(outcome(behind()), (Stage::Ready, "wait"));
1787 let strict = Facts {
1788 require_up_to_date: true,
1789 ..behind()
1790 };
1791 assert_eq!(outcome(strict), (Stage::CatchingUp, "catch up"));
1792 let current = Facts {
1793 review: reviewed(Some(Verdict::Approve)),
1794 ..facts()
1795 };
1796 assert_eq!(outcome(current), (Stage::Ready, "wait"));
1797 }
1798
1799 #[test]
1800 fn a_ready_change_lands_by_itself_only_where_the_repository_says_so() {
1801 let ready = || Facts {
1802 review: reviewed(Some(Verdict::Approve)),
1803 ..facts()
1804 };
1805 assert_eq!(outcome(ready()), (Stage::Ready, "wait"));
1806 let automatic = Facts {
1807 auto_merge: true,
1808 ..ready()
1809 };
1810 assert_eq!(outcome(automatic), (Stage::Ready, "merge"));
1811 // One that is behind is merged too: merging brings it up to date.
1812 let behind = Facts {
1813 auto_merge: true,
1814 behind: true,
1815 ..ready()
1816 };
1817 assert_eq!(outcome(behind), (Stage::Ready, "merge"));
1818 // Unless the repository wants it caught up and checked again first.
1819 let strict = Facts {
1820 auto_merge: true,
1821 behind: true,
1822 require_up_to_date: true,
1823 ..ready()
1824 };
1825 assert_eq!(outcome(strict), (Stage::CatchingUp, "catch up"));
1826 // Nothing short of approved is merged, whatever the setting.
1827 let failing = Facts {
1828 auto_merge: true,
1829 check_status: Some(CheckStatus::Failed),
1830 ..ready()
1831 };
1832 assert_eq!(outcome(failing), (Stage::Revising, "revise for checks"));
1833 let unreviewed = Facts {
1834 auto_merge: true,
1835 ..facts()
1836 };
1837 assert_eq!(outcome(unreviewed), (Stage::Reviewing, "review"));
1838 }
1839
1840 #[test]
1841 fn a_conflict_found_ahead_of_time_is_resolved_before_merging() {
1842 let conflicting = || Facts {
1843 review: reviewed(Some(Verdict::Approve)),
1844 behind: true,
1845 conflicting: true,
1846 ..facts()
1847 };
1848 // Even where the repository would merge one that is merely behind.
1849 assert_eq!(outcome(conflicting()), (Stage::CatchingUp, "catch up"));
1850 let automatic = Facts {
1851 auto_merge: true,
1852 ..conflicting()
1853 };
1854 assert_eq!(outcome(automatic), (Stage::CatchingUp, "catch up"));
1855 // Failed checks come first: a revision merges the branch in too.
1856 let failing = Facts {
1857 check_status: Some(CheckStatus::Failed),
1858 ..conflicting()
1859 };
1860 assert_eq!(outcome(failing), (Stage::Revising, "revise for checks"));
1861 }
1862
1863 #[test]
1864 fn catching_up_reruns_the_checks_but_not_the_review() {
1865 // The merge moved the head, so the checks are waited for again.
1866 let merged_in = Facts {
1867 review: reviewed(Some(Verdict::Approve)),
1868 workflows: checks(&[], &["CI"]),
1869 ..facts()
1870 };
1871 assert_eq!(outcome(merged_in), (Stage::Checking, "wait"));
1872 }
1873
1874 #[test]
1875 fn a_step_under_way_is_not_started_again() {
1876 for (step, stage) in [
1877 ("review", Stage::Reviewing),
1878 ("revision", Stage::Revising),
1879 ("catch_up", Stage::CatchingUp),
1880 ] {
1881 let busy = Facts {
1882 working_on: Some(step.to_owned()),
1883 // Whatever else is true, the step in hand comes first.
1884 check_status: Some(CheckStatus::Failed),
1885 ..facts()
1886 };
1887 assert_eq!(outcome(busy), (stage, "wait"));
1888 }
1889 let asked = Facts {
1890 review_pending: true,
1891 ..facts()
1892 };
1893 assert_eq!(outcome(asked), (Stage::Reviewing, "wait"));
1894 }
1895
1896 #[test]
1897 fn a_low_confidence_change_waits_for_a_person_instead_of_merging() {
1898 let held = || Facts {
1899 review: reviewed(Some(Verdict::Approve)),
1900 auto_merge: true,
1901 low_confidence: Some("tests not added, 3 revisions".to_owned()),
1902 ..facts()
1903 };
1904 let (lifecycle, next) = decide(held());
1905 assert_eq!(lifecycle.stage, Stage::NeedsYou);
1906 assert!(matches!(next, Next::Wait), "auto-merge must not land it");
1907 assert_eq!(
1908 lifecycle.detail,
1909 "The agent's confidence in this change is low (tests not added, 3 revisions). This repository asks a person before merging it: approve it to let it land, or ask for changes."
1910 );
1911 // Without auto-merge it needs someone too, and says why.
1912 assert_eq!(outcome(Facts { auto_merge: false, ..held() }), (Stage::NeedsYou, "wait"));
1913 // Not held (the setting is off, or a person approved): it lands.
1914 assert_eq!(outcome(Facts { low_confidence: None, ..held() }), (Stage::Ready, "merge"));
1915 // It holds only a change that is otherwise ready: what comes first,
1916 // such as failed checks, is still dealt with first.
1917 let failing = Facts {
1918 check_status: Some(CheckStatus::Failed),
1919 ..held()
1920 };
1921 assert_eq!(outcome(failing), (Stage::Revising, "revise for checks"));
1922 let unapproved = Facts {
1923 approvals_missing: Some("This repository requires 1 approving review.".to_owned()),
1924 ..held()
1925 };
1926 assert_eq!(decide(unapproved).0.detail, "This repository requires 1 approving review.");
1927 }
1928
1929 #[test]
1930 fn once_stopped_it_stays_stopped() {
1931 let (lifecycle, next) = decide(Facts {
1932 stalled: Some("The agent could not catch up.".to_owned()),
1933 review: reviewed(Some(Verdict::Approve)),
1934 behind: true,
1935 ..facts()
1936 });
1937 assert_eq!(lifecycle.stage, Stage::NeedsYou);
1938 assert_eq!(lifecycle.detail, "The agent could not catch up.");
1939 assert!(matches!(next, Next::Wait));
1940 }
1941}