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,975 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.

Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1//! 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};
10use 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};
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow13use 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;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow23use 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};
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow26
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
Merge branch 'worktree-agent-a39a116f38472effe' into claude/cool-newton-7nm8r933/// in the same statement (SQLite runs one write at a time). `?4` is the
34/// runner's number for the batch (null from runners that send none): a
35/// batch sent again, after an answer the runner did not hear, has the
36/// number of one already kept, and is not kept twice (migration 0016).
37const LOG_INSERT: &str = "INSERT INTO logs (job_id, seq, step, text, chunk)
38 SELECT ?1, COALESCE(MAX(seq), 0) + 1, ?2, ?3, ?4 FROM logs WHERE job_id = ?1
39 ON CONFLICT (job_id, chunk) DO NOTHING";
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)40/// The most one step's job summary keeps, in bytes, and how many steps of a
41/// job may have one, as on GitHub.
42pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
43pub(crate) const MAX_SUMMARIES: u32 = 20;
44
45/// Whether a step's summary of `adding` bytes is kept: a job keeps up to
46/// 20 steps' summaries (`steps` it has, `mine` this step's bytes so far)
47/// of up to 1 MiB each.
48pub(crate) fn summary_fits(steps: u32, mine: Option<usize>, adding: usize) -> bool {
49 if adding == 0 {
50 return false;
51 }
52 match mine {
53 Some(used) => used + adding <= MAX_SUMMARY_BYTES,
54 None => steps < MAX_SUMMARIES && adding <= MAX_SUMMARY_BYTES,
55 }
56}
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow57
58/// The repository whose unfinished runs `event` stops, by id: one deleted,
59/// or archived (not unarchived).
60pub fn stops_runs(event: &g1t_contracts::events::Event) -> Option<String> {
61 let stops = match event.kind.as_str() {
62 "repo.deleted" => true,
63 "repo.archived" => event.data["archived"].as_bool() != Some(false),
64 _ => false,
65 };
66 if !stops {
67 return None;
68 }
69 event.repo_id.clone().or_else(|| event.data["repoId"].as_str().map(str::to_owned))
70}
71
72pub struct NewRun {
73 pub repo: Repo,
74 pub path: String,
75 pub source: String,
76 pub workflow: Workflow,
77 pub info: RunInfo,
78 pub action: Option<String>,
79 pub pull: Option<u32>,
80 pub title: String,
81 pub inputs: Map<String, Value>,
82 pub event_key: String,
83 pub actor_id: Option<String>,
84 pub actor: Option<String>,
85 pub trusted: bool,
Merge branch 'worktree-agent-a3abfcce648e87dca'86 /// Why the run waits for someone with the Write role to approve it
87 /// before anything starts: a pull request from outside (protection.rs).
88 pub approval: Option<String>,
A pull request's `pull_request` runs build the pull request merged into its base branch, as on GitHub. g1t keeps `refs/pull/<number>/head` at every open pull request's head (a fork's or working copy's head commit is copied into the repository) and `refs/pull/<number>/merge` at a test merge commit of that head into the base branch's latest commit, moved whenever the head is pushed to, the pull request is opened or reopened, or its base changes; both can be fetched from the repository. A run's `github.sha` and `GITHUB_SHA` are that merge commit, `github.event.pull_request.merge_commit_sha` names it, `actions/checkout` fetches it as `+<sha>:refs/remotes/pull/<number>/merge` from the repository itself, and the workflow files and local reusable workflows are read from it; the run, its statuses and its check runs stay on the head, so required checks hold merges as before, and `pull_request_target` is unchanged. A pull request with conflicts starts no `pull_request` workflows, as on GitHub, and its checks box says why. g1t makes the merge without a sandbox, merging a file both sides changed line by line as git does; what it cannot merge itself (a rename on one side against a change on the other, more than 300 changed files) builds the head as before. The Actions and Checks guides say so.89 /// The commit the run is listed, and its statuses and checks reported,
90 /// on when it is not the one it builds (`info.sha`): a pull request's
91 /// head, when the run builds its test merge commit, as on GitHub.
92 pub head_sha: Option<String>,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow93}
94
95#[derive(Clone, Deserialize)]
96pub struct RunRow {
97 pub id: String,
98 pub workflow_id: String,
99 pub repo_id: String,
100 pub repo: String,
101 pub path: String,
102 pub name: String,
103 pub title: String,
104 pub number: u64,
105 pub attempt: u64,
106 pub event: String,
107 pub action: Option<String>,
108 pub git_ref: String,
109 pub sha: String,
110 pub pull: Option<u32>,
111 pub status: String,
112 pub conclusion: Option<String>,
113 pub error: Option<String>,
114 pub actor: Option<String>,
115 pub actor_id: Option<String>,
116 pub source: String,
117 pub info: String,
118 pub inputs: String,
119 pub trusted: u32,
120 pub concurrency_group: Option<String>,
121 pub created_at: String,
122 pub started_at: Option<String>,
123 pub finished_at: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'124 /// JSON `RunApproval`, for a run that needed approval (migration 0006).
125 #[serde(default)]
126 pub approval: Option<String>,
127 /// Whether its concurrency group cancels what it replaces.
128 #[serde(default)]
129 pub cancel_in_progress: u32,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)130 /// Who started the current attempt, once it is re-run (migration 0009).
131 #[serde(default)]
132 pub triggering_actor: Option<String>,
133 /// The current attempt runs with debug logging.
134 #[serde(default)]
135 pub debug: u32,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow136}
137
138impl RunRow {
Merge branch 'worktree-agent-a3abfcce648e87dca'139 pub fn approval(&self) -> Option<RunApproval> {
140 self.approval.as_deref().and_then(|text| serde_json::from_str(text).ok())
141 }
142
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow143 pub fn info(&self) -> RunInfo {
144 let mut info: RunInfo = serde_json::from_str(&self.info).unwrap_or_default();
145 info.run_id = self.id.clone();
146 info.run_number = self.number;
147 info.run_attempt = self.attempt;
148 info.workflow_path = self.path.clone();
149 if info.workflow.is_empty() {
150 info.workflow = self.name.clone();
151 }
152 info
153 }
154
155 pub fn inputs(&self) -> Map<String, Value> {
156 serde_json::from_str(&self.inputs).unwrap_or_default()
157 }
158
159 pub fn summary(&self) -> WorkflowRun {
160 WorkflowRun {
161 id: self.id.clone(),
162 workflow_id: self.workflow_id.clone(),
163 path: self.path.clone(),
164 name: self.name.clone(),
165 title: self.title.clone(),
166 number: self.number,
167 attempt: self.attempt,
168 event: self.event.clone(),
169 git_ref: self.git_ref.clone(),
170 sha: self.sha.clone(),
171 pull: self.pull,
172 status: self.status.clone(),
173 conclusion: self.conclusion.clone(),
174 error: self.error.clone(),
175 actor: self.actor.clone(),
176 created_at: self.created_at.clone(),
177 started_at: self.started_at.clone(),
178 finished_at: self.finished_at.clone(),
179 }
180 }
181}
182
183#[derive(Clone, Deserialize)]
184pub struct JobRow {
185 pub id: String,
186 pub run_id: String,
187 pub repo_id: String,
188 pub namespace: String,
189 pub key: String,
190 pub ordinal: u32,
191 pub name: String,
192 pub needs: String,
193 pub matrix: Option<String>,
194 pub status: String,
195 pub conclusion: Option<String>,
196 pub steps: String,
197 pub annotations: String,
198 pub outputs: String,
199 pub reason: Option<String>,
200 pub token_hash: Option<String>,
201 pub timeout_minutes: u32,
202 pub continue_on_error: u32,
203 pub max_parallel: Option<u32>,
204 /// Set for a job that calls a reusable workflow, and for that
205 /// workflow's jobs (see migration 0002).
206 pub call: Option<String>,
207 pub seen_at: Option<String>,
208 pub started_at: Option<String>,
209 pub finished_at: Option<String>,
Fast pages, required checks on the branch, self-hosted runners, honest incidents210 /// For a job whose `runs-on` names self-hosted runners: what it asks
211 /// for, as a JSON array (see `g1t_contracts::runners::Wanted`), when it
212 /// started waiting, and the runner that took it (migration 0004).
213 #[serde(default)]
214 pub labels: Option<String>,
215 #[serde(default)]
216 pub queued_at: Option<String>,
217 #[serde(default)]
218 pub runner_id: Option<String>,
219 #[serde(default)]
220 pub runner_name: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'221 /// The environment it names, read when its needs were done; a job its
222 /// rules hold is `pending` (migration 0006).
223 #[serde(default)]
224 pub environment: Option<String>,
225 /// Its own `concurrency` group, and whether that cancels what it replaces.
226 #[serde(default)]
227 pub concurrency_group: Option<String>,
228 #[serde(default)]
229 pub cancel_in_progress: u32,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)230 /// When it was told to stop, while it runs its cleanup steps
231 /// (migration 0009).
232 #[serde(default)]
233 pub cancel_requested_at: Option<String>,
234}
235
236/// How long a cancelled job has to run its `if: always()` and `cancelled()`
237/// steps and its post steps before it is stopped outright, as on GitHub.
238pub const CANCEL_GRACE_MS: u64 = 5 * 60 * 1000;
239
240/// Which jobs a re-run runs again.
241#[derive(Clone, Copy, Debug, PartialEq)]
242pub enum Rerun<'a> {
243 All,
244 /// Those that did not succeed.
245 Failed,
246 /// One job's key.
247 Job(&'a str),
248}
249
250/// The keys a re-run runs again, in `order`: those `which` names and,
251/// transitively, every key that needs one of them. `needs` gives a key's
252/// needs; `succeeded` whether every job of a key succeeded.
253pub fn rerun_keys(order: &[&str], needs: impl Fn(&str) -> Vec<String>, succeeded: impl Fn(&str) -> bool, which: Rerun) -> Vec<String> {
254 let mut again: Vec<String> = Vec::new();
255 for key in order {
256 let named = match which {
257 Rerun::All => true,
258 Rerun::Failed => !succeeded(key),
259 Rerun::Job(job) => *key == job,
260 };
261 if named || needs(key).iter().any(|need| again.contains(need)) {
262 again.push((*key).to_owned());
263 }
264 }
265 again
266}
267
268/// The key under `jobs:` a job belongs to: a called workflow's jobs
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9269/// (`build/test`, or `build[1]/test` for a matrix's second call) run again
270/// with the job that calls it (`build`).
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)271pub fn top_key(key: &str) -> &str {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9272 let first = key.split('/').next().unwrap_or(key);
273 first.split('[').next().unwrap_or(first)
274}
275
276/// What the keys of the jobs a caller's call makes start with: the
277/// caller's key, and for a matrix the combination's index after it
278/// (`build[1]`), so each combination's call has jobs of its own. A job id
279/// has no `[`, so the two never meet.
280fn call_prefix(key: &str, index: Option<usize>) -> String {
281 match index {
282 Some(index) => format!("{key}[{index}]"),
283 None => key.to_owned(),
284 }
285}
286
287/// Whether a run may call one more reusable workflow, `file`, `depth`
288/// levels below the workflow it started from, having called `called`
289/// already (one name per call; repeats are one workflow). As on GitHub: ten
290/// levels of workflows at most, the top one and nine reusable ones below
291/// it, and 50 different reusable workflows in a run.
292fn call_allowed(depth: u64, called: &[&str], file: &str) -> std::result::Result<(), String> {
293 if depth > MAX_CALL_DEPTH {
294 return Err(format!(
295 "`{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.",
296 depth + 1
297 ));
298 }
299 let mut unique: Vec<&str> = called.to_vec();
300 unique.push(file);
301 unique.sort_unstable();
302 unique.dedup();
303 if unique.len() > MAX_CALLED_WORKFLOWS {
304 return Err(format!("Calling `{file}` would make {} reusable workflows in this run: a run calls at most {MAX_CALLED_WORKFLOWS} different ones.", unique.len()));
305 }
306 Ok(())
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)307}
308
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9309/// What a job is shown as: its `name` (else its id), under the job whose
310/// call made it (`caller`), read with its matrix when it has an
311/// expression, else with the matrix's values after it.
312fn shown_name(job: &workflow::Job, caller: Option<&str>, combination: &Map<String, Value>, scope: &Scope) -> String {
313 let base = job.name.clone().unwrap_or(job.id.clone());
314 let base = match caller {
315 Some(caller) => format!("{caller} / {base}"),
316 None => base,
317 };
318 if expr::has_expression(&base) {
319 expr::interpolate(&base, scope).unwrap_or(base)
320 } else if job.matrix.is_some() {
321 matrix::job_name(&base, combination)
322 } else {
323 base
324 }
325}
326
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)327/// Debug logging for a job, as a re-run with it turned on gives GitHub's:
328/// `RUNNER_DEBUG=1` and `runner.debug`, and `ACTIONS_STEP_DEBUG` and
329/// `ACTIONS_RUNNER_DEBUG` set, which the runner reads as the secrets of
330/// those names (`::debug::` lines shown, and how each step's `if` read).
331pub fn debug_logging(variables: &mut Map<String, Value>, runner: &mut Value) {
332 variables.insert("RUNNER_DEBUG".into(), json!("1"));
333 variables.insert("ACTIONS_STEP_DEBUG".into(), json!("true"));
334 variables.insert("ACTIONS_RUNNER_DEBUG".into(), json!("true"));
335 runner["debug"] = json!("1");
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow336}
337
338impl JobRow {
339 pub fn needs(&self) -> Vec<String> {
340 serde_json::from_str(&self.needs).unwrap_or_default()
341 }
342
343 pub fn call(&self) -> Option<Value> {
344 self.call.as_deref().and_then(|call| serde_json::from_str(call).ok())
345 }
346
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9347 /// For a job that calls a reusable workflow: what the keys of the jobs
348 /// its call makes start with (`call_prefix`).
349 pub fn call_prefix(&self) -> Option<String> {
350 let call = self.call().filter(|call| call["role"] == "caller")?;
351 Some(call["prefix"].as_str().unwrap_or(&self.key).to_owned())
352 }
353
354 /// A job that calls a reusable workflow and has not called it yet: it
355 /// waits for room under its matrix's `max-parallel`, or in its
356 /// concurrency group.
357 pub fn holds_call(&self) -> bool {
358 self.status == "waiting" && self.call_prefix().is_some()
359 }
360
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow361 /// For a job of a called workflow: that workflow, the job's own id in
362 /// it, and the job.
363 pub fn callee(&self) -> Option<(Workflow, workflow::Job, Value)> {
364 let call = self.call().filter(|call| call["role"] == "callee")?;
365 let called = workflow::parse(call["source"].as_str()?).ok()?;
366 let job = called.jobs.iter().find(|job| call["job"].as_str() == Some(job.id.as_str()))?.clone();
367 Some((called, job, call))
368 }
369}
370
371/// A job to decide on: its key, its definition, and what it needs, as
372/// (name in `needs`, key of the jobs).
373type Unit = (String, workflow::Job, Vec<(String, String)>);
374
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9375/// How many levels of reusable workflows may sit below the workflow a run
376/// started from, as on GitHub (ten levels of workflows in all).
377const MAX_CALL_DEPTH: u64 = 9;
378/// How many different reusable workflows a run may call, as on GitHub.
379const MAX_CALLED_WORKFLOWS: usize = 50;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow380
381/// What the jobs of one key came to, for `needs.<key>`.
382fn key_result(rows: &[&JobRow]) -> &'static str {
383 let failed = |row: &&&JobRow| row.conclusion.as_deref() == Some("failure") && row.continue_on_error == 0;
384 if rows.iter().any(|row| failed(&row)) {
385 "failure"
386 } else if rows.iter().any(|row| row.conclusion.as_deref() == Some("cancelled")) {
387 "cancelled"
388 } else if rows.iter().all(|row| row.conclusion.as_deref() == Some("skipped")) {
389 "skipped"
390 } else {
391 "success"
392 }
393}
394
Fast pages, required checks on the branch, self-hosted runners, honest incidents395/// Whether any job before `key` failed: one it needs, or one those need,
396/// however far back. A job after a skipped one still sees the failure
397/// that skipped it, as GitHub's failure() does. `needs_of`: each key's
398/// needs, as keys; `failed`: whether a key's jobs came to a failure.
399fn ancestor_failed(needs_of: &std::collections::HashMap<&str, Vec<&str>>, key: &str, failed: impl Fn(&str) -> bool) -> bool {
400 let mut seen = std::collections::HashSet::new();
401 let mut stack: Vec<&str> = needs_of.get(key).cloned().unwrap_or_default();
402 while let Some(next) = stack.pop() {
403 if !seen.insert(next) {
404 continue;
405 }
406 if failed(next) {
407 return true;
408 }
409 stack.extend(needs_of.get(next).into_iter().flatten().copied());
410 }
411 false
412}
413
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97414/// A job's `environment:`, as deployments read it: its name, the address in
415/// `environment.url` once its expressions are filled in from the run, and
416/// whether the job deploys to it. A job with `deployment: false` only reads
417/// the environment's secrets and variables, and makes no deployment.
418#[derive(Clone, Debug, Default, PartialEq)]
419pub(crate) struct JobEnvironment {
420 pub(crate) name: String,
421 pub(crate) url: Option<String>,
422 pub(crate) deploys: bool,
423}
424
425/// The environment `environment:` names, as written in `raw` (a job), with
426/// `contexts` to fill in expressions: null when it names none, or only
427/// through an expression this cannot read before the job runs.
428pub(crate) fn environment_of(raw: &Value, contexts: &Map<String, Value>) -> Option<JobEnvironment> {
429 let scope = Scope { contexts, status: Status::Success, hash_files: None };
430 let plain = |value: &Value| -> Option<String> {
431 let text = match value {
432 Value::String(text) if text.contains("${{") => expr::interpolate_value(value, &scope).ok().map(|v| expr::to_text(&v))?,
433 Value::String(text) => text.clone(),
434 _ => return None,
435 };
436 let text = text.trim().to_owned();
437 (!text.is_empty()).then_some(text)
438 };
439 match raw.get("environment")? {
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9440 name @ Value::String(_) => Some(JobEnvironment { name: plain(name)?, url: None, deploys: workflow::records_deployment(raw) }),
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97441 Value::Object(env) => {
442 let name = plain(env.get("name")?)?;
443 let url = env
444 .get("url")
445 .and_then(plain)
446 .filter(|url| url.starts_with("https://") || url.starts_with("http://"));
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9447 Some(JobEnvironment { name, url, deploys: workflow::records_deployment(raw) })
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97448 }
449 _ => None,
450 }
451}
452
453/// The contexts a job's `runs-on` and `environment` are read with before it
454/// runs: the run's `github`, its inputs, and the job's matrix.
455fn start_contexts(run: &RunRow, job: &JobRow, key: &str) -> Map<String, Value> {
456 let mut contexts = Map::new();
457 contexts.insert("github".into(), run.info().context(key, "", run.action.as_deref()));
458 contexts.insert("inputs".into(), Value::Object(run.inputs()));
459 contexts.insert("matrix".into(), job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or_else(|| json!({})));
460 contexts
461}
462
463/// The job as its workflow (or the workflow it calls) defines it.
464fn job_spec(run: &RunRow, job: &JobRow) -> Option<workflow::Job> {
465 match job.callee() {
466 Some((_, spec, _)) => Some(spec),
467 None => workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)),
468 }
469}
470
Merge branch 'worktree-agent-a3abfcce648e87dca'471/// The environment a job deploys to, if it deploys: by the name read when
472/// its needs were done (which may have needed their outputs), else as
473/// its `environment:` reads now.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97474pub(crate) fn deploys_to(run: &RunRow, job: &JobRow) -> Option<JobEnvironment> {
475 let spec = job_spec(run, job)?;
Merge branch 'worktree-agent-a3abfcce648e87dca'476 let read = environment_of(&spec.raw, &start_contexts(run, job, &spec.id));
477 let env = match (read, &job.environment) {
478 (Some(env), Some(name)) => Some(JobEnvironment { name: name.clone(), ..env }),
479 (Some(env), None) => Some(env),
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9480 (None, Some(name)) => Some(JobEnvironment { name: name.clone(), url: None, deploys: workflow::records_deployment(&spec.raw) }),
Merge branch 'worktree-agent-a3abfcce648e87dca'481 (None, None) => None,
482 };
483 env.filter(|env| env.deploys)
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97484}
485
Fast pages, required checks on the branch, self-hosted runners, honest incidents486/// What the runner is told about a job it starts, from its workflow: the
487/// environment it names plainly, and the machine its `runs-on` asks for
488/// (`instance_for`; none for the standard one).
489#[derive(Default)]
490struct StartDetails {
491 environment: Option<String>,
492 instance: Option<String>,
493}
494
495fn start_details(run: &RunRow, job: &JobRow) -> StartDetails {
496 let spec = match job.callee() {
497 Some((_, spec, _)) => spec,
498 None => match workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)) {
499 Some(spec) => spec,
500 None => return StartDetails::default(),
501 },
502 };
Merge branch 'worktree-agent-a3abfcce648e87dca'503 // As read when its needs were done, an expression's included.
504 let environment = job.environment.clone();
Fast pages, required checks on the branch, self-hosted runners, honest incidents505 // `runs-on` as the job was queued with: its matrix and the run's
506 // inputs. A label that needs more than those is the standard machine.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97507 let contexts = start_contexts(run, job, &spec.id);
Fast pages, required checks on the branch, self-hosted runners, honest incidents508 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
509 let labels: Vec<String> = match expr::interpolate_value(&spec.runs_on, &scope).unwrap_or(Value::Null) {
510 Value::String(label) => vec![label],
511 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
512 Value::Object(given) => match given.get("labels") {
513 Some(Value::Array(labels)) => labels.iter().map(expr::to_text).collect(),
514 Some(label) => vec![expr::to_text(label)],
515 None => Vec::new(),
516 },
517 _ => Vec::new(),
518 };
519 let instance = g1t_contracts::actions::instance_for(&labels);
520 StartDetails {
521 environment,
522 instance: (instance != g1t_contracts::actions::STANDARD_INSTANCE).then(|| instance.label.to_owned()),
523 }
524}
525
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow526fn now() -> String {
527 rfc3339(now_ms())
528}
529
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97530/// How a finished run went for one environment, from the conclusions of
531/// the jobs that deploy to it: failed if any failed, an error if any was
532/// cancelled, else a success; nothing when none ran (all skipped).
533pub(crate) fn deployment_outcome(conclusions: &[Option<String>]) -> Option<&'static str> {
534 let ran: Vec<&str> = conclusions.iter().flatten().map(String::as_str).filter(|c| *c != "skipped").collect();
535 if ran.is_empty() {
536 return None;
537 }
538 if ran.iter().any(|c| *c == "failure" || *c == "timed_out") {
539 return Some("failure");
540 }
541 if ran.contains(&"cancelled") {
542 return Some("error");
543 }
544 Some("success")
545}
546
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow547impl Actions {
548 pub async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {
549 self.db.prepare("SELECT * FROM runs WHERE id = ?").bind(&[id.into()])?.first::<RunRow>(None).await
550 }
551
552 pub async fn job_rows(&self, run_id: &str) -> Result<Vec<JobRow>> {
553 self.db
554 .prepare("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid")
555 .bind(&[run_id.into()])?
556 .all()
557 .await?
558 .results::<JobRow>()
559 }
560
561 pub async fn run_summary(&self, id: &str) -> Result<Outcome<WorkflowRun>> {
562 Ok(match self.run_row(id).await? {
563 Some(row) => Outcome::Ok(row.summary()),
564 None => fail(FailureCode::NotFound, "No such run."),
565 })
566 }
567
568 /// The contexts every expression outside a job's steps may use.
569 fn base_contexts(run: &RunRow, vars: &Map<String, Value>, job: &str) -> Map<String, Value> {
570 let mut contexts = Map::new();
571 contexts.insert("github".into(), run.info().context(job, "", run.action.as_deref()));
572 contexts.insert("inputs".into(), Value::Object(run.inputs()));
573 contexts.insert("vars".into(), Value::Object(vars.clone()));
574 contexts.insert("needs".into(), json!({}));
575 contexts.insert("runner".into(), runner_context());
576 contexts
577 }
578
A workflow's event payload (`github.event` and the file at `GITHUB_EVENT_PATH`) reads as GitHub's webhook payload does. Ids are whole numbers, the same for an object wherever it appears and safe in JavaScript: `repository.id`, `pull_request.id`, `issue.id`, `review.id`, `sender.id` and every user's and owner's `id`, with g1t's own id as `node_id`; `github.repository_id`, `github.repository_owner_id`, `github.actor_id` and their `GITHUB_*` variables are the same numbers. What g1t's API takes back by id (comments, releases, deployments, workflow runs) keeps g1t's id, so `github.event.comment.id` still edits the comment. A repository in a person's own workspace is owned by a `User`, not an `Organization`. A push says whether it was `forced`, `created` or `deleted`, and its `head_commit` is the commit pushed. A pull request's `base.sha` is the tip of the branch it merges into at the event, not the merge base, and `merge_commit_sha` is the commit it landed as once merged; pull requests and issues carry `node_id`, `assignee`, `locked`, dates and label and milestone ids, and a fork's head names the fork. `workflow_run` has the followed run's attempt, times, actor, head commit, workflow id and URLs. Tests compare each payload's keys and types with trimmed copies of GitHub's documented payloads, and the Actions guide says what differs. OIDC claims keep g1t's ids, so trust policies already written still match.579 /// Whether `namespace` is a person's own workspace: usernames and
580 /// workspaces share one namespace, so it is when someone has that
581 /// username. Kept in the isolate for a few minutes, so a burst of
582 /// runs asks identity once.
583 async fn personal_namespace(&self, namespace: &str) -> Result<bool> {
584 const KEEP_MS: u64 = 5 * 60 * 1000;
585 thread_local! {
586 static PERSONAL: std::cell::RefCell<std::collections::HashMap<String, (bool, u64)>> = Default::default();
587 }
588 let now = now_ms();
589 let key = namespace.to_ascii_lowercase();
590 if let Some(personal) = PERSONAL.with(|kept| kept.borrow().get(&key).filter(|(_, at)| now.saturating_sub(*at) < KEEP_MS).map(|(personal, _)| *personal)) {
591 return Ok(personal);
592 }
593 let found: g1t_contracts::Viewer = g1t_kit::call(
594 &self.identity,
595 "user_by_username",
596 &g1t_contracts::identity::UsernameArgs { username: key.clone() },
597 )
598 .await?;
599 let personal = found.is_some_and(|user| user.username.eq_ignore_ascii_case(namespace));
600 PERSONAL.with(|kept| {
601 let mut kept = kept.borrow_mut();
602 if kept.len() > 1000 {
603 kept.clear();
604 }
605 kept.insert(key, (personal, now));
606 });
607 Ok(personal)
608 }
609
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow610 /// Makes a run and its jobs, and starts what can start. `None` when the
611 /// event already started this workflow.
612 pub async fn create_run(&self, new: NewRun) -> Result<Option<String>> {
613 // Nothing starts on an archived repository. A deleted one is never
614 // found to start anything on.
615 if new.repo.archived() {
616 return Ok(None);
617 }
618 let workflow_row = self.workflow_row(&new.repo, &new.path, &new.workflow.display_name(&new.path), &new.source).await?;
619 let numbered = self
620 .db
621 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
622 .bind(&[workflow_row.id.as_str().into()])?
623 .first::<Count>(None)
624 .await?
625 .map_or(1, |count| count.n);
626 let id = new_id("run", now_ms());
627 let mut info = new.info.clone();
A workflow's event payload (`github.event` and the file at `GITHUB_EVENT_PATH`) reads as GitHub's webhook payload does. Ids are whole numbers, the same for an object wherever it appears and safe in JavaScript: `repository.id`, `pull_request.id`, `issue.id`, `review.id`, `sender.id` and every user's and owner's `id`, with g1t's own id as `node_id`; `github.repository_id`, `github.repository_owner_id`, `github.actor_id` and their `GITHUB_*` variables are the same numbers. What g1t's API takes back by id (comments, releases, deployments, workflow runs) keeps g1t's id, so `github.event.comment.id` still edits the comment. A repository in a person's own workspace is owned by a `User`, not an `Organization`. A push says whether it was `forced`, `created` or `deleted`, and its `head_commit` is the commit pushed. A pull request's `base.sha` is the tip of the branch it merges into at the event, not the merge base, and `merge_commit_sha` is the commit it landed as once merged; pull requests and issues carry `node_id`, `assignee`, `locked`, dates and label and milestone ids, and a fork's head names the fork. `workflow_run` has the followed run's attempt, times, actor, head commit, workflow id and URLs. Tests compare each payload's keys and types with trimmed copies of GitHub's documented payloads, and the Actions guide says what differs. OIDC claims keep g1t's ids, so trust policies already written still match.628 // A personal workspace's repositories are a `User`'s, as on GitHub.
629 if self.personal_namespace(&new.repo.namespace).await? {
630 crate::payload::owned_by_person(&mut info.event, &new.repo.namespace);
631 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow632 info.workflow = new.workflow.display_name(&new.path);
633 info.workflow_path = new.path.clone();
634 info.run_id = id.clone();
635 info.run_number = u64::from(numbered);
636 let vars = self
637 .variables_for(&new.repo.id, &format!("{}/{}", new.repo.namespace, new.repo.name), None, new.trusted)
638 .await?;
639
640 // run-name and the concurrency group read github, inputs and vars.
641 let mut contexts = Map::new();
642 contexts.insert("github".into(), info.context("", "", new.action.as_deref()));
643 contexts.insert("inputs".into(), Value::Object(new.inputs.clone()));
644 contexts.insert("vars".into(), Value::Object(vars));
645 let scope = Scope {
646 contexts: &contexts,
647 status: Status::Success,
648 hash_files: None,
649 };
650 let title = new
651 .workflow
652 .run_name
653 .as_deref()
654 .and_then(|run_name| expr::interpolate(run_name, &scope).ok())
655 .filter(|title| !title.trim().is_empty())
656 .unwrap_or(new.title.clone());
657 let group = new.workflow.concurrency.as_ref().and_then(|c| expr::interpolate(&c.group, &scope).ok());
658 let cancel_in_progress = new
659 .workflow
660 .concurrency
661 .as_ref()
662 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
663 .is_some_and(|value| expr::truthy(&value));
664
Merge branch 'worktree-agent-a3abfcce648e87dca'665 let approval = new.approval.as_ref().map(|reason| RunApproval { state: "required".into(), reason: reason.clone(), approved_by: None });
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow666 let inserted = self
667 .db
668 .prepare(
669 "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'670 pull, status, actor, actor_id, source, info, inputs, trusted, concurrency_group, event_key, created_at,
671 approval, cancel_in_progress)
672 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow673 )
674 .bind(&[
675 id.as_str().into(),
676 workflow_row.id.as_str().into(),
677 new.repo.id.as_str().into(),
678 format!("{}/{}", new.repo.namespace, new.repo.name).into(),
679 new.path.as_str().into(),
680 info.workflow.as_str().into(),
681 title.as_str().into(),
682 numbered.into(),
683 info.event_name.as_str().into(),
684 optional(new.action.as_deref()),
685 info.git_ref.as_str().into(),
A pull request's `pull_request` runs build the pull request merged into its base branch, as on GitHub. g1t keeps `refs/pull/<number>/head` at every open pull request's head (a fork's or working copy's head commit is copied into the repository) and `refs/pull/<number>/merge` at a test merge commit of that head into the base branch's latest commit, moved whenever the head is pushed to, the pull request is opened or reopened, or its base changes; both can be fetched from the repository. A run's `github.sha` and `GITHUB_SHA` are that merge commit, `github.event.pull_request.merge_commit_sha` names it, `actions/checkout` fetches it as `+<sha>:refs/remotes/pull/<number>/merge` from the repository itself, and the workflow files and local reusable workflows are read from it; the run, its statuses and its check runs stay on the head, so required checks hold merges as before, and `pull_request_target` is unchanged. A pull request with conflicts starts no `pull_request` workflows, as on GitHub, and its checks box says why. g1t makes the merge without a sandbox, merging a file both sides changed line by line as git does; what it cannot merge itself (a rename on one side against a change on the other, more than 300 changed files) builds the head as before. The Actions and Checks guides say so.686 new.head_sha.as_deref().unwrap_or(&info.sha).into(),
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow687 new.pull.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
Merge branch 'worktree-agent-a3abfcce648e87dca'688 if approval.is_some() { "action_required" } else { "queued" }.into(),
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow689 optional(new.actor.as_deref()),
690 optional(new.actor_id.as_deref()),
691 new.source.as_str().into(),
692 serde_json::to_string(&info)?.into(),
693 serde_json::to_string(&new.inputs)?.into(),
694 u32::from(new.trusted).into(),
695 optional(group.as_deref()),
696 new.event_key.as_str().into(),
697 now().into(),
Merge branch 'worktree-agent-a3abfcce648e87dca'698 approval.as_ref().map(serde_json::to_string).transpose()?.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
699 u32::from(cancel_in_progress).into(),
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow700 ])?
701 .first::<Value>(None)
702 .await?;
703 if inserted.is_none() {
704 return Ok(None);
705 }
706 if let Some(run) = self.run_row(&id).await? {
707 self.report_pending(&run).await?;
708 }
709
710 // Every job, waiting; each is expanded when the jobs it needs are done.
711 let mut statements = Vec::new();
712 for job in &new.workflow.jobs {
713 statements.push(
714 self.db
715 .prepare(
716 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting')",
717 )
718 .bind(&[
719 new_id("job", now_ms()).into(),
720 id.as_str().into(),
721 new.repo.id.as_str().into(),
722 new.repo.namespace.as_str().into(),
723 job.id.as_str().into(),
724 job.name.clone().filter(|n| !expr::has_expression(n)).unwrap_or(job.id.clone()).into(),
725 serde_json::to_string(&job.needs)?.into(),
726 ])?,
727 );
728 }
729 self.db.batch(statements).await?;
730
Merge branch 'worktree-agent-a3abfcce648e87dca'731 // A pull request's run from outside waits to be approved; it joins
732 // its concurrency group once it is (protection.rs, approve_run).
733 if approval.is_some() {
734 return Ok(Some(id));
735 }
736 if let Some(run) = self.run_row(&id).await? {
737 self.enter_group(&run).await?;
738 }
739 Ok(Some(id))
740 }
741
742 /// A run about to start: one run at a time per concurrency group, as on
743 /// GitHub. A newer run replaces one that waits in the group, and with
744 /// `cancel-in-progress` the one that runs; otherwise it waits as
745 /// `pending` for the one that runs. Then it moves along.
746 pub(crate) async fn enter_group(&self, run: &RunRow) -> Result<()> {
747 if let Some(group) = &run.concurrency_group {
748 let cancel_in_progress = run.cancel_in_progress != 0;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow749 let others = self
750 .db
Merge branch 'worktree-agent-a3abfcce648e87dca'751 .prepare(
752 "SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND id != ? AND status NOT IN ('completed', 'action_required') ORDER BY id",
753 )
754 .bind(&[run.repo_id.as_str().into(), group.as_str().into(), run.id.as_str().into()])?
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow755 .all()
756 .await?
757 .results::<RunRow>()?;
758 for other in &others {
759 if cancel_in_progress || other.status == "pending" {
760 // 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)761 self.cancel_run(other, "A newer run in the same concurrency group replaced it.", false).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow762 }
763 }
764 if !cancel_in_progress && others.iter().any(|other| other.status != "pending") {
765 self.db
766 .prepare("UPDATE runs SET status = 'pending' WHERE id = ?")
Merge branch 'worktree-agent-a3abfcce648e87dca'767 .bind(&[run.id.as_str().into()])?
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow768 .run()
769 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'770 return Ok(());
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow771 }
772 }
Merge branch 'worktree-agent-a3abfcce648e87dca'773 self.advance(&run.id).await
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow774 }
775
776 /// A run that could not start, such as for a workflow file that does not read.
777 #[allow(clippy::too_many_arguments)]
778 pub async fn record_failed_run(
779 &self,
780 row: &WorkflowRow,
781 git_ref: &str,
782 sha: &str,
783 event_key: &str,
784 actor_id: Option<&str>,
785 actor: &str,
786 problem: &str,
787 ) -> Result<()> {
788 let numbered = self
789 .db
790 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
791 .bind(&[row.id.as_str().into()])?
792 .first::<Count>(None)
793 .await?
794 .map_or(1, |count| count.n);
795 let at = now();
796 self.db
797 .prepare(
798 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, git_ref, sha, status,
799 conclusion, error, actor, actor_id, source, info, event_key, created_at, finished_at)
800 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'push', ?, ?, 'completed', 'failure', ?, ?, ?, ?, '{}', ?, ?, ?)",
801 )
802 .bind(&[
803 new_id("run", now_ms()).into(),
804 row.id.as_str().into(),
805 row.repo_id.as_str().into(),
806 row.repo.as_str().into(),
807 row.path.as_str().into(),
808 row.path.as_str().into(),
809 "Invalid workflow file".into(),
810 numbered.into(),
811 git_ref.into(),
812 sha.into(),
813 problem.into(),
814 actor.into(),
815 optional(actor_id),
816 row.source.as_str().into(),
817 event_key.into(),
818 at.as_str().into(),
819 at.as_str().into(),
820 ])?
821 .run()
822 .await?;
823 Ok(())
824 }
825
826 /// Moves a run along: jobs whose needs are done are decided on, and
827 /// jobs that can start are started.
828 pub async fn advance(&self, run_id: &str) -> Result<()> {
829 // Each pass may finish jobs (skipped ones), which may free others.
830 for _ in 0..20 {
831 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'832 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow833 return Ok(());
834 }
835 let workflow = match workflow::parse(&run.source) {
836 Ok(workflow) => workflow,
837 Err(problem) => {
838 self.finish_run(&run, Some(&problem)).await?;
839 return Ok(());
840 }
841 };
842 let jobs = self.job_rows(run_id).await?;
843 let mut changed = false;
844 // What to decide on: the workflow's jobs, and the jobs of the
845 // workflows they call, each with its needs as (name, key).
846 let mut units: Vec<Unit> = workflow
847 .jobs
848 .iter()
849 .map(|job| (job.id.clone(), job.clone(), job.needs.iter().map(|n| (n.clone(), n.clone())).collect()))
850 .collect();
851 let mut seen = std::collections::HashSet::new();
852 for row in &jobs {
853 if !seen.insert(row.key.clone()) {
854 continue;
855 }
856 if let Some((_, job, call)) = row.callee() {
857 let parent = call["parent"].as_str().unwrap_or_default().to_owned();
858 let needs = job.needs.iter().map(|n| (n.clone(), format!("{parent}/{n}"))).collect();
859 units.push((row.key.clone(), job, needs));
860 }
861 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents862 // Each key's needs, as keys, to look back through for failure().
863 let needs_of: std::collections::HashMap<&str, Vec<&str>> =
864 units.iter().map(|(key, _, needs)| (key.as_str(), needs.iter().map(|(_, need)| need.as_str()).collect())).collect();
865 let key_failed = |key: &str| key_result(&jobs.iter().filter(|row| row.key == key).collect::<Vec<_>>()) == "failure";
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow866 for (key, job, needs) in &units {
867 let rows: Vec<&JobRow> = jobs.iter().filter(|row| &row.key == key).collect();
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9868 if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting" && !row.holds_call()) {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow869 continue;
870 }
871 let needed: Vec<(&String, Vec<&JobRow>)> =
872 needs.iter().map(|(name, need)| (name, jobs.iter().filter(|row| &row.key == need).collect())).collect();
873 if !needed.iter().all(|(_, rows)| rows.iter().all(|row| row.status == "completed")) {
874 continue;
875 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents876 let failed_before = ancestor_failed(&needs_of, key, key_failed);
877 self.decide(&run, job, rows[0], &needed, failed_before).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow878 changed = true;
879 }
880 // A job that called a workflow finishes with that workflow's jobs.
881 for row in jobs.iter().filter(|row| row.status == "calling") {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9882 let prefix = row.call_prefix().unwrap_or(row.key.clone());
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow883 let children: Vec<&JobRow> = jobs
884 .iter()
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9885 .filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(prefix.as_str())))
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow886 .collect();
887 if !children.is_empty() && children.iter().all(|child| child.status == "completed") {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9888 self.finish_call(&run, row, &children).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow889 changed = true;
890 }
891 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9892 // Calls that wait start while their matrix's `max-parallel` and
893 // their concurrency group have room.
894 let mut calling: std::collections::HashMap<&str, u32> = std::collections::HashMap::new();
895 for row in jobs.iter().filter(|row| row.status == "calling") {
896 *calling.entry(row.key.as_str()).or_default() += 1;
897 }
898 for row in jobs.iter().filter(|row| row.holds_call()) {
899 let running = calling.get(row.key.as_str()).copied().unwrap_or(0);
900 if row.max_parallel.is_some_and(|max| running >= max) || self.group_busy(row).await? {
901 continue;
902 }
903 if self.start_call(row).await? {
904 calling.insert(row.key.as_str(), running + 1);
905 changed = true;
906 }
907 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow908 if !changed {
909 break;
910 }
911 }
912 self.start_queued().await?;
913 self.finish_if_done(run_id).await
914 }
915
916 /// 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 incidents917 /// it into its matrix and queue it. `failed_before`: whether any job
918 /// before it failed, however far back (`ancestor_failed`).
919 async fn decide(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, needed: &[(&String, Vec<&JobRow>)], failed_before: bool) -> Result<()> {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow920 let vars = self.variables_for(&run.repo_id, &run.repo, None, run.trusted != 0).await?;
921 let mut contexts = Self::base_contexts(run, &vars, &job.id);
922 // A called workflow's jobs read the inputs they were called with.
923 let call = row.call();
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9924 // Shown under the job whose call made it, by that job's name.
925 let caller = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| {
926 c["title"].as_str().map(str::to_owned).or_else(|| c["parent"].as_str().map(|parent| parent.replace('/', " / ")))
927 });
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow928 if let Some(call) = call.as_ref().filter(|c| c["role"] == "callee") {
929 contexts.insert("inputs".into(), call["inputs"].clone());
930 }
931 let mut needs = Map::new();
Fast pages, required checks on the branch, self-hosted runners, honest incidents932 let mut results = Vec::new();
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow933 for (key, rows) in needed {
934 let result = key_result(rows);
935 let mut outputs = Map::new();
936 for row in rows {
937 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
938 outputs.extend(more);
939 }
940 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents941 results.push(result);
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow942 needs.insert((*key).clone(), json!({ "result": result, "outputs": outputs }));
943 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents944 // As on GitHub: a need that was skipped makes success() false but
945 // not failure(); failure() is a failure anywhere before the job.
946 let status = expr::job_status(results, failed_before, run.conclusion.as_deref() == Some("cancelled"));
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow947 contexts.insert("needs".into(), Value::Object(needs));
948 let scope = Scope {
949 contexts: &contexts,
950 status,
951 hash_files: None,
952 };
953 let condition = job.condition.as_deref().unwrap_or_default();
954 match expr::condition(condition, &scope) {
955 Ok(true) => {}
956 Ok(false) => return self.skip_job(row, None).await,
957 Err(problem) => return self.fail_job(row, &format!("Its `if` does not read: {problem}")).await,
958 }
959 // Its matrix, which may come from a needed job's outputs.
960 let combinations = match &job.matrix {
961 None => vec![Map::new()],
962 Some(matrix) => {
963 let value = match expr::interpolate_value(matrix, &scope) {
964 Ok(value) => value,
965 Err(problem) => return self.fail_job(row, &format!("Its matrix does not read: {problem}")).await,
966 };
967 match matrix::expand(&value) {
968 Ok(combinations) if !combinations.is_empty() => combinations,
969 Ok(_) => return self.fail_job(row, "Its matrix makes no jobs.").await,
970 Err(problem) => return self.fail_job(row, &problem).await,
971 }
972 }
973 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9974 // A job that calls a reusable workflow calls it once for each.
975 if let Some(uses) = &job.uses {
976 return self.call_workflow(run, job, row, uses, &contexts, &combinations, caller.as_deref()).await;
977 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow978 let total = combinations.len();
979 let raw = job.raw.as_object().cloned().unwrap_or_default();
Fast pages, required checks on the branch, self-hosted runners, honest incidents980 // Whether pull requests from forks may use self-hosted runners here,
981 // asked once, and only for a run that is not trusted.
982 let mut forks_allowed: Option<bool> = None;
Merge branch 'worktree-agent-a3abfcce648e87dca'983 // Each concurrency group its jobs join, and whether it cancels.
984 let mut groups: Vec<(String, bool)> = Vec::new();
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow985 let mut statements = Vec::new();
986 for (index, combination) in combinations.iter().enumerate() {
987 let mut contexts = contexts.clone();
988 contexts.insert("matrix".into(), Value::Object(combination.clone()));
989 contexts.insert(
990 "strategy".into(),
991 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
992 );
993 let scope = Scope {
994 contexts: &contexts,
995 status: Status::Success,
996 hash_files: None,
997 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9998 let name = shown_name(job, caller.as_deref(), combination, &scope);
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow999 let runs_on = expr::interpolate_value(&job.runs_on, &scope).unwrap_or(Value::Null);
1000 let labels = match &runs_on {
1001 Value::String(label) => vec![label.clone()],
1002 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
1003 Value::Object(spec) => spec.get("labels").map(|l| match l {
1004 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
1005 other => vec![expr::to_text(other)],
1006 }).unwrap_or_default(),
1007 _ => Vec::new(),
1008 };
Fast pages, required checks on the branch, self-hosted runners, honest incidents1009 // `runs-on: self-hosted` (or a group): the workspace's own
1010 // machines, which may run any OS. Otherwise g1t's Linux sandboxes.
1011 let group = match &runs_on {
1012 Value::Object(spec) => spec.get("group").map(expr::to_text),
1013 _ => None,
1014 };
1015 let wanted = Wanted::of(&labels, group.as_deref());
1016 let mut reason = if wanted.self_hosted {
1017 None
1018 } else {
1019 labels
1020 .iter()
1021 .find(|label| {
1022 let lower = label.to_ascii_lowercase();
1023 lower.contains("windows") || lower.contains("macos")
1024 })
1025 .map(|label| {
1026 let os = if label.to_ascii_lowercase().contains("windows") { "windows" } else { "macos" };
1027 format!("`runs-on: {label}`: g1t's own runners are Linux. A self-hosted runner can run it: `runs-on: [self-hosted, {os}]`.")
1028 })
1029 };
1030 // A pull request from a fork runs code anyone could write: never
1031 // on the workspace's machines unless it said they may.
1032 if wanted.self_hosted && run.trusted == 0 {
1033 let allowed = match forks_allowed {
1034 Some(allowed) => allowed,
1035 None => {
1036 let allowed = self.effective_runner_settings(&row.namespace, Some(&run.repo_id)).await?.fork_pull_requests;
1037 forks_allowed = Some(allowed);
1038 allowed
1039 }
1040 };
1041 if !allowed {
1042 reason = Some(
1043 "Pull requests from forks do not run on self-hosted runners here. An admin can allow it under Settings, Runners.".to_owned(),
1044 );
1045 }
1046 }
1047 let max_minutes = if wanted.self_hosted { SELF_HOSTED_MAX_TIMEOUT_MINUTES } else { MAX_TIMEOUT_MINUTES };
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1048 let timeout = raw
1049 .get("timeout-minutes")
1050 .and_then(|value| expr::interpolate_value(value, &scope).ok())
1051 .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.1052 .map_or(g1t_actions::workflow::DEFAULT_TIMEOUT_MINUTES.min(max_minutes), |minutes| (minutes.ceil() as u32).clamp(1, max_minutes));
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1053 let continue_on_error = raw
1054 .get("continue-on-error")
1055 .and_then(|value| expr::interpolate_value(value, &scope).ok())
1056 .is_some_and(|value| expr::truthy(&value));
Merge branch 'worktree-agent-a3abfcce648e87dca'1057 let (mut status, mut conclusion, mut finished) = match &reason {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1058 Some(_) => ("completed", Some("failure"), Some(now())),
1059 None => ("queued", None, None),
1060 };
Fast pages, required checks on the branch, self-hosted runners, honest incidents1061 // A self-hosted job waits, saying for what, until a runner takes it.
1062 let (labels_json, queued_at) = if wanted.self_hosted && reason.is_none() {
1063 reason = Some(waiting_reason(&wanted));
1064 (Some(serde_json::to_string(&wanted.stored())?), Some(now()))
1065 } else {
1066 (None, None)
1067 };
Merge branch 'worktree-agent-a3abfcce648e87dca'1068 // The environment it names, an expression read now, and what
1069 // that environment's protection rules say: it starts, waits as
1070 // `pending` (protection.rs), or may not deploy there.
1071 let environment = environment_of(&job.raw, &contexts).map(|env| env.name);
1072 if status == "queued"
1073 && let Some(name) = &environment
1074 {
1075 match self.gate(run, name).await? {
1076 Gate::Open => {}
1077 Gate::Held(why) => {
1078 status = "pending";
1079 reason = Some(why);
1080 }
1081 Gate::Refused(why) => {
1082 (status, conclusion, finished) = ("completed", Some("failure"), Some(now()));
1083 reason = Some(why);
1084 }
1085 }
1086 }
1087 // Its own concurrency group.
1088 let job_group = job
1089 .concurrency
1090 .as_ref()
1091 .and_then(|c| expr::interpolate(&c.group, &scope).ok())
1092 .map(|group| group.trim().to_owned())
1093 .filter(|group| !group.is_empty());
1094 let job_cancel = job
1095 .concurrency
1096 .as_ref()
1097 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
1098 .is_some_and(|value| expr::truthy(&value));
1099 if status != "completed"
1100 && let Some(group) = &job_group
1101 && !groups.iter().any(|(known, _)| known == group)
1102 {
1103 groups.push((group.clone(), job_cancel));
1104 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1105 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
1106 name.into(),
1107 serde_json::to_string(combination)?.into(),
1108 status.into(),
1109 optional(conclusion),
1110 optional(reason.as_deref()),
1111 timeout.into(),
1112 u32::from(continue_on_error).into(),
1113 job.max_parallel.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1114 optional(finished.as_deref()),
Fast pages, required checks on the branch, self-hosted runners, honest incidents1115 optional(labels_json.as_deref()),
1116 optional(queued_at.as_deref()),
Merge branch 'worktree-agent-a3abfcce648e87dca'1117 optional(environment.as_deref()),
1118 optional(job_group.as_deref()),
1119 u32::from(job_cancel).into(),
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1120 ];
1121 if index == 0 {
1122 let mut bound = values;
1123 bound.push(row.id.as_str().into());
1124 statements.push(
1125 self.db
1126 .prepare(
1127 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, timeout_minutes = ?,
Merge branch 'worktree-agent-a3abfcce648e87dca'1128 continue_on_error = ?, max_parallel = ?, finished_at = ?, labels = ?, queued_at = ?, environment = ?,
1129 concurrency_group = ?, cancel_in_progress = ? WHERE id = ?",
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1130 )
1131 .bind(&bound)?,
1132 );
1133 } else {
1134 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
1135 new_id("job", now_ms()).into(),
1136 row.run_id.as_str().into(),
1137 row.repo_id.as_str().into(),
1138 row.namespace.as_str().into(),
1139 row.key.as_str().into(),
1140 (index as u32).into(),
1141 row.needs.as_str().into(),
1142 optional(row.call.as_deref()),
1143 ];
1144 bound.extend(values);
1145 statements.push(
1146 self.db
1147 .prepare(
1148 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, call, name, matrix, status, conclusion, reason,
Merge branch 'worktree-agent-a3abfcce648e87dca'1149 timeout_minutes, continue_on_error, max_parallel, finished_at, labels, queued_at, environment,
1150 concurrency_group, cancel_in_progress)
1151 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1152 )
1153 .bind(&bound)?,
1154 );
1155 }
1156 }
1157 self.db.batch(statements).await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1158 self.replace_in_groups(run, &row.key, &groups).await
1159 }
1160
1161 /// A job queued in its own concurrency group: with `cancel-in-progress`
1162 /// it cancels the group's other jobs; otherwise it replaces any that
1163 /// are queued and not started, and waits for the one running
1164 /// (`start_queued` starts one of a group at a time). As on GitHub.
1165 async fn replace_in_groups(&self, run: &RunRow, key: &str, groups: &[(String, bool)]) -> Result<()> {
1166 if groups.is_empty() {
1167 return Ok(());
1168 }
1169 #[derive(Deserialize)]
1170 struct Id {
1171 id: String,
1172 }
1173 let mine: Vec<String> = self
1174 .db
1175 .prepare("SELECT id FROM jobs WHERE run_id = ? AND key = ?")
1176 .bind(&[run.id.as_str().into(), key.into()])?
1177 .all()
1178 .await?
1179 .results::<Id>()?
1180 .into_iter()
1181 .map(|row| row.id)
1182 .collect();
1183 let mut moved: Vec<String> = Vec::new();
1184 for (group, cancel) in groups {
1185 let others = self
1186 .db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91187 // A call that waits in the group, or calls, is in it too.
1188 .prepare("SELECT * FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('waiting', 'queued', 'pending', 'in_progress', 'calling')")
Merge branch 'worktree-agent-a3abfcce648e87dca'1189 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1190 .all()
1191 .await?
1192 .results::<JobRow>()?;
1193 for other in others.iter().filter(|other| !mine.contains(&other.id)) {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91194 if *cancel || matches!(other.status.as_str(), "queued" | "waiting") {
Merge branch 'worktree-agent-a3abfcce648e87dca'1195 self.stop_job(other, "A newer job in the same concurrency group replaced it.").await?;
1196 if !moved.contains(&other.run_id) {
1197 moved.push(other.run_id.clone());
1198 }
1199 }
1200 }
1201 }
1202 for other_run in moved.iter().filter(|id| **id != run.id) {
1203 Box::pin(self.advance(other_run)).await?;
1204 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1205 Ok(())
1206 }
1207
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1208 /// A job that calls a reusable workflow, in the repository or another
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91209 /// (reach.rs): once for each combination of its matrix, as on GitHub,
1210 /// each with the inputs and secrets it passes read with that
1211 /// combination's `matrix`. Each call waits (`holds_call`) until its
1212 /// matrix's `max-parallel` and its concurrency group have room, when
1213 /// `start_call` puts the workflow's jobs in the run under it.
1214 #[allow(clippy::too_many_arguments)]
1215 async fn call_workflow(
1216 &self,
1217 run: &RunRow,
1218 job: &workflow::Job,
1219 row: &JobRow,
1220 uses: &str,
1221 contexts: &Map<String, Value>,
1222 combinations: &[Map<String, Value>],
1223 caller: Option<&str>,
1224 ) -> Result<()> {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1225 let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1;
1226 if depth > MAX_CALL_DEPTH {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91227 return self.fail_job(row, &call_allowed(depth, &[], uses).unwrap_err()).await;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1228 }
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1229 let (file, source, origin) = match self.called_workflow(run, row, uses).await? {
1230 Ok(found) => found,
1231 Err(why) => return self.fail_job(row, &why).await,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1232 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91233 let jobs = self.job_rows(&run.id).await?;
1234 // Each call so far, by the workflow it called.
1235 let called_paths: Vec<String> = jobs
1236 .iter()
1237 .filter(|other| other.call_prefix().is_some())
1238 .filter_map(JobRow::call)
1239 .filter_map(|call| call["path"].as_str().map(str::to_owned))
1240 .collect();
1241 let called_already: Vec<&str> = called_paths.iter().map(String::as_str).collect();
1242 if let Err(why) = call_allowed(depth, &called_already, &file) {
1243 return self.fail_job(row, &why).await;
1244 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1245 let called = match workflow::parse(&source) {
1246 Ok(called) => called,
1247 Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await,
1248 };
1249 let Some(trigger) = called.trigger("workflow_call") else {
1250 return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await;
1251 };
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1252 // Secrets: none but the job's token unless `secrets:` passes them,
1253 // by name or with `inherit`; read when each job starts (job_spec).
1254 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-7nm8r91255 let total = combinations.len();
1256 let mut groups: Vec<(String, bool)> = Vec::new();
1257 let mut statements = Vec::new();
1258 for (index, combination) in combinations.iter().enumerate() {
1259 let mut contexts = contexts.clone();
1260 contexts.insert("matrix".into(), Value::Object(combination.clone()));
1261 contexts.insert(
1262 "strategy".into(),
1263 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
1264 );
1265 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1266 let name = shown_name(job, caller, combination, &scope);
1267 // Inputs: what `with:` passes, as the called workflow types them.
1268 let given = match job.raw.get("with") {
1269 Some(with) => match expr::interpolate_value(with, &scope) {
1270 Ok(Value::Object(given)) => Ok(given),
1271 Ok(_) => Ok(Map::new()),
1272 Err(problem) => Err(format!("Its `with` does not read: {problem}")),
1273 },
1274 None => Ok(Map::new()),
1275 };
1276 let inputs = given.and_then(|given| crate::reach::call_inputs(&file, &trigger.inputs, &given));
1277 let secrets = crate::reach::secrets_plan(&job.raw, &contexts, outer.clone());
1278 let problem = match &inputs {
1279 Err(why) => Some(why.clone()),
1280 Ok(_) => crate::reach::missing_secrets(&called.raw, &secrets)
1281 .first()
1282 .map(|name| format!("`{file}` needs the secret `{name}`: pass it under `secrets:`, or use `secrets: inherit`."))
1283 .or_else(|| {
1284 crate::reach::undeclared_secrets(&called.raw, &secrets)
1285 .first()
1286 .map(|name| format!("Invalid secret: `{name}` is not defined in `{file}`."))
1287 }),
1288 };
1289 // Its own concurrency group, which it holds while its workflow runs.
1290 let group = job
1291 .concurrency
1292 .as_ref()
1293 .and_then(|c| expr::interpolate(&c.group, &scope).ok())
1294 .map(|group| group.trim().to_owned())
1295 .filter(|group| !group.is_empty());
1296 let cancel = job
1297 .concurrency
1298 .as_ref()
1299 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
1300 .is_some_and(|value| expr::truthy(&value));
1301 if problem.is_none()
1302 && let Some(group) = &group
1303 && !groups.iter().any(|(known, _)| known == group)
1304 {
1305 groups.push((group.clone(), cancel));
1306 }
1307 let call = json!({
1308 "role": "caller", "path": file, "source": source, "origin": origin, "depth": depth,
1309 "prefix": call_prefix(&row.key, job.matrix.is_some().then_some(index)), "title": name,
1310 "inputs": inputs.unwrap_or_default(), "secrets": secrets, "fail_fast": job.fail_fast,
1311 });
1312 let (status, conclusion, finished) = match &problem {
1313 Some(_) => ("completed", Some("failure"), Some(now())),
1314 None => ("waiting", None, None),
1315 };
1316 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
1317 name.as_str().into(),
1318 serde_json::to_string(combination)?.into(),
1319 status.into(),
1320 optional(conclusion),
1321 optional(problem.as_deref()),
1322 job.max_parallel.map(|max| max.max(1)).map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1323 optional(finished.as_deref()),
1324 optional(group.as_deref()),
1325 u32::from(cancel).into(),
1326 serde_json::to_string(&call)?.into(),
1327 ];
1328 if index == 0 {
1329 let mut bound = values;
1330 bound.push(row.id.as_str().into());
1331 statements.push(
1332 self.db
1333 .prepare(
1334 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, max_parallel = ?, finished_at = ?,
1335 concurrency_group = ?, cancel_in_progress = ?, call = ? WHERE id = ?",
1336 )
1337 .bind(&bound)?,
1338 );
1339 } else {
1340 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
1341 new_id("job", now_ms()).into(),
1342 row.run_id.as_str().into(),
1343 row.repo_id.as_str().into(),
1344 row.namespace.as_str().into(),
1345 row.key.as_str().into(),
1346 (index as u32).into(),
1347 row.needs.as_str().into(),
1348 ];
1349 bound.extend(values);
1350 statements.push(
1351 self.db
1352 .prepare(
1353 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, name, matrix, status, conclusion, reason,
1354 max_parallel, finished_at, concurrency_group, cancel_in_progress, call)
1355 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
1356 )
1357 .bind(&bound)?,
1358 );
1359 }
1360 }
1361 self.db.batch(statements).await?;
1362 self.replace_in_groups(run, &row.key, &groups).await
1363 }
1364
1365 /// Whether another job of `row`'s concurrency group runs, or calls a
1366 /// workflow, so that `row` waits for it.
1367 async fn group_busy(&self, row: &JobRow) -> Result<bool> {
1368 let Some(group) = &row.concurrency_group else { return Ok(false) };
1369 let busy = self
1370 .db
1371 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?")
1372 .bind(&[row.repo_id.as_str().into(), group.as_str().into(), row.id.as_str().into()])?
1373 .first::<Count>(None)
1374 .await?
1375 .map_or(0, |count| count.n);
1376 Ok(busy > 0)
1377 }
1378
1379 /// A call that waited (`holds_call`) is made: the called workflow's
1380 /// jobs join the run under it, named under its name and keyed under its
1381 /// prefix. False when it no longer waits.
1382 async fn start_call(&self, row: &JobRow) -> Result<bool> {
1383 let Some(call) = row.call() else { return Ok(false) };
1384 let Some(prefix) = row.call_prefix() else { return Ok(false) };
1385 let Some(called) = call["source"].as_str().and_then(|source| workflow::parse(source).ok()) else { return Ok(false) };
1386 let file = call["path"].as_str().unwrap_or_default();
1387 let started = self
1388 .db
1389 .prepare("UPDATE jobs SET status = 'calling', reason = ?, started_at = ? WHERE id = ? AND status = 'waiting' RETURNING id")
1390 .bind(&[format!("Calls `{file}`.").into(), now().into(), row.id.as_str().into()])?
1391 .first::<Value>(None)
1392 .await?;
1393 if started.is_none() {
1394 return Ok(false);
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1395 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1396 let mut statements = Vec::new();
1397 for called_job in &called.jobs {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91398 let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{prefix}/{n}")).collect();
1399 let callee = json!({
1400 "role": "callee", "parent": prefix, "title": row.name, "job": called_job.id, "path": file,
1401 "source": call["source"], "inputs": call["inputs"], "depth": call["depth"], "origin": call["origin"], "secrets": call["secrets"],
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1402 });
1403 statements.push(
1404 self.db
1405 .prepare("INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status, call) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting', ?)")
1406 .bind(&[
1407 new_id("job", now_ms()).into(),
1408 row.run_id.as_str().into(),
1409 row.repo_id.as_str().into(),
1410 row.namespace.as_str().into(),
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91411 format!("{prefix}/{}", called_job.id).into(),
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1412 format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(),
1413 serde_json::to_string(&needs)?.into(),
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91414 serde_json::to_string(&callee)?.into(),
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1415 ])?,
1416 );
1417 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91418 if !statements.is_empty() {
1419 self.db.batch(statements).await?;
1420 }
1421 Ok(true)
1422 }
1423
1424 /// The runs with a call waiting in `group`, but `run_id`, moved along
1425 /// now that a job of the group has finished.
1426 async fn wake_group(&self, repo_id: &str, group: &str, run_id: &str) -> Result<()> {
1427 #[derive(Deserialize)]
1428 struct RunId {
1429 run_id: String,
1430 }
1431 let waiting = self
1432 .db
1433 .prepare("SELECT DISTINCT run_id FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'waiting' AND run_id != ?")
1434 .bind(&[repo_id.into(), group.into(), run_id.into()])?
1435 .all()
1436 .await?
1437 .results::<RunId>()?;
1438 for other in waiting {
1439 Box::pin(self.advance(&other.run_id)).await?;
1440 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1441 Ok(())
1442 }
1443
1444 /// A job that called a workflow, finished with its jobs: their result,
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91445 /// and the outputs the workflow declares. A failed call of a fail-fast
1446 /// matrix stops the matrix's other calls.
1447 async fn finish_call(&self, run: &RunRow, row: &JobRow, children: &[&JobRow]) -> Result<()> {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1448 let call = row.call().unwrap_or_default();
1449 let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok());
1450 let mut jobs_context = Map::new();
1451 let mut by_key: std::collections::BTreeMap<String, Vec<&JobRow>> = std::collections::BTreeMap::new();
1452 for child in children {
1453 by_key.entry(child.key.clone()).or_default().push(child);
1454 }
1455 for (key, rows) in &by_key {
1456 let mut outputs = Map::new();
1457 for child in rows {
1458 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&child.outputs) {
1459 outputs.extend(more);
1460 }
1461 }
1462 let id = key.rsplit('/').next().unwrap_or(key);
1463 jobs_context.insert(id.to_owned(), json!({ "result": key_result(rows), "outputs": outputs }));
1464 }
1465 let inputs = children.first().and_then(|c| c.call()).map(|c| c["inputs"].clone()).unwrap_or(json!({}));
1466 let mut contexts = Map::new();
1467 contexts.insert("jobs".into(), Value::Object(jobs_context));
1468 contexts.insert("inputs".into(), inputs);
1469 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1470 let mut outputs = Map::new();
1471 if let Some(Value::Object(declared)) = called.as_ref().map(|w| {
1472 let on = w.raw.get("on").or_else(|| w.raw.get("true")).cloned().unwrap_or(Value::Null);
1473 on.get("workflow_call").and_then(|c| c.get("outputs")).cloned().unwrap_or(Value::Null)
1474 }) {
1475 for (name, spec) in declared {
1476 if let Some(value) = spec.get("value") {
1477 let value = expr::interpolate_value(value, &scope).unwrap_or(Value::Null);
1478 outputs.insert(name, Value::String(expr::to_text(&value)));
1479 }
1480 }
1481 }
1482 let conclusion = key_result(children);
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91483 let finished = self
1484 .db
1485 .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling' RETURNING id")
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1486 .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-7nm8r91487 .first::<Value>(None)
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1488 .await?;
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91489 if finished.is_none() {
1490 return Ok(());
1491 }
1492 if conclusion == "failure" && row.matrix.as_deref().is_some_and(|m| m != "{}") && call["fail_fast"] != json!(false) {
1493 let siblings = self
1494 .db
1495 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1496 .bind(&[row.run_id.as_str().into(), row.key.as_str().into()])?
1497 .all()
1498 .await?
1499 .results::<JobRow>()?;
1500 for sibling in siblings {
1501 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1502 }
1503 }
1504 if let Some(group) = &row.concurrency_group {
1505 self.wake_group(&run.repo_id, group, &run.id).await?;
1506 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1507 Ok(())
1508 }
1509
1510 /// A file's text at a commit, if it is there.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1511 pub(crate) async fn read_file(&self, path: &RepoPath, ws: &g1t_contracts::User, sha: &str, file: &str) -> Result<Option<String>> {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1512 let blob: Outcome<g1t_contracts::repos::BlobView> = g1t_kit::call(
1513 &self.repos,
1514 "blob",
1515 &g1t_contracts::repos::BlobArgs {
1516 path: path.clone(),
1517 viewer: Some(ws.clone()),
1518 git_ref: sha.to_owned(),
1519 file_path: file.to_owned(),
1520 },
1521 )
1522 .await?;
1523 Ok(match blob {
1524 Outcome::Ok(view) => view.text,
1525 Outcome::Fail(_) => None,
1526 })
1527 }
1528
1529 async fn skip_job(&self, row: &JobRow, reason: Option<&str>) -> Result<()> {
1530 self.db
1531 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'skipped', reason = ?, finished_at = ? WHERE id = ?")
1532 .bind(&[optional(reason), now().into(), row.id.as_str().into()])?
1533 .run()
1534 .await?;
1535 Ok(())
1536 }
1537
1538 async fn fail_job(&self, row: &JobRow, reason: &str) -> Result<()> {
1539 self.db
1540 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'failure', reason = ?, finished_at = ? WHERE id = ? AND status != 'completed'")
1541 .bind(&[reason.into(), now().into(), row.id.as_str().into()])?
1542 .run()
1543 .await?;
1544 Ok(())
1545 }
1546
1547 /// Starts queued jobs, oldest first, while their workspace has room.
1548 pub async fn start_queued(&self) -> Result<()> {
1549 let queued = self
1550 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1551 // Self-hosted jobs are taken by their runners (runners.rs).
1552 .prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NULL ORDER BY rowid LIMIT 50")
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1553 .all()
1554 .await?
1555 .results::<JobRow>()?;
1556 let mut running: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
1557 for job in queued {
1558 let in_workspace = match running.get(&job.namespace) {
1559 Some(n) => *n,
1560 None => {
1561 let n = self
1562 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1563 // Only g1t's own sandboxes count against the workspace's room.
1564 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE status = 'in_progress' AND namespace = ? AND runner_id IS NULL")
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1565 .bind(&[job.namespace.as_str().into()])?
1566 .first::<Count>(None)
1567 .await?
1568 .map_or(0, |count| count.n);
1569 running.insert(job.namespace.clone(), n);
1570 n
1571 }
1572 };
1573 if in_workspace >= RUNNING_PER_WORKSPACE {
1574 continue;
1575 }
1576 if let Some(max) = job.max_parallel {
1577 let siblings = self
1578 .db
1579 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")
1580 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1581 .first::<Count>(None)
1582 .await?
1583 .map_or(0, |count| count.n);
1584 if siblings >= max {
1585 continue;
1586 }
1587 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1588 // One job of a concurrency group runs at a time.
1589 if let Some(group) = &job.concurrency_group {
1590 let running = self
1591 .db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91592 .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'1593 .bind(&[job.repo_id.as_str().into(), group.as_str().into(), job.id.as_str().into()])?
1594 .first::<Count>(None)
1595 .await?
1596 .map_or(0, |count| count.n);
1597 if running > 0 {
1598 continue;
1599 }
1600 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1601 let token = random_hex(24);
1602 let at = now();
1603 let claimed = self
1604 .db
1605 .prepare(
1606 "UPDATE jobs SET status = 'in_progress', token_hash = ?, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
1607 )
1608 .bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
1609 .first::<Value>(None)
1610 .await?;
1611 if claimed.is_none() {
1612 continue;
1613 }
1614 running.insert(job.namespace.clone(), in_workspace + 1);
1615 self.db
1616 .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")
1617 .bind(&[at.as_str().into(), job.run_id.as_str().into()])?
1618 .run()
1619 .await?;
1620 let run = self.run_row(&job.run_id).await?;
1621 let repo: RepoPath = run.as_ref().map(|run| repo_path(&run.repo)).unwrap_or(RepoPath {
1622 namespace: job.namespace.clone(),
1623 name: String::new(),
1624 });
Fast pages, required checks on the branch, self-hosted runners, honest incidents1625 // Its environment and the machine it asked for, for the runner.
1626 let details = run.as_ref().map(|run| start_details(run, &job)).unwrap_or_default();
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1627 let started: Outcome<Value> = g1t_kit::call(
1628 &self.runner,
1629 "start_actions_job",
1630 &StartJobArgs {
1631 job: job.id.clone(),
1632 token,
1633 repo,
1634 timeout_minutes: job.timeout_minutes,
Fast pages, required checks on the branch, self-hosted runners, honest incidents1635 workflow: run.as_ref().map(|run| run.path.clone()),
1636 environment: details.environment,
1637 trusted: run.as_ref().is_some_and(|run| run.trusted != 0),
1638 instance: details.instance,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1639 },
1640 )
1641 .await
1642 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
1643 if let Outcome::Fail(refused) = started {
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1644 // Billing would not pay for a scheduled run's job: the
1645 // schedule waits rather than making runs only to refuse them.
1646 if refused.code == FailureCode::PaymentRequired
1647 && let Some(run) = run.as_ref().filter(|run| run.event == "schedule")
1648 {
1649 self.db
1650 .prepare("UPDATE workflows SET schedule_refused_until = ? WHERE id = ?")
1651 .bind(&[rfc3339(now_ms() + crate::SCHEDULE_REFUSED_MS).into(), run.workflow_id.as_str().into()])?
1652 .run()
1653 .await?;
1654 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1655 Box::pin(self.finish_job(&job.id, "failure", Some(&refused.message), None)).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971656 } else {
1657 self.job_started(&job.id).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1658 }
1659 }
1660 Ok(())
1661 }
1662
1663 /// Finishes a job and moves its run along.
1664 pub async fn finish_job(&self, job_id: &str, conclusion: &str, reason: Option<&str>, outputs: Option<&Map<String, Value>>) -> Result<()> {
1665 let finished = self
1666 .db
1667 .prepare(
1668 "UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),
1669 finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
1670 )
1671 .bind(&[
1672 conclusion.into(),
1673 optional(reason),
1674 outputs.map(|o| serde_json::to_string(o).unwrap_or_default()).as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1675 now().into(),
1676 job_id.into(),
1677 ])?
1678 .first::<JobRow>(None)
1679 .await?;
1680 let Some(job) = finished else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1681 // Its G1T_TOKEN stops working with it.
1682 self.revoke_job_tokens(&job.id).await;
Merge branch 'worktree-agent-ab6bf4195fcb97f7c' into claude/cool-newton-7nm8r91683 // A g1t.page build it opened and did not finish fails now, not at
1684 // the deployments sweep. Only a job that ran can have opened one.
1685 if job.started_at.is_some() {
1686 self.end_job_builds(&job.id, conclusion).await;
1687 }
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971688 // A job that deploys failed: so did its run's deployment, now.
1689 if conclusion == "failure" && job.continue_on_error == 0 && job.started_at.is_some()
1690 && let Some(run) = self.run_row(&job.run_id).await?
1691 && let Some(env) = deploys_to(&run, &job)
1692 {
1693 self.report_deployment(&run, &env, "failure", false).await;
1694 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1695 // A self-hosted runner's job: the runner is free again, and its time
1696 // is recorded, at nothing.
1697 if job.runner_id.is_some() {
1698 self.released(&job).await?;
1699 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1700 // Steps still marked as going are not going any more.
1701 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
1702 let mut touched = false;
1703 for step in steps.iter_mut() {
1704 if step["status"] != "completed" {
1705 let was_running = step["status"] == "in_progress";
1706 step["status"] = json!("completed");
1707 step["conclusion"] = json!(if was_running { conclusion } else { "skipped" });
1708 touched = true;
1709 }
1710 }
1711 if touched {
1712 self.db
1713 .prepare("UPDATE jobs SET steps = ? WHERE id = ?")
1714 .bind(&[serde_json::to_string(&steps)?.into(), job.id.as_str().into()])?
1715 .run()
1716 .await?;
1717 }
1718 // fail-fast: one failed combination stops the rest of its matrix.
1719 if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") {
1720 let run = self.run_row(&job.run_id).await?;
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91721 // As its workflow, or the workflow it is a job of, says.
1722 let fail_fast = run.as_ref().and_then(|run| job_spec(run, &job)).is_none_or(|j| j.fail_fast);
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1723 if fail_fast {
1724 let siblings = self
1725 .db
1726 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1727 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1728 .all()
1729 .await?
1730 .results::<JobRow>()?;
1731 for sibling in siblings {
1732 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1733 }
1734 }
1735 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91736 // A call waiting in its concurrency group, in another run, may start.
1737 if let Some(group) = &job.concurrency_group {
1738 self.wake_group(&job.repo_id, group, &job.run_id).await?;
1739 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1740 Box::pin(self.advance(&job.run_id)).await
1741 }
1742
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1743 /// Cancels a job. One that is running is told to stop (the answer to
1744 /// its next report), and runs its `if: always()` and `cancelled()`
1745 /// steps and its post steps before it ends `cancelled`; one that has
1746 /// not started is cancelled at once.
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1747 async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91748 // A job calling a workflow stops with that workflow's jobs.
1749 if job.status == "calling"
1750 && let Some(prefix) = job.call_prefix()
1751 {
1752 let under = format!("{prefix}/");
1753 for child in self.job_rows(&job.run_id).await?.iter().filter(|child| child.key.starts_with(&under) && child.status != "completed") {
1754 Box::pin(self.stop_job(child, reason)).await?;
1755 }
1756 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1757 if job.status == "in_progress" && job.started_at.is_some() {
1758 self.db
1759 .prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'")
1760 .bind(&[now().into(), reason.into(), job.id.as_str().into()])?
1761 .run()
1762 .await?;
1763 return Ok(());
1764 }
1765 self.hard_stop(job, reason).await
1766 }
1767
1768 /// Cancels a job outright, stopping its sandbox if it has one.
1769 async fn hard_stop(&self, job: &JobRow, reason: &str) -> Result<()> {
Fast pages, required checks on the branch, self-hosted runners, honest incidents1770 // A self-hosted runner hears it was cancelled on its next poll.
1771 if job.status == "in_progress" && job.runner_id.is_none() {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1772 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
1773 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1774 let stopped = self
1775 .db
1776 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *")
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1777 .bind(&[reason.into(), now().into(), job.id.as_str().into()])?
Fast pages, required checks on the branch, self-hosted runners, honest incidents1778 .first::<JobRow>(None)
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1779 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1780 if stopped.as_ref().is_some_and(|row| row.started_at.is_some()) {
1781 self.revoke_job_tokens(&job.id).await;
Merge branch 'worktree-agent-ab6bf4195fcb97f7c' into claude/cool-newton-7nm8r91782 self.end_job_builds(&job.id, "cancelled").await;
Merge branch 'worktree-agent-a3abfcce648e87dca'1783 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1784 if let Some(stopped) = stopped.filter(|row| row.runner_id.is_some()) {
1785 self.released(&stopped).await?;
1786 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1787 Ok(())
1788 }
1789
Merge branch 'worktree-agent-a3abfcce648e87dca'1790 /// Ends a job's tokens at once. A failure is logged: the token expires
1791 /// on its own soon after the job's time limit.
1792 async fn revoke_job_tokens(&self, job_id: &str) {
1793 let revoked: Result<bool> =
1794 g1t_kit::call(&self.identity, "revoke_job_tokens", &RevokeJobTokensArgs { job_id: job_id.to_owned() }).await;
1795 if let Err(error) = revoked {
1796 worker::console_error!("actions: the tokens of job {job_id} were not revoked: {error}");
1797 }
1798 }
1799
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1800 /// Finishes the run when every job has.
1801 async fn finish_if_done(&self, run_id: &str) -> Result<()> {
1802 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1803 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1804 return Ok(());
1805 }
1806 let jobs = self.job_rows(run_id).await?;
1807 if !jobs.iter().all(|job| job.status == "completed") {
1808 return Ok(());
1809 }
1810 self.finish_run(&run, None).await
1811 }
1812
1813 async fn finish_run(&self, run: &RunRow, error: Option<&str>) -> Result<()> {
1814 let jobs = self.job_rows(&run.id).await?;
1815 let rows: Vec<&JobRow> = jobs.iter().collect();
1816 let conclusion = if error.is_some() {
1817 "failure"
1818 } else if run.conclusion.as_deref() == Some("cancelled") {
1819 "cancelled"
1820 } else if rows.is_empty() {
1821 "skipped"
1822 } else {
1823 key_result(&rows)
1824 };
1825 let done = self
1826 .db
1827 .prepare("UPDATE runs SET status = 'completed', conclusion = ?, error = COALESCE(?, error), finished_at = ? WHERE id = ? AND status != 'completed' RETURNING id")
1828 .bind(&[conclusion.into(), optional(error), now().into(), run.id.as_str().into()])?
1829 .first::<Value>(None)
1830 .await?;
1831 if done.is_none() {
1832 return Ok(());
1833 }
1834 self.report_status(run, conclusion).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971835 self.settle_deployments(run, &jobs, conclusion == "cancelled").await;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1836 let published: Result<()> = g1t_kit::call(
1837 &self.events,
1838 "publish",
1839 &g1t_contracts::events::Publish {
1840 events: vec![g1t_contracts::events::NewEvent {
1841 kind: "workflow.completed",
1842 source: "actions",
1843 repo_id: Some(run.repo_id.clone()),
1844 actor: run.actor_id.clone(),
1845 data: g1t_contracts::events::WorkflowEvent {
1846 run_id: run.id.clone(),
1847 repo_id: run.repo_id.clone(),
1848 workflow: run.name.clone(),
1849 path: run.path.clone(),
1850 number: run.number,
1851 event: run.event.clone(),
1852 conclusion: conclusion.to_owned(),
1853 git_ref: run.git_ref.clone(),
1854 sha: run.sha.clone(),
1855 pull: run.pull,
1856 },
1857 }],
1858 },
1859 )
1860 .await;
1861 if let Err(error) = published {
1862 worker::console_error!("actions: could not publish workflow.completed: {error}");
1863 }
1864 // The next run waiting in its concurrency group.
1865 if let Some(group) = &run.concurrency_group {
1866 let next = self
1867 .db
1868 .prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND status = 'pending' ORDER BY id LIMIT 1")
1869 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1870 .first::<RunRow>(None)
1871 .await?;
1872 if let Some(next) = next {
1873 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[next.id.as_str().into()])?.run().await?;
1874 Box::pin(self.advance(&next.id)).await?;
1875 }
1876 }
1877 Ok(())
1878 }
1879
Merge branch 'worktree-agent-ab6bf4195fcb97f7c' into claude/cool-newton-7nm8r91880 /// Tells the deployments service that a job ended (`workflow_job_ended`),
1881 /// so a build it opened with `start_from_workflow` and left open fails
1882 /// at once. Never fails the job: an answer that does not come is
1883 /// logged, and the sweep fails the build later.
1884 async fn end_job_builds(&self, job_id: &str, conclusion: &str) {
1885 let ended: Result<Value> =
1886 g1t_kit::call(&self.deployments, "workflow_job_ended", &json!({ "jobId": job_id, "conclusion": conclusion })).await;
1887 if let Err(error) = ended {
1888 worker::console_log!("actions: could not tell deployments that job {job_id} ended: {error}");
1889 }
1890 }
1891
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971892 /// Tells the deployments service how a run's deployment to `env` stands:
1893 /// made when its first job naming the environment starts, failed when one
1894 /// of them fails, and settled (`last`) when the run finishes. Never fails
1895 /// the run: a deployment that could not be recorded is logged.
1896 async fn report_deployment(&self, run: &RunRow, env: &JobEnvironment, state: &str, last: bool) {
1897 let path = repo_path(&run.repo);
1898 let reported: Result<Value> = g1t_kit::call(
1899 &self.deployments,
1900 "actions_deployment",
1901 &json!({
1902 "repoId": run.repo_id,
1903 "repo": { "namespace": path.namespace, "name": path.name },
1904 "runId": run.id,
1905 "attempt": run.attempt,
1906 "runUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
1907 "environment": env.name,
1908 "url": env.url,
1909 "ref": run.git_ref,
1910 "sha": run.sha,
1911 "state": state,
1912 "final": last,
1913 "creator": run.actor,
1914 "workflow": run.name,
1915 }),
1916 )
1917 .await;
1918 if let Err(error) = reported {
1919 worker::console_error!("actions: deployment to {} not recorded for run {}: {error}", env.name, run.id);
1920 }
1921 }
1922
1923 /// A job that deploys has started: its run's deployment to the
1924 /// environment is under way.
1925 pub(crate) async fn job_started(&self, job_id: &str) -> Result<()> {
1926 let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[job_id.into()])?.first::<JobRow>(None).await? else {
1927 return Ok(());
1928 };
1929 let Some(run) = self.run_row(&job.run_id).await? else { return Ok(()) };
1930 if let Some(env) = deploys_to(&run, &job) {
1931 self.report_deployment(&run, &env, "in_progress", false).await;
1932 }
1933 Ok(())
1934 }
1935
1936 /// The run is over: each environment its jobs deployed to takes the
1937 /// outcome of those jobs (`deployment_outcome`).
1938 async fn settle_deployments(&self, run: &RunRow, jobs: &[JobRow], cancelled: bool) {
1939 let mut seen: Vec<(JobEnvironment, Vec<Option<String>>)> = Vec::new();
1940 for job in jobs {
1941 let Some(env) = deploys_to(run, job) else { continue };
1942 let conclusion = if cancelled && job.conclusion.as_deref() != Some("skipped") && job.started_at.is_some() {
1943 Some("cancelled".to_owned())
1944 } else if job.started_at.is_none() {
1945 Some("skipped".to_owned())
1946 } else {
1947 job.conclusion.clone()
1948 };
1949 match seen.iter_mut().find(|(known, _)| known.name.eq_ignore_ascii_case(&env.name)) {
1950 Some((known, conclusions)) => {
1951 if known.url.is_none() {
1952 known.url = env.url.clone();
1953 }
1954 conclusions.push(conclusion);
1955 }
1956 None => seen.push((env, vec![conclusion])),
1957 }
1958 }
1959 for (env, conclusions) in seen {
1960 if let Some(state) = deployment_outcome(&conclusions) {
1961 self.report_deployment(run, &env, state, true).await;
1962 }
1963 }
1964 }
1965
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1966 /// Tells the pull request (or commit) how the run went, as a status.
1967 async fn report_status(&self, run: &RunRow, conclusion: &str) -> Result<()> {
1968 let state = match conclusion {
1969 "success" | "skipped" => "success",
1970 "cancelled" => "error",
1971 _ => "failure",
1972 };
1973 let _: Result<Value> = g1t_kit::call(
1974 &self.work,
1975 "set_commit_status",
1976 &json!({
1977 "repoId": run.repo_id,
1978 "sha": run.sha,
1979 "context": format!("{} / {}", run.name, run.event),
1980 "state": state,
1981 "description": format!("{} {}", run.name, match conclusion {
1982 "success" => "passed",
1983 "skipped" => "was skipped",
1984 "cancelled" => "was cancelled",
1985 _ => "failed",
1986 }),
1987 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge1988 "source": "actions",
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow1989 }),
1990 )
1991 .await;
1992 Ok(())
1993 }
1994
1995 /// Tells the pull request a run has started on its head.
1996 pub async fn report_pending(&self, run: &RunRow) -> Result<()> {
1997 let _: Result<Value> = g1t_kit::call(
1998 &self.work,
1999 "set_commit_status",
2000 &json!({
2001 "repoId": run.repo_id,
2002 "sha": run.sha,
2003 "context": format!("{} / {}", run.name, run.event),
2004 "state": "pending",
Merge branch 'worktree-agent-a3abfcce648e87dca'2005 "description": if run.status == "action_required" {
2006 format!("{} is waiting for approval", run.name)
2007 } else {
2008 format!("{} is running", run.name)
2009 },
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2010 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge2011 "source": "actions",
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2012 }),
2013 )
2014 .await;
2015 Ok(())
2016 }
2017
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2018 /// Cancels a run: its waiting and queued jobs at once, and its running
2019 /// ones gracefully (`stop_job`), or outright when `force`.
2020 pub async fn cancel_run(&self, run: &RunRow, reason: &str, force: bool) -> Result<()> {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2021 self.db
2022 .prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")
2023 .bind(&[run.id.as_str().into()])?
2024 .run()
2025 .await?;
2026 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)2027 if force {
2028 self.hard_stop(job, reason).await?;
2029 } else {
2030 self.stop_job(job, reason).await?;
2031 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2032 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2033 if run.status == "pending" || run.status == "action_required" {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2034 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;
2035 }
2036 self.advance(&run.id).await
2037 }
2038
2039 /// Cancels every run of the repository that has not finished, for
2040 /// `repo.deleted` and `repo.archived`.
2041 pub async fn stop_runs(&self, repo_id: &str) -> Result<()> {
2042 let runs = self
2043 .db
2044 .prepare("SELECT * FROM runs WHERE repo_id = ? AND status != 'completed'")
2045 .bind(&[repo_id.into()])?
2046 .all()
2047 .await?
2048 .results::<RunRow>()?;
2049 for run in runs {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2050 self.cancel_run(&run, "The repository was archived or deleted.", true).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2051 }
2052 Ok(())
2053 }
2054
2055 pub async fn cancel(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
2056 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
2057 return Ok(Outcome::Fail(refused));
2058 }
2059 let run = check!(self.run_in(&a.repo, &a.id).await?);
2060 if run.status == "completed" {
2061 return Ok(fail(FailureCode::Conflict, "The run has already finished."));
2062 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2063 // Cancelling a run that is already cancelling stops its jobs
2064 // outright, without waiting for their cleanup steps.
2065 let force = a.force || run.conclusion.as_deref() == Some("cancelled");
2066 let reason = if force {
2067 format!("{} stopped the run without waiting for its cleanup steps.", a.actor.username)
2068 } else {
2069 format!("{} cancelled the run.", a.actor.username)
2070 };
2071 self.cancel_run(&run, &reason, force).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2072 self.run_summary(&run.id).await
2073 }
2074
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2075 /// Runs again, as a new attempt: every job, with `failed_only` those
2076 /// that did not succeed, or with `job` that one; each with the jobs that
2077 /// need them. The attempt that ends is kept, its jobs and their logs and
2078 /// summaries with it (`job_attempts`, `run_attempts`).
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2079 pub async fn rerun(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
2080 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
2081 return Ok(Outcome::Fail(refused));
2082 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2083 // A job alone (`POST …/jobs/{job}/rerun`) names its run.
2084 let run_id = match (&a.job, a.id.is_empty()) {
2085 (Some(job), true) => {
2086 #[derive(Deserialize)]
2087 struct Of {
2088 run_id: String,
2089 }
2090 let of = self.db.prepare("SELECT run_id FROM jobs WHERE id = ?").bind(&[job.as_str().into()])?.first::<Of>(None).await?;
2091 match of {
2092 Some(of) => of.run_id,
2093 None => return Ok(fail(FailureCode::NotFound, "No such job.")),
2094 }
2095 }
2096 _ => a.id.clone(),
2097 };
2098 let run = check!(self.run_in(&a.repo, &run_id).await?);
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2099 if run.status != "completed" {
2100 return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));
2101 }
2102 // Nothing starts again on an archived repository.
2103 match self.visible_repo(&a.repo, &Some(a.actor.clone())).await? {
2104 Some(repo) if repo.archived() => {
2105 return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));
2106 }
2107 Some(_) => {}
2108 None => return Ok(fail(FailureCode::NotFound, "There is no such repository.")),
2109 }
2110 if run.error.is_some() {
2111 return Ok(fail(FailureCode::Conflict, "This run never started: fix the workflow file and push again."));
2112 }
2113 let jobs = self.job_rows(&run.id).await?;
2114 let workflow = workflow::parse(&run.source).ok();
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2115 let target = match &a.job {
2116 Some(id) => match jobs.iter().find(|job| &job.id == id) {
2117 Some(job) => Some(top_key(&job.key).to_owned()),
2118 None => return Ok(fail(FailureCode::NotFound, "That job is not in the run's latest attempt.")),
2119 },
2120 None => None,
2121 };
2122 let which = match (&target, a.failed_only) {
2123 (Some(key), _) => Rerun::Job(key),
2124 (None, true) => Rerun::Failed,
2125 (None, false) => Rerun::All,
2126 };
2127 let order = workflow.as_ref().map(|w| w.job_order()).unwrap_or_default();
2128 let again = rerun_keys(
2129 &order,
2130 |key| jobs.iter().find(|j| j.key == key).map(JobRow::needs).unwrap_or_default(),
2131 |key| jobs.iter().filter(|j| j.key == key).all(|j| j.conclusion.as_deref() == Some("success")),
2132 which,
2133 );
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2134 if again.is_empty() {
2135 return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));
2136 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2137 // The jobs that run again, a called workflow's with the job calling it.
2138 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();
2139 let ids = serde_json::to_string(&rerun_ids)?;
2140 let ended = run.attempt.to_string();
2141 let ended = ended.as_str();
2142 let mut statements = vec![
2143 // The attempt that ends, as it ended.
2144 self.db
2145 .prepare(
2146 "INSERT OR REPLACE INTO run_attempts (run_id, attempt, repo_id, conclusion, actor, debug, started_at, finished_at)
2147 SELECT id, attempt, repo_id, conclusion, COALESCE(triggering_actor, actor), debug, started_at, finished_at FROM runs WHERE id = ?",
2148 )
2149 .bind(&[run.id.as_str().into()])?,
2150 // Earlier attempts that showed a job's logs from where it ran
2151 // then now find them where they move to.
2152 self.db
2153 .prepare("UPDATE job_attempts SET log_id = log_id || '.' || ?1 WHERE run_id = ?2 AND log_id IN (SELECT value FROM json_each(?3))")
2154 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
2155 // Each of its jobs: one that runs again keeps its logs under
2156 // `{id}.{attempt}`; one left alone is still the live job's.
2157 self.db
2158 .prepare(
2159 "INSERT OR REPLACE INTO job_attempts (id, run_id, repo_id, attempt, job_id, log_id, key, ordinal, name, needs, status, conclusion,
2160 steps, annotations, reason, environment, labels, runner_name, started_at, finished_at)
2161 SELECT id || '.' || ?1, run_id, repo_id, ?1, id,
2162 CASE WHEN id IN (SELECT value FROM json_each(?3)) THEN id || '.' || ?1 ELSE id END,
2163 key, ordinal, name, needs, status, conclusion, steps, annotations, reason, environment, labels, runner_name, started_at, finished_at
2164 FROM jobs WHERE run_id = ?2 ORDER BY rowid",
2165 )
2166 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
2167 self.db
2168 .prepare("UPDATE logs SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
2169 .bind(&[ended.into(), ids.as_str().into()])?,
2170 self.db
2171 .prepare("UPDATE job_summaries SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
2172 .bind(&[ended.into(), ids.as_str().into()])?,
2173 ];
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2174 for key in &again {
2175 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-7nm8r92176 // The jobs of a workflow it called (each matrix combination's
2177 // under `key[i]/`) are made again when it calls it again.
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2178 statements.push(
2179 self.db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92180 .prepare("DELETE FROM jobs WHERE run_id = ? AND (key LIKE ? OR key LIKE ?)")
2181 .bind(&[run.id.as_str().into(), format!("{key}/%").into(), format!("{key}[%").into()])?,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2182 );
2183 statements.push(
2184 self.db
2185 .prepare(
2186 "UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
Fast pages, required checks on the branch, self-hosted runners, honest incidents2187 matrix = NULL, call = NULL, token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
Merge branch 'worktree-agent-a3abfcce648e87dca'2188 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)2189 concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2190 )
2191 .bind(&[run.id.as_str().into(), key.as_str().into()])?,
2192 );
2193 }
2194 statements.push(
2195 self.db
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2196 .prepare(
2197 "UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL,
2198 triggering_actor = ?, debug = ? WHERE id = ?",
2199 )
2200 .bind(&[a.actor.username.as_str().into(), u32::from(a.debug).into(), run.id.as_str().into()])?,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2201 );
2202 self.db.batch(statements).await?;
2203 if let Some(run) = self.run_row(&run.id).await? {
2204 self.report_pending(&run).await?;
2205 }
2206 self.advance(&run.id).await?;
2207 self.run_summary(&run.id).await
2208 }
2209
2210 pub async fn run_in(&self, repo: &RepoPath, id: &str) -> Result<Outcome<RunRow>> {
2211 let row = self
2212 .db
2213 .prepare("SELECT * FROM runs WHERE id = ? AND lower(repo) = lower(?)")
2214 .bind(&[id.into(), format!("{}/{}", repo.namespace, repo.name).into()])?
2215 .first::<RunRow>(None)
2216 .await?;
2217 Ok(row.map_or_else(|| fail(FailureCode::NotFound, "No such run."), Outcome::Ok))
2218 }
2219
2220 // --- The sandbox's side -----------------------------------------------------
2221
Fast pages, required checks on the branch, self-hosted runners, honest incidents2222 pub(crate) async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2223 let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;
2224 Ok(match job {
2225 Some(job) if job.status == "in_progress" && job.token_hash.as_deref().is_some_and(|hash| same(hash, &sha256_hex(&a.token))) => {
2226 Outcome::Ok(job)
2227 }
2228 _ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),
2229 })
2230 }
2231
2232 /// `job_auth`: which run and repository a running job's token is for,
2233 /// so the API can keep its artifacts and cache.
2234 pub async fn job_auth(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2235 let job = check!(self.job_for_token(&a).await?);
2236 Ok(Outcome::Ok(json!({ "run": job.run_id, "repoId": job.repo_id })))
2237 }
2238
2239 /// `job_spec`: everything the sandbox needs to run the job.
2240 pub async fn job_spec(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2241 let job = check!(self.job_for_token(&a).await?);
2242 let Some(run) = self.run_row(&job.run_id).await? else {
2243 return Ok(fail(FailureCode::NotFound, "No such run."));
2244 };
2245 let Ok(caller) = workflow::parse(&run.source) else {
2246 return Ok(fail(FailureCode::Invalid, "The workflow no longer reads."));
2247 };
2248 // A called workflow's job runs as that workflow defines it.
2249 let callee = job.callee();
2250 let (workflow, spec, call_inputs) = match callee {
2251 Some((called, spec, call)) => (called, spec, Some(call["inputs"].clone())),
2252 None => match caller.jobs.iter().find(|j| j.id == job.key) {
2253 Some(spec) => (caller.clone(), spec.clone(), None),
2254 None => return Ok(fail(FailureCode::NotFound, "The job is not in the workflow.")),
2255 },
2256 };
2257 let spec = &spec;
2258 let repo = repo_path(&run.repo);
2259 let trusted = run.trusted != 0;
Merge branch 'worktree-agent-a3abfcce648e87dca'2260 // The job's `environment:`, by name, as read when its needs were done
2261 // (an expression included), once the environment's protection rules
2262 // let it start: entries with a value for it give that value instead
2263 // of their default, as GitHub's environment secrets do.
2264 let environment: Option<String> = job.environment.clone();
2265 // What its token may do: its `permissions:` (a called workflow's
2266 // jobs no more than the job that calls it), else the repository's
2267 // default; read-only for a pull request from outside.
2268 // The repository's default, its workspace's taken in, and whether
2269 // its jobs may open and approve pull requests.
2270 let (default, pull_requests) = self.token_policy(&run.repo_id).await?;
2271 let mut permissions = spec.permissions(&workflow, default);
2272 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-7nm8r92273 let top = top_key(parent);
Merge branch 'worktree-agent-a3abfcce648e87dca'2274 if let Some(caller_job) = caller.jobs.iter().find(|j| j.id == top) {
2275 permissions = permissions.capped_by(&caller_job.permissions(&caller, default));
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2276 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2277 }
2278 if !trusted {
2279 permissions = permissions.read_only();
2280 }
2281 // G1T_TOKEN, and GITHUB_TOKEN as its alias: a token of the
2282 // workspace's that reaches this repository only, with the scopes
2283 // its permissions give, until the job ends.
2284 let token = match self.workspace_actor(&repo.namespace).await? {
2285 Some(workspace) => {
2286 let created: CreatedAccessToken = g1t_kit::call(
2287 &self.identity,
2288 "create_job_token",
2289 &CreateJobTokenArgs {
2290 workspace,
2291 repo: repo.clone(),
2292 run_id: run.id.clone(),
2293 job_id: job.id.clone(),
2294 name: format!("G1T_TOKEN for {} run {}", run.repo, run.number),
2295 ttl_seconds: u64::from(job.timeout_minutes) * 60 + 600,
2296 scopes: permissions.scopes().into_iter().map(str::to_owned).collect(),
2297 pull_requests: pull_requests && trusted,
2298 },
2299 )
2300 .await?;
2301 created.token
2302 }
2303 None => String::new(),
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2304 };
2305 // A run that is not trusted (a pull request from outside the
2306 // workspace) gets no secrets and an empty token.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts2307 let passed = job.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());
2308 let mut secrets = match (trusted, passed) {
2309 (false, _) => Map::new(),
2310 (true, None) => self.secrets_for(&run.repo_id, &run.repo, environment.as_deref(), true).await?,
2311 // A called workflow's job: what its callers passed it (reach.rs),
2312 // and its own environment's secrets over them.
2313 (true, Some(plan)) => {
2314 let base = self.secrets_for(&run.repo_id, &run.repo, None, true).await?;
2315 let vars = self.variables_for(&run.repo_id, &run.repo, None, true).await?;
2316 let github = run.info().context(&job.key, "", run.action.as_deref());
2317 let mut passed = crate::reach::resolve_secrets(&plan, &base, &github, &vars);
2318 if let Some(name) = environment.as_deref() {
2319 let own = self.secrets_for(&run.repo_id, &run.repo, Some(name), true).await?;
2320 for (key, value) in own {
2321 if base.get(&key) != Some(&value) {
2322 passed.insert(key, value);
2323 }
2324 }
2325 }
2326 passed
2327 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2328 };
2329 secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));
2330 secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));
Merge branch 'worktree-agent-a3abfcce648e87dca'2331 // Each secret as it is, a line at a time, base64 and JSON-escaped.
2332 let mut masks: Vec<String> = g1t_actions::mask::all_variants(secrets.values().filter_map(|v| v.as_str()));
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2333 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 R22334 // The toolkit's runtime token (runtime.rs), for as long as the job
2335 // may run; the API puts it and the toolkit's addresses in the
2336 // job's variables.
2337 let runtime_token = crate::runtime::runtime_token(
2338 &job.id,
2339 &job.run_id,
2340 job.token_hash.as_deref().unwrap_or_default(),
2341 now_ms() / 1000,
2342 u64::from(job.timeout_minutes) * 60 + 600,
2343 );
2344 masks.push(runtime_token.clone());
2345 let retention_days = self.retention_setting(&run.repo_id).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2346
2347 let jobs = self.job_rows(&run.id).await?;
2348 let mut needs = Map::new();
2349 // In a called workflow, its jobs' keys sit under the job that called it.
2350 let parent = job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));
2351 for need in &spec.needs {
2352 let key = match &parent {
2353 Some(parent) => format!("{parent}/{need}"),
2354 None => need.clone(),
2355 };
2356 let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();
2357 let mut outputs = Map::new();
2358 for row in &rows {
2359 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
2360 outputs.extend(more);
2361 }
2362 }
2363 needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));
2364 }
2365 let siblings = jobs.iter().filter(|row| row.key == job.key).count();
2366 let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));
2367 let info = run.info();
2368 let mut github = info.context(&job.key, &token, run.action.as_deref());
2369 github["token"] = json!(token);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22370 github["retention_days"] = json!(retention_days);
Fast pages, required checks on the branch, self-hosted runners, honest incidents2371 // On a self-hosted runner, `runner` and `RUNNER_*` describe that
2372 // machine rather than g1t's sandbox.
2373 let mut variables = info.variables(&job.key);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22374 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)2375 let mut runner = match &job.runner_id {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2376 Some(id) => self.runner_context_for(id, &mut variables).await?,
2377 None => runner_context(),
2378 };
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2379 // A re-run with debug logging: what GitHub sets for one.
2380 if run.debug != 0 {
2381 debug_logging(&mut variables, &mut runner);
2382 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2383
A pull request's `pull_request` runs build the pull request merged into its base branch, as on GitHub. g1t keeps `refs/pull/<number>/head` at every open pull request's head (a fork's or working copy's head commit is copied into the repository) and `refs/pull/<number>/merge` at a test merge commit of that head into the base branch's latest commit, moved whenever the head is pushed to, the pull request is opened or reopened, or its base changes; both can be fetched from the repository. A run's `github.sha` and `GITHUB_SHA` are that merge commit, `github.event.pull_request.merge_commit_sha` names it, `actions/checkout` fetches it as `+<sha>:refs/remotes/pull/<number>/merge` from the repository itself, and the workflow files and local reusable workflows are read from it; the run, its statuses and its check runs stay on the head, so required checks hold merges as before, and `pull_request_target` is unchanged. A pull request with conflicts starts no `pull_request` workflows, as on GitHub, and its checks box says why. g1t makes the merge without a sandbox, merging a file both sides changed line by line as git does; what it cannot merge itself (a rename on one side against a change on the other, more than 300 changed files) builds the head as before. The Actions and Checks guides say so.2384 // Where to check out: the repository, which holds a pull request's
2385 // test merge commit; a pull request's fork when the run builds its
2386 // head instead (`run.sha` is the head, `info.sha` what is built).
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2387 let clone_url = match run.pull {
A pull request's `pull_request` runs build the pull request merged into its base branch, as on GitHub. g1t keeps `refs/pull/<number>/head` at every open pull request's head (a fork's or working copy's head commit is copied into the repository) and `refs/pull/<number>/merge` at a test merge commit of that head into the base branch's latest commit, moved whenever the head is pushed to, the pull request is opened or reopened, or its base changes; both can be fetched from the repository. A run's `github.sha` and `GITHUB_SHA` are that merge commit, `github.event.pull_request.merge_commit_sha` names it, `actions/checkout` fetches it as `+<sha>:refs/remotes/pull/<number>/merge` from the repository itself, and the workflow files and local reusable workflows are read from it; the run, its statuses and its check runs stay on the head, so required checks hold merges as before, and `pull_request_target` is unchanged. A pull request with conflicts starts no `pull_request` workflows, as on GitHub, and its checks box says why. g1t makes the merge without a sandbox, merging a file both sides changed line by line as git does; what it cannot merge itself (a rename on one side against a change on the other, more than 300 changed files) builds the head as before. The Actions and Checks guides say so.2388 Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" && info.sha == run.sha => {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2389 let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(
2390 &self.work,
2391 "get_pull",
2392 &g1t_contracts::work::ViewArgs {
2393 repo: repo.clone(),
2394 number,
2395 viewer: self.workspace_actor(&repo.namespace).await?,
2396 after_seq: 0,
2397 },
2398 )
2399 .await?;
2400 match located {
2401 Outcome::Ok(detail) => match detail.pull.fork {
2402 Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),
2403 None => format!("{SITE}/{}.git", run.repo),
2404 },
2405 Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),
2406 }
2407 }
2408 _ => format!("{SITE}/{}.git", run.repo),
2409 };
2410
2411 Ok(Outcome::Ok(json!({
2412 "job": job.id,
2413 "run": run.id,
2414 "key": job.key,
2415 "name": job.name,
2416 "spec": spec.raw,
2417 "workflow": {
2418 "env": workflow.env,
2419 "defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),
2420 },
2421 "github": github,
Fast pages, required checks on the branch, self-hosted runners, honest incidents2422 "variables": variables,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2423 "event": info.event,
2424 "contexts": {
2425 "vars": vars,
2426 "secrets": secrets,
2427 "inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),
2428 "matrix": matrix,
2429 "needs": needs,
2430 "strategy": {
2431 "fail-fast": spec.fail_fast,
2432 "job-index": job.ordinal,
2433 "job-total": siblings,
2434 "max-parallel": spec.max_parallel.unwrap_or(siblings as u32),
2435 },
Fast pages, required checks on the branch, self-hosted runners, honest incidents2436 "runner": runner,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2437 },
2438 "checkout": {
2439 "repository": run.repo,
2440 "url": clone_url,
A pull request's `pull_request` runs build the pull request merged into its base branch, as on GitHub. g1t keeps `refs/pull/<number>/head` at every open pull request's head (a fork's or working copy's head commit is copied into the repository) and `refs/pull/<number>/merge` at a test merge commit of that head into the base branch's latest commit, moved whenever the head is pushed to, the pull request is opened or reopened, or its base changes; both can be fetched from the repository. A run's `github.sha` and `GITHUB_SHA` are that merge commit, `github.event.pull_request.merge_commit_sha` names it, `actions/checkout` fetches it as `+<sha>:refs/remotes/pull/<number>/merge` from the repository itself, and the workflow files and local reusable workflows are read from it; the run, its statuses and its check runs stay on the head, so required checks hold merges as before, and `pull_request_target` is unchanged. A pull request with conflicts starts no `pull_request` workflows, as on GitHub, and its checks box says why. g1t makes the merge without a sandbox, merging a file both sides changed line by line as git does; what it cannot merge itself (a rename on one side against a change on the other, more than 300 changed files) builds the head as before. The Actions and Checks guides say so.2441 "sha": info.sha,
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2442 "ref": run.git_ref,
2443 "token": token,
2444 },
2445 "timeoutMinutes": job.timeout_minutes,
2446 "masks": masks,
Merge branch 'worktree-agent-a3abfcce648e87dca'2447 // As the job's log lists them at its start.
2448 "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 R22449 // Whether the job may ask for an OIDC token decides whether it
2450 // is told where to.
2451 "runtime": {
2452 "token": runtime_token,
2453 "idToken": self.oidc_allowed(&run, &job),
2454 },
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2455 })))
2456 }
2457
2458 /// `job_report`: the sandbox telling how the job is going.
2459 pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2460 let job = check!(self.job_for_token(&a).await?);
2461 let report = &a.report;
2462 let at = now();
2463 match report["kind"].as_str().unwrap_or_default() {
2464 "steps" => {
2465 // The list can grow as the job goes (post steps), so steps
2466 // already reported keep where they stand.
2467 let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2468 let steps: Vec<Value> = report["steps"]
2469 .as_array()
2470 .map(|names| {
2471 names
2472 .iter()
2473 .enumerate()
2474 .map(|(i, name)| match known.get(i) {
2475 Some(step) if step["status"] != "queued" => step.clone(),
2476 _ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),
2477 })
2478 .collect()
2479 })
2480 .unwrap_or_default();
2481 self.db
2482 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2483 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2484 .run()
2485 .await?;
2486 }
2487 "step" => {
2488 let number = report["number"].as_u64().unwrap_or(0) as usize;
2489 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2490 if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {
2491 let status = report["status"].as_str().unwrap_or("in_progress");
2492 step["status"] = json!(status);
2493 if status == "in_progress" {
2494 step["startedAt"] = json!(at);
2495 }
2496 if status == "completed" {
2497 step["finishedAt"] = json!(at);
2498 step["conclusion"] = report["conclusion"].clone();
2499 }
2500 if let Some(name) = report["name"].as_str() {
2501 step["name"] = json!(name);
2502 }
2503 }
2504 self.db
2505 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2506 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2507 .run()
2508 .await?;
2509 }
2510 "log" => {
2511 let mut text = report["text"].as_str().unwrap_or_default().to_owned();
2512 if text.len() > MAX_CHUNK_BYTES {
2513 let mut cut = MAX_CHUNK_BYTES;
2514 while !text.is_char_boundary(cut) {
2515 cut -= 1;
2516 }
2517 text.truncate(cut);
2518 }
2519 #[derive(Deserialize)]
2520 struct Size {
2521 n: Option<f64>,
2522 }
2523 let size = self
2524 .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.2525 .prepare("SELECT SUM(LENGTH(text)) AS n FROM logs WHERE job_id = ?")
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2526 .bind(&[job.id.as_str().into()])?
2527 .first::<Size>(None)
2528 .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.2529 let used = size.and_then(|s| s.n).unwrap_or(0.0) as usize;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2530 if used < MAX_LOG_BYTES {
2531 if used + text.len() >= MAX_LOG_BYTES {
2532 text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");
2533 }
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.2534 // The next `seq` is taken in the insert itself, so two
2535 // batches at once (a report sent again while the first
2536 // was still being written) never reach for the same one
2537 // and lose a chunk to the primary key.
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2538 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.2539 .prepare(LOG_INSERT)
Merge branch 'worktree-agent-a39a116f38472effe' into claude/cool-newton-7nm8r92540 .bind(&[
2541 job.id.as_str().into(),
2542 (report["step"].as_u64().unwrap_or(0) as u32).into(),
2543 text.into(),
2544 report["chunk"].as_u64().map_or(worker::wasm_bindgen::JsValue::NULL, |n| (n as f64).into()),
2545 ])?
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2546 .run()
2547 .await?;
2548 }
2549 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2550 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2551 "summary" => {
2552 // `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the
2553 // step's summary, up to 1 MiB a step and 20 steps a job.
2554 let step = report["step"].as_u64().unwrap_or(0) as u32;
2555 let markdown = report["markdown"].as_str().unwrap_or_default();
2556 #[derive(Deserialize)]
2557 struct Held {
2558 steps: u32,
2559 mine: Option<f64>,
2560 }
2561 let held = self
2562 .db
2563 .prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")
2564 .bind(&[step.into(), job.id.as_str().into()])?
2565 .first::<Held>(None)
2566 .await?;
2567 let (steps, mine) = held.map_or((0, None), |held| (held.steps, held.mine.map(|n| n as usize)));
2568 if summary_fits(steps, mine, markdown.len()) {
2569 self.db
2570 .prepare(
2571 "INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)
2572 ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown || excluded.markdown",
2573 )
2574 .bind(&[job.id.as_str().into(), step.into(), markdown.into()])?
2575 .run()
2576 .await?;
2577 }
2578 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2579 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2580 "annotation" => {
2581 let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
2582 if annotations.len() < MAX_ANNOTATIONS {
2583 annotations.push(json!({
2584 "level": report["level"].as_str().unwrap_or("notice"),
2585 "message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),
2586 "title": report["title"],
2587 "file": report["file"],
2588 "line": report["line"],
2589 }));
2590 self.db
2591 .prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")
2592 .bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?
2593 .run()
2594 .await?;
2595 }
2596 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2597 // Nothing to tell; the answer says whether to stop.
2598 "ping" => {
2599 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2600 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2601 "done" => {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2602 // A job told to stop ends cancelled, however its cleanup went.
2603 let conclusion = if job.cancel_requested_at.is_some() {
2604 "cancelled"
2605 } else {
2606 report["conclusion"]
2607 .as_str()
2608 .filter(|c| matches!(*c, "success" | "failure" | "cancelled"))
2609 .unwrap_or("failure")
2610 };
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2611 let outputs = report["outputs"].as_object().cloned();
2612 Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
2613 }
2614 other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
2615 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2616 // `cancelled`: the run was cancelled, and the runner should stop
2617 // the step it is on and run only its cleanup steps.
2618 Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2619 }
2620
2621 // --- Every minute ---------------------------------------------------------------
2622
2623 pub async fn on_minute(&self, now_ms: u64) -> Result<()> {
2624 let minute = now_ms / 60_000 * 60_000;
Merge branch 'worktree-agent-a209db8188a251e77' into claude/cool-newton-7nm8r92625 // Repositories a workspace's managed workflows changed in, beyond
2626 // those synced at once (managed.rs): their schedules wait for it.
2627 if let Err(error) = self.catch_up_syncs().await {
2628 worker::console_error!("actions: catching up syncs failed: {error}");
2629 }
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)2630 // Staff (or billing's usage watcher) paused scheduled runs across
2631 // g1t: this minute's schedules are skipped, not queued for later.
2632 // Kept 30 seconds in the isolate (g1t_kit::pause).
2633 if g1t_kit::pause::paused(&self.billing, g1t_contracts::billing::PauseLevel::Schedules).await {
2634 worker::console_log!("actions: schedules are paused across g1t; skipped this minute's");
2635 } else if let Err(error) = self.run_schedules(minute).await {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2636 worker::console_error!("actions: schedules failed: {error}");
2637 }
2638 // Jobs whose sandbox went quiet or ran past their time.
2639 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
2640 for job in running {
2641 // Times in g1t's format compare as text.
2642 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
2643 let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());
2644 let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
2645 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)2646 let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(|at| at < before(CANCEL_GRACE_MS).as_str());
2647 if cancel_overdue {
2648 // Cancelled, and still going after its grace period.
2649 self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;
2650 self.advance(&job.run_id).await?;
2651 } else if over {
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2652 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 incidents2653 // A self-hosted runner is told to stop on its next poll.
2654 if job.runner_id.is_none() {
2655 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2656 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2657 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2658 } else if silent {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2659 let reason = match &job.runner_name {
2660 Some(name) => format!("The self-hosted runner {name} stopped answering."),
2661 None => "The runner stopped answering.".to_owned(),
2662 };
2663 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2664 }
2665 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92666 // Calls of reusable workflows waiting in a concurrency group, in
2667 // case the job they waited for was cancelled in another run.
2668 #[derive(Deserialize)]
2669 struct RunId {
2670 run_id: String,
2671 }
2672 let held = self
2673 .db
2674 .prepare("SELECT DISTINCT run_id FROM jobs WHERE status = 'waiting' AND concurrency_group IS NOT NULL LIMIT 20")
2675 .all()
2676 .await?
2677 .results::<RunId>()?;
2678 for run in held {
2679 self.advance(&run.run_id).await?;
2680 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2681 if let Err(error) = self.sweep_runners(now_ms).await {
2682 worker::console_error!("actions: the runners' sweep failed: {error}");
2683 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2684 // Jobs held at an environment whose wait timer has run out.
2685 if let Err(error) = self.release_gates().await {
2686 worker::console_error!("actions: environments' gates failed: {error}");
2687 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2688 // Once an hour: the cache's expired entries, and its storage.
2689 if (now_ms / 60_000) % 60 == 7
2690 && let Err(error) = self.sweep_cache(now_ms).await
2691 {
2692 worker::console_error!("actions: the cache's sweep failed: {error}");
2693 }
Merge remote-tracking branch 'origin/claude/peaceful-franklin-7qtkwx'2694 // And the archives of GitHub's actions no job uses any more.
2695 if (now_ms / 60_000) % 60 == 47
2696 && let Err(error) = self.sweep_action_archives(now_ms).await
2697 {
2698 worker::console_error!("actions: the action archives' sweep failed: {error}");
2699 }
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22700 // And artifacts past their time, and the toolkit's abandoned parts.
2701 if (now_ms / 60_000) % 60 == 37 {
2702 if let Err(error) = self.sweep_artifacts(now_ms).await {
2703 worker::console_error!("actions: the artifacts' sweep failed: {error}");
2704 }
2705 if let Err(error) = self.sweep_blob_parts(now_ms).await {
2706 worker::console_error!("actions: the blob parts' sweep failed: {error}");
2707 }
2708 }
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2709 self.start_queued().await
2710 }
2711}
2712
2713
2714#[cfg(test)]
Merge branch 'worktree-agent-a39a116f38472effe' into claude/cool-newton-7nm8r92715mod log_chunks {
2716 use rusqlite::{Connection, params};
2717
2718 use super::LOG_INSERT;
2719
2720 /// The logs table as the migrations leave it.
2721 fn database() -> Connection {
2722 let db = Connection::open_in_memory().unwrap();
2723 for name in ["0001_init.sql", "0016_log_chunks.sql"] {
2724 let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));
2725 db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
2726 }
2727 db
2728 }
2729
2730 fn log(db: &Connection, job: &str) -> Vec<(i64, i64, String)> {
2731 let mut rows = db.prepare("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq").unwrap();
2732 rows.query_map([job], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).unwrap().map(Result::unwrap).collect()
2733 }
2734
2735 #[test]
2736 fn a_batch_sent_again_is_kept_once() {
2737 let db = database();
2738 let insert = |job: &str, step: i64, text: &str, chunk: Option<i64>| db.execute(LOG_INSERT, params![job, step, text, chunk]).unwrap();
2739 assert_eq!(insert("job_1", 0, "one\n", Some(1)), 1);
2740 // The runner did not hear the answer and sent the batch again.
2741 assert_eq!(insert("job_1", 0, "one\n", Some(1)), 0);
2742 assert_eq!(insert("job_1", 1, "two\n", Some(2)), 1);
2743 // Another job numbers its batches from 1 too.
2744 assert_eq!(insert("job_2", 0, "other\n", Some(1)), 1);
2745 // A runner that numbers nothing has every batch kept, as before.
2746 assert_eq!(insert("job_3", 0, "a\n", None), 1);
2747 assert_eq!(insert("job_3", 0, "a\n", None), 1);
2748 assert_eq!(log(&db, "job_1"), [(1, 0, "one\n".into()), (2, 1, "two\n".into())]);
2749 assert_eq!(log(&db, "job_2"), [(1, 0, "other\n".into())]);
2750 assert_eq!(log(&db, "job_3").len(), 2);
2751 // A job run again moves its log aside (`{id}.{attempt}`), and the
2752 // new attempt's runner numbers from 1 again.
2753 db.execute("UPDATE logs SET job_id = job_id || '.1' WHERE job_id = 'job_1'", []).unwrap();
2754 assert_eq!(insert("job_1", 0, "again\n", Some(1)), 1);
2755 assert_eq!(log(&db, "job_1"), [(1, 0, "again\n".into())]);
2756 assert_eq!(log(&db, "job_1.1").len(), 2);
2757 }
2758}
2759
2760#[cfg(test)]
Deploys as code: a manifest of every Worker, a deploy tool that ships only what changed in parallel stages, and a g1t Actions workflow2761mod stopping {
2762 use super::stops_runs;
2763 use g1t_contracts::events::Event;
2764 use serde_json::{Value, json};
2765
2766 fn event(kind: &str, data: Value) -> Event {
2767 Event {
2768 id: "evt_1".into(),
2769 kind: kind.into(),
2770 source: "repos".into(),
2771 time: "2026-10-05T00:00:00Z".into(),
2772 repo_id: Some("rep_1".into()),
2773 actor: None,
2774 data,
2775 }
2776 }
2777
2778 #[test]
2779 fn deleting_or_archiving_stops_runs() {
2780 assert_eq!(stops_runs(&event("repo.deleted", json!({ "repoId": "rep_1" }))).as_deref(), Some("rep_1"));
2781 assert_eq!(stops_runs(&event("repo.archived", json!({ "archived": true }))).as_deref(), Some("rep_1"));
2782 assert_eq!(stops_runs(&event("repo.unarchived", json!({ "archived": false }))), None);
2783 assert_eq!(stops_runs(&event("repo.restored", json!({}))), None);
2784 assert_eq!(stops_runs(&event("git.push", json!({}))), None);
2785 }
2786}
Fast pages, required checks on the branch, self-hosted runners, honest incidents2787
2788#[cfg(test)]
2789mod status_of_needs {
2790 use std::collections::HashMap;
2791
2792 use super::ancestor_failed;
2793
2794 /// check -> plan -> (migrate) -> core -> edge, as deploy.yml has them,
2795 /// and a job that needs only the last.
2796 fn graph() -> HashMap<&'static str, Vec<&'static str>> {
2797 HashMap::from([
2798 ("check", vec![]),
2799 ("plan", vec!["check"]),
2800 ("migrate", vec!["plan"]),
2801 ("core", vec!["plan", "migrate"]),
2802 ("edge", vec!["plan", "migrate", "core"]),
2803 ("notify", vec!["edge"]),
2804 ])
2805 }
2806
2807 #[test]
2808 fn a_failure_is_seen_however_far_back() {
2809 let needs = graph();
2810 let failed = |which: &'static str| move |key: &str| key == which;
2811 // check failed; plan, and everything after, was skipped for it.
2812 assert!(ancestor_failed(&needs, "notify", failed("check")));
2813 assert!(ancestor_failed(&needs, "core", failed("check")));
2814 assert!(ancestor_failed(&needs, "edge", failed("core")));
2815 // Nothing before a job failed: a skipped migrate is not a failure.
2816 assert!(!ancestor_failed(&needs, "edge", |_| false));
2817 assert!(!ancestor_failed(&needs, "core", failed("edge")));
2818 assert!(!ancestor_failed(&needs, "check", failed("check")));
2819 }
2820
2821 #[test]
2822 fn cycles_and_unknown_keys_end() {
2823 let needs = HashMap::from([("a", vec!["b"]), ("b", vec!["a"])]);
2824 assert!(!ancestor_failed(&needs, "a", |_| false));
2825 assert!(!ancestor_failed(&needs, "missing", |_| true));
2826 }
2827}
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb972828
2829#[cfg(test)]
2830mod deployments {
2831 use serde_json::{Map, Value, json};
2832
2833 use super::{JobEnvironment, deployment_outcome, environment_of};
2834
2835 #[test]
2836 fn a_jobs_environment_is_read_for_deployments() {
2837 let contexts: Map<String, Value> = serde_json::from_value(json!({
2838 "github": { "ref_name": "main", "repository": "acme/web" },
2839 "inputs": { "target": "staging" },
2840 "matrix": {},
2841 }))
2842 .unwrap();
2843 let read = |raw: Value| environment_of(&raw, &contexts);
2844 assert_eq!(read(json!({})), None);
2845 assert_eq!(
2846 read(json!({ "environment": "production" })),
2847 Some(JobEnvironment { name: "production".into(), url: None, deploys: true })
2848 );
2849 assert_eq!(
2850 read(json!({ "environment": { "name": "production", "url": "https://g1t.sh" } })),
2851 Some(JobEnvironment { name: "production".into(), url: Some("https://g1t.sh".into()), deploys: true })
2852 );
2853 // Expressions are filled in from the run.
2854 assert_eq!(
2855 read(json!({ "environment": { "name": "${{ inputs.target }}", "url": "https://${{ github.ref_name }}.example.com" } })),
2856 Some(JobEnvironment { name: "staging".into(), url: Some("https://main.example.com".into()), deploys: true })
2857 );
2858 // Secrets only: no deployment.
2859 assert!(!read(json!({ "environment": { "name": "production", "deployment": false } })).unwrap().deploys);
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r92860 // The same, said in the job's g1t settings.
2861 assert!(!read(json!({ "environment": "production", "g1t": { "deployment": false } })).unwrap().deploys);
2862 assert!(!read(json!({ "environment": { "name": "production" }, "g1t": { "deployment": false } })).unwrap().deploys);
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb972863 // Only http(s) addresses.
2864 assert_eq!(read(json!({ "environment": { "name": "production", "url": "javascript:alert(1)" } })).unwrap().url, None);
2865 }
2866
2867 #[test]
2868 fn a_runs_outcome_for_an_environment() {
2869 let of = |list: &[&str]| deployment_outcome(&list.iter().map(|c| Some((*c).to_owned())).collect::<Vec<_>>());
2870 assert_eq!(of(&["success", "skipped"]), Some("success"));
2871 assert_eq!(of(&["success", "failure"]), Some("failure"));
2872 assert_eq!(of(&["success", "cancelled"]), Some("error"));
2873 assert_eq!(of(&["skipped"]), None);
2874 assert_eq!(deployment_outcome(&[None]), None);
2875 }
2876}
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2877
2878#[cfg(test)]
2879mod reruns {
2880 use serde_json::{Map, Value, json};
2881
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92882 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)2883
2884 /// build ← test ← deploy, and lint on its own.
2885 fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {
2886 let order = ["build", "lint", "test", "deploy"];
2887 let needs = |key: &str| -> Vec<String> {
2888 match key {
2889 "test" => vec!["build".into()],
2890 "deploy" => vec!["test".into()],
2891 _ => Vec::new(),
2892 }
2893 };
2894 rerun_keys(&order, needs, |key| !failed.contains(&key), which)
2895 }
2896
2897 #[test]
2898 fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {
2899 assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);
2900 assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);
2901 assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);
2902 assert!(keys(Rerun::Failed, &[]).is_empty());
2903 // One job, whatever it came to, and what depends on it.
2904 assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);
2905 assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);
2906 assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);
2907 assert!(keys(Rerun::Job("missing"), &[]).is_empty());
2908 }
2909
2910 #[test]
2911 fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {
2912 assert_eq!(top_key("build/test"), "build");
2913 assert_eq!(top_key("build/inner/test"), "build");
2914 assert_eq!(top_key("lint"), "lint");
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92915 // A matrix's calls each have their own jobs, under `key[i]/`.
2916 assert_eq!(top_key("build[1]/test"), "build");
2917 assert_eq!(top_key("build/inner[0]/test"), "build");
2918 assert_eq!(call_prefix("build", Some(1)), "build[1]");
2919 assert_eq!(call_prefix("build/inner", None), "build/inner");
2920 }
2921
2922 #[test]
2923 fn a_run_connects_ten_levels_of_workflows_and_fifty_reusable_ones() {
2924 assert!(call_allowed(1, &[], "a.yml").is_ok());
2925 assert!(call_allowed(9, &[], "a.yml").is_ok());
2926 let deep = call_allowed(10, &[], "a.yml").unwrap_err();
2927 assert!(deep.contains("at most ten") && deep.contains("nine levels"), "{deep}");
2928 let names: Vec<String> = (0..50).map(|n| format!("w{n}.yml")).collect();
2929 let fifty: Vec<&str> = names.iter().map(String::as_str).collect();
2930 // A workflow called again, as by each combination of a matrix, is one.
2931 assert!(call_allowed(1, &fifty, "w3.yml").is_ok());
2932 assert!(call_allowed(1, &fifty, "another.yml").unwrap_err().contains("at most 50 different"));
2933 }
2934
2935 #[test]
2936 fn a_matrixs_calls_are_named_by_the_caller_and_its_values() {
2937 let workflow = super::workflow::parse(
2938 "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' }]",
2939 )
2940 .unwrap();
2941 let job = |id: &str| workflow.jobs.iter().find(|job| job.id == id).unwrap();
2942 let combination: Map<String, Value> = serde_json::from_value(json!({ "os": "linux", "node": 20 })).unwrap();
2943 let mut contexts = Map::new();
2944 contexts.insert("matrix".into(), Value::Object(combination.clone()));
2945 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
2946 assert_eq!(shown_name(job("build"), None, &combination, &scope), "Build (linux, 20)");
2947 assert_eq!(shown_name(job("plain"), None, &Map::new(), &scope), "plain");
2948 // A called workflow's job, under the call that made it.
2949 assert_eq!(shown_name(job("plain"), Some("Build (linux, 20)"), &Map::new(), &scope), "Build (linux, 20) / plain");
2950 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)2951 }
2952
2953 #[test]
2954 fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {
2955 assert!(summary_fits(0, None, 10));
2956 assert!(!summary_fits(0, None, 0), "nothing to keep");
2957 assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");
2958 assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");
2959 assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));
2960 assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));
2961 assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));
2962 }
2963
2964 #[test]
2965 fn a_debug_rerun_sets_what_github_sets() {
2966 let mut variables = Map::new();
2967 let mut runner = json!({ "name": "g1t", "debug": "" });
2968 debug_logging(&mut variables, &mut runner);
2969 assert_eq!(variables["RUNNER_DEBUG"], "1");
2970 assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");
2971 assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");
2972 assert_eq!(runner["debug"], Value::String("1".into()));
2973 assert_eq!(runner["name"], "g1t");
2974 }
2975}

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