Sign in

flagon-io/g1t

public

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

gitagentscloudflarerustforgemcp

g1t/services/work/src/lib.rs

1,851 lines69,656 bytesCodeBlame

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.

Issues and pull requests replace intents and attempts1//! The work service: issues, pull requests, comments and sessions.
Work service in Rust, with RFC 3339 timestamps2//!
3//! Other services reach it over `POST /rpc/<method>`; see
4//! `g1t_contracts::work` for the methods and their arguments. It also
5//! consumes its queue of events from the bus.
6
Acceptance checks in sandboxes, line comments and review verdicts7mod checks;
Agents as a team: lifecycle, merge queue, billing and a new shell8mod lifecycle;
9mod plans;
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request10mod messages;
Agents as a team: lifecycle, merge queue, billing and a new shell11mod queue;
12mod reviews;
Work service in Rust, with RFC 3339 timestamps13mod rows;
Agents as a team: lifecycle, merge queue, billing and a new shell14mod settings;
GitHub Actions on g1t, part two: running workflows15mod statuses;
Work service in Rust, with RFC 3339 timestamps16
17use g1t_contracts::events::{
Events service in Rust, with RFC 3339 times and accurate push events18 CommentCreated, Event, IssueEvent, NewEvent, Publish, PullEvent, SessionAppended,
Work service in Rust, with RFC 3339 timestamps19};
Agents as a team: lifecycle, merge queue, billing and a new shell20use g1t_contracts::identity::UsernameArgs;
Pull requests from branches21use g1t_contracts::repos::{ForkArgs, GetArgs, HeadArgs, LandArgs, Landed, Repo, RepoPath};
Work service in Rust, with RFC 3339 timestamps22use g1t_contracts::time::rfc3339;
23use g1t_contracts::work::*;
Agents as a team: lifecycle, merge queue, billing and a new shell24use futures_util::future::{try_join, try_join3, try_join_all};
Work service in Rust, with RFC 3339 timestamps25use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
Events service in Rust, with RFC 3339 times and accurate push events26use g1t_kit::{args, now_ms, reply, rpc_method};
Work service in Rust, with RFC 3339 timestamps27use serde::Serialize;
28use worker::wasm_bindgen::JsValue;
29use worker::{
30 Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, event,
31};
32
Agents as a team: lifecycle, merge queue, billing and a new shell33use rows::{CommentRow, IssueRow, MovedRow, NumberRow, PullRow, SessionRow, Snapshot, ValueRow};
Work service in Rust, with RFC 3339 timestamps34
35const SOURCE: &str = "work";
36const MAX_ENTRY_BATCH: usize = 200;
37const MAX_ENTRY_CHARS: usize = 64_000;
Issues and pull requests replace intents and attempts38const MAX_TITLE_CHARS: usize = 200;
Work service in Rust, with RFC 3339 timestamps39const SESSION_PAGE: u32 = 500;
Issues and pull requests replace intents and attempts40const LIST_PAGE: u32 = 100;
Agents as a team: lifecycle, merge queue, billing and a new shell41const MAX_ASSIGNEES: usize = 10;
Work service in Rust, with RFC 3339 timestamps42const UNVERIFIED: &str = "Confirm your email address first. Check your inbox, or resend the link from the banner on g1t.sh.";
43
Issues and pull requests replace intents and attempts44const ISSUE_COLUMNS: &str = "issues.*,
45 (SELECT count(*) FROM pulls WHERE pulls.issue_id = issues.id) AS pull_count,
Agents as a team: lifecycle, merge queue, billing and a new shell46 (SELECT agent FROM pulls
47 WHERE pulls.issue_id = issues.id AND pulls.status IN ('draft', 'open')
48 AND pulls.fork_repo_id IS NOT NULL
49 ORDER BY pulls.number DESC LIMIT 1) AS agent,
Issues and pull requests replace intents and attempts50 (SELECT count(*) FROM comments
Agents as a team: lifecycle, merge queue, billing and a new shell51 WHERE comments.repo_id = issues.repo_id AND comments.number = issues.number
52 AND comments.kind = 'comment') AS comment_count";
Work service in Rust, with RFC 3339 timestamps53
Issues and pull requests replace intents and attempts54fn no_issue<T>() -> Outcome<T> {
55 Outcome::fail(FailureCode::NotFound, "Issue not found.")
Work service in Rust, with RFC 3339 timestamps56}
57
Issues and pull requests replace intents and attempts58fn no_pull<T>() -> Outcome<T> {
59 Outcome::fail(FailureCode::NotFound, "Pull request not found.")
Work service in Rust, with RFC 3339 timestamps60}
61
62fn optional(value: &Option<String>) -> JsValue {
63 value.as_deref().map_or(JsValue::NULL, JsValue::from)
64}
65
Issues and pull requests replace intents and attempts66fn optional_number(value: Option<u32>) -> JsValue {
67 value.map_or(JsValue::NULL, JsValue::from)
68}
69
70/// The lowercase name a `State` is stored and sent as.
71fn state_name(state: Option<State>) -> Option<&'static str> {
72 state.map(|state| match state {
73 State::Open => "open",
74 State::Closed => "closed",
75 })
76}
77
78/// A trimmed title, or why it cannot be used.
79fn valid_title(title: &str) -> std::result::Result<&str, &'static str> {
80 let title = title.trim();
81 if title.is_empty() {
82 Err("A title is required.")
83 } else if title.chars().count() > MAX_TITLE_CHARS {
84 Err("That title is too long.")
85 } else {
86 Ok(title)
87 }
88}
89
90/// Unwraps an `Outcome`, returning its failure from the enclosing method.
91macro_rules! check {
92 ($outcome:expr) => {
93 match $outcome {
94 Outcome::Ok(value) => value,
95 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
96 }
97 };
98}
99
Work service in Rust, with RFC 3339 timestamps100struct Work {
101 db: D1Database,
Agents as a team: lifecycle, merge queue, billing and a new shell102 identity: Fetcher,
Work service in Rust, with RFC 3339 timestamps103 repos: Fetcher,
Events service in Rust, with RFC 3339 times and accurate push events104 events: Fetcher,
Sidebar: the panels really slide105 /// GitHub Actions: runs a merge queue's `merge_group` workflows.
106 actions: Fetcher,
Work service in Rust, with RFC 3339 timestamps107}
108
109impl Work {
Issues and pull requests replace intents and attempts110 async fn publish<T: Serialize>(
111 &self,
112 kind: &'static str,
113 repo_id: &str,
114 actor: &User,
115 data: T,
116 ) -> Result<()> {
Acceptance checks in sandboxes, line comments and review verdicts117 self.publish_as(kind, repo_id, Some(actor.id.clone()), data)
118 .await
119 }
120
Agents move along on private repositories too121 /// A pull request's author as a viewer who can read its repository and
122 /// source. Stored authors carry no memberships, so a private repository
123 /// would otherwise look missing to them.
124 pub(crate) async fn author_viewer(&self, pull: &Pull) -> Result<Viewer> {
125 let path: Option<RepoPath> = g1t_kit::call(
126 &self.repos,
127 "path_by_id",
128 &g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
129 )
130 .await?;
131 let mut author = pull.author.clone();
132 if let Some(path) = path
133 && !author.is_member(&path.namespace.to_lowercase())
134 {
Workspace names and icons, and a component kit for every control135 author.workspaces.push(g1t_contracts::Membership::member(path.namespace.to_lowercase()));
Agents move along on private repositories too136 }
137 Ok(Some(author))
138 }
139
Acceptance checks in sandboxes, line comments and review verdicts140 /// Publishes an event caused by `actor`, or by g1t itself.
141 async fn publish_as<T: Serialize>(
142 &self,
143 kind: &'static str,
144 repo_id: &str,
145 actor: Option<String>,
146 data: T,
147 ) -> Result<()> {
Issues and pull requests replace intents and attempts148 let event = NewEvent {
149 kind,
150 source: SOURCE,
151 repo_id: Some(repo_id.to_owned()),
Acceptance checks in sandboxes, line comments and review verdicts152 actor,
Issues and pull requests replace intents and attempts153 data,
154 };
Events service in Rust, with RFC 3339 times and accurate push events155 g1t_kit::call(
156 &self.events,
157 "publish",
158 &Publish {
159 events: vec![event],
160 },
161 )
162 .await
Work service in Rust, with RFC 3339 timestamps163 }
164
Issues and pull requests replace intents and attempts165 /// The repository, if the viewer may see it. Whether they may is
166 /// decided by the repos service.
167 async fn repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
Work service in Rust, with RFC 3339 timestamps168 g1t_kit::call(
169 &self.repos,
170 "get",
171 &GetArgs {
172 path: path.clone(),
173 viewer: viewer.clone(),
174 },
175 )
176 .await
177 }
178
Issues and pull requests replace intents and attempts179 /// The next number in the repository's sequence. Taking it is one
180 /// statement, so concurrent opens cannot be given the same number.
181 async fn next_number(&self, repo_id: &str) -> Result<u32> {
182 let row = self
183 .db
184 .prepare(
185 "INSERT INTO counters (repo_id, last) VALUES (?, 1)
186 ON CONFLICT (repo_id) DO UPDATE SET last = last + 1
187 RETURNING last AS n",
188 )
189 .bind(&[repo_id.into()])?
190 .first::<NumberRow>(None)
191 .await?;
192 row.map(|row| row.n)
193 .ok_or_else(|| worker::Error::RustError("no number was assigned".into()))
Work service in Rust, with RFC 3339 timestamps194 }
195
Issues and pull requests replace intents and attempts196 async fn issue(&self, repo_id: &str, number: u32) -> Result<Option<Issue>> {
Work service in Rust, with RFC 3339 timestamps197 Ok(self
198 .db
Issues and pull requests replace intents and attempts199 .prepare(format!(
200 "SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = ? AND number = ?"
201 ))
202 .bind(&[repo_id.into(), number.into()])?
203 .first::<IssueRow>(None)
Work service in Rust, with RFC 3339 timestamps204 .await?
Issues and pull requests replace intents and attempts205 .map(Issue::from))
Work service in Rust, with RFC 3339 timestamps206 }
207
Issues and pull requests replace intents and attempts208 async fn pull(&self, repo_id: &str, number: u32) -> Result<Option<Pull>> {
Work service in Rust, with RFC 3339 timestamps209 Ok(self
210 .db
Issues and pull requests replace intents and attempts211 .prepare("SELECT * FROM pulls WHERE repo_id = ? AND number = ?")
212 .bind(&[repo_id.into(), number.into()])?
213 .first::<PullRow>(None)
214 .await?
215 .map(Pull::from))
216 }
217
218 async fn comments(&self, repo_id: &str, number: u32) -> Result<Vec<Comment>> {
219 let rows = self
220 .db
221 .prepare(
222 "SELECT * FROM comments WHERE repo_id = ? AND number = ? ORDER BY id LIMIT 500",
223 )
224 .bind(&[repo_id.into(), number.into()])?
225 .all()
Work service in Rust, with RFC 3339 timestamps226 .await?
Issues and pull requests replace intents and attempts227 .results::<CommentRow>()?;
228 Ok(rows.into_iter().map(Comment::from).collect())
Work service in Rust, with RFC 3339 timestamps229 }
230
Issues and pull requests replace intents and attempts231 /// The repository and one of its issues, as seen by `viewer`.
232 async fn issue_at(
233 &self,
234 path: &RepoPath,
235 number: u32,
236 viewer: &Viewer,
237 ) -> Result<Outcome<(Repo, Issue)>> {
238 let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
239 return Ok(no_issue());
Work service in Rust, with RFC 3339 timestamps240 };
Issues and pull requests replace intents and attempts241 Ok(match self.issue(&repo.id, number).await? {
242 Some(issue) => Outcome::Ok((repo, issue)),
243 None => no_issue(),
244 })
245 }
246
247 /// The repository and one of its pull requests, as seen by `viewer`.
248 async fn pull_at(
249 &self,
250 path: &RepoPath,
251 number: u32,
252 viewer: &Viewer,
253 ) -> Result<Outcome<(Repo, Pull)>> {
254 let Outcome::Ok(repo) = self.repo(path, viewer).await? else {
255 return Ok(no_pull());
256 };
257 Ok(match self.pull(&repo.id, number).await? {
258 Some(pull) => Outcome::Ok((repo, pull)),
259 None => no_pull(),
260 })
261 }
262
Agents as a team: lifecycle, merge queue, billing and a new shell263 /// Records something that happened to an issue or a pull request, so
264 /// that it shows in the conversation where it happened. `text` is what
265 /// `author` did, as the rest of a sentence starting with their name.
266 pub(crate) async fn note(
267 &self,
268 repo_id: &str,
269 number: u32,
270 author: (&str, &str),
271 text: &str,
272 ) -> Result<()> {
273 let now = now_ms();
274 self.db
275 .prepare(
276 "INSERT INTO comments
277 (id, repo_id, number, author_id, author_name, body, kind, created_at)
278 VALUES (?, ?, ?, ?, ?, ?, 'event', ?)",
279 )
280 .bind(&[
281 new_id("cmt", now).into(),
282 repo_id.into(),
283 number.into(),
284 author.0.into(),
285 author.1.into(),
286 text.into(),
287 rfc3339(now).into(),
288 ])?
289 .run()
290 .await?;
291 Ok(())
292 }
293
294 /// Notes who was added to and removed from a list of people, such as
295 /// "assigned ana" or "requested a review from g1t-agent".
296 async fn note_changes(
297 &self,
298 repo_id: &str,
299 number: u32,
300 actor: &User,
301 before: &[String],
302 after: &[String],
303 (added, removed): (&str, &str),
304 ) -> Result<()> {
305 let joined = |names: Vec<&String>| {
306 names
307 .into_iter()
308 .map(String::as_str)
309 .collect::<Vec<_>>()
310 .join(", ")
311 };
312 let new: Vec<&String> = after.iter().filter(|name| !before.contains(name)).collect();
313 let gone: Vec<&String> = before.iter().filter(|name| !after.contains(name)).collect();
314 let who = (actor.id.as_str(), actor.username.as_str());
315 if !new.is_empty() {
316 // Taking something on oneself reads better said that way.
317 let text = if added == "assigned" && new == [&actor.username] {
318 "self-assigned this".to_owned()
319 } else {
320 format!("{added} {}", joined(new))
321 };
322 self.note(repo_id, number, who, &text).await?;
323 }
324 if !gone.is_empty() {
325 self.note(repo_id, number, who, &format!("{removed} {}", joined(gone)))
326 .await?;
327 }
328 Ok(())
329 }
330
Issues and pull requests replace intents and attempts331 fn issue_event(issue: &Issue) -> IssueEvent {
332 IssueEvent {
333 issue_id: issue.id.clone(),
334 repo_id: issue.repo_id.clone(),
335 number: issue.number,
336 ..IssueEvent::default()
Work service in Rust, with RFC 3339 timestamps337 }
338 }
339
Workflows run when an agent's pull request is marked ready340 /// The commit a pull request's change is at in git right now: its
341 /// fork's default branch, or its branch.
342 async fn live_head(&self, pull: &Pull) -> Result<Option<String>> {
343 g1t_kit::call(
344 &self.repos,
345 "head",
346 &HeadArgs {
347 repo_id: pull.fork_repo_id.clone().unwrap_or_else(|| pull.repo_id.clone()),
348 branch: pull.branch.clone().unwrap_or_default(),
349 },
350 )
351 .await
352 }
353
Issues and pull requests replace intents and attempts354 fn pull_event(pull: &Pull) -> PullEvent {
355 PullEvent {
356 pull_id: pull.id.clone(),
357 repo_id: pull.repo_id.clone(),
358 number: pull.number,
359 issue: pull.issue,
360 ..PullEvent::default()
Work service in Rust, with RFC 3339 timestamps361 }
362 }
363
Issues and pull requests replace intents and attempts364 // --- Issues ------------------------------------------------------------
365
366 async fn open_issue(&self, a: OpenIssueArgs) -> Result<Outcome<Issue>> {
Work service in Rust, with RFC 3339 timestamps367 if !a.actor.verified {
368 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
369 }
Issues and pull requests replace intents and attempts370 let title = match valid_title(&a.title) {
371 Ok(title) => title,
372 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
373 };
374 let Some(labels) = normalize_labels(&a.labels) else {
Work service in Rust, with RFC 3339 timestamps375 return Ok(Outcome::fail(
376 FailureCode::Invalid,
Issues and pull requests replace intents and attempts377 "An issue can have up to 10 labels of up to 40 characters each.",
Work service in Rust, with RFC 3339 timestamps378 ));
379 };
Issues and pull requests replace intents and attempts380 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
Work service in Rust, with RFC 3339 timestamps381 let checks: Vec<&str> = a
382 .checks
383 .iter()
384 .map(|check| check.trim())
385 .filter(|check| !check.is_empty())
386 .collect();
387
388 let now = now_ms();
Issues and pull requests replace intents and attempts389 let id = new_id("iss", now);
390 let number = self.next_number(&repo.id).await?;
391 let timestamp = rfc3339(now);
Work service in Rust, with RFC 3339 timestamps392 self.db
393 .prepare(
Issues and pull requests replace intents and attempts394 "INSERT INTO issues
395 (id, repo_id, number, title, body, labels, checks, author_id, author_name,
396 created_at, updated_at)
397 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Work service in Rust, with RFC 3339 timestamps398 )
399 .bind(&[
400 id.as_str().into(),
401 repo.id.as_str().into(),
Issues and pull requests replace intents and attempts402 number.into(),
Work service in Rust, with RFC 3339 timestamps403 title.into(),
Issues and pull requests replace intents and attempts404 a.body.trim().into(),
405 serde_json::to_string(&labels)?.into(),
Work service in Rust, with RFC 3339 timestamps406 serde_json::to_string(&checks)?.into(),
407 a.actor.id.as_str().into(),
408 a.actor.username.as_str().into(),
Issues and pull requests replace intents and attempts409 timestamp.as_str().into(),
410 timestamp.as_str().into(),
Work service in Rust, with RFC 3339 timestamps411 ])?
412 .run()
413 .await?;
Issues and pull requests replace intents and attempts414 let Some(issue) = self.issue(&repo.id, number).await? else {
415 return Ok(no_issue());
Work service in Rust, with RFC 3339 timestamps416 };
Issues and pull requests replace intents and attempts417 self.publish(
418 "issue.opened",
419 &repo.id,
420 &a.actor,
421 IssueEvent {
422 title: Some(issue.title.clone()),
423 ..Self::issue_event(&issue)
Work service in Rust, with RFC 3339 timestamps424 },
Issues and pull requests replace intents and attempts425 )
Work service in Rust, with RFC 3339 timestamps426 .await?;
Issues and pull requests replace intents and attempts427 Ok(Outcome::Ok(issue))
Work service in Rust, with RFC 3339 timestamps428 }
429
Issues and pull requests replace intents and attempts430 async fn list_issues(&self, a: ListIssuesArgs) -> Result<Outcome<Vec<Issue>>> {
431 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
432 let state = state_name(a.state).map_or(JsValue::NULL, JsValue::from);
433 let label = a
434 .label
435 .map(|label| label.trim().to_lowercase())
436 .filter(|label| !label.is_empty());
Work service in Rust, with RFC 3339 timestamps437 let rows = self
438 .db
439 .prepare(format!(
Issues and pull requests replace intents and attempts440 "SELECT {ISSUE_COLUMNS} FROM issues
441 WHERE repo_id = ? AND (? IS NULL OR state = ?)
442 AND (? IS NULL OR EXISTS
443 (SELECT 1 FROM json_each(issues.labels) WHERE json_each.value = ?))
444 ORDER BY number DESC LIMIT ?"
Work service in Rust, with RFC 3339 timestamps445 ))
Issues and pull requests replace intents and attempts446 .bind(&[
447 repo.id.into(),
448 state.clone(),
449 state,
450 optional(&label),
451 optional(&label),
452 LIST_PAGE.into(),
453 ])?
Work service in Rust, with RFC 3339 timestamps454 .all()
455 .await?
Issues and pull requests replace intents and attempts456 .results::<IssueRow>()?;
457 Ok(Outcome::Ok(rows.into_iter().map(Issue::from).collect()))
Work service in Rust, with RFC 3339 timestamps458 }
459
Issues and pull requests replace intents and attempts460 async fn get_issue(&self, a: ViewArgs) -> Result<Outcome<IssueDetail>> {
461 let (repo, issue) = check!(self.issue_at(&a.repo, a.number, &a.viewer).await?);
462 let pulls = self
Work service in Rust, with RFC 3339 timestamps463 .db
Issues and pull requests replace intents and attempts464 .prepare("SELECT * FROM pulls WHERE issue_id = ? ORDER BY number")
465 .bind(&[issue.id.as_str().into()])?
Work service in Rust, with RFC 3339 timestamps466 .all()
467 .await?
Issues and pull requests replace intents and attempts468 .results::<PullRow>()?;
469 Ok(Outcome::Ok(IssueDetail {
470 comments: self.comments(&repo.id, issue.number).await?,
471 pulls: pulls.into_iter().map(Pull::from).collect(),
472 issue,
Work service in Rust, with RFC 3339 timestamps473 }))
474 }
475
Issues and pull requests replace intents and attempts476 /// The issue, if `actor` wrote it or belongs to the repository's workspace.
477 async fn manageable_issue(
478 &self,
479 actor: &User,
480 path: &RepoPath,
481 number: u32,
482 ) -> Result<Outcome<Issue>> {
483 let (repo, issue) = check!(self.issue_at(path, number, &Some(actor.clone())).await?);
484 if issue.author.id != actor.id && !actor.is_member(&repo.namespace) {
Work service in Rust, with RFC 3339 timestamps485 return Ok(Outcome::fail(
486 FailureCode::Forbidden,
Issues and pull requests replace intents and attempts487 "Only the author or a member of the workspace can change an issue.",
Work service in Rust, with RFC 3339 timestamps488 ));
489 }
Issues and pull requests replace intents and attempts490 Ok(Outcome::Ok(issue))
491 }
492
493 async fn update_issue(&self, a: UpdateIssueArgs) -> Result<Outcome<Issue>> {
494 let issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
495 let title = match a.title.as_deref().map(valid_title) {
496 Some(Err(message)) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
497 Some(Ok(title)) => Some(title.to_owned()),
498 None => None,
499 };
500 let labels = match a.labels.as_deref().map(normalize_labels) {
501 Some(None) => {
502 return Ok(Outcome::fail(
503 FailureCode::Invalid,
504 "An issue can have up to 10 labels of up to 40 characters each.",
505 ));
506 }
507 Some(Some(labels)) => Some(serde_json::to_string(&labels)?),
508 None => None,
509 };
Agents as a team: lifecycle, merge queue, billing and a new shell510 let assignees = match a.assignees {
511 Some(names) => Some(check!(self.valid_assignees(names).await?)),
512 None => None,
513 };
514 let assigned = assignees.as_ref().map(serde_json::to_string).transpose()?;
Issues and pull requests replace intents and attempts515 let body = a.body.map(|body| body.trim().to_owned());
516 self.db
517 .prepare(
518 "UPDATE issues
519 SET title = COALESCE(?, title), body = COALESCE(?, body),
Agents as a team: lifecycle, merge queue, billing and a new shell520 labels = COALESCE(?, labels), assignees = COALESCE(?, assignees),
521 updated_at = ?
Issues and pull requests replace intents and attempts522 WHERE id = ?",
523 )
524 .bind(&[
525 optional(&title),
526 optional(&body),
527 optional(&labels),
Agents as a team: lifecycle, merge queue, billing and a new shell528 optional(&assigned),
Issues and pull requests replace intents and attempts529 rfc3339(now_ms()).into(),
530 issue.id.as_str().into(),
531 ])?
532 .run()
533 .await?;
Agents as a team: lifecycle, merge queue, billing and a new shell534 let before = issue.assignees.clone();
Issues and pull requests replace intents and attempts535 let Some(issue) = self.issue(&issue.repo_id, issue.number).await? else {
536 return Ok(no_issue());
537 };
538 self.publish(
539 "issue.updated",
540 &issue.repo_id,
541 &a.actor,
542 Self::issue_event(&issue),
543 )
544 .await?;
Agents as a team: lifecycle, merge queue, billing and a new shell545 if let Some(assignees) = assignees {
546 self.note_changes(
547 &issue.repo_id,
548 issue.number,
549 &a.actor,
550 &before,
551 &assignees,
552 ("assigned", "unassigned"),
553 )
554 .await?;
555 self.publish(
556 "issue.assigned",
557 &issue.repo_id,
558 &a.actor,
559 IssueEvent {
560 assignees: Some(assignees),
561 ..Self::issue_event(&issue)
562 },
563 )
564 .await?;
565 }
Issues and pull requests replace intents and attempts566 Ok(Outcome::Ok(issue))
567 }
568
Agents as a team: lifecycle, merge queue, billing and a new shell569 /// Usernames as given, tidied, if each names an account.
570 async fn valid_assignees(&self, names: Vec<String>) -> Result<Outcome<Vec<String>>> {
571 let mut assignees: Vec<String> = Vec::new();
572 for name in names {
573 let name = name.trim().trim_start_matches('@').to_lowercase();
574 if name.is_empty() || assignees.contains(&name) {
575 continue;
576 }
577 if assignees.len() == MAX_ASSIGNEES {
578 return Ok(Outcome::fail(
579 FailureCode::Invalid,
580 format!("An issue can be assigned to at most {MAX_ASSIGNEES} people."),
581 ));
582 }
583 let account: Viewer = g1t_kit::call(
584 &self.identity,
585 "user_by_username",
586 &UsernameArgs {
587 username: name.clone(),
588 },
589 )
590 .await?;
591 if account.is_none() {
592 return Ok(Outcome::fail(
593 FailureCode::Invalid,
594 format!("There is no account named {name}."),
595 ));
596 }
597 assignees.push(name);
598 }
599 Ok(Outcome::Ok(assignees))
600 }
601
602 /// Open issues assigned to the viewer, in every repository. Callers
603 /// show only those in repositories the viewer can still see.
604 async fn list_assigned_issues(&self, a: ViewerArgs) -> Result<Vec<Issue>> {
605 let Some(viewer) = a.viewer else {
606 return Ok(Vec::new());
607 };
608 let rows = self
609 .db
610 .prepare(format!(
611 "SELECT {ISSUE_COLUMNS} FROM issues
612 WHERE state = 'open' AND EXISTS (
613 SELECT 1 FROM json_each(issues.assignees) WHERE json_each.value = ?)
614 ORDER BY updated_at DESC LIMIT 50"
615 ))
616 .bind(&[viewer.username.into()])?
617 .all()
618 .await?
619 .results::<IssueRow>()?;
620 Ok(rows.into_iter().map(Issue::from).collect())
621 }
622
Issues and pull requests replace intents and attempts623 async fn close_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
624 let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
625 if issue.state == State::Closed {
Work service in Rust, with RFC 3339 timestamps626 return Ok(Outcome::fail(
627 FailureCode::Conflict,
Issues and pull requests replace intents and attempts628 "This issue is already closed.",
Work service in Rust, with RFC 3339 timestamps629 ));
630 }
Issues and pull requests replace intents and attempts631 let reason = a.reason.unwrap_or(IssueReason::Completed);
632 let now = rfc3339(now_ms());
Work service in Rust, with RFC 3339 timestamps633 self.db
Issues and pull requests replace intents and attempts634 .prepare(
635 "UPDATE issues SET state = 'closed', reason = ?, closed_at = ?, updated_at = ?
636 WHERE id = ?",
637 )
638 .bind(&[
639 reason.as_str().into(),
640 now.as_str().into(),
641 now.as_str().into(),
642 issue.id.as_str().into(),
643 ])?
Work service in Rust, with RFC 3339 timestamps644 .run()
645 .await?;
Issues and pull requests replace intents and attempts646 self.publish(
647 "issue.closed",
648 &issue.repo_id,
649 &a.actor,
650 IssueEvent {
651 reason: Some(reason.as_str()),
652 ..Self::issue_event(&issue)
Work service in Rust, with RFC 3339 timestamps653 },
Issues and pull requests replace intents and attempts654 )
Work service in Rust, with RFC 3339 timestamps655 .await?;
Agents as a team: lifecycle, merge queue, billing and a new shell656 self.note(
657 &issue.repo_id,
658 issue.number,
659 (&a.actor.id, &a.actor.username),
660 match reason {
661 IssueReason::Completed => "closed this as completed",
662 IssueReason::NotPlanned => "closed this as not planned",
663 },
664 )
665 .await?;
Issues and pull requests replace intents and attempts666 issue.state = State::Closed;
667 issue.reason = Some(reason);
668 issue.closed_at = Some(now.clone());
669 issue.updated_at = now;
670 Ok(Outcome::Ok(issue))
671 }
672
673 async fn reopen_issue(&self, a: IssueActionArgs) -> Result<Outcome<Issue>> {
674 let mut issue = check!(self.manageable_issue(&a.actor, &a.repo, a.number).await?);
675 if issue.state == State::Open {
676 return Ok(Outcome::fail(
677 FailureCode::Conflict,
678 "This issue is already open.",
679 ));
680 }
681 let now = rfc3339(now_ms());
682 self.db
683 .prepare(
684 "UPDATE issues
685 SET state = 'open', reason = NULL, resolved_by = NULL, closed_at = NULL,
686 updated_at = ?
687 WHERE id = ?",
688 )
689 .bind(&[now.as_str().into(), issue.id.as_str().into()])?
690 .run()
691 .await?;
692 self.publish(
693 "issue.reopened",
694 &issue.repo_id,
695 &a.actor,
696 Self::issue_event(&issue),
697 )
698 .await?;
Agents as a team: lifecycle, merge queue, billing and a new shell699 self.note(
700 &issue.repo_id,
701 issue.number,
702 (&a.actor.id, &a.actor.username),
703 "reopened this",
704 )
705 .await?;
Issues and pull requests replace intents and attempts706 issue.state = State::Open;
707 issue.reason = None;
708 issue.resolved_by = None;
709 issue.closed_at = None;
710 issue.updated_at = now;
711 Ok(Outcome::Ok(issue))
Work service in Rust, with RFC 3339 timestamps712 }
713
Issues and pull requests replace intents and attempts714 /// The default labels, then every other label in use on the repository.
715 async fn list_labels(&self, a: ViewArgs) -> Result<Outcome<Vec<String>>> {
716 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
717 let used = self
718 .db
719 .prepare(
720 "SELECT DISTINCT json_each.value AS value
721 FROM issues, json_each(issues.labels)
722 WHERE issues.repo_id = ? ORDER BY 1 LIMIT 200",
723 )
724 .bind(&[repo.id.into()])?
725 .all()
726 .await?
727 .results::<ValueRow>()?;
728 let mut labels: Vec<String> = DEFAULT_LABELS.iter().map(|label| (*label).into()).collect();
729 for row in used {
730 if !labels.contains(&row.value) {
731 labels.push(row.value);
732 }
733 }
734 Ok(Outcome::Ok(labels))
735 }
736
737 async fn counts(&self, a: ViewArgs) -> Result<Outcome<Counts>> {
738 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
739 let counts = self
740 .db
741 .prepare(
742 "SELECT
743 (SELECT count(*) FROM issues WHERE repo_id = ? AND state = 'open') AS issues,
744 (SELECT count(*) FROM pulls
745 WHERE repo_id = ? AND status IN ('draft', 'open')) AS pulls",
746 )
747 .bind(&[repo.id.as_str().into(), repo.id.as_str().into()])?
748 .first::<Counts>(None)
749 .await?;
750 Ok(Outcome::Ok(counts.unwrap_or(Counts {
751 issues: 0,
752 pulls: 0,
753 })))
754 }
755
756 // --- Comments ----------------------------------------------------------
757
758 async fn add_comment(&self, a: AddCommentArgs) -> Result<Outcome<Comment>> {
Work service in Rust, with RFC 3339 timestamps759 if !a.actor.verified {
760 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
761 }
Issues and pull requests replace intents and attempts762 let body = a.body.trim();
Acceptance checks in sandboxes, line comments and review verdicts763 // An approval speaks for itself; anything else has to say something.
764 if body.is_empty() && a.verdict != Some(Verdict::Approve) {
Issues and pull requests replace intents and attempts765 return Ok(Outcome::fail(
766 FailureCode::Invalid,
767 "A comment cannot be empty.",
768 ));
769 }
Acceptance checks in sandboxes, line comments and review verdicts770 let path = a
771 .path
772 .as_deref()
773 .map(str::trim)
774 .filter(|path| !path.is_empty());
775 let line = a.line.filter(|line| *line > 0 && path.is_some());
Issues and pull requests replace intents and attempts776 if body.chars().count() > MAX_ENTRY_CHARS {
777 return Ok(Outcome::fail(
778 FailureCode::Invalid,
779 "That comment is too long.",
780 ));
781 }
782 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
783 // The number names an issue or a pull request, never both.
Agents as a team: lifecycle, merge queue, billing and a new shell784 let mut pull_id = None;
Issues and pull requests replace intents and attempts785 let table = if self.issue(&repo.id, a.number).await?.is_some() {
Acceptance checks in sandboxes, line comments and review verdicts786 if path.is_some() || a.verdict.is_some() {
787 return Ok(Outcome::fail(
788 FailureCode::Invalid,
789 "Only a pull request can be reviewed or commented on by line.",
790 ));
791 }
Issues and pull requests replace intents and attempts792 "issues"
Acceptance checks in sandboxes, line comments and review verdicts793 } else if let Some(pull) = self.pull(&repo.id, a.number).await? {
794 if a.verdict.is_some() && pull.author.id == a.actor.id {
795 return Ok(Outcome::fail(
796 FailureCode::Forbidden,
797 "You cannot approve or request changes on your own pull request.",
798 ));
799 }
Agents as a team: lifecycle, merge queue, billing and a new shell800 pull_id = Some(pull.id.clone());
Issues and pull requests replace intents and attempts801 "pulls"
802 } else {
Work service in Rust, with RFC 3339 timestamps803 return Ok(Outcome::fail(
Issues and pull requests replace intents and attempts804 FailureCode::NotFound,
805 "No issue or pull request has that number.",
Work service in Rust, with RFC 3339 timestamps806 ));
Issues and pull requests replace intents and attempts807 };
808
809 let now = now_ms();
810 let comment = Comment {
Agents as a team: lifecycle, merge queue, billing and a new shell811 kind: CommentKind::Comment,
Issues and pull requests replace intents and attempts812 id: new_id("cmt", now),
813 author: a.actor.clone(),
814 body: body.to_owned(),
Acceptance checks in sandboxes, line comments and review verdicts815 path: path.map(str::to_owned),
816 line,
817 verdict: a.verdict,
Issues and pull requests replace intents and attempts818 created_at: rfc3339(now),
819 };
820 self.db
821 .batch(vec![
822 self.db
823 .prepare(
824 "INSERT INTO comments
Acceptance checks in sandboxes, line comments and review verdicts825 (id, repo_id, number, author_id, author_name, body, path, line,
826 verdict, created_at)
827 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Issues and pull requests replace intents and attempts828 )
829 .bind(&[
830 comment.id.as_str().into(),
831 repo.id.as_str().into(),
832 a.number.into(),
833 a.actor.id.as_str().into(),
834 a.actor.username.as_str().into(),
835 body.into(),
Acceptance checks in sandboxes, line comments and review verdicts836 optional(&comment.path),
837 optional_number(line),
838 a.verdict
839 .map_or(JsValue::NULL, |verdict| verdict.as_str().into()),
Issues and pull requests replace intents and attempts840 comment.created_at.as_str().into(),
841 ])?,
842 self.db
843 .prepare(format!(
844 "UPDATE {table} SET updated_at = ? WHERE repo_id = ? AND number = ?"
845 ))
846 .bind(&[
847 comment.created_at.as_str().into(),
848 repo.id.as_str().into(),
849 a.number.into(),
850 ])?,
851 ])
852 .await?;
853 self.publish(
854 "comment.created",
855 &repo.id,
856 &a.actor,
857 CommentCreated {
858 comment_id: comment.id.clone(),
859 repo_id: repo.id.clone(),
860 number: a.number,
Agents as a team: lifecycle, merge queue, billing and a new shell861 pull_id,
862 verdict: a.verdict,
Issues and pull requests replace intents and attempts863 },
864 )
865 .await?;
866 Ok(Outcome::Ok(comment))
867 }
868
869 // --- Pull requests -----------------------------------------------------
870
871 async fn open_pull(&self, a: OpenPullArgs) -> Result<Outcome<Pull>> {
872 if !a.actor.verified {
873 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
Work service in Rust, with RFC 3339 timestamps874 }
Issues and pull requests replace intents and attempts875 let repo = check!(self.repo(&a.repo, &Some(a.actor.clone())).await?);
876 let issue = match a.issue {
877 Some(number) => match self.issue(&repo.id, number).await? {
878 Some(issue) if issue.state == State::Open => Some(issue),
879 Some(_) => {
880 return Ok(Outcome::fail(
881 FailureCode::Conflict,
882 "This issue is closed.",
883 ));
884 }
885 None => return Ok(no_issue()),
886 },
887 None => None,
888 };
889 // A pull request for an issue takes the issue's title unless given one.
890 let title = match (a.title.trim(), &issue) {
891 ("", Some(issue)) => issue.title.clone(),
892 (title, _) => match valid_title(title) {
893 Ok(title) => title.to_owned(),
894 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
895 },
896 };
Work service in Rust, with RFC 3339 timestamps897 let agent = match a.agent.trim() {
898 "" => "agent",
899 agent => agent,
900 };
901 let runtime = match a.runtime {
Issues and pull requests replace intents and attempts902 Runtime::Hosted => "hosted",
903 Runtime::External => "external",
Work service in Rust, with RFC 3339 timestamps904 };
905
906 let now = now_ms();
Issues and pull requests replace intents and attempts907 let id = new_id("pr", now);
Pull requests from branches908 let branch = a
909 .branch
910 .as_deref()
911 .map(str::trim)
912 .filter(|branch| !branch.is_empty());
913 // The change is on a branch already pushed to the repository, or
914 // will be made in a fork created for this pull request.
915 let (fork, head) = match branch {
916 Some(branch) => {
917 if branch == repo.default_branch {
918 return Ok(Outcome::fail(
919 FailureCode::Invalid,
920 format!("Choose a branch other than {branch}."),
921 ));
922 }
923 let head: Option<String> = g1t_kit::call(
924 &self.repos,
925 "head",
926 &HeadArgs {
927 repo_id: repo.id.clone(),
928 branch: branch.to_owned(),
929 },
930 )
931 .await?;
932 let Some(head) = head else {
933 return Ok(Outcome::fail(
934 FailureCode::NotFound,
935 format!("There is no branch named {branch}. Push it first."),
936 ));
937 };
938 let existing = self
939 .db
940 .prepare(
941 "SELECT number AS n FROM pulls
942 WHERE repo_id = ? AND source_branch = ? AND status IN ('draft', 'open')",
943 )
944 .bind(&[repo.id.as_str().into(), branch.into()])?
945 .first::<NumberRow>(None)
946 .await?;
947 if let Some(existing) = existing {
948 return Ok(Outcome::fail(
949 FailureCode::Conflict,
950 format!("Pull request #{} is already open for {branch}.", existing.n),
951 ));
952 }
953 (None, Some(head))
954 }
955 None => {
956 let fork: Outcome<Repo> = g1t_kit::call(
957 &self.repos,
958 "fork_for_pull",
959 &ForkArgs {
960 source_id: repo.id.clone(),
961 pull_id: id.clone(),
962 actor: a.actor.clone(),
963 },
964 )
965 .await?;
966 (Some(check!(fork)), None)
967 }
968 };
969 // A branch already holds the work, so its pull request is ready for
970 // review from the start; one with a fork starts as a draft.
971 let status = if branch.is_some() { "open" } else { "draft" };
972 let body = Some(a.body.trim().to_owned()).filter(|body| !body.is_empty());
Work service in Rust, with RFC 3339 timestamps973
Issues and pull requests replace intents and attempts974 let number = self.next_number(&repo.id).await?;
Work service in Rust, with RFC 3339 timestamps975 let timestamp = rfc3339(now);
976 self.db
977 .prepare(
Issues and pull requests replace intents and attempts978 "INSERT INTO pulls
Pull requests from branches979 (id, repo_id, number, issue_id, issue_number, title, body, agent, runtime,
980 status, fork_repo_id, fork_namespace, fork_name, source_branch, head_commit,
981 author_id, author_name, created_at, updated_at)
982 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Work service in Rust, with RFC 3339 timestamps983 )
984 .bind(&[
985 id.as_str().into(),
Issues and pull requests replace intents and attempts986 repo.id.as_str().into(),
987 number.into(),
988 optional(&issue.as_ref().map(|issue| issue.id.clone())),
989 optional_number(issue.as_ref().map(|issue| issue.number)),
990 title.into(),
Pull requests from branches991 optional(&body),
Work service in Rust, with RFC 3339 timestamps992 agent.into(),
993 runtime.into(),
Pull requests from branches994 status.into(),
995 optional(&fork.as_ref().map(|fork| fork.id.clone())),
996 optional(&fork.as_ref().map(|fork| fork.namespace.clone())),
997 optional(&fork.as_ref().map(|fork| fork.name.clone())),
998 optional(&branch.map(str::to_owned)),
999 optional(&head),
Work service in Rust, with RFC 3339 timestamps1000 a.actor.id.as_str().into(),
1001 a.actor.username.as_str().into(),
1002 timestamp.as_str().into(),
1003 timestamp.as_str().into(),
1004 ])?
1005 .run()
1006 .await?;
Issues and pull requests replace intents and attempts1007 let Some(pull) = self.pull(&repo.id, number).await? else {
1008 return Ok(no_pull());
Work service in Rust, with RFC 3339 timestamps1009 };
Agents as a team: lifecycle, merge queue, billing and a new shell1010 self.manage(&pull).await?;
1011 // Someone is on it now, so it is no longer waiting for an agent.
1012 if let Some(issue) = pull.issue {
1013 self.db
1014 .prepare("UPDATE issues SET queued_by = NULL WHERE repo_id = ? AND number = ?")
1015 .bind(&[repo.id.as_str().into(), issue.into()])?
1016 .run()
1017 .await?;
1018 }
1019 if let Some(issue) = pull.issue {
1020 let text = if lifecycle::made_by_g1t(&pull) {
1021 format!("assigned this to g1t-agent, which opened #{}", pull.number)
1022 } else {
1023 format!("opened #{} for this", pull.number)
1024 };
1025 self.note(&repo.id, issue, (&a.actor.id, &a.actor.username), &text)
1026 .await?;
1027 }
Issues and pull requests replace intents and attempts1028 self.publish(
1029 "pull.opened",
1030 &repo.id,
1031 &a.actor,
1032 PullEvent {
1033 agent: Some(pull.agent.clone()),
1034 ..Self::pull_event(&pull)
Work service in Rust, with RFC 3339 timestamps1035 },
Issues and pull requests replace intents and attempts1036 )
Work service in Rust, with RFC 3339 timestamps1037 .await?;
Issues and pull requests replace intents and attempts1038 Ok(Outcome::Ok(pull))
Work service in Rust, with RFC 3339 timestamps1039 }
1040
Issues and pull requests replace intents and attempts1041 async fn list_pulls(&self, a: ListPullsArgs) -> Result<Outcome<Vec<Pull>>> {
1042 let repo = check!(self.repo(&a.repo, &a.viewer).await?);
1043 let filter = match a.state {
1044 Some(State::Open) => "AND status IN ('draft', 'open')",
1045 Some(State::Closed) => "AND status IN ('merged', 'closed')",
1046 None => "",
Work service in Rust, with RFC 3339 timestamps1047 };
Issues and pull requests replace intents and attempts1048 let rows = self
1049 .db
1050 .prepare(format!(
1051 "SELECT * FROM pulls WHERE repo_id = ? {filter} ORDER BY number DESC LIMIT ?"
1052 ))
1053 .bind(&[repo.id.into(), LIST_PAGE.into()])?
1054 .all()
1055 .await?
1056 .results::<PullRow>()?;
1057 Ok(Outcome::Ok(rows.into_iter().map(Pull::from).collect()))
1058 }
1059
1060 async fn get_pull(&self, a: ViewArgs) -> Result<Outcome<PullDetail>> {
1061 let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
1062 let issue = match pull.issue {
1063 Some(number) => self.issue(&repo.id, number).await?,
1064 None => None,
Work service in Rust, with RFC 3339 timestamps1065 };
Agents as a team: lifecycle, merge queue, billing and a new shell1066 let mut pull = pull;
1067 // Worked out on each push; this covers a pull request from before
1068 // that was recorded.
1069 if pull.files.is_empty() && pull.head_commit.is_some() {
1070 pull.files = self.refresh_files(&pull).await?;
1071 }
1072 // Everything else at once: none of it depends on the rest, and each
1073 // is a round trip of its own.
1074 let standing = async {
1075 let behind = self.is_behind(&repo.id, &pull).await?;
1076 let lifecycle = self
1077 .assess(&pull, &issue, behind)
1078 .await?
1079 .map(|(lifecycle, _)| lifecycle);
1080 Ok::<_, worker::Error>((behind, lifecycle))
1081 };
1082 let (((behind, lifecycle), (landing, stalled), comments), (checks, overlaps, review_pending)) =
1083 try_join(
1084 try_join3(standing, self.landing_state(&pull.id), self.comments(&repo.id, pull.number)),
1085 try_join3(
1086 self.latest_checks(&pull.id),
1087 self.overlaps(&pull),
1088 self.review_pending(&pull.id),
1089 ),
1090 )
1091 .await?;
Issues and pull requests replace intents and attempts1092 Ok(Outcome::Ok(PullDetail {
Agents as a team: lifecycle, merge queue, billing and a new shell1093 comments,
1094 checks,
1095 overlaps,
1096 behind,
1097 review_pending,
1098 lifecycle,
1099 landing,
1100 stalled,
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request1101 messages: self.messages(&pull.id).await?,
GitHub Actions on g1t, part two: running workflows1102 statuses: self.statuses(&repo.id, pull.head_commit.as_deref()).await?,
Issues and pull requests replace intents and attempts1103 issue,
1104 pull,
1105 }))
Work service in Rust, with RFC 3339 timestamps1106 }
1107
Issues and pull requests replace intents and attempts1108 /// The pull request, if it is still active and `actor` opened it or
1109 /// belongs to the repository's workspace.
1110 async fn manageable_pull(
Work service in Rust, with RFC 3339 timestamps1111 &self,
Issues and pull requests replace intents and attempts1112 actor: &User,
1113 path: &RepoPath,
1114 number: u32,
1115 ) -> Result<Outcome<Pull>> {
1116 let (repo, pull) = check!(self.pull_at(path, number, &Some(actor.clone())).await?);
1117 if pull.author.id != actor.id && !actor.is_member(&repo.namespace) {
Work service in Rust, with RFC 3339 timestamps1118 return Ok(Outcome::fail(
Issues and pull requests replace intents and attempts1119 FailureCode::Forbidden,
1120 "Only whoever opened a pull request, or a member of the workspace, can change it.",
1121 ));
1122 }
1123 if !pull.status.is_active() {
1124 return Ok(Outcome::fail(
Work service in Rust, with RFC 3339 timestamps1125 FailureCode::Conflict,
Issues and pull requests replace intents and attempts1126 format!("This pull request is already {}.", pull.status.as_str()),
Work service in Rust, with RFC 3339 timestamps1127 ));
1128 }
Issues and pull requests replace intents and attempts1129 Ok(Outcome::Ok(pull))
1130 }
1131
Agents as a team: lifecycle, merge queue, billing and a new shell1132 async fn update_pull(&self, a: UpdatePullArgs) -> Result<Outcome<Pull>> {
1133 let pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1134 let assignees = match a.assignees {
1135 Some(names) => Some(check!(self.valid_assignees(names).await?)),
1136 None => None,
1137 };
1138 let reviewers = match a.reviewers {
1139 Some(names) => {
1140 // A g1t agent is not an account; everyone else has to be.
1141 let agent = names
1142 .iter()
1143 .any(|name| name.trim().eq_ignore_ascii_case(reviews::AGENT_NAME));
1144 let people = names
1145 .into_iter()
1146 .filter(|name| !name.trim().eq_ignore_ascii_case(reviews::AGENT_NAME))
1147 .collect();
1148 let mut reviewers = check!(self.valid_assignees(people).await?);
1149 reviewers.retain(|name| *name != pull.author.username);
1150 if agent {
1151 reviewers.insert(0, reviews::AGENT_NAME.to_owned());
1152 }
1153 Some(reviewers)
1154 }
1155 None => None,
1156 };
1157 self.db
1158 .prepare(
1159 "UPDATE pulls
1160 SET assignees = COALESCE(?, assignees), reviewers = COALESCE(?, reviewers),
1161 updated_at = ?
1162 WHERE id = ?",
1163 )
1164 .bind(&[
1165 optional(&assignees.as_ref().map(serde_json::to_string).transpose()?),
1166 optional(&reviewers.as_ref().map(serde_json::to_string).transpose()?),
1167 rfc3339(now_ms()).into(),
1168 pull.id.as_str().into(),
1169 ])?
1170 .run()
1171 .await?;
1172 if let Some(assignees) = &assignees {
1173 self.note_changes(
1174 &pull.repo_id,
1175 pull.number,
1176 &a.actor,
1177 &pull.assignees,
1178 assignees,
1179 ("assigned", "unassigned"),
1180 )
1181 .await?;
1182 }
1183 if let Some(reviewers) = &reviewers {
1184 self.note_changes(
1185 &pull.repo_id,
1186 pull.number,
1187 &a.actor,
1188 &pull.reviewers,
1189 reviewers,
1190 (
1191 "requested a review from",
1192 "withdrew the request for a review from",
1193 ),
1194 )
1195 .await?;
1196 }
1197 Ok(match self.pull(&pull.repo_id, pull.number).await? {
1198 Some(pull) => Outcome::Ok(pull),
1199 None => no_pull(),
1200 })
1201 }
1202
Issues and pull requests replace intents and attempts1203 /// Marks a draft ready for review, or updates the description of one
1204 /// that already is.
1205 async fn ready_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1206 let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
Work service in Rust, with RFC 3339 timestamps1207 let summary = Some(a.summary.trim().to_owned()).filter(|summary| !summary.is_empty());
1208 let now = rfc3339(now_ms());
1209 self.db
1210 .prepare(
Issues and pull requests replace intents and attempts1211 "UPDATE pulls SET status = 'open', body = COALESCE(?, body), updated_at = ?
Work service in Rust, with RFC 3339 timestamps1212 WHERE id = ?",
1213 )
1214 .bind(&[
1215 optional(&summary),
1216 now.as_str().into(),
Issues and pull requests replace intents and attempts1217 pull.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps1218 ])?
1219 .run()
1220 .await?;
Issues and pull requests replace intents and attempts1221 if pull.status == PullStatus::Draft {
Workflows run when an agent's pull request is marked ready1222 // The head as it is now: the push that came just before may not
1223 // have reached `head_commit` yet, and workflows run on it.
1224 let commit = self.live_head(&pull).await?.or_else(|| pull.head_commit.clone());
Issues and pull requests replace intents and attempts1225 self.publish(
1226 "pull.ready",
1227 &pull.repo_id,
1228 &a.actor,
Workflows run when an agent's pull request is marked ready1229 PullEvent {
1230 commit,
1231 ..Self::pull_event(&pull)
1232 },
Issues and pull requests replace intents and attempts1233 )
1234 .await?;
1235 }
Agents as a team: lifecycle, merge queue, billing and a new shell1236 if pull.status == PullStatus::Draft {
1237 self.note(
1238 &pull.repo_id,
1239 pull.number,
1240 (&a.actor.id, &a.actor.username),
1241 "marked this ready for review",
1242 )
1243 .await?;
1244 }
Issues and pull requests replace intents and attempts1245 pull.status = PullStatus::Open;
1246 pull.body = summary.or(pull.body);
1247 pull.updated_at = now;
1248 Ok(Outcome::Ok(pull))
1249 }
1250
1251 async fn close_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
1252 let mut pull = check!(self.manageable_pull(&a.actor, &a.repo, a.number).await?);
1253 let now = rfc3339(now_ms());
1254 self.db
1255 .prepare("UPDATE pulls SET status = 'closed', updated_at = ? WHERE id = ?")
1256 .bind(&[now.as_str().into(), pull.id.as_str().into()])?
1257 .run()
1258 .await?;
1259 self.publish(
1260 "pull.closed",
1261 &pull.repo_id,
1262 &a.actor,
1263 Self::pull_event(&pull),
1264 )
Work service in Rust, with RFC 3339 timestamps1265 .await?;
Agents as a team: lifecycle, merge queue, billing and a new shell1266 self.note(
1267 &pull.repo_id,
1268 pull.number,
1269 (&a.actor.id, &a.actor.username),
1270 "closed this",
1271 )
1272 .await?;
1273 // A closed pull request leaves the merge queue.
1274 if self
1275 .leave(&pull.repo_id, &pull, QueueState::Removed, Some("It was closed."))
1276 .await?
1277 {
1278 self.publish_as(
1279 "queue.changed",
1280 &pull.repo_id,
1281 None,
1282 g1t_contracts::events::QueueChanged {
1283 repo_id: pull.repo_id.clone(),
1284 },
1285 )
1286 .await?;
1287 }
Issues and pull requests replace intents and attempts1288 pull.status = PullStatus::Closed;
1289 pull.updated_at = now;
1290 Ok(Outcome::Ok(pull))
Work service in Rust, with RFC 3339 timestamps1291 }
1292
Issues and pull requests replace intents and attempts1293 /// Lands the pull request on the repository's default branch. Unless
1294 /// told to keep it open, that resolves the issue it was for: the issue
1295 /// closes naming this pull request, and the others still in progress
1296 /// for it close as superseded.
1297 async fn merge_pull(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
Work service in Rust, with RFC 3339 timestamps1298 let viewer = Some(a.actor.clone());
Agents as a team: lifecycle, merge queue, billing and a new shell1299 let (repo, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
Issues and pull requests replace intents and attempts1300 match pull.status {
1301 PullStatus::Open => {}
1302 PullStatus::Draft => {
1303 return Ok(Outcome::fail(
1304 FailureCode::Conflict,
1305 "This pull request is still a draft. Mark it ready for review first.",
1306 ));
1307 }
1308 status => {
1309 return Ok(Outcome::fail(
1310 FailureCode::Conflict,
1311 format!("This pull request is already {}.", status.as_str()),
1312 ));
1313 }
Work service in Rust, with RFC 3339 timestamps1314 }
Agents as a team: lifecycle, merge queue, billing and a new shell1315 let settings = self.settings(&repo.id).await?;
1316 // Where the repository does not allow it, asking to ignore the
1317 // checks changes nothing.
1318 if !a.ignore_checks || !settings.allow_ignoring_checks {
Acceptance checks in sandboxes, line comments and review verdicts1319 let waiting = match pull.check_status {
1320 Some(CheckStatus::Queued | CheckStatus::Running) => {
1321 Some("The acceptance checks are still running.")
1322 }
1323 Some(CheckStatus::Failed) => Some("The acceptance checks did not pass."),
1324 Some(CheckStatus::Errored) => Some("The acceptance checks could not be run."),
1325 Some(CheckStatus::Passed) | None => None,
1326 };
GitHub Actions on g1t, part two: running workflows1327 // Workflows run on its head count as checks too.
1328 let workflows = statuses::WorkflowFacts::of(&self.statuses(&repo.id, pull.head_commit.as_deref()).await?).refusal();
1329 let waiting = waiting.map(str::to_owned).or(workflows);
Acceptance checks in sandboxes, line comments and review verdicts1330 if let Some(reason) = waiting {
Agents as a team: lifecycle, merge queue, billing and a new shell1331 let remedy = if settings.allow_ignoring_checks {
1332 "Wait or fix them, or merge anyway by ignoring the checks."
1333 } else {
1334 "This repository only merges pull requests whose checks pass."
1335 };
Acceptance checks in sandboxes, line comments and review verdicts1336 return Ok(Outcome::fail(
1337 FailureCode::Conflict,
Agents as a team: lifecycle, merge queue, billing and a new shell1338 format!("{reason} {remedy}"),
Acceptance checks in sandboxes, line comments and review verdicts1339 ));
1340 }
1341 }
Agents as a team: lifecycle, merge queue, billing and a new shell1342 if let Some(missing) = self.approvals_gap(&settings, &pull).await? {
1343 return Ok(Outcome::fail(FailureCode::Conflict, missing));
1344 }
Work service in Rust, with RFC 3339 timestamps1345
Agents as a team: lifecycle, merge queue, billing and a new shell1346 // A repository that merges through a queue: it joins the queue, and
1347 // lands once its state together with everything ahead has passed.
1348 if settings.merge_queue {
1349 if !a.actor.verified {
1350 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1351 }
1352 if !a.actor.is_member(&repo.namespace) {
1353 return Ok(Outcome::fail(
1354 FailureCode::Forbidden,
1355 "Only members of the repository's workspace can merge a pull request.",
1356 ));
1357 }
1358 return self.enqueue(&repo, &pull, &a.actor, a.keep_issue_open).await;
1359 }
1360
1361 // The default branch has moved under it. Unless the repository
1362 // insists on that being dealt with first, bring it up to date and
1363 // land it when that is done.
1364 if self.is_behind(&repo.id, &pull).await? {
1365 if settings.require_up_to_date {
1366 return Ok(Outcome::fail(
1367 FailureCode::Conflict,
1368 format!(
1369 "{} has moved since this pull request was made, and this repository requires pull requests to be up to date before they merge. Catch up with {0} first.",
1370 repo.default_branch
1371 ),
1372 ));
1373 }
1374 if !a.actor.verified {
1375 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
1376 }
1377 if !a.actor.is_member(&repo.namespace) {
1378 return Ok(Outcome::fail(
1379 FailureCode::Forbidden,
1380 "Only members of the repository's workspace can merge a pull request.",
1381 ));
1382 }
1383 self.request_landing(&pull, &a.actor, a.keep_issue_open)
1384 .await?;
1385 return Ok(Outcome::Ok(pull));
1386 }
1387
Issues and pull requests replace intents and attempts1388 // Whether the actor may write to the repository is decided by repos.
Work service in Rust, with RFC 3339 timestamps1389 let landed: Outcome<Landed> = g1t_kit::call(
1390 &self.repos,
1391 "land",
1392 &LandArgs {
Pull requests from branches1393 // A pull request from a branch lands from the repository itself.
1394 source_id: pull.fork_repo_id.clone().unwrap_or_else(|| repo.id.clone()),
1395 branch: pull.branch.clone(),
Work service in Rust, with RFC 3339 timestamps1396 actor: a.actor.clone(),
1397 },
1398 )
1399 .await?;
Issues and pull requests replace intents and attempts1400 let landed = check!(landed);
Agents as a team: lifecycle, merge queue, billing and a new shell1401 Ok(Outcome::Ok(
1402 self.record_merge(&repo, pull, &a.actor, a.keep_issue_open, landed)
1403 .await?,
1404 ))
1405 }
Work service in Rust, with RFC 3339 timestamps1406
Agents as a team: lifecycle, merge queue, billing and a new shell1407 /// Records a pull request as merged once the default branch holds it:
1408 /// closes its issue, supersedes the others for it, and says so.
1409 pub(crate) async fn record_merge(
1410 &self,
1411 repo: &Repo,
1412 mut pull: Pull,
1413 actor: &User,
1414 keep_issue_open: bool,
1415 landed: Landed,
1416 ) -> Result<Pull> {
1417 let issue = match pull.issue {
1418 Some(number) if !keep_issue_open => self
1419 .issue(&repo.id, number)
1420 .await?
1421 .filter(|issue| issue.state == State::Open),
1422 _ => None,
1423 };
Work service in Rust, with RFC 3339 timestamps1424 let now = rfc3339(now_ms());
Issues and pull requests replace intents and attempts1425 let mut statements = vec![
1426 self.db
1427 .prepare(
1428 "UPDATE pulls
1429 SET status = 'merged', head_commit = ?, merge_base = ?, merged_by = ?,
1430 merged_at = ?, updated_at = ?
1431 WHERE id = ?",
1432 )
1433 .bind(&[
1434 landed.commit.as_str().into(),
1435 optional(&landed.previous),
Agents as a team: lifecycle, merge queue, billing and a new shell1436 actor.username.as_str().into(),
Issues and pull requests replace intents and attempts1437 now.as_str().into(),
1438 now.as_str().into(),
1439 pull.id.as_str().into(),
1440 ])?,
1441 ];
1442 if let Some(issue) = &issue {
1443 statements.push(
Work service in Rust, with RFC 3339 timestamps1444 self.db
1445 .prepare(
Issues and pull requests replace intents and attempts1446 "UPDATE issues
1447 SET state = 'closed', reason = 'completed', resolved_by = ?,
1448 closed_at = ?, updated_at = ?
Work service in Rust, with RFC 3339 timestamps1449 WHERE id = ?",
1450 )
1451 .bind(&[
Issues and pull requests replace intents and attempts1452 pull.number.into(),
1453 now.as_str().into(),
Work service in Rust, with RFC 3339 timestamps1454 now.as_str().into(),
Issues and pull requests replace intents and attempts1455 issue.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps1456 ])?,
Issues and pull requests replace intents and attempts1457 );
1458 statements.push(
Work service in Rust, with RFC 3339 timestamps1459 self.db
Issues and pull requests replace intents and attempts1460 .prepare(
1461 "UPDATE pulls SET status = 'closed', superseded_by = ?, updated_at = ?
1462 WHERE issue_id = ? AND id != ? AND status IN ('draft', 'open')",
1463 )
1464 .bind(&[
1465 pull.number.into(),
1466 now.as_str().into(),
1467 issue.id.as_str().into(),
1468 pull.id.as_str().into(),
1469 ])?,
1470 );
1471 }
1472 self.db.batch(statements).await?;
1473
1474 self.publish(
1475 "pull.merged",
1476 &repo.id,
Agents as a team: lifecycle, merge queue, billing and a new shell1477 actor,
Issues and pull requests replace intents and attempts1478 PullEvent {
Work service in Rust, with RFC 3339 timestamps1479 commit: Some(landed.commit.clone()),
Issues and pull requests replace intents and attempts1480 ..Self::pull_event(&pull)
Work service in Rust, with RFC 3339 timestamps1481 },
Issues and pull requests replace intents and attempts1482 )
Work service in Rust, with RFC 3339 timestamps1483 .await?;
Issues and pull requests replace intents and attempts1484 if let Some(issue) = &issue {
1485 self.publish(
1486 "issue.closed",
1487 &repo.id,
Agents as a team: lifecycle, merge queue, billing and a new shell1488 actor,
Issues and pull requests replace intents and attempts1489 IssueEvent {
1490 reason: Some(IssueReason::Completed.as_str()),
1491 resolved_by: Some(pull.number),
1492 ..Self::issue_event(issue)
1493 },
1494 )
1495 .await?;
1496 }
Work service in Rust, with RFC 3339 timestamps1497
Agents as a team: lifecycle, merge queue, billing and a new shell1498 let who = (actor.id.as_str(), actor.username.as_str());
1499 self.note(&repo.id, pull.number, who, "merged this").await?;
1500 if let Some(issue) = &issue {
1501 self.note(
1502 &repo.id,
1503 issue.number,
1504 who,
1505 &format!("closed this by merging #{}", pull.number),
1506 )
1507 .await?;
1508 }
Issues and pull requests replace intents and attempts1509 pull.status = PullStatus::Merged;
Agents as a team: lifecycle, merge queue, billing and a new shell1510 pull.head_commit = Some(landed.commit.clone());
Issues and pull requests replace intents and attempts1511 pull.merge_base = landed.previous;
Agents as a team: lifecycle, merge queue, billing and a new shell1512 pull.merged_by = Some(actor.username.clone());
Issues and pull requests replace intents and attempts1513 pull.merged_at = Some(now.clone());
1514 pull.updated_at = now;
Agents as a team: lifecycle, merge queue, billing and a new shell1515 Ok(pull)
Work service in Rust, with RFC 3339 timestamps1516 }
1517
Issues and pull requests replace intents and attempts1518 async fn list_active_pulls(&self, a: ViewerArgs) -> Result<Vec<ActivePull>> {
Work service in Rust, with RFC 3339 timestamps1519 let Some(viewer) = a.viewer else {
1520 return Ok(Vec::new());
1521 };
Agents as a team: lifecycle, merge queue, billing and a new shell1522 let found = self
Work service in Rust, with RFC 3339 timestamps1523 .db
1524 .prepare(
Issues and pull requests replace intents and attempts1525 "SELECT * FROM pulls
1526 WHERE author_id = ? AND status IN ('draft', 'open')
Work service in Rust, with RFC 3339 timestamps1527 ORDER BY updated_at DESC LIMIT 50",
1528 )
1529 .bind(&[viewer.id.into()])?
1530 .all()
Agents as a team: lifecycle, merge queue, billing and a new shell1531 .await?;
1532 let snapshots = found.results::<Snapshot>()?;
1533 let pulls: Vec<Pull> = found.results::<PullRow>()?.into_iter().map(Pull::from).collect();
1534 // Each one at once: its issue, and where it stands. That is the
1535 // remembered assessment when there is one, and worked out otherwise.
1536 try_join_all(pulls.into_iter().zip(snapshots).map(|(pull, snapshot)| async move {
Issues and pull requests replace intents and attempts1537 let issue = match pull.issue {
1538 Some(number) => self.issue(&pull.repo_id, number).await?,
1539 None => None,
1540 };
Agents as a team: lifecycle, merge queue, billing and a new shell1541 // Only a pull request g1t is seeing through has a lifecycle.
1542 let lifecycle = if !lifecycle::made_by_g1t(&pull) || snapshot.managed == 0 {
1543 None
1544 } else if let (Some(stage), Some(detail)) = (snapshot.stage, snapshot.stage_detail) {
1545 Some(Lifecycle {
1546 stage,
1547 detail,
1548 revisions: snapshot.revisions,
1549 })
1550 } else {
1551 let behind = self.is_behind(&pull.repo_id, &pull).await?;
1552 self.assess(&pull, &issue, behind)
1553 .await?
1554 .map(|(lifecycle, _)| lifecycle)
1555 };
1556 Ok::<_, worker::Error>(ActivePull {
1557 pull,
1558 issue,
1559 lifecycle,
1560 })
1561 }))
1562 .await
Work service in Rust, with RFC 3339 timestamps1563 }
1564
Issues and pull requests replace intents and attempts1565 // --- Sessions ----------------------------------------------------------
1566
Work service in Rust, with RFC 3339 timestamps1567 async fn append_session(&self, a: AppendSessionArgs) -> Result<Outcome<Appended>> {
1568 if a.entries.is_empty() {
1569 return Ok(Outcome::Ok(Appended { count: 0 }));
1570 }
1571 if a.entries.len() > MAX_ENTRY_BATCH {
1572 return Ok(Outcome::fail(
1573 FailureCode::Invalid,
1574 format!("Send at most {MAX_ENTRY_BATCH} entries at a time."),
1575 ));
1576 }
Issues and pull requests replace intents and attempts1577 let viewer = Some(a.actor.clone());
1578 let (_, pull) = check!(self.pull_at(&a.repo, a.number, &viewer).await?);
1579 if pull.author.id != a.actor.id {
1580 return Ok(Outcome::fail(
1581 FailureCode::Forbidden,
1582 "Only whoever opened a pull request can record its session.",
1583 ));
1584 }
Work service in Rust, with RFC 3339 timestamps1585
1586 let now = rfc3339(now_ms());
1587 let count = a.entries.len() as u32;
1588 let mut statements = Vec::with_capacity(a.entries.len() + 1);
1589 for entry in a.entries {
1590 let kind = serde_json::to_value(entry.kind)?;
1591 let text: String = entry.text.chars().take(MAX_ENTRY_CHARS).collect();
1592 // Each insert takes the next sequence number itself, so two
1593 // writers appending at once cannot collide.
1594 statements.push(
1595 self.db
1596 .prepare(
Issues and pull requests replace intents and attempts1597 "INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
Work service in Rust, with RFC 3339 timestamps1598 SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
Issues and pull requests replace intents and attempts1599 FROM session_entries WHERE pull_id = ?",
Work service in Rust, with RFC 3339 timestamps1600 )
1601 .bind(&[
Issues and pull requests replace intents and attempts1602 pull.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps1603 kind.as_str().unwrap_or("note").into(),
1604 text.into(),
1605 optional(&entry.tool),
Issues and pull requests replace intents and attempts1606 optional(&entry.commit.or_else(|| pull.head_commit.clone())),
Work service in Rust, with RFC 3339 timestamps1607 now.as_str().into(),
Issues and pull requests replace intents and attempts1608 pull.id.as_str().into(),
Work service in Rust, with RFC 3339 timestamps1609 ])?,
1610 );
1611 }
1612 statements.push(
1613 self.db
Issues and pull requests replace intents and attempts1614 .prepare("UPDATE pulls SET updated_at = ? WHERE id = ?")
1615 .bind(&[now.as_str().into(), pull.id.as_str().into()])?,
Work service in Rust, with RFC 3339 timestamps1616 );
1617 self.db.batch(statements).await?;
Issues and pull requests replace intents and attempts1618 self.publish(
1619 "session.appended",
1620 &pull.repo_id,
1621 &a.actor,
1622 SessionAppended {
1623 pull_id: pull.id.clone(),
1624 repo_id: pull.repo_id.clone(),
1625 number: pull.number,
Work service in Rust, with RFC 3339 timestamps1626 count,
1627 },
Issues and pull requests replace intents and attempts1628 )
Work service in Rust, with RFC 3339 timestamps1629 .await?;
1630 Ok(Outcome::Ok(Appended { count }))
1631 }
1632
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request1633 /// Adds entries to a pull request's session, each taking the next
1634 /// sequence number, without announcing it.
1635 pub(crate) async fn append_entries(&self, pull: &Pull, entries: &[NewSessionEntry]) -> Result<()> {
1636 let now = rfc3339(now_ms());
1637 let mut statements = Vec::with_capacity(entries.len());
1638 for entry in entries {
1639 let kind = serde_json::to_value(entry.kind)?;
1640 let text: String = entry.text.chars().take(MAX_ENTRY_CHARS).collect();
1641 statements.push(
1642 self.db
1643 .prepare(
1644 "INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
1645 SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ?, ?, ?, ?
1646 FROM session_entries WHERE pull_id = ?",
1647 )
1648 .bind(&[
1649 pull.id.as_str().into(),
1650 kind.as_str().unwrap_or("note").into(),
1651 text.into(),
1652 optional(&entry.tool),
1653 optional(&entry.commit.clone().or_else(|| pull.head_commit.clone())),
1654 now.as_str().into(),
1655 pull.id.as_str().into(),
1656 ])?,
1657 );
1658 }
1659 self.db.batch(statements).await?;
1660 Ok(())
1661 }
1662
Issues and pull requests replace intents and attempts1663 async fn read_session(&self, a: ViewArgs) -> Result<Outcome<Vec<SessionEntry>>> {
1664 let (_, pull) = check!(self.pull_at(&a.repo, a.number, &a.viewer).await?);
Work service in Rust, with RFC 3339 timestamps1665 let rows = self
1666 .db
1667 .prepare(
1668 "SELECT seq, kind, text, tool, \"commit\", at FROM session_entries
Issues and pull requests replace intents and attempts1669 WHERE pull_id = ? AND seq > ? ORDER BY seq LIMIT ?",
Work service in Rust, with RFC 3339 timestamps1670 )
Issues and pull requests replace intents and attempts1671 .bind(&[pull.id.into(), a.after_seq.into(), SESSION_PAGE.into()])?
Work service in Rust, with RFC 3339 timestamps1672 .all()
1673 .await?
1674 .results::<SessionRow>()?;
1675 Ok(Outcome::Ok(
1676 rows.into_iter().map(SessionEntry::from).collect(),
1677 ))
1678 }
1679
Events service in Rust, with RFC 3339 times and accurate push events1680 /// A push moves the head of the pull request it concerns: the one whose
1681 /// fork was pushed to, or the one opened from the branch that moved.
1682 async fn on_event(&self, event: &Event) -> Result<()> {
Work service in Rust, with RFC 3339 timestamps1683 if event.kind != "git.push" {
1684 return Ok(());
1685 }
Events service in Rust, with RFC 3339 times and accurate push events1686 let (Some(repo_id), Some(after), Some(git_ref)) = (
1687 event.repo_id.as_deref(),
1688 event.data["after"].as_str(),
1689 event.data["ref"].as_str(),
1690 ) else {
Work service in Rust, with RFC 3339 timestamps1691 return Ok(());
1692 };
Pull requests from branches1693 let now = rfc3339(now_ms());
Agents as a team: lifecycle, merge queue, billing and a new shell1694 // The head moved, so whatever the checks said no longer applies, and
1695 // whatever step g1t was waiting on has been taken.
Acceptance checks in sandboxes, line comments and review verdicts1696 let moved = "UPDATE pulls
Agents as a team: lifecycle, merge queue, billing and a new shell1697 SET head_commit = ?, updated_at = ?, check_status = NULL, check_run_id = NULL,
1698 working_on = NULL, working_until = NULL, stalled = NULL";
Acceptance checks in sandboxes, line comments and review verdicts1699 let active = "status IN ('draft', 'open') AND head_commit IS NOT ?";
1700 let returning = "RETURNING id, repo_id, number, issue_number, status";
1701 let mut pulls: Vec<MovedRow> = Vec::new();
Events service in Rust, with RFC 3339 times and accurate push events1702 // A fork carries its pull request on its default branch.
1703 if event.data["defaultBranch"].as_bool() == Some(true) {
Acceptance checks in sandboxes, line comments and review verdicts1704 pulls.extend(
Events service in Rust, with RFC 3339 times and accurate push events1705 self.db
1706 .prepare(format!(
Acceptance checks in sandboxes, line comments and review verdicts1707 "{moved} WHERE fork_repo_id = ? AND {active} {returning}"
Events service in Rust, with RFC 3339 times and accurate push events1708 ))
Acceptance checks in sandboxes, line comments and review verdicts1709 .bind(&[
1710 after.into(),
1711 now.as_str().into(),
1712 repo_id.into(),
1713 after.into(),
1714 ])?
1715 .all()
1716 .await?
1717 .results::<MovedRow>()?,
Events service in Rust, with RFC 3339 times and accurate push events1718 );
1719 }
1720 if let Some(branch) = git_ref.strip_prefix("refs/heads/") {
Acceptance checks in sandboxes, line comments and review verdicts1721 pulls.extend(
Pull requests from branches1722 self.db
Events service in Rust, with RFC 3339 times and accurate push events1723 .prepare(format!(
Acceptance checks in sandboxes, line comments and review verdicts1724 "{moved} WHERE repo_id = ? AND source_branch = ? AND {active} {returning}"
Events service in Rust, with RFC 3339 times and accurate push events1725 ))
1726 .bind(&[
1727 after.into(),
1728 now.as_str().into(),
1729 repo_id.into(),
1730 branch.into(),
Acceptance checks in sandboxes, line comments and review verdicts1731 after.into(),
1732 ])?
1733 .all()
1734 .await?
1735 .results::<MovedRow>()?,
Events service in Rust, with RFC 3339 times and accurate push events1736 );
Pull requests from branches1737 }
Agents as a team: lifecycle, merge queue, billing and a new shell1738 // What each now changes, so overlaps show while the work is under way.
1739 for moved in &pulls {
1740 if let Some(pull) = self.pull_by_id(&moved.id).await? {
1741 self.refresh_files(&pull).await?;
1742 }
1743 }
1744 // A merge that was waiting for this push to bring it up to date.
1745 for moved in &pulls {
1746 self.land_if_requested(&moved.id).await?;
1747 }
Acceptance checks in sandboxes, line comments and review verdicts1748 // A draft is announced when it is marked ready instead.
1749 for pull in pulls
1750 .into_iter()
1751 .filter(|pull| pull.status == PullStatus::Open)
1752 {
1753 self.publish_as(
1754 "pull.updated",
1755 &pull.repo_id,
1756 event.actor.clone(),
1757 PullEvent {
1758 pull_id: pull.id,
1759 repo_id: pull.repo_id.clone(),
1760 number: pull.number,
1761 issue: pull.issue_number,
1762 commit: Some(after.to_owned()),
1763 ..PullEvent::default()
1764 },
1765 )
1766 .await?;
1767 }
Work service in Rust, with RFC 3339 timestamps1768 Ok(())
1769 }
1770}
1771
1772fn service(env: &Env) -> Result<Work> {
1773 Ok(Work {
1774 db: env.d1("DB")?,
Agents as a team: lifecycle, merge queue, billing and a new shell1775 identity: env.service("IDENTITY")?,
Work service in Rust, with RFC 3339 timestamps1776 repos: env.service("REPOS")?,
Events service in Rust, with RFC 3339 times and accurate push events1777 events: env.service("EVENTS")?,
Sidebar: the panels really slide1778 actions: env.service("ACTIONS")?,
Work service in Rust, with RFC 3339 timestamps1779 })
1780}
1781
1782#[event(fetch)]
1783async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
1784 let Some(method) = rpc_method(&request) else {
1785 return Response::error("Not found", 404);
1786 };
1787 let body: serde_json::Value = request.json().await?;
1788 let work = service(&env)?;
1789
1790 match method.as_str() {
Issues and pull requests replace intents and attempts1791 "open_issue" => reply(&work.open_issue(args(body)?).await?),
1792 "list_issues" => reply(&work.list_issues(args(body)?).await?),
1793 "get_issue" => reply(&work.get_issue(args(body)?).await?),
1794 "update_issue" => reply(&work.update_issue(args(body)?).await?),
1795 "close_issue" => reply(&work.close_issue(args(body)?).await?),
1796 "reopen_issue" => reply(&work.reopen_issue(args(body)?).await?),
1797 "list_labels" => reply(&work.list_labels(args(body)?).await?),
1798 "counts" => reply(&work.counts(args(body)?).await?),
1799 "add_comment" => reply(&work.add_comment(args(body)?).await?),
Acceptance checks in sandboxes, line comments and review verdicts1800 "start_checks" => reply(&work.start_checks(args(body)?).await?),
1801 "report_checks" => reply(&work.report_checks(args(body)?).await?),
GitHub Actions on g1t, part two: running workflows1802 "set_commit_status" => reply(&work.set_commit_status(args(body)?).await?),
Agents as a team: lifecycle, merge queue, billing and a new shell1803 "start_review" => reply(&work.start_review(args(body)?).await?),
1804 "advance" => reply(&work.advance(args(body)?).await?),
1805 "stall" => reply(&work.stall(args(body)?).await?),
1806 "managed_pulls" => reply(&work.managed_pulls(args(body)?).await?),
1807 "queue" => reply(&work.queue(args(body)?).await?),
1808 "queue_build" => reply(&work.queue_build(args(body)?).await?),
1809 "report_queue" => reply(&work.report_queue(args(body)?).await?),
1810 "remove_from_queue" => reply(&work.remove_from_queue(args(body)?).await?),
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request1811 "message_agent" => reply(&work.message_agent(args(body)?).await?),
Record your own agent's sessions automatically1812 "locate_pull" => reply(&work.locate_pull(args(body)?).await?),
Agents ask each other, hand each other work, and answer1813 "answer_message" => reply(&work.answer_message(args(body)?).await?),
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request1814 "take_messages" => reply(&work.take_messages(args(body)?).await?),
Agents asked while not at work are woken to answer1815 "wake_for_messages" => reply(&work.wake_for_messages(args(body)?).await?),
Agents as a team: lifecycle, merge queue, billing and a new shell1816 "catch_up_job" => reply(&work.catch_up_job(args(body)?).await?),
1817 "get_settings" => reply(&work.get_settings(args(body)?).await?),
1818 "update_settings" => reply(&work.update_settings(args(body)?).await?),
1819 "report_review" => reply(&work.report_review(args(body)?).await?),
Issues and pull requests replace intents and attempts1820 "open_pull" => reply(&work.open_pull(args(body)?).await?),
1821 "list_pulls" => reply(&work.list_pulls(args(body)?).await?),
1822 "get_pull" => reply(&work.get_pull(args(body)?).await?),
Agents as a team: lifecycle, merge queue, billing and a new shell1823 "update_pull" => reply(&work.update_pull(args(body)?).await?),
Issues and pull requests replace intents and attempts1824 "ready_pull" => reply(&work.ready_pull(args(body)?).await?),
1825 "close_pull" => reply(&work.close_pull(args(body)?).await?),
1826 "merge_pull" => reply(&work.merge_pull(args(body)?).await?),
1827 "list_active_pulls" => reply(&work.list_active_pulls(args(body)?).await?),
Agents as a team: lifecycle, merge queue, billing and a new shell1828 "start_plan" => reply(&work.start_plan(args(body)?).await?),
1829 "report_plan" => reply(&work.report_plan(args(body)?).await?),
1830 "get_plan" => reply(&work.get_plan(args(body)?).await?),
1831 "list_plans" => reply(&work.list_plans(args(body)?).await?),
1832 "apply_plan" => reply(&work.apply_plan(args(body)?).await?),
1833 "queue_issue" => reply(&work.queue_issue(args(body)?).await?),
1834 "ready_issues" => reply(&work.ready_issues(args(body)?).await?),
1835 "list_assigned_issues" => reply(&work.list_assigned_issues(args(body)?).await?),
Work service in Rust, with RFC 3339 timestamps1836 "append_session" => reply(&work.append_session(args(body)?).await?),
1837 "read_session" => reply(&work.read_session(args(body)?).await?),
1838 _ => Response::error("Unknown method", 404),
1839 }
1840}
1841
1842/// Events from the bus, delivered on this service's own queue.
1843#[event(queue)]
Events service in Rust, with RFC 3339 times and accurate push events1844async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
Work service in Rust, with RFC 3339 timestamps1845 let work = service(&env)?;
1846 for message in batch.messages()? {
1847 work.on_event(message.body()).await?;
1848 message.ack();
1849 }
1850 Ok(())
1851}