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