Skip to content

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

474 linesCodeBlameRaw
1//! `get_work`: everything about a work item an agent needs to pick it up,
2//! in one call. What it is for and what done means, the repositories it is
3//! linked to with their pull requests for it and what landed where, what
4//! is still missing, where else it may belong (suggested targets), who is
5//! on it, what it waits on and what waits on it, the decisions made about
6//! it, and its latest coordination events.
7
8use std::collections::HashMap;
9
10use futures_util::future::{try_join, try_join_all, try_join3};
11use g1t_contracts::access::Capability;
12use g1t_contracts::decisions::{Decision, DecisionKind, ListDecisionsArgs};
13use g1t_contracts::events::{Event, ListArgs as ListEventsArgs};
14use g1t_contracts::repos::{ReadableArgs, Repo};
15use g1t_contracts::work::*;
16use g1t_contracts::{Outcome, Viewer};
17use g1t_kit::now_ms;
18use serde::Deserialize;
19use worker::Result;
20
21use super::{SESSION_COLUMNS, SessionRow, unwrap_or_fail};
22use crate::Work;
23use crate::items::{self, Held, PastLanding};
24use crate::rows::{PULL_COLUMNS, PullRow};
25
26/// How many sessions, decisions, events and suggestions `get_work` gives
27/// at most.
28const SESSIONS: u32 = 50;
29const DECISIONS: u32 = 20;
30const EVENTS: u32 = 20;
31const SUGGESTIONS: usize = 5;
32/// How many earlier landings suggestions are drawn from.
33const LANDINGS: u32 = 200;
34
35#[derive(Deserialize)]
36struct PlaceRow {
37 state: QueueState,
38 position: u32,
39}
40
41/// A repository of the workspace, and how many files sessions on the work
42/// item touched in it.
43#[derive(Clone, Debug, Deserialize)]
44pub(crate) struct TouchedRow {
45 pub repo_id: String,
46 pub files: u32,
47}
48
49/// The repositories where sessions on work item `?1` touched files.
50const TOUCHED_FOR: &str = "SELECT s.repo_id AS repo_id, count(*) AS files
51 FROM session_touches t JOIN sessions s ON s.id = t.session_id
52 WHERE t.kind = 'file'
53 AND (s.item_id = ?1 OR s.id IN (SELECT session_id FROM session_links WHERE kind = 'item' AND item_id = ?1))
54 GROUP BY s.repo_id";
55
56/// The words of `text` that say something: lowercase, four letters or
57/// more.
58fn words(text: &str) -> Vec<String> {
59 let mut found: Vec<String> = Vec::new();
60 for word in text.split(|c: char| !c.is_alphanumeric()).map(str::to_lowercase) {
61 if word.chars().count() >= 4 && !found.contains(&word) {
62 found.push(word);
63 }
64 }
65 found
66}
67
68/// Where a work item may belong besides where it is meant to land already:
69/// the workspace's repositories (`candidates`, by id and path), ranked by
70/// where work items sharing its labels or title words landed before, by a
71/// repository's name in its title or description, and by where sessions on
72/// it touched files. Each says why; the likeliest come first.
73pub(crate) fn suggest_targets(
74 item: &WorkItem,
75 candidates: &[(String, String)],
76 landed: &[PastLanding],
77 touched: &[TouchedRow],
78) -> Vec<SuggestedTarget> {
79 // A closed work item needs nowhere else to land.
80 if item.state == State::Closed {
81 return Vec::new();
82 }
83 let title_words = words(&item.title);
84 let text = format!("{}\n{}", item.title, item.body).to_lowercase();
85 let mut suggested: Vec<SuggestedTarget> = Vec::new();
86 let counted = |n: u32| if n == 1 { "1 work item".to_owned() } else { format!("{n} work items") };
87 for (repo_id, path) in candidates {
88 // Linked already, as its origin, a target or touched.
89 if item.repos.iter().any(|repo| repo.repo_id == *repo_id) {
90 continue;
91 }
92 let mut score = 0;
93 let mut why: Vec<String> = Vec::new();
94 let landings: Vec<&PastLanding> = landed.iter().filter(|landing| landing.repo_id == *repo_id).collect();
95 let labelled = landings
96 .iter()
97 .filter(|landing| {
98 let labels: Vec<String> = serde_json::from_str(&landing.labels).unwrap_or_default();
99 labels.iter().any(|label| item.labels.contains(label))
100 })
101 .count() as u32;
102 if labelled > 0 {
103 score += 3 * labelled;
104 why.push(format!("{} with the same labels landed there", counted(labelled)));
105 }
106 let alike = landings.iter().filter(|landing| words(&landing.title).iter().any(|word| title_words.contains(word))).count() as u32;
107 if alike > 0 {
108 score += 2 * alike;
109 why.push(format!("{} with like titles landed there", counted(alike)));
110 }
111 let name = path.rsplit('/').next().unwrap_or(path).to_lowercase();
112 let names_it = text.split(|c: char| !(c.is_alphanumeric() || c == '-' || c == '_' || c == '.')).any(|word| word == name);
113 if names_it || text.contains(&path.to_lowercase()) {
114 score += 5;
115 why.push(format!("it names {name}"));
116 }
117 if let Some(found) = touched.iter().find(|found| found.repo_id == *repo_id) {
118 score += 4;
119 why.push(format!("sessions on it touched {} files there", found.files));
120 }
121 if score > 0 {
122 let mut why = why.join("; ");
123 if let Some(first) = why.get_mut(0..1) {
124 first.make_ascii_uppercase();
125 }
126 suggested.push(SuggestedTarget { repo_id: repo_id.clone(), path: path.clone(), why: format!("{why}."), score });
127 }
128 }
129 suggested.sort_by(|a, b| b.score.cmp(&a.score).then_with(|| a.path.cmp(&b.path)));
130 suggested.truncate(SUGGESTIONS);
131 suggested
132}
133
134/// Who is on a work item of `workspace`, as `viewer` is shown it: the
135/// claims and sessions in repositories they may read (`readable`), each
136/// session as `shown` lets them see it.
137fn shown_people(claims: Vec<Claim>, mut sessions: Vec<Session>, readable: &[String], viewer: &Viewer, workspace: &str) -> (Vec<Claim>, Vec<Session>) {
138 let may_read = |repo_id: &str| readable.iter().any(|id| id == repo_id);
139 let claims = claims.into_iter().filter(|claim| claim.repo_id.as_deref().is_none_or(may_read)).collect();
140 sessions.retain(|session| may_read(&session.repo_id));
141 super::shown::sessions(&mut sessions, viewer, workspace);
142 (claims, sessions)
143}
144
145impl Work {
146 pub(crate) async fn get_work(&self, a: WorkItemViewArgs) -> Result<Outcome<WorkContext>> {
147 let held = unwrap_or_fail!(items::for_viewer(self, &a.viewer, &a.workspace, a.number, Capability::Read).await?);
148 // What went quiet in its repositories is read as it is now.
149 for repo_id in held.repo_ids().iter().take(10) {
150 self.sweep(repo_id).await?;
151 }
152 let pulls: Vec<Pull> = self
153 .db
154 .prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE item_id = ? ORDER BY created_at"))
155 .bind(&[held.item.id.as_str().into()])?
156 .all()
157 .await?
158 .results::<PullRow>()?
159 .into_iter()
160 .map(Pull::from)
161 .collect();
162 // The repositories it is linked to and has pull requests in, as far
163 // as the viewer may read them.
164 let mut ids = held.repo_ids();
165 for pull in &pulls {
166 if !ids.contains(&pull.repo_id) {
167 ids.push(pull.repo_id.clone());
168 }
169 }
170 let readable: Vec<Repo> = g1t_kit::call(&self.repos, "readable", &ReadableArgs { ids, viewer: a.viewer.clone() }).await?;
171 let pulls: Vec<Pull> = pulls.into_iter().filter(|pull| readable.iter().any(|repo| repo.id == pull.repo_id)).collect();
172 let mut numbers = vec![held.item.number];
173 numbers.extend(pulls.iter().map(|pull| pull.number));
174 let viewer = a.viewer.as_ref();
175 let may_read = |repo_id: &str| readable.iter().any(|repo| repo.id == repo_id);
176
177 let people = try_join3(
178 self.active_claims(&held.item.id),
179 self.work_sessions(&held, &pulls),
180 self.suggestions(&held, viewer),
181 );
182 let dependencies = try_join(self.blockers(viewer, &held.item.workspace, &held.item.blocked_by), items::blocked_items(self, &held, viewer));
183 // Only the repositories the viewer may read: the events service
184 // takes the ids it is given as checked.
185 let linked: Vec<String> = held.repo_ids().into_iter().filter(|id| may_read(id)).collect();
186 let history = try_join(
187 self.coordination_events(&linked, &numbers, COORDINATION_EVENTS.to_vec(), EVENTS),
188 self.decisions_on(&held, &a.viewer),
189 );
190 let repos = try_join_all(held.item.repos.iter().filter_map(|link| {
191 let repo = readable.iter().find(|repo| repo.id == link.repo_id)?;
192 let mine: Vec<&Pull> = pulls.iter().filter(|pull| pull.repo_id == repo.id).collect();
193 Some(self.work_repo(link.clone(), repo, mine))
194 }));
195 let (((claims, sessions, suggested_targets), (blocked_by, blocks)), ((recent, decisions), repos)) =
196 try_join(try_join(people, dependencies), try_join(history, repos)).await?;
197 let readable_ids: Vec<String> = readable.iter().map(|repo| repo.id.clone()).collect();
198 let (claims, sessions) = shown_people(claims, sessions, &readable_ids, &a.viewer, &held.item.workspace);
199 let item = held.shown(viewer);
200 Ok(Outcome::Ok(WorkContext {
201 definition_of_done: definition_of_done(&item.body),
202 missing: item.missing(),
203 item,
204 repos,
205 suggested_targets,
206 claims,
207 sessions,
208 blocked_by,
209 blocks,
210 last_integration_failure: decisions.iter().find(|decision| decision.kind == DecisionKind::IntegrationFailure).cloned(),
211 decisions,
212 recent,
213 }))
214 }
215
216 /// One repository a work item is linked to: its required checks, and
217 /// its pull requests for the item.
218 async fn work_repo(&self, link: ItemRepo, repo: &Repo, pulls: Vec<&Pull>) -> Result<WorkRepo> {
219 let settings = self.default_branch_settings(repo).await?;
220 let pulls = try_join_all(pulls.into_iter().map(|pull| self.work_pull(repo, pull))).await?;
221 Ok(WorkRepo { link, required_checks: settings.required_checks, pulls })
222 }
223
224 /// Where else a work item may belong ([`suggest_targets`]), among the
225 /// repositories `viewer` may read.
226 async fn suggestions(&self, held: &Held, viewer: Option<&g1t_contracts::User>) -> Result<Vec<SuggestedTarget>> {
227 #[derive(Deserialize)]
228 struct Candidate {
229 repo_id: String,
230 path: String,
231 workspace: String,
232 private: u8,
233 }
234 if held.item.state == State::Closed {
235 return Ok(Vec::new());
236 }
237 let results = self
238 .db
239 .batch(vec![
240 self.db
241 .prepare("SELECT repo_id, path, workspace, private FROM repo_workspaces WHERE workspace = ? ORDER BY path LIMIT 500")
242 .bind(&[held.item.workspace.as_str().into()])?,
243 self.db.prepare(TOUCHED_FOR).bind(&[held.item.id.as_str().into()])?,
244 ])
245 .await?;
246 let mut results = results.into_iter();
247 let candidates: Vec<(String, String)> = match results.next() {
248 Some(result) => result
249 .results::<Candidate>()?
250 .into_iter()
251 .filter(|row| items::repo_readable(viewer, &row.repo_id, Some(&row.workspace), Some(row.private)))
252 .map(|row| (row.repo_id, row.path))
253 .collect(),
254 None => Vec::new(),
255 };
256 let landed = items::landed_before(self, held, LANDINGS).await?;
257 let touched: Vec<TouchedRow> = match results.next() {
258 Some(result) => result.results()?,
259 None => Vec::new(),
260 };
261 Ok(suggest_targets(&held.item, &candidates, &landed, &touched))
262 }
263
264 /// Sessions on a work item or its pull requests, in any of their
265 /// repositories, most recently heard from first. Subagents are listed
266 /// beside the sessions that started them, so sibling agents on one
267 /// work item see each other; so are sessions tied to it any other way
268 /// (a claim, a branch, a trailer, `#N`).
269 async fn work_sessions(&self, held: &Held, pulls: &[Pull]) -> Result<Vec<Session>> {
270 let now = now_ms();
271 let mut pushed: Vec<String> = Vec::new();
272 let mut by_repo: HashMap<&str, Vec<u32>> = HashMap::new();
273 for pull in pulls {
274 by_repo.entry(pull.repo_id.as_str()).or_default().push(pull.number);
275 }
276 for (repo_id, numbers) in &by_repo {
277 pushed.extend(self.branch_sessions(repo_id, numbers).await?);
278 }
279 let pull_ids: Vec<&str> = pulls.iter().map(|pull| pull.id.as_str()).collect();
280 let mut sessions: Vec<Session> = self
281 .db
282 .prepare(format!(
283 "SELECT {SESSION_COLUMNS} FROM sessions
284 WHERE item_id = ?1
285 OR id IN (SELECT value FROM json_each(?3))
286 OR EXISTS (SELECT 1 FROM pulls p WHERE p.id IN (SELECT value FROM json_each(?2))
287 AND p.repo_id = sessions.repo_id AND p.number = sessions.pull)
288 OR id IN (SELECT session_id FROM session_links WHERE kind = 'item' AND item_id = ?1)
289 OR id IN (SELECT l.session_id FROM session_links l JOIN pulls p ON p.repo_id = l.repo_id AND p.number = l.number
290 WHERE l.kind = 'pull' AND p.id IN (SELECT value FROM json_each(?2)))
291 ORDER BY state = 'ended', last_seen_at DESC LIMIT ?4"
292 ))
293 .bind(&[
294 held.item.id.as_str().into(),
295 serde_json::to_string(&pull_ids)?.into(),
296 serde_json::to_string(&pushed)?.into(),
297 SESSIONS.into(),
298 ])?
299 .all()
300 .await?
301 .results::<SessionRow>()?
302 .into_iter()
303 .map(|row| row.at(now))
304 .collect();
305 self.fill_subagents(&mut sessions).await?;
306 Ok(sessions)
307 }
308
309 /// The latest events of `types` about `numbers` in `repos`, newest
310 /// first.
311 async fn coordination_events(&self, repos: &[String], numbers: &[u32], types: Vec<&str>, limit: u32) -> Result<Vec<Event>> {
312 if repos.is_empty() {
313 return Ok(Vec::new());
314 }
315 g1t_kit::call(
316 &self.events,
317 "list",
318 &ListEventsArgs {
319 repo_ids: repos.to_vec(),
320 types: types.into_iter().map(str::to_owned).collect(),
321 numbers: numbers.to_vec(),
322 limit: Some(limit),
323 ..ListEventsArgs::default()
324 },
325 )
326 .await
327 }
328
329 /// Decision records on a work item and its pull requests, newest
330 /// first, up to [`DECISIONS`].
331 async fn decisions_on(&self, held: &Held, viewer: &Viewer) -> Result<Vec<Decision>> {
332 let args = ListDecisionsArgs {
333 repo: None,
334 workspace: held.item.workspace.clone(),
335 viewer: viewer.clone(),
336 number: held.item.number,
337 };
338 let mut decisions = match self.list_decisions(args).await? {
339 Outcome::Ok(decisions) => decisions,
340 Outcome::Fail(_) => Vec::new(),
341 };
342 decisions.reverse();
343 decisions.truncate(DECISIONS as usize);
344 Ok(decisions)
345 }
346
347 /// One of the work item's pull requests and where it stands: its
348 /// required checks, whether it merges, what it overlaps and its place
349 /// in the merge queue, for those still in progress.
350 async fn work_pull(&self, repo: &Repo, pull: &Pull) -> Result<WorkPull> {
351 let mut pull = pull.clone();
352 crate::fill_base(&mut pull, repo);
353 let brief = |pull: &Pull| WorkPull {
354 repo: items::path_of(repo),
355 number: pull.number,
356 title: pull.title.clone(),
357 status: pull.status,
358 branch: pull.branch.clone(),
359 head_commit: pull.head_commit.clone(),
360 author: pull.owner().into(),
361 check_status: pull.check_status,
362 required_checks: Vec::new(),
363 mergeable: Mergeable::Unknown,
364 overlaps: Vec::new(),
365 queue: None,
366 };
367 if !pull.status.is_active() {
368 return Ok(brief(&pull));
369 }
370 let required = async {
371 let settings = self.settings_on(repo, &pull).await?;
372 if settings.required_checks.is_empty() {
373 return Ok(Vec::new());
374 }
375 let statuses = self.statuses(&repo.id, pull.head_commit.as_deref()).await?;
376 let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses).await?;
377 Ok::<_, worker::Error>(required_checks(&settings.required_checks, &checked))
378 };
379 let place_query = self
380 .db
381 .prepare(
382 "SELECT mine.state AS state,
383 (SELECT count(*) FROM queue_entries ahead
384 WHERE ahead.repo_id = mine.repo_id AND ahead.state IN ('waiting', 'testing', 'passed')
385 AND (ahead.created_at < mine.created_at
386 OR (ahead.created_at = mine.created_at AND ahead.id < mine.id))) + 1 AS position
387 FROM queue_entries mine
388 WHERE mine.pull_id = ? AND mine.state IN ('waiting', 'testing', 'passed') LIMIT 1",
389 )
390 .bind(&[pull.id.as_str().into()])?;
391 let place = place_query.first::<PlaceRow>(None);
392 let ((required_checks, (mergeable, _)), (overlaps, place)) =
393 try_join(try_join(required, self.mergeability(&pull)), try_join(self.overlaps(&pull), place)).await?;
394 Ok(WorkPull {
395 required_checks,
396 mergeable,
397 overlaps,
398 queue: place.map(|place| QueuePlace { state: place.state, position: place.position }),
399 ..brief(&pull)
400 })
401 }
402}
403
404#[cfg(test)]
405mod tests {
406 use super::*;
407
408 fn item(title: &str, body: &str, labels: &[&str], repos: Vec<ItemRepo>) -> WorkItem {
409 serde_json::from_value(serde_json::json!({
410 "id": "wki_1", "workspace": "acme", "number": 7, "title": title, "body": body,
411 "labels": labels, "state": "open", "reason": null, "author": { "id": "usr_1", "username": "ana" },
412 "createdAt": "", "updatedAt": "", "closedAt": null, "pullCount": 0, "commentCount": 0,
413 "repos": repos
414 }))
415 .unwrap()
416 }
417
418 fn landing(repo: &str, title: &str, labels: &str) -> PastLanding {
419 PastLanding { repo_id: repo.into(), title: title.into(), labels: labels.into() }
420 }
421
422 #[test]
423 fn suggestions_say_why_and_come_likeliest_first() {
424 let candidates = vec![
425 ("rep_api".to_owned(), "acme/api".to_owned()),
426 ("rep_web".to_owned(), "acme/web".to_owned()),
427 ("rep_docs".to_owned(), "acme/docs".to_owned()),
428 ];
429 let item = item("Login page shows a stale avatar", "The web client caches it.", &["bug"], Vec::new());
430 let landed = vec![
431 landing("rep_api", "Avatar upload fails", r#"["bug"]"#),
432 landing("rep_api", "Rate limits", r#"["enhancement"]"#),
433 landing("rep_docs", "Typo", "[]"),
434 ];
435 let touched = vec![TouchedRow { repo_id: "rep_web".into(), files: 3 }];
436 let suggested = suggest_targets(&item, &candidates, &landed, &touched);
437 let paths: Vec<&str> = suggested.iter().map(|s| s.path.as_str()).collect();
438 assert_eq!(paths, ["acme/web", "acme/api"]);
439 assert_eq!(suggested[0].why, "It names web; sessions on it touched 3 files there.");
440 assert_eq!(suggested[1].why, "1 work item with the same labels landed there; 1 work item with like titles landed there.");
441 assert!(suggested[0].score > suggested[1].score);
442 }
443
444 #[test]
445 fn a_repository_already_linked_is_not_suggested() {
446 let candidates = vec![("rep_web".to_owned(), "acme/web".to_owned())];
447 for role in LinkRole::ALL {
448 let linked = ItemRepo { repo_id: "rep_web".into(), path: "acme/web".into(), role, legacy_number: None, landed: false };
449 let item = item("Fix web", "", &[], vec![linked]);
450 assert!(suggest_targets(&item, &candidates, &[], &[]).is_empty(), "{role:?}");
451 }
452 }
453
454 #[test]
455 fn sessions_and_claims_are_shown_as_the_viewer_may_see_them() {
456 let public = super::super::shown::tests::a_session();
457 let private = Session { id: "ses_2".into(), repo_id: "rep_secret".into(), ..public.clone() };
458 let (_, sessions) = shown_people(Vec::new(), vec![public, private], &["rep_1".to_owned()], &None, "acme");
459 // Only the one in a repository they may read, and not what is inside it.
460 assert_eq!(sessions.len(), 1);
461 assert_eq!(sessions[0].id, "ses_1");
462 assert_eq!(sessions[0].branch, None);
463 assert_eq!(sessions[0].title, None);
464 }
465
466 #[test]
467 fn a_closed_work_item_gets_no_suggestions() {
468 let candidates = vec![("rep_web".to_owned(), "acme/web".to_owned())];
469 let mut closed = item("Fix web", "", &[], Vec::new());
470 assert_eq!(suggest_targets(&closed, &candidates, &[], &[]).len(), 1);
471 closed.state = State::Closed;
472 assert!(suggest_targets(&closed, &candidates, &[], &[]).is_empty());
473 }
474}