Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
923 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Checkout tries a failed fetch twice more, after 2 s and 5 s1//! The merge queue: pull requests are tested together with the ones ahead
2//! of them, and only a combination that passed reaches the default branch.
3//!
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge4//! A batch of up to the merge queue rule's `max_entries_to_build` entries (4 unless
5//! the default branch's rules say otherwise) is tested at once, speculatively: one
Checkout tries a failed fetch twice more, after 2 s and 5 s6//! sandbox per entry builds the default branch with that entry and every
7//! entry ahead of it merged in, and pushes the result to
8//! `g1t-queue/<entry>`. The repository's `merge_group` workflows then run
9//! on that commit, and the default branch's required checks must pass
10//! there. Entries land in order, each by moving the default branch to its
11//! tested state, once everything ahead has landed. One that fails leaves
The work, events, webhooks, actions, repos and packages services no longer run g1t's own agents. Work drops the 35 calls that queued issues for g1t, delegated them, messaged and reviewed through the agent, planned, captured memories automatically, rated confidence and capped agent spend, and its pull request lifecycle no longer revises or reviews: a pull request only advances to a merge by the repository's auto-merge setting. Merging a pull request that is behind now catches it up itself, through the repository's branch update, and lands it when the push arrives; when it can't be caught up cleanly the merge says to merge the base in and try again. Recording a session on a pull request, memory recall and remember, overlaps, the merge queue and required checks stay as they were.12//! the queue with a failed check run; the entries behind it are tested
Merge branch 'worktree-agent-ab8033ab973f8d99a'13//! again without it. Which entry answers for a failure, which entries
14//! ahead it breaks with, who is told and how it rejoins: integration.rs.
Checkout tries a failed fetch twice more, after 2 s and 5 s15
16use futures_util::future::try_join_all;
17use g1t_contracts::events::{ChecksEvent, QueueChanged};
18use g1t_contracts::repos::{DeleteBranchArgs, GetByIdArgs, HeadArgs, LandArgs, Landed, Repo, RepoPath};
19use g1t_contracts::time::rfc3339;
20use g1t_contracts::work::*;
21use g1t_contracts::{FailureCode, Outcome, User, new_id};
22use g1t_kit::now_ms;
23use serde::Deserialize;
24use worker::Result;
25use worker::wasm_bindgen::JsValue;
26
27use crate::Work;
28use crate::checks::{hash, new_token};
29
30/// How many entries that left the queue are shown.
31const RECENT: u32 = 20;
32/// Where tested states are pushed, in the repository itself.
Merge branch 'worktree-agent-ab8033ab973f8d99a'33pub(crate) const BRANCH_PREFIX: &str = "g1t-queue/";
Checkout tries a failed fetch twice more, after 2 s and 5 s34
35#[derive(Clone, Deserialize)]
36pub(crate) struct EntryRow {
Merge branch 'worktree-agent-ab8033ab973f8d99a'37 pub(crate) id: String,
38 pub(crate) repo_id: String,
39 pub(crate) pull_id: String,
40 pub(crate) number: u32,
Checkout tries a failed fetch twice more, after 2 s and 5 s41 state: String,
42 enqueued_by: String,
Merge branch 'worktree-agent-ab8033ab973f8d99a'43 pub(crate) keep_issue_open: u32,
44 pub(crate) head_commit: Option<String>,
45 pub(crate) base_commit: Option<String>,
Checkout tries a failed fetch twice more, after 2 s and 5 s46 ahead: Option<String>,
47 combined_commit: Option<String>,
48 results: Option<String>,
49 error: Option<String>,
Merge branch 'worktree-agent-a0796e9a7a41f51e5'50 pub(crate) tested_at: Option<String>,
Checkout tries a failed fetch twice more, after 2 s and 5 s51 created_at: String,
52 finished_at: Option<String>,
Merge branch 'worktree-agent-ab8033ab973f8d99a'53 /// What failed its combined state while entries ahead were still
54 /// being tested (integration.rs), as JSON.
55 #[serde(default)]
56 pub(crate) held: Option<String>,
57 /// What the queue worked out about its failure (integration.rs).
58 #[serde(default)]
59 pub(crate) diagnosis: Option<String>,
Checkout tries a failed fetch twice more, after 2 s and 5 s60}
61
62impl EntryRow {
Merge branch 'worktree-agent-ab8033ab973f8d99a'63 pub(crate) fn state(&self) -> QueueState {
Checkout tries a failed fetch twice more, after 2 s and 5 s64 match self.state.as_str() {
65 "testing" => QueueState::Testing,
66 "passed" => QueueState::Passed,
67 "failed" => QueueState::Failed,
68 "landed" => QueueState::Landed,
69 "removed" => QueueState::Removed,
70 _ => QueueState::Waiting,
71 }
72 }
73
Merge branch 'worktree-agent-ab8033ab973f8d99a'74 pub(crate) fn actor(&self) -> Option<User> {
Checkout tries a failed fetch twice more, after 2 s and 5 s75 serde_json::from_str(&self.enqueued_by).ok()
76 }
77
Merge branch 'worktree-agent-ab8033ab973f8d99a'78 pub(crate) fn ahead(&self) -> Vec<u32> {
Checkout tries a failed fetch twice more, after 2 s and 5 s79 self.ahead
80 .as_deref()
81 .and_then(|ahead| serde_json::from_str(ahead).ok())
82 .unwrap_or_default()
83 }
84
85 fn branch(&self) -> String {
86 format!("{BRANCH_PREFIX}{}", self.id)
87 }
88}
89
90fn minutes_ago(minutes: u64) -> String {
91 rfc3339(now_ms().saturating_sub(minutes * 60 * 1000))
92}
93
94impl Work {
Merge branch 'worktree-agent-ab8033ab973f8d99a'95 pub(crate) async fn entries(&self, repo_id: &str, active: bool) -> Result<Vec<EntryRow>> {
Checkout tries a failed fetch twice more, after 2 s and 5 s96 let sql = if active {
97 "SELECT * FROM queue_entries
98 WHERE repo_id = ? AND state IN ('waiting', 'testing', 'passed')
99 ORDER BY created_at, id"
100 } else {
101 "SELECT * FROM queue_entries
102 WHERE repo_id = ? AND state IN ('failed', 'landed', 'removed')
103 ORDER BY finished_at DESC LIMIT ?"
104 };
105 let statement = self.db.prepare(sql);
106 let statement = if active {
107 statement.bind(&[repo_id.into()])?
108 } else {
109 statement.bind(&[repo_id.into(), RECENT.into()])?
110 };
111 statement.all().await?.results::<EntryRow>()
112 }
113
114 /// The entry a pull request has in the queue now, if any.
115 pub(crate) async fn queued_entry(&self, pull_id: &str) -> Result<Option<(QueueState, Vec<u32>)>> {
116 let row = match self.prefetched_pull(pull_id) {
117 Some(found) => found.first::<EntryRow>(crate::prefetch::Slot::Queued)?,
118 None => self
119 .db
120 .prepare(
121 "SELECT * FROM queue_entries
122 WHERE pull_id = ? AND state IN ('waiting', 'testing', 'passed') LIMIT 1",
123 )
124 .bind(&[pull_id.into()])?
125 .first::<EntryRow>(None)
126 .await?,
127 };
128 Ok(row.map(|row| (row.state(), row.ahead())))
129 }
130
Merge branch 'worktree-agent-ab8033ab973f8d99a'131 pub(crate) async fn changed(&self, repo_id: &str) -> Result<()> {
Checkout tries a failed fetch twice more, after 2 s and 5 s132 self.publish_as(
133 "queue.changed",
134 repo_id,
135 None,
136 QueueChanged {
137 repo_id: repo_id.to_owned(),
138 },
139 )
140 .await
141 }
142
143 /// Puts a pull request that may merge into the queue instead. Merging
144 /// it again while it is queued changes nothing.
145 pub(crate) async fn enqueue(
146 &self,
147 repo: &Repo,
148 pull: &Pull,
149 actor: &User,
150 keep_issue_open: bool,
151 ) -> Result<Outcome<Pull>> {
152 if self.queued_entry(&pull.id).await?.is_some() {
153 return Ok(Outcome::Ok(pull.clone()));
154 }
155 let now = now_ms();
156 self.db
157 .prepare(
158 "INSERT INTO queue_entries
159 (id, repo_id, pull_id, number, state, enqueued_by, keep_issue_open, created_at)
160 VALUES (?, ?, ?, ?, 'waiting', ?, ?, ?)",
161 )
162 .bind(&[
163 new_id("qen", now).into(),
164 repo.id.as_str().into(),
165 pull.id.as_str().into(),
166 pull.number.into(),
167 serde_json::to_string(actor)?.into(),
168 u32::from(keep_issue_open).into(),
169 rfc3339(now).into(),
170 ])?
171 .run()
172 .await?;
173 let who = (actor.id.as_str(), actor.username.as_str());
174 self.note(&repo.id, pull.number, who, "added this to the merge queue")
175 .await?;
176 self.changed(&repo.id).await?;
177 Ok(Outcome::Ok(pull.clone()))
178 }
179
180 pub(crate) async fn queue(&self, a: QueueArgs) -> Result<Outcome<QueueView>> {
181 let repo = match self.repo(&a.repo, &a.viewer).await? {
182 Outcome::Ok(repo) => repo,
183 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
184 };
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge185 let enabled = self.default_branch_settings(&repo).await?.merge_queue;
Checkout tries a failed fetch twice more, after 2 s and 5 s186 let (active, recent) = (self.entries(&repo.id, true).await?, self.entries(&repo.id, false).await?);
187 let numbers: Vec<u32> = active.iter().chain(&recent).map(|row| row.number).collect();
188 let pulls = try_join_all(numbers.iter().map(|number| self.pull(&repo.id, *number))).await?;
189 let view = |rows: Vec<EntryRow>, pulls: &[Option<Pull>]| -> Vec<QueueEntry> {
190 rows.into_iter()
191 .zip(pulls)
192 .map(|(row, pull)| QueueEntry {
193 state: row.state(),
194 ahead: row.ahead(),
195 enqueued_by: row.actor().map_or_else(|| "g1t".to_owned(), |actor| actor.username),
196 title: pull.as_ref().map(|p| p.title.clone()).unwrap_or_default(),
197 agent: pull.as_ref().map(|p| p.agent.clone()).unwrap_or_default(),
198 results: row
199 .results
200 .as_deref()
201 .and_then(|results| serde_json::from_str(results).ok())
202 .unwrap_or_default(),
203 id: row.id,
204 number: row.number,
205 base_commit: row.base_commit,
206 combined_commit: row.combined_commit,
207 error: row.error,
208 created_at: row.created_at,
209 finished_at: row.finished_at,
210 })
211 .collect()
212 };
213 let split = active.len();
214 Ok(Outcome::Ok(QueueView {
215 enabled,
216 active: view(active, &pulls[..split]),
217 recent: view(recent, &pulls[split..]),
218 }))
219 }
220
221 /// Takes a pull request out of the queue, by the hand of someone who
222 /// may merge.
223 pub(crate) async fn remove_from_queue(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
224 let viewer = Some(a.actor.clone());
225 let (repo, pull) = match self.pull_at(&a.repo, a.number, &viewer).await? {
226 Outcome::Ok(found) => found,
227 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
228 };
229 if let Outcome::Fail(failure) = crate::retired::writable(&repo) {
230 return Ok(Outcome::Fail(failure));
231 }
232 if !a.actor.verified {
233 return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED));
234 }
235 if let Outcome::Fail(failure) =
236 crate::allowed(Some(&a.actor), &repo, g1t_contracts::access::Capability::Merge)
237 {
238 return Ok(Outcome::Fail(failure));
239 }
240 if self.leave(&repo.id, &pull, QueueState::Removed, None).await? {
241 let who = (a.actor.id.as_str(), a.actor.username.as_str());
242 self.note(&repo.id, pull.number, who, "removed this from the merge queue")
243 .await?;
244 self.changed(&repo.id).await?;
245 }
246 Ok(Outcome::Ok(pull))
247 }
248
249 /// Takes a pull request's entry out of the queue, and sends every entry
250 /// whose tested state included it back to waiting. Whether it had one.
251 pub(crate) async fn leave(
252 &self,
253 repo_id: &str,
254 pull: &Pull,
255 state: QueueState,
256 error: Option<&str>,
257 ) -> Result<bool> {
258 let active = self.entries(repo_id, true).await?;
259 let Some(entry) = active.iter().find(|row| row.pull_id == pull.id) else {
260 return Ok(false);
261 };
262 let now = rfc3339(now_ms());
263 self.db
264 .prepare(
265 "UPDATE queue_entries SET state = ?, error = COALESCE(?, error), finished_at = ?
266 WHERE id = ?",
267 )
268 .bind(&[
269 state.as_str().into(),
270 error.map_or(JsValue::NULL, JsValue::from),
271 now.as_str().into(),
272 entry.id.as_str().into(),
273 ])?
274 .run()
275 .await?;
276 self.drop_branch(entry).await;
277 let behind: Vec<&EntryRow> = active
278 .iter()
279 .filter(|row| row.state() != QueueState::Waiting && row.ahead().contains(&pull.number))
280 .collect();
281 self.retest(&behind).await?;
282 Ok(true)
283 }
284
285 /// Removes an entry's tested state from the repository once it has
286 /// left the queue, as GitHub does with its queue's branches. A branch
287 /// left behind is untidy, not wrong, so a failure only logs.
288 async fn drop_branch(&self, row: &EntryRow) {
Merge branch 'worktree-agent-ab8033ab973f8d99a'289 self.drop_branch_named(&row.repo_id, &row.branch()).await;
290 }
291
292 /// Removes one of the queue's branches.
293 pub(crate) async fn drop_branch_named(&self, repo_id: &str, branch: &str) {
Checkout tries a failed fetch twice more, after 2 s and 5 s294 let deleted: Result<Outcome<bool>> = g1t_kit::call(
295 &self.repos,
296 "delete_branch",
297 &DeleteBranchArgs {
Merge branch 'worktree-agent-ab8033ab973f8d99a'298 repo_id: repo_id.to_owned(),
299 branch: branch.to_owned(),
Merge update PRs close themselves: g1t closes its security and version updates once they are no longer needed, and deletes their branches300 head: None,
Checkout tries a failed fetch twice more, after 2 s and 5 s301 },
302 )
303 .await;
304 match deleted {
305 Ok(Outcome::Ok(_)) => {}
Merge branch 'worktree-agent-ab8033ab973f8d99a'306 Ok(Outcome::Fail(failure)) => worker::console_warn!("{branch}: {}", failure.message),
307 Err(error) => worker::console_warn!("{branch}: {error}"),
Checkout tries a failed fetch twice more, after 2 s and 5 s308 }
309 }
310
311 /// Sends entries back to waiting, to be tested again.
312 async fn retest(&self, rows: &[&EntryRow]) -> Result<()> {
313 for row in rows {
314 self.db
315 .prepare(
316 "UPDATE queue_entries
317 SET state = 'waiting', token_hash = NULL, combined_commit = NULL,
Merge branch 'worktree-agent-ab8033ab973f8d99a'318 results = NULL, error = NULL, ahead = NULL, held = NULL
Checkout tries a failed fetch twice more, after 2 s and 5 s319 WHERE id = ? AND state IN ('testing', 'passed')",
320 )
321 .bind(&[row.id.as_str().into()])?
322 .run()
323 .await?;
324 }
325 Ok(())
326 }
327
328 /// The next batch to test, if nothing is being tested now: one job per
329 /// entry, each building the default branch with that entry and every
330 /// entry ahead of it.
Merge branch 'worktree-agent-ab8033ab973f8d99a'331 ///
332 /// Smaller combinations a failure is being narrowed down with
333 /// (integration.rs) are built alongside, whatever the queue is doing.
Checkout tries a failed fetch twice more, after 2 s and 5 s334 pub(crate) async fn queue_build(&self, a: QueueBuildArgs) -> Result<Vec<QueueJob>> {
Merge branch 'worktree-agent-ab8033ab973f8d99a'335 let mut jobs = self.probe_jobs(&a.repo_id).await?;
336 jobs.extend(self.queue_batch(a).await?);
337 Ok(jobs)
338 }
339
340 async fn queue_batch(&self, a: QueueBuildArgs) -> Result<Vec<QueueJob>> {
Checkout tries a failed fetch twice more, after 2 s and 5 s341 let active = self.entries(&a.repo_id, true).await?;
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge342 // How the default branch's merge queue rule batches (rulesets.rs).
343 let rule = self.queue_rule(&a.repo_id).await?;
Checkout tries a failed fetch twice more, after 2 s and 5 s344 // A batch that has taken too long is tested again.
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge345 let stale = minutes_ago(u64::from(rule.check_response_timeout_minutes).max(1));
Checkout tries a failed fetch twice more, after 2 s and 5 s346 let stuck: Vec<&EntryRow> = active
347 .iter()
348 .filter(|row| {
349 row.state() == QueueState::Testing
350 && row.tested_at.as_deref().is_none_or(|at| at < stale.as_str())
351 })
352 .collect();
353 if !stuck.is_empty() {
354 self.retest(&stuck).await?;
Merge branch 'worktree-agent-ab8033ab973f8d99a'355 return Box::pin(self.queue_batch(a)).await;
Checkout tries a failed fetch twice more, after 2 s and 5 s356 }
357 if active.iter().any(|row| row.state() != QueueState::Waiting) {
358 return Ok(Vec::new());
359 }
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge360 let batch: Vec<EntryRow> = active.into_iter().take(rule.max_entries_to_build.max(1) as usize).collect();
Checkout tries a failed fetch twice more, after 2 s and 5 s361 let Some(first) = batch.first() else {
362 return Ok(Vec::new());
363 };
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge364 // Too few to start yet, and the oldest has not waited long enough.
365 if batch.len() < rule.min_entries_to_merge as usize
366 && first.created_at.as_str() > minutes_ago(u64::from(rule.min_entries_wait_minutes)).as_str()
367 {
368 return Ok(Vec::new());
369 }
Checkout tries a failed fetch twice more, after 2 s and 5 s370 let Some(actor) = first.actor() else {
371 return Ok(Vec::new());
372 };
373 let repo: Outcome<Repo> = g1t_kit::call(
374 &self.repos,
375 "get_by_id",
376 &GetByIdArgs {
377 id: a.repo_id.clone(),
378 viewer: Some(actor),
379 },
380 )
381 .await?;
382 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
383 return Ok(Vec::new());
384 };
385 let base: Option<String> = g1t_kit::call(
386 &self.repos,
387 "head",
388 &HeadArgs {
389 repo_id: repo.id.clone(),
390 branch: repo.default_branch.clone(),
391 },
392 )
393 .await?;
394 let Some(base) = base else {
395 return Ok(Vec::new());
396 };
397
398 // Each entry's change, as it is now.
399 let mut items: Vec<(EntryRow, Pull, QueueStackItem)> = Vec::new();
400 for row in batch {
401 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
402 continue;
403 };
404 if pull.status != PullStatus::Open {
405 self.leave(&repo.id, &pull, QueueState::Removed, Some("It was closed."))
406 .await?;
407 continue;
408 }
409 let head: Option<String> = g1t_kit::call(
410 &self.repos,
411 "head",
412 &HeadArgs {
413 repo_id: pull.fork_repo_id.clone().unwrap_or_else(|| repo.id.clone()),
414 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
415 },
416 )
417 .await?;
418 let Some(commit) = head else {
419 continue;
420 };
421 let source = pull.fork.clone().unwrap_or_else(|| RepoPath {
422 namespace: repo.namespace.clone(),
423 name: repo.name.clone(),
424 });
425 let item = QueueStackItem {
426 number: pull.number,
427 title: pull.title.clone(),
428 source,
429 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
430 commit,
431 };
432 items.push((row, pull, item));
433 }
434
435 let now = rfc3339(now_ms());
436 let mut jobs = Vec::new();
437 for index in 0..items.len() {
438 let (row, pull, _) = &items[index];
439 let stack: Vec<QueueStackItem> = items[..=index].iter().map(|(_, _, item)| item.clone()).collect();
440 let ahead: Vec<u32> = stack[..index].iter().map(|item| item.number).collect();
441 let token = new_token();
442 self.db
443 .prepare(
444 "UPDATE queue_entries
445 SET state = 'testing', token_hash = ?, head_commit = ?, base_commit = ?,
446 ahead = ?, combined_commit = NULL, results = NULL, error = NULL,
447 tested_at = ?
448 WHERE id = ? AND state = 'waiting'",
449 )
450 .bind(&[
451 hash(&token).into(),
452 stack[index].commit.as_str().into(),
453 base.as_str().into(),
454 serde_json::to_string(&ahead)?.into(),
455 now.as_str().into(),
456 row.id.as_str().into(),
457 ])?
458 .run()
459 .await?;
460 // The sandbox reads the change and pushes the tested state as
461 // the pull request's owner (whoever asked g1t for it, or its
462 // author): a real account, whose token carries its memberships.
463 // (Whoever queued it may be g1t itself.)
464 let actor = pull.owner().clone();
465 jobs.push(QueueJob {
466 entry_id: row.id.clone(),
467 token,
468 repo: RepoPath {
469 namespace: repo.namespace.clone(),
470 name: repo.name.clone(),
471 },
472 default_branch: repo.default_branch.clone(),
473 base_commit: base.clone(),
474 branch: row.branch(),
475 // The state is checked by its merge_group workflows.
476 contract_checks: Vec::new(),
477 stack,
478 checks: Vec::new(),
479 actor,
480 });
481 }
482 Ok(jobs)
483 }
484
485 /// A sandbox's result for one combined state.
486 pub(crate) async fn report_queue(&self, a: ReportQueueArgs) -> Result<Outcome<QueueState>> {
Merge branch 'worktree-agent-ab8033ab973f8d99a'487 // A smaller combination built to narrow down a failure.
488 if a.entry_id.starts_with("qpr_") {
489 return self.report_probe(a).await;
490 }
Checkout tries a failed fetch twice more, after 2 s and 5 s491 let row = self
492 .db
493 .prepare("SELECT * FROM queue_entries WHERE id = ? AND token_hash = ?")
494 .bind(&[a.entry_id.as_str().into(), hash(&a.token).into()])?
495 .first::<EntryRow>(None)
496 .await?;
497 let Some(row) = row else {
498 return Ok(Outcome::fail(FailureCode::NotFound, "No such queue entry."));
499 };
500 if row.state() != QueueState::Testing {
501 return Ok(Outcome::fail(
502 FailureCode::Conflict,
503 "This state is no longer being tested.",
504 ));
505 }
506 let built = a.error.is_none() && a.results.iter().all(|result| result.passed);
507 // A state that was built runs the repository's `merge_group`
508 // workflows; it stays in testing until they finish (see `statuses`),
509 // and the branch's required checks must pass on it.
510 let workflows = match (&a.combined_commit, built) {
Merge branch 'worktree-agent-ab8033ab973f8d99a'511 (Some(commit), true) => self
512 .run_merge_group(
513 &row.repo_id,
514 &row.id,
515 &row.branch(),
516 row.base_commit.as_deref(),
517 row.number,
518 &row.ahead(),
519 commit,
520 )
521 .await
522 .unwrap_or(0),
Checkout tries a failed fetch twice more, after 2 s and 5 s523 _ => 0,
524 };
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge525 let required = self.settings_by_id(&row.repo_id).await?.required_checks;
Checkout tries a failed fetch twice more, after 2 s and 5 s526 // Nothing runs on it, so the required checks never would report.
527 let unchecked = (built && workflows == 0 && !required.is_empty()).then(|| {
528 format!(
529 "the required {} {} cannot report on it: no workflow runs on merge_group events. Add merge_group to the on: of the workflows the branch requires",
530 if required.len() == 1 { "check" } else { "checks" },
531 crate::statuses::list(&required)
532 )
533 });
534 let passed = built && unchecked.is_none();
Merge branch 'worktree-agent-ab8033ab973f8d99a'535 let error = a.error.clone().or(unchecked.clone());
536 let report = ReportQueueArgs { error: error.clone(), ..a };
537 let failure = (!passed).then(|| {
538 crate::integration::Failure::of_report(&report, unchecked.is_some().then_some(&required[..]))
539 });
540 // Failed while an entry ahead of it is still being tested: held
541 // until those ahead are settled, so that it is not blamed for
542 // theirs (integration.rs).
543 let held = match &failure {
544 Some(failure) if self.ahead_unsettled(&row).await? => Some(serde_json::to_string(failure)?),
545 _ => None,
546 };
547 let state = if passed && workflows == 0 {
548 QueueState::Passed
549 } else if passed || held.is_some() {
Checkout tries a failed fetch twice more, after 2 s and 5 s550 QueueState::Testing
551 } else {
Merge branch 'worktree-agent-ab8033ab973f8d99a'552 QueueState::Failed
Checkout tries a failed fetch twice more, after 2 s and 5 s553 };
554 self.db
555 .prepare(
556 "UPDATE queue_entries
Merge branch 'worktree-agent-ab8033ab973f8d99a'557 SET state = ?, combined_commit = ?, results = ?, error = ?, token_hash = NULL, held = ?,
Checkout tries a failed fetch twice more, after 2 s and 5 s558 finished_at = CASE WHEN ? = 'failed' THEN ? ELSE NULL END
559 WHERE id = ?",
560 )
561 .bind(&[
562 state.as_str().into(),
Merge branch 'worktree-agent-ab8033ab973f8d99a'563 report.combined_commit.as_deref().map_or(JsValue::NULL, JsValue::from),
564 serde_json::to_string(&report.results)?.into(),
Checkout tries a failed fetch twice more, after 2 s and 5 s565 error.as_deref().map_or(JsValue::NULL, JsValue::from),
Merge branch 'worktree-agent-ab8033ab973f8d99a'566 held.as_deref().map_or(JsValue::NULL, JsValue::from),
Checkout tries a failed fetch twice more, after 2 s and 5 s567 state.as_str().into(),
568 rfc3339(now_ms()).into(),
569 row.id.as_str().into(),
570 ])?
571 .run()
572 .await?;
Merge branch 'worktree-agent-ab8033ab973f8d99a'573 if let Some(failure) = failure.filter(|_| held.is_none()) {
Checkout tries a failed fetch twice more, after 2 s and 5 s574 self.eject(&row, &report).await?;
Merge branch 'worktree-agent-ab8033ab973f8d99a'575 self.diagnose(&row, failure).await?;
576 }
577 if state == QueueState::Passed {
578 self.integration_passed(&row, report.combined_commit.as_deref()).await?;
579 self.resolve_held(&row.repo_id).await?;
Checkout tries a failed fetch twice more, after 2 s and 5 s580 }
581 self.settle(&row.repo_id).await?;
582 self.changed(&row.repo_id).await?;
583 Ok(Outcome::Ok(state))
584 }
585
586 /// Asks the actions service to run the repository's `merge_group`
Merge branch 'worktree-agent-ab8033ab973f8d99a'587 /// workflows on a combined state: `entry`'s, pushed to `branch`, with
588 /// `ahead` merged in before `number`. Returns how many runs started.
589 #[allow(clippy::too_many_arguments)]
590 pub(crate) async fn run_merge_group(
591 &self,
592 repo_id: &str,
593 entry: &str,
594 branch: &str,
595 base_commit: Option<&str>,
596 number: u32,
597 ahead: &[u32],
598 commit: &str,
599 ) -> Result<u32> {
Checkout tries a failed fetch twice more, after 2 s and 5 s600 #[derive(serde::Deserialize)]
601 struct Started {
602 runs: u32,
603 }
604 let started: Outcome<Started> = g1t_kit::call(
605 &self.actions,
606 "merge_group",
607 &serde_json::json!({
Merge branch 'worktree-agent-ab8033ab973f8d99a'608 "repoId": repo_id,
609 "entry": entry,
Checkout tries a failed fetch twice more, after 2 s and 5 s610 "sha": commit,
Merge branch 'worktree-agent-ab8033ab973f8d99a'611 "headRef": format!("refs/heads/{branch}"),
612 "baseSha": base_commit,
613 "number": number,
614 "ahead": ahead,
Checkout tries a failed fetch twice more, after 2 s and 5 s615 }),
616 )
617 .await?;
618 Ok(match started {
619 Outcome::Ok(started) => started.runs,
620 Outcome::Fail(_) => 0,
621 })
622 }
623
624 /// Workflows on a combined state finished: it passes and lands in turn
625 /// when they all passed and so did every required check, or fails and
626 /// leaves the queue.
627 pub(crate) async fn merge_group_finished(
628 &self,
629 repo_id: &str,
630 commit: &str,
631 facts: &crate::statuses::WorkflowFacts,
632 ) -> Result<()> {
633 let row = self
634 .db
635 .prepare(
Merge branch 'worktree-agent-ab8033ab973f8d99a'636 "SELECT * FROM queue_entries
637 WHERE repo_id = ? AND combined_commit = ? AND state = 'testing' AND token_hash IS NULL AND held IS NULL",
Checkout tries a failed fetch twice more, after 2 s and 5 s638 )
639 .bind(&[repo_id.into(), commit.into()])?
640 .first::<EntryRow>(None)
641 .await?;
642 let missing = facts.expected();
643 let passed = facts.failed.is_empty() && facts.required_failed().is_empty() && missing.is_empty();
Merge branch 'worktree-agent-ab8033ab973f8d99a'644 let Some(row) = row else {
645 // Or a smaller combination built to narrow down a failure.
646 if self.probe_finished(repo_id, commit, passed).await? {
647 self.changed(repo_id).await?;
648 }
649 return Ok(());
650 };
651 let failure = if passed {
652 None
653 } else {
654 let error = if facts.failed.is_empty() && facts.required_failed().is_empty() {
655 format!(
656 "the required {} {} did not report on it. Add merge_group to the on: of the workflows the branch requires",
657 if missing.len() == 1 { "check" } else { "checks" },
658 crate::statuses::list(&missing)
659 )
660 } else if facts.failed.is_empty() {
661 format!("the required {} failed on it", crate::statuses::list(&facts.required_failed()))
662 } else {
663 format!("the workflow {} failed on it", crate::statuses::list(&facts.failed))
664 };
665 // Where to see the check that failed: its run's page.
666 let statuses = self.statuses(repo_id, Some(commit)).await?;
667 let url = |name: &str| {
668 statuses
669 .iter()
670 .find(|status| status.context == name || check_name(&status.context).0 == name)
671 .and_then(|status| status.target_url.clone())
672 };
673 Some(crate::integration::Failure::of_workflows(
674 commit,
675 &facts.failed,
676 &facts.required_failed(),
677 &missing,
678 error,
679 url,
680 ))
681 };
682 let held = match &failure {
683 Some(failure) if self.ahead_unsettled(&row).await? => Some(serde_json::to_string(failure)?),
684 _ => None,
685 };
686 let state = if passed {
687 "passed"
688 } else if held.is_some() {
689 "testing"
690 } else {
691 "failed"
692 };
693 let moved = self
694 .db
695 .prepare(
696 "UPDATE queue_entries SET state = ?, held = ?, finished_at = CASE WHEN ? = 'failed' THEN ? ELSE NULL END
697 WHERE id = ? AND state = 'testing' AND held IS NULL RETURNING id AS value",
698 )
Checkout tries a failed fetch twice more, after 2 s and 5 s699 .bind(&[
Merge branch 'worktree-agent-ab8033ab973f8d99a'700 state.into(),
701 held.as_deref().map_or(JsValue::NULL, JsValue::from),
702 state.into(),
Checkout tries a failed fetch twice more, after 2 s and 5 s703 rfc3339(now_ms()).into(),
704 row.id.as_str().into(),
705 ])?
Merge branch 'worktree-agent-ab8033ab973f8d99a'706 .first::<crate::rows::ValueRow>(None)
Checkout tries a failed fetch twice more, after 2 s and 5 s707 .await?;
Merge branch 'worktree-agent-ab8033ab973f8d99a'708 if moved.is_none() {
709 return Ok(());
710 }
711 if let Some(failure) = failure.filter(|_| held.is_none()) {
712 self.eject(&row, &failure.report(&row.id)).await?;
713 self.diagnose(&row, failure).await?;
Checkout tries a failed fetch twice more, after 2 s and 5 s714 }
Merge branch 'worktree-agent-ab8033ab973f8d99a'715 if passed {
716 self.integration_passed(&row, Some(commit)).await?;
717 self.resolve_held(repo_id).await?;
718 }
Checkout tries a failed fetch twice more, after 2 s and 5 s719 self.settle(repo_id).await?;
720 self.changed(repo_id).await?;
721 Ok(())
722 }
723
724 /// An entry whose combined state failed: the entries tested on top of
725 /// it are tested again without it, and the failure is recorded as a
The work, events, webhooks, actions, repos and packages services no longer run g1t's own agents. Work drops the 35 calls that queued issues for g1t, delegated them, messaged and reviewed through the agent, planned, captured memories automatically, rated confidence and capped agent spend, and its pull request lifecycle no longer revises or reviews: a pull request only advances to a merge by the repository's auto-merge setting. Merging a pull request that is behind now catches it up itself, through the repository's branch update, and lands it when the push arrives; when it can't be caught up cleanly the merge says to merge the base in and try again. Recording a session on a pull request, memory recall and remember, overlaps, the merge queue and required checks stay as they were.726 /// failed check run of its pull request.
Merge branch 'worktree-agent-ab8033ab973f8d99a'727 pub(crate) async fn eject(&self, row: &EntryRow, report: &ReportQueueArgs) -> Result<()> {
Checkout tries a failed fetch twice more, after 2 s and 5 s728 self.drop_branch(row).await;
729 let active = self.entries(&row.repo_id, true).await?;
730 let behind: Vec<&EntryRow> = active
731 .iter()
732 .filter(|other| other.state() != QueueState::Waiting && other.ahead().contains(&row.number))
733 .collect();
734 self.retest(&behind).await?;
735
736 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
737 return Ok(());
738 };
739 let ahead = row.ahead();
740 let state = if ahead.is_empty() {
741 "the default branch as it is now".to_owned()
742 } else {
743 format!(
744 "the default branch with {} merged in first",
745 ahead.iter().map(|n| format!("#{n}")).collect::<Vec<_>>().join(", ")
746 )
747 };
748 // Named in backticks, so the conversation can link each to the diff.
749 let files = crate::mergeability::tidy(report.conflicts.clone())
750 .iter()
751 .map(|path| format!("`{path}`"))
752 .collect::<Vec<_>>()
753 .join(", ");
754 let why = match (&report.error, report.conflict_with) {
755 (_, Some(other)) if other != row.number && !files.is_empty() => format!(
Merge branch 'worktree-agent-ab8033ab973f8d99a'756 "Its change conflicts with #{other}, which is ahead of it in the merge queue, in {files}. Bring it up to date with the default branch once #{other} lands and push: it rejoins the merge queue once its checks pass."
Checkout tries a failed fetch twice more, after 2 s and 5 s757 ),
758 (_, Some(other)) if other != row.number => format!(
Merge branch 'worktree-agent-ab8033ab973f8d99a'759 "Its change conflicts with #{other}, which is ahead of it in the merge queue. Bring it up to date with the default branch once #{other} lands and push: it rejoins the merge queue once its checks pass."
Checkout tries a failed fetch twice more, after 2 s and 5 s760 ),
761 (Some(_), None) if !files.is_empty() => format!(
Merge branch 'worktree-agent-ab8033ab973f8d99a'762 "Its change conflicts with the default branch in {files}. Bring it up to date with the default branch and push: it rejoins the merge queue once its checks pass."
Checkout tries a failed fetch twice more, after 2 s and 5 s763 ),
764 (Some(error), _) if error.starts_with("the workflow ") => {
765 format!("{} when it was combined with {state}.", error.replacen("the workflow", "The workflow", 1).trim_end_matches(" on it"))
766 }
767 (Some(error), _) if error.starts_with("the required ") => {
768 format!("Combined with {state}, {error}.")
769 }
770 (Some(error), _) => format!("Its combined state could not be built or checked: {error}"),
771 (None, _) => format!(
772 "It failed when it was combined with {state}, though it may pass on its own."
773 ),
774 };
775 let now = now_ms();
776 let run_id = new_id("chk", now);
777 let mut results = report.results.clone();
778 for result in &mut results {
779 result.command = format!("{} (merge queue, on {state})", result.command);
780 }
781 self.db
782 .batch(vec![
783 self.db
784 .prepare(
785 "INSERT INTO check_runs
786 (id, pull_id, head_commit, status, results, error, token_hash,
787 created_at, finished_at)
788 VALUES (?, ?, ?, 'failed', ?, ?, ?, ?, ?)",
789 )
790 .bind(&[
791 run_id.as_str().into(),
792 pull.id.as_str().into(),
793 row.head_commit.as_deref().unwrap_or_default().into(),
794 serde_json::to_string(&results)?.into(),
795 why.as_str().into(),
796 hash(&new_token()).into(),
797 rfc3339(now).into(),
798 rfc3339(now).into(),
799 ])?,
800 self.db
801 .prepare("UPDATE pulls SET check_status = 'failed', check_run_id = ? WHERE id = ?")
802 .bind(&[run_id.as_str().into(), pull.id.as_str().into()])?,
Merge branch 'worktree-agent-ab8033ab973f8d99a'803 // Its merge stays asked for: it rejoins once a fix's checks
804 // pass (integration.rs `requeue_if_fixed`).
805 self.db
806 .prepare("UPDATE queue_entries SET requeue = 1 WHERE id = ?")
807 .bind(&[row.id.as_str().into()])?,
Checkout tries a failed fetch twice more, after 2 s and 5 s808 ])
809 .await?;
810 self.note(
811 &row.repo_id,
812 pull.number,
813 ("g1t", "g1t"),
814 &format!("was taken out of the merge queue. {why}"),
815 )
816 .await?;
817 self.publish_as(
818 "checks.completed",
819 &row.repo_id,
820 None,
821 ChecksEvent {
822 pull_id: pull.id.clone(),
823 repo_id: row.repo_id.clone(),
824 number: pull.number,
825 status: "failed",
826 commit: row.head_commit.clone().unwrap_or_default(),
827 },
828 )
829 .await?;
830 Ok(())
831 }
832
833 /// Lands every entry at the front of the queue whose tested state
834 /// passed, in order.
835 async fn settle(&self, repo_id: &str) -> Result<()> {
836 let active = self.entries(repo_id, true).await?;
837 let Some(viewer) = active.first().and_then(EntryRow::actor) else {
838 return Ok(());
839 };
840 let repo: Outcome<Repo> = g1t_kit::call(
841 &self.repos,
842 "get_by_id",
843 &GetByIdArgs {
844 id: repo_id.to_owned(),
845 viewer: Some(viewer),
846 },
847 )
848 .await?;
849 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
850 return Ok(());
851 };
852 for row in active {
853 if row.state() != QueueState::Passed {
854 // The front has not passed yet: nothing behind it may land.
855 return Ok(());
856 }
857 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
858 continue;
859 };
860 let Some(actor) = row.actor() else {
861 continue;
862 };
863 // Pushed to since it was tested: test it again as it is now.
864 if pull.status != PullStatus::Open {
865 self.leave(repo_id, &pull, QueueState::Removed, Some("It was closed."))
866 .await?;
867 continue;
868 }
869 let head: Option<String> = g1t_kit::call(
870 &self.repos,
871 "head",
872 &HeadArgs {
873 repo_id: pull.fork_repo_id.clone().unwrap_or_else(|| repo_id.to_owned()),
874 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
875 },
876 )
877 .await?;
878 if head.is_some() && head != row.head_commit {
879 let active = self.entries(repo_id, true).await?;
880 let again: Vec<&EntryRow> = active
881 .iter()
882 .filter(|other| other.id == row.id || other.ahead().contains(&row.number))
883 .collect();
884 self.retest(&again).await?;
885 return Ok(());
886 }
887 let landed: Outcome<Landed> = g1t_kit::call(
888 &self.repos,
889 "land",
890 &LandArgs {
891 source_id: repo_id.to_owned(),
892 branch: Some(row.branch()),
893 actor: actor.clone(),
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar894 // The queue lands on the default branch.
895 target_branch: None,
Checkout tries a failed fetch twice more, after 2 s and 5 s896 },
897 )
898 .await?;
899 let landed = match landed {
900 Outcome::Ok(landed) => landed,
901 Outcome::Fail(_) => {
902 // The default branch moved outside the queue: every
903 // tested state is built on something that is gone.
904 let active = self.entries(repo_id, true).await?;
905 let all: Vec<&EntryRow> = active.iter().collect();
906 self.retest(&all).await?;
907 return Ok(());
908 }
909 };
910 self.db
911 .prepare(
912 "UPDATE queue_entries SET state = 'landed', finished_at = ? WHERE id = ?",
913 )
914 .bind(&[rfc3339(now_ms()).into(), row.id.as_str().into()])?
915 .run()
916 .await?;
917 self.drop_branch(&row).await;
918 self.record_merge(&repo, pull, &actor, row.keep_issue_open != 0, landed)
919 .await?;
920 }
921 Ok(())
922 }
923}

This file's history is long; its oldest lines are credited to the oldest commit read.