Skip to content
Sign up

flagon-io/g1t

public
0

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

g1t.sh
gitagentscloudflarerustforgemcp
450 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
10use g1t_contracts::agents::*;
11use g1t_contracts::repos::{Repo, RepoPath};
12use g1t_contracts::time::rfc3339;
13use g1t_contracts::{FailureCode, Membership, Outcome, User, Viewer, new_id};
14use g1t_kit::now_ms;
15use serde::Deserialize;
16use worker::Result;
17use worker::wasm_bindgen::JsValue;
18
19use crate::checks::{hash, new_token};
20use crate::{Work, optional};
21
22/// How the runner starts a step that notes something about the sandbox
23/// rather than what the run is doing: a host it was refused (the runner's
24/// egress.ts `blockedStep`).
25const NOTE_PREFIXES: [&str; 1] = ["Blocked: "];
26
27/// Where a report leaves the run: its newest step that says what the run
28/// is doing. A note (a refused host) stays in the steps but never stands
29/// for the run, which carries on past it. `None` when there is only notes.
30fn current_step(steps: &[String]) -> Option<String> {
31 steps.iter().rev().find(|step| !NOTE_PREFIXES.iter().any(|prefix| step.starts_with(prefix))).cloned()
32}
33
34/// The most steps a run keeps; older ones fall off the start.
35const MAX_STEPS: u32 = 200;
36/// The most steps taken from one report.
37const MAX_STEPS_PER_REPORT: usize = 20;
38/// The longest a step is kept.
39const MAX_STEP_CHARS: usize = 240;
40/// A run that has said nothing for this long is taken to have died.
41const SILENT_HOURS: u64 = 3;
42const DEFAULT_LIST: u32 = 50;
43const MAX_LIST: u32 = 200;
44
45/// Every column but the steps, which only a run's own page needs.
46const COLUMNS: &str = "id, workspace, repo_id, repo, number, pull_id, kind, agent, model, status,
47 step, step_count, cost_usd, turns, sandbox, token_hash, started_by, error, created_at,
48 started_at, finished_at, updated_at, budget_usd, time_cap_minutes, halted,
49 COALESCE((SELECT title FROM pulls WHERE pulls.id = agent_runs.pull_id), agent_runs.title) AS title,
50 (SELECT detail FROM run_confidence WHERE run_confidence.run_id = agent_runs.id) AS confidence";
51
52#[derive(Deserialize)]
53pub(crate) struct RunRow {
54 id: String,
55 #[allow(dead_code)]
56 workspace: String,
57 #[allow(dead_code)]
58 repo_id: String,
59 repo: String,
60 number: Option<u32>,
61 pub(crate) pull_id: Option<String>,
62 kind: String,
63 agent: String,
64 model: Option<String>,
65 status: String,
66 step: Option<String>,
67 #[serde(default)]
68 steps: Option<String>,
69 step_count: u32,
70 cost_usd: Option<f64>,
71 turns: Option<u32>,
72 token_hash: String,
73 started_by: Option<String>,
74 error: Option<String>,
75 created_at: String,
76 started_at: Option<String>,
77 finished_at: Option<String>,
78 updated_at: String,
79 title: Option<String>,
80 #[serde(default)]
81 budget_usd: Option<f64>,
82 #[serde(default)]
83 time_cap_minutes: Option<u32>,
84 #[serde(default)]
85 halted: Option<String>,
86 /// JSON of how sure g1t was of the change as the run left it.
87 #[serde(default)]
88 confidence: Option<String>,
89}
90
91impl RunRow {
92 fn status(&self) -> RunStatus {
93 RunStatus::parse(&self.status).unwrap_or(RunStatus::Failed)
94 }
95
96 /// The run as `member` (or not) may see it: model and cost are the
97 /// workspace's business.
98 fn into_run(self, member: bool) -> AgentRun {
99 let (namespace, name) = self.repo.split_once('/').unwrap_or((&self.repo, ""));
100 AgentRun {
101 repo: RepoPath {
102 namespace: namespace.to_owned(),
103 name: name.to_owned(),
104 },
105 status: self.status(),
106 kind: RunKind::parse(&self.kind).unwrap_or(RunKind::Implement),
107 steps: self
108 .steps
109 .as_deref()
110 .and_then(|steps| serde_json::from_str(steps).ok())
111 .unwrap_or_default(),
112 id: self.id,
113 number: self.number.filter(|number| *number > 0),
114 title: self.title,
115 agent: self.agent,
116 model: self.model.filter(|_| member),
117 step: self.step,
118 step_count: self.step_count,
119 started_by: self.started_by,
120 error: self.error,
121 cost_usd: self.cost_usd.filter(|_| member),
122 turns: self.turns,
123 budget_usd: self.budget_usd.filter(|_| member),
124 time_cap_minutes: self.time_cap_minutes,
125 halted: self.halted,
126 confidence: self
127 .confidence
128 .as_deref()
129 .and_then(|detail| serde_json::from_str(detail).ok()),
130 created_at: self.created_at,
131 started_at: self.started_at,
132 finished_at: self.finished_at,
133 updated_at: self.updated_at,
134 }
135 }
136}
137
138/// One line, short enough to read at a glance.
139pub(crate) fn one_line(text: &str, limit: usize) -> String {
140 let line = text.split_whitespace().collect::<Vec<_>>().join(" ");
141 if line.chars().count() <= limit {
142 return line;
143 }
144 let mut short: String = line.chars().take(limit.saturating_sub(1)).collect();
145 short.push('…');
146 short
147}
148
149/// A principal that can read any repository of `namespace`, for lookups a
150/// trusted service asks for. What it is given reads and nothing more.
151pub(crate) fn member_of(actor: &User, namespace: &str) -> Viewer {
152 let mut viewer = actor.clone();
153 let slug = namespace.to_lowercase();
154 if !viewer.is_member(&slug) {
155 viewer.workspaces.push(Membership {
156 base_permission: Some(g1t_contracts::access::BasePermission::Read),
157 ..Membership::member(slug)
158 });
159 }
160 Some(viewer)
161}
162
163/// Whether the viewer belongs to the workspace: what a run spent, its model
164/// and its budget are the workspace's business, not every reader's (an
165/// outside collaborator sees the work, not the bill).
166fn is_member(viewer: &Viewer, namespace: &str) -> bool {
167 viewer
168 .as_ref()
169 .is_some_and(|viewer| viewer.is_member(&namespace.to_lowercase()))
170}
171
172impl Work {
173 /// The repository at `path`, if `viewer` may see it, and whether they
174 /// are a member of its workspace (which shows what runs cost).
175 async fn visible_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<(Repo, bool)>> {
176 Ok(match self.repo(path, viewer).await? {
177 Outcome::Ok(repo) => {
178 let member = is_member(viewer, &repo.namespace);
179 Outcome::Ok((repo, member))
180 }
181 Outcome::Fail(failure) => Outcome::Fail(failure),
182 })
183 }
184
185 async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {
186 self.db
187 .prepare(format!("SELECT {COLUMNS}, steps FROM agent_runs WHERE id = ?"))
188 .bind(&[id.into()])?
189 .first::<RunRow>(None)
190 .await
191 }
192
193 pub(crate) async fn open_run(&self, a: OpenRunArgs) -> Result<Outcome<AgentRunTicket>> {
194 // Only sandboxes that are not agents are started now; earlier
195 // kinds stay readable on the runs already recorded.
196 if !matches!(a.kind, RunKind::Checks | RunKind::Queue | RunKind::Mergecheck) {
197 return Ok(Outcome::fail(
198 FailureCode::Invalid,
199 format!("A {} run is no longer started.", a.kind.as_str()),
200 ));
201 }
202 // The runner is trusted: it names the repository it is starting a
203 // sandbox in, whoever the sandbox acts as. The run records where
204 // the repository is now, looked up by its id, not the path it was
205 // named by, which may be from before a transfer or a rename.
206 let (repo_id, path) = match &a.pull_id {
207 Some(pull_id) => match self.pull_by_id(pull_id).await? {
208 Some(pull) => {
209 let now: Option<RepoPath> = g1t_kit::call(
210 &self.repos,
211 "path_by_id",
212 &g1t_contracts::repos::PathByIdArgs { id: pull.repo_id.clone() },
213 )
214 .await?;
215 (pull.repo_id, now.unwrap_or_else(|| a.repo.clone()))
216 }
217 None => return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found.")),
218 },
219 None => match self.repo(&a.repo, &member_of(&a.actor, &a.repo.namespace)).await? {
220 Outcome::Ok(repo) => (repo.id, RepoPath { namespace: repo.namespace, name: repo.name }),
221 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
222 },
223 };
224 let namespace = path.namespace.to_lowercase();
225 // Nothing new starts on an archived or deleted repository.
226 if !self.repo_active(&repo_id).await? {
227 return Ok(Outcome::fail(
228 FailureCode::Forbidden,
229 format!("{}/{} is archived or deleted, so nothing new starts on it.", path.namespace, path.name),
230 ));
231 }
232 let now = now_ms();
233 let id = new_id("arn", now);
234 let token = new_token();
235 let timestamp = rfc3339(now);
236 let agent = a
237 .agent
238 .clone()
239 .unwrap_or_else(|| g1t_contracts::identity::AGENT_NAME.to_owned());
240 self.db
241 .prepare(
242 "INSERT INTO agent_runs
243 (id, workspace, repo_id, repo, number, pull_id, title, kind, agent, model, status,
244 sandbox, token_hash, started_by, created_at, updated_at, budget_usd, time_cap_minutes)
245 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?)",
246 )
247 .bind(&[
248 id.as_str().into(),
249 namespace.into(),
250 repo_id.into(),
251 format!("{}/{}", path.namespace, path.name).into(),
252 a.number.filter(|n| *n > 0).map_or(JsValue::NULL, JsValue::from),
253 optional(&a.pull_id),
254 optional(&a.title.map(|title| one_line(&title, 200))),
255 a.kind.as_str().into(),
256 agent.into(),
257 optional(&a.model),
258 a.sandbox.into(),
259 hash(&token).into(),
260 optional(&a.started_by),
261 timestamp.as_str().into(),
262 timestamp.as_str().into(),
263 a.budget_usd.filter(|usd| usd.is_finite() && *usd > 0.0).map_or(JsValue::NULL, JsValue::from),
264 a.time_cap_minutes.map_or(JsValue::NULL, JsValue::from),
265 ])?
266 .run()
267 .await?;
268 Ok(Outcome::Ok(AgentRunTicket { run_id: id, token }))
269 }
270
271 /// Adds `text` to a run's steps, keeping the latest `MAX_STEPS`.
272 pub(crate) fn add_step(&self, run_id: &str, at: &str, text: &str) -> Result<worker::D1PreparedStatement> {
273 self.db
274 .prepare(
275 "UPDATE agent_runs SET
276 steps = CASE WHEN json_array_length(steps) >= ?4
277 THEN json_insert(json_remove(steps, '$[0]'), '$[#]', json_object('at', ?1, 'text', ?2))
278 ELSE json_insert(steps, '$[#]', json_object('at', ?1, 'text', ?2)) END,
279 step_count = step_count + 1
280 WHERE id = ?3",
281 )
282 .bind(&[at.into(), text.into(), run_id.into(), MAX_STEPS.into()])
283 }
284
285 pub(crate) async fn report_run(&self, a: ReportRunArgs) -> Result<Outcome<RunStatus>> {
286 let Some(run) = self
287 .run_row(&a.run_id)
288 .await?
289 .filter(|run| run.token_hash == hash(&a.token))
290 else {
291 return Ok(Outcome::fail(FailureCode::NotFound, "Run not found."));
292 };
293 // Finished, or stopped by a person: nothing more is taken, and the
294 // sandbox learns why.
295 if run.finished_at.is_some() {
296 return Ok(Outcome::Ok(run.status()));
297 }
298 // It reached a cap of its guardrails (guardrails.rs).
299 if let Some(halt) = a.halt {
300 let pull_id = run.pull_id.as_deref();
301 return self
302 .halt_run(&run.id, &run.repo_id, pull_id, run.number, &run.kind, halt, a.error, a.cost_usd)
303 .await;
304 }
305 let now = rfc3339(now_ms());
306 let steps: Vec<String> = a
307 .steps
308 .iter()
309 .map(|step| one_line(step, MAX_STEP_CHARS))
310 .filter(|step| !step.is_empty())
311 .collect();
312 // A burst keeps its end: that is where the run is now.
313 let skip = steps.len().saturating_sub(MAX_STEPS_PER_REPORT);
314 let mut statements = Vec::new();
315 for step in &steps[skip..] {
316 statements.push(self.add_step(&run.id, &now, step)?);
317 }
318 let current = a
319 .step
320 .as_deref()
321 .map(|step| one_line(step, MAX_STEP_CHARS))
322 .filter(|step| !step.is_empty())
323 .or_else(|| current_step(&steps));
324 let outcome = a
325 .outcome
326 .filter(|outcome| matches!(outcome, RunStatus::Succeeded | RunStatus::Failed));
327 let cost = a.cost_usd.filter(|cost| cost.is_finite() && *cost >= 0.0);
328 statements.push(
329 self.db
330 .prepare(
331 "UPDATE agent_runs SET
332 status = COALESCE(?1, CASE WHEN status = 'queued' THEN 'running' ELSE status END),
333 started_at = COALESCE(started_at, ?2),
334 step = COALESCE(?3, step),
335 cost_usd = COALESCE(?4, cost_usd),
336 turns = COALESCE(?5, turns),
337 error = COALESCE(?6, error),
338 finished_at = CASE WHEN ?1 IS NULL THEN finished_at ELSE ?2 END,
339 updated_at = ?2
340 WHERE id = ?7 AND finished_at IS NULL",
341 )
342 .bind(&[
343 outcome.map_or(JsValue::NULL, |outcome| outcome.as_str().into()),
344 now.as_str().into(),
345 optional(&current),
346 cost.map_or(JsValue::NULL, JsValue::from),
347 a.turns.map_or(JsValue::NULL, JsValue::from),
348 optional(&a.error.map(|error| one_line(&error, 1000))),
349 run.id.as_str().into(),
350 ])?,
351 );
352 self.db.batch(statements).await?;
353 Ok(Outcome::Ok(outcome.unwrap_or(RunStatus::Running)))
354 }
355
356 /// Runs whose sandbox died without anyone noticing, marked failed.
357 pub(crate) async fn sweep_silent(&self) -> Result<()> {
358 let now = now_ms();
359 let cutoff = rfc3339(now.saturating_sub(SILENT_HOURS * 3_600_000));
360 self.db
361 .prepare(
362 "UPDATE agent_runs
363 SET status = 'failed', error = COALESCE(error, 'It stopped reporting.'),
364 finished_at = ?, updated_at = ?
365 WHERE status IN ('queued', 'running') AND updated_at < ?",
366 )
367 .bind(&[rfc3339(now).into(), rfc3339(now).into(), cutoff.into()])?
368 .run()
369 .await?;
370 Ok(())
371 }
372
373 pub(crate) async fn list_runs(&self, a: ListRunsArgs) -> Result<Outcome<Vec<AgentRun>>> {
374 let (column, key, member) = if let Some(workspace) = &a.workspace {
375 let slug = workspace.to_lowercase();
376 if !is_member(&a.viewer, &slug) {
377 return Ok(Outcome::fail(
378 FailureCode::Forbidden,
379 "Only members of the workspace can see its fleet.",
380 ));
381 }
382 ("workspace", slug, true)
383 } else if let Some(path) = &a.repo {
384 match self.visible_repo(path, &a.viewer).await? {
385 Outcome::Ok((repo, member)) => ("repo_id", repo.id, member),
386 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
387 }
388 } else {
389 return Ok(Outcome::fail(FailureCode::Invalid, "Name a repository or a workspace."));
390 };
391 self.sweep_silent().await?;
392 let limit = a.limit.unwrap_or(DEFAULT_LIST).clamp(1, MAX_LIST);
393 let rows = self
394 .db
395 .prepare(format!(
396 "SELECT {COLUMNS} FROM agent_runs
397 WHERE {column} = ?1
398 AND (?2 = 0 OR status IN ('queued', 'running'))
399 AND (?3 IS NULL OR kind = ?3)
400 AND (?4 IS NULL OR status = ?4)
401 AND (?5 IS NULL OR number = ?5)
402 ORDER BY created_at DESC LIMIT ?6"
403 ))
404 .bind(&[
405 key.into(),
406 u32::from(a.active).into(),
407 a.kind.map_or(JsValue::NULL, |kind| kind.as_str().into()),
408 a.status.map_or(JsValue::NULL, |status| status.as_str().into()),
409 a.number.map_or(JsValue::NULL, JsValue::from),
410 limit.into(),
411 ])?
412 .all()
413 .await?
414 .results::<RunRow>()?;
415 Ok(Outcome::Ok(rows.into_iter().map(|row| row.into_run(member)).collect()))
416 }
417
418 pub(crate) async fn get_run(&self, a: GetRunArgs) -> Result<Outcome<AgentRun>> {
419 let (repo, member) = match self.visible_repo(&a.repo, &a.viewer).await? {
420 Outcome::Ok(found) => found,
421 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
422 };
423 Ok(match self.run_row(&a.id).await?.filter(|run| run.repo_id == repo.id) {
424 Some(row) => Outcome::Ok(row.into_run(member)),
425 None => Outcome::fail(FailureCode::NotFound, "Run not found."),
426 })
427 }
428}
429
430#[cfg(test)]
431mod tests {
432 use super::*;
433
434 #[test]
435 fn a_refused_host_is_noted_but_never_where_the_run_is() {
436 let steps = |list: &[&str]| list.iter().map(|step| (*step).to_owned()).collect::<Vec<String>>();
437 assert_eq!(current_step(&steps(&["Running npm ci", "Blocked: sparrow.cloudflare.com (not an allowed domain)"])).as_deref(), Some("Running npm ci"));
438 assert_eq!(current_step(&steps(&["Blocked: a.com (not an allowed domain)"])), None);
439 assert_eq!(current_step(&steps(&["Blocked: a.com (not an allowed domain)", "Typecheck"])).as_deref(), Some("Typecheck"));
440 assert_eq!(current_step(&[]), None);
441 }
442
443 #[test]
444 fn steps_are_one_short_line() {
445 assert_eq!(one_line(" Read\n src/lib.rs ", 40), "Read src/lib.rs");
446 let long = one_line(&"x".repeat(300), 10);
447 assert_eq!(long.chars().count(), 10);
448 assert!(long.ends_with('…'));
449 }
450}