| 1 | //! Coordinating work between agents and people |
| 2 | //! (`g1t_contracts::work::coordination`): sessions of their own harnesses |
| 3 | //! on a repository (sessions.rs), advisory claims on work items |
| 4 | //! (claims.rs), what a work item is blocked by (dependencies.rs), pull |
| 5 | //! requests that start changing the same files (overlaps.rs), and the one |
| 6 | //! call that gathers it all for a work item (context.rs). |
| 7 | //! |
| 8 | //! Nothing here is a lock and nothing waits on a harness: sessions and |
| 9 | //! claims that went quiet are found when the repository's coordination is |
| 10 | //! next touched (`sweep`), and announced then as ended or expired. |
| 11 | |
| 12 | pub(crate) mod updates; |
| 13 | mod claims; |
| 14 | mod context; |
| 15 | mod dependencies; |
| 16 | mod links; |
| 17 | mod overlaps; |
| 18 | mod sessions; |
| 19 | pub(crate) mod shown; |
| 20 | pub(crate) mod titles; |
| 21 | mod workspace; |
| 22 | #[cfg(test)] |
| 23 | mod tests; |
| 24 | |
| 25 | use g1t_contracts::credentials::Principal; |
| 26 | use g1t_contracts::repos::{Repo, RepoPath}; |
| 27 | use g1t_contracts::time::{parse_rfc3339, rfc3339}; |
| 28 | use g1t_contracts::work::*; |
| 29 | use g1t_contracts::{FailureCode, Outcome, User, Viewer}; |
| 30 | use g1t_kit::now_ms; |
| 31 | use serde::Deserialize; |
| 32 | use worker::Result; |
| 33 | |
| 34 | use crate::Work; |
| 35 | |
| 36 | /// Unwraps an `Outcome`, returning its failure from the enclosing method. |
| 37 | macro_rules! unwrap_or_fail { |
| 38 | ($outcome:expr) => { |
| 39 | match $outcome { |
| 40 | g1t_contracts::Outcome::Ok(value) => value, |
| 41 | g1t_contracts::Outcome::Fail(failure) => return Ok(g1t_contracts::Outcome::Fail(failure)), |
| 42 | } |
| 43 | }; |
| 44 | } |
| 45 | pub(crate) use unwrap_or_fail; |
| 46 | |
| 47 | /// A session's columns, with its work item's number beside its id. |
| 48 | const SESSION_COLUMNS: &str = concat!( |
| 49 | "id, repo_id, item_id, |
| 50 | ", |
| 51 | crate::items::sql::item_number!("sessions.item_id"), |
| 52 | " AS item, |
| 53 | pull, branch, actor_id, actor_name, harness, harness_session_id, |
| 54 | title, state, head_commit, started_at, last_seen_at, ended_at, end_reason, |
| 55 | parent_id, agent_type, task, prompt, result, model, title_source, title_settled_at" |
| 56 | ); |
| 57 | |
| 58 | /// A claim's columns, with its work item's workspace and number. |
| 59 | const CLAIM_COLUMNS: &str = concat!( |
| 60 | "id, item_id, repo_id, |
| 61 | ", |
| 62 | crate::items::sql::item_workspace!("claims.item_id"), |
| 63 | " AS workspace, |
| 64 | ", |
| 65 | crate::items::sql::item_number!("claims.item_id"), |
| 66 | " AS item, |
| 67 | session_id, actor_id, actor_name, harness, note, state, |
| 68 | ttl_minutes, created_at, expires_at, ended_at, end_reason, handoff_to" |
| 69 | ); |
| 70 | |
| 71 | #[derive(Clone, Deserialize)] |
| 72 | pub(crate) struct SessionRow { |
| 73 | id: String, |
| 74 | repo_id: String, |
| 75 | #[serde(default)] |
| 76 | item_id: Option<String>, |
| 77 | #[serde(default)] |
| 78 | item: Option<u32>, |
| 79 | pull: Option<u32>, |
| 80 | branch: Option<String>, |
| 81 | actor_id: String, |
| 82 | actor_name: String, |
| 83 | harness: String, |
| 84 | harness_session_id: Option<String>, |
| 85 | title: Option<String>, |
| 86 | state: String, |
| 87 | head_commit: Option<String>, |
| 88 | started_at: String, |
| 89 | last_seen_at: String, |
| 90 | ended_at: Option<String>, |
| 91 | end_reason: Option<String>, |
| 92 | #[serde(default)] |
| 93 | parent_id: Option<String>, |
| 94 | #[serde(default)] |
| 95 | agent_type: Option<String>, |
| 96 | #[serde(default)] |
| 97 | task: Option<String>, |
| 98 | #[serde(default)] |
| 99 | prompt: Option<String>, |
| 100 | #[serde(default)] |
| 101 | result: Option<String>, |
| 102 | #[serde(default)] |
| 103 | model: Option<String>, |
| 104 | #[serde(default)] |
| 105 | title_source: Option<String>, |
| 106 | #[serde(default)] |
| 107 | title_settled_at: Option<String>, |
| 108 | } |
| 109 | |
| 110 | impl SessionRow { |
| 111 | /// The session as it reads at `now`. |
| 112 | pub(crate) fn at(self, now: u64) -> Session { |
| 113 | let last_seen = parse_rfc3339(&self.last_seen_at).unwrap_or(now); |
| 114 | Session { |
| 115 | state: SessionState::at(self.state == "ended", last_seen, now), |
| 116 | id: self.id, |
| 117 | repo_id: self.repo_id, |
| 118 | item: self.item, |
| 119 | pull: self.pull, |
| 120 | branch: self.branch, |
| 121 | actor: Principal { id: self.actor_id, username: self.actor_name }, |
| 122 | harness: Harness::parse(&self.harness).unwrap_or(Harness::Other), |
| 123 | harness_session_id: self.harness_session_id, |
| 124 | title: self.title, |
| 125 | head_commit: self.head_commit, |
| 126 | started_at: self.started_at, |
| 127 | last_seen_at: self.last_seen_at, |
| 128 | ended_at: self.ended_at, |
| 129 | end_reason: self.end_reason, |
| 130 | parent_id: self.parent_id, |
| 131 | agent_type: self.agent_type, |
| 132 | task: self.task, |
| 133 | prompt: self.prompt, |
| 134 | result: self.result, |
| 135 | model: self.model, |
| 136 | subagents: 0, |
| 137 | first_prompt: None, |
| 138 | title_source: self.title_source.as_deref().and_then(TitleSource::parse), |
| 139 | title_settled_at: self.title_settled_at, |
| 140 | } |
| 141 | } |
| 142 | } |
| 143 | |
| 144 | #[derive(Deserialize)] |
| 145 | pub(crate) struct ClaimRow { |
| 146 | id: String, |
| 147 | pub(crate) item_id: String, |
| 148 | #[serde(default)] |
| 149 | repo_id: Option<String>, |
| 150 | #[serde(default)] |
| 151 | workspace: Option<String>, |
| 152 | #[serde(default)] |
| 153 | item: Option<u32>, |
| 154 | session_id: Option<String>, |
| 155 | actor_id: String, |
| 156 | actor_name: String, |
| 157 | harness: Option<String>, |
| 158 | note: Option<String>, |
| 159 | state: String, |
| 160 | #[allow(dead_code)] |
| 161 | ttl_minutes: u32, |
| 162 | created_at: String, |
| 163 | expires_at: String, |
| 164 | ended_at: Option<String>, |
| 165 | end_reason: Option<String>, |
| 166 | handoff_to: Option<String>, |
| 167 | } |
| 168 | |
| 169 | impl From<ClaimRow> for Claim { |
| 170 | fn from(row: ClaimRow) -> Self { |
| 171 | Claim { |
| 172 | id: row.id, |
| 173 | repo_id: row.repo_id, |
| 174 | workspace: row.workspace.unwrap_or_default(), |
| 175 | item: row.item.unwrap_or_default(), |
| 176 | session_id: row.session_id, |
| 177 | actor: Principal { id: row.actor_id, username: row.actor_name }, |
| 178 | harness: row.harness.as_deref().and_then(Harness::parse), |
| 179 | note: row.note, |
| 180 | state: ClaimState::parse(&row.state).unwrap_or(ClaimState::Expired), |
| 181 | created_at: row.created_at, |
| 182 | expires_at: row.expires_at, |
| 183 | ended_at: row.ended_at, |
| 184 | end_reason: row.end_reason, |
| 185 | handoff_to: row.handoff_to, |
| 186 | } |
| 187 | } |
| 188 | } |
| 189 | |
| 190 | /// Text as given, trimmed and cut to `limit` characters; none when empty. |
| 191 | pub(crate) fn tidy(text: Option<&str>, limit: usize) -> Option<String> { |
| 192 | let text = text?.trim(); |
| 193 | (!text.is_empty()).then(|| text.chars().take(limit).collect()) |
| 194 | } |
| 195 | |
| 196 | /// The most sessions, and claims, one sweep ends; the rest wait for the next. |
| 197 | const SWEEP_LIMIT: u32 = 50; |
| 198 | |
| 199 | /// Ends a repository's (`?1`) session `?3` if last seen at or before `?2`, |
| 200 | /// as `timed_out`, as of when it was last seen; its title is settled a |
| 201 | /// last time (titles.rs). |
| 202 | pub(crate) fn time_out_sessions_sql() -> String { |
| 203 | format!( |
| 204 | "UPDATE sessions SET state = 'ended', ended_at = last_seen_at, end_reason = 'timed_out', {final_due} |
| 205 | WHERE repo_id = ?1 AND state = 'active' AND last_seen_at <= ?2 AND id = ?3 |
| 206 | RETURNING {SESSION_COLUMNS}", |
| 207 | final_due = titles::final_due("last_seen_at") |
| 208 | ) |
| 209 | } |
| 210 | |
| 211 | /// Expires active claim `?3` if due at or before `?2` (`?1` unused, so |
| 212 | /// the binds are the sessions'). |
| 213 | pub(crate) fn expire_claims_sql() -> String { |
| 214 | format!( |
| 215 | "UPDATE claims SET state = 'expired', ended_at = expires_at, end_reason = 'expired' |
| 216 | WHERE ?1 IS NOT NULL AND state = 'active' AND expires_at <= ?2 AND id = ?3 |
| 217 | RETURNING {CLAIM_COLUMNS}" |
| 218 | ) |
| 219 | } |
| 220 | |
| 221 | /// RFC 3339 for `minutes` after `now`. |
| 222 | pub(crate) fn after_minutes(now: u64, minutes: u32) -> String { |
| 223 | rfc3339(now + u64::from(minutes) * 60_000) |
| 224 | } |
| 225 | |
| 226 | impl Work { |
| 227 | /// The repository as `viewer` sees it, for reading its coordination. |
| 228 | pub(crate) async fn coordination_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> { |
| 229 | let repo = unwrap_or_fail!(self.repo(path, viewer).await?); |
| 230 | self.sweep(&repo.id).await?; |
| 231 | Ok(Outcome::Ok(repo)) |
| 232 | } |
| 233 | |
| 234 | /// The repository, if `actor` may take part in it: a confirmed email |
| 235 | /// address, a repository that is not archived, and a role that may |
| 236 | /// open work items and pull requests. |
| 237 | pub(crate) async fn participating_repo(&self, path: &RepoPath, actor: &User) -> Result<Outcome<Repo>> { |
| 238 | if !actor.verified { |
| 239 | return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED)); |
| 240 | } |
| 241 | let repo = unwrap_or_fail!(self.repo(path, &Some(actor.clone())).await?); |
| 242 | unwrap_or_fail!(crate::retired::writable(&repo)); |
| 243 | unwrap_or_fail!(crate::allowed(Some(actor), &repo, g1t_contracts::access::Capability::Participate)); |
| 244 | self.sweep(&repo.id).await?; |
| 245 | Ok(Outcome::Ok(repo)) |
| 246 | } |
| 247 | |
| 248 | /// Ends the repository's sessions that have gone 30 minutes without a |
| 249 | /// heartbeat (`timed_out`) and expires the claims made from it or on |
| 250 | /// work items linked to it past their time, and announces each. Run |
| 251 | /// before the repository's coordination is read or changed, so that |
| 252 | /// nobody sees a stale one as live. |
| 253 | pub(crate) async fn sweep(&self, repo_id: &str) -> Result<()> { |
| 254 | let now = now_ms(); |
| 255 | let quiet_since = rfc3339(now.saturating_sub(SESSION_TIMEOUT_MS)); |
| 256 | let at = rfc3339(now); |
| 257 | // Read first: most of the time nothing is due, and nothing is written. |
| 258 | let due = self |
| 259 | .db |
| 260 | .batch(vec![ |
| 261 | self.db |
| 262 | .prepare(format!( |
| 263 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 264 | WHERE repo_id = ?1 AND state = 'active' AND last_seen_at <= ?2 LIMIT ?3" |
| 265 | )) |
| 266 | .bind(&[repo_id.into(), quiet_since.as_str().into(), SWEEP_LIMIT.into()])?, |
| 267 | self.db |
| 268 | .prepare(format!( |
| 269 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 270 | WHERE state = 'active' AND expires_at <= ?2 AND (repo_id = ?1 OR item_id IN {linked}) |
| 271 | LIMIT ?3", |
| 272 | linked = crate::items::sql::items_linked_to!("?1"), |
| 273 | )) |
| 274 | .bind(&[repo_id.into(), at.as_str().into(), SWEEP_LIMIT.into()])?, |
| 275 | ]) |
| 276 | .await?; |
| 277 | let mut due = due.into_iter(); |
| 278 | let quiet: Vec<SessionRow> = match due.next() { |
| 279 | Some(result) => result.results()?, |
| 280 | None => Vec::new(), |
| 281 | }; |
| 282 | let lapsed: Vec<ClaimRow> = match due.next() { |
| 283 | Some(result) => result.results()?, |
| 284 | None => Vec::new(), |
| 285 | }; |
| 286 | if quiet.is_empty() && lapsed.is_empty() { |
| 287 | return Ok(()); |
| 288 | } |
| 289 | // Each one's end and its updates are one change: a sweep running at |
| 290 | // the same moment ends it, and delivers it, once. |
| 291 | let mut statements = Vec::new(); |
| 292 | let mut sessions = Vec::new(); |
| 293 | for row in quiet { |
| 294 | statements.push( |
| 295 | self.db |
| 296 | .prepare(time_out_sessions_sql()) |
| 297 | .bind(&[repo_id.into(), quiet_since.as_str().into(), row.id.as_str().into()])?, |
| 298 | ); |
| 299 | let index = statements.len() - 1; |
| 300 | let ended_at = row.last_seen_at.clone(); |
| 301 | let expected = SessionRow { |
| 302 | state: "ended".to_owned(), |
| 303 | ended_at: Some(ended_at.clone()), |
| 304 | end_reason: Some("timed_out".to_owned()), |
| 305 | ..row |
| 306 | } |
| 307 | .at(now); |
| 308 | let delivery = self.delivery( |
| 309 | "session.ended", |
| 310 | repo_id, |
| 311 | serde_json::to_value(SessionEvent::of(&expected))?, |
| 312 | &format!("{}:ended:{ended_at}", expected.id), |
| 313 | "EXISTS (SELECT 1 FROM sessions WHERE id = ?7 AND state = 'ended' AND ended_at = ?8)", |
| 314 | vec![expected.id.as_str().into(), ended_at.as_str().into()], |
| 315 | )?; |
| 316 | sessions.push((index, updates::push_delivery(&mut statements, delivery))); |
| 317 | } |
| 318 | let mut claims = Vec::new(); |
| 319 | for row in lapsed { |
| 320 | statements.push( |
| 321 | self.db.prepare(expire_claims_sql()).bind(&[repo_id.into(), at.as_str().into(), row.id.as_str().into()])?, |
| 322 | ); |
| 323 | let index = statements.len() - 1; |
| 324 | let held = Claim::from(row); |
| 325 | let expected = Claim { |
| 326 | state: ClaimState::Expired, |
| 327 | ended_at: Some(held.expires_at.clone()), |
| 328 | end_reason: Some("expired".to_owned()), |
| 329 | ..held |
| 330 | }; |
| 331 | let delivery = self.delivery( |
| 332 | "work.claim_expired", |
| 333 | repo_id, |
| 334 | serde_json::to_value(ClaimEvent::of(&expected, &[repo_id.to_owned()]))?, |
| 335 | &format!("{}:{}", expected.id, ClaimState::Expired.as_str()), |
| 336 | "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = 'expired')", |
| 337 | vec![expected.id.as_str().into()], |
| 338 | )?; |
| 339 | claims.push((index, updates::push_delivery(&mut statements, delivery))); |
| 340 | } |
| 341 | let results = self.db.batch(statements).await?; |
| 342 | let returned = |index: usize| results.get(index); |
| 343 | for (index, sent) in sessions { |
| 344 | let Some(row) = returned(index).map(|result| result.results::<SessionRow>()).transpose()?.and_then(|rows| rows.into_iter().next()) |
| 345 | else { |
| 346 | continue; |
| 347 | }; |
| 348 | let session = row.at(now); |
| 349 | self.delivered(sent, "session.ended", repo_id, None, serde_json::to_value(SessionEvent::of(&session))?).await?; |
| 350 | } |
| 351 | for (index, sent) in claims { |
| 352 | let Some(row) = returned(index).map(|result| result.results::<ClaimRow>()).transpose()?.and_then(|rows| rows.into_iter().next()) |
| 353 | else { |
| 354 | continue; |
| 355 | }; |
| 356 | let claim = Claim::from(row); |
| 357 | let data = serde_json::to_value(ClaimEvent::of(&claim, &[repo_id.to_owned()]))?; |
| 358 | self.delivered(sent, "work.claim_expired", repo_id, None, data).await?; |
| 359 | } |
| 360 | Ok(()) |
| 361 | } |
| 362 | |
| 363 | pub(crate) async fn session_row(&self, repo_id: &str, id: &str) -> Result<Option<SessionRow>> { |
| 364 | self.db |
| 365 | .prepare(format!("SELECT {SESSION_COLUMNS} FROM sessions WHERE repo_id = ? AND id = ?")) |
| 366 | .bind(&[repo_id.into(), id.into()])? |
| 367 | .first::<SessionRow>(None) |
| 368 | .await |
| 369 | } |
| 370 | |
| 371 | /// Active claims on a work item, oldest first. |
| 372 | pub(crate) async fn active_claims(&self, item_id: &str) -> Result<Vec<Claim>> { |
| 373 | Ok(self |
| 374 | .db |
| 375 | .prepare(format!( |
| 376 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 377 | WHERE item_id = ? AND state = 'active' ORDER BY created_at, id" |
| 378 | )) |
| 379 | .bind(&[item_id.into()])? |
| 380 | .all() |
| 381 | .await? |
| 382 | .results::<ClaimRow>()? |
| 383 | .into_iter() |
| 384 | .map(Claim::from) |
| 385 | .collect()) |
| 386 | } |
| 387 | } |