| 1 | //! Sessions: one run of someone's own harness on a repository. |
| 2 | //! |
| 3 | //! Starting one is idempotent on the harness's own session id, so a hook |
| 4 | //! that fires twice, or a harness that reconnects, gets the same session |
| 5 | //! back. Every update, and every batch of entries, is a heartbeat: it keeps |
| 6 | //! the session active, brings back one that timed out, and renews the |
| 7 | //! claims made from it. Only the person whose token started a session |
| 8 | //! changes it. |
| 9 | //! |
| 10 | //! A subagent is a session of its own with a parent (`parent_id`): a |
| 11 | //! Claude Code subagent, a Managed Agents thread. It starts on what its |
| 12 | //! parent is on unless it says otherwise, keeps the task, prompt and |
| 13 | //! result its parent gave and got (cut short), and once ended stays ended |
| 14 | //! even if a late hook starts it again: a subagent never comes back. Its |
| 15 | //! heartbeats keep its parent seen, and ending a session ends the |
| 16 | //! subagents it left going. |
| 17 | |
| 18 | use g1t_contracts::events::SessionAppended; |
| 19 | use g1t_contracts::repos::Repo; |
| 20 | use g1t_contracts::time::rfc3339; |
| 21 | use g1t_contracts::work::*; |
| 22 | use g1t_contracts::{FailureCode, Outcome, User, new_id}; |
| 23 | use g1t_kit::now_ms; |
| 24 | use serde::Deserialize; |
| 25 | use worker::Result; |
| 26 | use worker::wasm_bindgen::JsValue; |
| 27 | |
| 28 | use super::links::{count_subagents, end_tree_sql}; |
| 29 | use super::{SESSION_COLUMNS, SessionRow, after_minutes, tidy, unwrap_or_fail}; |
| 30 | use crate::Work; |
| 31 | use crate::rows::NumberRow; |
| 32 | |
| 33 | /// The most entries one look at a session returns. |
| 34 | const ENTRY_PAGE: u32 = 500; |
| 35 | /// The most sessions one list returns. |
| 36 | const LIST_LIMIT: u32 = 100; |
| 37 | const MAX_TITLE_CHARS: usize = 200; |
| 38 | const MAX_BRANCH_CHARS: usize = 255; |
| 39 | const MAX_ID_CHARS: usize = 200; |
| 40 | const MAX_COMMIT_CHARS: usize = 64; |
| 41 | const MAX_AGENT_CHARS: usize = 100; |
| 42 | const MAX_MODEL_CHARS: usize = 100; |
| 43 | /// How long one prompt entry may be before its `#N` go unread. |
| 44 | const MAX_REF_TEXT: usize = 8_000; |
| 45 | |
| 46 | fn not_found<T>() -> Outcome<T> { |
| 47 | Outcome::fail(FailureCode::NotFound, "Session not found.") |
| 48 | } |
| 49 | |
| 50 | fn text_or_null(value: &Option<String>) -> JsValue { |
| 51 | value.as_deref().map_or(JsValue::NULL, JsValue::from) |
| 52 | } |
| 53 | |
| 54 | fn number_or_null(value: Option<u32>) -> JsValue { |
| 55 | value.map_or(JsValue::NULL, JsValue::from) |
| 56 | } |
| 57 | |
| 58 | /// Long text as kept: trimmed, cut to `limit` characters with an ellipsis. |
| 59 | fn clip(text: Option<&str>, limit: usize) -> Option<String> { |
| 60 | let text = text?.trim(); |
| 61 | if text.is_empty() { |
| 62 | return None; |
| 63 | } |
| 64 | if text.chars().count() <= limit { |
| 65 | return Some(text.to_owned()); |
| 66 | } |
| 67 | let cut: String = text.chars().take(limit.saturating_sub(1)).collect(); |
| 68 | Some(format!("{}…", cut.trim_end())) |
| 69 | } |
| 70 | |
| 71 | /// What a session's heartbeat may change, tidied. |
| 72 | #[derive(Default)] |
| 73 | struct Fields { |
| 74 | issue: Option<u32>, |
| 75 | pull: Option<u32>, |
| 76 | branch: Option<String>, |
| 77 | head_commit: Option<String>, |
| 78 | title: Option<String>, |
| 79 | agent_type: Option<String>, |
| 80 | task: Option<String>, |
| 81 | prompt: Option<String>, |
| 82 | model: Option<String>, |
| 83 | result: Option<String>, |
| 84 | } |
| 85 | |
| 86 | impl Fields { |
| 87 | fn binds(&self) -> Vec<JsValue> { |
| 88 | vec![ |
| 89 | number_or_null(self.issue), |
| 90 | number_or_null(self.pull), |
| 91 | text_or_null(&self.branch), |
| 92 | text_or_null(&self.head_commit), |
| 93 | text_or_null(&self.title), |
| 94 | text_or_null(&self.agent_type), |
| 95 | text_or_null(&self.task), |
| 96 | text_or_null(&self.prompt), |
| 97 | text_or_null(&self.model), |
| 98 | text_or_null(&self.result), |
| 99 | ] |
| 100 | } |
| 101 | |
| 102 | /// The text whose `#N` tie the session to issues and pull requests. |
| 103 | fn named(&self) -> String { |
| 104 | [&self.title, &self.task, &self.prompt].into_iter().flatten().cloned().collect::<Vec<_>>().join("\n") |
| 105 | } |
| 106 | } |
| 107 | |
| 108 | #[derive(Deserialize)] |
| 109 | struct RenewRow { |
| 110 | id: String, |
| 111 | ttl_minutes: u32, |
| 112 | expires_at: String, |
| 113 | } |
| 114 | |
| 115 | /// Starts a session; nothing when one with the same harness session id |
| 116 | /// was started already. |
| 117 | pub(crate) fn insert_session_sql() -> String { |
| 118 | format!( |
| 119 | "INSERT INTO sessions |
| 120 | (id, repo_id, issue, pull, branch, actor_id, actor_name, harness, harness_session_id, |
| 121 | title, state, head_commit, started_at, last_seen_at, parent_id, agent_type, task, prompt, model) |
| 122 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 'active', ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18) |
| 123 | ON CONFLICT (repo_id, actor_id, harness, harness_session_id) |
| 124 | WHERE harness_session_id IS NOT NULL DO NOTHING |
| 125 | RETURNING {SESSION_COLUMNS}" |
| 126 | ) |
| 127 | } |
| 128 | |
| 129 | /// A person's session by its harness's id for it. |
| 130 | pub(crate) fn harness_session_sql() -> String { |
| 131 | format!( |
| 132 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 133 | WHERE repo_id = ?1 AND actor_id = ?2 AND harness = ?3 AND harness_session_id = ?4" |
| 134 | ) |
| 135 | } |
| 136 | |
| 137 | /// The fields `?1`–`?10` of [`Fields::binds`], each kept when not given. |
| 138 | const SET_FIELDS: &str = "issue = COALESCE(?1, issue), pull = COALESCE(?2, pull), branch = COALESCE(?3, branch), |
| 139 | head_commit = COALESCE(?4, head_commit), title = COALESCE(?5, title), |
| 140 | agent_type = COALESCE(?6, agent_type), task = COALESCE(?7, task), prompt = COALESCE(?8, prompt), |
| 141 | model = COALESCE(?9, model), result = COALESCE(?10, result)"; |
| 142 | |
| 143 | /// A heartbeat: seen now (`?11`), active again, with whichever fields are |
| 144 | /// given, of session `?12`. |
| 145 | pub(crate) fn resume_sql() -> String { |
| 146 | format!( |
| 147 | "UPDATE sessions |
| 148 | SET {SET_FIELDS}, |
| 149 | last_seen_at = ?11, state = 'active', ended_at = NULL, end_reason = NULL |
| 150 | WHERE id = ?12 |
| 151 | RETURNING {SESSION_COLUMNS}" |
| 152 | ) |
| 153 | } |
| 154 | |
| 155 | /// The fields of session `?12`, changed without bringing it back: for a |
| 156 | /// subagent's late hooks (`?11` unused, so the binds are the same). |
| 157 | pub(crate) fn refresh_sql() -> String { |
| 158 | format!( |
| 159 | "UPDATE sessions SET {SET_FIELDS} |
| 160 | WHERE id = ?12 AND ?11 IS NOT NULL |
| 161 | RETURNING {SESSION_COLUMNS}" |
| 162 | ) |
| 163 | } |
| 164 | |
| 165 | /// A subagent's heartbeat keeps its parent `?2` seen at `?1`. |
| 166 | pub(crate) const SEE_PARENT: &str = "UPDATE sessions SET last_seen_at = ?1 WHERE id = ?2 AND state = 'active' AND last_seen_at < ?1"; |
| 167 | |
| 168 | /// Records one entry in session `?7`, and in pull request `?1` when it |
| 169 | /// is on one. Each takes the next number of the session, and of its pull |
| 170 | /// request, itself, so two batches at once cannot collide; an entry whose |
| 171 | /// key (`?8`) the session has recorded already is skipped. |
| 172 | pub(crate) const INSERT_ENTRY: &str = "INSERT INTO session_entries |
| 173 | (pull_id, seq, kind, text, tool, \"commit\", at, session_id, session_seq, key) |
| 174 | SELECT ?1, |
| 175 | CASE WHEN ?1 IS NULL |
| 176 | THEN (SELECT COALESCE(MAX(session_seq), 0) + 1 FROM session_entries WHERE session_id = ?7) |
| 177 | ELSE (SELECT COALESCE(MAX(seq), 0) + 1 FROM session_entries WHERE pull_id = ?1) END, |
| 178 | ?2, ?3, ?4, ?5, ?6, ?7, |
| 179 | (SELECT COALESCE(MAX(session_seq), 0) + 1 FROM session_entries WHERE session_id = ?7), |
| 180 | ?8 |
| 181 | WHERE true |
| 182 | ON CONFLICT (session_id, key) WHERE session_id IS NOT NULL AND key IS NOT NULL DO NOTHING |
| 183 | RETURNING session_seq AS n"; |
| 184 | |
| 185 | /// Entries of session `?3` recorded before it named its pull request |
| 186 | /// `?1` join it, numbered after its last (`?2`), in their order. |
| 187 | pub(crate) const MOVE_ENTRIES: &str = "UPDATE session_entries SET pull_id = ?1, seq = ?2 + session_seq |
| 188 | WHERE session_id = ?3 AND pull_id IS NULL"; |
| 189 | |
| 190 | /// Renews active claim `?2` to `?1`. |
| 191 | pub(crate) const RENEW_CLAIM: &str = "UPDATE claims SET expires_at = ?1 WHERE id = ?2 AND state = 'active'"; |
| 192 | |
| 193 | /// Marks session `?2` as announced at `?1` when it was last announced at |
| 194 | /// or before `?3` (or never): a row back means it is time to announce. |
| 195 | pub(crate) const MARK_PROGRESS: &str = "UPDATE sessions SET progressed_at = ?1 |
| 196 | WHERE id = ?2 AND (progressed_at IS NULL OR progressed_at <= ?3) RETURNING progressed_at"; |
| 197 | |
| 198 | /// When a claim of `minutes` renewed at `now` lapses, if that is later |
| 199 | /// than it does already (`expires_at`): a heartbeat never shortens one. |
| 200 | pub(crate) fn renewal(now: u64, minutes: u32, expires_at: &str) -> Option<String> { |
| 201 | let renewed = after_minutes(now, minutes); |
| 202 | (renewed.as_str() > expires_at).then_some(renewed) |
| 203 | } |
| 204 | |
| 205 | /// Whether a session is a subagent that ended on purpose: it stays ended. |
| 206 | fn finished_subagent(row: &SessionRow) -> bool { |
| 207 | row.parent_id.is_some() && row.state == "ended" && row.end_reason.as_deref() != Some("timed_out") |
| 208 | } |
| 209 | |
| 210 | impl Work { |
| 211 | /// The issue and pull request a session names, checked: each is one of |
| 212 | /// the repository's, of the right kind. A pull request for an issue |
| 213 | /// names that issue when no issue was given. |
| 214 | async fn session_subjects(&self, repo: &Repo, issue: Option<u32>, pull: Option<u32>) -> Result<Outcome<(Option<u32>, Option<u32>)>> { |
| 215 | let mut issue = issue.filter(|number| *number > 0); |
| 216 | let pull = pull.filter(|number| *number > 0); |
| 217 | if let Some(number) = issue |
| 218 | && self.issue(&repo.id, number).await?.is_none() |
| 219 | { |
| 220 | return Ok(Outcome::fail(FailureCode::Invalid, format!("#{number} is not an issue in this repository."))); |
| 221 | } |
| 222 | if let Some(number) = pull { |
| 223 | match self.pull(&repo.id, number).await? { |
| 224 | Some(found) => issue = issue.or(found.issue), |
| 225 | None => { |
| 226 | return Ok(Outcome::fail(FailureCode::Invalid, format!("#{number} is not a pull request in this repository."))); |
| 227 | } |
| 228 | } |
| 229 | } |
| 230 | Ok(Outcome::Ok((issue, pull))) |
| 231 | } |
| 232 | |
| 233 | /// Ties a session to what it said it is on, and what its text names. |
| 234 | async fn link_fields(&self, session: &Session, fields: &Fields) -> Result<()> { |
| 235 | if let Some(number) = fields.issue { |
| 236 | self.link(&session.id, &session.repo_id, LinkKind::Issue, number, LinkVia::Attached).await; |
| 237 | } |
| 238 | if let Some(number) = fields.pull { |
| 239 | self.link(&session.id, &session.repo_id, LinkKind::Pull, number, LinkVia::Attached).await; |
| 240 | } |
| 241 | if let Some(branch) = &fields.branch { |
| 242 | self.link_branch(&session.id, &session.repo_id, branch).await?; |
| 243 | } |
| 244 | let named = fields.named(); |
| 245 | if !named.is_empty() { |
| 246 | self.link_refs(&session.id, &session.repo_id, &named).await?; |
| 247 | } |
| 248 | Ok(()) |
| 249 | } |
| 250 | |
| 251 | pub(crate) async fn start_session(&self, a: StartSessionArgs) -> Result<Outcome<Session>> { |
| 252 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 253 | // A subagent's parent: one of the actor's sessions here. |
| 254 | let parent = match a.parent_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) { |
| 255 | Some(id) => match self.session_row(&repo.id, id).await? { |
| 256 | Some(row) if row.actor_id == a.actor.id => Some(row), |
| 257 | _ => return Ok(Outcome::fail(FailureCode::Invalid, "parent_id is not one of your sessions in this repository.")), |
| 258 | }, |
| 259 | None => None, |
| 260 | }; |
| 261 | let (issue, pull) = unwrap_or_fail!(self.session_subjects(&repo, a.issue, a.pull).await?); |
| 262 | let mut fields = Fields { |
| 263 | issue, |
| 264 | pull, |
| 265 | branch: tidy(a.branch.as_deref(), MAX_BRANCH_CHARS), |
| 266 | head_commit: tidy(a.head_commit.as_deref(), MAX_COMMIT_CHARS), |
| 267 | title: tidy(a.title.as_deref(), MAX_TITLE_CHARS), |
| 268 | agent_type: tidy(a.agent_type.as_deref(), MAX_AGENT_CHARS), |
| 269 | task: clip(a.task.as_deref(), MAX_TASK_CHARS), |
| 270 | prompt: clip(a.prompt.as_deref(), MAX_PROMPT_CHARS), |
| 271 | model: tidy(a.model.as_deref(), MAX_MODEL_CHARS), |
| 272 | result: None, |
| 273 | }; |
| 274 | let harness_session_id = tidy(a.harness_session_id.as_deref(), MAX_ID_CHARS); |
| 275 | let now = now_ms(); |
| 276 | let at = rfc3339(now); |
| 277 | // The same harness session again: the one already started. |
| 278 | if let Some(harness_id) = &harness_session_id |
| 279 | && let Some(row) = self.harness_session(&repo.id, &a.actor, a.harness, harness_id).await? |
| 280 | { |
| 281 | return self.resume(&repo, row, &a.actor, fields).await; |
| 282 | } |
| 283 | // A subagent works on what its parent does unless it says. |
| 284 | if let Some(parent) = &parent { |
| 285 | fields.issue = fields.issue.or(parent.issue); |
| 286 | fields.pull = fields.pull.or(parent.pull); |
| 287 | fields.branch = fields.branch.clone().or(parent.branch.clone()); |
| 288 | fields.head_commit = fields.head_commit.clone().or(parent.head_commit.clone()); |
| 289 | } |
| 290 | let parent_id = parent.as_ref().map(|parent| parent.id.clone()); |
| 291 | let inserted = self |
| 292 | .db |
| 293 | .prepare(insert_session_sql()) |
| 294 | .bind(&[ |
| 295 | new_id("ses", now).into(), |
| 296 | repo.id.as_str().into(), |
| 297 | number_or_null(fields.issue), |
| 298 | number_or_null(fields.pull), |
| 299 | text_or_null(&fields.branch), |
| 300 | a.actor.id.as_str().into(), |
| 301 | a.actor.username.as_str().into(), |
| 302 | a.harness.as_str().into(), |
| 303 | text_or_null(&harness_session_id), |
| 304 | text_or_null(&fields.title), |
| 305 | text_or_null(&fields.head_commit), |
| 306 | at.as_str().into(), |
| 307 | at.as_str().into(), |
| 308 | text_or_null(&parent_id), |
| 309 | text_or_null(&fields.agent_type), |
| 310 | text_or_null(&fields.task), |
| 311 | text_or_null(&fields.prompt), |
| 312 | text_or_null(&fields.model), |
| 313 | ])? |
| 314 | .first::<SessionRow>(None) |
| 315 | .await?; |
| 316 | let Some(row) = inserted else { |
| 317 | // Started at the same moment by another call: that one. |
| 318 | let Some(harness_id) = &harness_session_id else { |
| 319 | return Ok(not_found()); |
| 320 | }; |
| 321 | return match self.harness_session(&repo.id, &a.actor, a.harness, harness_id).await? { |
| 322 | Some(row) => Ok(Outcome::Ok(row.at(now))), |
| 323 | None => Ok(not_found()), |
| 324 | }; |
| 325 | }; |
| 326 | let session = row.at(now); |
| 327 | self.link_fields(&session, &fields).await?; |
| 328 | if let Some(parent) = &parent_id { |
| 329 | self.db.prepare(SEE_PARENT).bind(&[at.as_str().into(), parent.as_str().into()])?.run().await?; |
| 330 | } |
| 331 | self.publish("session.started", &repo.id, &a.actor, SessionEvent::of(&session)).await?; |
| 332 | Ok(Outcome::Ok(session)) |
| 333 | } |
| 334 | |
| 335 | async fn harness_session(&self, repo_id: &str, actor: &User, harness: Harness, harness_id: &str) -> Result<Option<SessionRow>> { |
| 336 | self.db |
| 337 | .prepare(harness_session_sql()) |
| 338 | .bind(&[repo_id.into(), actor.id.as_str().into(), harness.as_str().into(), harness_id.into()])? |
| 339 | .first::<SessionRow>(None) |
| 340 | .await |
| 341 | } |
| 342 | |
| 343 | /// A heartbeat, with whatever `fields` change: the session is seen now, |
| 344 | /// brought back if it had ended (announced as started again), its |
| 345 | /// claims renewed, and entries recorded before it named its pull |
| 346 | /// request moved onto it. A subagent that finished only takes the |
| 347 | /// fields, and stays ended. |
| 348 | async fn resume(&self, repo: &Repo, row: SessionRow, actor: &User, fields: Fields) -> Result<Outcome<Session>> { |
| 349 | let now = now_ms(); |
| 350 | let was_ended = row.state == "ended"; |
| 351 | let finished = finished_subagent(&row); |
| 352 | let had_pull = row.pull; |
| 353 | let parent_id = row.parent_id.clone(); |
| 354 | let id = row.id.clone(); |
| 355 | let at = rfc3339(now); |
| 356 | let mut binds = fields.binds(); |
| 357 | binds.push(at.as_str().into()); |
| 358 | binds.push(id.as_str().into()); |
| 359 | let sql = if finished { refresh_sql() } else { resume_sql() }; |
| 360 | let updated = self.db.prepare(sql).bind(&binds)?.first::<SessionRow>(None).await?; |
| 361 | let Some(updated) = updated else { |
| 362 | return Ok(not_found()); |
| 363 | }; |
| 364 | let session = updated.at(now); |
| 365 | self.link_fields(&session, &fields).await?; |
| 366 | if finished { |
| 367 | return Ok(Outcome::Ok(session)); |
| 368 | } |
| 369 | self.renew_claims(&id, now).await?; |
| 370 | if let Some(parent) = &parent_id { |
| 371 | self.db.prepare(SEE_PARENT).bind(&[at.as_str().into(), parent.as_str().into()])?.run().await?; |
| 372 | } |
| 373 | if let Some(number) = session.pull |
| 374 | && had_pull != Some(number) |
| 375 | { |
| 376 | self.move_entries_onto_pull(&repo.id, &id, number).await?; |
| 377 | } |
| 378 | if was_ended { |
| 379 | self.publish("session.started", &repo.id, actor, SessionEvent::of(&session)).await?; |
| 380 | } |
| 381 | Ok(Outcome::Ok(session)) |
| 382 | } |
| 383 | |
| 384 | /// Renews the active claims made from a session by their own length. |
| 385 | async fn renew_claims(&self, session_id: &str, now: u64) -> Result<()> { |
| 386 | let claims = self |
| 387 | .db |
| 388 | .prepare("SELECT id, ttl_minutes, expires_at FROM claims WHERE session_id = ? AND state = 'active'") |
| 389 | .bind(&[session_id.into()])? |
| 390 | .all() |
| 391 | .await? |
| 392 | .results::<RenewRow>()?; |
| 393 | let mut statements = Vec::new(); |
| 394 | for claim in claims { |
| 395 | if let Some(renewed) = renewal(now, claim.ttl_minutes, &claim.expires_at) { |
| 396 | statements.push(self.db.prepare(RENEW_CLAIM).bind(&[renewed.into(), claim.id.into()])?); |
| 397 | } |
| 398 | } |
| 399 | if !statements.is_empty() { |
| 400 | self.db.batch(statements).await?; |
| 401 | } |
| 402 | Ok(()) |
| 403 | } |
| 404 | |
| 405 | /// Entries a session recorded before it named its pull request join |
| 406 | /// that pull request's session, after what it has, in their order. |
| 407 | async fn move_entries_onto_pull(&self, repo_id: &str, session_id: &str, number: u32) -> Result<()> { |
| 408 | let Some(pull) = self.pull(repo_id, number).await? else { |
| 409 | return Ok(()); |
| 410 | }; |
| 411 | let base = self |
| 412 | .db |
| 413 | .prepare("SELECT COALESCE(MAX(seq), 0) AS n FROM session_entries WHERE pull_id = ?") |
| 414 | .bind(&[pull.id.as_str().into()])? |
| 415 | .first::<NumberRow>(None) |
| 416 | .await? |
| 417 | .map_or(0, |row| row.n); |
| 418 | let moved = self |
| 419 | .db |
| 420 | .prepare(MOVE_ENTRIES) |
| 421 | .bind(&[pull.id.as_str().into(), base.into(), session_id.into()])? |
| 422 | .run() |
| 423 | .await; |
| 424 | // An entry recorded on the pull request at the same moment took a |
| 425 | // number these needed: they stay the session's alone. |
| 426 | if let Err(error) = moved { |
| 427 | worker::console_error!("session {session_id}: entries not moved onto #{number}: {error}"); |
| 428 | } |
| 429 | Ok(()) |
| 430 | } |
| 431 | |
| 432 | /// The caller's own session, as stored. |
| 433 | async fn own_session(&self, repo: &Repo, actor: &User, id: &str) -> Result<Outcome<SessionRow>> { |
| 434 | Ok(match self.session_row(&repo.id, id.trim()).await? { |
| 435 | Some(row) if row.actor_id == actor.id => Outcome::Ok(row), |
| 436 | _ => not_found(), |
| 437 | }) |
| 438 | } |
| 439 | |
| 440 | pub(crate) async fn update_session(&self, a: UpdateSessionArgs) -> Result<Outcome<Session>> { |
| 441 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 442 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 443 | if row.state == "ended" && row.end_reason.as_deref() != Some("timed_out") && !finished_subagent(&row) { |
| 444 | return Ok(Outcome::fail(FailureCode::Conflict, "This session has ended. Start a new one.")); |
| 445 | } |
| 446 | let (issue, pull) = unwrap_or_fail!(self.session_subjects(&repo, a.issue, a.pull).await?); |
| 447 | let fields = Fields { |
| 448 | issue, |
| 449 | pull, |
| 450 | branch: tidy(a.branch.as_deref(), MAX_BRANCH_CHARS), |
| 451 | head_commit: tidy(a.head_commit.as_deref(), MAX_COMMIT_CHARS), |
| 452 | title: tidy(a.title.as_deref(), MAX_TITLE_CHARS), |
| 453 | agent_type: tidy(a.agent_type.as_deref(), MAX_AGENT_CHARS), |
| 454 | task: clip(a.task.as_deref(), MAX_TASK_CHARS), |
| 455 | prompt: clip(a.prompt.as_deref(), MAX_PROMPT_CHARS), |
| 456 | model: tidy(a.model.as_deref(), MAX_MODEL_CHARS), |
| 457 | result: clip(a.result.as_deref(), MAX_RESULT_CHARS), |
| 458 | }; |
| 459 | self.resume(&repo, row, &a.actor, fields).await |
| 460 | } |
| 461 | |
| 462 | pub(crate) async fn end_session(&self, a: EndSessionArgs) -> Result<Outcome<Session>> { |
| 463 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 464 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 465 | let now = now_ms(); |
| 466 | let at = rfc3339(now); |
| 467 | let result = clip(a.result.as_deref(), MAX_RESULT_CHARS); |
| 468 | if row.state == "ended" { |
| 469 | // Ended already: it keeps what it reported, or takes it now. |
| 470 | if row.result.is_none() |
| 471 | && let Some(result) = &result |
| 472 | { |
| 473 | let updated = self |
| 474 | .db |
| 475 | .prepare(format!("UPDATE sessions SET result = ? WHERE id = ? RETURNING {SESSION_COLUMNS}")) |
| 476 | .bind(&[result.as_str().into(), row.id.as_str().into()])? |
| 477 | .first::<SessionRow>(None) |
| 478 | .await?; |
| 479 | if let Some(updated) = updated { |
| 480 | return Ok(Outcome::Ok(updated.at(now))); |
| 481 | } |
| 482 | } |
| 483 | return Ok(Outcome::Ok(row.at(now))); |
| 484 | } |
| 485 | let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(|| "ended".to_owned()); |
| 486 | let ended = self |
| 487 | .db |
| 488 | .prepare(format!( |
| 489 | "UPDATE sessions SET state = 'ended', ended_at = ?, end_reason = ?, last_seen_at = ?, result = COALESCE(?, result) |
| 490 | WHERE id = ? AND state = 'active' |
| 491 | RETURNING {SESSION_COLUMNS}" |
| 492 | )) |
| 493 | .bind(&[at.as_str().into(), reason.into(), at.as_str().into(), text_or_null(&result), row.id.as_str().into()])? |
| 494 | .first::<SessionRow>(None) |
| 495 | .await?; |
| 496 | let Some(ended) = ended else { |
| 497 | // Ended by another call meanwhile. |
| 498 | return Ok(match self.session_row(&repo.id, &row.id).await? { |
| 499 | Some(row) => Outcome::Ok(row.at(now)), |
| 500 | None => not_found(), |
| 501 | }); |
| 502 | }; |
| 503 | let session = ended.at(now); |
| 504 | if let Some(result) = &result { |
| 505 | self.link_refs(&session.id, &session.repo_id, &result.chars().take(MAX_REF_TEXT).collect::<String>()).await?; |
| 506 | } |
| 507 | self.publish("session.ended", &repo.id, &a.actor, SessionEvent::of(&session)).await?; |
| 508 | // The subagents it left going end with it. |
| 509 | let orphans = self |
| 510 | .db |
| 511 | .prepare(end_tree_sql()) |
| 512 | .bind(&[session.id.as_str().into(), at.as_str().into(), "parent ended".into()])? |
| 513 | .all() |
| 514 | .await? |
| 515 | .results::<SessionRow>()?; |
| 516 | for orphan in orphans { |
| 517 | self.publish("session.ended", &repo.id, &a.actor, SessionEvent::of(&orphan.at(now))).await?; |
| 518 | } |
| 519 | Ok(Outcome::Ok(session)) |
| 520 | } |
| 521 | |
| 522 | pub(crate) async fn get_work_session(&self, a: WorkSessionArgs) -> Result<Outcome<SessionDetail>> { |
| 523 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 524 | let Some(row) = self.session_row(&repo.id, a.id.trim()).await? else { |
| 525 | return Ok(not_found()); |
| 526 | }; |
| 527 | let now = now_ms(); |
| 528 | let entries = self |
| 529 | .db |
| 530 | .prepare( |
| 531 | "SELECT session_seq AS seq, kind, text, tool, \"commit\", at FROM session_entries |
| 532 | WHERE session_id = ? AND session_seq > ? ORDER BY session_seq LIMIT ?", |
| 533 | ) |
| 534 | .bind(&[row.id.as_str().into(), a.after_seq.into(), ENTRY_PAGE.into()])? |
| 535 | .all() |
| 536 | .await? |
| 537 | .results::<crate::rows::SessionRow>()? |
| 538 | .into_iter() |
| 539 | .map(SessionEntry::from) |
| 540 | .collect(); |
| 541 | let mut session = row.at(now); |
| 542 | let children = self.subagents_of(&session.id, now).await?; |
| 543 | session.subagents = children.iter().filter(|child| child.parent_id.as_deref() == Some(session.id.as_str())).count() as u32; |
| 544 | // A subagent's page names the session at the top of its tree. |
| 545 | let root = match &session.parent_id { |
| 546 | Some(_) => { |
| 547 | let mut top = None; |
| 548 | let mut at = session.parent_id.clone(); |
| 549 | for _ in 0..8 { |
| 550 | let Some(id) = at else { break }; |
| 551 | let Some(found) = self.session_row(&repo.id, &id).await? else { break }; |
| 552 | at = found.parent_id.clone(); |
| 553 | top = Some(found.at(now)); |
| 554 | } |
| 555 | top |
| 556 | } |
| 557 | None => None, |
| 558 | }; |
| 559 | let mut ids: Vec<String> = vec![session.id.clone()]; |
| 560 | ids.extend(children.iter().map(|child| child.id.clone())); |
| 561 | let (touched, links, skills) = self.tree_facts(&repo.id, &ids).await?; |
| 562 | Ok(Outcome::Ok(SessionDetail { session, entries, children, touched, links, skills, root })) |
| 563 | } |
| 564 | |
| 565 | pub(crate) async fn list_work_sessions(&self, a: ListWorkSessionsArgs) -> Result<Outcome<Vec<Session>>> { |
| 566 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 567 | let now = now_ms(); |
| 568 | let idle_since = rfc3339(now.saturating_sub(SESSION_IDLE_MS)); |
| 569 | // Every statement names each bind it is given. |
| 570 | let state = match a.state { |
| 571 | None => "AND ?2 IS NOT NULL", |
| 572 | Some(SessionState::Active) => "AND state = 'active' AND last_seen_at > ?2", |
| 573 | Some(SessionState::Idle) => "AND state = 'active' AND last_seen_at <= ?2", |
| 574 | Some(SessionState::Ended) => "AND state = 'ended' AND ?2 IS NOT NULL", |
| 575 | }; |
| 576 | // On an issue or pull request: attached, or tied to it any way. |
| 577 | let mut linked: Vec<String> = Vec::new(); |
| 578 | if let Some(number) = a.issue { |
| 579 | linked.extend(self.linked_sessions(&repo.id, LinkKind::Issue, number).await?); |
| 580 | // And those on its pull requests. |
| 581 | let pulls: Vec<u32> = self |
| 582 | .db |
| 583 | .prepare("SELECT number AS n FROM pulls WHERE repo_id = ? AND issue_number = ?") |
| 584 | .bind(&[repo.id.as_str().into(), number.into()])? |
| 585 | .all() |
| 586 | .await? |
| 587 | .results::<NumberRow>()? |
| 588 | .into_iter() |
| 589 | .map(|row| row.n) |
| 590 | .collect(); |
| 591 | for pull in &pulls { |
| 592 | linked.extend(self.linked_sessions(&repo.id, LinkKind::Pull, *pull).await?); |
| 593 | } |
| 594 | linked.extend(self.branch_sessions(&repo.id, &pulls).await?); |
| 595 | } |
| 596 | if let Some(number) = a.pull { |
| 597 | linked.extend(self.linked_sessions(&repo.id, LinkKind::Pull, number).await?); |
| 598 | linked.extend(self.branch_sessions(&repo.id, &[number]).await?); |
| 599 | } |
| 600 | let rows = self |
| 601 | .db |
| 602 | .prepare(format!( |
| 603 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 604 | WHERE repo_id = ?1 {state} |
| 605 | AND (?6 = 1 OR parent_id IS NULL OR id IN (SELECT value FROM json_each(?7))) |
| 606 | AND (?3 IS NULL OR issue = ?3 OR id IN (SELECT value FROM json_each(?7))) |
| 607 | AND (?4 IS NULL OR pull = ?4 OR id IN (SELECT value FROM json_each(?7))) |
| 608 | AND (?8 IS NULL OR id IN (SELECT session_id FROM session_touches WHERE repo_id = ?1 AND kind = 'file' |
| 609 | AND (value = ?8 OR (substr(?8, -1) = '/' AND substr(value, 1, length(?8)) = ?8)))) |
| 610 | ORDER BY last_seen_at DESC LIMIT ?5" |
| 611 | )) |
| 612 | .bind(&[ |
| 613 | repo.id.as_str().into(), |
| 614 | idle_since.as_str().into(), |
| 615 | number_or_null(a.issue), |
| 616 | number_or_null(a.pull), |
| 617 | LIST_LIMIT.into(), |
| 618 | u32::from(a.subagents).into(), |
| 619 | serde_json::to_string(&linked)?.into(), |
| 620 | text_or_null(&tidy(a.path.as_deref().map(|path| path.trim_start_matches("./")), MAX_TOUCH_CHARS)), |
| 621 | ])? |
| 622 | .all() |
| 623 | .await? |
| 624 | .results::<SessionRow>()?; |
| 625 | let mut sessions: Vec<Session> = rows.into_iter().map(|row| row.at(now)).collect(); |
| 626 | // With subagents, each tree whole: its subagents, and its top. |
| 627 | if a.subagents { |
| 628 | let mut more: Vec<Session> = Vec::new(); |
| 629 | for session in sessions.iter().filter(|session| session.parent_id.is_none()).take(20) { |
| 630 | more.extend(self.subagents_of(&session.id, now).await?); |
| 631 | } |
| 632 | let parents: Vec<String> = sessions |
| 633 | .iter() |
| 634 | .filter_map(|session| session.parent_id.clone()) |
| 635 | .filter(|parent| !sessions.iter().any(|session| &session.id == parent)) |
| 636 | .collect(); |
| 637 | more.extend(self.sessions_by_id(&repo.id, &parents).await?); |
| 638 | for session in more { |
| 639 | if !sessions.iter().any(|known| known.id == session.id) { |
| 640 | sessions.push(session); |
| 641 | } |
| 642 | } |
| 643 | count_subagents(&mut sessions); |
| 644 | } else { |
| 645 | self.fill_subagents(&mut sessions).await?; |
| 646 | } |
| 647 | self.fill_first_prompts(&mut sessions).await?; |
| 648 | Ok(Outcome::Ok(sessions)) |
| 649 | } |
| 650 | |
| 651 | /// Notes `text` (from g1t) in the live harness sessions on `pull` or |
| 652 | /// its issue, so an agent reading its session hears of it. Never fails |
| 653 | /// what it follows: a problem is logged. |
| 654 | pub(crate) async fn note_sessions(&self, pull: &Pull, text: &str) { |
| 655 | let noted = async { |
| 656 | let sessions: Vec<SessionRow> = self |
| 657 | .db |
| 658 | .prepare(format!( |
| 659 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 660 | WHERE repo_id = ?1 AND state = 'active' AND (pull = ?2 OR (?3 IS NOT NULL AND issue = ?3))" |
| 661 | )) |
| 662 | .bind(&[pull.repo_id.as_str().into(), pull.number.into(), number_or_null(pull.issue)])? |
| 663 | .all() |
| 664 | .await? |
| 665 | .results()?; |
| 666 | let at = rfc3339(now_ms()); |
| 667 | let mut statements = Vec::new(); |
| 668 | for session in sessions { |
| 669 | // Kept by the session alone: the pull request's own session |
| 670 | // has the note already. |
| 671 | statements.push(self.db.prepare(INSERT_ENTRY).bind(&[ |
| 672 | JsValue::NULL, |
| 673 | "note".into(), |
| 674 | text.into(), |
| 675 | JsValue::NULL, |
| 676 | text_or_null(&pull.head_commit), |
| 677 | at.as_str().into(), |
| 678 | session.id.as_str().into(), |
| 679 | JsValue::NULL, |
| 680 | ])?); |
| 681 | } |
| 682 | if !statements.is_empty() { |
| 683 | self.db.batch(statements).await?; |
| 684 | } |
| 685 | Ok::<(), worker::Error>(()) |
| 686 | }; |
| 687 | if let Err(error) = noted.await { |
| 688 | worker::console_error!("#{}: sessions were not told: {error}", pull.number); |
| 689 | } |
| 690 | } |
| 691 | |
| 692 | pub(crate) async fn record_session_entries(&self, a: RecordEntriesArgs) -> Result<Outcome<Recorded>> { |
| 693 | if a.entries.len() > crate::MAX_ENTRY_BATCH { |
| 694 | return Ok(Outcome::fail( |
| 695 | FailureCode::Invalid, |
| 696 | format!("Send at most {} entries at a time.", crate::MAX_ENTRY_BATCH), |
| 697 | )); |
| 698 | } |
| 699 | if a.touched.len() > MAX_TOUCHES { |
| 700 | return Ok(Outcome::fail(FailureCode::Invalid, format!("Say at most {MAX_TOUCHES} things touched at a time."))); |
| 701 | } |
| 702 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 703 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 704 | if row.state == "ended" && row.end_reason.as_deref() != Some("timed_out") { |
| 705 | return Ok(Outcome::fail(FailureCode::Conflict, "This session has ended. Start a new one.")); |
| 706 | } |
| 707 | // Recording is a heartbeat too. |
| 708 | let session = unwrap_or_fail!(self.resume(&repo, row, &a.actor, Fields::default()).await?); |
| 709 | let touched = self.record_touches(&session, &a.touched).await?; |
| 710 | if a.entries.is_empty() { |
| 711 | if touched > 0 { |
| 712 | self.announce_progress(&session).await?; |
| 713 | } |
| 714 | return Ok(Outcome::Ok(Recorded::default())); |
| 715 | } |
| 716 | let pull = match session.pull { |
| 717 | Some(number) => self.pull(&repo.id, number).await?, |
| 718 | None => None, |
| 719 | }; |
| 720 | let pull_id: JsValue = pull.as_ref().map_or(JsValue::NULL, |pull| pull.id.as_str().into()); |
| 721 | let at = rfc3339(now_ms()); |
| 722 | let total = a.entries.len() as u32; |
| 723 | let mut statements = Vec::with_capacity(a.entries.len()); |
| 724 | let mut prompts = String::new(); |
| 725 | for keyed in a.entries { |
| 726 | let entry = keyed.entry; |
| 727 | let kind = serde_json::to_value(entry.kind)?; |
| 728 | let mut text: String = entry.text.chars().take(crate::MAX_ENTRY_CHARS).collect(); |
| 729 | if entry.kind == SessionEntryKind::Skill { |
| 730 | // A skill is its name, nothing more. |
| 731 | text = text.trim().chars().take(MAX_AGENT_CHARS).collect(); |
| 732 | } |
| 733 | if entry.kind == SessionEntryKind::Prompt && prompts.len() < MAX_REF_TEXT { |
| 734 | prompts.push_str(&text); |
| 735 | prompts.push('\n'); |
| 736 | } |
| 737 | let key = tidy(keyed.key.as_deref(), MAX_KEY_CHARS); |
| 738 | // Each takes the next number of the session, and of its pull |
| 739 | // request when it has one, itself, so two batches at once |
| 740 | // cannot collide; a key already recorded is skipped. |
| 741 | statements.push( |
| 742 | self.db |
| 743 | .prepare(INSERT_ENTRY) |
| 744 | .bind(&[ |
| 745 | pull_id.clone(), |
| 746 | kind.as_str().unwrap_or("note").into(), |
| 747 | text.into(), |
| 748 | text_or_null(&entry.tool), |
| 749 | text_or_null(&entry.commit.or_else(|| session.head_commit.clone())), |
| 750 | at.as_str().into(), |
| 751 | session.id.as_str().into(), |
| 752 | text_or_null(&key), |
| 753 | ])?, |
| 754 | ); |
| 755 | } |
| 756 | let mut recorded = 0; |
| 757 | for result in self.db.batch(statements).await? { |
| 758 | recorded += result.results::<NumberRow>()?.len() as u32; |
| 759 | } |
| 760 | if !prompts.is_empty() && recorded > 0 { |
| 761 | self.link_refs(&session.id, &repo.id, &prompts).await?; |
| 762 | } |
| 763 | if let Some(pull) = pull |
| 764 | && recorded > 0 |
| 765 | { |
| 766 | self.publish( |
| 767 | "session.appended", |
| 768 | &repo.id, |
| 769 | &a.actor, |
| 770 | SessionAppended { pull_id: pull.id, repo_id: repo.id.clone(), number: pull.number, count: recorded }, |
| 771 | ) |
| 772 | .await?; |
| 773 | } |
| 774 | if recorded > 0 || touched > 0 { |
| 775 | self.announce_progress(&session).await?; |
| 776 | } |
| 777 | Ok(Outcome::Ok(Recorded { recorded, duplicates: total - recorded })) |
| 778 | } |
| 779 | |
| 780 | /// `session.progressed`, at most every [`PROGRESS_EVERY_MS`] for a |
| 781 | /// session, so the pages showing it follow along. |
| 782 | async fn announce_progress(&self, session: &Session) -> Result<()> { |
| 783 | let now = now_ms(); |
| 784 | let due = self |
| 785 | .db |
| 786 | .prepare(MARK_PROGRESS) |
| 787 | .bind(&[rfc3339(now).into(), session.id.as_str().into(), rfc3339(now.saturating_sub(PROGRESS_EVERY_MS)).into()])? |
| 788 | .first::<serde_json::Value>(None) |
| 789 | .await?; |
| 790 | if due.is_none() { |
| 791 | return Ok(()); |
| 792 | } |
| 793 | let mut root_id = None; |
| 794 | let mut up = session.parent_id.clone(); |
| 795 | for _ in 0..8 { |
| 796 | let Some(id) = up else { break }; |
| 797 | up = self.session_row(&session.repo_id, &id).await?.and_then(|row| row.parent_id); |
| 798 | root_id = Some(id); |
| 799 | } |
| 800 | self.publish_as( |
| 801 | "session.progressed", |
| 802 | &session.repo_id, |
| 803 | Some(session.actor.id.clone()), |
| 804 | SessionProgressed { |
| 805 | repo_id: session.repo_id.clone(), |
| 806 | number: session.pull.or(session.issue), |
| 807 | session_id: session.id.clone(), |
| 808 | root_id, |
| 809 | }, |
| 810 | ) |
| 811 | .await |
| 812 | } |
| 813 | } |