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