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
716 linesCodeBlameRaw
1//! Runs: every sandbox g1t starts for checks, the merge queue and finding
2//! out whether a pull request merges cleanly, as a record people can watch
3//! and look back on. Runs of g1t's former agents are kept and listed too.
4//!
5//! The runner service opens a run as it starts a sandbox (`open_run`) and
6//! the sandbox reports its steps through the API with the run's one-time
7//! token (`report_run`), as checks do. The runner closes it when the
8//! sandbox stops.
9//!
10//! Sessions are the read side of what coding agents record on a pull
11//! request (`append_session`): `list_sessions` and `get_session`.
12
13use g1t_contracts::agents::*;
14use g1t_contracts::repos::{Repo, RepoPath};
15use g1t_contracts::time::rfc3339;
16use g1t_contracts::work::{PullStatus, SessionEntry, SessionEntryKind};
17use g1t_scan::redact::soften;
18use g1t_contracts::{FailureCode, Membership, Outcome, User, Viewer, new_id};
19use g1t_kit::now_ms;
20use serde::Deserialize;
21use worker::Result;
22use worker::wasm_bindgen::JsValue;
23
24use crate::checks::{hash, new_token};
25use crate::coordination::shown::{self, Sight};
26use crate::rows::SessionRow;
27use crate::{Work, optional};
28
29/// How the runner starts a step that notes something about the sandbox
30/// rather than what the run is doing: a host it was refused (the runner's
31/// egress.ts `blockedStep`).
32/// A workspace renamed: its runs move to the slug it has now. `?1` is the
33/// current slug, `?2` a stale one (see `g1t_kit::rename`).
34pub(crate) const RENAMED: &[&str] = &[
35 "UPDATE agent_runs SET workspace = ?1 WHERE workspace = ?2",
36 "UPDATE agent_runs SET repo = ?1 || substr(repo, length(?2) + 1) WHERE substr(repo, 1, length(?2) + 1) = ?2 || '/'",
37];
38
39/// A repository transferred (see `g1t_kit::transfer`): its runs go with
40/// it, as its pull requests do by id. `?1`/`?2` the path now
41/// and before, `?3`/`?4` the workspaces, `?5` the repository's id.
42pub(crate) const TRANSFERRED: &[&str] = &[
43 "UPDATE agent_runs SET workspace = ?3, repo = ?1 WHERE repo_id = ?5 AND repo = ?2",
44];
45
46const NOTE_PREFIXES: [&str; 1] = ["Blocked: "];
47
48/// Where a report leaves the run: its newest step that says what the run
49/// is doing. A note (a refused host) stays in the steps but never stands
50/// for the run, which carries on past it. `None` when there is only notes.
51fn current_step(steps: &[String]) -> Option<String> {
52 steps.iter().rev().find(|step| !NOTE_PREFIXES.iter().any(|prefix| step.starts_with(prefix))).cloned()
53}
54
55/// The most steps a run keeps; older ones fall off the start.
56const MAX_STEPS: u32 = 200;
57/// The most steps taken from one report.
58const MAX_STEPS_PER_REPORT: usize = 20;
59/// The longest a step is kept.
60const MAX_STEP_CHARS: usize = 240;
61/// A run that has said nothing for this long is taken to have died.
62const SILENT_HOURS: u64 = 3;
63const DEFAULT_LIST: u32 = 50;
64const MAX_LIST: u32 = 200;
65/// The most session entries one view returns.
66const SESSION_ENTRIES: u32 = 2000;
67
68/// Every column but the steps, which only a run's own page needs.
69const COLUMNS: &str = "id, workspace, repo_id, repo, number, pull_id, kind, agent, model, status,
70 step, step_count, cost_usd, turns, sandbox, token_hash, started_by, error, created_at,
71 started_at, finished_at, updated_at, budget_usd, time_cap_minutes, halted,
72 COALESCE((SELECT title FROM pulls WHERE pulls.id = agent_runs.pull_id), agent_runs.title) AS title";
73
74/// The pull requests of the repositories in `?1` (a JSON array of ids)
75/// with entries recorded on them directly (not by a session moved onto
76/// it), most recently recorded first, at most `?2`.
77pub(crate) const RECORDED_IN: &str = "SELECT p.id AS pull_id, p.repo_id, p.number, p.title, p.status, p.agent, p.author_name,
78 (SELECT number FROM work_items WHERE work_items.id = p.item_id) AS item,
79 min(e.at) AS started_at, max(e.at) AS last_at,
80 EXISTS (SELECT 1 FROM agent_runs r WHERE r.pull_id = p.id) AS has_runs
81 FROM pulls p JOIN session_entries e ON e.pull_id = p.id
82 WHERE p.repo_id IN (SELECT value FROM json_each(?1)) AND e.session_id IS NULL
83 GROUP BY p.id ORDER BY last_at DESC LIMIT ?2";
84
85/// A pull request's recorded session, for a workspace's sessions.
86#[derive(Clone, Debug, Deserialize)]
87pub(crate) struct RecordedRow {
88 pub(crate) pull_id: String,
89 pub(crate) repo_id: String,
90 pub(crate) number: u32,
91 pub(crate) title: String,
92 pub(crate) status: g1t_contracts::work::PullStatus,
93 pub(crate) agent: String,
94 pub(crate) author_name: String,
95 #[serde(default)]
96 pub(crate) item: Option<u32>,
97 pub(crate) started_at: String,
98 pub(crate) last_at: String,
99 pub(crate) has_runs: u32,
100}
101
102#[derive(Deserialize)]
103pub(crate) struct RunRow {
104 id: String,
105 #[allow(dead_code)]
106 workspace: String,
107 #[allow(dead_code)]
108 repo_id: String,
109 repo: String,
110 number: Option<u32>,
111 pub(crate) pull_id: Option<String>,
112 kind: String,
113 agent: String,
114 model: Option<String>,
115 status: String,
116 step: Option<String>,
117 #[serde(default)]
118 steps: Option<String>,
119 step_count: u32,
120 cost_usd: Option<f64>,
121 turns: Option<u32>,
122 token_hash: String,
123 started_by: Option<String>,
124 error: Option<String>,
125 created_at: String,
126 started_at: Option<String>,
127 finished_at: Option<String>,
128 updated_at: String,
129 title: Option<String>,
130 #[serde(default)]
131 budget_usd: Option<f64>,
132 #[serde(default)]
133 time_cap_minutes: Option<u32>,
134 #[serde(default)]
135 halted: Option<String>,
136}
137
138impl RunRow {
139 fn status(&self) -> RunStatus {
140 RunStatus::parse(&self.status).unwrap_or(RunStatus::Failed)
141 }
142
143 /// The run as `member` (or not) may see it: model and cost are the
144 /// workspace's business.
145 fn into_run(self, member: bool) -> AgentRun {
146 let (namespace, name) = self.repo.split_once('/').unwrap_or((&self.repo, ""));
147 AgentRun {
148 repo: RepoPath {
149 namespace: namespace.to_owned(),
150 name: name.to_owned(),
151 },
152 status: self.status(),
153 kind: RunKind::parse(&self.kind).unwrap_or(RunKind::Implement),
154 steps: self
155 .steps
156 .as_deref()
157 .and_then(|steps| serde_json::from_str(steps).ok())
158 .unwrap_or_default(),
159 id: self.id,
160 number: self.number.filter(|number| *number > 0),
161 title: self.title,
162 agent: self.agent,
163 model: self.model.filter(|_| member),
164 step: self.step,
165 step_count: self.step_count,
166 started_by: self.started_by,
167 error: self.error,
168 cost_usd: self.cost_usd.filter(|_| member),
169 turns: self.turns,
170 budget_usd: self.budget_usd.filter(|_| member),
171 time_cap_minutes: self.time_cap_minutes,
172 halted: self.halted,
173 created_at: self.created_at,
174 started_at: self.started_at,
175 finished_at: self.finished_at,
176 updated_at: self.updated_at,
177 }
178 }
179}
180
181/// One line, short enough to read at a glance.
182pub(crate) fn one_line(text: &str, limit: usize) -> String {
183 let line = text.split_whitespace().collect::<Vec<_>>().join(" ");
184 if line.chars().count() <= limit {
185 return line;
186 }
187 let mut short: String = line.chars().take(limit.saturating_sub(1)).collect();
188 short.push('…');
189 short
190}
191
192/// A principal that can read any repository of `namespace`, for lookups a
193/// trusted service asks for. What it is given reads and nothing more.
194pub(crate) fn member_of(actor: &User, namespace: &str) -> Viewer {
195 let mut viewer = actor.clone();
196 let slug = namespace.to_lowercase();
197 if !viewer.is_member(&slug) {
198 viewer.workspaces.push(Membership {
199 base_permission: Some(g1t_contracts::access::BasePermission::Read),
200 ..Membership::member(slug)
201 });
202 }
203 Some(viewer)
204}
205
206/// Whether the viewer belongs to the workspace: what a run spent, its model
207/// and its budget are the workspace's business, not every reader's (an
208/// outside collaborator sees the work, not the bill).
209fn is_member(viewer: &Viewer, namespace: &str) -> bool {
210 viewer
211 .as_ref()
212 .is_some_and(|viewer| viewer.is_member(&namespace.to_lowercase()))
213}
214
215#[derive(Deserialize)]
216struct SessionListRow {
217 pull_id: String,
218 number: u32,
219 title: String,
220 status: PullStatus,
221 agent: String,
222 entries: u32,
223 tools: u32,
224 started_at: String,
225 last_at: String,
226 prompt: Option<String>,
227}
228
229#[derive(Deserialize)]
230struct RunSumRow {
231 pull_id: String,
232 kind: String,
233 status: String,
234 cost_usd: Option<f64>,
235}
236
237impl Work {
238 /// The repository at `path`, if `viewer` may see it, and whether they
239 /// are a member of its workspace (which shows what runs cost).
240 async fn visible_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<(Repo, bool)>> {
241 Ok(match self.repo(path, viewer).await? {
242 Outcome::Ok(repo) => {
243 let member = is_member(viewer, &repo.namespace);
244 Outcome::Ok((repo, member))
245 }
246 Outcome::Fail(failure) => Outcome::Fail(failure),
247 })
248 }
249
250 async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {
251 self.db
252 .prepare(format!("SELECT {COLUMNS}, steps FROM agent_runs WHERE id = ?"))
253 .bind(&[id.into()])?
254 .first::<RunRow>(None)
255 .await
256 }
257
258 pub(crate) async fn open_run(&self, a: OpenRunArgs) -> Result<Outcome<AgentRunTicket>> {
259 // Only sandboxes that are not agents are started now; earlier
260 // kinds stay readable on the runs already recorded.
261 if !matches!(a.kind, RunKind::Checks | RunKind::Queue | RunKind::Mergecheck) {
262 return Ok(Outcome::fail(
263 FailureCode::Invalid,
264 format!("A {} run is no longer started.", a.kind.as_str()),
265 ));
266 }
267 // The runner is trusted: it names the repository it is starting a
268 // sandbox in, whoever the sandbox acts as. The run records where
269 // the repository is now, looked up by its id, not the path it was
270 // named by, which may be from before a transfer or a rename.
271 let (repo_id, path) = match &a.pull_id {
272 Some(pull_id) => match self.pull_by_id(pull_id).await? {
273 Some(pull) => {
274 let now: Option<RepoPath> = g1t_kit::call(
275 &self.repos,
276 "path_by_id",
277 &g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
278 )
279 .await?;
280 (pull.repo_id, now.unwrap_or_else(|| a.repo.clone()))
281 }
282 None => return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found.")),
283 },
284 None => match self.repo(&a.repo, &member_of(&a.actor, &a.repo.namespace)).await? {
285 Outcome::Ok(repo) => (repo.id, RepoPath { namespace: repo.namespace, name: repo.name }),
286 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
287 },
288 };
289 let namespace = path.namespace.to_lowercase();
290 // Nothing new starts on an archived or deleted repository.
291 if !self.repo_active(&repo_id).await? {
292 return Ok(Outcome::fail(
293 FailureCode::Forbidden,
294 format!("{}/{} is archived or deleted, so nothing new starts on it.", path.namespace, path.name),
295 ));
296 }
297 let now = now_ms();
298 let id = new_id("arn", now);
299 let token = new_token();
300 let timestamp = rfc3339(now);
301 let agent = a
302 .agent
303 .clone()
304 .unwrap_or_else(|| g1t_contracts::identity::AGENT_NAME.to_owned());
305 self.db
306 .prepare(
307 "INSERT INTO agent_runs
308 (id, workspace, repo_id, repo, number, pull_id, title, kind, agent, model, status,
309 sandbox, token_hash, started_by, created_at, updated_at, budget_usd, time_cap_minutes)
310 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?)",
311 )
312 .bind(&[
313 id.as_str().into(),
314 namespace.into(),
315 repo_id.into(),
316 format!("{}/{}", path.namespace, path.name).into(),
317 a.number.filter(|n| *n > 0).map_or(JsValue::NULL, JsValue::from),
318 optional(&a.pull_id),
319 optional(&a.title.map(|title| one_line(&title, 200))),
320 a.kind.as_str().into(),
321 agent.into(),
322 optional(&a.model),
323 a.sandbox.into(),
324 hash(&token).into(),
325 optional(&a.started_by),
326 timestamp.as_str().into(),
327 timestamp.as_str().into(),
328 a.budget_usd.filter(|usd| usd.is_finite() && *usd > 0.0).map_or(JsValue::NULL, JsValue::from),
329 a.time_cap_minutes.map_or(JsValue::NULL, JsValue::from),
330 ])?
331 .run()
332 .await?;
333 Ok(Outcome::Ok(AgentRunTicket { run_id: id, token }))
334 }
335
336 /// Adds `text` to a run's steps, keeping the latest `MAX_STEPS`.
337 pub(crate) fn add_step(&self, run_id: &str, at: &str, text: &str) -> Result<worker::D1PreparedStatement> {
338 self.db
339 .prepare(
340 "UPDATE agent_runs SET
341 steps = CASE WHEN json_array_length(steps) >= ?4
342 THEN json_insert(json_remove(steps, '$[0]'), '$[#]', json_object('at', ?1, 'text', ?2))
343 ELSE json_insert(steps, '$[#]', json_object('at', ?1, 'text', ?2)) END,
344 step_count = step_count + 1
345 WHERE id = ?3",
346 )
347 .bind(&[at.into(), text.into(), run_id.into(), MAX_STEPS.into()])
348 }
349
350 pub(crate) async fn report_run(&self, a: ReportRunArgs) -> Result<Outcome<RunStatus>> {
351 let Some(run) = self
352 .run_row(&a.run_id)
353 .await?
354 .filter(|run| run.token_hash == hash(&a.token))
355 else {
356 return Ok(Outcome::fail(FailureCode::NotFound, "Run not found."));
357 };
358 // Finished, or stopped by a person: nothing more is taken, and the
359 // sandbox learns why.
360 if run.finished_at.is_some() {
361 return Ok(Outcome::Ok(run.status()));
362 }
363 // It reached a cap of its guardrails (guardrails.rs).
364 if let Some(halt) = a.halt {
365 let pull_id = run.pull_id.as_deref();
366 return self
367 .halt_run(&run.id, &run.repo_id, pull_id, run.number, &run.kind, halt, a.error, a.cost_usd)
368 .await;
369 }
370 let now = rfc3339(now_ms());
371 let steps: Vec<String> = a
372 .steps
373 .iter()
374 .map(|step| one_line(step, MAX_STEP_CHARS))
375 .filter(|step| !step.is_empty())
376 .collect();
377 // A burst keeps its end: that is where the run is now.
378 let skip = steps.len().saturating_sub(MAX_STEPS_PER_REPORT);
379 let mut statements = Vec::new();
380 for step in &steps[skip..] {
381 statements.push(self.add_step(&run.id, &now, step)?);
382 }
383 let current = a
384 .step
385 .as_deref()
386 .map(|step| one_line(step, MAX_STEP_CHARS))
387 .filter(|step| !step.is_empty())
388 .or_else(|| current_step(&steps));
389 let outcome = a
390 .outcome
391 .filter(|outcome| matches!(outcome, RunStatus::Succeeded | RunStatus::Failed));
392 let cost = a.cost_usd.filter(|cost| cost.is_finite() && *cost >= 0.0);
393 statements.push(
394 self.db
395 .prepare(
396 "UPDATE agent_runs SET
397 status = COALESCE(?1, CASE WHEN status = 'queued' THEN 'running' ELSE status END),
398 started_at = COALESCE(started_at, ?2),
399 step = COALESCE(?3, step),
400 cost_usd = COALESCE(?4, cost_usd),
401 turns = COALESCE(?5, turns),
402 error = COALESCE(?6, error),
403 finished_at = CASE WHEN ?1 IS NULL THEN finished_at ELSE ?2 END,
404 updated_at = ?2
405 WHERE id = ?7 AND finished_at IS NULL",
406 )
407 .bind(&[
408 outcome.map_or(JsValue::NULL, |outcome| outcome.as_str().into()),
409 now.as_str().into(),
410 optional(&current),
411 cost.map_or(JsValue::NULL, JsValue::from),
412 a.turns.map_or(JsValue::NULL, JsValue::from),
413 optional(&a.error.map(|error| one_line(&error, 1000))),
414 run.id.as_str().into(),
415 ])?,
416 );
417 self.db.batch(statements).await?;
418 Ok(Outcome::Ok(outcome.unwrap_or(RunStatus::Running)))
419 }
420
421 /// Runs whose sandbox died without anyone noticing, marked failed.
422 pub(crate) async fn sweep_silent(&self) -> Result<()> {
423 let now = now_ms();
424 let cutoff = rfc3339(now.saturating_sub(SILENT_HOURS * 3_600_000));
425 self.db
426 .prepare(
427 "UPDATE agent_runs
428 SET status = 'failed', error = COALESCE(error, 'It stopped reporting.'),
429 finished_at = ?, updated_at = ?
430 WHERE status IN ('queued', 'running') AND updated_at < ?",
431 )
432 .bind(&[rfc3339(now).into(), rfc3339(now).into(), cutoff.into()])?
433 .run()
434 .await?;
435 Ok(())
436 }
437
438 pub(crate) async fn list_runs(&self, a: ListRunsArgs) -> Result<Outcome<Vec<AgentRun>>> {
439 let (column, key, member) = if let Some(workspace) = &a.workspace {
440 let slug = workspace.to_lowercase();
441 if !is_member(&a.viewer, &slug) {
442 return Ok(Outcome::fail(
443 FailureCode::Forbidden,
444 "Only members of the workspace can see its fleet.",
445 ));
446 }
447 ("workspace", slug, true)
448 } else if let Some(path) = &a.repo {
449 match self.visible_repo(path, &a.viewer).await? {
450 Outcome::Ok((repo, member)) => ("repo_id", repo.id, member),
451 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
452 }
453 } else {
454 return Ok(Outcome::fail(FailureCode::Invalid, "Name a repository or a workspace."));
455 };
456 self.sweep_silent().await?;
457 let limit = a.limit.unwrap_or(DEFAULT_LIST).clamp(1, MAX_LIST);
458 let rows = self
459 .db
460 .prepare(format!(
461 "SELECT {COLUMNS} FROM agent_runs
462 WHERE {column} = ?1
463 AND (?2 = 0 OR status IN ('queued', 'running'))
464 AND (?3 IS NULL OR kind = ?3)
465 AND (?4 IS NULL OR status = ?4)
466 AND (?5 IS NULL OR number = ?5)
467 ORDER BY created_at DESC LIMIT ?6"
468 ))
469 .bind(&[
470 key.into(),
471 u32::from(a.active).into(),
472 a.kind.map_or(JsValue::NULL, |kind| kind.as_str().into()),
473 a.status.map_or(JsValue::NULL, |status| status.as_str().into()),
474 a.number.map_or(JsValue::NULL, JsValue::from),
475 limit.into(),
476 ])?
477 .all()
478 .await?
479 .results::<RunRow>()?;
480 Ok(Outcome::Ok(rows.into_iter().map(|row| row.into_run(member)).collect()))
481 }
482
483 /// The runs of g1t's own agent in some repositories of a workspace
484 /// (sandboxes for checks, the queue and merge checks are not agents),
485 /// newest first, with each one's repository id. Only `live` ones: those
486 /// queued or running.
487 pub(crate) async fn agent_runs_in(&self, workspace: &str, repo_ids: &[String], live: bool, member: bool, limit: u32) -> Result<Vec<(String, AgentRun)>> {
488 if repo_ids.is_empty() {
489 return Ok(Vec::new());
490 }
491 let rows = self
492 .db
493 .prepare(format!(
494 "SELECT {COLUMNS} FROM agent_runs
495 WHERE workspace = ?1 AND repo_id IN (SELECT value FROM json_each(?2))
496 AND kind IN ('implement', 'revise', 'review', 'answer', 'update', 'plan')
497 AND (?3 = 0 OR status IN ('queued', 'running'))
498 ORDER BY updated_at DESC LIMIT ?4"
499 ))
500 .bind(&[workspace.to_lowercase().into(), serde_json::to_string(repo_ids)?.into(), u32::from(live).into(), limit.into()])?
501 .all()
502 .await?
503 .results::<RunRow>()?;
504 Ok(rows.into_iter().map(|row| (row.repo_id.clone(), row.into_run(member))).collect())
505 }
506
507 /// The pull requests of some repositories with a recorded session,
508 /// most recently recorded first: each one's id, number, title, status,
509 /// agent label, author, first and last entry.
510 pub(crate) async fn recorded_in(&self, repo_ids: &[String], limit: u32) -> Result<Vec<RecordedRow>> {
511 if repo_ids.is_empty() {
512 return Ok(Vec::new());
513 }
514 self.db
515 .prepare(RECORDED_IN)
516 .bind(&[serde_json::to_string(repo_ids)?.into(), limit.into()])?
517 .all()
518 .await?
519 .results::<RecordedRow>()
520 }
521
522 pub(crate) async fn get_run(&self, a: GetRunArgs) -> Result<Outcome<AgentRun>> {
523 let (repo, member) = match self.visible_repo(&a.repo, &a.viewer).await? {
524 Outcome::Ok(found) => found,
525 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
526 };
527 Ok(match self.run_row(&a.id).await?.filter(|run| run.repo_id == repo.id) {
528 Some(row) => Outcome::Ok(row.into_run(member)),
529 None => Outcome::fail(FailureCode::NotFound, "Run not found."),
530 })
531 }
532
533 // --- Sessions ----------------------------------------------------------
534
535 pub(crate) async fn list_sessions(&self, a: ListSessionsArgs) -> Result<Outcome<Vec<SessionSummary>>> {
536 let Some(path) = &a.repo else {
537 return Ok(Outcome::fail(FailureCode::Invalid, "Name a repository."));
538 };
539 let (repo, member) = match self.visible_repo(path, &a.viewer).await? {
540 Outcome::Ok(found) => found,
541 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
542 };
543 let rows = self
544 .db
545 .prepare(
546 "SELECT p.id AS pull_id, p.number, p.title, p.status, p.agent,
547 count(*) AS entries,
548 sum(CASE WHEN e.kind = 'tool_call' THEN 1 ELSE 0 END) AS tools,
549 min(e.at) AS started_at, max(e.at) AS last_at,
550 (SELECT substr(f.text, 1, 300) FROM session_entries f
551 WHERE f.pull_id = p.id AND f.kind = 'prompt' ORDER BY f.seq LIMIT 1) AS prompt
552 FROM pulls p JOIN session_entries e ON e.pull_id = p.id
553 WHERE p.repo_id = ?1
554 AND (?2 IS NULL OR p.status = ?2)
555 AND (?3 IS NULL OR p.number = ?3)
556 AND (?4 IS NULL OR EXISTS
557 (SELECT 1 FROM agent_runs r WHERE r.pull_id = p.id AND r.kind = ?4))
558 GROUP BY p.id ORDER BY last_at DESC LIMIT 100",
559 )
560 .bind(&[
561 repo.id.as_str().into(),
562 a.outcome.map_or(JsValue::NULL, |status| status.as_str().into()),
563 a.number.map_or(JsValue::NULL, JsValue::from),
564 a.kind.map_or(JsValue::NULL, |kind| kind.as_str().into()),
565 ])?
566 .all()
567 .await?
568 .results::<SessionListRow>()?;
569 let ids: Vec<&str> = rows.iter().map(|row| row.pull_id.as_str()).collect();
570 let runs = if ids.is_empty() {
571 Vec::new()
572 } else {
573 self.db
574 .prepare(
575 "SELECT pull_id, kind, status, cost_usd FROM agent_runs
576 WHERE repo_id = ? AND pull_id IN (SELECT value FROM json_each(?))",
577 )
578 .bind(&[repo.id.as_str().into(), serde_json::to_string(&ids)?.into()])?
579 .all()
580 .await?
581 .results::<RunSumRow>()?
582 };
583 Ok(Outcome::Ok(
584 rows.into_iter()
585 .map(|row| {
586 let mine: Vec<&RunSumRow> = runs.iter().filter(|run| run.pull_id == row.pull_id).collect();
587 let mut kinds: Vec<RunKind> = Vec::new();
588 for run in &mine {
589 if let Some(kind) = RunKind::parse(&run.kind)
590 && !kinds.contains(&kind)
591 {
592 kinds.push(kind);
593 }
594 }
595 let spent: f64 = mine.iter().filter_map(|run| run.cost_usd).sum();
596 SessionSummary {
597 number: row.number,
598 title: row.title,
599 status: row.status,
600 agent: row.agent,
601 entries: row.entries,
602 tools: row.tools,
603 // Someone's words to their agent: not for outsiders,
604 // and to members without credentials or swearing.
605 prompt: row.prompt.filter(|_| member).map(|prompt| soften(&one_line(&prompt, 200))).filter(|prompt| !prompt.is_empty()),
606 runs: mine.len() as u32,
607 cost_usd: (member && mine.iter().any(|run| run.cost_usd.is_some())).then_some(spent),
608 active: mine
609 .iter()
610 .any(|run| RunStatus::parse(&run.status).is_some_and(RunStatus::is_active)),
611 kinds,
612 started_at: row.started_at,
613 last_at: row.last_at,
614 }
615 })
616 .collect(),
617 ))
618 }
619
620 pub(crate) async fn get_session(&self, a: GetSessionArgs) -> Result<Outcome<SessionView>> {
621 let (repo, pull) = match self.pull_at(&a.repo, a.number, &a.viewer).await? {
622 Outcome::Ok(found) => found,
623 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
624 };
625 let member = is_member(&a.viewer, &repo.namespace);
626 // Each entry as the viewer may see it: a prompt or a tool's output
627 // only by whoever's session recorded it, nothing to an outsider
628 // but g1t's own notes (coordination/shown.rs).
629 let entries: Vec<SessionEntry> = self
630 .db
631 .prepare(
632 "SELECT e.seq, e.kind, e.text, e.tool, e.\"commit\", e.at, s.actor_id FROM session_entries e
633 LEFT JOIN sessions s ON s.id = e.session_id
634 WHERE e.pull_id = ? ORDER BY e.seq LIMIT ?",
635 )
636 .bind(&[pull.id.as_str().into(), SESSION_ENTRIES.into()])?
637 .all()
638 .await?
639 .results::<SessionRow>()?
640 .into_iter()
641 .filter_map(|row| {
642 let note = row.kind == SessionEntryKind::Note && row.actor_id.is_none();
643 let sight = match &row.actor_id {
644 Some(actor) => Sight::of(&a.viewer, &repo.namespace, actor),
645 None if member => Sight::Member,
646 None => Sight::Outside,
647 };
648 let entry = SessionEntry::from(row);
649 if note && sight == Sight::Outside {
650 return Some(SessionEntry { text: shown::clean(&entry.text), ..entry });
651 }
652 shown::entry(entry, sight)
653 })
654 .collect();
655 let runs: Vec<AgentRun> = self
656 .db
657 .prepare(format!(
658 "SELECT {COLUMNS}, steps FROM agent_runs WHERE pull_id = ? ORDER BY created_at DESC LIMIT 50"
659 ))
660 .bind(&[pull.id.as_str().into()])?
661 .all()
662 .await?
663 .results::<RunRow>()?
664 .into_iter()
665 .map(|row| row.into_run(member))
666 .collect();
667 let cost_usd = (member && runs.iter().any(|run| run.cost_usd.is_some()))
668 .then(|| runs.iter().filter_map(|run| run.cost_usd).sum());
669 Ok(Outcome::Ok(SessionView {
670 pull,
671 entries,
672 runs,
673 cost_usd,
674 }))
675 }
676}
677
678#[cfg(test)]
679mod tests {
680 use super::*;
681
682 #[test]
683 fn transfer_and_deletion_statements_take_what_they_name() {
684 for sql in TRANSFERRED.iter().chain(crate::guardrails::TRANSFERRED) {
685 let n = g1t_kit::transfer::parameters(sql);
686 assert!((1..=5).contains(&n), "{sql}");
687 }
688 for sql in crate::guardrails::DELETED {
689 assert_eq!(g1t_kit::rename::parameters(sql), 1, "{sql}");
690 }
691 }
692
693 #[test]
694 fn rename_statements_take_the_two_slugs() {
695 for sql in RENAMED.iter().chain(crate::guardrails::RENAMED) {
696 assert_eq!(g1t_kit::rename::parameters(sql), 2, "{sql}");
697 }
698 }
699
700 #[test]
701 fn a_refused_host_is_noted_but_never_where_the_run_is() {
702 let steps = |list: &[&str]| list.iter().map(|step| (*step).to_owned()).collect::<Vec<String>>();
703 assert_eq!(current_step(&steps(&["Running npm ci", "Blocked: sparrow.cloudflare.com (not an allowed domain)"])).as_deref(), Some("Running npm ci"));
704 assert_eq!(current_step(&steps(&["Blocked: a.com (not an allowed domain)"])), None);
705 assert_eq!(current_step(&steps(&["Blocked: a.com (not an allowed domain)", "Typecheck"])).as_deref(), Some("Typecheck"));
706 assert_eq!(current_step(&[]), None);
707 }
708
709 #[test]
710 fn steps_are_one_short_line() {
711 assert_eq!(one_line(" Read\n src/lib.rs ", 40), "Read src/lib.rs");
712 let long = one_line(&"x".repeat(300), 10);
713 assert_eq!(long.chars().count(), 10);
714 assert!(long.ends_with('…'));
715 }
716}