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