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