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