Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
736 linesCodeBlameRaw
1//! Sessions across a workspace (`g1t_contracts::work::workspace_sessions`):
2//! every kind of session in every repository of it the viewer may read,
3//! in one list, with the merge queues waiting and what just landed.
4//!
5//! Harness and hosted sessions are the `sessions` table's; g1t's agent is
6//! `agent_runs`; a recorded one is a pull request with session entries
7//! and nothing else behind it. Who may see what inside a session is the
8//! repository's rule (shown.rs), applied to each as it is in its own list.
9
10use std::collections::{HashMap, HashSet};
11
12use g1t_contracts::agents::AgentRun;
13use g1t_contracts::repos::{ReadableArgs, Repo, RepoPath};
14use g1t_contracts::time::{parse_rfc3339, rfc3339};
15use g1t_contracts::work::*;
16use g1t_contracts::{FailureCode, Outcome, Viewer};
17use g1t_kit::now_ms;
18use serde::Deserialize;
19use worker::Result;
20use worker::wasm_bindgen::JsValue;
21
22use super::links::count_subagents;
23use super::{SESSION_COLUMNS, SessionRow, shown};
24use crate::Work;
25use crate::rows::{PULL_COLUMNS, PullRow};
26use crate::runs::RecordedRow;
27
28/// The most repositories of a workspace looked in.
29const MAX_REPOS: u32 = 500;
30/// The most pull requests Just landed gives.
31const LANDED: u32 = 8;
32/// The most merge queues given.
33const QUEUES: usize = 12;
34/// The most trees given whole with `subagents`.
35const TREES: usize = 30;
36
37/// The workspace's sessions in the repositories `?1` (a JSON array of
38/// ids): `?2` is the time before which a quiet one reads as ended, `?3`
39/// whether subagents are listed, `?4` a harness, `?5` a username, `?6` a
40/// work item's id, `?7` how many, `?8` a kind (`harness` or `hosted`).
41pub(crate) fn workspace_sessions_sql(shown: SessionsShown) -> String {
42 let state = match shown {
43 SessionsShown::Live => "AND state = 'active' AND last_seen_at > ?2",
44 SessionsShown::Recent => "AND ?2 IS NOT NULL",
45 SessionsShown::Ended => "AND (state = 'ended' OR last_seen_at <= ?2)",
46 };
47 format!(
48 "SELECT {SESSION_COLUMNS} FROM sessions
49 WHERE repo_id IN (SELECT value FROM json_each(?1)) {state}
50 AND (?3 = 1 OR parent_id IS NULL)
51 AND (?4 IS NULL OR harness = ?4)
52 AND (?5 IS NULL OR lower(actor_name) = lower(?5))
53 AND (?6 IS NULL OR item_id = ?6
54 OR id IN (SELECT session_id FROM session_links WHERE kind = 'item' AND item_id = ?6))
55 AND (?8 IS NULL OR (?8 = 'hosted') = (harness = 'claude_managed_agents'))
56 ORDER BY last_seen_at DESC LIMIT ?7"
57 )
58}
59
60/// The titles and states of some pull requests, by `{r, n}` pairs in `?1`.
61pub(crate) const PULL_FACTS: &str = "SELECT p.repo_id, p.number, p.title, p.status,
62 (SELECT number FROM work_items WHERE work_items.id = p.item_id) AS item
63 FROM json_each(?1) j JOIN pulls p ON p.repo_id = json_extract(j.value, '$.r') AND p.number = json_extract(j.value, '$.n')";
64
65/// The repositories of `?1` (a JSON array of ids) with something waiting
66/// in their merge queue, the busiest first.
67pub(crate) const QUEUED_REPOS: &str = "SELECT repo_id AS value FROM queue_entries
68 WHERE repo_id IN (SELECT value FROM json_each(?1)) AND state IN ('waiting', 'testing', 'passed')
69 GROUP BY repo_id ORDER BY COUNT(*) DESC, MIN(created_at) LIMIT ?2";
70
71#[derive(Debug, Deserialize)]
72pub(crate) struct PullFact {
73 pub(crate) repo_id: String,
74 pub(crate) number: u32,
75 pub(crate) title: String,
76 pub(crate) status: PullStatus,
77 #[serde(default)]
78 pub(crate) item: Option<u32>,
79}
80
81#[derive(Deserialize)]
82struct Value {
83 value: String,
84}
85
86#[derive(Deserialize)]
87struct Titled {
88 number: u32,
89 title: String,
90}
91
92#[derive(Deserialize)]
93struct Placed {
94 repo_id: String,
95 path: String,
96}
97
98/// A harness or hosted session, as one of the workspace's.
99pub(crate) fn from_session(mut session: Session, path: &str, now: u64) -> WorkspaceSession {
100 let last = parse_rfc3339(&session.last_seen_at).unwrap_or(now);
101 session.state = settled_state(session.state, last, now);
102 if session.state == SessionState::Ended && session.ended_at.is_none() {
103 session.end_reason.get_or_insert_with(|| "timed_out".to_owned());
104 }
105 WorkspaceSession {
106 kind: SessionKind::of(session.harness),
107 id: session.id.clone(),
108 repo: path.to_owned(),
109 repo_id: session.repo_id.clone(),
110 title: session.title.clone(),
111 actor: Some(session.actor.username.clone()),
112 harness: Some(session.harness),
113 agent: None,
114 branch: session.branch.clone(),
115 item: session.item,
116 item_title: None,
117 pull: session.pull,
118 pull_title: None,
119 pull_status: None,
120 state: session.state,
121 started_at: session.started_at.clone(),
122 last_activity_at: session.last_seen_at.clone(),
123 subagents: session.subagents,
124 step: None,
125 session: Some(session),
126 children: Vec::new(),
127 }
128}
129
130/// A run of g1t's agent, as one of the workspace's sessions.
131pub(crate) fn from_run(run: AgentRun, repo_id: &str) -> WorkspaceSession {
132 let state = if run.status.is_active() { SessionState::Active } else { SessionState::Ended };
133 WorkspaceSession {
134 kind: SessionKind::Agent,
135 id: run.id,
136 repo: format!("{}/{}", run.repo.namespace, run.repo.name),
137 repo_id: repo_id.to_owned(),
138 title: run.title,
139 actor: Some(run.started_by.unwrap_or_else(|| "g1t".to_owned())),
140 harness: None,
141 agent: Some(run.agent),
142 branch: None,
143 item: None,
144 item_title: None,
145 pull: run.number,
146 pull_title: None,
147 pull_status: None,
148 state,
149 started_at: run.started_at.unwrap_or_else(|| run.created_at.clone()),
150 last_activity_at: run.updated_at,
151 subagents: 0,
152 step: run.step,
153 session: None,
154 children: Vec::new(),
155 }
156}
157
158/// A pull request's recorded session, as one of the workspace's.
159pub(crate) fn from_recorded(row: RecordedRow, path: &str, now: u64) -> WorkspaceSession {
160 let last = parse_rfc3339(&row.last_at).unwrap_or(0);
161 WorkspaceSession {
162 kind: SessionKind::Recorded,
163 id: row.pull_id,
164 repo: path.to_owned(),
165 repo_id: row.repo_id,
166 title: Some(row.title.clone()),
167 actor: Some(row.author_name),
168 harness: None,
169 agent: Some(row.agent),
170 branch: None,
171 item: row.item,
172 item_title: None,
173 pull: Some(row.number),
174 pull_title: Some(row.title),
175 pull_status: Some(row.status),
176 state: quiet_state(false, last, now),
177 started_at: row.started_at,
178 last_activity_at: row.last_at,
179 subagents: 0,
180 step: None,
181 session: None,
182 children: Vec::new(),
183 }
184}
185
186/// Filters for the kinds that are not in the `sessions` table.
187pub(crate) struct Wanted<'a> {
188 pub(crate) shown: SessionsShown,
189 pub(crate) kind: Option<SessionKind>,
190 pub(crate) harness: Option<Harness>,
191 pub(crate) person: Option<&'a str>,
192 pub(crate) item: Option<u32>,
193}
194
195impl Wanted<'_> {
196 fn admits(&self, session: &WorkspaceSession) -> bool {
197 self.shown.admits(session.state)
198 && self.kind.is_none_or(|kind| kind == session.kind)
199 && self.harness.is_none_or(|harness| session.harness == Some(harness))
200 && self.person.is_none_or(|person| session.actor.as_deref().is_some_and(|actor| actor.eq_ignore_ascii_case(person)))
201 && self.item.is_none_or(|item| session.item == Some(item))
202 }
203}
204
205/// One list of every kind: a recorded session of a pull request that a
206/// harness session or a run of g1t's agent is already on is that one,
207/// and is left out. Filtered, most recently heard from first, at most
208/// `limit`.
209pub(crate) fn merged(
210 sessions: Vec<WorkspaceSession>,
211 runs: Vec<WorkspaceSession>,
212 recorded: Vec<(WorkspaceSession, bool)>,
213 wanted: &Wanted,
214 limit: usize,
215) -> Vec<WorkspaceSession> {
216 let covered: HashSet<(String, u32)> =
217 sessions.iter().chain(&runs).filter_map(|session| session.pull.map(|pull| (session.repo_id.clone(), pull))).collect();
218 let mut all: Vec<WorkspaceSession> = sessions.into_iter().filter(|session| wanted.admits(session)).collect();
219 all.extend(runs.into_iter().filter(|run| wanted.admits(run)));
220 all.extend(
221 recorded
222 .into_iter()
223 .filter(|(session, has_runs)| !has_runs && !session.pull.is_some_and(|pull| covered.contains(&(session.repo_id.clone(), pull))))
224 .map(|(session, _)| session)
225 .filter(|session| wanted.admits(session)),
226 );
227 all.sort_by(|a, b| b.last_activity_at.cmp(&a.last_activity_at).then_with(|| a.id.cmp(&b.id)));
228 all.truncate(limit);
229 all
230}
231
232impl Work {
233 /// The repositories of `workspace` the viewer may read, or of them the
234 /// one named `only`.
235 async fn readable_in(&self, workspace: &str, viewer: &Viewer, only: Option<&str>) -> Result<Vec<Repo>> {
236 let placed = self
237 .db
238 .prepare("SELECT repo_id, path FROM repo_workspaces WHERE workspace = ?1 AND (?2 IS NULL OR lower(path) = lower(?1 || '/' || ?2)) ORDER BY path LIMIT ?3")
239 .bind(&[workspace.into(), only.map_or(JsValue::NULL, JsValue::from), MAX_REPOS.into()])?
240 .all()
241 .await?
242 .results::<Placed>()?;
243 if placed.is_empty() {
244 return Ok(Vec::new());
245 }
246 let ids: Vec<String> = placed.iter().map(|row| row.repo_id.clone()).collect();
247 let mut readable: Vec<Repo> = g1t_kit::call(&self.repos, "readable", &ReadableArgs { ids, viewer: viewer.clone() }).await?;
248 readable.retain(|repo| repo.namespace.eq_ignore_ascii_case(workspace));
249 readable.sort_by(|a, b| a.name.cmp(&b.name));
250 Ok(readable)
251 }
252
253 pub(crate) async fn list_workspace_sessions(&self, a: ListWorkspaceSessionsArgs) -> Result<Outcome<WorkspaceSessions>> {
254 let workspace = a.workspace.trim().to_lowercase();
255 if workspace.is_empty() {
256 return Ok(Outcome::fail(FailureCode::Invalid, "Name the workspace by its slug."));
257 }
258 let only = a.repo.as_deref().map(str::trim).filter(|name| !name.is_empty()).map(|name| name.rsplit('/').next().unwrap_or(name).to_owned());
259 let repos = self.readable_in(&workspace, &a.viewer, only.as_deref()).await?;
260 if repos.is_empty() && only.is_some() {
261 return Ok(Outcome::fail(FailureCode::NotFound, "Repository not found."));
262 }
263 let ids: Vec<String> = repos.iter().map(|repo| repo.id.clone()).collect();
264 let path_of: HashMap<&str, String> = repos.iter().map(|repo| (repo.id.as_str(), format!("{}/{}", repo.namespace, repo.name))).collect();
265 let member = a.viewer.as_ref().is_some_and(|viewer| viewer.is_member(&workspace));
266 let limit = a.limit.unwrap_or(DEFAULT_WORKSPACE_SESSIONS).clamp(1, MAX_WORKSPACE_SESSIONS);
267 let now = now_ms();
268 let person = a.person.as_deref().map(|person| person.trim().trim_start_matches('@')).filter(|person| !person.is_empty());
269 // A work item by number, in this workspace.
270 let item_id = match a.item {
271 Some(number) => match crate::items::by_number(self, &workspace, number).await? {
272 Some(held) => Some(held.item.id),
273 None => return Ok(Outcome::Ok(WorkspaceSessions { repos: sessions_repos(&repos), ..WorkspaceSessions::default() })),
274 },
275 None => None,
276 };
277
278 // Harness and hosted sessions.
279 let from_table = !matches!(a.kind, Some(SessionKind::Agent | SessionKind::Recorded));
280 let mut sessions: Vec<Session> = if from_table && !ids.is_empty() {
281 self.db
282 .prepare(workspace_sessions_sql(a.shown))
283 .bind(&[
284 serde_json::to_string(&ids)?.into(),
285 rfc3339(now.saturating_sub(SESSION_TIMEOUT_MS)).into(),
286 u32::from(false).into(),
287 a.harness.map_or(JsValue::NULL, |harness| harness.as_str().into()),
288 person.map_or(JsValue::NULL, JsValue::from),
289 item_id.as_deref().map_or(JsValue::NULL, JsValue::from),
290 limit.into(),
291 a.kind.map_or(JsValue::NULL, |kind| kind.as_str().into()),
292 ])?
293 .all()
294 .await?
295 .results::<SessionRow>()?
296 .into_iter()
297 .map(|row| row.at(now))
298 .collect()
299 } else {
300 Vec::new()
301 };
302 self.fill_subagents(&mut sessions).await?;
303 self.fill_first_prompts(&mut sessions).await?;
304 shown::sessions(&mut sessions, &a.viewer, &workspace);
305 let mut trees: HashMap<String, Vec<Session>> = HashMap::new();
306 if a.subagents {
307 for session in sessions.iter().filter(|session| session.subagents > 0).take(TREES) {
308 let mut children = self.subagents_of(&session.id, now).await?;
309 count_subagents(&mut children);
310 shown::sessions(&mut children, &a.viewer, &workspace);
311 trees.insert(session.id.clone(), children);
312 }
313 }
314 let table: Vec<WorkspaceSession> = sessions
315 .into_iter()
316 .filter_map(|session| {
317 let path = path_of.get(session.repo_id.as_str())?.clone();
318 let mut listed = from_session(session, &path, now);
319 listed.children = trees.remove(&listed.id).unwrap_or_default();
320 Some(listed)
321 })
322 .collect();
323
324 // g1t's agent: the workspace's business, as its fleet is.
325 let wants = |kind: SessionKind| a.kind.is_none_or(|wanted| wanted == kind) && a.harness.is_none();
326 let runs: Vec<WorkspaceSession> = if member && wants(SessionKind::Agent) {
327 self.agent_runs_in(&workspace, &ids, a.shown == SessionsShown::Live, member, limit)
328 .await?
329 .into_iter()
330 .map(|(repo_id, run)| from_run(run, &repo_id))
331 .collect()
332 } else {
333 Vec::new()
334 };
335 let recorded: Vec<(WorkspaceSession, bool)> = if wants(SessionKind::Recorded) {
336 self.recorded_in(&ids, limit)
337 .await?
338 .into_iter()
339 .filter_map(|row| {
340 let path = path_of.get(row.repo_id.as_str())?.clone();
341 let has_runs = row.has_runs != 0;
342 Some((from_recorded(row, &path, now), has_runs))
343 })
344 .collect()
345 } else {
346 Vec::new()
347 };
348 let mut runs = runs;
349 self.fill_pulls(&mut runs).await?;
350 let wanted = Wanted { shown: a.shown, kind: a.kind, harness: a.harness, person, item: a.item };
351 let mut listed = merged(table, runs, recorded, &wanted, limit as usize);
352 self.fill_pulls(&mut listed).await?;
353 if member {
354 self.fill_item_titles(&workspace, &mut listed).await?;
355 }
356
357 let (queue, landed) = if a.sessions_only { (Vec::new(), Vec::new()) } else { (self.queues_in(&repos, &a.viewer).await?, self.landed_in(&repos).await?) };
358 Ok(Outcome::Ok(WorkspaceSessions { sessions: listed, queue, landed, repos: sessions_repos(&repos) }))
359 }
360
361 /// Each session's pull request's title and status, and, for one with
362 /// no work item of its own, the pull request's.
363 async fn fill_pulls(&self, sessions: &mut [WorkspaceSession]) -> Result<()> {
364 let pairs: Vec<serde_json::Value> = sessions
365 .iter()
366 .filter(|session| session.pull_title.is_none() || session.item.is_none())
367 .filter_map(|session| session.pull.map(|number| serde_json::json!({ "r": session.repo_id, "n": number })))
368 .collect();
369 if pairs.is_empty() {
370 return Ok(());
371 }
372 let facts = self.db.prepare(PULL_FACTS).bind(&[serde_json::to_string(&pairs)?.into()])?.all().await?.results::<PullFact>()?;
373 for session in sessions.iter_mut() {
374 let Some(fact) = session.pull.and_then(|number| facts.iter().find(|fact| fact.repo_id == session.repo_id && fact.number == number)) else {
375 continue;
376 };
377 session.pull_title.get_or_insert_with(|| fact.title.clone());
378 session.pull_status.get_or_insert(fact.status);
379 if session.item.is_none() {
380 session.item = fact.item;
381 }
382 if session.title.is_none() && session.kind == SessionKind::Agent {
383 session.title = Some(fact.title.clone());
384 }
385 }
386 Ok(())
387 }
388
389 /// Each session's work item's title, for a member of the workspace.
390 async fn fill_item_titles(&self, workspace: &str, sessions: &mut [WorkspaceSession]) -> Result<()> {
391 let mut numbers: Vec<u32> = sessions.iter().filter_map(|session| session.item).collect();
392 numbers.sort_unstable();
393 numbers.dedup();
394 if numbers.is_empty() {
395 return Ok(());
396 }
397 let titles: HashMap<u32, String> = self
398 .db
399 .prepare("SELECT number, title FROM work_items WHERE workspace = ?1 AND number IN (SELECT value FROM json_each(?2))")
400 .bind(&[workspace.into(), serde_json::to_string(&numbers)?.into()])?
401 .all()
402 .await?
403 .results::<Titled>()?
404 .into_iter()
405 .map(|row| (row.number, row.title))
406 .collect();
407 for session in sessions {
408 session.item_title = session.item.and_then(|number| titles.get(&number).cloned());
409 }
410 Ok(())
411 }
412
413 /// The merge queues of `repos` with something waiting, busiest first.
414 async fn queues_in(&self, repos: &[Repo], viewer: &Viewer) -> Result<Vec<RepoQueue>> {
415 if repos.is_empty() {
416 return Ok(Vec::new());
417 }
418 let ids: Vec<&str> = repos.iter().map(|repo| repo.id.as_str()).collect();
419 let busy = self
420 .db
421 .prepare(QUEUED_REPOS)
422 .bind(&[serde_json::to_string(&ids)?.into(), (QUEUES as u32).into()])?
423 .all()
424 .await?
425 .results::<Value>()?;
426 let mut queues = Vec::new();
427 for id in busy {
428 let Some(repo) = repos.iter().find(|repo| repo.id == id.value) else { continue };
429 let path = RepoPath { namespace: repo.namespace.clone(), name: repo.name.clone() };
430 if let Outcome::Ok(queue) = self.queue(QueueArgs { repo: path, viewer: viewer.clone() }).await? {
431 queues.push(RepoQueue { repo: format!("{}/{}", repo.namespace, repo.name), default_branch: repo.default_branch.clone(), queue });
432 }
433 }
434 Ok(queues)
435 }
436
437 /// The pull requests of `repos` merged most recently.
438 async fn landed_in(&self, repos: &[Repo]) -> Result<Vec<RepoPull>> {
439 if repos.is_empty() {
440 return Ok(Vec::new());
441 }
442 let ids: Vec<&str> = repos.iter().map(|repo| repo.id.as_str()).collect();
443 let pulls = self
444 .db
445 .prepare(format!(
446 "SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id IN (SELECT value FROM json_each(?1)) AND status = 'merged' AND merged_at IS NOT NULL
447 ORDER BY merged_at DESC LIMIT ?2"
448 ))
449 .bind(&[serde_json::to_string(&ids)?.into(), LANDED.into()])?
450 .all()
451 .await?
452 .results::<PullRow>()?;
453 Ok(pulls
454 .into_iter()
455 .map(Pull::from)
456 .filter_map(|mut pull| {
457 let repo = repos.iter().find(|repo| repo.id == pull.repo_id)?;
458 crate::fill_base(&mut pull, repo);
459 Some(RepoPull { repo: format!("{}/{}", repo.namespace, repo.name), pull })
460 })
461 .collect())
462 }
463
464 /// One harness or hosted session of a workspace by its id, read as its
465 /// repository's own list reads it.
466 pub(crate) async fn get_workspace_session(&self, a: GetWorkspaceSessionArgs) -> Result<Outcome<WorkspaceSessionDetail>> {
467 let workspace = a.workspace.trim().to_lowercase();
468 let placed = self
469 .db
470 .prepare(
471 "SELECT s.repo_id, rw.path FROM sessions s JOIN repo_workspaces rw ON rw.repo_id = s.repo_id
472 WHERE s.id = ?1 AND rw.workspace = ?2",
473 )
474 .bind(&[a.id.trim().into(), workspace.as_str().into()])?
475 .first::<Placed>(None)
476 .await?;
477 let Some(placed) = placed else {
478 return Ok(Outcome::fail(FailureCode::NotFound, "Session not found."));
479 };
480 let Some((namespace, name)) = placed.path.split_once('/') else {
481 return Ok(Outcome::fail(FailureCode::NotFound, "Session not found."));
482 };
483 let repo = RepoPath { namespace: namespace.to_owned(), name: name.to_owned() };
484 let args = WorkSessionArgs { viewer: a.viewer, repo, id: a.id.trim().to_owned(), after_seq: a.after_seq };
485 Ok(match self.get_work_session(args).await? {
486 Outcome::Ok(detail) => Outcome::Ok(WorkspaceSessionDetail { repo: placed.path, detail }),
487 Outcome::Fail(failure) => Outcome::Fail(failure),
488 })
489 }
490}
491
492fn sessions_repos(repos: &[Repo]) -> Vec<SessionsRepo> {
493 repos
494 .iter()
495 .map(|repo| SessionsRepo { id: repo.id.clone(), path: format!("{}/{}", repo.namespace, repo.name), default_branch: repo.default_branch.clone() })
496 .collect()
497}
498
499#[cfg(test)]
500mod tests {
501 use super::*;
502 use crate::runs::RECORDED_IN;
503 use rusqlite::{Connection, params};
504
505 const NOW: u64 = 1_791_800_000_000;
506 const MINUTE: u64 = 60_000;
507
508 fn at(ago_minutes: u64) -> String {
509 rfc3339(NOW - ago_minutes * MINUTE)
510 }
511
512 /// Two repositories of acme and one of another workspace, a work item,
513 /// and pull requests in each.
514 fn database() -> Connection {
515 let db = crate::migrated::database();
516 db.execute_batch(
517 "INSERT INTO repo_workspaces (repo_id, workspace, path, private) VALUES
518 ('rep_web', 'acme', 'acme/web', 1), ('rep_api', 'acme', 'acme/api', 1), ('rep_x', 'other', 'other/x', 0);
519 INSERT INTO work_items (id, workspace, number, title, body, author_id, author_name, created_at, updated_at)
520 VALUES ('wki_4', 'acme', 4, 'Greet people', '', 'usr_1', 'ana', '', '');
521 INSERT INTO pulls (id, repo_id, number, title, agent, runtime, author_id, author_name, created_at, updated_at, item_id) VALUES
522 ('pr_web_9', 'rep_web', 9, 'Say hello', 'claude-code', 'external', 'usr_1', 'ana', '', '', 'wki_4'),
523 ('pr_api_7', 'rep_api', 7, 'Recorded only', 'cursor', 'external', 'usr_2', 'bo', '', '', NULL),
524 ('pr_api_8', 'rep_api', 8, 'By g1t', 'g1t', 'external', 'usr_2', 'bo', '', '', NULL),
525 ('pr_x_1', 'rep_x', 1, 'Elsewhere', 'codex', 'external', 'usr_3', 'cy', '', '', NULL);",
526 )
527 .unwrap();
528 db
529 }
530
531 #[allow(clippy::too_many_arguments)]
532 fn session(db: &Connection, id: &str, repo: &str, item: Option<&str>, pull: Option<u32>, actor: &str, harness: &str, state: &str, last: u64, parent: Option<&str>) {
533 db.execute(
534 "INSERT INTO sessions (id, repo_id, item_id, pull, branch, actor_id, actor_name, harness, harness_session_id,
535 title, state, head_commit, started_at, last_seen_at, parent_id)
536 VALUES (?1, ?2, ?3, ?4, 'main', ?5, ?5, ?6, ?1, NULL, ?7, NULL, ?8, ?8, ?9)",
537 params![id, repo, item, pull, actor, harness, state, at(last), parent],
538 )
539 .unwrap();
540 }
541
542 #[allow(clippy::too_many_arguments)]
543 fn listed(db: &Connection, shown: SessionsShown, repos: &[&str], harness: Option<&str>, person: Option<&str>, item: Option<&str>, kind: Option<&str>) -> Vec<String> {
544 let sql = workspace_sessions_sql(shown);
545 let mut statement = db.prepare(&sql).unwrap();
546 statement
547 .query_map(
548 params![serde_json::to_string(repos).unwrap(), rfc3339(NOW - SESSION_TIMEOUT_MS), 0, harness, person, item, 100, kind],
549 |row| row.get::<_, String>("id"),
550 )
551 .unwrap()
552 .map(Result::unwrap)
553 .collect()
554 }
555
556 fn seeded() -> Connection {
557 let db = database();
558 session(&db, "ses_cc", "rep_web", Some("wki_4"), Some(9), "ana", "claude_code", "active", 1, None);
559 session(&db, "ses_sub", "rep_web", Some("wki_4"), None, "ana", "claude_code", "active", 1, Some("ses_cc"));
560 session(&db, "ses_codex", "rep_api", None, None, "bo", "codex", "active", 20, None);
561 session(&db, "ses_cursor", "rep_api", None, None, "bo", "cursor", "active", 45, None);
562 session(&db, "ses_hosted", "rep_api", None, None, "ana", "claude_managed_agents", "active", 3, None);
563 session(&db, "ses_done", "rep_web", None, None, "cy", "other", "ended", 90, None);
564 session(&db, "ses_x", "rep_x", None, None, "cy", "codex", "active", 1, None);
565 db
566 }
567
568 #[test]
569 fn only_the_repositories_looked_in_are_listed_and_subagents_stay_under_their_parent() {
570 let db = seeded();
571 let all = listed(&db, SessionsShown::Recent, &["rep_web", "rep_api"], None, None, None, None);
572 assert_eq!(all, ["ses_cc", "ses_hosted", "ses_codex", "ses_cursor", "ses_done"]);
573 // A repository the viewer may not read is never looked in.
574 let web = listed(&db, SessionsShown::Recent, &["rep_web"], None, None, None, None);
575 assert_eq!(web, ["ses_cc", "ses_done"]);
576 assert!(!all.contains(&"ses_x".to_owned()));
577 }
578
579 #[test]
580 fn live_is_what_is_going_and_ended_takes_those_gone_quiet_too() {
581 let db = seeded();
582 let repos = ["rep_web", "rep_api"];
583 // Idle (20 minutes) is still live; quiet past the timeout (45) is not.
584 assert_eq!(listed(&db, SessionsShown::Live, &repos, None, None, None, None), ["ses_cc", "ses_hosted", "ses_codex"]);
585 assert_eq!(listed(&db, SessionsShown::Ended, &repos, None, None, None, None), ["ses_cursor", "ses_done"]);
586 }
587
588 #[test]
589 fn sessions_filter_by_harness_person_kind_and_work_item() {
590 let db = seeded();
591 let repos = ["rep_web", "rep_api"];
592 assert_eq!(listed(&db, SessionsShown::Recent, &repos, Some("codex"), None, None, None), ["ses_codex"]);
593 assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, Some("BO"), None, None), ["ses_codex", "ses_cursor"]);
594 assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, None, Some("hosted")), ["ses_hosted"]);
595 assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, None, Some("harness")), ["ses_cc", "ses_codex", "ses_cursor", "ses_done"]);
596 assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, Some("wki_4"), None), ["ses_cc"]);
597 // Tied to the work item another way: a link.
598 db.execute(super::super::links::INSERT_LINK, params!["ses_codex", "rep_api", "item", 0, "ref", at(1), "wki_4"]).unwrap();
599 assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, Some("wki_4"), None), ["ses_cc", "ses_codex"]);
600 }
601
602 fn recorded(db: &Connection, repos: &[&str]) -> Vec<RecordedRow> {
603 let mut statement = db.prepare(RECORDED_IN).unwrap();
604 let rows = statement
605 .query_map(params![serde_json::to_string(repos).unwrap(), 50], |row| {
606 Ok(RecordedRow {
607 pull_id: row.get("pull_id")?,
608 repo_id: row.get("repo_id")?,
609 number: row.get("number")?,
610 title: row.get("title")?,
611 status: PullStatus::Open,
612 agent: row.get("agent")?,
613 author_name: row.get("author_name")?,
614 item: row.get("item")?,
615 started_at: row.get("started_at")?,
616 last_at: row.get("last_at")?,
617 has_runs: row.get("has_runs")?,
618 })
619 })
620 .unwrap();
621 rows.map(Result::unwrap).collect()
622 }
623
624 fn run(number: u32, status: &str) -> AgentRun {
625 serde_json::from_value(serde_json::json!({
626 "id": "run_1", "repo": { "namespace": "acme", "name": "api" }, "number": number, "title": null, "kind": "implement",
627 "agent": "g1t", "model": null, "status": status, "step": "Running tests", "stepCount": 3, "startedBy": null,
628 "error": null, "costUsd": null, "turns": null, "createdAt": at(10), "startedAt": at(10), "finishedAt": null, "updatedAt": at(1)
629 }))
630 .unwrap()
631 }
632
633 #[test]
634 fn a_pull_request_recorded_directly_is_a_session_and_one_moved_from_a_harness_is_not() {
635 let db = seeded();
636 db.execute_batch(&format!(
637 "INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_api_7', 1, 'prompt', 'Fix it', '{a}'), ('pr_api_7', 2, 'message', 'Done', '{b}');
638 INSERT INTO session_entries (pull_id, seq, kind, text, at, session_id, session_seq) VALUES ('pr_web_9', 1, 'message', 'Moved', '{b}', 'ses_cc', 1);
639 INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_x_1', 1, 'message', 'Elsewhere', '{b}');
640 INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_api_8', 1, 'message', 'g1t', '{c}');
641 INSERT INTO agent_runs (id, workspace, repo_id, repo, number, pull_id, kind, agent, status, sandbox, token_hash, created_at, updated_at)
642 VALUES ('run_1', 'acme', 'rep_api', 'acme/api', 8, 'pr_api_8', 'implement', 'g1t', 'running', 'sb', 'h', '{a}', '{b}');",
643 a = at(10),
644 b = at(2),
645 c = at(5)
646 ))
647 .unwrap();
648 let rows = recorded(&db, &["rep_web", "rep_api"]);
649 let numbers: Vec<u32> = rows.iter().map(|row| row.number).collect();
650 assert_eq!(numbers, [7, 8]);
651 assert_eq!(rows[0].started_at, at(10));
652 assert_eq!(rows[0].has_runs, 0);
653 assert_eq!(rows[1].has_runs, 1);
654 // Together: the run of g1t's agent stands for its pull request, and
655 // the recorded one with nothing behind it is a session of its own.
656 let recorded: Vec<(WorkspaceSession, bool)> = rows
657 .into_iter()
658 .map(|row| {
659 let runs = row.has_runs != 0;
660 (from_recorded(row, "acme/api", NOW), runs)
661 })
662 .collect();
663 let wanted = Wanted { shown: SessionsShown::Recent, kind: None, harness: None, person: None, item: None };
664 let all = merged(Vec::new(), vec![from_run(run(8, "running"), "rep_api")], recorded, &wanted, 10);
665 let kinds: Vec<(SessionKind, &str)> = all.iter().map(|session| (session.kind, session.id.as_str())).collect();
666 assert_eq!(kinds, [(SessionKind::Agent, "run_1"), (SessionKind::Recorded, "pr_api_7")]);
667 assert_eq!(all[0].actor.as_deref(), Some("g1t"));
668 assert_eq!(all[0].state, SessionState::Active);
669 assert_eq!(all[1].state, SessionState::Active);
670 let live_by_bo = Wanted { shown: SessionsShown::Live, kind: Some(SessionKind::Recorded), harness: None, person: Some("bo"), item: None };
671 assert_eq!(merged(Vec::new(), Vec::new(), vec![(all[1].clone(), false)], &live_by_bo, 10).len(), 1);
672 let by_harness = Wanted { shown: SessionsShown::Recent, kind: None, harness: Some(Harness::Codex), person: None, item: None };
673 assert!(merged(Vec::new(), Vec::new(), vec![(all[1].clone(), false)], &by_harness, 10).is_empty());
674 // A finished run is not live.
675 let live = Wanted { shown: SessionsShown::Live, kind: None, harness: None, person: None, item: None };
676 assert!(merged(Vec::new(), vec![from_run(run(8, "succeeded"), "rep_api")], Vec::new(), &live, 10).is_empty());
677 }
678
679 #[test]
680 fn every_kind_reads_as_one_shape_and_a_quiet_session_reads_as_ended() {
681 let session: Session = serde_json::from_value(serde_json::json!({
682 "id": "ses_q", "repoId": "rep_api", "item": null, "pull": 7, "branch": "fix", "actor": { "id": "usr_2", "username": "bo" },
683 "harness": "claude_managed_agents", "harnessSessionId": null, "title": "Fix it", "state": "idle", "headCommit": null,
684 "startedAt": at(60), "lastSeenAt": at(40), "endedAt": null, "endReason": null, "subagents": 2
685 }))
686 .unwrap();
687 let listed = from_session(session, "acme/api", NOW);
688 assert_eq!(listed.kind, SessionKind::Hosted);
689 assert_eq!(listed.state, SessionState::Ended);
690 assert_eq!(listed.session.as_ref().unwrap().end_reason.as_deref(), Some("timed_out"));
691 assert_eq!(listed.subagents, 2);
692 assert_eq!(listed.repo, "acme/api");
693 // A recorded one on the pull request this session is on is this session.
694 let recorded = RecordedRow {
695 pull_id: "pr_api_7".into(),
696 repo_id: "rep_api".into(),
697 number: 7,
698 title: "Fix it".into(),
699 status: PullStatus::Open,
700 agent: "cursor".into(),
701 author_name: "bo".into(),
702 item: None,
703 started_at: at(60),
704 last_at: at(50),
705 has_runs: 0,
706 };
707 let wanted = Wanted { shown: SessionsShown::Recent, kind: None, harness: None, person: None, item: None };
708 let all = merged(vec![listed], Vec::new(), vec![(from_recorded(recorded, "acme/api", NOW), false)], &wanted, 10);
709 assert_eq!(all.len(), 1);
710 assert_eq!(all[0].id, "ses_q");
711 }
712
713 #[test]
714 fn pull_facts_and_busy_queues_are_read_across_repositories() {
715 let db = seeded();
716 let pairs = serde_json::json!([{ "r": "rep_web", "n": 9 }, { "r": "rep_api", "n": 7 }, { "r": "rep_api", "n": 99 }]).to_string();
717 let mut statement = db.prepare(PULL_FACTS).unwrap();
718 let mut facts: Vec<(String, u32, String, Option<u32>)> = statement
719 .query_map(params![pairs], |row| Ok((row.get("repo_id")?, row.get("number")?, row.get("title")?, row.get("item")?)))
720 .unwrap()
721 .map(Result::unwrap)
722 .collect();
723 facts.sort();
724 assert_eq!(facts, [("rep_api".into(), 7, "Recorded only".into(), None), ("rep_web".into(), 9, "Say hello".into(), Some(4))]);
725 db.execute_batch(
726 "INSERT INTO queue_entries (id, repo_id, pull_id, number, state, enqueued_by, created_at) VALUES
727 ('q1', 'rep_api', 'pr_api_7', 7, 'waiting', '{}', '1'), ('q2', 'rep_api', 'pr_api_8', 8, 'testing', '{}', '2'),
728 ('q3', 'rep_web', 'pr_web_9', 9, 'landed', '{}', '3'), ('q4', 'rep_x', 'pr_x_1', 1, 'waiting', '{}', '4');",
729 )
730 .unwrap();
731 let mut statement = db.prepare(QUEUED_REPOS).unwrap();
732 let busy: Vec<String> =
733 statement.query_map(params![r#"["rep_web","rep_api"]"#, 12], |row| row.get::<_, String>("value")).unwrap().map(Result::unwrap).collect();
734 assert_eq!(busy, ["rep_api"]);
735 }
736}