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
2,856 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1//! A run's life: made, its jobs waiting on the jobs they need, each job
2//! skipped or expanded into its matrix and queued, started in a sandbox
3//! when its workspace has room, reporting its steps and logs as it goes,
4//! and finished; the run finishes with its last job.
5
6use g1t_actions::events::{RunInfo, runner_context};
7use g1t_actions::expr::{self, Scope, Status};
8use g1t_actions::matrix;
9use g1t_actions::workflow::{self, Workflow};
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look10use g1t_contracts::access::Capability;
Merge branch 'worktree-agent-a3abfcce648e87dca'11use g1t_contracts::actions::{JobCallArgs, RunActionArgs, RunApproval, StartJobArgs, WorkflowRun};
12use g1t_contracts::identity::{CreateJobTokenArgs, CreatedAccessToken, RevokeJobTokensArgs};
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API13use g1t_contracts::repos::{Repo, RepoPath};
14use g1t_contracts::time::rfc3339;
15use g1t_contracts::{FailureCode, Outcome, new_id};
16use g1t_kit::now_ms;
17use g1t_secrets::{random_hex, same, sha256_hex};
18use serde::Deserialize;
19use serde_json::{Map, Value, json};
20use worker::Result;
21
Merge branch 'worktree-agent-a3abfcce648e87dca'22use crate::protection::Gate;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API23use crate::sync::WorkflowRow;
Fast pages, required checks on the branch, self-hosted runners, honest incidents24use crate::{Actions, Count, MAX_TIMEOUT_MINUTES, RUNNING_PER_WORKSPACE, SELF_HOSTED_MAX_TIMEOUT_MINUTES, SILENT_MS, SITE, check, fail, optional, repo_path};
25use g1t_contracts::runners::{Wanted, waiting_reason};
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API26
27/// The most log one job keeps, in bytes; past it, the log says so and stops.
28const MAX_LOG_BYTES: usize = 4 * 1024 * 1024;
29/// The most a single log report may add.
30const MAX_CHUNK_BYTES: usize = 256 * 1024;
31const MAX_ANNOTATIONS: usize = 50;
A running job's log is live, as on GitHub. A step's output reaches its run's page within about two seconds: the runner sends what is printed every second, and the page asks every second while a step runs, every three while none does; an answer that fails (a 429, a 5xx, the network) is asked again after a wait that doubles to half a minute, or as long as Retry-After says, where before one such answer stopped the log for good and the step said "No output yet." until the page was reloaded, and a signed-out viewer's run page, refreshing beside its log, ran into the 30-a-minute limit within a minute; the live log now counts as a page, and only log downloads as costly. A hidden tab stops asking and catches up when shown, and the page says when it is reconnecting. The running step opens by itself, its timer counts, its log follows the newest line while its end is on screen and offers Jump to latest once the viewer scrolls up, a step that has printed nothing for a while says "Running · no output for 2m 10s", and the job's header names the step it is on. Colour that tools print (bold, dim, italic, underline, the 16 colours as theme tokens for light and dark, 256 colours and 24-bit colour, text and background) is drawn as printed, other escape codes are dropped and links keep their text, and search reads the plain text and keeps the colours around what it marks. Every line is stamped with the time it was printed, as GitHub's raw logs are, so log.txt and the run's zip have a time at the start of each line, and a clock button beside Search logs shows the times, remembered by the browser. Set up job now gives the runner's version and name, the operating system, the runner image, the machine's architecture, CPUs and memory, the token's permissions under G1T_TOKEN Permissions, the secret source and the actions it downloads; each run step opens with GitHub's folded Run group, the script in cyan, the shell's whole path and the step's env, each uses step with its with inputs (an action's defaults among them) and env; cache and artifact transfers say how far they have got every five seconds; and a last Complete job step sets the job's outputs and stops what the steps left running, found by RUNNER_TRACKING_ID as GitHub's runner finds it. Two batches of a job's log written at once can no longer reach for the same seq and lose one. The Actions guide and the rate limits page say so.32/// A batch of a job's log, under the next `seq` of that job's, worked out
33/// in the same statement (SQLite runs one write at a time).
34const LOG_INSERT: &str =
35 "INSERT INTO logs (job_id, seq, step, text) SELECT ?1, COALESCE(MAX(seq), 0) + 1, ?2, ?3 FROM logs WHERE job_id = ?1";
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)36/// The most one step's job summary keeps, in bytes, and how many steps of a
37/// job may have one, as on GitHub.
38pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
39pub(crate) const MAX_SUMMARIES: u32 = 20;
40
41/// Whether a step's summary of `adding` bytes is kept: a job keeps up to
42/// 20 steps' summaries (`steps` it has, `mine` this step's bytes so far)
43/// of up to 1 MiB each.
44pub(crate) fn summary_fits(steps: u32, mine: Option<usize>, adding: usize) -> bool {
45 if adding == 0 {
46 return false;
47 }
48 match mine {
49 Some(used) => used + adding <= MAX_SUMMARY_BYTES,
50 None => steps < MAX_SUMMARIES && adding <= MAX_SUMMARY_BYTES,
51 }
52}
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API53
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look54/// The repository whose unfinished runs `event` stops, by id: one deleted,
55/// or archived (not unarchived).
56pub fn stops_runs(event: &g1t_contracts::events::Event) -> Option<String> {
57 let stops = match event.kind.as_str() {
58 "repo.deleted" => true,
59 "repo.archived" => event.data["archived"].as_bool() != Some(false),
60 _ => false,
61 };
62 if !stops {
63 return None;
64 }
65 event.repo_id.clone().or_else(|| event.data["repoId"].as_str().map(str::to_owned))
66}
67
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API68pub struct NewRun {
69 pub repo: Repo,
70 pub path: String,
71 pub source: String,
72 pub workflow: Workflow,
73 pub info: RunInfo,
74 pub action: Option<String>,
75 pub pull: Option<u32>,
76 pub title: String,
77 pub inputs: Map<String, Value>,
78 pub event_key: String,
79 pub actor_id: Option<String>,
80 pub actor: Option<String>,
81 pub trusted: bool,
Merge branch 'worktree-agent-a3abfcce648e87dca'82 /// Why the run waits for someone with the Write role to approve it
83 /// before anything starts: a pull request from outside (protection.rs).
84 pub approval: Option<String>,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API85}
86
87#[derive(Clone, Deserialize)]
88pub struct RunRow {
89 pub id: String,
90 pub workflow_id: String,
91 pub repo_id: String,
92 pub repo: String,
93 pub path: String,
94 pub name: String,
95 pub title: String,
96 pub number: u64,
97 pub attempt: u64,
98 pub event: String,
99 pub action: Option<String>,
100 pub git_ref: String,
101 pub sha: String,
102 pub pull: Option<u32>,
103 pub status: String,
104 pub conclusion: Option<String>,
105 pub error: Option<String>,
106 pub actor: Option<String>,
107 pub actor_id: Option<String>,
108 pub source: String,
109 pub info: String,
110 pub inputs: String,
111 pub trusted: u32,
112 pub concurrency_group: Option<String>,
113 pub created_at: String,
114 pub started_at: Option<String>,
115 pub finished_at: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'116 /// JSON `RunApproval`, for a run that needed approval (migration 0006).
117 #[serde(default)]
118 pub approval: Option<String>,
119 /// Whether its concurrency group cancels what it replaces.
120 #[serde(default)]
121 pub cancel_in_progress: u32,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)122 /// Who started the current attempt, once it is re-run (migration 0009).
123 #[serde(default)]
124 pub triggering_actor: Option<String>,
125 /// The current attempt runs with debug logging.
126 #[serde(default)]
127 pub debug: u32,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API128}
129
130impl RunRow {
Merge branch 'worktree-agent-a3abfcce648e87dca'131 pub fn approval(&self) -> Option<RunApproval> {
132 self.approval.as_deref().and_then(|text| serde_json::from_str(text).ok())
133 }
134
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API135 pub fn info(&self) -> RunInfo {
136 let mut info: RunInfo = serde_json::from_str(&self.info).unwrap_or_default();
137 info.run_id = self.id.clone();
138 info.run_number = self.number;
139 info.run_attempt = self.attempt;
140 info.workflow_path = self.path.clone();
141 if info.workflow.is_empty() {
142 info.workflow = self.name.clone();
143 }
144 info
145 }
146
147 pub fn inputs(&self) -> Map<String, Value> {
148 serde_json::from_str(&self.inputs).unwrap_or_default()
149 }
150
151 pub fn summary(&self) -> WorkflowRun {
152 WorkflowRun {
153 id: self.id.clone(),
154 workflow_id: self.workflow_id.clone(),
155 path: self.path.clone(),
156 name: self.name.clone(),
157 title: self.title.clone(),
158 number: self.number,
159 attempt: self.attempt,
160 event: self.event.clone(),
161 git_ref: self.git_ref.clone(),
162 sha: self.sha.clone(),
163 pull: self.pull,
164 status: self.status.clone(),
165 conclusion: self.conclusion.clone(),
166 error: self.error.clone(),
167 actor: self.actor.clone(),
168 created_at: self.created_at.clone(),
169 started_at: self.started_at.clone(),
170 finished_at: self.finished_at.clone(),
171 }
172 }
173}
174
175#[derive(Clone, Deserialize)]
176pub struct JobRow {
177 pub id: String,
178 pub run_id: String,
179 pub repo_id: String,
180 pub namespace: String,
181 pub key: String,
182 pub ordinal: u32,
183 pub name: String,
184 pub needs: String,
185 pub matrix: Option<String>,
186 pub status: String,
187 pub conclusion: Option<String>,
188 pub steps: String,
189 pub annotations: String,
190 pub outputs: String,
191 pub reason: Option<String>,
192 pub token_hash: Option<String>,
193 pub timeout_minutes: u32,
194 pub continue_on_error: u32,
195 pub max_parallel: Option<u32>,
196 /// Set for a job that calls a reusable workflow, and for that
197 /// workflow's jobs (see migration 0002).
198 pub call: Option<String>,
199 pub seen_at: Option<String>,
200 pub started_at: Option<String>,
201 pub finished_at: Option<String>,
Fast pages, required checks on the branch, self-hosted runners, honest incidents202 /// For a job whose `runs-on` names self-hosted runners: what it asks
203 /// for, as a JSON array (see `g1t_contracts::runners::Wanted`), when it
204 /// started waiting, and the runner that took it (migration 0004).
205 #[serde(default)]
206 pub labels: Option<String>,
207 #[serde(default)]
208 pub queued_at: Option<String>,
209 #[serde(default)]
210 pub runner_id: Option<String>,
211 #[serde(default)]
212 pub runner_name: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'213 /// The environment it names, read when its needs were done; a job its
214 /// rules hold is `pending` (migration 0006).
215 #[serde(default)]
216 pub environment: Option<String>,
217 /// Its own `concurrency` group, and whether that cancels what it replaces.
218 #[serde(default)]
219 pub concurrency_group: Option<String>,
220 #[serde(default)]
221 pub cancel_in_progress: u32,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)222 /// When it was told to stop, while it runs its cleanup steps
223 /// (migration 0009).
224 #[serde(default)]
225 pub cancel_requested_at: Option<String>,
226}
227
228/// How long a cancelled job has to run its `if: always()` and `cancelled()`
229/// steps and its post steps before it is stopped outright, as on GitHub.
230pub const CANCEL_GRACE_MS: u64 = 5 * 60 * 1000;
231
232/// Which jobs a re-run runs again.
233#[derive(Clone, Copy, Debug, PartialEq)]
234pub enum Rerun<'a> {
235 All,
236 /// Those that did not succeed.
237 Failed,
238 /// One job's key.
239 Job(&'a str),
240}
241
242/// The keys a re-run runs again, in `order`: those `which` names and,
243/// transitively, every key that needs one of them. `needs` gives a key's
244/// needs; `succeeded` whether every job of a key succeeded.
245pub fn rerun_keys(order: &[&str], needs: impl Fn(&str) -> Vec<String>, succeeded: impl Fn(&str) -> bool, which: Rerun) -> Vec<String> {
246 let mut again: Vec<String> = Vec::new();
247 for key in order {
248 let named = match which {
249 Rerun::All => true,
250 Rerun::Failed => !succeeded(key),
251 Rerun::Job(job) => *key == job,
252 };
253 if named || needs(key).iter().any(|need| again.contains(need)) {
254 again.push((*key).to_owned());
255 }
256 }
257 again
258}
259
260/// The key under `jobs:` a job belongs to: a called workflow's jobs
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9261/// (`build/test`, or `build[1]/test` for a matrix's second call) run again
262/// with the job that calls it (`build`).
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)263pub fn top_key(key: &str) -> &str {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9264 let first = key.split('/').next().unwrap_or(key);
265 first.split('[').next().unwrap_or(first)
266}
267
268/// What the keys of the jobs a caller's call makes start with: the
269/// caller's key, and for a matrix the combination's index after it
270/// (`build[1]`), so each combination's call has jobs of its own. A job id
271/// has no `[`, so the two never meet.
272fn call_prefix(key: &str, index: Option<usize>) -> String {
273 match index {
274 Some(index) => format!("{key}[{index}]"),
275 None => key.to_owned(),
276 }
277}
278
279/// Whether a run may call one more reusable workflow, `file`, `depth`
280/// levels below the workflow it started from, having called `called`
281/// already (one name per call; repeats are one workflow). As on GitHub: ten
282/// levels of workflows at most, the top one and nine reusable ones below
283/// it, and 50 different reusable workflows in a run.
284fn call_allowed(depth: u64, called: &[&str], file: &str) -> std::result::Result<(), String> {
285 if depth > MAX_CALL_DEPTH {
286 return Err(format!(
287 "`{file}` would be the {}th level of workflows: a run connects at most ten, the workflow it started from and nine levels of reusable workflows below it.",
288 depth + 1
289 ));
290 }
291 let mut unique: Vec<&str> = called.to_vec();
292 unique.push(file);
293 unique.sort_unstable();
294 unique.dedup();
295 if unique.len() > MAX_CALLED_WORKFLOWS {
296 return Err(format!("Calling `{file}` would make {} reusable workflows in this run: a run calls at most {MAX_CALLED_WORKFLOWS} different ones.", unique.len()));
297 }
298 Ok(())
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)299}
300
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9301/// What a job is shown as: its `name` (else its id), under the job whose
302/// call made it (`caller`), read with its matrix when it has an
303/// expression, else with the matrix's values after it.
304fn shown_name(job: &workflow::Job, caller: Option<&str>, combination: &Map<String, Value>, scope: &Scope) -> String {
305 let base = job.name.clone().unwrap_or(job.id.clone());
306 let base = match caller {
307 Some(caller) => format!("{caller} / {base}"),
308 None => base,
309 };
310 if expr::has_expression(&base) {
311 expr::interpolate(&base, scope).unwrap_or(base)
312 } else if job.matrix.is_some() {
313 matrix::job_name(&base, combination)
314 } else {
315 base
316 }
317}
318
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)319/// Debug logging for a job, as a re-run with it turned on gives GitHub's:
320/// `RUNNER_DEBUG=1` and `runner.debug`, and `ACTIONS_STEP_DEBUG` and
321/// `ACTIONS_RUNNER_DEBUG` set, which the runner reads as the secrets of
322/// those names (`::debug::` lines shown, and how each step's `if` read).
323pub fn debug_logging(variables: &mut Map<String, Value>, runner: &mut Value) {
324 variables.insert("RUNNER_DEBUG".into(), json!("1"));
325 variables.insert("ACTIONS_STEP_DEBUG".into(), json!("true"));
326 variables.insert("ACTIONS_RUNNER_DEBUG".into(), json!("true"));
327 runner["debug"] = json!("1");
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API328}
329
330impl JobRow {
331 pub fn needs(&self) -> Vec<String> {
332 serde_json::from_str(&self.needs).unwrap_or_default()
333 }
334
335 pub fn call(&self) -> Option<Value> {
336 self.call.as_deref().and_then(|call| serde_json::from_str(call).ok())
337 }
338
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9339 /// For a job that calls a reusable workflow: what the keys of the jobs
340 /// its call makes start with (`call_prefix`).
341 pub fn call_prefix(&self) -> Option<String> {
342 let call = self.call().filter(|call| call["role"] == "caller")?;
343 Some(call["prefix"].as_str().unwrap_or(&self.key).to_owned())
344 }
345
346 /// A job that calls a reusable workflow and has not called it yet: it
347 /// waits for room under its matrix's `max-parallel`, or in its
348 /// concurrency group.
349 pub fn holds_call(&self) -> bool {
350 self.status == "waiting" && self.call_prefix().is_some()
351 }
352
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API353 /// For a job of a called workflow: that workflow, the job's own id in
354 /// it, and the job.
355 pub fn callee(&self) -> Option<(Workflow, workflow::Job, Value)> {
356 let call = self.call().filter(|call| call["role"] == "callee")?;
357 let called = workflow::parse(call["source"].as_str()?).ok()?;
358 let job = called.jobs.iter().find(|job| call["job"].as_str() == Some(job.id.as_str()))?.clone();
359 Some((called, job, call))
360 }
361}
362
363/// A job to decide on: its key, its definition, and what it needs, as
364/// (name in `needs`, key of the jobs).
365type Unit = (String, workflow::Job, Vec<(String, String)>);
366
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9367/// How many levels of reusable workflows may sit below the workflow a run
368/// started from, as on GitHub (ten levels of workflows in all).
369const MAX_CALL_DEPTH: u64 = 9;
370/// How many different reusable workflows a run may call, as on GitHub.
371const MAX_CALLED_WORKFLOWS: usize = 50;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API372
373/// What the jobs of one key came to, for `needs.<key>`.
374fn key_result(rows: &[&JobRow]) -> &'static str {
375 let failed = |row: &&&JobRow| row.conclusion.as_deref() == Some("failure") && row.continue_on_error == 0;
376 if rows.iter().any(|row| failed(&row)) {
377 "failure"
378 } else if rows.iter().any(|row| row.conclusion.as_deref() == Some("cancelled")) {
379 "cancelled"
380 } else if rows.iter().all(|row| row.conclusion.as_deref() == Some("skipped")) {
381 "skipped"
382 } else {
383 "success"
384 }
385}
386
Fast pages, required checks on the branch, self-hosted runners, honest incidents387/// Whether any job before `key` failed: one it needs, or one those need,
388/// however far back. A job after a skipped one still sees the failure
389/// that skipped it, as GitHub's failure() does. `needs_of`: each key's
390/// needs, as keys; `failed`: whether a key's jobs came to a failure.
391fn ancestor_failed(needs_of: &std::collections::HashMap<&str, Vec<&str>>, key: &str, failed: impl Fn(&str) -> bool) -> bool {
392 let mut seen = std::collections::HashSet::new();
393 let mut stack: Vec<&str> = needs_of.get(key).cloned().unwrap_or_default();
394 while let Some(next) = stack.pop() {
395 if !seen.insert(next) {
396 continue;
397 }
398 if failed(next) {
399 return true;
400 }
401 stack.extend(needs_of.get(next).into_iter().flatten().copied());
402 }
403 false
404}
405
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97406/// A job's `environment:`, as deployments read it: its name, the address in
407/// `environment.url` once its expressions are filled in from the run, and
408/// whether the job deploys to it. A job with `deployment: false` only reads
409/// the environment's secrets and variables, and makes no deployment.
410#[derive(Clone, Debug, Default, PartialEq)]
411pub(crate) struct JobEnvironment {
412 pub(crate) name: String,
413 pub(crate) url: Option<String>,
414 pub(crate) deploys: bool,
415}
416
417/// The environment `environment:` names, as written in `raw` (a job), with
418/// `contexts` to fill in expressions: null when it names none, or only
419/// through an expression this cannot read before the job runs.
420pub(crate) fn environment_of(raw: &Value, contexts: &Map<String, Value>) -> Option<JobEnvironment> {
421 let scope = Scope { contexts, status: Status::Success, hash_files: None };
422 let plain = |value: &Value| -> Option<String> {
423 let text = match value {
424 Value::String(text) if text.contains("${{") => expr::interpolate_value(value, &scope).ok().map(|v| expr::to_text(&v))?,
425 Value::String(text) => text.clone(),
426 _ => return None,
427 };
428 let text = text.trim().to_owned();
429 (!text.is_empty()).then_some(text)
430 };
431 match raw.get("environment")? {
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9432 name @ Value::String(_) => Some(JobEnvironment { name: plain(name)?, url: None, deploys: workflow::records_deployment(raw) }),
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97433 Value::Object(env) => {
434 let name = plain(env.get("name")?)?;
435 let url = env
436 .get("url")
437 .and_then(plain)
438 .filter(|url| url.starts_with("https://") || url.starts_with("http://"));
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9439 Some(JobEnvironment { name, url, deploys: workflow::records_deployment(raw) })
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97440 }
441 _ => None,
442 }
443}
444
445/// The contexts a job's `runs-on` and `environment` are read with before it
446/// runs: the run's `github`, its inputs, and the job's matrix.
447fn start_contexts(run: &RunRow, job: &JobRow, key: &str) -> Map<String, Value> {
448 let mut contexts = Map::new();
449 contexts.insert("github".into(), run.info().context(key, "", run.action.as_deref()));
450 contexts.insert("inputs".into(), Value::Object(run.inputs()));
451 contexts.insert("matrix".into(), job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or_else(|| json!({})));
452 contexts
453}
454
455/// The job as its workflow (or the workflow it calls) defines it.
456fn job_spec(run: &RunRow, job: &JobRow) -> Option<workflow::Job> {
457 match job.callee() {
458 Some((_, spec, _)) => Some(spec),
459 None => workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)),
460 }
461}
462
Merge branch 'worktree-agent-a3abfcce648e87dca'463/// The environment a job deploys to, if it deploys: by the name read when
464/// its needs were done (which may have needed their outputs), else as
465/// its `environment:` reads now.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97466pub(crate) fn deploys_to(run: &RunRow, job: &JobRow) -> Option<JobEnvironment> {
467 let spec = job_spec(run, job)?;
Merge branch 'worktree-agent-a3abfcce648e87dca'468 let read = environment_of(&spec.raw, &start_contexts(run, job, &spec.id));
469 let env = match (read, &job.environment) {
470 (Some(env), Some(name)) => Some(JobEnvironment { name: name.clone(), ..env }),
471 (Some(env), None) => Some(env),
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9472 (None, Some(name)) => Some(JobEnvironment { name: name.clone(), url: None, deploys: workflow::records_deployment(&spec.raw) }),
Merge branch 'worktree-agent-a3abfcce648e87dca'473 (None, None) => None,
474 };
475 env.filter(|env| env.deploys)
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97476}
477
Fast pages, required checks on the branch, self-hosted runners, honest incidents478/// What the runner is told about a job it starts, from its workflow: the
479/// environment it names plainly, and the machine its `runs-on` asks for
480/// (`instance_for`; none for the standard one).
481#[derive(Default)]
482struct StartDetails {
483 environment: Option<String>,
484 instance: Option<String>,
485}
486
487fn start_details(run: &RunRow, job: &JobRow) -> StartDetails {
488 let spec = match job.callee() {
489 Some((_, spec, _)) => spec,
490 None => match workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)) {
491 Some(spec) => spec,
492 None => return StartDetails::default(),
493 },
494 };
Merge branch 'worktree-agent-a3abfcce648e87dca'495 // As read when its needs were done, an expression's included.
496 let environment = job.environment.clone();
Fast pages, required checks on the branch, self-hosted runners, honest incidents497 // `runs-on` as the job was queued with: its matrix and the run's
498 // inputs. A label that needs more than those is the standard machine.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97499 let contexts = start_contexts(run, job, &spec.id);
Fast pages, required checks on the branch, self-hosted runners, honest incidents500 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
501 let labels: Vec<String> = match expr::interpolate_value(&spec.runs_on, &scope).unwrap_or(Value::Null) {
502 Value::String(label) => vec![label],
503 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
504 Value::Object(given) => match given.get("labels") {
505 Some(Value::Array(labels)) => labels.iter().map(expr::to_text).collect(),
506 Some(label) => vec![expr::to_text(label)],
507 None => Vec::new(),
508 },
509 _ => Vec::new(),
510 };
511 let instance = g1t_contracts::actions::instance_for(&labels);
512 StartDetails {
513 environment,
514 instance: (instance != g1t_contracts::actions::STANDARD_INSTANCE).then(|| instance.label.to_owned()),
515 }
516}
517
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API518fn now() -> String {
519 rfc3339(now_ms())
520}
521
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97522/// How a finished run went for one environment, from the conclusions of
523/// the jobs that deploy to it: failed if any failed, an error if any was
524/// cancelled, else a success; nothing when none ran (all skipped).
525pub(crate) fn deployment_outcome(conclusions: &[Option<String>]) -> Option<&'static str> {
526 let ran: Vec<&str> = conclusions.iter().flatten().map(String::as_str).filter(|c| *c != "skipped").collect();
527 if ran.is_empty() {
528 return None;
529 }
530 if ran.iter().any(|c| *c == "failure" || *c == "timed_out") {
531 return Some("failure");
532 }
533 if ran.contains(&"cancelled") {
534 return Some("error");
535 }
536 Some("success")
537}
538
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API539impl Actions {
540 pub async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {
541 self.db.prepare("SELECT * FROM runs WHERE id = ?").bind(&[id.into()])?.first::<RunRow>(None).await
542 }
543
544 pub async fn job_rows(&self, run_id: &str) -> Result<Vec<JobRow>> {
545 self.db
546 .prepare("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid")
547 .bind(&[run_id.into()])?
548 .all()
549 .await?
550 .results::<JobRow>()
551 }
552
553 pub async fn run_summary(&self, id: &str) -> Result<Outcome<WorkflowRun>> {
554 Ok(match self.run_row(id).await? {
555 Some(row) => Outcome::Ok(row.summary()),
556 None => fail(FailureCode::NotFound, "No such run."),
557 })
558 }
559
560 /// The contexts every expression outside a job's steps may use.
561 fn base_contexts(run: &RunRow, vars: &Map<String, Value>, job: &str) -> Map<String, Value> {
562 let mut contexts = Map::new();
563 contexts.insert("github".into(), run.info().context(job, "", run.action.as_deref()));
564 contexts.insert("inputs".into(), Value::Object(run.inputs()));
565 contexts.insert("vars".into(), Value::Object(vars.clone()));
566 contexts.insert("needs".into(), json!({}));
567 contexts.insert("runner".into(), runner_context());
568 contexts
569 }
570
571 /// Makes a run and its jobs, and starts what can start. `None` when the
572 /// event already started this workflow.
573 pub async fn create_run(&self, new: NewRun) -> Result<Option<String>> {
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look574 // Nothing starts on an archived repository. A deleted one is never
575 // found to start anything on.
576 if new.repo.archived() {
577 return Ok(None);
578 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API579 let workflow_row = self.workflow_row(&new.repo, &new.path, &new.workflow.display_name(&new.path), &new.source).await?;
580 let numbered = self
581 .db
582 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
583 .bind(&[workflow_row.id.as_str().into()])?
584 .first::<Count>(None)
585 .await?
586 .map_or(1, |count| count.n);
587 let id = new_id("run", now_ms());
588 let mut info = new.info.clone();
589 info.workflow = new.workflow.display_name(&new.path);
590 info.workflow_path = new.path.clone();
591 info.run_id = id.clone();
592 info.run_number = u64::from(numbered);
593 let vars = self
594 .variables_for(&new.repo.id, &format!("{}/{}", new.repo.namespace, new.repo.name), None, new.trusted)
595 .await?;
596
597 // run-name and the concurrency group read github, inputs and vars.
598 let mut contexts = Map::new();
599 contexts.insert("github".into(), info.context("", "", new.action.as_deref()));
600 contexts.insert("inputs".into(), Value::Object(new.inputs.clone()));
601 contexts.insert("vars".into(), Value::Object(vars));
602 let scope = Scope {
603 contexts: &contexts,
604 status: Status::Success,
605 hash_files: None,
606 };
607 let title = new
608 .workflow
609 .run_name
610 .as_deref()
611 .and_then(|run_name| expr::interpolate(run_name, &scope).ok())
612 .filter(|title| !title.trim().is_empty())
613 .unwrap_or(new.title.clone());
614 let group = new.workflow.concurrency.as_ref().and_then(|c| expr::interpolate(&c.group, &scope).ok());
615 let cancel_in_progress = new
616 .workflow
617 .concurrency
618 .as_ref()
619 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
620 .is_some_and(|value| expr::truthy(&value));
621
Merge branch 'worktree-agent-a3abfcce648e87dca'622 let approval = new.approval.as_ref().map(|reason| RunApproval { state: "required".into(), reason: reason.clone(), approved_by: None });
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API623 let inserted = self
624 .db
625 .prepare(
626 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, action, git_ref, sha,
Merge branch 'worktree-agent-a3abfcce648e87dca'627 pull, status, actor, actor_id, source, info, inputs, trusted, concurrency_group, event_key, created_at,
628 approval, cancel_in_progress)
629 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API630 )
631 .bind(&[
632 id.as_str().into(),
633 workflow_row.id.as_str().into(),
634 new.repo.id.as_str().into(),
635 format!("{}/{}", new.repo.namespace, new.repo.name).into(),
636 new.path.as_str().into(),
637 info.workflow.as_str().into(),
638 title.as_str().into(),
639 numbered.into(),
640 info.event_name.as_str().into(),
641 optional(new.action.as_deref()),
642 info.git_ref.as_str().into(),
643 info.sha.as_str().into(),
644 new.pull.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
Merge branch 'worktree-agent-a3abfcce648e87dca'645 if approval.is_some() { "action_required" } else { "queued" }.into(),
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API646 optional(new.actor.as_deref()),
647 optional(new.actor_id.as_deref()),
648 new.source.as_str().into(),
649 serde_json::to_string(&info)?.into(),
650 serde_json::to_string(&new.inputs)?.into(),
651 u32::from(new.trusted).into(),
652 optional(group.as_deref()),
653 new.event_key.as_str().into(),
654 now().into(),
Merge branch 'worktree-agent-a3abfcce648e87dca'655 approval.as_ref().map(serde_json::to_string).transpose()?.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
656 u32::from(cancel_in_progress).into(),
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API657 ])?
658 .first::<Value>(None)
659 .await?;
660 if inserted.is_none() {
661 return Ok(None);
662 }
663 if let Some(run) = self.run_row(&id).await? {
664 self.report_pending(&run).await?;
665 }
666
667 // Every job, waiting; each is expanded when the jobs it needs are done.
668 let mut statements = Vec::new();
669 for job in &new.workflow.jobs {
670 statements.push(
671 self.db
672 .prepare(
673 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting')",
674 )
675 .bind(&[
676 new_id("job", now_ms()).into(),
677 id.as_str().into(),
678 new.repo.id.as_str().into(),
679 new.repo.namespace.as_str().into(),
680 job.id.as_str().into(),
681 job.name.clone().filter(|n| !expr::has_expression(n)).unwrap_or(job.id.clone()).into(),
682 serde_json::to_string(&job.needs)?.into(),
683 ])?,
684 );
685 }
686 self.db.batch(statements).await?;
687
Merge branch 'worktree-agent-a3abfcce648e87dca'688 // A pull request's run from outside waits to be approved; it joins
689 // its concurrency group once it is (protection.rs, approve_run).
690 if approval.is_some() {
691 return Ok(Some(id));
692 }
693 if let Some(run) = self.run_row(&id).await? {
694 self.enter_group(&run).await?;
695 }
696 Ok(Some(id))
697 }
698
699 /// A run about to start: one run at a time per concurrency group, as on
700 /// GitHub. A newer run replaces one that waits in the group, and with
701 /// `cancel-in-progress` the one that runs; otherwise it waits as
702 /// `pending` for the one that runs. Then it moves along.
703 pub(crate) async fn enter_group(&self, run: &RunRow) -> Result<()> {
704 if let Some(group) = &run.concurrency_group {
705 let cancel_in_progress = run.cancel_in_progress != 0;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API706 let others = self
707 .db
Merge branch 'worktree-agent-a3abfcce648e87dca'708 .prepare(
709 "SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND id != ? AND status NOT IN ('completed', 'action_required') ORDER BY id",
710 )
711 .bind(&[run.repo_id.as_str().into(), group.as_str().into(), run.id.as_str().into()])?
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API712 .all()
713 .await?
714 .results::<RunRow>()?;
715 for other in &others {
716 if cancel_in_progress || other.status == "pending" {
717 // A newer run replaces a waiting one, as on GitHub.
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)718 self.cancel_run(other, "A newer run in the same concurrency group replaced it.", false).await?;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API719 }
720 }
721 if !cancel_in_progress && others.iter().any(|other| other.status != "pending") {
722 self.db
723 .prepare("UPDATE runs SET status = 'pending' WHERE id = ?")
Merge branch 'worktree-agent-a3abfcce648e87dca'724 .bind(&[run.id.as_str().into()])?
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API725 .run()
726 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'727 return Ok(());
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API728 }
729 }
Merge branch 'worktree-agent-a3abfcce648e87dca'730 self.advance(&run.id).await
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API731 }
732
733 /// A run that could not start, such as for a workflow file that does not read.
734 #[allow(clippy::too_many_arguments)]
735 pub async fn record_failed_run(
736 &self,
737 row: &WorkflowRow,
738 git_ref: &str,
739 sha: &str,
740 event_key: &str,
741 actor_id: Option<&str>,
742 actor: &str,
743 problem: &str,
744 ) -> Result<()> {
745 let numbered = self
746 .db
747 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
748 .bind(&[row.id.as_str().into()])?
749 .first::<Count>(None)
750 .await?
751 .map_or(1, |count| count.n);
752 let at = now();
753 self.db
754 .prepare(
755 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, git_ref, sha, status,
756 conclusion, error, actor, actor_id, source, info, event_key, created_at, finished_at)
757 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'push', ?, ?, 'completed', 'failure', ?, ?, ?, ?, '{}', ?, ?, ?)",
758 )
759 .bind(&[
760 new_id("run", now_ms()).into(),
761 row.id.as_str().into(),
762 row.repo_id.as_str().into(),
763 row.repo.as_str().into(),
764 row.path.as_str().into(),
765 row.path.as_str().into(),
766 "Invalid workflow file".into(),
767 numbered.into(),
768 git_ref.into(),
769 sha.into(),
770 problem.into(),
771 actor.into(),
772 optional(actor_id),
773 row.source.as_str().into(),
774 event_key.into(),
775 at.as_str().into(),
776 at.as_str().into(),
777 ])?
778 .run()
779 .await?;
780 Ok(())
781 }
782
783 /// Moves a run along: jobs whose needs are done are decided on, and
784 /// jobs that can start are started.
785 pub async fn advance(&self, run_id: &str) -> Result<()> {
786 // Each pass may finish jobs (skipped ones), which may free others.
787 for _ in 0..20 {
788 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'789 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API790 return Ok(());
791 }
792 let workflow = match workflow::parse(&run.source) {
793 Ok(workflow) => workflow,
794 Err(problem) => {
795 self.finish_run(&run, Some(&problem)).await?;
796 return Ok(());
797 }
798 };
799 let jobs = self.job_rows(run_id).await?;
800 let mut changed = false;
801 // What to decide on: the workflow's jobs, and the jobs of the
802 // workflows they call, each with its needs as (name, key).
803 let mut units: Vec<Unit> = workflow
804 .jobs
805 .iter()
806 .map(|job| (job.id.clone(), job.clone(), job.needs.iter().map(|n| (n.clone(), n.clone())).collect()))
807 .collect();
808 let mut seen = std::collections::HashSet::new();
809 for row in &jobs {
810 if !seen.insert(row.key.clone()) {
811 continue;
812 }
813 if let Some((_, job, call)) = row.callee() {
814 let parent = call["parent"].as_str().unwrap_or_default().to_owned();
815 let needs = job.needs.iter().map(|n| (n.clone(), format!("{parent}/{n}"))).collect();
816 units.push((row.key.clone(), job, needs));
817 }
818 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents819 // Each key's needs, as keys, to look back through for failure().
820 let needs_of: std::collections::HashMap<&str, Vec<&str>> =
821 units.iter().map(|(key, _, needs)| (key.as_str(), needs.iter().map(|(_, need)| need.as_str()).collect())).collect();
822 let key_failed = |key: &str| key_result(&jobs.iter().filter(|row| row.key == key).collect::<Vec<_>>()) == "failure";
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API823 for (key, job, needs) in &units {
824 let rows: Vec<&JobRow> = jobs.iter().filter(|row| &row.key == key).collect();
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9825 if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting" && !row.holds_call()) {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API826 continue;
827 }
828 let needed: Vec<(&String, Vec<&JobRow>)> =
829 needs.iter().map(|(name, need)| (name, jobs.iter().filter(|row| &row.key == need).collect())).collect();
830 if !needed.iter().all(|(_, rows)| rows.iter().all(|row| row.status == "completed")) {
831 continue;
832 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents833 let failed_before = ancestor_failed(&needs_of, key, key_failed);
834 self.decide(&run, job, rows[0], &needed, failed_before).await?;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API835 changed = true;
836 }
837 // A job that called a workflow finishes with that workflow's jobs.
838 for row in jobs.iter().filter(|row| row.status == "calling") {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9839 let prefix = row.call_prefix().unwrap_or(row.key.clone());
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API840 let children: Vec<&JobRow> = jobs
841 .iter()
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9842 .filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(prefix.as_str())))
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API843 .collect();
844 if !children.is_empty() && children.iter().all(|child| child.status == "completed") {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9845 self.finish_call(&run, row, &children).await?;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API846 changed = true;
847 }
848 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9849 // Calls that wait start while their matrix's `max-parallel` and
850 // their concurrency group have room.
851 let mut calling: std::collections::HashMap<&str, u32> = std::collections::HashMap::new();
852 for row in jobs.iter().filter(|row| row.status == "calling") {
853 *calling.entry(row.key.as_str()).or_default() += 1;
854 }
855 for row in jobs.iter().filter(|row| row.holds_call()) {
856 let running = calling.get(row.key.as_str()).copied().unwrap_or(0);
857 if row.max_parallel.is_some_and(|max| running >= max) || self.group_busy(row).await? {
858 continue;
859 }
860 if self.start_call(row).await? {
861 calling.insert(row.key.as_str(), running + 1);
862 changed = true;
863 }
864 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API865 if !changed {
866 break;
867 }
868 }
869 self.start_queued().await?;
870 self.finish_if_done(run_id).await
871 }
872
873 /// Decides on one job whose needs are done: skip it, fail it, or expand
Fast pages, required checks on the branch, self-hosted runners, honest incidents874 /// it into its matrix and queue it. `failed_before`: whether any job
875 /// before it failed, however far back (`ancestor_failed`).
876 async fn decide(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, needed: &[(&String, Vec<&JobRow>)], failed_before: bool) -> Result<()> {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API877 let vars = self.variables_for(&run.repo_id, &run.repo, None, run.trusted != 0).await?;
878 let mut contexts = Self::base_contexts(run, &vars, &job.id);
879 // A called workflow's jobs read the inputs they were called with.
880 let call = row.call();
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9881 // Shown under the job whose call made it, by that job's name.
882 let caller = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| {
883 c["title"].as_str().map(str::to_owned).or_else(|| c["parent"].as_str().map(|parent| parent.replace('/', " / ")))
884 });
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API885 if let Some(call) = call.as_ref().filter(|c| c["role"] == "callee") {
886 contexts.insert("inputs".into(), call["inputs"].clone());
887 }
888 let mut needs = Map::new();
Fast pages, required checks on the branch, self-hosted runners, honest incidents889 let mut results = Vec::new();
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API890 for (key, rows) in needed {
891 let result = key_result(rows);
892 let mut outputs = Map::new();
893 for row in rows {
894 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
895 outputs.extend(more);
896 }
897 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents898 results.push(result);
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API899 needs.insert((*key).clone(), json!({ "result": result, "outputs": outputs }));
900 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents901 // As on GitHub: a need that was skipped makes success() false but
902 // not failure(); failure() is a failure anywhere before the job.
903 let status = expr::job_status(results, failed_before, run.conclusion.as_deref() == Some("cancelled"));
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API904 contexts.insert("needs".into(), Value::Object(needs));
905 let scope = Scope {
906 contexts: &contexts,
907 status,
908 hash_files: None,
909 };
910 let condition = job.condition.as_deref().unwrap_or_default();
911 match expr::condition(condition, &scope) {
912 Ok(true) => {}
913 Ok(false) => return self.skip_job(row, None).await,
914 Err(problem) => return self.fail_job(row, &format!("Its `if` does not read: {problem}")).await,
915 }
916 // Its matrix, which may come from a needed job's outputs.
917 let combinations = match &job.matrix {
918 None => vec![Map::new()],
919 Some(matrix) => {
920 let value = match expr::interpolate_value(matrix, &scope) {
921 Ok(value) => value,
922 Err(problem) => return self.fail_job(row, &format!("Its matrix does not read: {problem}")).await,
923 };
924 match matrix::expand(&value) {
925 Ok(combinations) if !combinations.is_empty() => combinations,
926 Ok(_) => return self.fail_job(row, "Its matrix makes no jobs.").await,
927 Err(problem) => return self.fail_job(row, &problem).await,
928 }
929 }
930 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9931 // A job that calls a reusable workflow calls it once for each.
932 if let Some(uses) = &job.uses {
933 return self.call_workflow(run, job, row, uses, &contexts, &combinations, caller.as_deref()).await;
934 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API935 let total = combinations.len();
936 let raw = job.raw.as_object().cloned().unwrap_or_default();
Fast pages, required checks on the branch, self-hosted runners, honest incidents937 // Whether pull requests from forks may use self-hosted runners here,
938 // asked once, and only for a run that is not trusted.
939 let mut forks_allowed: Option<bool> = None;
Merge branch 'worktree-agent-a3abfcce648e87dca'940 // Each concurrency group its jobs join, and whether it cancels.
941 let mut groups: Vec<(String, bool)> = Vec::new();
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API942 let mut statements = Vec::new();
943 for (index, combination) in combinations.iter().enumerate() {
944 let mut contexts = contexts.clone();
945 contexts.insert("matrix".into(), Value::Object(combination.clone()));
946 contexts.insert(
947 "strategy".into(),
948 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
949 );
950 let scope = Scope {
951 contexts: &contexts,
952 status: Status::Success,
953 hash_files: None,
954 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9955 let name = shown_name(job, caller.as_deref(), combination, &scope);
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API956 let runs_on = expr::interpolate_value(&job.runs_on, &scope).unwrap_or(Value::Null);
957 let labels = match &runs_on {
958 Value::String(label) => vec![label.clone()],
959 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
960 Value::Object(spec) => spec.get("labels").map(|l| match l {
961 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
962 other => vec![expr::to_text(other)],
963 }).unwrap_or_default(),
964 _ => Vec::new(),
965 };
Fast pages, required checks on the branch, self-hosted runners, honest incidents966 // `runs-on: self-hosted` (or a group): the workspace's own
967 // machines, which may run any OS. Otherwise g1t's Linux sandboxes.
968 let group = match &runs_on {
969 Value::Object(spec) => spec.get("group").map(expr::to_text),
970 _ => None,
971 };
972 let wanted = Wanted::of(&labels, group.as_deref());
973 let mut reason = if wanted.self_hosted {
974 None
975 } else {
976 labels
977 .iter()
978 .find(|label| {
979 let lower = label.to_ascii_lowercase();
980 lower.contains("windows") || lower.contains("macos")
981 })
982 .map(|label| {
983 let os = if label.to_ascii_lowercase().contains("windows") { "windows" } else { "macos" };
984 format!("`runs-on: {label}`: g1t's own runners are Linux. A self-hosted runner can run it: `runs-on: [self-hosted, {os}]`.")
985 })
986 };
987 // A pull request from a fork runs code anyone could write: never
988 // on the workspace's machines unless it said they may.
989 if wanted.self_hosted && run.trusted == 0 {
990 let allowed = match forks_allowed {
991 Some(allowed) => allowed,
992 None => {
993 let allowed = self.effective_runner_settings(&row.namespace, Some(&run.repo_id)).await?.fork_pull_requests;
994 forks_allowed = Some(allowed);
995 allowed
996 }
997 };
998 if !allowed {
999 reason = Some(
1000 "Pull requests from forks do not run on self-hosted runners here. An admin can allow it under Settings, Runners.".to_owned(),
1001 );
1002 }
1003 }
1004 let max_minutes = if wanted.self_hosted { SELF_HOSTED_MAX_TIMEOUT_MINUTES } else { MAX_TIMEOUT_MINUTES };
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1005 let timeout = raw
1006 .get("timeout-minutes")
1007 .and_then(|value| expr::interpolate_value(value, &scope).ok())
1008 .and_then(|value| value.as_f64().or_else(|| expr::to_text(&value).parse().ok()))
What runs differently on g1t's machines is said before it surprises anyone, and four things now behave as on GitHub. A commit whose message says `[skip ci]`, `[ci skip]`, `[no ci]`, `[skip actions]` or `[actions skip]`, or ends with `skip-checks: true`, starts no `push` or `pull_request` workflows. A `pull_request_review` event's `review.state` is `approved` or `changes_requested`, not `approve` or `requestchanges`. A self-hosted job that names no `timeout-minutes` runs for 360 minutes, not 60. A JavaScript action's `::debug::` lines show on a debug run, as `run:` steps' do. The workflow's page now warns when a job asks for more than the 60 minutes g1t's machines give it, when an Arm label will run on x64, and when a matrix on a job that calls a reusable workflow will run it once; it no longer says artifacts are kept up to 60 MB (they are kept up to 5 GiB), nor that `setup-node` with `cache:` runs without its cache (the toolkit's cache works). The desktop release's two hosted jobs ask for the 60 minutes they get. The Actions and self-hosted runners guides say so.1009 .map_or(g1t_actions::workflow::DEFAULT_TIMEOUT_MINUTES.min(max_minutes), |minutes| (minutes.ceil() as u32).clamp(1, max_minutes));
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1010 let continue_on_error = raw
1011 .get("continue-on-error")
1012 .and_then(|value| expr::interpolate_value(value, &scope).ok())
1013 .is_some_and(|value| expr::truthy(&value));
Merge branch 'worktree-agent-a3abfcce648e87dca'1014 let (mut status, mut conclusion, mut finished) = match &reason {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1015 Some(_) => ("completed", Some("failure"), Some(now())),
1016 None => ("queued", None, None),
1017 };
Fast pages, required checks on the branch, self-hosted runners, honest incidents1018 // A self-hosted job waits, saying for what, until a runner takes it.
1019 let (labels_json, queued_at) = if wanted.self_hosted && reason.is_none() {
1020 reason = Some(waiting_reason(&wanted));
1021 (Some(serde_json::to_string(&wanted.stored())?), Some(now()))
1022 } else {
1023 (None, None)
1024 };
Merge branch 'worktree-agent-a3abfcce648e87dca'1025 // The environment it names, an expression read now, and what
1026 // that environment's protection rules say: it starts, waits as
1027 // `pending` (protection.rs), or may not deploy there.
1028 let environment = environment_of(&job.raw, &contexts).map(|env| env.name);
1029 if status == "queued"
1030 && let Some(name) = &environment
1031 {
1032 match self.gate(run, name).await? {
1033 Gate::Open => {}
1034 Gate::Held(why) => {
1035 status = "pending";
1036 reason = Some(why);
1037 }
1038 Gate::Refused(why) => {
1039 (status, conclusion, finished) = ("completed", Some("failure"), Some(now()));
1040 reason = Some(why);
1041 }
1042 }
1043 }
1044 // Its own concurrency group.
1045 let job_group = job
1046 .concurrency
1047 .as_ref()
1048 .and_then(|c| expr::interpolate(&c.group, &scope).ok())
1049 .map(|group| group.trim().to_owned())
1050 .filter(|group| !group.is_empty());
1051 let job_cancel = job
1052 .concurrency
1053 .as_ref()
1054 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
1055 .is_some_and(|value| expr::truthy(&value));
1056 if status != "completed"
1057 && let Some(group) = &job_group
1058 && !groups.iter().any(|(known, _)| known == group)
1059 {
1060 groups.push((group.clone(), job_cancel));
1061 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1062 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
1063 name.into(),
1064 serde_json::to_string(combination)?.into(),
1065 status.into(),
1066 optional(conclusion),
1067 optional(reason.as_deref()),
1068 timeout.into(),
1069 u32::from(continue_on_error).into(),
1070 job.max_parallel.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1071 optional(finished.as_deref()),
Fast pages, required checks on the branch, self-hosted runners, honest incidents1072 optional(labels_json.as_deref()),
1073 optional(queued_at.as_deref()),
Merge branch 'worktree-agent-a3abfcce648e87dca'1074 optional(environment.as_deref()),
1075 optional(job_group.as_deref()),
1076 u32::from(job_cancel).into(),
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1077 ];
1078 if index == 0 {
1079 let mut bound = values;
1080 bound.push(row.id.as_str().into());
1081 statements.push(
1082 self.db
1083 .prepare(
1084 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, timeout_minutes = ?,
Merge branch 'worktree-agent-a3abfcce648e87dca'1085 continue_on_error = ?, max_parallel = ?, finished_at = ?, labels = ?, queued_at = ?, environment = ?,
1086 concurrency_group = ?, cancel_in_progress = ? WHERE id = ?",
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1087 )
1088 .bind(&bound)?,
1089 );
1090 } else {
1091 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
1092 new_id("job", now_ms()).into(),
1093 row.run_id.as_str().into(),
1094 row.repo_id.as_str().into(),
1095 row.namespace.as_str().into(),
1096 row.key.as_str().into(),
1097 (index as u32).into(),
1098 row.needs.as_str().into(),
1099 optional(row.call.as_deref()),
1100 ];
1101 bound.extend(values);
1102 statements.push(
1103 self.db
1104 .prepare(
1105 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, call, name, matrix, status, conclusion, reason,
Merge branch 'worktree-agent-a3abfcce648e87dca'1106 timeout_minutes, continue_on_error, max_parallel, finished_at, labels, queued_at, environment,
1107 concurrency_group, cancel_in_progress)
1108 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1109 )
1110 .bind(&bound)?,
1111 );
1112 }
1113 }
1114 self.db.batch(statements).await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1115 self.replace_in_groups(run, &row.key, &groups).await
1116 }
1117
1118 /// A job queued in its own concurrency group: with `cancel-in-progress`
1119 /// it cancels the group's other jobs; otherwise it replaces any that
1120 /// are queued and not started, and waits for the one running
1121 /// (`start_queued` starts one of a group at a time). As on GitHub.
1122 async fn replace_in_groups(&self, run: &RunRow, key: &str, groups: &[(String, bool)]) -> Result<()> {
1123 if groups.is_empty() {
1124 return Ok(());
1125 }
1126 #[derive(Deserialize)]
1127 struct Id {
1128 id: String,
1129 }
1130 let mine: Vec<String> = self
1131 .db
1132 .prepare("SELECT id FROM jobs WHERE run_id = ? AND key = ?")
1133 .bind(&[run.id.as_str().into(), key.into()])?
1134 .all()
1135 .await?
1136 .results::<Id>()?
1137 .into_iter()
1138 .map(|row| row.id)
1139 .collect();
1140 let mut moved: Vec<String> = Vec::new();
1141 for (group, cancel) in groups {
1142 let others = self
1143 .db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91144 // A call that waits in the group, or calls, is in it too.
1145 .prepare("SELECT * FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('waiting', 'queued', 'pending', 'in_progress', 'calling')")
Merge branch 'worktree-agent-a3abfcce648e87dca'1146 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1147 .all()
1148 .await?
1149 .results::<JobRow>()?;
1150 for other in others.iter().filter(|other| !mine.contains(&other.id)) {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91151 if *cancel || matches!(other.status.as_str(), "queued" | "waiting") {
Merge branch 'worktree-agent-a3abfcce648e87dca'1152 self.stop_job(other, "A newer job in the same concurrency group replaced it.").await?;
1153 if !moved.contains(&other.run_id) {
1154 moved.push(other.run_id.clone());
1155 }
1156 }
1157 }
1158 }
1159 for other_run in moved.iter().filter(|id| **id != run.id) {
1160 Box::pin(self.advance(other_run)).await?;
1161 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1162 Ok(())
1163 }
1164
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1165 /// A job that calls a reusable workflow, in the repository or another
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91166 /// (reach.rs): once for each combination of its matrix, as on GitHub,
1167 /// each with the inputs and secrets it passes read with that
1168 /// combination's `matrix`. Each call waits (`holds_call`) until its
1169 /// matrix's `max-parallel` and its concurrency group have room, when
1170 /// `start_call` puts the workflow's jobs in the run under it.
1171 #[allow(clippy::too_many_arguments)]
1172 async fn call_workflow(
1173 &self,
1174 run: &RunRow,
1175 job: &workflow::Job,
1176 row: &JobRow,
1177 uses: &str,
1178 contexts: &Map<String, Value>,
1179 combinations: &[Map<String, Value>],
1180 caller: Option<&str>,
1181 ) -> Result<()> {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1182 let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1;
1183 if depth > MAX_CALL_DEPTH {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91184 return self.fail_job(row, &call_allowed(depth, &[], uses).unwrap_err()).await;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1185 }
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1186 let (file, source, origin) = match self.called_workflow(run, row, uses).await? {
1187 Ok(found) => found,
1188 Err(why) => return self.fail_job(row, &why).await,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1189 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91190 let jobs = self.job_rows(&run.id).await?;
1191 // Each call so far, by the workflow it called.
1192 let called_paths: Vec<String> = jobs
1193 .iter()
1194 .filter(|other| other.call_prefix().is_some())
1195 .filter_map(JobRow::call)
1196 .filter_map(|call| call["path"].as_str().map(str::to_owned))
1197 .collect();
1198 let called_already: Vec<&str> = called_paths.iter().map(String::as_str).collect();
1199 if let Err(why) = call_allowed(depth, &called_already, &file) {
1200 return self.fail_job(row, &why).await;
1201 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1202 let called = match workflow::parse(&source) {
1203 Ok(called) => called,
1204 Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await,
1205 };
1206 let Some(trigger) = called.trigger("workflow_call") else {
1207 return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await;
1208 };
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1209 // Secrets: none but the job's token unless `secrets:` passes them,
1210 // by name or with `inherit`; read when each job starts (job_spec).
1211 let outer = row.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91212 let total = combinations.len();
1213 let mut groups: Vec<(String, bool)> = Vec::new();
1214 let mut statements = Vec::new();
1215 for (index, combination) in combinations.iter().enumerate() {
1216 let mut contexts = contexts.clone();
1217 contexts.insert("matrix".into(), Value::Object(combination.clone()));
1218 contexts.insert(
1219 "strategy".into(),
1220 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
1221 );
1222 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1223 let name = shown_name(job, caller, combination, &scope);
1224 // Inputs: what `with:` passes, as the called workflow types them.
1225 let given = match job.raw.get("with") {
1226 Some(with) => match expr::interpolate_value(with, &scope) {
1227 Ok(Value::Object(given)) => Ok(given),
1228 Ok(_) => Ok(Map::new()),
1229 Err(problem) => Err(format!("Its `with` does not read: {problem}")),
1230 },
1231 None => Ok(Map::new()),
1232 };
1233 let inputs = given.and_then(|given| crate::reach::call_inputs(&file, &trigger.inputs, &given));
1234 let secrets = crate::reach::secrets_plan(&job.raw, &contexts, outer.clone());
1235 let problem = match &inputs {
1236 Err(why) => Some(why.clone()),
1237 Ok(_) => crate::reach::missing_secrets(&called.raw, &secrets)
1238 .first()
1239 .map(|name| format!("`{file}` needs the secret `{name}`: pass it under `secrets:`, or use `secrets: inherit`."))
1240 .or_else(|| {
1241 crate::reach::undeclared_secrets(&called.raw, &secrets)
1242 .first()
1243 .map(|name| format!("Invalid secret: `{name}` is not defined in `{file}`."))
1244 }),
1245 };
1246 // Its own concurrency group, which it holds while its workflow runs.
1247 let group = job
1248 .concurrency
1249 .as_ref()
1250 .and_then(|c| expr::interpolate(&c.group, &scope).ok())
1251 .map(|group| group.trim().to_owned())
1252 .filter(|group| !group.is_empty());
1253 let cancel = job
1254 .concurrency
1255 .as_ref()
1256 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
1257 .is_some_and(|value| expr::truthy(&value));
1258 if problem.is_none()
1259 && let Some(group) = &group
1260 && !groups.iter().any(|(known, _)| known == group)
1261 {
1262 groups.push((group.clone(), cancel));
1263 }
1264 let call = json!({
1265 "role": "caller", "path": file, "source": source, "origin": origin, "depth": depth,
1266 "prefix": call_prefix(&row.key, job.matrix.is_some().then_some(index)), "title": name,
1267 "inputs": inputs.unwrap_or_default(), "secrets": secrets, "fail_fast": job.fail_fast,
1268 });
1269 let (status, conclusion, finished) = match &problem {
1270 Some(_) => ("completed", Some("failure"), Some(now())),
1271 None => ("waiting", None, None),
1272 };
1273 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
1274 name.as_str().into(),
1275 serde_json::to_string(combination)?.into(),
1276 status.into(),
1277 optional(conclusion),
1278 optional(problem.as_deref()),
1279 job.max_parallel.map(|max| max.max(1)).map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1280 optional(finished.as_deref()),
1281 optional(group.as_deref()),
1282 u32::from(cancel).into(),
1283 serde_json::to_string(&call)?.into(),
1284 ];
1285 if index == 0 {
1286 let mut bound = values;
1287 bound.push(row.id.as_str().into());
1288 statements.push(
1289 self.db
1290 .prepare(
1291 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, max_parallel = ?, finished_at = ?,
1292 concurrency_group = ?, cancel_in_progress = ?, call = ? WHERE id = ?",
1293 )
1294 .bind(&bound)?,
1295 );
1296 } else {
1297 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
1298 new_id("job", now_ms()).into(),
1299 row.run_id.as_str().into(),
1300 row.repo_id.as_str().into(),
1301 row.namespace.as_str().into(),
1302 row.key.as_str().into(),
1303 (index as u32).into(),
1304 row.needs.as_str().into(),
1305 ];
1306 bound.extend(values);
1307 statements.push(
1308 self.db
1309 .prepare(
1310 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, name, matrix, status, conclusion, reason,
1311 max_parallel, finished_at, concurrency_group, cancel_in_progress, call)
1312 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
1313 )
1314 .bind(&bound)?,
1315 );
1316 }
1317 }
1318 self.db.batch(statements).await?;
1319 self.replace_in_groups(run, &row.key, &groups).await
1320 }
1321
1322 /// Whether another job of `row`'s concurrency group runs, or calls a
1323 /// workflow, so that `row` waits for it.
1324 async fn group_busy(&self, row: &JobRow) -> Result<bool> {
1325 let Some(group) = &row.concurrency_group else { return Ok(false) };
1326 let busy = self
1327 .db
1328 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?")
1329 .bind(&[row.repo_id.as_str().into(), group.as_str().into(), row.id.as_str().into()])?
1330 .first::<Count>(None)
1331 .await?
1332 .map_or(0, |count| count.n);
1333 Ok(busy > 0)
1334 }
1335
1336 /// A call that waited (`holds_call`) is made: the called workflow's
1337 /// jobs join the run under it, named under its name and keyed under its
1338 /// prefix. False when it no longer waits.
1339 async fn start_call(&self, row: &JobRow) -> Result<bool> {
1340 let Some(call) = row.call() else { return Ok(false) };
1341 let Some(prefix) = row.call_prefix() else { return Ok(false) };
1342 let Some(called) = call["source"].as_str().and_then(|source| workflow::parse(source).ok()) else { return Ok(false) };
1343 let file = call["path"].as_str().unwrap_or_default();
1344 let started = self
1345 .db
1346 .prepare("UPDATE jobs SET status = 'calling', reason = ?, started_at = ? WHERE id = ? AND status = 'waiting' RETURNING id")
1347 .bind(&[format!("Calls `{file}`.").into(), now().into(), row.id.as_str().into()])?
1348 .first::<Value>(None)
1349 .await?;
1350 if started.is_none() {
1351 return Ok(false);
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1352 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1353 let mut statements = Vec::new();
1354 for called_job in &called.jobs {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91355 let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{prefix}/{n}")).collect();
1356 let callee = json!({
1357 "role": "callee", "parent": prefix, "title": row.name, "job": called_job.id, "path": file,
1358 "source": call["source"], "inputs": call["inputs"], "depth": call["depth"], "origin": call["origin"], "secrets": call["secrets"],
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1359 });
1360 statements.push(
1361 self.db
1362 .prepare("INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status, call) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting', ?)")
1363 .bind(&[
1364 new_id("job", now_ms()).into(),
1365 row.run_id.as_str().into(),
1366 row.repo_id.as_str().into(),
1367 row.namespace.as_str().into(),
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91368 format!("{prefix}/{}", called_job.id).into(),
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1369 format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(),
1370 serde_json::to_string(&needs)?.into(),
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91371 serde_json::to_string(&callee)?.into(),
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1372 ])?,
1373 );
1374 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91375 if !statements.is_empty() {
1376 self.db.batch(statements).await?;
1377 }
1378 Ok(true)
1379 }
1380
1381 /// The runs with a call waiting in `group`, but `run_id`, moved along
1382 /// now that a job of the group has finished.
1383 async fn wake_group(&self, repo_id: &str, group: &str, run_id: &str) -> Result<()> {
1384 #[derive(Deserialize)]
1385 struct RunId {
1386 run_id: String,
1387 }
1388 let waiting = self
1389 .db
1390 .prepare("SELECT DISTINCT run_id FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'waiting' AND run_id != ?")
1391 .bind(&[repo_id.into(), group.into(), run_id.into()])?
1392 .all()
1393 .await?
1394 .results::<RunId>()?;
1395 for other in waiting {
1396 Box::pin(self.advance(&other.run_id)).await?;
1397 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1398 Ok(())
1399 }
1400
1401 /// A job that called a workflow, finished with its jobs: their result,
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91402 /// and the outputs the workflow declares. A failed call of a fail-fast
1403 /// matrix stops the matrix's other calls.
1404 async fn finish_call(&self, run: &RunRow, row: &JobRow, children: &[&JobRow]) -> Result<()> {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1405 let call = row.call().unwrap_or_default();
1406 let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok());
1407 let mut jobs_context = Map::new();
1408 let mut by_key: std::collections::BTreeMap<String, Vec<&JobRow>> = std::collections::BTreeMap::new();
1409 for child in children {
1410 by_key.entry(child.key.clone()).or_default().push(child);
1411 }
1412 for (key, rows) in &by_key {
1413 let mut outputs = Map::new();
1414 for child in rows {
1415 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&child.outputs) {
1416 outputs.extend(more);
1417 }
1418 }
1419 let id = key.rsplit('/').next().unwrap_or(key);
1420 jobs_context.insert(id.to_owned(), json!({ "result": key_result(rows), "outputs": outputs }));
1421 }
1422 let inputs = children.first().and_then(|c| c.call()).map(|c| c["inputs"].clone()).unwrap_or(json!({}));
1423 let mut contexts = Map::new();
1424 contexts.insert("jobs".into(), Value::Object(jobs_context));
1425 contexts.insert("inputs".into(), inputs);
1426 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1427 let mut outputs = Map::new();
1428 if let Some(Value::Object(declared)) = called.as_ref().map(|w| {
1429 let on = w.raw.get("on").or_else(|| w.raw.get("true")).cloned().unwrap_or(Value::Null);
1430 on.get("workflow_call").and_then(|c| c.get("outputs")).cloned().unwrap_or(Value::Null)
1431 }) {
1432 for (name, spec) in declared {
1433 if let Some(value) = spec.get("value") {
1434 let value = expr::interpolate_value(value, &scope).unwrap_or(Value::Null);
1435 outputs.insert(name, Value::String(expr::to_text(&value)));
1436 }
1437 }
1438 }
1439 let conclusion = key_result(children);
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91440 let finished = self
1441 .db
1442 .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling' RETURNING id")
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1443 .bind(&[conclusion.into(), serde_json::to_string(&outputs)?.into(), now().into(), row.id.as_str().into()])?
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91444 .first::<Value>(None)
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1445 .await?;
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91446 if finished.is_none() {
1447 return Ok(());
1448 }
1449 if conclusion == "failure" && row.matrix.as_deref().is_some_and(|m| m != "{}") && call["fail_fast"] != json!(false) {
1450 let siblings = self
1451 .db
1452 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1453 .bind(&[row.run_id.as_str().into(), row.key.as_str().into()])?
1454 .all()
1455 .await?
1456 .results::<JobRow>()?;
1457 for sibling in siblings {
1458 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1459 }
1460 }
1461 if let Some(group) = &row.concurrency_group {
1462 self.wake_group(&run.repo_id, group, &run.id).await?;
1463 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1464 Ok(())
1465 }
1466
1467 /// A file's text at a commit, if it is there.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1468 pub(crate) async fn read_file(&self, path: &RepoPath, ws: &g1t_contracts::User, sha: &str, file: &str) -> Result<Option<String>> {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1469 let blob: Outcome<g1t_contracts::repos::BlobView> = g1t_kit::call(
1470 &self.repos,
1471 "blob",
1472 &g1t_contracts::repos::BlobArgs {
1473 path: path.clone(),
1474 viewer: Some(ws.clone()),
1475 git_ref: sha.to_owned(),
1476 file_path: file.to_owned(),
1477 },
1478 )
1479 .await?;
1480 Ok(match blob {
1481 Outcome::Ok(view) => view.text,
1482 Outcome::Fail(_) => None,
1483 })
1484 }
1485
1486 async fn skip_job(&self, row: &JobRow, reason: Option<&str>) -> Result<()> {
1487 self.db
1488 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'skipped', reason = ?, finished_at = ? WHERE id = ?")
1489 .bind(&[optional(reason), now().into(), row.id.as_str().into()])?
1490 .run()
1491 .await?;
1492 Ok(())
1493 }
1494
1495 async fn fail_job(&self, row: &JobRow, reason: &str) -> Result<()> {
1496 self.db
1497 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'failure', reason = ?, finished_at = ? WHERE id = ? AND status != 'completed'")
1498 .bind(&[reason.into(), now().into(), row.id.as_str().into()])?
1499 .run()
1500 .await?;
1501 Ok(())
1502 }
1503
1504 /// Starts queued jobs, oldest first, while their workspace has room.
1505 pub async fn start_queued(&self) -> Result<()> {
1506 let queued = self
1507 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1508 // Self-hosted jobs are taken by their runners (runners.rs).
1509 .prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NULL ORDER BY rowid LIMIT 50")
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1510 .all()
1511 .await?
1512 .results::<JobRow>()?;
1513 let mut running: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
1514 for job in queued {
1515 let in_workspace = match running.get(&job.namespace) {
1516 Some(n) => *n,
1517 None => {
1518 let n = self
1519 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1520 // Only g1t's own sandboxes count against the workspace's room.
1521 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE status = 'in_progress' AND namespace = ? AND runner_id IS NULL")
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1522 .bind(&[job.namespace.as_str().into()])?
1523 .first::<Count>(None)
1524 .await?
1525 .map_or(0, |count| count.n);
1526 running.insert(job.namespace.clone(), n);
1527 n
1528 }
1529 };
1530 if in_workspace >= RUNNING_PER_WORKSPACE {
1531 continue;
1532 }
1533 if let Some(max) = job.max_parallel {
1534 let siblings = self
1535 .db
1536 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")
1537 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1538 .first::<Count>(None)
1539 .await?
1540 .map_or(0, |count| count.n);
1541 if siblings >= max {
1542 continue;
1543 }
1544 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1545 // One job of a concurrency group runs at a time.
1546 if let Some(group) = &job.concurrency_group {
1547 let running = self
1548 .db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91549 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?")
Merge branch 'worktree-agent-a3abfcce648e87dca'1550 .bind(&[job.repo_id.as_str().into(), group.as_str().into(), job.id.as_str().into()])?
1551 .first::<Count>(None)
1552 .await?
1553 .map_or(0, |count| count.n);
1554 if running > 0 {
1555 continue;
1556 }
1557 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1558 let token = random_hex(24);
1559 let at = now();
1560 let claimed = self
1561 .db
1562 .prepare(
1563 "UPDATE jobs SET status = 'in_progress', token_hash = ?, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
1564 )
1565 .bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
1566 .first::<Value>(None)
1567 .await?;
1568 if claimed.is_none() {
1569 continue;
1570 }
1571 running.insert(job.namespace.clone(), in_workspace + 1);
1572 self.db
1573 .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")
1574 .bind(&[at.as_str().into(), job.run_id.as_str().into()])?
1575 .run()
1576 .await?;
1577 let run = self.run_row(&job.run_id).await?;
1578 let repo: RepoPath = run.as_ref().map(|run| repo_path(&run.repo)).unwrap_or(RepoPath {
1579 namespace: job.namespace.clone(),
1580 name: String::new(),
1581 });
Fast pages, required checks on the branch, self-hosted runners, honest incidents1582 // Its environment and the machine it asked for, for the runner.
1583 let details = run.as_ref().map(|run| start_details(run, &job)).unwrap_or_default();
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1584 let started: Outcome<Value> = g1t_kit::call(
1585 &self.runner,
1586 "start_actions_job",
1587 &StartJobArgs {
1588 job: job.id.clone(),
1589 token,
1590 repo,
1591 timeout_minutes: job.timeout_minutes,
Fast pages, required checks on the branch, self-hosted runners, honest incidents1592 workflow: run.as_ref().map(|run| run.path.clone()),
1593 environment: details.environment,
1594 trusted: run.as_ref().is_some_and(|run| run.trusted != 0),
1595 instance: details.instance,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1596 },
1597 )
1598 .await
1599 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
1600 if let Outcome::Fail(refused) = started {
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1601 // Billing would not pay for a scheduled run's job: the
1602 // schedule waits rather than making runs only to refuse them.
1603 if refused.code == FailureCode::PaymentRequired
1604 && let Some(run) = run.as_ref().filter(|run| run.event == "schedule")
1605 {
1606 self.db
1607 .prepare("UPDATE workflows SET schedule_refused_until = ? WHERE id = ?")
1608 .bind(&[rfc3339(now_ms() + crate::SCHEDULE_REFUSED_MS).into(), run.workflow_id.as_str().into()])?
1609 .run()
1610 .await?;
1611 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1612 Box::pin(self.finish_job(&job.id, "failure", Some(&refused.message), None)).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971613 } else {
1614 self.job_started(&job.id).await?;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1615 }
1616 }
1617 Ok(())
1618 }
1619
1620 /// Finishes a job and moves its run along.
1621 pub async fn finish_job(&self, job_id: &str, conclusion: &str, reason: Option<&str>, outputs: Option<&Map<String, Value>>) -> Result<()> {
1622 let finished = self
1623 .db
1624 .prepare(
1625 "UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),
1626 finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
1627 )
1628 .bind(&[
1629 conclusion.into(),
1630 optional(reason),
1631 outputs.map(|o| serde_json::to_string(o).unwrap_or_default()).as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1632 now().into(),
1633 job_id.into(),
1634 ])?
1635 .first::<JobRow>(None)
1636 .await?;
1637 let Some(job) = finished else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1638 // Its G1T_TOKEN stops working with it.
1639 self.revoke_job_tokens(&job.id).await;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971640 // A job that deploys failed: so did its run's deployment, now.
1641 if conclusion == "failure" && job.continue_on_error == 0 && job.started_at.is_some()
1642 && let Some(run) = self.run_row(&job.run_id).await?
1643 && let Some(env) = deploys_to(&run, &job)
1644 {
1645 self.report_deployment(&run, &env, "failure", false).await;
1646 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1647 // A self-hosted runner's job: the runner is free again, and its time
1648 // is recorded, at nothing.
1649 if job.runner_id.is_some() {
1650 self.released(&job).await?;
1651 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1652 // Steps still marked as going are not going any more.
1653 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
1654 let mut touched = false;
1655 for step in steps.iter_mut() {
1656 if step["status"] != "completed" {
1657 let was_running = step["status"] == "in_progress";
1658 step["status"] = json!("completed");
1659 step["conclusion"] = json!(if was_running { conclusion } else { "skipped" });
1660 touched = true;
1661 }
1662 }
1663 if touched {
1664 self.db
1665 .prepare("UPDATE jobs SET steps = ? WHERE id = ?")
1666 .bind(&[serde_json::to_string(&steps)?.into(), job.id.as_str().into()])?
1667 .run()
1668 .await?;
1669 }
1670 // fail-fast: one failed combination stops the rest of its matrix.
1671 if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") {
1672 let run = self.run_row(&job.run_id).await?;
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91673 // As its workflow, or the workflow it is a job of, says.
1674 let fail_fast = run.as_ref().and_then(|run| job_spec(run, &job)).is_none_or(|j| j.fail_fast);
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1675 if fail_fast {
1676 let siblings = self
1677 .db
1678 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1679 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1680 .all()
1681 .await?
1682 .results::<JobRow>()?;
1683 for sibling in siblings {
1684 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1685 }
1686 }
1687 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91688 // A call waiting in its concurrency group, in another run, may start.
1689 if let Some(group) = &job.concurrency_group {
1690 self.wake_group(&job.repo_id, group, &job.run_id).await?;
1691 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1692 Box::pin(self.advance(&job.run_id)).await
1693 }
1694
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1695 /// Cancels a job. One that is running is told to stop (the answer to
1696 /// its next report), and runs its `if: always()` and `cancelled()`
1697 /// steps and its post steps before it ends `cancelled`; one that has
1698 /// not started is cancelled at once.
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1699 async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91700 // A job calling a workflow stops with that workflow's jobs.
1701 if job.status == "calling"
1702 && let Some(prefix) = job.call_prefix()
1703 {
1704 let under = format!("{prefix}/");
1705 for child in self.job_rows(&job.run_id).await?.iter().filter(|child| child.key.starts_with(&under) && child.status != "completed") {
1706 Box::pin(self.stop_job(child, reason)).await?;
1707 }
1708 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1709 if job.status == "in_progress" && job.started_at.is_some() {
1710 self.db
1711 .prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'")
1712 .bind(&[now().into(), reason.into(), job.id.as_str().into()])?
1713 .run()
1714 .await?;
1715 return Ok(());
1716 }
1717 self.hard_stop(job, reason).await
1718 }
1719
1720 /// Cancels a job outright, stopping its sandbox if it has one.
1721 async fn hard_stop(&self, job: &JobRow, reason: &str) -> Result<()> {
Fast pages, required checks on the branch, self-hosted runners, honest incidents1722 // A self-hosted runner hears it was cancelled on its next poll.
1723 if job.status == "in_progress" && job.runner_id.is_none() {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1724 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
1725 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1726 let stopped = self
1727 .db
1728 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *")
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1729 .bind(&[reason.into(), now().into(), job.id.as_str().into()])?
Fast pages, required checks on the branch, self-hosted runners, honest incidents1730 .first::<JobRow>(None)
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1731 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1732 if stopped.as_ref().is_some_and(|row| row.started_at.is_some()) {
1733 self.revoke_job_tokens(&job.id).await;
1734 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1735 if let Some(stopped) = stopped.filter(|row| row.runner_id.is_some()) {
1736 self.released(&stopped).await?;
1737 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1738 Ok(())
1739 }
1740
Merge branch 'worktree-agent-a3abfcce648e87dca'1741 /// Ends a job's tokens at once. A failure is logged: the token expires
1742 /// on its own soon after the job's time limit.
1743 async fn revoke_job_tokens(&self, job_id: &str) {
1744 let revoked: Result<bool> =
1745 g1t_kit::call(&self.identity, "revoke_job_tokens", &RevokeJobTokensArgs { job_id: job_id.to_owned() }).await;
1746 if let Err(error) = revoked {
1747 worker::console_error!("actions: the tokens of job {job_id} were not revoked: {error}");
1748 }
1749 }
1750
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1751 /// Finishes the run when every job has.
1752 async fn finish_if_done(&self, run_id: &str) -> Result<()> {
1753 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1754 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1755 return Ok(());
1756 }
1757 let jobs = self.job_rows(run_id).await?;
1758 if !jobs.iter().all(|job| job.status == "completed") {
1759 return Ok(());
1760 }
1761 self.finish_run(&run, None).await
1762 }
1763
1764 async fn finish_run(&self, run: &RunRow, error: Option<&str>) -> Result<()> {
1765 let jobs = self.job_rows(&run.id).await?;
1766 let rows: Vec<&JobRow> = jobs.iter().collect();
1767 let conclusion = if error.is_some() {
1768 "failure"
1769 } else if run.conclusion.as_deref() == Some("cancelled") {
1770 "cancelled"
1771 } else if rows.is_empty() {
1772 "skipped"
1773 } else {
1774 key_result(&rows)
1775 };
1776 let done = self
1777 .db
1778 .prepare("UPDATE runs SET status = 'completed', conclusion = ?, error = COALESCE(?, error), finished_at = ? WHERE id = ? AND status != 'completed' RETURNING id")
1779 .bind(&[conclusion.into(), optional(error), now().into(), run.id.as_str().into()])?
1780 .first::<Value>(None)
1781 .await?;
1782 if done.is_none() {
1783 return Ok(());
1784 }
1785 self.report_status(run, conclusion).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971786 self.settle_deployments(run, &jobs, conclusion == "cancelled").await;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1787 let published: Result<()> = g1t_kit::call(
1788 &self.events,
1789 "publish",
1790 &g1t_contracts::events::Publish {
1791 events: vec![g1t_contracts::events::NewEvent {
1792 kind: "workflow.completed",
1793 source: "actions",
1794 repo_id: Some(run.repo_id.clone()),
1795 actor: run.actor_id.clone(),
1796 data: g1t_contracts::events::WorkflowEvent {
1797 run_id: run.id.clone(),
1798 repo_id: run.repo_id.clone(),
1799 workflow: run.name.clone(),
1800 path: run.path.clone(),
1801 number: run.number,
1802 event: run.event.clone(),
1803 conclusion: conclusion.to_owned(),
1804 git_ref: run.git_ref.clone(),
1805 sha: run.sha.clone(),
1806 pull: run.pull,
1807 },
1808 }],
1809 },
1810 )
1811 .await;
1812 if let Err(error) = published {
1813 worker::console_error!("actions: could not publish workflow.completed: {error}");
1814 }
1815 // The next run waiting in its concurrency group.
1816 if let Some(group) = &run.concurrency_group {
1817 let next = self
1818 .db
1819 .prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND status = 'pending' ORDER BY id LIMIT 1")
1820 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1821 .first::<RunRow>(None)
1822 .await?;
1823 if let Some(next) = next {
1824 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[next.id.as_str().into()])?.run().await?;
1825 Box::pin(self.advance(&next.id)).await?;
1826 }
1827 }
1828 Ok(())
1829 }
1830
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971831 /// Tells the deployments service how a run's deployment to `env` stands:
1832 /// made when its first job naming the environment starts, failed when one
1833 /// of them fails, and settled (`last`) when the run finishes. Never fails
1834 /// the run: a deployment that could not be recorded is logged.
1835 async fn report_deployment(&self, run: &RunRow, env: &JobEnvironment, state: &str, last: bool) {
1836 let path = repo_path(&run.repo);
1837 let reported: Result<Value> = g1t_kit::call(
1838 &self.deployments,
1839 "actions_deployment",
1840 &json!({
1841 "repoId": run.repo_id,
1842 "repo": { "namespace": path.namespace, "name": path.name },
1843 "runId": run.id,
1844 "attempt": run.attempt,
1845 "runUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
1846 "environment": env.name,
1847 "url": env.url,
1848 "ref": run.git_ref,
1849 "sha": run.sha,
1850 "state": state,
1851 "final": last,
1852 "creator": run.actor,
1853 "workflow": run.name,
1854 }),
1855 )
1856 .await;
1857 if let Err(error) = reported {
1858 worker::console_error!("actions: deployment to {} not recorded for run {}: {error}", env.name, run.id);
1859 }
1860 }
1861
1862 /// A job that deploys has started: its run's deployment to the
1863 /// environment is under way.
1864 pub(crate) async fn job_started(&self, job_id: &str) -> Result<()> {
1865 let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[job_id.into()])?.first::<JobRow>(None).await? else {
1866 return Ok(());
1867 };
1868 let Some(run) = self.run_row(&job.run_id).await? else { return Ok(()) };
1869 if let Some(env) = deploys_to(&run, &job) {
1870 self.report_deployment(&run, &env, "in_progress", false).await;
1871 }
1872 Ok(())
1873 }
1874
1875 /// The run is over: each environment its jobs deployed to takes the
1876 /// outcome of those jobs (`deployment_outcome`).
1877 async fn settle_deployments(&self, run: &RunRow, jobs: &[JobRow], cancelled: bool) {
1878 let mut seen: Vec<(JobEnvironment, Vec<Option<String>>)> = Vec::new();
1879 for job in jobs {
1880 let Some(env) = deploys_to(run, job) else { continue };
1881 let conclusion = if cancelled && job.conclusion.as_deref() != Some("skipped") && job.started_at.is_some() {
1882 Some("cancelled".to_owned())
1883 } else if job.started_at.is_none() {
1884 Some("skipped".to_owned())
1885 } else {
1886 job.conclusion.clone()
1887 };
1888 match seen.iter_mut().find(|(known, _)| known.name.eq_ignore_ascii_case(&env.name)) {
1889 Some((known, conclusions)) => {
1890 if known.url.is_none() {
1891 known.url = env.url.clone();
1892 }
1893 conclusions.push(conclusion);
1894 }
1895 None => seen.push((env, vec![conclusion])),
1896 }
1897 }
1898 for (env, conclusions) in seen {
1899 if let Some(state) = deployment_outcome(&conclusions) {
1900 self.report_deployment(run, &env, state, true).await;
1901 }
1902 }
1903 }
1904
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1905 /// Tells the pull request (or commit) how the run went, as a status.
1906 async fn report_status(&self, run: &RunRow, conclusion: &str) -> Result<()> {
1907 let state = match conclusion {
1908 "success" | "skipped" => "success",
1909 "cancelled" => "error",
1910 _ => "failure",
1911 };
1912 let _: Result<Value> = g1t_kit::call(
1913 &self.work,
1914 "set_commit_status",
1915 &json!({
1916 "repoId": run.repo_id,
1917 "sha": run.sha,
1918 "context": format!("{} / {}", run.name, run.event),
1919 "state": state,
1920 "description": format!("{} {}", run.name, match conclusion {
1921 "success" => "passed",
1922 "skipped" => "was skipped",
1923 "cancelled" => "was cancelled",
1924 _ => "failed",
1925 }),
1926 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge1927 "source": "actions",
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1928 }),
1929 )
1930 .await;
1931 Ok(())
1932 }
1933
1934 /// Tells the pull request a run has started on its head.
1935 pub async fn report_pending(&self, run: &RunRow) -> Result<()> {
1936 let _: Result<Value> = g1t_kit::call(
1937 &self.work,
1938 "set_commit_status",
1939 &json!({
1940 "repoId": run.repo_id,
1941 "sha": run.sha,
1942 "context": format!("{} / {}", run.name, run.event),
1943 "state": "pending",
Merge branch 'worktree-agent-a3abfcce648e87dca'1944 "description": if run.status == "action_required" {
1945 format!("{} is waiting for approval", run.name)
1946 } else {
1947 format!("{} is running", run.name)
1948 },
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1949 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge1950 "source": "actions",
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1951 }),
1952 )
1953 .await;
1954 Ok(())
1955 }
1956
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1957 /// Cancels a run: its waiting and queued jobs at once, and its running
1958 /// ones gracefully (`stop_job`), or outright when `force`.
1959 pub async fn cancel_run(&self, run: &RunRow, reason: &str, force: bool) -> Result<()> {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1960 self.db
1961 .prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")
1962 .bind(&[run.id.as_str().into()])?
1963 .run()
1964 .await?;
1965 for job in self.job_rows(&run.id).await?.iter().filter(|job| job.status != "completed") {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1966 if force {
1967 self.hard_stop(job, reason).await?;
1968 } else {
1969 self.stop_job(job, reason).await?;
1970 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1971 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1972 if run.status == "pending" || run.status == "action_required" {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1973 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;
1974 }
1975 self.advance(&run.id).await
1976 }
1977
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1978 /// Cancels every run of the repository that has not finished, for
1979 /// `repo.deleted` and `repo.archived`.
1980 pub async fn stop_runs(&self, repo_id: &str) -> Result<()> {
1981 let runs = self
1982 .db
1983 .prepare("SELECT * FROM runs WHERE repo_id = ? AND status != 'completed'")
1984 .bind(&[repo_id.into()])?
1985 .all()
1986 .await?
1987 .results::<RunRow>()?;
1988 for run in runs {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1989 self.cancel_run(&run, "The repository was archived or deleted.", true).await?;
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1990 }
1991 Ok(())
1992 }
1993
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1994 pub async fn cancel(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1995 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API1996 return Ok(Outcome::Fail(refused));
1997 }
1998 let run = check!(self.run_in(&a.repo, &a.id).await?);
1999 if run.status == "completed" {
2000 return Ok(fail(FailureCode::Conflict, "The run has already finished."));
2001 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2002 // Cancelling a run that is already cancelling stops its jobs
2003 // outright, without waiting for their cleanup steps.
2004 let force = a.force || run.conclusion.as_deref() == Some("cancelled");
2005 let reason = if force {
2006 format!("{} stopped the run without waiting for its cleanup steps.", a.actor.username)
2007 } else {
2008 format!("{} cancelled the run.", a.actor.username)
2009 };
2010 self.cancel_run(&run, &reason, force).await?;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2011 self.run_summary(&run.id).await
2012 }
2013
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2014 /// Runs again, as a new attempt: every job, with `failed_only` those
2015 /// that did not succeed, or with `job` that one; each with the jobs that
2016 /// need them. The attempt that ends is kept, its jobs and their logs and
2017 /// summaries with it (`job_attempts`, `run_attempts`).
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2018 pub async fn rerun(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look2019 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2020 return Ok(Outcome::Fail(refused));
2021 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2022 // A job alone (`POST …/jobs/{job}/rerun`) names its run.
2023 let run_id = match (&a.job, a.id.is_empty()) {
2024 (Some(job), true) => {
2025 #[derive(Deserialize)]
2026 struct Of {
2027 run_id: String,
2028 }
2029 let of = self.db.prepare("SELECT run_id FROM jobs WHERE id = ?").bind(&[job.as_str().into()])?.first::<Of>(None).await?;
2030 match of {
2031 Some(of) => of.run_id,
2032 None => return Ok(fail(FailureCode::NotFound, "No such job.")),
2033 }
2034 }
2035 _ => a.id.clone(),
2036 };
2037 let run = check!(self.run_in(&a.repo, &run_id).await?);
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2038 if run.status != "completed" {
2039 return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));
2040 }
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look2041 // Nothing starts again on an archived repository.
2042 match self.visible_repo(&a.repo, &Some(a.actor.clone())).await? {
2043 Some(repo) if repo.archived() => {
2044 return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));
2045 }
2046 Some(_) => {}
2047 None => return Ok(fail(FailureCode::NotFound, "There is no such repository.")),
2048 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2049 if run.error.is_some() {
2050 return Ok(fail(FailureCode::Conflict, "This run never started: fix the workflow file and push again."));
2051 }
2052 let jobs = self.job_rows(&run.id).await?;
2053 let workflow = workflow::parse(&run.source).ok();
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2054 let target = match &a.job {
2055 Some(id) => match jobs.iter().find(|job| &job.id == id) {
2056 Some(job) => Some(top_key(&job.key).to_owned()),
2057 None => return Ok(fail(FailureCode::NotFound, "That job is not in the run's latest attempt.")),
2058 },
2059 None => None,
2060 };
2061 let which = match (&target, a.failed_only) {
2062 (Some(key), _) => Rerun::Job(key),
2063 (None, true) => Rerun::Failed,
2064 (None, false) => Rerun::All,
2065 };
2066 let order = workflow.as_ref().map(|w| w.job_order()).unwrap_or_default();
2067 let again = rerun_keys(
2068 &order,
2069 |key| jobs.iter().find(|j| j.key == key).map(JobRow::needs).unwrap_or_default(),
2070 |key| jobs.iter().filter(|j| j.key == key).all(|j| j.conclusion.as_deref() == Some("success")),
2071 which,
2072 );
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2073 if again.is_empty() {
2074 return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));
2075 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2076 // The jobs that run again, a called workflow's with the job calling it.
2077 let rerun_ids: Vec<&str> = jobs.iter().filter(|job| again.iter().any(|key| key == top_key(&job.key))).map(|job| job.id.as_str()).collect();
2078 let ids = serde_json::to_string(&rerun_ids)?;
2079 let ended = run.attempt.to_string();
2080 let ended = ended.as_str();
2081 let mut statements = vec![
2082 // The attempt that ends, as it ended.
2083 self.db
2084 .prepare(
2085 "INSERT OR REPLACE INTO run_attempts (run_id, attempt, repo_id, conclusion, actor, debug, started_at, finished_at)
2086 SELECT id, attempt, repo_id, conclusion, COALESCE(triggering_actor, actor), debug, started_at, finished_at FROM runs WHERE id = ?",
2087 )
2088 .bind(&[run.id.as_str().into()])?,
2089 // Earlier attempts that showed a job's logs from where it ran
2090 // then now find them where they move to.
2091 self.db
2092 .prepare("UPDATE job_attempts SET log_id = log_id || '.' || ?1 WHERE run_id = ?2 AND log_id IN (SELECT value FROM json_each(?3))")
2093 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
2094 // Each of its jobs: one that runs again keeps its logs under
2095 // `{id}.{attempt}`; one left alone is still the live job's.
2096 self.db
2097 .prepare(
2098 "INSERT OR REPLACE INTO job_attempts (id, run_id, repo_id, attempt, job_id, log_id, key, ordinal, name, needs, status, conclusion,
2099 steps, annotations, reason, environment, labels, runner_name, started_at, finished_at)
2100 SELECT id || '.' || ?1, run_id, repo_id, ?1, id,
2101 CASE WHEN id IN (SELECT value FROM json_each(?3)) THEN id || '.' || ?1 ELSE id END,
2102 key, ordinal, name, needs, status, conclusion, steps, annotations, reason, environment, labels, runner_name, started_at, finished_at
2103 FROM jobs WHERE run_id = ?2 ORDER BY rowid",
2104 )
2105 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
2106 self.db
2107 .prepare("UPDATE logs SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
2108 .bind(&[ended.into(), ids.as_str().into()])?,
2109 self.db
2110 .prepare("UPDATE job_summaries SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
2111 .bind(&[ended.into(), ids.as_str().into()])?,
2112 ];
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2113 for key in &again {
2114 statements.push(self.db.prepare("DELETE FROM jobs WHERE run_id = ? AND key = ? AND ordinal > 0").bind(&[run.id.as_str().into(), key.as_str().into()])?);
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92115 // The jobs of a workflow it called (each matrix combination's
2116 // under `key[i]/`) are made again when it calls it again.
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2117 statements.push(
2118 self.db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92119 .prepare("DELETE FROM jobs WHERE run_id = ? AND (key LIKE ? OR key LIKE ?)")
2120 .bind(&[run.id.as_str().into(), format!("{key}/%").into(), format!("{key}[%").into()])?,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2121 );
2122 statements.push(
2123 self.db
2124 .prepare(
2125 "UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
Fast pages, required checks on the branch, self-hosted runners, honest incidents2126 matrix = NULL, call = NULL, token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
Merge branch 'worktree-agent-a3abfcce648e87dca'2127 labels = NULL, queued_at = NULL, runner_id = NULL, runner_name = NULL, environment = NULL,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2128 concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2129 )
2130 .bind(&[run.id.as_str().into(), key.as_str().into()])?,
2131 );
2132 }
2133 statements.push(
2134 self.db
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2135 .prepare(
2136 "UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL,
2137 triggering_actor = ?, debug = ? WHERE id = ?",
2138 )
2139 .bind(&[a.actor.username.as_str().into(), u32::from(a.debug).into(), run.id.as_str().into()])?,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2140 );
2141 self.db.batch(statements).await?;
2142 if let Some(run) = self.run_row(&run.id).await? {
2143 self.report_pending(&run).await?;
2144 }
2145 self.advance(&run.id).await?;
2146 self.run_summary(&run.id).await
2147 }
2148
2149 pub async fn run_in(&self, repo: &RepoPath, id: &str) -> Result<Outcome<RunRow>> {
2150 let row = self
2151 .db
2152 .prepare("SELECT * FROM runs WHERE id = ? AND lower(repo) = lower(?)")
2153 .bind(&[id.into(), format!("{}/{}", repo.namespace, repo.name).into()])?
2154 .first::<RunRow>(None)
2155 .await?;
2156 Ok(row.map_or_else(|| fail(FailureCode::NotFound, "No such run."), Outcome::Ok))
2157 }
2158
2159 // --- The sandbox's side -----------------------------------------------------
2160
Fast pages, required checks on the branch, self-hosted runners, honest incidents2161 pub(crate) async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2162 let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;
2163 Ok(match job {
2164 Some(job) if job.status == "in_progress" && job.token_hash.as_deref().is_some_and(|hash| same(hash, &sha256_hex(&a.token))) => {
2165 Outcome::Ok(job)
2166 }
2167 _ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),
2168 })
2169 }
2170
2171 /// `job_auth`: which run and repository a running job's token is for,
2172 /// so the API can keep its artifacts and cache.
2173 pub async fn job_auth(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2174 let job = check!(self.job_for_token(&a).await?);
2175 Ok(Outcome::Ok(json!({ "run": job.run_id, "repoId": job.repo_id })))
2176 }
2177
2178 /// `job_spec`: everything the sandbox needs to run the job.
2179 pub async fn job_spec(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2180 let job = check!(self.job_for_token(&a).await?);
2181 let Some(run) = self.run_row(&job.run_id).await? else {
2182 return Ok(fail(FailureCode::NotFound, "No such run."));
2183 };
2184 let Ok(caller) = workflow::parse(&run.source) else {
2185 return Ok(fail(FailureCode::Invalid, "The workflow no longer reads."));
2186 };
2187 // A called workflow's job runs as that workflow defines it.
2188 let callee = job.callee();
2189 let (workflow, spec, call_inputs) = match callee {
2190 Some((called, spec, call)) => (called, spec, Some(call["inputs"].clone())),
2191 None => match caller.jobs.iter().find(|j| j.id == job.key) {
2192 Some(spec) => (caller.clone(), spec.clone(), None),
2193 None => return Ok(fail(FailureCode::NotFound, "The job is not in the workflow.")),
2194 },
2195 };
2196 let spec = &spec;
2197 let repo = repo_path(&run.repo);
2198 let trusted = run.trusted != 0;
Merge branch 'worktree-agent-a3abfcce648e87dca'2199 // The job's `environment:`, by name, as read when its needs were done
2200 // (an expression included), once the environment's protection rules
2201 // let it start: entries with a value for it give that value instead
2202 // of their default, as GitHub's environment secrets do.
2203 let environment: Option<String> = job.environment.clone();
2204 // What its token may do: its `permissions:` (a called workflow's
2205 // jobs no more than the job that calls it), else the repository's
2206 // default; read-only for a pull request from outside.
2207 // The repository's default, its workspace's taken in, and whether
2208 // its jobs may open and approve pull requests.
2209 let (default, pull_requests) = self.token_policy(&run.repo_id).await?;
2210 let mut permissions = spec.permissions(&workflow, default);
2211 if let Some(parent) = &job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned)) {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92212 let top = top_key(parent);
Merge branch 'worktree-agent-a3abfcce648e87dca'2213 if let Some(caller_job) = caller.jobs.iter().find(|j| j.id == top) {
2214 permissions = permissions.capped_by(&caller_job.permissions(&caller, default));
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2215 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2216 }
2217 if !trusted {
2218 permissions = permissions.read_only();
2219 }
2220 // G1T_TOKEN, and GITHUB_TOKEN as its alias: a token of the
2221 // workspace's that reaches this repository only, with the scopes
2222 // its permissions give, until the job ends.
2223 let token = match self.workspace_actor(&repo.namespace).await? {
2224 Some(workspace) => {
2225 let created: CreatedAccessToken = g1t_kit::call(
2226 &self.identity,
2227 "create_job_token",
2228 &CreateJobTokenArgs {
2229 workspace,
2230 repo: repo.clone(),
2231 run_id: run.id.clone(),
2232 job_id: job.id.clone(),
2233 name: format!("G1T_TOKEN for {} run {}", run.repo, run.number),
2234 ttl_seconds: u64::from(job.timeout_minutes) * 60 + 600,
2235 scopes: permissions.scopes().into_iter().map(str::to_owned).collect(),
2236 pull_requests: pull_requests && trusted,
2237 },
2238 )
2239 .await?;
2240 created.token
2241 }
2242 None => String::new(),
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2243 };
2244 // A run that is not trusted (a pull request from outside the
2245 // workspace) gets no secrets and an empty token.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts2246 let passed = job.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());
2247 let mut secrets = match (trusted, passed) {
2248 (false, _) => Map::new(),
2249 (true, None) => self.secrets_for(&run.repo_id, &run.repo, environment.as_deref(), true).await?,
2250 // A called workflow's job: what its callers passed it (reach.rs),
2251 // and its own environment's secrets over them.
2252 (true, Some(plan)) => {
2253 let base = self.secrets_for(&run.repo_id, &run.repo, None, true).await?;
2254 let vars = self.variables_for(&run.repo_id, &run.repo, None, true).await?;
2255 let github = run.info().context(&job.key, "", run.action.as_deref());
2256 let mut passed = crate::reach::resolve_secrets(&plan, &base, &github, &vars);
2257 if let Some(name) = environment.as_deref() {
2258 let own = self.secrets_for(&run.repo_id, &run.repo, Some(name), true).await?;
2259 for (key, value) in own {
2260 if base.get(&key) != Some(&value) {
2261 passed.insert(key, value);
2262 }
2263 }
2264 }
2265 passed
2266 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2267 };
2268 secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));
2269 secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));
Merge branch 'worktree-agent-a3abfcce648e87dca'2270 // Each secret as it is, a line at a time, base64 and JSON-escaped.
2271 let mut masks: Vec<String> = g1t_actions::mask::all_variants(secrets.values().filter_map(|v| v.as_str()));
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2272 let vars = self.variables_for(&run.repo_id, &run.repo, environment.as_deref(), trusted).await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22273 // The toolkit's runtime token (runtime.rs), for as long as the job
2274 // may run; the API puts it and the toolkit's addresses in the
2275 // job's variables.
2276 let runtime_token = crate::runtime::runtime_token(
2277 &job.id,
2278 &job.run_id,
2279 job.token_hash.as_deref().unwrap_or_default(),
2280 now_ms() / 1000,
2281 u64::from(job.timeout_minutes) * 60 + 600,
2282 );
2283 masks.push(runtime_token.clone());
2284 let retention_days = self.retention_setting(&run.repo_id).await?;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2285
2286 let jobs = self.job_rows(&run.id).await?;
2287 let mut needs = Map::new();
2288 // In a called workflow, its jobs' keys sit under the job that called it.
2289 let parent = job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));
2290 for need in &spec.needs {
2291 let key = match &parent {
2292 Some(parent) => format!("{parent}/{need}"),
2293 None => need.clone(),
2294 };
2295 let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();
2296 let mut outputs = Map::new();
2297 for row in &rows {
2298 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
2299 outputs.extend(more);
2300 }
2301 }
2302 needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));
2303 }
2304 let siblings = jobs.iter().filter(|row| row.key == job.key).count();
2305 let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));
2306 let info = run.info();
2307 let mut github = info.context(&job.key, &token, run.action.as_deref());
2308 github["token"] = json!(token);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22309 github["retention_days"] = json!(retention_days);
Fast pages, required checks on the branch, self-hosted runners, honest incidents2310 // On a self-hosted runner, `runner` and `RUNNER_*` describe that
2311 // machine rather than g1t's sandbox.
2312 let mut variables = info.variables(&job.key);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22313 variables.insert("GITHUB_RETENTION_DAYS".into(), json!(retention_days.to_string()));
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2314 let mut runner = match &job.runner_id {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2315 Some(id) => self.runner_context_for(id, &mut variables).await?,
2316 None => runner_context(),
2317 };
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2318 // A re-run with debug logging: what GitHub sets for one.
2319 if run.debug != 0 {
2320 debug_logging(&mut variables, &mut runner);
2321 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2322
2323 // Where to check out: a pull request's fork, or the repository.
2324 let clone_url = match run.pull {
2325 Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" => {
2326 let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(
2327 &self.work,
2328 "get_pull",
2329 &g1t_contracts::work::ViewArgs {
2330 repo: repo.clone(),
2331 number,
2332 viewer: self.workspace_actor(&repo.namespace).await?,
2333 after_seq: 0,
2334 },
2335 )
2336 .await?;
2337 match located {
2338 Outcome::Ok(detail) => match detail.pull.fork {
2339 Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),
2340 None => format!("{SITE}/{}.git", run.repo),
2341 },
2342 Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),
2343 }
2344 }
2345 _ => format!("{SITE}/{}.git", run.repo),
2346 };
2347
2348 Ok(Outcome::Ok(json!({
2349 "job": job.id,
2350 "run": run.id,
2351 "key": job.key,
2352 "name": job.name,
2353 "spec": spec.raw,
2354 "workflow": {
2355 "env": workflow.env,
2356 "defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),
2357 },
2358 "github": github,
Fast pages, required checks on the branch, self-hosted runners, honest incidents2359 "variables": variables,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2360 "event": info.event,
2361 "contexts": {
2362 "vars": vars,
2363 "secrets": secrets,
2364 "inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),
2365 "matrix": matrix,
2366 "needs": needs,
2367 "strategy": {
2368 "fail-fast": spec.fail_fast,
2369 "job-index": job.ordinal,
2370 "job-total": siblings,
2371 "max-parallel": spec.max_parallel.unwrap_or(siblings as u32),
2372 },
Fast pages, required checks on the branch, self-hosted runners, honest incidents2373 "runner": runner,
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2374 },
2375 "checkout": {
2376 "repository": run.repo,
2377 "url": clone_url,
2378 "sha": run.sha,
2379 "ref": run.git_ref,
2380 "token": token,
2381 },
2382 "timeoutMinutes": job.timeout_minutes,
2383 "masks": masks,
Merge branch 'worktree-agent-a3abfcce648e87dca'2384 // As the job's log lists them at its start.
2385 "permissions": permissions.listed().into_iter().map(|(name, access)| (name.to_owned(), json!(access.as_str()))).collect::<Map<String, Value>>(),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22386 // Whether the job may ask for an OIDC token decides whether it
2387 // is told where to.
2388 "runtime": {
2389 "token": runtime_token,
2390 "idToken": self.oidc_allowed(&run, &job),
2391 },
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2392 })))
2393 }
2394
2395 /// `job_report`: the sandbox telling how the job is going.
2396 pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2397 let job = check!(self.job_for_token(&a).await?);
2398 let report = &a.report;
2399 let at = now();
2400 match report["kind"].as_str().unwrap_or_default() {
2401 "steps" => {
2402 // The list can grow as the job goes (post steps), so steps
2403 // already reported keep where they stand.
2404 let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2405 let steps: Vec<Value> = report["steps"]
2406 .as_array()
2407 .map(|names| {
2408 names
2409 .iter()
2410 .enumerate()
2411 .map(|(i, name)| match known.get(i) {
2412 Some(step) if step["status"] != "queued" => step.clone(),
2413 _ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),
2414 })
2415 .collect()
2416 })
2417 .unwrap_or_default();
2418 self.db
2419 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2420 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2421 .run()
2422 .await?;
2423 }
2424 "step" => {
2425 let number = report["number"].as_u64().unwrap_or(0) as usize;
2426 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2427 if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {
2428 let status = report["status"].as_str().unwrap_or("in_progress");
2429 step["status"] = json!(status);
2430 if status == "in_progress" {
2431 step["startedAt"] = json!(at);
2432 }
2433 if status == "completed" {
2434 step["finishedAt"] = json!(at);
2435 step["conclusion"] = report["conclusion"].clone();
2436 }
2437 if let Some(name) = report["name"].as_str() {
2438 step["name"] = json!(name);
2439 }
2440 }
2441 self.db
2442 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2443 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2444 .run()
2445 .await?;
2446 }
2447 "log" => {
2448 let mut text = report["text"].as_str().unwrap_or_default().to_owned();
2449 if text.len() > MAX_CHUNK_BYTES {
2450 let mut cut = MAX_CHUNK_BYTES;
2451 while !text.is_char_boundary(cut) {
2452 cut -= 1;
2453 }
2454 text.truncate(cut);
2455 }
2456 #[derive(Deserialize)]
2457 struct Size {
2458 n: Option<f64>,
2459 }
2460 let size = self
2461 .db
A running job's log is live, as on GitHub. A step's output reaches its run's page within about two seconds: the runner sends what is printed every second, and the page asks every second while a step runs, every three while none does; an answer that fails (a 429, a 5xx, the network) is asked again after a wait that doubles to half a minute, or as long as Retry-After says, where before one such answer stopped the log for good and the step said "No output yet." until the page was reloaded, and a signed-out viewer's run page, refreshing beside its log, ran into the 30-a-minute limit within a minute; the live log now counts as a page, and only log downloads as costly. A hidden tab stops asking and catches up when shown, and the page says when it is reconnecting. The running step opens by itself, its timer counts, its log follows the newest line while its end is on screen and offers Jump to latest once the viewer scrolls up, a step that has printed nothing for a while says "Running · no output for 2m 10s", and the job's header names the step it is on. Colour that tools print (bold, dim, italic, underline, the 16 colours as theme tokens for light and dark, 256 colours and 24-bit colour, text and background) is drawn as printed, other escape codes are dropped and links keep their text, and search reads the plain text and keeps the colours around what it marks. Every line is stamped with the time it was printed, as GitHub's raw logs are, so log.txt and the run's zip have a time at the start of each line, and a clock button beside Search logs shows the times, remembered by the browser. Set up job now gives the runner's version and name, the operating system, the runner image, the machine's architecture, CPUs and memory, the token's permissions under G1T_TOKEN Permissions, the secret source and the actions it downloads; each run step opens with GitHub's folded Run group, the script in cyan, the shell's whole path and the step's env, each uses step with its with inputs (an action's defaults among them) and env; cache and artifact transfers say how far they have got every five seconds; and a last Complete job step sets the job's outputs and stops what the steps left running, found by RUNNER_TRACKING_ID as GitHub's runner finds it. Two batches of a job's log written at once can no longer reach for the same seq and lose one. The Actions guide and the rate limits page say so.2462 .prepare("SELECT SUM(LENGTH(text)) AS n FROM logs WHERE job_id = ?")
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2463 .bind(&[job.id.as_str().into()])?
2464 .first::<Size>(None)
2465 .await?;
A running job's log is live, as on GitHub. A step's output reaches its run's page within about two seconds: the runner sends what is printed every second, and the page asks every second while a step runs, every three while none does; an answer that fails (a 429, a 5xx, the network) is asked again after a wait that doubles to half a minute, or as long as Retry-After says, where before one such answer stopped the log for good and the step said "No output yet." until the page was reloaded, and a signed-out viewer's run page, refreshing beside its log, ran into the 30-a-minute limit within a minute; the live log now counts as a page, and only log downloads as costly. A hidden tab stops asking and catches up when shown, and the page says when it is reconnecting. The running step opens by itself, its timer counts, its log follows the newest line while its end is on screen and offers Jump to latest once the viewer scrolls up, a step that has printed nothing for a while says "Running · no output for 2m 10s", and the job's header names the step it is on. Colour that tools print (bold, dim, italic, underline, the 16 colours as theme tokens for light and dark, 256 colours and 24-bit colour, text and background) is drawn as printed, other escape codes are dropped and links keep their text, and search reads the plain text and keeps the colours around what it marks. Every line is stamped with the time it was printed, as GitHub's raw logs are, so log.txt and the run's zip have a time at the start of each line, and a clock button beside Search logs shows the times, remembered by the browser. Set up job now gives the runner's version and name, the operating system, the runner image, the machine's architecture, CPUs and memory, the token's permissions under G1T_TOKEN Permissions, the secret source and the actions it downloads; each run step opens with GitHub's folded Run group, the script in cyan, the shell's whole path and the step's env, each uses step with its with inputs (an action's defaults among them) and env; cache and artifact transfers say how far they have got every five seconds; and a last Complete job step sets the job's outputs and stops what the steps left running, found by RUNNER_TRACKING_ID as GitHub's runner finds it. Two batches of a job's log written at once can no longer reach for the same seq and lose one. The Actions guide and the rate limits page say so.2466 let used = size.and_then(|s| s.n).unwrap_or(0.0) as usize;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2467 if used < MAX_LOG_BYTES {
2468 if used + text.len() >= MAX_LOG_BYTES {
2469 text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");
2470 }
A running job's log is live, as on GitHub. A step's output reaches its run's page within about two seconds: the runner sends what is printed every second, and the page asks every second while a step runs, every three while none does; an answer that fails (a 429, a 5xx, the network) is asked again after a wait that doubles to half a minute, or as long as Retry-After says, where before one such answer stopped the log for good and the step said "No output yet." until the page was reloaded, and a signed-out viewer's run page, refreshing beside its log, ran into the 30-a-minute limit within a minute; the live log now counts as a page, and only log downloads as costly. A hidden tab stops asking and catches up when shown, and the page says when it is reconnecting. The running step opens by itself, its timer counts, its log follows the newest line while its end is on screen and offers Jump to latest once the viewer scrolls up, a step that has printed nothing for a while says "Running · no output for 2m 10s", and the job's header names the step it is on. Colour that tools print (bold, dim, italic, underline, the 16 colours as theme tokens for light and dark, 256 colours and 24-bit colour, text and background) is drawn as printed, other escape codes are dropped and links keep their text, and search reads the plain text and keeps the colours around what it marks. Every line is stamped with the time it was printed, as GitHub's raw logs are, so log.txt and the run's zip have a time at the start of each line, and a clock button beside Search logs shows the times, remembered by the browser. Set up job now gives the runner's version and name, the operating system, the runner image, the machine's architecture, CPUs and memory, the token's permissions under G1T_TOKEN Permissions, the secret source and the actions it downloads; each run step opens with GitHub's folded Run group, the script in cyan, the shell's whole path and the step's env, each uses step with its with inputs (an action's defaults among them) and env; cache and artifact transfers say how far they have got every five seconds; and a last Complete job step sets the job's outputs and stops what the steps left running, found by RUNNER_TRACKING_ID as GitHub's runner finds it. Two batches of a job's log written at once can no longer reach for the same seq and lose one. The Actions guide and the rate limits page say so.2471 // The next `seq` is taken in the insert itself, so two
2472 // batches at once (a report sent again while the first
2473 // was still being written) never reach for the same one
2474 // and lose a chunk to the primary key.
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2475 self.db
A running job's log is live, as on GitHub. A step's output reaches its run's page within about two seconds: the runner sends what is printed every second, and the page asks every second while a step runs, every three while none does; an answer that fails (a 429, a 5xx, the network) is asked again after a wait that doubles to half a minute, or as long as Retry-After says, where before one such answer stopped the log for good and the step said "No output yet." until the page was reloaded, and a signed-out viewer's run page, refreshing beside its log, ran into the 30-a-minute limit within a minute; the live log now counts as a page, and only log downloads as costly. A hidden tab stops asking and catches up when shown, and the page says when it is reconnecting. The running step opens by itself, its timer counts, its log follows the newest line while its end is on screen and offers Jump to latest once the viewer scrolls up, a step that has printed nothing for a while says "Running · no output for 2m 10s", and the job's header names the step it is on. Colour that tools print (bold, dim, italic, underline, the 16 colours as theme tokens for light and dark, 256 colours and 24-bit colour, text and background) is drawn as printed, other escape codes are dropped and links keep their text, and search reads the plain text and keeps the colours around what it marks. Every line is stamped with the time it was printed, as GitHub's raw logs are, so log.txt and the run's zip have a time at the start of each line, and a clock button beside Search logs shows the times, remembered by the browser. Set up job now gives the runner's version and name, the operating system, the runner image, the machine's architecture, CPUs and memory, the token's permissions under G1T_TOKEN Permissions, the secret source and the actions it downloads; each run step opens with GitHub's folded Run group, the script in cyan, the shell's whole path and the step's env, each uses step with its with inputs (an action's defaults among them) and env; cache and artifact transfers say how far they have got every five seconds; and a last Complete job step sets the job's outputs and stops what the steps left running, found by RUNNER_TRACKING_ID as GitHub's runner finds it. Two batches of a job's log written at once can no longer reach for the same seq and lose one. The Actions guide and the rate limits page say so.2476 .prepare(LOG_INSERT)
2477 .bind(&[job.id.as_str().into(), (report["step"].as_u64().unwrap_or(0) as u32).into(), text.into()])?
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2478 .run()
2479 .await?;
2480 }
2481 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2482 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2483 "summary" => {
2484 // `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the
2485 // step's summary, up to 1 MiB a step and 20 steps a job.
2486 let step = report["step"].as_u64().unwrap_or(0) as u32;
2487 let markdown = report["markdown"].as_str().unwrap_or_default();
2488 #[derive(Deserialize)]
2489 struct Held {
2490 steps: u32,
2491 mine: Option<f64>,
2492 }
2493 let held = self
2494 .db
2495 .prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")
2496 .bind(&[step.into(), job.id.as_str().into()])?
2497 .first::<Held>(None)
2498 .await?;
2499 let (steps, mine) = held.map_or((0, None), |held| (held.steps, held.mine.map(|n| n as usize)));
2500 if summary_fits(steps, mine, markdown.len()) {
2501 self.db
2502 .prepare(
2503 "INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)
2504 ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown || excluded.markdown",
2505 )
2506 .bind(&[job.id.as_str().into(), step.into(), markdown.into()])?
2507 .run()
2508 .await?;
2509 }
2510 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2511 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2512 "annotation" => {
2513 let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
2514 if annotations.len() < MAX_ANNOTATIONS {
2515 annotations.push(json!({
2516 "level": report["level"].as_str().unwrap_or("notice"),
2517 "message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),
2518 "title": report["title"],
2519 "file": report["file"],
2520 "line": report["line"],
2521 }));
2522 self.db
2523 .prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")
2524 .bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?
2525 .run()
2526 .await?;
2527 }
2528 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2529 // Nothing to tell; the answer says whether to stop.
2530 "ping" => {
2531 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2532 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2533 "done" => {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2534 // A job told to stop ends cancelled, however its cleanup went.
2535 let conclusion = if job.cancel_requested_at.is_some() {
2536 "cancelled"
2537 } else {
2538 report["conclusion"]
2539 .as_str()
2540 .filter(|c| matches!(*c, "success" | "failure" | "cancelled"))
2541 .unwrap_or("failure")
2542 };
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2543 let outputs = report["outputs"].as_object().cloned();
2544 Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
2545 }
2546 other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
2547 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2548 // `cancelled`: the run was cancelled, and the runner should stop
2549 // the step it is on and run only its cleanup steps.
2550 Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2551 }
2552
2553 // --- Every minute ---------------------------------------------------------------
2554
2555 pub async fn on_minute(&self, now_ms: u64) -> Result<()> {
2556 let minute = now_ms / 60_000 * 60_000;
Merge platform pause and the hourly usage watcher: staff can pause compute, schedules, indexing or renders for everyone, the watcher emails on a breach and is never blind quietly, and the models proxy holds each run to its cap (billing 0051, integrations 0006)2557 // Staff (or billing's usage watcher) paused scheduled runs across
2558 // g1t: this minute's schedules are skipped, not queued for later.
2559 // Kept 30 seconds in the isolate (g1t_kit::pause).
2560 if g1t_kit::pause::paused(&self.billing, g1t_contracts::billing::PauseLevel::Schedules).await {
2561 worker::console_log!("actions: schedules are paused across g1t; skipped this minute's");
2562 } else if let Err(error) = self.run_schedules(minute).await {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2563 worker::console_error!("actions: schedules failed: {error}");
2564 }
2565 // Jobs whose sandbox went quiet or ran past their time.
2566 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
2567 for job in running {
2568 // Times in g1t's format compare as text.
2569 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
2570 let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());
2571 let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
2572 let over = job.started_at.as_deref().is_some_and(|started| started < before(limit).as_str());
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2573 let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(|at| at < before(CANCEL_GRACE_MS).as_str());
2574 if cancel_overdue {
2575 // Cancelled, and still going after its grace period.
2576 self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;
2577 self.advance(&job.run_id).await?;
2578 } else if over {
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2579 let reason = format!("It ran longer than its time limit of {} minutes.", job.timeout_minutes);
Fast pages, required checks on the branch, self-hosted runners, honest incidents2580 // A self-hosted runner is told to stop on its next poll.
2581 if job.runner_id.is_none() {
2582 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2583 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2584 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2585 } else if silent {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2586 let reason = match &job.runner_name {
2587 Some(name) => format!("The self-hosted runner {name} stopped answering."),
2588 None => "The runner stopped answering.".to_owned(),
2589 };
2590 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2591 }
2592 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92593 // Calls of reusable workflows waiting in a concurrency group, in
2594 // case the job they waited for was cancelled in another run.
2595 #[derive(Deserialize)]
2596 struct RunId {
2597 run_id: String,
2598 }
2599 let held = self
2600 .db
2601 .prepare("SELECT DISTINCT run_id FROM jobs WHERE status = 'waiting' AND concurrency_group IS NOT NULL LIMIT 20")
2602 .all()
2603 .await?
2604 .results::<RunId>()?;
2605 for run in held {
2606 self.advance(&run.run_id).await?;
2607 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2608 if let Err(error) = self.sweep_runners(now_ms).await {
2609 worker::console_error!("actions: the runners' sweep failed: {error}");
2610 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2611 // Jobs held at an environment whose wait timer has run out.
2612 if let Err(error) = self.release_gates().await {
2613 worker::console_error!("actions: environments' gates failed: {error}");
2614 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2615 // Once an hour: the cache's expired entries, and its storage.
2616 if (now_ms / 60_000) % 60 == 7
2617 && let Err(error) = self.sweep_cache(now_ms).await
2618 {
2619 worker::console_error!("actions: the cache's sweep failed: {error}");
2620 }
Merge remote-tracking branch 'origin/claude/peaceful-franklin-7qtkwx'2621 // And the archives of GitHub's actions no job uses any more.
2622 if (now_ms / 60_000) % 60 == 47
2623 && let Err(error) = self.sweep_action_archives(now_ms).await
2624 {
2625 worker::console_error!("actions: the action archives' sweep failed: {error}");
2626 }
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22627 // And artifacts past their time, and the toolkit's abandoned parts.
2628 if (now_ms / 60_000) % 60 == 37 {
2629 if let Err(error) = self.sweep_artifacts(now_ms).await {
2630 worker::console_error!("actions: the artifacts' sweep failed: {error}");
2631 }
2632 if let Err(error) = self.sweep_blob_parts(now_ms).await {
2633 worker::console_error!("actions: the blob parts' sweep failed: {error}");
2634 }
2635 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API2636 self.start_queued().await
2637 }
2638}
2639
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look2640
2641#[cfg(test)]
2642mod stopping {
2643 use super::stops_runs;
2644 use g1t_contracts::events::Event;
2645 use serde_json::{Value, json};
2646
2647 fn event(kind: &str, data: Value) -> Event {
2648 Event {
2649 id: "evt_1".into(),
2650 kind: kind.into(),
2651 source: "repos".into(),
2652 time: "2026-10-05T00:00:00Z".into(),
2653 repo_id: Some("rep_1".into()),
2654 actor: None,
2655 data,
2656 }
2657 }
2658
2659 #[test]
2660 fn deleting_or_archiving_stops_runs() {
2661 assert_eq!(stops_runs(&event("repo.deleted", json!({ "repoId": "rep_1" }))).as_deref(), Some("rep_1"));
2662 assert_eq!(stops_runs(&event("repo.archived", json!({ "archived": true }))).as_deref(), Some("rep_1"));
2663 assert_eq!(stops_runs(&event("repo.unarchived", json!({ "archived": false }))), None);
2664 assert_eq!(stops_runs(&event("repo.restored", json!({}))), None);
2665 assert_eq!(stops_runs(&event("git.push", json!({}))), None);
2666 }
2667}
Fast pages, required checks on the branch, self-hosted runners, honest incidents2668
2669#[cfg(test)]
2670mod status_of_needs {
2671 use std::collections::HashMap;
2672
2673 use super::ancestor_failed;
2674
2675 /// check -> plan -> (migrate) -> core -> edge, as deploy.yml has them,
2676 /// and a job that needs only the last.
2677 fn graph() -> HashMap<&'static str, Vec<&'static str>> {
2678 HashMap::from([
2679 ("check", vec![]),
2680 ("plan", vec!["check"]),
2681 ("migrate", vec!["plan"]),
2682 ("core", vec!["plan", "migrate"]),
2683 ("edge", vec!["plan", "migrate", "core"]),
2684 ("notify", vec!["edge"]),
2685 ])
2686 }
2687
2688 #[test]
2689 fn a_failure_is_seen_however_far_back() {
2690 let needs = graph();
2691 let failed = |which: &'static str| move |key: &str| key == which;
2692 // check failed; plan, and everything after, was skipped for it.
2693 assert!(ancestor_failed(&needs, "notify", failed("check")));
2694 assert!(ancestor_failed(&needs, "core", failed("check")));
2695 assert!(ancestor_failed(&needs, "edge", failed("core")));
2696 // Nothing before a job failed: a skipped migrate is not a failure.
2697 assert!(!ancestor_failed(&needs, "edge", |_| false));
2698 assert!(!ancestor_failed(&needs, "core", failed("edge")));
2699 assert!(!ancestor_failed(&needs, "check", failed("check")));
2700 }
2701
2702 #[test]
2703 fn cycles_and_unknown_keys_end() {
2704 let needs = HashMap::from([("a", vec!["b"]), ("b", vec!["a"])]);
2705 assert!(!ancestor_failed(&needs, "a", |_| false));
2706 assert!(!ancestor_failed(&needs, "missing", |_| true));
2707 }
2708}
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb972709
2710#[cfg(test)]
2711mod deployments {
2712 use serde_json::{Map, Value, json};
2713
2714 use super::{JobEnvironment, deployment_outcome, environment_of};
2715
2716 #[test]
2717 fn a_jobs_environment_is_read_for_deployments() {
2718 let contexts: Map<String, Value> = serde_json::from_value(json!({
2719 "github": { "ref_name": "main", "repository": "acme/web" },
2720 "inputs": { "target": "staging" },
2721 "matrix": {},
2722 }))
2723 .unwrap();
2724 let read = |raw: Value| environment_of(&raw, &contexts);
2725 assert_eq!(read(json!({})), None);
2726 assert_eq!(
2727 read(json!({ "environment": "production" })),
2728 Some(JobEnvironment { name: "production".into(), url: None, deploys: true })
2729 );
2730 assert_eq!(
2731 read(json!({ "environment": { "name": "production", "url": "https://g1t.sh" } })),
2732 Some(JobEnvironment { name: "production".into(), url: Some("https://g1t.sh".into()), deploys: true })
2733 );
2734 // Expressions are filled in from the run.
2735 assert_eq!(
2736 read(json!({ "environment": { "name": "${{ inputs.target }}", "url": "https://${{ github.ref_name }}.example.com" } })),
2737 Some(JobEnvironment { name: "staging".into(), url: Some("https://main.example.com".into()), deploys: true })
2738 );
2739 // Secrets only: no deployment.
2740 assert!(!read(json!({ "environment": { "name": "production", "deployment": false } })).unwrap().deploys);
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r92741 // The same, said in the job's g1t settings.
2742 assert!(!read(json!({ "environment": "production", "g1t": { "deployment": false } })).unwrap().deploys);
2743 assert!(!read(json!({ "environment": { "name": "production" }, "g1t": { "deployment": false } })).unwrap().deploys);
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb972744 // Only http(s) addresses.
2745 assert_eq!(read(json!({ "environment": { "name": "production", "url": "javascript:alert(1)" } })).unwrap().url, None);
2746 }
2747
2748 #[test]
2749 fn a_runs_outcome_for_an_environment() {
2750 let of = |list: &[&str]| deployment_outcome(&list.iter().map(|c| Some((*c).to_owned())).collect::<Vec<_>>());
2751 assert_eq!(of(&["success", "skipped"]), Some("success"));
2752 assert_eq!(of(&["success", "failure"]), Some("failure"));
2753 assert_eq!(of(&["success", "cancelled"]), Some("error"));
2754 assert_eq!(of(&["skipped"]), None);
2755 assert_eq!(deployment_outcome(&[None]), None);
2756 }
2757}
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2758
2759#[cfg(test)]
2760mod reruns {
2761 use serde_json::{Map, Value, json};
2762
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92763 use super::{MAX_SUMMARIES, MAX_SUMMARY_BYTES, Rerun, Scope, Status, call_allowed, call_prefix, debug_logging, rerun_keys, shown_name, summary_fits, top_key};
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2764
2765 /// build ← test ← deploy, and lint on its own.
2766 fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {
2767 let order = ["build", "lint", "test", "deploy"];
2768 let needs = |key: &str| -> Vec<String> {
2769 match key {
2770 "test" => vec!["build".into()],
2771 "deploy" => vec!["test".into()],
2772 _ => Vec::new(),
2773 }
2774 };
2775 rerun_keys(&order, needs, |key| !failed.contains(&key), which)
2776 }
2777
2778 #[test]
2779 fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {
2780 assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);
2781 assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);
2782 assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);
2783 assert!(keys(Rerun::Failed, &[]).is_empty());
2784 // One job, whatever it came to, and what depends on it.
2785 assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);
2786 assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);
2787 assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);
2788 assert!(keys(Rerun::Job("missing"), &[]).is_empty());
2789 }
2790
2791 #[test]
2792 fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {
2793 assert_eq!(top_key("build/test"), "build");
2794 assert_eq!(top_key("build/inner/test"), "build");
2795 assert_eq!(top_key("lint"), "lint");
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92796 // A matrix's calls each have their own jobs, under `key[i]/`.
2797 assert_eq!(top_key("build[1]/test"), "build");
2798 assert_eq!(top_key("build/inner[0]/test"), "build");
2799 assert_eq!(call_prefix("build", Some(1)), "build[1]");
2800 assert_eq!(call_prefix("build/inner", None), "build/inner");
2801 }
2802
2803 #[test]
2804 fn a_run_connects_ten_levels_of_workflows_and_fifty_reusable_ones() {
2805 assert!(call_allowed(1, &[], "a.yml").is_ok());
2806 assert!(call_allowed(9, &[], "a.yml").is_ok());
2807 let deep = call_allowed(10, &[], "a.yml").unwrap_err();
2808 assert!(deep.contains("at most ten") && deep.contains("nine levels"), "{deep}");
2809 let names: Vec<String> = (0..50).map(|n| format!("w{n}.yml")).collect();
2810 let fifty: Vec<&str> = names.iter().map(String::as_str).collect();
2811 // A workflow called again, as by each combination of a matrix, is one.
2812 assert!(call_allowed(1, &fifty, "w3.yml").is_ok());
2813 assert!(call_allowed(1, &fifty, "another.yml").unwrap_err().contains("at most 50 different"));
2814 }
2815
2816 #[test]
2817 fn a_matrixs_calls_are_named_by_the_caller_and_its_values() {
2818 let workflow = super::workflow::parse(
2819 "on: push\njobs:\n build:\n name: Build\n strategy:\n matrix: { os: [linux], node: [20] }\n uses: ./.g1t/workflows/b.yml\n plain:\n uses: ./.g1t/workflows/b.yml\n test:\n name: Test ${{ matrix.os }}\n strategy:\n matrix: { os: [linux] }\n runs-on: ubuntu-latest\n steps: [{ run: 'true' }]",
2820 )
2821 .unwrap();
2822 let job = |id: &str| workflow.jobs.iter().find(|job| job.id == id).unwrap();
2823 let combination: Map<String, Value> = serde_json::from_value(json!({ "os": "linux", "node": 20 })).unwrap();
2824 let mut contexts = Map::new();
2825 contexts.insert("matrix".into(), Value::Object(combination.clone()));
2826 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
2827 assert_eq!(shown_name(job("build"), None, &combination, &scope), "Build (linux, 20)");
2828 assert_eq!(shown_name(job("plain"), None, &Map::new(), &scope), "plain");
2829 // A called workflow's job, under the call that made it.
2830 assert_eq!(shown_name(job("plain"), Some("Build (linux, 20)"), &Map::new(), &scope), "Build (linux, 20) / plain");
2831 assert_eq!(shown_name(job("test"), Some("Build"), &combination, &scope), "Build / Test linux");
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2832 }
2833
2834 #[test]
2835 fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {
2836 assert!(summary_fits(0, None, 10));
2837 assert!(!summary_fits(0, None, 0), "nothing to keep");
2838 assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");
2839 assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");
2840 assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));
2841 assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));
2842 assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));
2843 }
2844
2845 #[test]
2846 fn a_debug_rerun_sets_what_github_sets() {
2847 let mut variables = Map::new();
2848 let mut runner = json!({ "name": "g1t", "debug": "" });
2849 debug_logging(&mut variables, &mut runner);
2850 assert_eq!(variables["RUNNER_DEBUG"], "1");
2851 assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");
2852 assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");
2853 assert_eq!(runner["debug"], Value::String("1".into()));
2854 assert_eq!(runner["name"], "g1t");
2855 }
2856}

This file's history is long; its oldest lines are credited to the oldest commit read.