| 1 | //! What sessions touched, and the work items and pull requests they worked |
| 2 | //! on. |
| 3 | //! |
| 4 | //! A session is tied to a work item or pull request in as many ways as it |
| 5 | //! was (`LinkVia`): attached when it started or updated, a claim it made, |
| 6 | //! a pull request it opened or a branch or commit of one, a |
| 7 | //! `G1t-Session:` trailer on a commit pushed to one, and `#N` in its |
| 8 | //! prompt, its task or a subagent's. Each way is a row of |
| 9 | //! `session_links`; reading a session gathers its whole tree's |
| 10 | //! (`gather_links`). Files two sessions touched make them overlap |
| 11 | //! (`overlap.detected`, once per pair). |
| 12 | |
| 13 | use std::collections::{HashMap, HashSet}; |
| 14 | |
| 15 | use g1t_contracts::Outcome; |
| 16 | use g1t_contracts::repos::{Commit, LogArgs, RepoPath}; |
| 17 | use g1t_contracts::time::rfc3339; |
| 18 | use g1t_contracts::work::*; |
| 19 | use g1t_kit::now_ms; |
| 20 | use serde::Deserialize; |
| 21 | use worker::Result; |
| 22 | |
| 23 | use super::updates::push_delivery; |
| 24 | use super::{SESSION_COLUMNS, SessionRow}; |
| 25 | use crate::items::{self, Named}; |
| 26 | use crate::Work; |
| 27 | |
| 28 | /// The start (`?2` characters) of the first prompt each session in `?1` |
| 29 | /// (a JSON list of ids) recorded, by the session's own numbering. |
| 30 | pub(crate) const FIRST_PROMPTS: &str = "SELECT session_id, text FROM ( |
| 31 | SELECT ids.value AS session_id, |
| 32 | (SELECT substr(e.text, 1, ?2) FROM session_entries e |
| 33 | WHERE e.session_id = ids.value AND e.kind = 'prompt' ORDER BY e.session_seq LIMIT 1) AS text |
| 34 | FROM json_each(?1) AS ids) |
| 35 | WHERE text IS NOT NULL"; |
| 36 | |
| 37 | /// The most `#N` of one text looked up. |
| 38 | const MAX_REFS: usize = 10; |
| 39 | /// The most work items and pull requests a session's links name. |
| 40 | const MAX_LINKED: usize = 40; |
| 41 | /// The most commits of one push read for trailers. |
| 42 | const MAX_TRAILER_COMMITS: u32 = 20; |
| 43 | /// The most subagents a tree shows. |
| 44 | pub(crate) const MAX_TREE: u32 = 200; |
| 45 | |
| 46 | /// Ties session `?1` of repository `?2` to pull request `?4` (`?3` `pull`, |
| 47 | /// `?7` empty) or work item `?7` (`?3` `item`, `?4` 0) by way of `?5`, at |
| 48 | /// `?6`; nothing when it is tied that way already. |
| 49 | pub(crate) const INSERT_LINK: &str = "INSERT INTO session_links (session_id, repo_id, kind, number, via, at, item_id) |
| 50 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) ON CONFLICT DO NOTHING"; |
| 51 | |
| 52 | /// Records that session `?1` of repository `?2` touched (`?3`) `?4`, at |
| 53 | /// `?5`; returns it only the first time. |
| 54 | pub(crate) const INSERT_TOUCH: &str = "INSERT INTO session_touches (session_id, repo_id, kind, value, at) |
| 55 | VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT DO NOTHING RETURNING kind, value"; |
| 56 | |
| 57 | /// Session `?1`'s subagents, and theirs, in the order they started. |
| 58 | pub(crate) fn tree_sql() -> String { |
| 59 | format!( |
| 60 | "WITH RECURSIVE tree(id) AS ( |
| 61 | SELECT id FROM sessions WHERE parent_id = ?1 |
| 62 | UNION SELECT s.id FROM sessions s JOIN tree ON s.parent_id = tree.id) |
| 63 | SELECT {SESSION_COLUMNS} FROM sessions WHERE id IN (SELECT id FROM tree) |
| 64 | ORDER BY started_at, id LIMIT {MAX_TREE}" |
| 65 | ) |
| 66 | } |
| 67 | |
| 68 | /// Ends session `?1`'s subagents still going, and theirs, at `?2` for |
| 69 | /// `?3`: only once `?1` itself ended at `?2`, so in the batch that ends it, |
| 70 | /// a call that lost the race to end it ends nothing. |
| 71 | pub(crate) fn end_tree_sql() -> String { |
| 72 | format!( |
| 73 | "WITH RECURSIVE tree(id) AS ( |
| 74 | SELECT id FROM sessions WHERE parent_id = ?1 |
| 75 | UNION SELECT s.id FROM sessions s JOIN tree ON s.parent_id = tree.id) |
| 76 | UPDATE sessions SET state = 'ended', ended_at = ?2, end_reason = ?3, last_seen_at = ?2 |
| 77 | WHERE id IN (SELECT id FROM tree) AND state = 'active' |
| 78 | AND EXISTS (SELECT 1 FROM sessions WHERE id = ?1 AND state = 'ended' AND ended_at = ?2) |
| 79 | RETURNING {SESSION_COLUMNS}" |
| 80 | ) |
| 81 | } |
| 82 | |
| 83 | /// Session `?1`'s line: its parents up to the top, and its subagents and |
| 84 | /// theirs. Never told of overlapping with each other. |
| 85 | pub(crate) const FAMILY: &str = "WITH RECURSIVE |
| 86 | up(id, parent) AS ( |
| 87 | SELECT id, parent_id FROM sessions WHERE id = ?1 |
| 88 | UNION SELECT s.id, s.parent_id FROM sessions s JOIN up ON s.id = up.parent), |
| 89 | down(id) AS ( |
| 90 | SELECT id FROM sessions WHERE parent_id = ?1 |
| 91 | UNION SELECT s.id FROM sessions s JOIN down ON s.parent_id = down.id) |
| 92 | SELECT id FROM up UNION SELECT id FROM down"; |
| 93 | |
| 94 | /// The files of `?2` (JSON) that active sessions of repository `?1` other |
| 95 | /// than `?3` touched. |
| 96 | pub(crate) const TOUCHED_BY_OTHERS: &str = "SELECT t.session_id AS session_id, t.value AS value |
| 97 | FROM session_touches t JOIN sessions s ON s.id = t.session_id |
| 98 | WHERE t.repo_id = ?1 AND t.kind = 'file' AND t.value IN (SELECT value FROM json_each(?2)) |
| 99 | AND t.session_id <> ?3 AND s.state = 'active'"; |
| 100 | |
| 101 | /// The sessions of repository `?1` that pushed the branch of one of the |
| 102 | /// pull requests `?2` (JSON numbers). |
| 103 | pub(crate) const BRANCH_SESSIONS: &str = "SELECT DISTINCT t.session_id AS id FROM session_touches t |
| 104 | JOIN pulls p ON p.repo_id = t.repo_id AND p.source_branch = t.value |
| 105 | AND p.fork_repo_id IS NULL AND p.head_repo_id IS NULL |
| 106 | WHERE t.repo_id = ?1 AND t.kind = 'branch' AND p.number IN (SELECT value FROM json_each(?2)) LIMIT 200"; |
| 107 | |
| 108 | /// The sessions tied to repository `?1`'s pull requests numbered in `?2` |
| 109 | /// (JSON): by a link, by saying so when they started, or by pushing the |
| 110 | /// branch. A subagent is named with the session that started it. |
| 111 | pub(crate) const PULL_SESSIONS: &str = "WITH asked(n) AS (SELECT value FROM json_each(?2)), |
| 112 | tied(number, session_id) AS ( |
| 113 | SELECT l.number, l.session_id FROM session_links l WHERE l.repo_id = ?1 AND l.kind = 'pull' AND l.number IN asked |
| 114 | UNION SELECT s.pull, s.id FROM sessions s WHERE s.repo_id = ?1 AND s.pull IN asked |
| 115 | UNION SELECT p.number, t.session_id FROM session_touches t |
| 116 | JOIN pulls p ON p.repo_id = t.repo_id AND p.source_branch = t.value AND p.fork_repo_id IS NULL AND p.head_repo_id IS NULL |
| 117 | WHERE t.repo_id = ?1 AND t.kind = 'branch' AND p.number IN asked) |
| 118 | SELECT tied.number AS number, s.id AS id, s.harness AS harness, s.parent_id AS parent_id, s.started_at AS started_at, |
| 119 | (SELECT p.harness FROM sessions p WHERE p.id = s.parent_id) AS parent_harness |
| 120 | FROM tied JOIN sessions s ON s.id = tied.session_id |
| 121 | ORDER BY tied.number, s.started_at LIMIT 1000"; |
| 122 | |
| 123 | /// A row of [`PULL_SESSIONS`]. |
| 124 | #[derive(Clone, Debug, Deserialize)] |
| 125 | pub(crate) struct PullSessionRow { |
| 126 | pub number: u32, |
| 127 | pub id: String, |
| 128 | pub harness: String, |
| 129 | pub parent_id: Option<String>, |
| 130 | pub parent_harness: Option<String>, |
| 131 | } |
| 132 | |
| 133 | /// Each pull request's sessions, a subagent's as the session that started |
| 134 | /// it, each once, oldest first. |
| 135 | pub(crate) fn pull_sessions_of(rows: Vec<PullSessionRow>) -> Vec<PullSessions> { |
| 136 | let mut out: Vec<PullSessions> = Vec::new(); |
| 137 | for row in rows { |
| 138 | let (id, harness) = match (row.parent_id, row.parent_harness) { |
| 139 | (Some(parent), Some(harness)) => (parent, harness), |
| 140 | _ => (row.id, row.harness), |
| 141 | }; |
| 142 | let session = SessionRef { id, harness: Harness::parse(&harness).unwrap_or(Harness::Other) }; |
| 143 | match out.iter_mut().find(|entry| entry.number == row.number) { |
| 144 | Some(entry) if entry.sessions.iter().any(|known| known.id == session.id) => {} |
| 145 | Some(entry) => entry.sessions.push(session), |
| 146 | None => out.push(PullSessions { number: row.number, sessions: vec![session] }), |
| 147 | } |
| 148 | } |
| 149 | out |
| 150 | } |
| 151 | |
| 152 | /// Announces the pair `?2`, `?3` of repository `?1` as overlapping on |
| 153 | /// `?4` (JSON) at `?5`: returns it the first time, and again whenever a |
| 154 | /// file it was not announced for joins, so a pair that first met on a |
| 155 | /// trivial file still hears when it starts changing the same code. |
| 156 | pub(crate) const INSERT_OVERLAP: &str = "INSERT INTO session_overlaps (repo_id, low, high, files, noticed_at) |
| 157 | VALUES (?1, ?2, ?3, ?4, ?5) |
| 158 | ON CONFLICT (repo_id, low, high) DO UPDATE SET |
| 159 | files = (SELECT json_group_array(value) FROM ( |
| 160 | SELECT value FROM json_each(session_overlaps.files) |
| 161 | UNION SELECT value FROM json_each(excluded.files))), |
| 162 | noticed_at = excluded.noticed_at |
| 163 | WHERE EXISTS (SELECT 1 FROM json_each(excluded.files) AS added |
| 164 | WHERE added.value NOT IN (SELECT value FROM json_each(session_overlaps.files))) |
| 165 | RETURNING low"; |
| 166 | |
| 167 | #[derive(Deserialize)] |
| 168 | struct IdRow { |
| 169 | id: String, |
| 170 | } |
| 171 | |
| 172 | #[derive(Deserialize)] |
| 173 | struct NumberOnly { |
| 174 | number: u32, |
| 175 | } |
| 176 | |
| 177 | #[derive(Deserialize)] |
| 178 | pub(crate) struct LinkRow { |
| 179 | session_id: String, |
| 180 | kind: String, |
| 181 | /// A pull request's number, or a work item's in its workspace. |
| 182 | #[serde(default)] |
| 183 | number: Option<u32>, |
| 184 | via: String, |
| 185 | } |
| 186 | |
| 187 | #[derive(Deserialize)] |
| 188 | pub(crate) struct TouchRow { |
| 189 | session_id: String, |
| 190 | kind: String, |
| 191 | value: String, |
| 192 | at: String, |
| 193 | } |
| 194 | |
| 195 | #[derive(Deserialize)] |
| 196 | struct PairRow { |
| 197 | session_id: String, |
| 198 | value: String, |
| 199 | } |
| 200 | |
| 201 | #[derive(Deserialize)] |
| 202 | struct SkillRow { |
| 203 | session_id: String, |
| 204 | name: String, |
| 205 | uses: u32, |
| 206 | } |
| 207 | |
| 208 | impl Work { |
| 209 | /// Ties `session_id` to a pull request or a work item one way. Never |
| 210 | /// fails what it follows: a problem is logged. |
| 211 | async fn link(&self, session_id: &str, repo_id: &str, kind: LinkKind, number: u32, item_id: &str, via: LinkVia) { |
| 212 | let at = rfc3339(now_ms()); |
| 213 | let linked = async { |
| 214 | let new = self |
| 215 | .db |
| 216 | .prepare(format!("{INSERT_LINK} RETURNING via")) |
| 217 | .bind(&[ |
| 218 | session_id.into(), |
| 219 | repo_id.into(), |
| 220 | kind.as_str().into(), |
| 221 | number.into(), |
| 222 | via.as_str().into(), |
| 223 | at.as_str().into(), |
| 224 | item_id.into(), |
| 225 | ])? |
| 226 | .all() |
| 227 | .await? |
| 228 | .results::<serde_json::Value>()?; |
| 229 | // Tied to what it works on for the first time: its title is due (titles.rs). |
| 230 | if !new.is_empty() && via != LinkVia::Ref { |
| 231 | self.db.prepare(super::titles::linked_sql()).bind(&[session_id.into(), at.as_str().into()])?.run().await?; |
| 232 | } |
| 233 | Ok::<(), worker::Error>(()) |
| 234 | }; |
| 235 | if let Err(error) = linked.await { |
| 236 | worker::console_error!("session {session_id}: not tied to {} #{number}{item_id}: {error}", kind.as_str()); |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | /// Ties `session_id` to pull request `number` of `repo_id` one way. |
| 241 | pub(crate) async fn link_pull(&self, session_id: &str, repo_id: &str, number: u32, via: LinkVia) { |
| 242 | self.link(session_id, repo_id, LinkKind::Pull, number, "", via).await; |
| 243 | } |
| 244 | |
| 245 | /// Ties `session_id`, of `repo_id`, to work item `item_id` one way. |
| 246 | pub(crate) async fn link_item(&self, session_id: &str, repo_id: &str, item_id: &str, via: LinkVia) { |
| 247 | self.link(session_id, repo_id, LinkKind::Item, 0, item_id, via).await; |
| 248 | } |
| 249 | |
| 250 | /// Ties a session to what the references in `text` name, read in its |
| 251 | /// repository as every reader of them reads them |
| 252 | /// (`g1t_contracts::work::refs`, outside code): one of its workspace's |
| 253 | /// pull requests, or work items. |
| 254 | pub(crate) async fn link_refs(&self, session_id: &str, repo_id: &str, text: &str) -> Result<()> { |
| 255 | let Some(workspace) = items::repo_workspace(self, repo_id).await? else { |
| 256 | return Ok(()); |
| 257 | }; |
| 258 | for reference in refs::refs_in_text(text).into_iter().take(MAX_REFS) { |
| 259 | let Some((read_in, named)) = items::resolve_in(self, repo_id, &workspace, &reference).await? else { continue }; |
| 260 | match named { |
| 261 | Named::PullHere => self.link_pull(session_id, &read_in, reference.number, LinkVia::Ref).await, |
| 262 | Named::Item(id) => self.link_item(session_id, repo_id, &id, LinkVia::Ref).await, |
| 263 | Named::PullThere(_) => {} |
| 264 | } |
| 265 | } |
| 266 | Ok(()) |
| 267 | } |
| 268 | |
| 269 | /// Ties a session to the pull requests of `branch`. |
| 270 | pub(crate) async fn link_branch(&self, session_id: &str, repo_id: &str, branch: &str) -> Result<()> { |
| 271 | let pulls = self |
| 272 | .db |
| 273 | .prepare( |
| 274 | "SELECT number FROM pulls WHERE repo_id = ? AND source_branch = ? |
| 275 | AND fork_repo_id IS NULL AND head_repo_id IS NULL", |
| 276 | ) |
| 277 | .bind(&[repo_id.into(), branch.into()])? |
| 278 | .all() |
| 279 | .await? |
| 280 | .results::<NumberOnly>()?; |
| 281 | for pull in pulls { |
| 282 | self.link_pull(session_id, repo_id, pull.number, LinkVia::Branch).await; |
| 283 | } |
| 284 | Ok(()) |
| 285 | } |
| 286 | |
| 287 | /// Ties a session to the pull requests whose head is `commit` (a full |
| 288 | /// or abbreviated hash). |
| 289 | pub(crate) async fn link_commit(&self, session_id: &str, repo_id: &str, commit: &str) -> Result<()> { |
| 290 | let pulls = self |
| 291 | .db |
| 292 | .prepare("SELECT number FROM pulls WHERE repo_id = ? AND head_commit IS NOT NULL AND substr(head_commit, 1, length(?2)) = ?2") |
| 293 | .bind(&[repo_id.into(), commit.into()])? |
| 294 | .all() |
| 295 | .await? |
| 296 | .results::<NumberOnly>()?; |
| 297 | for pull in pulls { |
| 298 | self.link_pull(session_id, repo_id, pull.number, LinkVia::Commit).await; |
| 299 | } |
| 300 | Ok(()) |
| 301 | } |
| 302 | |
| 303 | /// Records what a session touched, ties it to what that names, and |
| 304 | /// tells it and the other sessions on the same files that they |
| 305 | /// overlap. Returns how many were new. |
| 306 | pub(crate) async fn record_touches(&self, session: &Session, touches: &[NewTouch]) -> Result<u32> { |
| 307 | let mut touches = tidy_touches(touches); |
| 308 | if touches.is_empty() { |
| 309 | return Ok(0); |
| 310 | } |
| 311 | // A work item is kept by its id: named by its number in the |
| 312 | // repository's workspace, one that is none is left out. |
| 313 | if touches.iter().any(|touch| touch.kind == TouchKind::Item) { |
| 314 | let workspace = items::repo_workspace(self, &session.repo_id).await?.unwrap_or_default(); |
| 315 | let mut kept = Vec::with_capacity(touches.len()); |
| 316 | for mut touch in touches { |
| 317 | if touch.kind == TouchKind::Item { |
| 318 | let Some(held) = items::by_number(self, &workspace, touch.value.parse().unwrap_or_default()).await? else { continue }; |
| 319 | touch.value = held.item.id; |
| 320 | } |
| 321 | kept.push(touch); |
| 322 | } |
| 323 | touches = kept; |
| 324 | } |
| 325 | let at = rfc3339(now_ms()); |
| 326 | let mut statements = Vec::with_capacity(touches.len()); |
| 327 | for touch in &touches { |
| 328 | statements.push(self.db.prepare(INSERT_TOUCH).bind(&[ |
| 329 | session.id.as_str().into(), |
| 330 | session.repo_id.as_str().into(), |
| 331 | touch.kind.as_str().into(), |
| 332 | touch.value.as_str().into(), |
| 333 | at.as_str().into(), |
| 334 | ])?); |
| 335 | } |
| 336 | let mut new: Vec<NewTouch> = Vec::new(); |
| 337 | for result in self.db.batch(statements).await? { |
| 338 | for row in result.results::<serde_json::Value>()? { |
| 339 | if let (Some(kind), Some(value)) = (row["kind"].as_str().and_then(TouchKind::parse), row["value"].as_str()) { |
| 340 | new.push(NewTouch { kind, value: value.to_owned() }); |
| 341 | } |
| 342 | } |
| 343 | } |
| 344 | let mut files = Vec::new(); |
| 345 | for touch in &new { |
| 346 | let number = touch.value.parse::<u32>().ok(); |
| 347 | match (touch.kind, number) { |
| 348 | (TouchKind::Pull, Some(number)) => self.link_pull(&session.id, &session.repo_id, number, LinkVia::Opened).await, |
| 349 | (TouchKind::Item, _) => self.link_item(&session.id, &session.repo_id, &touch.value, LinkVia::Ref).await, |
| 350 | (TouchKind::Branch, _) => self.link_branch(&session.id, &session.repo_id, &touch.value).await?, |
| 351 | (TouchKind::Commit, _) => self.link_commit(&session.id, &session.repo_id, &touch.value).await?, |
| 352 | (TouchKind::File, _) => files.push(touch.value.clone()), |
| 353 | _ => {} |
| 354 | } |
| 355 | } |
| 356 | if !files.is_empty() { |
| 357 | self.announce_session_overlaps(session, &files).await?; |
| 358 | } |
| 359 | Ok(new.len() as u32) |
| 360 | } |
| 361 | |
| 362 | /// The ids of a session's line: its parents and its subagents. |
| 363 | pub(crate) async fn family(&self, session_id: &str) -> Result<HashSet<String>> { |
| 364 | Ok(self |
| 365 | .db |
| 366 | .prepare(FAMILY) |
| 367 | .bind(&[session_id.into()])? |
| 368 | .all() |
| 369 | .await? |
| 370 | .results::<IdRow>()? |
| 371 | .into_iter() |
| 372 | .map(|row| row.id) |
| 373 | .collect()) |
| 374 | } |
| 375 | |
| 376 | /// Tells `session` and each other active session that touched one of |
| 377 | /// `files` (outside its own line) that they overlap, once per pair. |
| 378 | async fn announce_session_overlaps(&self, session: &Session, files: &[String]) -> Result<()> { |
| 379 | let touched: Vec<(String, String)> = self |
| 380 | .db |
| 381 | .prepare(TOUCHED_BY_OTHERS) |
| 382 | .bind(&[session.repo_id.as_str().into(), serde_json::to_string(files)?.into(), session.id.as_str().into()])? |
| 383 | .all() |
| 384 | .await? |
| 385 | .results::<PairRow>()? |
| 386 | .into_iter() |
| 387 | .map(|row| (row.session_id, row.value)) |
| 388 | .collect(); |
| 389 | if touched.is_empty() { |
| 390 | return Ok(()); |
| 391 | } |
| 392 | let family = self.family(&session.id).await?; |
| 393 | let at = rfc3339(now_ms()); |
| 394 | for (other_id, files) in session_overlaps(&session.id, &family, &touched) { |
| 395 | let (low, high) = session_pair(&session.id, &other_id); |
| 396 | let Some(other) = self.session_row(&session.repo_id, &other_id).await? else { |
| 397 | continue; |
| 398 | }; |
| 399 | let other = other.at(now_ms()); |
| 400 | let ours = SessionOverlapDetected { |
| 401 | repo_id: session.repo_id.clone(), |
| 402 | number: session.pull.or(session.item), |
| 403 | session_id: session.id.clone(), |
| 404 | other_session_id: other.id.clone(), |
| 405 | other_actor: other.actor.username.clone(), |
| 406 | files: files.clone(), |
| 407 | item: session.item, |
| 408 | other_item: other.item, |
| 409 | }; |
| 410 | let theirs = SessionOverlapDetected { |
| 411 | number: other.pull.or(other.item), |
| 412 | session_id: other.id.clone(), |
| 413 | other_session_id: session.id.clone(), |
| 414 | other_actor: session.actor.username.clone(), |
| 415 | item: other.item, |
| 416 | other_item: session.item, |
| 417 | ..ours.clone() |
| 418 | }; |
| 419 | // The pair is noticed once, and each side told, in one change. |
| 420 | let mut statements = vec![self |
| 421 | .db |
| 422 | .prepare(INSERT_OVERLAP) |
| 423 | .bind(&[session.repo_id.as_str().into(), low.as_str().into(), high.as_str().into(), serde_json::to_string(&files)?.into(), at.as_str().into()])?]; |
| 424 | let guard = "EXISTS (SELECT 1 FROM session_overlaps WHERE repo_id = ?7 AND low = ?8 AND high = ?9 AND noticed_at = ?10)"; |
| 425 | let mut sent = Vec::new(); |
| 426 | for event in [&ours, &theirs] { |
| 427 | let delivery = self.delivery( |
| 428 | "overlap.detected", |
| 429 | &session.repo_id, |
| 430 | serde_json::to_value(event)?, |
| 431 | &format!("{low}:{high}:{at}:{}", event.session_id), |
| 432 | guard, |
| 433 | vec![session.repo_id.as_str().into(), low.as_str().into(), high.as_str().into(), at.as_str().into()], |
| 434 | )?; |
| 435 | sent.push(push_delivery(&mut statements, delivery)); |
| 436 | } |
| 437 | let new = match self.db.batch(statements).await?.into_iter().next() { |
| 438 | Some(result) => !result.results::<serde_json::Value>()?.is_empty(), |
| 439 | None => false, |
| 440 | }; |
| 441 | if !new { |
| 442 | continue; |
| 443 | } |
| 444 | let actor = Some(session.actor.id.clone()); |
| 445 | for (event, sent) in [ours, theirs].into_iter().zip(sent) { |
| 446 | self.delivered(sent, "overlap.detected", &session.repo_id, actor.clone(), serde_json::to_value(event)?).await?; |
| 447 | } |
| 448 | } |
| 449 | Ok(()) |
| 450 | } |
| 451 | |
| 452 | /// A session's subagents, and theirs, in the order they started, each |
| 453 | /// with how many it started. |
| 454 | pub(crate) async fn subagents_of(&self, session_id: &str, now: u64) -> Result<Vec<Session>> { |
| 455 | let rows = self.db.prepare(tree_sql()).bind(&[session_id.into()])?.all().await?.results::<SessionRow>()?; |
| 456 | let mut children: Vec<Session> = rows.into_iter().map(|row| row.at(now)).collect(); |
| 457 | count_subagents(&mut children); |
| 458 | Ok(children) |
| 459 | } |
| 460 | |
| 461 | /// How many subagents each of `sessions` started, filled in. |
| 462 | pub(crate) async fn fill_subagents(&self, sessions: &mut [Session]) -> Result<()> { |
| 463 | if sessions.is_empty() { |
| 464 | return Ok(()); |
| 465 | } |
| 466 | let ids: Vec<&str> = sessions.iter().map(|session| session.id.as_str()).collect(); |
| 467 | #[derive(Deserialize)] |
| 468 | struct Count { |
| 469 | parent_id: String, |
| 470 | n: u32, |
| 471 | } |
| 472 | let counts = self |
| 473 | .db |
| 474 | .prepare("SELECT parent_id, COUNT(*) AS n FROM sessions WHERE parent_id IN (SELECT value FROM json_each(?)) GROUP BY parent_id") |
| 475 | .bind(&[serde_json::to_string(&ids)?.into()])? |
| 476 | .all() |
| 477 | .await? |
| 478 | .results::<Count>()?; |
| 479 | let counts: HashMap<String, u32> = counts.into_iter().map(|count| (count.parent_id, count.n)).collect(); |
| 480 | for session in sessions { |
| 481 | session.subagents = counts.get(&session.id).copied().unwrap_or_default(); |
| 482 | } |
| 483 | Ok(()) |
| 484 | } |
| 485 | |
| 486 | /// Gives each session with no title the start of the first prompt it |
| 487 | /// recorded, as `first_prompt`, for saying what it is for. |
| 488 | pub(crate) async fn fill_first_prompts(&self, sessions: &mut [Session]) -> Result<()> { |
| 489 | let ids: Vec<&str> = sessions.iter().filter(|session| session.title.is_none()).map(|session| session.id.as_str()).collect(); |
| 490 | if ids.is_empty() { |
| 491 | return Ok(()); |
| 492 | } |
| 493 | #[derive(Deserialize)] |
| 494 | struct First { |
| 495 | session_id: String, |
| 496 | text: String, |
| 497 | } |
| 498 | let found = self |
| 499 | .db |
| 500 | .prepare(FIRST_PROMPTS) |
| 501 | .bind(&[serde_json::to_string(&ids)?.into(), (MAX_FIRST_PROMPT_CHARS as u32).into()])? |
| 502 | .all() |
| 503 | .await? |
| 504 | .results::<First>()?; |
| 505 | let found: HashMap<String, String> = found.into_iter().map(|first| (first.session_id, first.text)).collect(); |
| 506 | for session in sessions.iter_mut().filter(|session| session.title.is_none()) { |
| 507 | session.first_prompt = found.get(&session.id).map(|text| text.trim().to_owned()).filter(|text| !text.is_empty()); |
| 508 | } |
| 509 | Ok(()) |
| 510 | } |
| 511 | |
| 512 | /// What a tree of sessions touched, its links, and the skills it used. |
| 513 | pub(crate) async fn tree_facts(&self, repo_id: &str, ids: &[String]) -> Result<(Vec<SessionTouch>, Vec<SessionLink>, Vec<SessionSkill>)> { |
| 514 | let ids_json = serde_json::to_string(ids)?; |
| 515 | // A work item touched is shown by its number. |
| 516 | let touches = self |
| 517 | .db |
| 518 | .prepare(concat!( |
| 519 | "SELECT session_id, kind, |
| 520 | CASE WHEN kind = 'item' |
| 521 | THEN COALESCE(CAST(", |
| 522 | items::sql::item_number!("session_touches.value"), |
| 523 | " AS TEXT), value) |
| 524 | ELSE value END AS value, |
| 525 | at FROM session_touches |
| 526 | WHERE session_id IN (SELECT value FROM json_each(?)) ORDER BY at, rowid LIMIT 1000", |
| 527 | )) |
| 528 | .bind(&[ids_json.as_str().into()])? |
| 529 | .all() |
| 530 | .await? |
| 531 | .results::<TouchRow>()?; |
| 532 | // A pull request opened on a branch after the session pushed it. |
| 533 | let branches: HashSet<(String, String)> = touches |
| 534 | .iter() |
| 535 | .filter(|touch| touch.kind == "branch") |
| 536 | .map(|touch| (touch.session_id.clone(), touch.value.clone())) |
| 537 | .collect(); |
| 538 | for (session_id, branch) in branches.iter().take(20) { |
| 539 | self.link_branch(session_id, repo_id, branch).await?; |
| 540 | } |
| 541 | let records: Vec<LinkRecord> = self |
| 542 | .db |
| 543 | .prepare(concat!( |
| 544 | "SELECT session_id, kind, |
| 545 | CASE WHEN kind = 'item' THEN ", |
| 546 | items::sql::item_number!("session_links.item_id"), |
| 547 | " |
| 548 | ELSE number END AS number, |
| 549 | via FROM session_links |
| 550 | WHERE session_id IN (SELECT value FROM json_each(?)) ORDER BY at", |
| 551 | )) |
| 552 | .bind(&[ids_json.as_str().into()])? |
| 553 | .all() |
| 554 | .await? |
| 555 | .results::<LinkRow>()? |
| 556 | .into_iter() |
| 557 | .filter_map(|row| { |
| 558 | let kind = match row.kind.as_str() { |
| 559 | "item" => LinkKind::Item, |
| 560 | "pull" => LinkKind::Pull, |
| 561 | _ => return None, |
| 562 | }; |
| 563 | Some(LinkRecord { session_id: row.session_id, kind, number: row.number?, via: LinkVia::parse(&row.via)? }) |
| 564 | }) |
| 565 | .collect(); |
| 566 | let workspace = items::repo_workspace(self, repo_id).await?.unwrap_or_default(); |
| 567 | let mut known = HashMap::new(); |
| 568 | let mut wanted: Vec<(LinkKind, u32)> = Vec::new(); |
| 569 | for record in &records { |
| 570 | if !wanted.contains(&(record.kind, record.number)) { |
| 571 | wanted.push((record.kind, record.number)); |
| 572 | } |
| 573 | } |
| 574 | for (kind, number) in wanted.into_iter().take(MAX_LINKED) { |
| 575 | let found = match kind { |
| 576 | LinkKind::Item => items::by_number(self, &workspace, number).await?.map(|held| (held.item.title, state_name(&held.item.state))), |
| 577 | LinkKind::Pull => self.pull(repo_id, number).await?.map(|pull| (pull.title, state_name(&pull.status))), |
| 578 | }; |
| 579 | if let Some(found) = found { |
| 580 | known.insert((kind, number), found); |
| 581 | } |
| 582 | } |
| 583 | let links = gather_links(&records, &known); |
| 584 | let skills = self |
| 585 | .db |
| 586 | .prepare( |
| 587 | "SELECT session_id, text AS name, COUNT(*) AS uses FROM session_entries |
| 588 | WHERE session_id IN (SELECT value FROM json_each(?)) AND kind = 'skill' |
| 589 | GROUP BY session_id, text ORDER BY MIN(session_seq)", |
| 590 | ) |
| 591 | .bind(&[ids_json.as_str().into()])? |
| 592 | .all() |
| 593 | .await? |
| 594 | .results::<SkillRow>()? |
| 595 | .into_iter() |
| 596 | .map(|row| SessionSkill { session_id: row.session_id, name: row.name, uses: row.uses }) |
| 597 | .collect(); |
| 598 | let touched = touches |
| 599 | .into_iter() |
| 600 | .filter_map(|row| { |
| 601 | Some(SessionTouch { session_id: row.session_id, kind: TouchKind::parse(&row.kind)?, value: row.value, at: row.at }) |
| 602 | }) |
| 603 | .collect(); |
| 604 | Ok((touched, links, skills)) |
| 605 | } |
| 606 | |
| 607 | /// After a push moved `pulls` on `branch` to `after`: the sessions that |
| 608 | /// pushed that branch, and those its new commits name in |
| 609 | /// `G1t-Session:` trailers, are tied to them. Never fails the push's |
| 610 | /// handling: a problem is logged. |
| 611 | pub(crate) async fn link_push(&self, repo_id: &str, branch: &str, after: &str, commits: Option<u32>, pulls: &[Pull]) { |
| 612 | if pulls.is_empty() { |
| 613 | return; |
| 614 | } |
| 615 | let linked = async { |
| 616 | let pushed_by = self |
| 617 | .db |
| 618 | .prepare("SELECT DISTINCT session_id AS id FROM session_touches WHERE repo_id = ? AND kind = 'branch' AND value = ?") |
| 619 | .bind(&[repo_id.into(), branch.into()])? |
| 620 | .all() |
| 621 | .await? |
| 622 | .results::<IdRow>()?; |
| 623 | for session in &pushed_by { |
| 624 | for pull in pulls { |
| 625 | self.link_pull(&session.id, repo_id, pull.number, LinkVia::Branch).await; |
| 626 | } |
| 627 | } |
| 628 | // Trailers are read only while someone records sessions here. |
| 629 | let recording = self |
| 630 | .db |
| 631 | .prepare("SELECT id FROM sessions WHERE repo_id = ? AND state = 'active' LIMIT 1") |
| 632 | .bind(&[repo_id.into()])? |
| 633 | .first::<IdRow>(None) |
| 634 | .await?; |
| 635 | if recording.is_none() { |
| 636 | return Ok::<(), worker::Error>(()); |
| 637 | } |
| 638 | let Some(pull) = pulls.first() else { return Ok(()) }; |
| 639 | let path: Option<RepoPath> = |
| 640 | g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: repo_id.to_owned() }).await?; |
| 641 | let Some(path) = path else { return Ok(()) }; |
| 642 | let log = LogArgs { |
| 643 | path, |
| 644 | viewer: self.owner_viewer(pull).await?, |
| 645 | git_ref: Some(after.to_owned()), |
| 646 | limit: commits.unwrap_or(MAX_TRAILER_COMMITS).clamp(1, MAX_TRAILER_COMMITS), |
| 647 | }; |
| 648 | let Outcome::Ok(history) = g1t_kit::call::<_, Outcome<Vec<Commit>>>(&self.repos, "log", &log).await? else { |
| 649 | return Ok(()); |
| 650 | }; |
| 651 | let mut named: Vec<String> = Vec::new(); |
| 652 | for commit in &history { |
| 653 | for id in session_trailers(&commit.message) { |
| 654 | if !named.contains(&id) { |
| 655 | named.push(id); |
| 656 | } |
| 657 | } |
| 658 | } |
| 659 | for id in named { |
| 660 | if self.session_row(repo_id, &id).await?.is_none() { |
| 661 | continue; |
| 662 | } |
| 663 | for pull in pulls { |
| 664 | self.link_pull(&id, repo_id, pull.number, LinkVia::Trailer).await; |
| 665 | } |
| 666 | } |
| 667 | Ok(()) |
| 668 | }; |
| 669 | if let Err(error) = linked.await { |
| 670 | worker::console_error!("push to {branch}: sessions not tied to its pull requests: {error}"); |
| 671 | } |
| 672 | } |
| 673 | |
| 674 | /// Sessions tied to a repository's pull request by a link, by id. |
| 675 | pub(crate) async fn linked_sessions(&self, repo_id: &str, number: u32) -> Result<Vec<String>> { |
| 676 | Ok(self |
| 677 | .db |
| 678 | .prepare("SELECT DISTINCT session_id AS id FROM session_links WHERE repo_id = ? AND kind = 'pull' AND number = ? LIMIT 200") |
| 679 | .bind(&[repo_id.into(), number.into()])? |
| 680 | .all() |
| 681 | .await? |
| 682 | .results::<IdRow>()? |
| 683 | .into_iter() |
| 684 | .map(|row| row.id) |
| 685 | .collect()) |
| 686 | } |
| 687 | |
| 688 | /// A repository's sessions tied to a work item by a link, by id. |
| 689 | pub(crate) async fn item_sessions(&self, repo_id: &str, item_id: &str) -> Result<Vec<String>> { |
| 690 | Ok(self |
| 691 | .db |
| 692 | .prepare("SELECT DISTINCT session_id AS id FROM session_links WHERE repo_id = ? AND kind = 'item' AND item_id = ? LIMIT 200") |
| 693 | .bind(&[repo_id.into(), item_id.into()])? |
| 694 | .all() |
| 695 | .await? |
| 696 | .results::<IdRow>()? |
| 697 | .into_iter() |
| 698 | .map(|row| row.id) |
| 699 | .collect()) |
| 700 | } |
| 701 | |
| 702 | /// Sessions that pushed one of `pulls`' branches, by id: tied to them |
| 703 | /// even when the pull request was opened after the push. |
| 704 | pub(crate) async fn branch_sessions(&self, repo_id: &str, pulls: &[u32]) -> Result<Vec<String>> { |
| 705 | if pulls.is_empty() { |
| 706 | return Ok(Vec::new()); |
| 707 | } |
| 708 | Ok(self |
| 709 | .db |
| 710 | .prepare(BRANCH_SESSIONS) |
| 711 | .bind(&[repo_id.into(), serde_json::to_string(pulls)?.into()])? |
| 712 | .all() |
| 713 | .await? |
| 714 | .results::<IdRow>()? |
| 715 | .into_iter() |
| 716 | .map(|row| row.id) |
| 717 | .collect()) |
| 718 | } |
| 719 | |
| 720 | /// Sessions by id, as they read now. |
| 721 | pub(crate) async fn sessions_by_id(&self, repo_id: &str, ids: &[String]) -> Result<Vec<Session>> { |
| 722 | if ids.is_empty() { |
| 723 | return Ok(Vec::new()); |
| 724 | } |
| 725 | let now = now_ms(); |
| 726 | Ok(self |
| 727 | .db |
| 728 | .prepare(format!( |
| 729 | "SELECT {SESSION_COLUMNS} FROM sessions WHERE repo_id = ? AND id IN (SELECT value FROM json_each(?))" |
| 730 | )) |
| 731 | .bind(&[repo_id.into(), serde_json::to_string(ids)?.into()])? |
| 732 | .all() |
| 733 | .await? |
| 734 | .results::<SessionRow>()? |
| 735 | .into_iter() |
| 736 | .map(|row| row.at(now)) |
| 737 | .collect()) |
| 738 | } |
| 739 | } |
| 740 | |
| 741 | /// A work item's state or a pull request's status, as its JSON name. |
| 742 | fn state_name<T: serde::Serialize>(state: &T) -> String { |
| 743 | serde_json::to_value(state).ok().and_then(|value| value.as_str().map(str::to_owned)).unwrap_or_default() |
| 744 | } |
| 745 | |
| 746 | /// Fills in how many subagents each session of a tree started, from the |
| 747 | /// tree itself. |
| 748 | pub(crate) fn count_subagents(sessions: &mut [Session]) { |
| 749 | let mut counts: HashMap<String, u32> = HashMap::new(); |
| 750 | for session in sessions.iter() { |
| 751 | if let Some(parent) = &session.parent_id { |
| 752 | *counts.entry(parent.clone()).or_default() += 1; |
| 753 | } |
| 754 | } |
| 755 | for session in sessions.iter_mut() { |
| 756 | session.subagents = counts.get(&session.id).copied().unwrap_or_default(); |
| 757 | } |
| 758 | } |
| 759 | |
| 760 | #[cfg(test)] |
| 761 | mod ref_tests { |
| 762 | use g1t_contracts::work::refs::{self, Ref, RefScope}; |
| 763 | |
| 764 | #[test] |
| 765 | fn a_session_reads_references_as_every_reader_does() { |
| 766 | // What link_refs reads: never inside code, and owner/name#N and |
| 767 | // workspace#N as what they name. |
| 768 | let found = refs::refs_in_text("Fixing `#5` in acme/web#7, see acme#9 and #123456789"); |
| 769 | let scopes: Vec<(String, u32)> = found |
| 770 | .iter() |
| 771 | .map(|Ref { scope, number }| { |
| 772 | let scope = match scope { |
| 773 | RefScope::Here => "here".to_owned(), |
| 774 | RefScope::Repo(path) => format!("{}/{}", path.namespace, path.name), |
| 775 | RefScope::Workspace(slug) => slug.clone(), |
| 776 | }; |
| 777 | (scope, *number) |
| 778 | }) |
| 779 | .collect(); |
| 780 | assert_eq!(scopes, vec![("acme/web".to_owned(), 7), ("acme".to_owned(), 9), ("here".to_owned(), 123_456_789)]); |
| 781 | } |
| 782 | } |