| 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; anyone with write access may also end it, for one left |
| 9 | //! behind (recorded as ended by hand by them). |
| 10 | //! |
| 11 | //! A subagent is a session of its own with a parent (`parent_id`): a |
| 12 | //! Claude Code subagent, a Managed Agents thread. It starts on what its |
| 13 | //! parent is on unless it says otherwise, keeps the task, prompt and |
| 14 | //! result its parent gave and got (cut short), and once ended stays ended |
| 15 | //! even if a late hook starts it again: a subagent never comes back. Its |
| 16 | //! heartbeats keep its parent seen, and ending a session ends the |
| 17 | //! subagents it left going. |
| 18 | |
| 19 | use g1t_contracts::events::SessionAppended; |
| 20 | use g1t_contracts::repos::Repo; |
| 21 | use g1t_contracts::time::rfc3339; |
| 22 | use g1t_contracts::work::*; |
| 23 | use g1t_contracts::{FailureCode, Outcome, User, new_id}; |
| 24 | use g1t_kit::now_ms; |
| 25 | use serde::Deserialize; |
| 26 | use worker::Result; |
| 27 | use worker::wasm_bindgen::JsValue; |
| 28 | |
| 29 | use super::links::{PULL_SESSIONS, PullSessionRow, count_subagents, end_tree_sql, pull_sessions_of, tree_sql}; |
| 30 | |
| 31 | /// The most pull requests one `pull_sessions` asks about. |
| 32 | const MAX_PULL_SESSIONS: usize = 100; |
| 33 | use super::updates::push_delivery; |
| 34 | use super::{SESSION_COLUMNS, SessionRow, after_minutes, shown, tidy, titles, unwrap_or_fail}; |
| 35 | use crate::Work; |
| 36 | use crate::items; |
| 37 | use crate::rows::NumberRow; |
| 38 | |
| 39 | /// The most entries one look at a session returns. |
| 40 | const ENTRY_PAGE: u32 = 500; |
| 41 | /// The most sessions one list returns. |
| 42 | const LIST_LIMIT: u32 = 100; |
| 43 | const MAX_TITLE_CHARS: usize = MAX_SESSION_TITLE_CHARS; |
| 44 | const MAX_BRANCH_CHARS: usize = 255; |
| 45 | const MAX_ID_CHARS: usize = 200; |
| 46 | const MAX_COMMIT_CHARS: usize = 64; |
| 47 | const MAX_AGENT_CHARS: usize = 100; |
| 48 | const MAX_MODEL_CHARS: usize = 100; |
| 49 | /// How long one prompt entry may be before its `#N` go unread. |
| 50 | const MAX_REF_TEXT: usize = 8_000; |
| 51 | |
| 52 | fn not_found<T>() -> Outcome<T> { |
| 53 | Outcome::fail(FailureCode::NotFound, "Session not found.") |
| 54 | } |
| 55 | |
| 56 | fn text_or_null(value: &Option<String>) -> JsValue { |
| 57 | value.as_deref().map_or(JsValue::NULL, JsValue::from) |
| 58 | } |
| 59 | |
| 60 | fn number_or_null(value: Option<u32>) -> JsValue { |
| 61 | value.map_or(JsValue::NULL, JsValue::from) |
| 62 | } |
| 63 | |
| 64 | /// Long text as kept: trimmed, cut to `limit` characters with an ellipsis. |
| 65 | fn clip(text: Option<&str>, limit: usize) -> Option<String> { |
| 66 | let text = text?.trim(); |
| 67 | if text.is_empty() { |
| 68 | return None; |
| 69 | } |
| 70 | if text.chars().count() <= limit { |
| 71 | return Some(text.to_owned()); |
| 72 | } |
| 73 | let cut: String = text.chars().take(limit.saturating_sub(1)).collect(); |
| 74 | Some(format!("{}…", cut.trim_end())) |
| 75 | } |
| 76 | |
| 77 | /// What a session's heartbeat may change, tidied. |
| 78 | #[derive(Default)] |
| 79 | struct Fields { |
| 80 | /// Its work item, by id and number. |
| 81 | item: Option<(String, u32)>, |
| 82 | pull: Option<u32>, |
| 83 | branch: Option<String>, |
| 84 | head_commit: Option<String>, |
| 85 | title: Option<String>, |
| 86 | agent_type: Option<String>, |
| 87 | task: Option<String>, |
| 88 | prompt: Option<String>, |
| 89 | model: Option<String>, |
| 90 | result: Option<String>, |
| 91 | } |
| 92 | |
| 93 | impl Fields { |
| 94 | fn binds(&self) -> Vec<JsValue> { |
| 95 | vec![ |
| 96 | self.item.as_ref().map_or(JsValue::NULL, |(id, _)| JsValue::from(id.as_str())), |
| 97 | number_or_null(self.pull), |
| 98 | text_or_null(&self.branch), |
| 99 | text_or_null(&self.head_commit), |
| 100 | text_or_null(&self.title), |
| 101 | text_or_null(&self.agent_type), |
| 102 | text_or_null(&self.task), |
| 103 | text_or_null(&self.prompt), |
| 104 | text_or_null(&self.model), |
| 105 | text_or_null(&self.result), |
| 106 | ] |
| 107 | } |
| 108 | |
| 109 | /// The text whose `#N` tie the session to work items and pull requests. |
| 110 | fn named(&self) -> String { |
| 111 | [&self.title, &self.task, &self.prompt].into_iter().flatten().cloned().collect::<Vec<_>>().join("\n") |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | #[derive(Deserialize)] |
| 116 | struct RenewRow { |
| 117 | id: String, |
| 118 | ttl_minutes: u32, |
| 119 | expires_at: String, |
| 120 | } |
| 121 | |
| 122 | /// Starts a session; nothing when one with the same harness session id |
| 123 | /// was started already. |
| 124 | pub(crate) fn insert_session_sql() -> String { |
| 125 | format!( |
| 126 | "INSERT INTO sessions |
| 127 | (id, repo_id, item_id, pull, branch, actor_id, actor_name, harness, harness_session_id, |
| 128 | title, state, head_commit, started_at, last_seen_at, parent_id, agent_type, task, prompt, model, |
| 129 | title_source, title_settled_at) |
| 130 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 'active', ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, |
| 131 | CASE WHEN ?10 IS NULL THEN NULL ELSE 'agent' END, CASE WHEN ?10 IS NULL THEN NULL ELSE ?12 END) |
| 132 | ON CONFLICT (repo_id, actor_id, harness, harness_session_id) |
| 133 | WHERE harness_session_id IS NOT NULL DO NOTHING |
| 134 | RETURNING {SESSION_COLUMNS}" |
| 135 | ) |
| 136 | } |
| 137 | |
| 138 | /// A person's session by its harness's id for it. |
| 139 | pub(crate) fn harness_session_sql() -> String { |
| 140 | format!( |
| 141 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 142 | WHERE repo_id = ?1 AND actor_id = ?2 AND harness = ?3 AND harness_session_id = ?4" |
| 143 | ) |
| 144 | } |
| 145 | |
| 146 | /// The fields `?1`–`?10` of [`Fields::binds`], each kept when not given; |
| 147 | /// a title as [`titles::SET_TITLE`] takes it, from `?13` (`?11` now). |
| 148 | const SET_FIELDS: &str = "item_id = COALESCE(?1, item_id), pull = COALESCE(?2, pull), branch = COALESCE(?3, branch), |
| 149 | head_commit = COALESCE(?4, head_commit), |
| 150 | agent_type = COALESCE(?6, agent_type), task = COALESCE(?7, task), prompt = COALESCE(?8, prompt), |
| 151 | model = COALESCE(?9, model), result = COALESCE(?10, result)"; |
| 152 | |
| 153 | /// A heartbeat: seen now (`?11`), active again, with whichever fields are |
| 154 | /// given, of session `?12`; a title given came from `?13`. One that ended |
| 155 | /// is named again as it ends again. |
| 156 | pub(crate) fn resume_sql() -> String { |
| 157 | format!( |
| 158 | "UPDATE sessions |
| 159 | SET {SET_FIELDS}, {set_title}, |
| 160 | title_final = CASE WHEN state = 'ended' THEN 0 ELSE title_final END, |
| 161 | last_seen_at = ?11, state = 'active', ended_at = NULL, end_reason = NULL |
| 162 | WHERE id = ?12 |
| 163 | RETURNING {SESSION_COLUMNS}", |
| 164 | set_title = titles::SET_TITLE, |
| 165 | ) |
| 166 | } |
| 167 | |
| 168 | /// The fields of session `?12`, changed without bringing it back: for a |
| 169 | /// subagent's late hooks (the binds are the same). |
| 170 | pub(crate) fn refresh_sql() -> String { |
| 171 | format!( |
| 172 | "UPDATE sessions SET {SET_FIELDS}, {set_title} |
| 173 | WHERE id = ?12 AND ?11 IS NOT NULL |
| 174 | RETURNING {SESSION_COLUMNS}", |
| 175 | set_title = titles::SET_TITLE, |
| 176 | ) |
| 177 | } |
| 178 | |
| 179 | /// Session `?3`'s token usage (`?1`, JSON as reported) and the cost |
| 180 | /// estimated from it (`?2`): cumulative, so each report replaces the last. |
| 181 | pub(crate) const SET_USAGE: &str = "UPDATE sessions SET usage = ?1, cost_usd = ?2 WHERE id = ?3"; |
| 182 | |
| 183 | /// A subagent's heartbeat keeps its parent `?2` seen at `?1`. |
| 184 | pub(crate) const SEE_PARENT: &str = "UPDATE sessions SET last_seen_at = ?1 WHERE id = ?2 AND state = 'active' AND last_seen_at < ?1"; |
| 185 | |
| 186 | /// Records one entry in session `?7`, and in pull request `?1` when it |
| 187 | /// is on one. Each takes the next number of the session, and of its pull |
| 188 | /// request, itself, so two batches at once cannot collide; an entry whose |
| 189 | /// key (`?8`) the session has recorded already is skipped. |
| 190 | pub(crate) const INSERT_ENTRY: &str = "INSERT INTO session_entries |
| 191 | (pull_id, seq, kind, text, tool, \"commit\", at, session_id, session_seq, key) |
| 192 | SELECT ?1, |
| 193 | CASE WHEN ?1 IS NULL |
| 194 | THEN (SELECT COALESCE(MAX(session_seq), 0) + 1 FROM session_entries WHERE session_id = ?7) |
| 195 | ELSE (SELECT COALESCE(MAX(seq), 0) + 1 FROM session_entries WHERE pull_id = ?1) END, |
| 196 | ?2, ?3, ?4, ?5, ?6, ?7, |
| 197 | (SELECT COALESCE(MAX(session_seq), 0) + 1 FROM session_entries WHERE session_id = ?7), |
| 198 | ?8 |
| 199 | WHERE true |
| 200 | ON CONFLICT (session_id, key) WHERE session_id IS NOT NULL AND key IS NOT NULL DO NOTHING |
| 201 | RETURNING session_seq AS n"; |
| 202 | |
| 203 | /// Entries of session `?3` recorded before it named its pull request |
| 204 | /// `?1` join it, numbered after its last (`?2`), in their order. |
| 205 | pub(crate) const MOVE_ENTRIES: &str = "UPDATE session_entries SET pull_id = ?1, seq = ?2 + session_seq |
| 206 | WHERE session_id = ?3 AND pull_id IS NULL"; |
| 207 | |
| 208 | /// Renews active claim `?2` to `?1`. |
| 209 | pub(crate) const RENEW_CLAIM: &str = "UPDATE claims SET expires_at = ?1 WHERE id = ?2 AND state = 'active'"; |
| 210 | |
| 211 | /// Marks session `?2` as announced at `?1` when it was last announced at |
| 212 | /// or before `?3` (or never): a row back means it is time to announce. |
| 213 | pub(crate) const MARK_PROGRESS: &str = "UPDATE sessions SET progressed_at = ?1 |
| 214 | WHERE id = ?2 AND (progressed_at IS NULL OR progressed_at <= ?3) RETURNING progressed_at"; |
| 215 | |
| 216 | /// When a claim of `minutes` renewed at `now` lapses, if that is later |
| 217 | /// than it does already (`expires_at`): a heartbeat never shortens one. |
| 218 | pub(crate) fn renewal(now: u64, minutes: u32, expires_at: &str) -> Option<String> { |
| 219 | let renewed = after_minutes(now, minutes); |
| 220 | (renewed.as_str() > expires_at).then_some(renewed) |
| 221 | } |
| 222 | |
| 223 | /// Whether a session is a subagent that ended on purpose: it stays ended. |
| 224 | fn finished_subagent(row: &SessionRow) -> bool { |
| 225 | row.parent_id.is_some() && row.state == "ended" && !revives(row.end_reason.as_deref()) |
| 226 | } |
| 227 | |
| 228 | impl Work { |
| 229 | /// The work item and pull request a session names, checked: a work |
| 230 | /// item of the repository's workspace, and a pull request of the |
| 231 | /// repository. A pull request for a work item names that item when no |
| 232 | /// item was given. |
| 233 | async fn session_subjects(&self, repo: &Repo, item: Option<u32>, pull: Option<u32>) -> Result<Outcome<(Option<(String, u32)>, Option<u32>)>> { |
| 234 | let workspace = repo.namespace.to_lowercase(); |
| 235 | let mut found_item = None; |
| 236 | if let Some(number) = item.filter(|number| *number > 0) { |
| 237 | match items::by_number(self, &workspace, number).await? { |
| 238 | Some(held) => found_item = Some((held.item.id, number)), |
| 239 | None => return Ok(Outcome::fail(FailureCode::Invalid, format!("#{number} is not a work item in {workspace}."))), |
| 240 | } |
| 241 | } |
| 242 | let pull = pull.filter(|number| *number > 0); |
| 243 | if let Some(number) = pull { |
| 244 | match self.pull(&repo.id, number).await? { |
| 245 | Some(found) => { |
| 246 | if found_item.is_none() |
| 247 | && let (Some(id), Some(number)) = (self.pull_item_id(&found.id).await?, found.item) |
| 248 | { |
| 249 | found_item = Some((id, number)); |
| 250 | } |
| 251 | } |
| 252 | None => { |
| 253 | return Ok(Outcome::fail(FailureCode::Invalid, format!("#{number} is not a pull request in this repository."))); |
| 254 | } |
| 255 | } |
| 256 | } |
| 257 | Ok(Outcome::Ok((found_item, pull))) |
| 258 | } |
| 259 | |
| 260 | /// Ties a session to what it said it is on, and what its text names. |
| 261 | async fn link_fields(&self, session: &Session, fields: &Fields) -> Result<()> { |
| 262 | if let Some((id, _)) = &fields.item { |
| 263 | self.link_item(&session.id, &session.repo_id, id, LinkVia::Attached).await; |
| 264 | } |
| 265 | if let Some(number) = fields.pull { |
| 266 | self.link_pull(&session.id, &session.repo_id, number, LinkVia::Attached).await; |
| 267 | } |
| 268 | if let Some(branch) = &fields.branch { |
| 269 | self.link_branch(&session.id, &session.repo_id, branch).await?; |
| 270 | } |
| 271 | let named = fields.named(); |
| 272 | if !named.is_empty() { |
| 273 | self.link_refs(&session.id, &session.repo_id, &named).await?; |
| 274 | } |
| 275 | Ok(()) |
| 276 | } |
| 277 | |
| 278 | pub(crate) async fn start_session(&self, a: StartSessionArgs) -> Result<Outcome<Session>> { |
| 279 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 280 | // A subagent's parent: one of the actor's sessions here. |
| 281 | let parent = match a.parent_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) { |
| 282 | Some(id) => match self.session_row(&repo.id, id).await? { |
| 283 | Some(row) if row.actor_id == a.actor.id => Some(row), |
| 284 | _ => return Ok(Outcome::fail(FailureCode::Invalid, "parent_id is not one of your sessions in this repository.")), |
| 285 | }, |
| 286 | None => None, |
| 287 | }; |
| 288 | let (item, pull) = unwrap_or_fail!(self.session_subjects(&repo, a.item, a.pull).await?); |
| 289 | let mut fields = Fields { |
| 290 | item, |
| 291 | pull, |
| 292 | branch: tidy(a.branch.as_deref(), MAX_BRANCH_CHARS), |
| 293 | head_commit: tidy(a.head_commit.as_deref(), MAX_COMMIT_CHARS), |
| 294 | title: tidy(a.title.as_deref().map(shown::clean).as_deref(), MAX_TITLE_CHARS), |
| 295 | agent_type: tidy(a.agent_type.as_deref(), MAX_AGENT_CHARS), |
| 296 | task: clip(a.task.as_deref().map(shown::clean).as_deref(), MAX_TASK_CHARS), |
| 297 | prompt: clip(a.prompt.as_deref().map(shown::clean).as_deref(), MAX_PROMPT_CHARS), |
| 298 | model: tidy(a.model.as_deref(), MAX_MODEL_CHARS), |
| 299 | result: None, |
| 300 | }; |
| 301 | let harness_session_id = tidy(a.harness_session_id.as_deref(), MAX_ID_CHARS); |
| 302 | let now = now_ms(); |
| 303 | let at = rfc3339(now); |
| 304 | // The same harness session again: the one already started. |
| 305 | if let Some(harness_id) = &harness_session_id |
| 306 | && let Some(row) = self.harness_session(&repo.id, &a.actor, a.harness, harness_id).await? |
| 307 | { |
| 308 | return self.resume(&repo, row, &a.actor, fields).await; |
| 309 | } |
| 310 | // A subagent works on what its parent does unless it says. |
| 311 | if let Some(parent) = &parent { |
| 312 | fields.item = fields.item.clone().or(parent.item_id.clone().zip(parent.item)); |
| 313 | fields.pull = fields.pull.or(parent.pull); |
| 314 | fields.branch = fields.branch.clone().or(parent.branch.clone()); |
| 315 | fields.head_commit = fields.head_commit.clone().or(parent.head_commit.clone()); |
| 316 | } |
| 317 | let parent_id = parent.as_ref().map(|parent| parent.id.clone()); |
| 318 | let id = new_id("ses", now); |
| 319 | // The session as the insert makes it, announced with it. |
| 320 | let expected = SessionRow { |
| 321 | id: id.clone(), |
| 322 | repo_id: repo.id.clone(), |
| 323 | item_id: fields.item.as_ref().map(|(id, _)| id.clone()), |
| 324 | item: fields.item.as_ref().map(|(_, number)| *number), |
| 325 | pull: fields.pull, |
| 326 | branch: fields.branch.clone(), |
| 327 | actor_id: a.actor.id.clone(), |
| 328 | actor_name: a.actor.username.clone(), |
| 329 | harness: a.harness.as_str().to_owned(), |
| 330 | harness_session_id: harness_session_id.clone(), |
| 331 | title: fields.title.clone(), |
| 332 | state: "active".to_owned(), |
| 333 | head_commit: fields.head_commit.clone(), |
| 334 | started_at: at.clone(), |
| 335 | last_seen_at: at.clone(), |
| 336 | ended_at: None, |
| 337 | end_reason: None, |
| 338 | parent_id: parent_id.clone(), |
| 339 | agent_type: fields.agent_type.clone(), |
| 340 | task: fields.task.clone(), |
| 341 | prompt: fields.prompt.clone(), |
| 342 | result: None, |
| 343 | model: fields.model.clone(), |
| 344 | title_source: fields.title.as_ref().map(|_| TitleSource::Agent.as_str().to_owned()), |
| 345 | title_settled_at: fields.title.as_ref().map(|_| at.clone()), |
| 346 | usage: None, |
| 347 | cost_usd: None, |
| 348 | } |
| 349 | .at(now); |
| 350 | let started = |session: &Session| g1t_contracts::events::marked(SessionEvent::of(session), Some(&a.actor)); |
| 351 | let mut statements = vec![self |
| 352 | .db |
| 353 | .prepare(insert_session_sql()) |
| 354 | .bind(&[ |
| 355 | id.as_str().into(), |
| 356 | repo.id.as_str().into(), |
| 357 | fields.item.as_ref().map_or(JsValue::NULL, |(id, _)| JsValue::from(id.as_str())), |
| 358 | number_or_null(fields.pull), |
| 359 | text_or_null(&fields.branch), |
| 360 | a.actor.id.as_str().into(), |
| 361 | a.actor.username.as_str().into(), |
| 362 | a.harness.as_str().into(), |
| 363 | text_or_null(&harness_session_id), |
| 364 | text_or_null(&fields.title), |
| 365 | text_or_null(&fields.head_commit), |
| 366 | at.as_str().into(), |
| 367 | at.as_str().into(), |
| 368 | text_or_null(&parent_id), |
| 369 | text_or_null(&fields.agent_type), |
| 370 | text_or_null(&fields.task), |
| 371 | text_or_null(&fields.prompt), |
| 372 | text_or_null(&fields.model), |
| 373 | ])?]; |
| 374 | // Started at the same moment by another call: none, and no update. |
| 375 | let sent = push_delivery( |
| 376 | &mut statements, |
| 377 | self.delivery( |
| 378 | "session.started", |
| 379 | &repo.id, |
| 380 | started(&expected), |
| 381 | &format!("{id}:started"), |
| 382 | "EXISTS (SELECT 1 FROM sessions WHERE id = ?7)", |
| 383 | vec![id.as_str().into()], |
| 384 | )?, |
| 385 | ); |
| 386 | let inserted = match self.db.batch(statements).await?.into_iter().next() { |
| 387 | Some(result) => result.results::<SessionRow>()?.into_iter().next(), |
| 388 | None => None, |
| 389 | }; |
| 390 | let Some(row) = inserted else { |
| 391 | // Started at the same moment by another call: that one. |
| 392 | let Some(harness_id) = &harness_session_id else { |
| 393 | return Ok(not_found()); |
| 394 | }; |
| 395 | return match self.harness_session(&repo.id, &a.actor, a.harness, harness_id).await? { |
| 396 | Some(row) => Ok(Outcome::Ok(row.at(now))), |
| 397 | None => Ok(not_found()), |
| 398 | }; |
| 399 | }; |
| 400 | let session = row.at(now); |
| 401 | self.link_fields(&session, &fields).await?; |
| 402 | if let Some(parent) = &parent_id { |
| 403 | self.db.prepare(SEE_PARENT).bind(&[at.as_str().into(), parent.as_str().into()])?.run().await?; |
| 404 | } |
| 405 | self.delivered(sent, "session.started", &repo.id, Some(a.actor.id.clone()), started(&session)).await?; |
| 406 | Ok(Outcome::Ok(session)) |
| 407 | } |
| 408 | |
| 409 | async fn harness_session(&self, repo_id: &str, actor: &User, harness: Harness, harness_id: &str) -> Result<Option<SessionRow>> { |
| 410 | self.db |
| 411 | .prepare(harness_session_sql()) |
| 412 | .bind(&[repo_id.into(), actor.id.as_str().into(), harness.as_str().into(), harness_id.into()])? |
| 413 | .first::<SessionRow>(None) |
| 414 | .await |
| 415 | } |
| 416 | |
| 417 | /// A heartbeat, with whatever `fields` change: the session is seen now, |
| 418 | /// brought back if it had ended (announced as started again), its |
| 419 | /// claims renewed, and entries recorded before it named its pull |
| 420 | /// request moved onto it. A subagent that finished only takes the |
| 421 | /// fields, and stays ended. |
| 422 | async fn resume(&self, repo: &Repo, row: SessionRow, actor: &User, fields: Fields) -> Result<Outcome<Session>> { |
| 423 | let now = now_ms(); |
| 424 | let was_ended = row.state == "ended"; |
| 425 | let finished = finished_subagent(&row); |
| 426 | let had_pull = row.pull; |
| 427 | let parent_id = row.parent_id.clone(); |
| 428 | let id = row.id.clone(); |
| 429 | let at = rfc3339(now); |
| 430 | let current_source = row.title_source.as_deref().and_then(TitleSource::parse); |
| 431 | // A title given: its agent's, unless it is what g1t calls it already. |
| 432 | let given = fields.title.as_deref().map(|title| titles::given_source(title, row.title.as_deref(), current_source)); |
| 433 | let pinned = current_source == Some(TitleSource::Person); |
| 434 | let mut binds = fields.binds(); |
| 435 | binds.push(at.as_str().into()); |
| 436 | binds.push(id.as_str().into()); |
| 437 | binds.push(given.map_or(JsValue::NULL, |source| source.as_str().into())); |
| 438 | let sql = if finished { refresh_sql() } else { resume_sql() }; |
| 439 | let mut statements = vec![self.db.prepare(sql).bind(&binds)?]; |
| 440 | // Brought back from having ended: announced with the change. |
| 441 | let sent = if was_ended && !finished { |
| 442 | let expected = SessionRow { |
| 443 | item_id: fields.item.as_ref().map(|(id, _)| id.clone()).or(row.item_id.clone()), |
| 444 | item: fields.item.as_ref().map(|(_, number)| *number).or(row.item), |
| 445 | pull: fields.pull.or(row.pull), |
| 446 | branch: fields.branch.clone().or(row.branch.clone()), |
| 447 | head_commit: fields.head_commit.clone().or(row.head_commit.clone()), |
| 448 | title: if pinned { row.title.clone() } else { fields.title.clone().or(row.title.clone()) }, |
| 449 | title_source: if pinned { row.title_source.clone() } else { given.map(|source| source.as_str().to_owned()).or(row.title_source.clone()) }, |
| 450 | agent_type: fields.agent_type.clone().or(row.agent_type.clone()), |
| 451 | task: fields.task.clone().or(row.task.clone()), |
| 452 | prompt: fields.prompt.clone().or(row.prompt.clone()), |
| 453 | model: fields.model.clone().or(row.model.clone()), |
| 454 | result: fields.result.clone().or(row.result.clone()), |
| 455 | last_seen_at: at.clone(), |
| 456 | state: "active".to_owned(), |
| 457 | ended_at: None, |
| 458 | end_reason: None, |
| 459 | ..row.clone() |
| 460 | } |
| 461 | .at(now); |
| 462 | push_delivery( |
| 463 | &mut statements, |
| 464 | self.delivery( |
| 465 | "session.started", |
| 466 | &repo.id, |
| 467 | g1t_contracts::events::marked(SessionEvent::of(&expected), Some(actor)), |
| 468 | &format!("{id}:started:{at}"), |
| 469 | "EXISTS (SELECT 1 FROM sessions WHERE id = ?7 AND state = 'active' AND last_seen_at = ?8)", |
| 470 | vec![id.as_str().into(), at.as_str().into()], |
| 471 | )?, |
| 472 | ) |
| 473 | } else { |
| 474 | None |
| 475 | }; |
| 476 | let updated = match self.db.batch(statements).await?.into_iter().next() { |
| 477 | Some(result) => result.results::<SessionRow>()?.into_iter().next(), |
| 478 | None => None, |
| 479 | }; |
| 480 | let Some(updated) = updated else { |
| 481 | return Ok(not_found()); |
| 482 | }; |
| 483 | let session = updated.at(now); |
| 484 | self.link_fields(&session, &fields).await?; |
| 485 | if finished { |
| 486 | return Ok(Outcome::Ok(session)); |
| 487 | } |
| 488 | self.renew_claims(&id, now).await?; |
| 489 | if let Some(parent) = &parent_id { |
| 490 | self.db.prepare(SEE_PARENT).bind(&[at.as_str().into(), parent.as_str().into()])?.run().await?; |
| 491 | } |
| 492 | if let Some(number) = session.pull |
| 493 | && had_pull != Some(number) |
| 494 | { |
| 495 | self.move_entries_onto_pull(&repo.id, &id, number).await?; |
| 496 | } |
| 497 | if was_ended { |
| 498 | let data = g1t_contracts::events::marked(SessionEvent::of(&session), Some(actor)); |
| 499 | self.delivered(sent, "session.started", &repo.id, Some(actor.id.clone()), data).await?; |
| 500 | } |
| 501 | Ok(Outcome::Ok(session)) |
| 502 | } |
| 503 | |
| 504 | /// Renews the active claims made from a session by their own length. |
| 505 | async fn renew_claims(&self, session_id: &str, now: u64) -> Result<()> { |
| 506 | let claims = self |
| 507 | .db |
| 508 | .prepare("SELECT id, ttl_minutes, expires_at FROM claims WHERE session_id = ? AND state = 'active'") |
| 509 | .bind(&[session_id.into()])? |
| 510 | .all() |
| 511 | .await? |
| 512 | .results::<RenewRow>()?; |
| 513 | let mut statements = Vec::new(); |
| 514 | for claim in claims { |
| 515 | if let Some(renewed) = renewal(now, claim.ttl_minutes, &claim.expires_at) { |
| 516 | statements.push(self.db.prepare(RENEW_CLAIM).bind(&[renewed.into(), claim.id.into()])?); |
| 517 | } |
| 518 | } |
| 519 | if !statements.is_empty() { |
| 520 | self.db.batch(statements).await?; |
| 521 | } |
| 522 | Ok(()) |
| 523 | } |
| 524 | |
| 525 | /// Entries a session recorded before it named its pull request join |
| 526 | /// that pull request's session, after what it has, in their order. |
| 527 | async fn move_entries_onto_pull(&self, repo_id: &str, session_id: &str, number: u32) -> Result<()> { |
| 528 | let Some(pull) = self.pull(repo_id, number).await? else { |
| 529 | return Ok(()); |
| 530 | }; |
| 531 | let base = self |
| 532 | .db |
| 533 | .prepare("SELECT COALESCE(MAX(seq), 0) AS n FROM session_entries WHERE pull_id = ?") |
| 534 | .bind(&[pull.id.as_str().into()])? |
| 535 | .first::<NumberRow>(None) |
| 536 | .await? |
| 537 | .map_or(0, |row| row.n); |
| 538 | let moved = self |
| 539 | .db |
| 540 | .prepare(MOVE_ENTRIES) |
| 541 | .bind(&[pull.id.as_str().into(), base.into(), session_id.into()])? |
| 542 | .run() |
| 543 | .await; |
| 544 | // An entry recorded on the pull request at the same moment took a |
| 545 | // number these needed: they stay the session's alone. |
| 546 | if let Err(error) = moved { |
| 547 | worker::console_error!("session {session_id}: entries not moved onto #{number}: {error}"); |
| 548 | } |
| 549 | Ok(()) |
| 550 | } |
| 551 | |
| 552 | /// The caller's own session, as stored. |
| 553 | async fn own_session(&self, repo: &Repo, actor: &User, id: &str) -> Result<Outcome<SessionRow>> { |
| 554 | Ok(match self.session_row(&repo.id, id.trim()).await? { |
| 555 | Some(row) if row.actor_id == actor.id => Outcome::Ok(row), |
| 556 | _ => not_found(), |
| 557 | }) |
| 558 | } |
| 559 | |
| 560 | pub(crate) async fn update_session(&self, a: UpdateSessionArgs) -> Result<Outcome<Session>> { |
| 561 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 562 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 563 | if row.state == "ended" && !revives(row.end_reason.as_deref()) && !finished_subagent(&row) { |
| 564 | return Ok(Outcome::fail(FailureCode::Conflict, "This session has ended. Start a new one.")); |
| 565 | } |
| 566 | let (item, pull) = unwrap_or_fail!(self.session_subjects(&repo, a.item, a.pull).await?); |
| 567 | let fields = Fields { |
| 568 | item, |
| 569 | pull, |
| 570 | branch: tidy(a.branch.as_deref(), MAX_BRANCH_CHARS), |
| 571 | head_commit: tidy(a.head_commit.as_deref(), MAX_COMMIT_CHARS), |
| 572 | title: tidy(a.title.as_deref().map(shown::clean).as_deref(), MAX_TITLE_CHARS), |
| 573 | agent_type: tidy(a.agent_type.as_deref(), MAX_AGENT_CHARS), |
| 574 | task: clip(a.task.as_deref().map(shown::clean).as_deref(), MAX_TASK_CHARS), |
| 575 | prompt: clip(a.prompt.as_deref().map(shown::clean).as_deref(), MAX_PROMPT_CHARS), |
| 576 | model: tidy(a.model.as_deref(), MAX_MODEL_CHARS), |
| 577 | result: clip(a.result.as_deref().map(shown::clean).as_deref(), MAX_RESULT_CHARS), |
| 578 | }; |
| 579 | // Its usage first, so the session returned carries the cost. |
| 580 | if let Some(usage) = &a.usage { |
| 581 | let usage = match g1t_contracts::model_prices::parse_usage(&serde_json::to_value(usage).unwrap_or_default()) { |
| 582 | Ok(usage) => usage, |
| 583 | Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)), |
| 584 | }; |
| 585 | let cost = g1t_contracts::model_prices::estimate(&usage).cost_usd; |
| 586 | let json = serde_json::to_string(&usage).unwrap_or_else(|_| "[]".to_owned()); |
| 587 | self.db |
| 588 | .prepare(SET_USAGE) |
| 589 | .bind(&[json.into(), cost.map_or(JsValue::NULL, JsValue::from), row.id.as_str().into()])? |
| 590 | .run() |
| 591 | .await?; |
| 592 | } |
| 593 | self.resume(&repo, row, &a.actor, fields).await |
| 594 | } |
| 595 | |
| 596 | /// A session `actor` may end: their own, or anyone's when they have |
| 597 | /// write access to the repository (one left behind by a harness that |
| 598 | /// was killed or never said it ended). |
| 599 | async fn endable_session(&self, repo: &Repo, actor: &User, id: &str) -> Result<Outcome<SessionRow>> { |
| 600 | let Some(row) = self.session_row(&repo.id, id.trim()).await? else { |
| 601 | return Ok(not_found()); |
| 602 | }; |
| 603 | let can_push = matches!(crate::allowed(Some(actor), repo, g1t_contracts::access::Capability::Push), Outcome::Ok(())); |
| 604 | Ok(if may_end_session(&row.actor_id, &actor.id, can_push) { |
| 605 | Outcome::Ok(row) |
| 606 | } else { |
| 607 | Outcome::fail(FailureCode::Forbidden, "Only whoever runs this session, or someone with write access to the repository, can end it.") |
| 608 | }) |
| 609 | } |
| 610 | |
| 611 | pub(crate) async fn end_session(&self, a: EndSessionArgs) -> Result<Outcome<Session>> { |
| 612 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 613 | let row = unwrap_or_fail!(self.endable_session(&repo, &a.actor, &a.id).await?); |
| 614 | let own = row.actor_id == a.actor.id; |
| 615 | let now = now_ms(); |
| 616 | let at = rfc3339(now); |
| 617 | // What it reported is its own to say: someone else ending it says nothing for it. |
| 618 | let result = if own { clip(a.result.as_deref(), MAX_RESULT_CHARS) } else { None }; |
| 619 | if row.state == "ended" { |
| 620 | // Ended already: it keeps what it reported, or takes it now. |
| 621 | if row.result.is_none() |
| 622 | && let Some(result) = &result |
| 623 | { |
| 624 | let updated = self |
| 625 | .db |
| 626 | .prepare(format!("UPDATE sessions SET result = ? WHERE id = ? RETURNING {SESSION_COLUMNS}")) |
| 627 | .bind(&[result.as_str().into(), row.id.as_str().into()])? |
| 628 | .first::<SessionRow>(None) |
| 629 | .await?; |
| 630 | if let Some(updated) = updated { |
| 631 | return Ok(Outcome::Ok(updated.at(now))); |
| 632 | } |
| 633 | } |
| 634 | return Ok(Outcome::Ok(row.at(now))); |
| 635 | } |
| 636 | let reason = session_end_reason(own, a.by_hand, &a.actor.username, tidy(a.reason.as_deref(), MAX_NOTE_CHARS).as_deref()); |
| 637 | let ended_event = |session: &Session| g1t_contracts::events::marked(SessionEvent::of(session), Some(&a.actor)); |
| 638 | // Its subagents still going end with it, all in one change with |
| 639 | // their updates. (One started meanwhile ends with it too, and is |
| 640 | // announced on its own below.) |
| 641 | let going: Vec<SessionRow> = self |
| 642 | .db |
| 643 | .prepare(tree_sql()) |
| 644 | .bind(&[row.id.as_str().into()])? |
| 645 | .all() |
| 646 | .await? |
| 647 | .results::<SessionRow>()? |
| 648 | .into_iter() |
| 649 | .filter(|child| child.state == "active") |
| 650 | .collect(); |
| 651 | let end_one = |session: SessionRow, reason: &str, result: Option<String>| { |
| 652 | SessionRow { |
| 653 | state: "ended".to_owned(), |
| 654 | ended_at: Some(at.clone()), |
| 655 | end_reason: Some(reason.to_owned()), |
| 656 | last_seen_at: at.clone(), |
| 657 | result: result.or(session.result.clone()), |
| 658 | ..session |
| 659 | } |
| 660 | .at(now) |
| 661 | }; |
| 662 | let guard = "EXISTS (SELECT 1 FROM sessions WHERE id = ?7 AND state = 'ended' AND ended_at = ?8)"; |
| 663 | let mut statements = vec![self |
| 664 | .db |
| 665 | .prepare(format!( |
| 666 | "UPDATE sessions SET state = 'ended', ended_at = ?1, end_reason = ?2, last_seen_at = ?1, result = COALESCE(?3, result), |
| 667 | {final_due} |
| 668 | WHERE id = ?4 AND state = 'active' |
| 669 | RETURNING {SESSION_COLUMNS}", |
| 670 | final_due = titles::final_due("?1"), |
| 671 | )) |
| 672 | .bind(&[at.as_str().into(), reason.as_str().into(), text_or_null(&result), row.id.as_str().into()])?]; |
| 673 | let expected = end_one(row.clone(), &reason, result.clone()); |
| 674 | let sent = push_delivery( |
| 675 | &mut statements, |
| 676 | self.delivery( |
| 677 | "session.ended", |
| 678 | &repo.id, |
| 679 | ended_event(&expected), |
| 680 | &format!("{}:ended:{at}", row.id), |
| 681 | guard, |
| 682 | vec![row.id.as_str().into(), at.as_str().into()], |
| 683 | )?, |
| 684 | ); |
| 685 | // Only once it ended here, so a call that lost the race ends nothing. |
| 686 | statements.push(self.db.prepare(end_tree_sql()).bind(&[row.id.as_str().into(), at.as_str().into(), "parent ended".into()])?); |
| 687 | let tree_index = statements.len() - 1; |
| 688 | let mut orphans_sent = Vec::new(); |
| 689 | for child in going { |
| 690 | let child_id = child.id.clone(); |
| 691 | let expected = end_one(child, "parent ended", None); |
| 692 | let delivery = self.delivery( |
| 693 | "session.ended", |
| 694 | &repo.id, |
| 695 | ended_event(&expected), |
| 696 | &format!("{child_id}:ended:{at}"), |
| 697 | guard, |
| 698 | vec![child_id.as_str().into(), at.as_str().into()], |
| 699 | )?; |
| 700 | orphans_sent.push((child_id, push_delivery(&mut statements, delivery))); |
| 701 | } |
| 702 | let mut results = self.db.batch(statements).await?; |
| 703 | let orphans: Vec<SessionRow> = match results.get(tree_index) { |
| 704 | Some(result) => result.results()?, |
| 705 | None => Vec::new(), |
| 706 | }; |
| 707 | let ended = match results.drain(..).next() { |
| 708 | Some(result) => result.results::<SessionRow>()?.into_iter().next(), |
| 709 | None => None, |
| 710 | }; |
| 711 | let Some(ended) = ended else { |
| 712 | // Ended by another call meanwhile. |
| 713 | return Ok(match self.session_row(&repo.id, &row.id).await? { |
| 714 | Some(row) => Outcome::Ok(row.at(now)), |
| 715 | None => not_found(), |
| 716 | }); |
| 717 | }; |
| 718 | let session = ended.at(now); |
| 719 | if let Some(result) = &result { |
| 720 | self.link_refs(&session.id, &session.repo_id, &result.chars().take(MAX_REF_TEXT).collect::<String>()).await?; |
| 721 | } |
| 722 | self.delivered(sent, "session.ended", &repo.id, Some(a.actor.id.clone()), ended_event(&session)).await?; |
| 723 | for orphan in orphans { |
| 724 | let orphan = orphan.at(now); |
| 725 | match orphans_sent.iter().position(|(id, _)| *id == orphan.id) { |
| 726 | Some(index) => { |
| 727 | let (_, sent) = orphans_sent.swap_remove(index); |
| 728 | self.delivered(sent, "session.ended", &repo.id, Some(a.actor.id.clone()), ended_event(&orphan)).await?; |
| 729 | } |
| 730 | None => self.publish("session.ended", &repo.id, &a.actor, SessionEvent::of(&orphan)).await?, |
| 731 | } |
| 732 | } |
| 733 | Ok(Outcome::Ok(session)) |
| 734 | } |
| 735 | |
| 736 | /// A person names a session from the site (`TitleSessionArgs`): pinned, |
| 737 | /// or with no title, unpinned for g1t to name again. |
| 738 | pub(crate) async fn title_session(&self, a: TitleSessionArgs) -> Result<Outcome<Session>> { |
| 739 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 740 | let Some(row) = self.session_row(&repo.id, a.id.trim()).await? else { |
| 741 | return Ok(not_found()); |
| 742 | }; |
| 743 | let can_push = matches!(crate::allowed(Some(&a.actor), &repo, g1t_contracts::access::Capability::Push), Outcome::Ok(())); |
| 744 | if !may_end_session(&row.actor_id, &a.actor.id, can_push) { |
| 745 | return Ok(Outcome::fail(FailureCode::Forbidden, "Only whoever runs this session, or someone with write access to the repository, can name it.")); |
| 746 | } |
| 747 | let at = rfc3339(now_ms()); |
| 748 | let title = tidy(a.title.as_deref().map(shown::clean).as_deref(), MAX_TITLE_CHARS); |
| 749 | let statement = match &title { |
| 750 | Some(title) => self.db.prepare(titles::PIN_TITLE).bind(&[title.as_str().into(), at.as_str().into(), row.id.as_str().into()])?, |
| 751 | None => self.db.prepare(titles::UNPIN_TITLE).bind(&[at.as_str().into(), row.id.as_str().into()])?, |
| 752 | }; |
| 753 | statement.run().await?; |
| 754 | let Some(row) = self.session_row(&repo.id, &row.id).await? else { |
| 755 | return Ok(not_found()); |
| 756 | }; |
| 757 | let session = row.at(now_ms()); |
| 758 | self.publish_as( |
| 759 | "session.progressed", |
| 760 | &repo.id, |
| 761 | Some(a.actor.id.clone()), |
| 762 | SessionProgressed { repo_id: repo.id.clone(), number: session.pull.or(session.item), session_id: session.id.clone(), root_id: None }, |
| 763 | ) |
| 764 | .await?; |
| 765 | let mut sessions = [session]; |
| 766 | shown::sessions(&mut sessions, &Some(a.actor.clone()), &repo.namespace); |
| 767 | let [session] = sessions; |
| 768 | Ok(Outcome::Ok(session)) |
| 769 | } |
| 770 | |
| 771 | pub(crate) async fn get_work_session(&self, a: WorkSessionArgs) -> Result<Outcome<SessionDetail>> { |
| 772 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 773 | let Some(row) = self.session_row(&repo.id, a.id.trim()).await? else { |
| 774 | return Ok(not_found()); |
| 775 | }; |
| 776 | let now = now_ms(); |
| 777 | let entries = self |
| 778 | .db |
| 779 | .prepare( |
| 780 | "SELECT session_seq AS seq, kind, text, tool, \"commit\", at FROM session_entries |
| 781 | WHERE session_id = ? AND session_seq > ? ORDER BY session_seq LIMIT ?", |
| 782 | ) |
| 783 | .bind(&[row.id.as_str().into(), a.after_seq.into(), ENTRY_PAGE.into()])? |
| 784 | .all() |
| 785 | .await? |
| 786 | .results::<crate::rows::SessionRow>()? |
| 787 | .into_iter() |
| 788 | .map(SessionEntry::from) |
| 789 | .collect(); |
| 790 | let mut session = row.at(now); |
| 791 | let children = self.subagents_of(&session.id, now).await?; |
| 792 | session.subagents = children.iter().filter(|child| child.parent_id.as_deref() == Some(session.id.as_str())).count() as u32; |
| 793 | // A subagent's page names the session at the top of its tree. |
| 794 | let root = match &session.parent_id { |
| 795 | Some(_) => { |
| 796 | let mut top = None; |
| 797 | let mut at = session.parent_id.clone(); |
| 798 | for _ in 0..8 { |
| 799 | let Some(id) = at else { break }; |
| 800 | let Some(found) = self.session_row(&repo.id, &id).await? else { break }; |
| 801 | at = found.parent_id.clone(); |
| 802 | top = Some(found.at(now)); |
| 803 | } |
| 804 | top |
| 805 | } |
| 806 | None => None, |
| 807 | }; |
| 808 | let mut ids: Vec<String> = vec![session.id.clone()]; |
| 809 | ids.extend(children.iter().map(|child| child.id.clone())); |
| 810 | let (touched, links, skills) = self.tree_facts(&repo.id, &ids).await?; |
| 811 | let updates = self.pending_updates(&session.id).await?; |
| 812 | let detail = SessionDetail { session, entries, children, touched, links, skills, root, updates, limited: false }; |
| 813 | Ok(Outcome::Ok(shown::detail(detail, &a.viewer, &repo.namespace))) |
| 814 | } |
| 815 | |
| 816 | /// The sessions tied to some pull requests, for Home (`PullSessionsArgs`). |
| 817 | pub(crate) async fn pull_sessions(&self, a: PullSessionsArgs) -> Result<Outcome<Vec<PullSessions>>> { |
| 818 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 819 | let mut numbers = a.numbers; |
| 820 | numbers.sort_unstable(); |
| 821 | numbers.dedup(); |
| 822 | numbers.truncate(MAX_PULL_SESSIONS); |
| 823 | if numbers.is_empty() { |
| 824 | return Ok(Outcome::Ok(Vec::new())); |
| 825 | } |
| 826 | let rows = self |
| 827 | .db |
| 828 | .prepare(PULL_SESSIONS) |
| 829 | .bind(&[repo.id.as_str().into(), serde_json::to_string(&numbers)?.into()])? |
| 830 | .all() |
| 831 | .await? |
| 832 | .results::<PullSessionRow>()?; |
| 833 | Ok(Outcome::Ok(pull_sessions_of(rows))) |
| 834 | } |
| 835 | |
| 836 | pub(crate) async fn list_work_sessions(&self, a: ListWorkSessionsArgs) -> Result<Outcome<Vec<Session>>> { |
| 837 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 838 | let now = now_ms(); |
| 839 | let idle_since = rfc3339(now.saturating_sub(SESSION_IDLE_MS)); |
| 840 | // Every statement names each bind it is given. |
| 841 | let state = match a.state { |
| 842 | None => "AND ?2 IS NOT NULL", |
| 843 | Some(SessionState::Active) => "AND state = 'active' AND last_seen_at > ?2", |
| 844 | Some(SessionState::Idle) => "AND state = 'active' AND last_seen_at <= ?2", |
| 845 | Some(SessionState::Ended) => "AND state = 'ended' AND ?2 IS NOT NULL", |
| 846 | }; |
| 847 | // On a work item or pull request: attached, or tied to it any way. |
| 848 | let mut linked: Vec<String> = Vec::new(); |
| 849 | let mut item_id = None; |
| 850 | if let Some(number) = a.item { |
| 851 | let Some(held) = items::by_number(self, &repo.namespace, number).await? else { |
| 852 | return Ok(Outcome::Ok(Vec::new())); |
| 853 | }; |
| 854 | linked.extend(self.item_sessions(&repo.id, &held.item.id).await?); |
| 855 | // And those on its pull requests here. |
| 856 | let pulls: Vec<u32> = self |
| 857 | .db |
| 858 | .prepare("SELECT number AS n FROM pulls WHERE repo_id = ? AND item_id = ?") |
| 859 | .bind(&[repo.id.as_str().into(), held.item.id.as_str().into()])? |
| 860 | .all() |
| 861 | .await? |
| 862 | .results::<NumberRow>()? |
| 863 | .into_iter() |
| 864 | .map(|row| row.n) |
| 865 | .collect(); |
| 866 | for pull in &pulls { |
| 867 | linked.extend(self.linked_sessions(&repo.id, *pull).await?); |
| 868 | } |
| 869 | linked.extend(self.branch_sessions(&repo.id, &pulls).await?); |
| 870 | item_id = Some(held.item.id); |
| 871 | } |
| 872 | if let Some(number) = a.pull { |
| 873 | linked.extend(self.linked_sessions(&repo.id, number).await?); |
| 874 | linked.extend(self.branch_sessions(&repo.id, &[number]).await?); |
| 875 | } |
| 876 | let rows = self |
| 877 | .db |
| 878 | .prepare(format!( |
| 879 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 880 | WHERE repo_id = ?1 {state} |
| 881 | AND (?6 = 1 OR parent_id IS NULL OR id IN (SELECT value FROM json_each(?7))) |
| 882 | AND (?3 IS NULL OR item_id = ?3 OR id IN (SELECT value FROM json_each(?7))) |
| 883 | AND (?4 IS NULL OR pull = ?4 OR id IN (SELECT value FROM json_each(?7))) |
| 884 | AND (?8 IS NULL OR id IN (SELECT session_id FROM session_touches WHERE repo_id = ?1 AND kind = 'file' |
| 885 | AND (value = ?8 OR (substr(?8, -1) = '/' AND substr(value, 1, length(?8)) = ?8)))) |
| 886 | ORDER BY last_seen_at DESC LIMIT ?5" |
| 887 | )) |
| 888 | .bind(&[ |
| 889 | repo.id.as_str().into(), |
| 890 | idle_since.as_str().into(), |
| 891 | text_or_null(&item_id), |
| 892 | number_or_null(a.pull), |
| 893 | LIST_LIMIT.into(), |
| 894 | u32::from(a.subagents).into(), |
| 895 | serde_json::to_string(&linked)?.into(), |
| 896 | text_or_null(&tidy(a.path.as_deref().map(|path| path.trim_start_matches("./")), MAX_TOUCH_CHARS)), |
| 897 | ])? |
| 898 | .all() |
| 899 | .await? |
| 900 | .results::<SessionRow>()?; |
| 901 | let mut sessions: Vec<Session> = rows.into_iter().map(|row| row.at(now)).collect(); |
| 902 | // With subagents, each tree whole: its subagents, and its top. |
| 903 | if a.subagents { |
| 904 | let mut more: Vec<Session> = Vec::new(); |
| 905 | for session in sessions.iter().filter(|session| session.parent_id.is_none()).take(20) { |
| 906 | more.extend(self.subagents_of(&session.id, now).await?); |
| 907 | } |
| 908 | let parents: Vec<String> = sessions |
| 909 | .iter() |
| 910 | .filter_map(|session| session.parent_id.clone()) |
| 911 | .filter(|parent| !sessions.iter().any(|session| &session.id == parent)) |
| 912 | .collect(); |
| 913 | more.extend(self.sessions_by_id(&repo.id, &parents).await?); |
| 914 | for session in more { |
| 915 | if !sessions.iter().any(|known| known.id == session.id) { |
| 916 | sessions.push(session); |
| 917 | } |
| 918 | } |
| 919 | count_subagents(&mut sessions); |
| 920 | } else { |
| 921 | self.fill_subagents(&mut sessions).await?; |
| 922 | } |
| 923 | self.fill_first_prompts(&mut sessions).await?; |
| 924 | shown::sessions(&mut sessions, &a.viewer, &repo.namespace); |
| 925 | Ok(Outcome::Ok(sessions)) |
| 926 | } |
| 927 | |
| 928 | /// Notes `text` (from g1t) in the live harness sessions on `pull` or |
| 929 | /// its work item, so an agent reading its session hears of it. Never |
| 930 | /// fails what it follows: a problem is logged. |
| 931 | pub(crate) async fn note_sessions(&self, pull: &Pull, text: &str) { |
| 932 | let noted = async { |
| 933 | let sessions: Vec<SessionRow> = self |
| 934 | .db |
| 935 | .prepare(format!( |
| 936 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 937 | WHERE state = 'active' AND ((repo_id = ?1 AND pull = ?2) |
| 938 | OR item_id = (SELECT item_id FROM pulls WHERE repo_id = ?1 AND number = ?2))" |
| 939 | )) |
| 940 | .bind(&[pull.repo_id.as_str().into(), pull.number.into()])? |
| 941 | .all() |
| 942 | .await? |
| 943 | .results()?; |
| 944 | let at = rfc3339(now_ms()); |
| 945 | let mut statements = Vec::new(); |
| 946 | for session in sessions { |
| 947 | // Kept by the session alone: the pull request's own session |
| 948 | // has the note already. |
| 949 | statements.push(self.db.prepare(INSERT_ENTRY).bind(&[ |
| 950 | JsValue::NULL, |
| 951 | "note".into(), |
| 952 | text.into(), |
| 953 | JsValue::NULL, |
| 954 | text_or_null(&pull.head_commit), |
| 955 | at.as_str().into(), |
| 956 | session.id.as_str().into(), |
| 957 | JsValue::NULL, |
| 958 | ])?); |
| 959 | } |
| 960 | if !statements.is_empty() { |
| 961 | self.db.batch(statements).await?; |
| 962 | } |
| 963 | Ok::<(), worker::Error>(()) |
| 964 | }; |
| 965 | if let Err(error) = noted.await { |
| 966 | worker::console_error!("#{}: sessions were not told: {error}", pull.number); |
| 967 | } |
| 968 | } |
| 969 | |
| 970 | pub(crate) async fn record_session_entries(&self, a: RecordEntriesArgs) -> Result<Outcome<Recorded>> { |
| 971 | if a.acknowledge.len() > 100 || a.acknowledge.iter().any(|id| id.len() > 100) { |
| 972 | return Ok(Outcome::fail(FailureCode::Invalid, "Acknowledge at most 100 update IDs.")); |
| 973 | } |
| 974 | if a.entries.len() > crate::MAX_ENTRY_BATCH { |
| 975 | return Ok(Outcome::fail( |
| 976 | FailureCode::Invalid, |
| 977 | format!("Send at most {} entries at a time.", crate::MAX_ENTRY_BATCH), |
| 978 | )); |
| 979 | } |
| 980 | if a.touched.len() > MAX_TOUCHES { |
| 981 | return Ok(Outcome::fail(FailureCode::Invalid, format!("Say at most {MAX_TOUCHES} things touched at a time."))); |
| 982 | } |
| 983 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 984 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 985 | if row.state == "ended" && !revives(row.end_reason.as_deref()) { |
| 986 | return Ok(Outcome::fail(FailureCode::Conflict, "This session has ended. Start a new one.")); |
| 987 | } |
| 988 | // Recording is a heartbeat too. |
| 989 | let session = unwrap_or_fail!(self.resume(&repo, row, &a.actor, Fields::default()).await?); |
| 990 | let touched = self.record_touches(&session, &a.touched).await?; |
| 991 | if a.entries.is_empty() { |
| 992 | if touched > 0 { |
| 993 | self.announce_progress(&session).await?; |
| 994 | } |
| 995 | let acknowledged = self.acknowledge_updates(&session.id, &a.acknowledge).await?; |
| 996 | return Ok(Outcome::Ok(Recorded { acknowledged, ..Recorded::default() })); |
| 997 | } |
| 998 | let pull = match session.pull { |
| 999 | Some(number) => self.pull(&repo.id, number).await?, |
| 1000 | None => None, |
| 1001 | }; |
| 1002 | let pull_id: JsValue = pull.as_ref().map_or(JsValue::NULL, |pull| pull.id.as_str().into()); |
| 1003 | let at = rfc3339(now_ms()); |
| 1004 | let total = a.entries.len() as u32; |
| 1005 | let mut statements = Vec::with_capacity(a.entries.len()); |
| 1006 | let mut prompts = String::new(); |
| 1007 | // Each prompt's place among the statements, and what the first one with one is for. |
| 1008 | let mut prompt_at: Vec<usize> = Vec::new(); |
| 1009 | let mut intent: Option<String> = None; |
| 1010 | for keyed in a.entries { |
| 1011 | let entry = keyed.entry; |
| 1012 | let kind = serde_json::to_value(entry.kind)?; |
| 1013 | // Kept without credentials: a session's page may show it to a |
| 1014 | // whole team, and its prompts to its author on any device. |
| 1015 | let mut text: String = shown::clean(&entry.text.chars().take(crate::MAX_ENTRY_CHARS).collect::<String>()); |
| 1016 | if entry.kind == SessionEntryKind::Skill { |
| 1017 | // A skill is its name, nothing more. |
| 1018 | text = text.trim().chars().take(MAX_AGENT_CHARS).collect(); |
| 1019 | } |
| 1020 | if entry.kind == SessionEntryKind::Prompt { |
| 1021 | if prompts.len() < MAX_REF_TEXT { |
| 1022 | prompts.push_str(&text); |
| 1023 | prompts.push('\n'); |
| 1024 | } |
| 1025 | prompt_at.push(statements.len()); |
| 1026 | if intent.is_none() { |
| 1027 | intent = titles::intent_title(&text); |
| 1028 | } |
| 1029 | } |
| 1030 | let key = tidy(keyed.key.as_deref(), MAX_KEY_CHARS); |
| 1031 | // Each takes the next number of the session, and of its pull |
| 1032 | // request when it has one, itself, so two batches at once |
| 1033 | // cannot collide; a key already recorded is skipped. |
| 1034 | statements.push( |
| 1035 | self.db |
| 1036 | .prepare(INSERT_ENTRY) |
| 1037 | .bind(&[ |
| 1038 | pull_id.clone(), |
| 1039 | kind.as_str().unwrap_or("note").into(), |
| 1040 | text.into(), |
| 1041 | text_or_null(&entry.tool), |
| 1042 | text_or_null(&entry.commit.or_else(|| session.head_commit.clone())), |
| 1043 | at.as_str().into(), |
| 1044 | session.id.as_str().into(), |
| 1045 | text_or_null(&key), |
| 1046 | ])?, |
| 1047 | ); |
| 1048 | } |
| 1049 | let mut recorded = 0; |
| 1050 | let mut prompted = 0; |
| 1051 | for (index, result) in self.db.batch(statements).await?.into_iter().enumerate() { |
| 1052 | let new = result.results::<NumberRow>()?.len() as u32; |
| 1053 | recorded += new; |
| 1054 | if prompt_at.contains(&index) { |
| 1055 | prompted += new; |
| 1056 | } |
| 1057 | } |
| 1058 | if prompted > 0 { |
| 1059 | // The first prompt names the session for now; enough of them settle its title (titles.rs). |
| 1060 | let now = now_ms(); |
| 1061 | self.db |
| 1062 | .prepare(titles::prompted_sql()) |
| 1063 | .bind(&[ |
| 1064 | session.id.as_str().into(), |
| 1065 | prompted.into(), |
| 1066 | text_or_null(&intent), |
| 1067 | rfc3339(now).into(), |
| 1068 | after_minutes(now, titles::SETTLE_AFTER_MINUTES).into(), |
| 1069 | ])? |
| 1070 | .run() |
| 1071 | .await?; |
| 1072 | } |
| 1073 | if !prompts.is_empty() && recorded > 0 { |
| 1074 | self.link_refs(&session.id, &repo.id, &prompts).await?; |
| 1075 | } |
| 1076 | if let Some(pull) = pull |
| 1077 | && recorded > 0 |
| 1078 | { |
| 1079 | self.publish( |
| 1080 | "session.appended", |
| 1081 | &repo.id, |
| 1082 | &a.actor, |
| 1083 | SessionAppended { pull_id: pull.id, repo_id: repo.id.clone(), number: pull.number, count: recorded }, |
| 1084 | ) |
| 1085 | .await?; |
| 1086 | } |
| 1087 | if recorded > 0 || touched > 0 { |
| 1088 | self.announce_progress(&session).await?; |
| 1089 | } |
| 1090 | let acknowledged = self.acknowledge_updates(&session.id, &a.acknowledge).await?; |
| 1091 | Ok(Outcome::Ok(Recorded { recorded, duplicates: total - recorded, acknowledged })) |
| 1092 | } |
| 1093 | |
| 1094 | /// `session.progressed`, at most every [`PROGRESS_EVERY_MS`] for a |
| 1095 | /// session, so the pages showing it follow along. |
| 1096 | async fn announce_progress(&self, session: &Session) -> Result<()> { |
| 1097 | let now = now_ms(); |
| 1098 | let due = self |
| 1099 | .db |
| 1100 | .prepare(MARK_PROGRESS) |
| 1101 | .bind(&[rfc3339(now).into(), session.id.as_str().into(), rfc3339(now.saturating_sub(PROGRESS_EVERY_MS)).into()])? |
| 1102 | .first::<serde_json::Value>(None) |
| 1103 | .await?; |
| 1104 | if due.is_none() { |
| 1105 | return Ok(()); |
| 1106 | } |
| 1107 | let mut root_id = None; |
| 1108 | let mut up = session.parent_id.clone(); |
| 1109 | for _ in 0..8 { |
| 1110 | let Some(id) = up else { break }; |
| 1111 | up = self.session_row(&session.repo_id, &id).await?.and_then(|row| row.parent_id); |
| 1112 | root_id = Some(id); |
| 1113 | } |
| 1114 | self.publish_as( |
| 1115 | "session.progressed", |
| 1116 | &session.repo_id, |
| 1117 | Some(session.actor.id.clone()), |
| 1118 | SessionProgressed { |
| 1119 | repo_id: session.repo_id.clone(), |
| 1120 | number: session.pull.or(session.item), |
| 1121 | session_id: session.id.clone(), |
| 1122 | root_id, |
| 1123 | }, |
| 1124 | ) |
| 1125 | .await |
| 1126 | } |
| 1127 | } |