Skip to content
Sign up

flagon-io/g1t

public
00

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

g1t.sh
gitagentscloudflarerustforgemcp
3,199 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.

Download ZIP reads blobs eight batches at a time and deflates files together1//! 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};
Download ZIP reads blobs eight batches at a time and deflates files together13use 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 brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.22use crate::broker::{self, Broker};
Merge branch 'worktree-agent-a3abfcce648e87dca'23use crate::protection::Gate;
Download ZIP reads blobs eight batches at a time and deflates files together24use crate::sync::WorkflowRow;
25use crate::{Actions, Count, MAX_TIMEOUT_MINUTES, RUNNING_PER_WORKSPACE, SELF_HOSTED_MAX_TIMEOUT_MINUTES, SILENT_MS, SITE, check, fail, optional, repo_path};
26use g1t_contracts::runners::{Wanted, waiting_reason};
27
28/// The most log one job keeps, in bytes; past it, the log says so and stops.
29const MAX_LOG_BYTES: usize = 4 * 1024 * 1024;
30/// The most a single log report may add.
31const MAX_CHUNK_BYTES: usize = 256 * 1024;
32const MAX_ANNOTATIONS: usize = 50;
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.33/// Swaps a running job's token for a new one, only while it still has the
34/// one the caller showed (`job_spec` with `rotate`), keeping the hash of
35/// the one it started with for `brokered_secrets` (migration 0020).
36const TOKEN_ROTATE: &str = "UPDATE jobs SET first_token_hash = COALESCE(first_token_hash, token_hash), token_hash = ?1
37 WHERE id = ?2 AND status = 'in_progress' AND token_hash = ?3 RETURNING id";
38
39/// Whether `token` is a running job's: its current one, or with `started`
40/// (`brokered_secrets` only) the one it started with before its harness
41/// swapped it.
42fn token_opens(current: Option<&str>, first: Option<&str>, token: &str, started: bool) -> bool {
43 let hash = sha256_hex(token);
44 current.is_some_and(|h| same(h, &hash)) || (started && first.is_some_and(|h| same(h, &hash)))
45}
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.46/// A batch of a job's log, under the next `seq` of that job's, worked out
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'47/// in the same statement (SQLite runs one write at a time). `?4` is the
48/// runner's number for the batch (null from runners that send none): a
49/// batch sent again, after an answer the runner did not hear, has the
50/// number of one already kept, and is not kept twice (migration 0016).
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.51pub(crate) const LOG_INSERT: &str = "INSERT INTO logs (job_id, seq, step, text, chunk)
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'52 SELECT ?1, COALESCE(MAX(seq), 0) + 1, ?2, ?3, ?4 FROM logs WHERE job_id = ?1
53 ON CONFLICT (job_id, chunk) DO NOTHING";
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)54/// The most one step's job summary keeps, in bytes, and how many steps of a
55/// job may have one, as on GitHub.
56pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
57pub(crate) const MAX_SUMMARIES: u32 = 20;
58
59/// Whether a step's summary of `adding` bytes is kept: a job keeps up to
60/// 20 steps' summaries (`steps` it has, `mine` this step's bytes so far)
61/// of up to 1 MiB each.
62pub(crate) fn summary_fits(steps: u32, mine: Option<usize>, adding: usize) -> bool {
63 if adding == 0 {
64 return false;
65 }
66 match mine {
67 Some(used) => used + adding <= MAX_SUMMARY_BYTES,
68 None => steps < MAX_SUMMARIES && adding <= MAX_SUMMARY_BYTES,
69 }
70}
Download ZIP reads blobs eight batches at a time and deflates files together71
72/// The repository whose unfinished runs `event` stops, by id: one deleted,
73/// or archived (not unarchived).
74pub fn stops_runs(event: &g1t_contracts::events::Event) -> Option<String> {
75 let stops = match event.kind.as_str() {
76 "repo.deleted" => true,
77 "repo.archived" => event.data["archived"].as_bool() != Some(false),
78 _ => false,
79 };
80 if !stops {
81 return None;
82 }
83 event.repo_id.clone().or_else(|| event.data["repoId"].as_str().map(str::to_owned))
84}
85
86pub struct NewRun {
87 pub repo: Repo,
88 pub path: String,
89 pub source: String,
90 pub workflow: Workflow,
91 pub info: RunInfo,
92 pub action: Option<String>,
93 pub pull: Option<u32>,
94 pub title: String,
95 pub inputs: Map<String, Value>,
96 pub event_key: String,
97 pub actor_id: Option<String>,
98 pub actor: Option<String>,
99 pub trusted: bool,
Merge branch 'worktree-agent-a3abfcce648e87dca'100 /// Why the run waits for someone with the Write role to approve it
101 /// before anything starts: a pull request from outside (protection.rs).
102 pub approval: Option<String>,
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'103 /// The commit the run is listed, and its statuses and checks reported,
104 /// on when it is not the one it builds (`info.sha`): a pull request's
105 /// head, when the run builds its test merge commit, as on GitHub.
106 pub head_sha: Option<String>,
Download ZIP reads blobs eight batches at a time and deflates files together107}
108
109#[derive(Clone, Deserialize)]
110pub struct RunRow {
111 pub id: String,
112 pub workflow_id: String,
113 pub repo_id: String,
114 pub repo: String,
115 pub path: String,
116 pub name: String,
117 pub title: String,
118 pub number: u64,
119 pub attempt: u64,
120 pub event: String,
121 pub action: Option<String>,
122 pub git_ref: String,
123 pub sha: String,
124 pub pull: Option<u32>,
125 pub status: String,
126 pub conclusion: Option<String>,
127 pub error: Option<String>,
128 pub actor: Option<String>,
129 pub actor_id: Option<String>,
130 pub source: String,
131 pub info: String,
132 pub inputs: String,
133 pub trusted: u32,
134 pub concurrency_group: Option<String>,
135 pub created_at: String,
136 pub started_at: Option<String>,
137 pub finished_at: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'138 /// JSON `RunApproval`, for a run that needed approval (migration 0006).
139 #[serde(default)]
140 pub approval: Option<String>,
141 /// Whether its concurrency group cancels what it replaces.
142 #[serde(default)]
143 pub cancel_in_progress: u32,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)144 /// Who started the current attempt, once it is re-run (migration 0009).
145 #[serde(default)]
146 pub triggering_actor: Option<String>,
147 /// The current attempt runs with debug logging.
148 #[serde(default)]
149 pub debug: u32,
Download ZIP reads blobs eight batches at a time and deflates files together150}
151
152impl RunRow {
Merge branch 'worktree-agent-a3abfcce648e87dca'153 pub fn approval(&self) -> Option<RunApproval> {
154 self.approval.as_deref().and_then(|text| serde_json::from_str(text).ok())
155 }
156
Download ZIP reads blobs eight batches at a time and deflates files together157 pub fn info(&self) -> RunInfo {
158 let mut info: RunInfo = serde_json::from_str(&self.info).unwrap_or_default();
159 info.run_id = self.id.clone();
160 info.run_number = self.number;
161 info.run_attempt = self.attempt;
162 info.workflow_path = self.path.clone();
163 if info.workflow.is_empty() {
164 info.workflow = self.name.clone();
165 }
166 info
167 }
168
169 pub fn inputs(&self) -> Map<String, Value> {
170 serde_json::from_str(&self.inputs).unwrap_or_default()
171 }
172
173 pub fn summary(&self) -> WorkflowRun {
174 WorkflowRun {
175 id: self.id.clone(),
176 workflow_id: self.workflow_id.clone(),
177 path: self.path.clone(),
178 name: self.name.clone(),
179 title: self.title.clone(),
180 number: self.number,
181 attempt: self.attempt,
182 event: self.event.clone(),
183 git_ref: self.git_ref.clone(),
184 sha: self.sha.clone(),
185 pull: self.pull,
186 status: self.status.clone(),
187 conclusion: self.conclusion.clone(),
188 error: self.error.clone(),
189 actor: self.actor.clone(),
190 created_at: self.created_at.clone(),
191 started_at: self.started_at.clone(),
192 finished_at: self.finished_at.clone(),
193 }
194 }
195}
196
197#[derive(Clone, Deserialize)]
198pub struct JobRow {
199 pub id: String,
200 pub run_id: String,
201 pub repo_id: String,
202 pub namespace: String,
203 pub key: String,
204 pub ordinal: u32,
205 pub name: String,
206 pub needs: String,
207 pub matrix: Option<String>,
208 pub status: String,
209 pub conclusion: Option<String>,
210 pub steps: String,
211 pub annotations: String,
212 pub outputs: String,
213 pub reason: Option<String>,
214 pub token_hash: Option<String>,
215 pub timeout_minutes: u32,
216 pub continue_on_error: u32,
217 pub max_parallel: Option<u32>,
218 /// Set for a job that calls a reusable workflow, and for that
219 /// workflow's jobs (see migration 0002).
220 pub call: Option<String>,
221 pub seen_at: Option<String>,
222 pub started_at: Option<String>,
223 pub finished_at: Option<String>,
224 /// For a job whose `runs-on` names self-hosted runners: what it asks
225 /// for, as a JSON array (see `g1t_contracts::runners::Wanted`), when it
226 /// started waiting, and the runner that took it (migration 0004).
227 #[serde(default)]
228 pub labels: Option<String>,
229 #[serde(default)]
230 pub queued_at: Option<String>,
231 #[serde(default)]
232 pub runner_id: Option<String>,
233 #[serde(default)]
234 pub runner_name: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'235 /// The environment it names, read when its needs were done; a job its
236 /// rules hold is `pending` (migration 0006).
237 #[serde(default)]
238 pub environment: Option<String>,
239 /// Its own `concurrency` group, and whether that cancels what it replaces.
240 #[serde(default)]
241 pub concurrency_group: Option<String>,
242 #[serde(default)]
243 pub cancel_in_progress: u32,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)244 /// When it was told to stop, while it runs its cleanup steps
245 /// (migration 0009).
246 #[serde(default)]
247 pub cancel_requested_at: Option<String>,
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.248 /// Whether it was given stand-ins for brokered secrets (migration
249 /// 0019), so its log is masked for their values too.
250 #[serde(default)]
251 pub brokered: u32,
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.252 /// The hash of the token it started with, once its harness swapped that
253 /// for another (migration 0020): good for `brokered_secrets` only.
254 #[serde(default)]
255 pub first_token_hash: Option<String>,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)256}
257
258/// How long a cancelled job has to run its `if: always()` and `cancelled()`
259/// steps and its post steps before it is stopped outright, as on GitHub.
260pub const CANCEL_GRACE_MS: u64 = 5 * 60 * 1000;
261
262/// Which jobs a re-run runs again.
263#[derive(Clone, Copy, Debug, PartialEq)]
264pub enum Rerun<'a> {
265 All,
266 /// Those that did not succeed.
267 Failed,
268 /// One job's key.
269 Job(&'a str),
270}
271
272/// The keys a re-run runs again, in `order`: those `which` names and,
273/// transitively, every key that needs one of them. `needs` gives a key's
274/// needs; `succeeded` whether every job of a key succeeded.
275pub fn rerun_keys(order: &[&str], needs: impl Fn(&str) -> Vec<String>, succeeded: impl Fn(&str) -> bool, which: Rerun) -> Vec<String> {
276 let mut again: Vec<String> = Vec::new();
277 for key in order {
278 let named = match which {
279 Rerun::All => true,
280 Rerun::Failed => !succeeded(key),
281 Rerun::Job(job) => *key == job,
282 };
283 if named || needs(key).iter().any(|need| again.contains(need)) {
284 again.push((*key).to_owned());
285 }
286 }
287 again
288}
289
290/// The key under `jobs:` a job belongs to: a called workflow's jobs
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9291/// (`build/test`, or `build[1]/test` for a matrix's second call) run again
292/// with the job that calls it (`build`).
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)293pub fn top_key(key: &str) -> &str {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9294 let first = key.split('/').next().unwrap_or(key);
295 first.split('[').next().unwrap_or(first)
296}
297
298/// What the keys of the jobs a caller's call makes start with: the
299/// caller's key, and for a matrix the combination's index after it
300/// (`build[1]`), so each combination's call has jobs of its own. A job id
301/// has no `[`, so the two never meet.
302fn call_prefix(key: &str, index: Option<usize>) -> String {
303 match index {
304 Some(index) => format!("{key}[{index}]"),
305 None => key.to_owned(),
306 }
307}
308
309/// Whether a run may call one more reusable workflow, `file`, `depth`
310/// levels below the workflow it started from, having called `called`
311/// already (one name per call; repeats are one workflow). As on GitHub: ten
312/// levels of workflows at most, the top one and nine reusable ones below
313/// it, and 50 different reusable workflows in a run.
314fn call_allowed(depth: u64, called: &[&str], file: &str) -> std::result::Result<(), String> {
315 if depth > MAX_CALL_DEPTH {
316 return Err(format!(
317 "`{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.",
318 depth + 1
319 ));
320 }
321 let mut unique: Vec<&str> = called.to_vec();
322 unique.push(file);
323 unique.sort_unstable();
324 unique.dedup();
325 if unique.len() > MAX_CALLED_WORKFLOWS {
326 return Err(format!("Calling `{file}` would make {} reusable workflows in this run: a run calls at most {MAX_CALLED_WORKFLOWS} different ones.", unique.len()));
327 }
328 Ok(())
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)329}
330
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9331/// What a job is shown as: its `name` (else its id), under the job whose
332/// call made it (`caller`), read with its matrix when it has an
333/// expression, else with the matrix's values after it.
334fn shown_name(job: &workflow::Job, caller: Option<&str>, combination: &Map<String, Value>, scope: &Scope) -> String {
335 let base = job.name.clone().unwrap_or(job.id.clone());
336 let base = match caller {
337 Some(caller) => format!("{caller} / {base}"),
338 None => base,
339 };
340 if expr::has_expression(&base) {
341 expr::interpolate(&base, scope).unwrap_or(base)
342 } else if job.matrix.is_some() {
343 matrix::job_name(&base, combination)
344 } else {
345 base
346 }
347}
348
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)349/// Debug logging for a job, as a re-run with it turned on gives GitHub's:
350/// `RUNNER_DEBUG=1` and `runner.debug`, and `ACTIONS_STEP_DEBUG` and
351/// `ACTIONS_RUNNER_DEBUG` set, which the runner reads as the secrets of
352/// those names (`::debug::` lines shown, and how each step's `if` read).
353pub fn debug_logging(variables: &mut Map<String, Value>, runner: &mut Value) {
354 variables.insert("RUNNER_DEBUG".into(), json!("1"));
355 variables.insert("ACTIONS_STEP_DEBUG".into(), json!("true"));
356 variables.insert("ACTIONS_RUNNER_DEBUG".into(), json!("true"));
357 runner["debug"] = json!("1");
Download ZIP reads blobs eight batches at a time and deflates files together358}
359
360impl JobRow {
361 pub fn needs(&self) -> Vec<String> {
362 serde_json::from_str(&self.needs).unwrap_or_default()
363 }
364
365 pub fn call(&self) -> Option<Value> {
366 self.call.as_deref().and_then(|call| serde_json::from_str(call).ok())
367 }
368
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9369 /// For a job that calls a reusable workflow: what the keys of the jobs
370 /// its call makes start with (`call_prefix`).
371 pub fn call_prefix(&self) -> Option<String> {
372 let call = self.call().filter(|call| call["role"] == "caller")?;
373 Some(call["prefix"].as_str().unwrap_or(&self.key).to_owned())
374 }
375
376 /// A job that calls a reusable workflow and has not called it yet: it
377 /// waits for room under its matrix's `max-parallel`, or in its
378 /// concurrency group.
379 pub fn holds_call(&self) -> bool {
380 self.status == "waiting" && self.call_prefix().is_some()
381 }
382
Download ZIP reads blobs eight batches at a time and deflates files together383 /// For a job of a called workflow: that workflow, the job's own id in
384 /// it, and the job.
385 pub fn callee(&self) -> Option<(Workflow, workflow::Job, Value)> {
386 let call = self.call().filter(|call| call["role"] == "callee")?;
387 let called = workflow::parse(call["source"].as_str()?).ok()?;
388 let job = called.jobs.iter().find(|job| call["job"].as_str() == Some(job.id.as_str()))?.clone();
389 Some((called, job, call))
390 }
391}
392
393/// A job to decide on: its key, its definition, and what it needs, as
394/// (name in `needs`, key of the jobs).
395type Unit = (String, workflow::Job, Vec<(String, String)>);
396
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9397/// How many levels of reusable workflows may sit below the workflow a run
398/// started from, as on GitHub (ten levels of workflows in all).
399const MAX_CALL_DEPTH: u64 = 9;
400/// How many different reusable workflows a run may call, as on GitHub.
401const MAX_CALLED_WORKFLOWS: usize = 50;
Download ZIP reads blobs eight batches at a time and deflates files together402
403/// What the jobs of one key came to, for `needs.<key>`.
404fn key_result(rows: &[&JobRow]) -> &'static str {
405 let failed = |row: &&&JobRow| row.conclusion.as_deref() == Some("failure") && row.continue_on_error == 0;
406 if rows.iter().any(|row| failed(&row)) {
407 "failure"
408 } else if rows.iter().any(|row| row.conclusion.as_deref() == Some("cancelled")) {
409 "cancelled"
410 } else if rows.iter().all(|row| row.conclusion.as_deref() == Some("skipped")) {
411 "skipped"
412 } else {
413 "success"
414 }
415}
416
417/// Whether any job before `key` failed: one it needs, or one those need,
418/// however far back. A job after a skipped one still sees the failure
419/// that skipped it, as GitHub's failure() does. `needs_of`: each key's
420/// needs, as keys; `failed`: whether a key's jobs came to a failure.
421fn ancestor_failed(needs_of: &std::collections::HashMap<&str, Vec<&str>>, key: &str, failed: impl Fn(&str) -> bool) -> bool {
422 let mut seen = std::collections::HashSet::new();
423 let mut stack: Vec<&str> = needs_of.get(key).cloned().unwrap_or_default();
424 while let Some(next) = stack.pop() {
425 if !seen.insert(next) {
426 continue;
427 }
428 if failed(next) {
429 return true;
430 }
431 stack.extend(needs_of.get(next).into_iter().flatten().copied());
432 }
433 false
434}
435
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97436/// A job's `environment:`, as deployments read it: its name, the address in
437/// `environment.url` once its expressions are filled in from the run, and
438/// whether the job deploys to it. A job with `deployment: false` only reads
439/// the environment's secrets and variables, and makes no deployment.
440#[derive(Clone, Debug, Default, PartialEq)]
441pub(crate) struct JobEnvironment {
442 pub(crate) name: String,
443 pub(crate) url: Option<String>,
444 pub(crate) deploys: bool,
445}
446
447/// The environment `environment:` names, as written in `raw` (a job), with
448/// `contexts` to fill in expressions: null when it names none, or only
449/// through an expression this cannot read before the job runs.
450pub(crate) fn environment_of(raw: &Value, contexts: &Map<String, Value>) -> Option<JobEnvironment> {
451 let scope = Scope { contexts, status: Status::Success, hash_files: None };
452 let plain = |value: &Value| -> Option<String> {
453 let text = match value {
454 Value::String(text) if text.contains("${{") => expr::interpolate_value(value, &scope).ok().map(|v| expr::to_text(&v))?,
455 Value::String(text) => text.clone(),
456 _ => return None,
457 };
458 let text = text.trim().to_owned();
459 (!text.is_empty()).then_some(text)
460 };
461 match raw.get("environment")? {
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9462 name @ Value::String(_) => Some(JobEnvironment { name: plain(name)?, url: None, deploys: workflow::records_deployment(raw) }),
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97463 Value::Object(env) => {
464 let name = plain(env.get("name")?)?;
465 let url = env
466 .get("url")
467 .and_then(plain)
468 .filter(|url| url.starts_with("https://") || url.starts_with("http://"));
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9469 Some(JobEnvironment { name, url, deploys: workflow::records_deployment(raw) })
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97470 }
471 _ => None,
472 }
473}
474
475/// The contexts a job's `runs-on` and `environment` are read with before it
476/// runs: the run's `github`, its inputs, and the job's matrix.
477fn start_contexts(run: &RunRow, job: &JobRow, key: &str) -> Map<String, Value> {
478 let mut contexts = Map::new();
479 contexts.insert("github".into(), run.info().context(key, "", run.action.as_deref()));
480 contexts.insert("inputs".into(), Value::Object(run.inputs()));
481 contexts.insert("matrix".into(), job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or_else(|| json!({})));
482 contexts
483}
484
485/// The job as its workflow (or the workflow it calls) defines it.
486fn job_spec(run: &RunRow, job: &JobRow) -> Option<workflow::Job> {
487 match job.callee() {
488 Some((_, spec, _)) => Some(spec),
489 None => workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)),
490 }
491}
492
Merge branch 'worktree-agent-a3abfcce648e87dca'493/// The environment a job deploys to, if it deploys: by the name read when
494/// its needs were done (which may have needed their outputs), else as
495/// its `environment:` reads now.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97496pub(crate) fn deploys_to(run: &RunRow, job: &JobRow) -> Option<JobEnvironment> {
497 let spec = job_spec(run, job)?;
Merge branch 'worktree-agent-a3abfcce648e87dca'498 let read = environment_of(&spec.raw, &start_contexts(run, job, &spec.id));
499 let env = match (read, &job.environment) {
500 (Some(env), Some(name)) => Some(JobEnvironment { name: name.clone(), ..env }),
501 (Some(env), None) => Some(env),
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r9502 (None, Some(name)) => Some(JobEnvironment { name: name.clone(), url: None, deploys: workflow::records_deployment(&spec.raw) }),
Merge branch 'worktree-agent-a3abfcce648e87dca'503 (None, None) => None,
504 };
505 env.filter(|env| env.deploys)
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97506}
507
Download ZIP reads blobs eight batches at a time and deflates files together508/// What the runner is told about a job it starts, from its workflow: the
509/// environment it names plainly, and the machine its `runs-on` asks for
510/// (`instance_for`; none for the standard one).
511#[derive(Default)]
512struct StartDetails {
513 environment: Option<String>,
514 instance: Option<String>,
515}
516
517fn start_details(run: &RunRow, job: &JobRow) -> StartDetails {
518 let spec = match job.callee() {
519 Some((_, spec, _)) => spec,
520 None => match workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)) {
521 Some(spec) => spec,
522 None => return StartDetails::default(),
523 },
524 };
Merge branch 'worktree-agent-a3abfcce648e87dca'525 // As read when its needs were done, an expression's included.
526 let environment = job.environment.clone();
Download ZIP reads blobs eight batches at a time and deflates files together527 // `runs-on` as the job was queued with: its matrix and the run's
528 // inputs. A label that needs more than those is the standard machine.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97529 let contexts = start_contexts(run, job, &spec.id);
Download ZIP reads blobs eight batches at a time and deflates files together530 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
531 let labels: Vec<String> = match expr::interpolate_value(&spec.runs_on, &scope).unwrap_or(Value::Null) {
532 Value::String(label) => vec![label],
533 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
534 Value::Object(given) => match given.get("labels") {
535 Some(Value::Array(labels)) => labels.iter().map(expr::to_text).collect(),
536 Some(label) => vec![expr::to_text(label)],
537 None => Vec::new(),
538 },
539 _ => Vec::new(),
540 };
541 let instance = g1t_contracts::actions::instance_for(&labels);
542 StartDetails {
543 environment,
544 instance: (instance != g1t_contracts::actions::STANDARD_INSTANCE).then(|| instance.label.to_owned()),
545 }
546}
547
548fn now() -> String {
549 rfc3339(now_ms())
550}
551
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97552/// How a finished run went for one environment, from the conclusions of
553/// the jobs that deploy to it: failed if any failed, an error if any was
554/// cancelled, else a success; nothing when none ran (all skipped).
555pub(crate) fn deployment_outcome(conclusions: &[Option<String>]) -> Option<&'static str> {
556 let ran: Vec<&str> = conclusions.iter().flatten().map(String::as_str).filter(|c| *c != "skipped").collect();
557 if ran.is_empty() {
558 return None;
559 }
560 if ran.iter().any(|c| *c == "failure" || *c == "timed_out") {
561 return Some("failure");
562 }
563 if ran.contains(&"cancelled") {
564 return Some("error");
565 }
566 Some("success")
567}
568
Download ZIP reads blobs eight batches at a time and deflates files together569impl Actions {
570 pub async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {
571 self.db.prepare("SELECT * FROM runs WHERE id = ?").bind(&[id.into()])?.first::<RunRow>(None).await
572 }
573
574 pub async fn job_rows(&self, run_id: &str) -> Result<Vec<JobRow>> {
575 self.db
576 .prepare("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid")
577 .bind(&[run_id.into()])?
578 .all()
579 .await?
580 .results::<JobRow>()
581 }
582
583 pub async fn run_summary(&self, id: &str) -> Result<Outcome<WorkflowRun>> {
584 Ok(match self.run_row(id).await? {
585 Some(row) => Outcome::Ok(row.summary()),
586 None => fail(FailureCode::NotFound, "No such run."),
587 })
588 }
589
590 /// The contexts every expression outside a job's steps may use.
591 fn base_contexts(run: &RunRow, vars: &Map<String, Value>, job: &str) -> Map<String, Value> {
592 let mut contexts = Map::new();
593 contexts.insert("github".into(), run.info().context(job, "", run.action.as_deref()));
594 contexts.insert("inputs".into(), Value::Object(run.inputs()));
595 contexts.insert("vars".into(), Value::Object(vars.clone()));
596 contexts.insert("needs".into(), json!({}));
597 contexts.insert("runner".into(), runner_context());
598 contexts
599 }
600
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'601 /// Whether `namespace` is a person's own workspace: usernames and
602 /// workspaces share one namespace, so it is when someone has that
603 /// username. Kept in the isolate for a few minutes, so a burst of
604 /// runs asks identity once.
605 async fn personal_namespace(&self, namespace: &str) -> Result<bool> {
606 const KEEP_MS: u64 = 5 * 60 * 1000;
607 thread_local! {
608 static PERSONAL: std::cell::RefCell<std::collections::HashMap<String, (bool, u64)>> = Default::default();
609 }
610 let now = now_ms();
611 let key = namespace.to_ascii_lowercase();
612 if let Some(personal) = PERSONAL.with(|kept| kept.borrow().get(&key).filter(|(_, at)| now.saturating_sub(*at) < KEEP_MS).map(|(personal, _)| *personal)) {
613 return Ok(personal);
614 }
615 let found: g1t_contracts::Viewer = g1t_kit::call(
616 &self.identity,
617 "user_by_username",
618 &g1t_contracts::identity::UsernameArgs { username: key.clone() },
619 )
620 .await?;
621 let personal = found.is_some_and(|user| user.username.eq_ignore_ascii_case(namespace));
622 PERSONAL.with(|kept| {
623 let mut kept = kept.borrow_mut();
624 if kept.len() > 1000 {
625 kept.clear();
626 }
627 kept.insert(key, (personal, now));
628 });
629 Ok(personal)
630 }
631
Download ZIP reads blobs eight batches at a time and deflates files together632 /// Makes a run and its jobs, and starts what can start. `None` when the
633 /// event already started this workflow.
634 pub async fn create_run(&self, new: NewRun) -> Result<Option<String>> {
635 // Nothing starts on an archived repository. A deleted one is never
636 // found to start anything on.
637 if new.repo.archived() {
638 return Ok(None);
639 }
640 let workflow_row = self.workflow_row(&new.repo, &new.path, &new.workflow.display_name(&new.path), &new.source).await?;
641 let numbered = self
642 .db
643 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
644 .bind(&[workflow_row.id.as_str().into()])?
645 .first::<Count>(None)
646 .await?
647 .map_or(1, |count| count.n);
648 let id = new_id("run", now_ms());
649 let mut info = new.info.clone();
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'650 // A personal workspace's repositories are a `User`'s, as on GitHub.
651 if self.personal_namespace(&new.repo.namespace).await? {
652 crate::payload::owned_by_person(&mut info.event, &new.repo.namespace);
653 }
Download ZIP reads blobs eight batches at a time and deflates files together654 info.workflow = new.workflow.display_name(&new.path);
655 info.workflow_path = new.path.clone();
656 info.run_id = id.clone();
657 info.run_number = u64::from(numbered);
658 let vars = self
659 .variables_for(&new.repo.id, &format!("{}/{}", new.repo.namespace, new.repo.name), None, new.trusted)
660 .await?;
661
662 // run-name and the concurrency group read github, inputs and vars.
663 let mut contexts = Map::new();
664 contexts.insert("github".into(), info.context("", "", new.action.as_deref()));
665 contexts.insert("inputs".into(), Value::Object(new.inputs.clone()));
666 contexts.insert("vars".into(), Value::Object(vars));
667 let scope = Scope {
668 contexts: &contexts,
669 status: Status::Success,
670 hash_files: None,
671 };
672 let title = new
673 .workflow
674 .run_name
675 .as_deref()
676 .and_then(|run_name| expr::interpolate(run_name, &scope).ok())
677 .filter(|title| !title.trim().is_empty())
678 .unwrap_or(new.title.clone());
679 let group = new.workflow.concurrency.as_ref().and_then(|c| expr::interpolate(&c.group, &scope).ok());
680 let cancel_in_progress = new
681 .workflow
682 .concurrency
683 .as_ref()
684 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
685 .is_some_and(|value| expr::truthy(&value));
686
Merge branch 'worktree-agent-a3abfcce648e87dca'687 let approval = new.approval.as_ref().map(|reason| RunApproval { state: "required".into(), reason: reason.clone(), approved_by: None });
Download ZIP reads blobs eight batches at a time and deflates files together688 let inserted = self
689 .db
690 .prepare(
691 "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'692 pull, status, actor, actor_id, source, info, inputs, trusted, concurrency_group, event_key, created_at,
693 approval, cancel_in_progress)
694 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
Download ZIP reads blobs eight batches at a time and deflates files together695 )
696 .bind(&[
697 id.as_str().into(),
698 workflow_row.id.as_str().into(),
699 new.repo.id.as_str().into(),
700 format!("{}/{}", new.repo.namespace, new.repo.name).into(),
701 new.path.as_str().into(),
702 info.workflow.as_str().into(),
703 title.as_str().into(),
704 numbered.into(),
705 info.event_name.as_str().into(),
706 optional(new.action.as_deref()),
707 info.git_ref.as_str().into(),
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'708 new.head_sha.as_deref().unwrap_or(&info.sha).into(),
Download ZIP reads blobs eight batches at a time and deflates files together709 new.pull.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
Merge branch 'worktree-agent-a3abfcce648e87dca'710 if approval.is_some() { "action_required" } else { "queued" }.into(),
Download ZIP reads blobs eight batches at a time and deflates files together711 optional(new.actor.as_deref()),
712 optional(new.actor_id.as_deref()),
713 new.source.as_str().into(),
714 serde_json::to_string(&info)?.into(),
715 serde_json::to_string(&new.inputs)?.into(),
716 u32::from(new.trusted).into(),
717 optional(group.as_deref()),
718 new.event_key.as_str().into(),
719 now().into(),
Merge branch 'worktree-agent-a3abfcce648e87dca'720 approval.as_ref().map(serde_json::to_string).transpose()?.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
721 u32::from(cancel_in_progress).into(),
Download ZIP reads blobs eight batches at a time and deflates files together722 ])?
723 .first::<Value>(None)
724 .await?;
725 if inserted.is_none() {
726 return Ok(None);
727 }
728 if let Some(run) = self.run_row(&id).await? {
729 self.report_pending(&run).await?;
730 }
731
732 // Every job, waiting; each is expanded when the jobs it needs are done.
733 let mut statements = Vec::new();
734 for job in &new.workflow.jobs {
735 statements.push(
736 self.db
737 .prepare(
738 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting')",
739 )
740 .bind(&[
741 new_id("job", now_ms()).into(),
742 id.as_str().into(),
743 new.repo.id.as_str().into(),
744 new.repo.namespace.as_str().into(),
745 job.id.as_str().into(),
746 job.name.clone().filter(|n| !expr::has_expression(n)).unwrap_or(job.id.clone()).into(),
747 serde_json::to_string(&job.needs)?.into(),
748 ])?,
749 );
750 }
751 self.db.batch(statements).await?;
752
Merge branch 'worktree-agent-a3abfcce648e87dca'753 // A pull request's run from outside waits to be approved; it joins
754 // its concurrency group once it is (protection.rs, approve_run).
755 if approval.is_some() {
756 return Ok(Some(id));
757 }
758 if let Some(run) = self.run_row(&id).await? {
759 self.enter_group(&run).await?;
760 }
761 Ok(Some(id))
762 }
763
764 /// A run about to start: one run at a time per concurrency group, as on
765 /// GitHub. A newer run replaces one that waits in the group, and with
766 /// `cancel-in-progress` the one that runs; otherwise it waits as
767 /// `pending` for the one that runs. Then it moves along.
768 pub(crate) async fn enter_group(&self, run: &RunRow) -> Result<()> {
769 if let Some(group) = &run.concurrency_group {
770 let cancel_in_progress = run.cancel_in_progress != 0;
Download ZIP reads blobs eight batches at a time and deflates files together771 let others = self
772 .db
Merge branch 'worktree-agent-a3abfcce648e87dca'773 .prepare(
774 "SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND id != ? AND status NOT IN ('completed', 'action_required') ORDER BY id",
775 )
776 .bind(&[run.repo_id.as_str().into(), group.as_str().into(), run.id.as_str().into()])?
Download ZIP reads blobs eight batches at a time and deflates files together777 .all()
778 .await?
779 .results::<RunRow>()?;
780 for other in &others {
781 if cancel_in_progress || other.status == "pending" {
782 // 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)783 self.cancel_run(other, "A newer run in the same concurrency group replaced it.", false).await?;
Download ZIP reads blobs eight batches at a time and deflates files together784 }
785 }
786 if !cancel_in_progress && others.iter().any(|other| other.status != "pending") {
787 self.db
788 .prepare("UPDATE runs SET status = 'pending' WHERE id = ?")
Merge branch 'worktree-agent-a3abfcce648e87dca'789 .bind(&[run.id.as_str().into()])?
Download ZIP reads blobs eight batches at a time and deflates files together790 .run()
791 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'792 return Ok(());
Download ZIP reads blobs eight batches at a time and deflates files together793 }
794 }
Merge branch 'worktree-agent-a3abfcce648e87dca'795 self.advance(&run.id).await
Download ZIP reads blobs eight batches at a time and deflates files together796 }
797
798 /// A run that could not start, such as for a workflow file that does not read.
799 #[allow(clippy::too_many_arguments)]
800 pub async fn record_failed_run(
801 &self,
802 row: &WorkflowRow,
803 git_ref: &str,
804 sha: &str,
805 event_key: &str,
806 actor_id: Option<&str>,
807 actor: &str,
808 problem: &str,
809 ) -> Result<()> {
810 let numbered = self
811 .db
812 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
813 .bind(&[row.id.as_str().into()])?
814 .first::<Count>(None)
815 .await?
816 .map_or(1, |count| count.n);
817 let at = now();
818 self.db
819 .prepare(
820 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, git_ref, sha, status,
821 conclusion, error, actor, actor_id, source, info, event_key, created_at, finished_at)
822 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'push', ?, ?, 'completed', 'failure', ?, ?, ?, ?, '{}', ?, ?, ?)",
823 )
824 .bind(&[
825 new_id("run", now_ms()).into(),
826 row.id.as_str().into(),
827 row.repo_id.as_str().into(),
828 row.repo.as_str().into(),
829 row.path.as_str().into(),
830 row.path.as_str().into(),
831 "Invalid workflow file".into(),
832 numbered.into(),
833 git_ref.into(),
834 sha.into(),
835 problem.into(),
836 actor.into(),
837 optional(actor_id),
838 row.source.as_str().into(),
839 event_key.into(),
840 at.as_str().into(),
841 at.as_str().into(),
842 ])?
843 .run()
844 .await?;
845 Ok(())
846 }
847
848 /// Moves a run along: jobs whose needs are done are decided on, and
849 /// jobs that can start are started.
850 pub async fn advance(&self, run_id: &str) -> Result<()> {
851 // Each pass may finish jobs (skipped ones), which may free others.
852 for _ in 0..20 {
853 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'854 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
Download ZIP reads blobs eight batches at a time and deflates files together855 return Ok(());
856 }
857 let workflow = match workflow::parse(&run.source) {
858 Ok(workflow) => workflow,
859 Err(problem) => {
860 self.finish_run(&run, Some(&problem)).await?;
861 return Ok(());
862 }
863 };
864 let jobs = self.job_rows(run_id).await?;
865 let mut changed = false;
866 // What to decide on: the workflow's jobs, and the jobs of the
867 // workflows they call, each with its needs as (name, key).
868 let mut units: Vec<Unit> = workflow
869 .jobs
870 .iter()
871 .map(|job| (job.id.clone(), job.clone(), job.needs.iter().map(|n| (n.clone(), n.clone())).collect()))
872 .collect();
873 let mut seen = std::collections::HashSet::new();
874 for row in &jobs {
875 if !seen.insert(row.key.clone()) {
876 continue;
877 }
878 if let Some((_, job, call)) = row.callee() {
879 let parent = call["parent"].as_str().unwrap_or_default().to_owned();
880 let needs = job.needs.iter().map(|n| (n.clone(), format!("{parent}/{n}"))).collect();
881 units.push((row.key.clone(), job, needs));
882 }
883 }
884 // Each key's needs, as keys, to look back through for failure().
885 let needs_of: std::collections::HashMap<&str, Vec<&str>> =
886 units.iter().map(|(key, _, needs)| (key.as_str(), needs.iter().map(|(_, need)| need.as_str()).collect())).collect();
887 let key_failed = |key: &str| key_result(&jobs.iter().filter(|row| row.key == key).collect::<Vec<_>>()) == "failure";
888 for (key, job, needs) in &units {
889 let rows: Vec<&JobRow> = jobs.iter().filter(|row| &row.key == key).collect();
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9890 if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting" && !row.holds_call()) {
Download ZIP reads blobs eight batches at a time and deflates files together891 continue;
892 }
893 let needed: Vec<(&String, Vec<&JobRow>)> =
894 needs.iter().map(|(name, need)| (name, jobs.iter().filter(|row| &row.key == need).collect())).collect();
895 if !needed.iter().all(|(_, rows)| rows.iter().all(|row| row.status == "completed")) {
896 continue;
897 }
898 let failed_before = ancestor_failed(&needs_of, key, key_failed);
899 self.decide(&run, job, rows[0], &needed, failed_before).await?;
900 changed = true;
901 }
902 // A job that called a workflow finishes with that workflow's jobs.
903 for row in jobs.iter().filter(|row| row.status == "calling") {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9904 let prefix = row.call_prefix().unwrap_or(row.key.clone());
Download ZIP reads blobs eight batches at a time and deflates files together905 let children: Vec<&JobRow> = jobs
906 .iter()
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9907 .filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(prefix.as_str())))
Download ZIP reads blobs eight batches at a time and deflates files together908 .collect();
909 if !children.is_empty() && children.iter().all(|child| child.status == "completed") {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9910 self.finish_call(&run, row, &children).await?;
Download ZIP reads blobs eight batches at a time and deflates files together911 changed = true;
912 }
913 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9914 // Calls that wait start while their matrix's `max-parallel` and
915 // their concurrency group have room.
916 let mut calling: std::collections::HashMap<&str, u32> = std::collections::HashMap::new();
917 for row in jobs.iter().filter(|row| row.status == "calling") {
918 *calling.entry(row.key.as_str()).or_default() += 1;
919 }
920 for row in jobs.iter().filter(|row| row.holds_call()) {
921 let running = calling.get(row.key.as_str()).copied().unwrap_or(0);
922 if row.max_parallel.is_some_and(|max| running >= max) || self.group_busy(row).await? {
923 continue;
924 }
925 if self.start_call(row).await? {
926 calling.insert(row.key.as_str(), running + 1);
927 changed = true;
928 }
929 }
Download ZIP reads blobs eight batches at a time and deflates files together930 if !changed {
931 break;
932 }
933 }
934 self.start_queued().await?;
935 self.finish_if_done(run_id).await
936 }
937
938 /// Decides on one job whose needs are done: skip it, fail it, or expand
939 /// it into its matrix and queue it. `failed_before`: whether any job
940 /// before it failed, however far back (`ancestor_failed`).
941 async fn decide(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, needed: &[(&String, Vec<&JobRow>)], failed_before: bool) -> Result<()> {
942 let vars = self.variables_for(&run.repo_id, &run.repo, None, run.trusted != 0).await?;
943 let mut contexts = Self::base_contexts(run, &vars, &job.id);
944 // A called workflow's jobs read the inputs they were called with.
945 let call = row.call();
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9946 // Shown under the job whose call made it, by that job's name.
947 let caller = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| {
948 c["title"].as_str().map(str::to_owned).or_else(|| c["parent"].as_str().map(|parent| parent.replace('/', " / ")))
949 });
Download ZIP reads blobs eight batches at a time and deflates files together950 if let Some(call) = call.as_ref().filter(|c| c["role"] == "callee") {
951 contexts.insert("inputs".into(), call["inputs"].clone());
952 }
953 let mut needs = Map::new();
954 let mut results = Vec::new();
955 for (key, rows) in needed {
956 let result = key_result(rows);
957 let mut outputs = Map::new();
958 for row in rows {
959 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
960 outputs.extend(more);
961 }
962 }
963 results.push(result);
964 needs.insert((*key).clone(), json!({ "result": result, "outputs": outputs }));
965 }
966 // As on GitHub: a need that was skipped makes success() false but
967 // not failure(); failure() is a failure anywhere before the job.
968 let status = expr::job_status(results, failed_before, run.conclusion.as_deref() == Some("cancelled"));
969 contexts.insert("needs".into(), Value::Object(needs));
970 let scope = Scope {
971 contexts: &contexts,
972 status,
973 hash_files: None,
974 };
975 let condition = job.condition.as_deref().unwrap_or_default();
976 match expr::condition(condition, &scope) {
977 Ok(true) => {}
978 Ok(false) => return self.skip_job(row, None).await,
979 Err(problem) => return self.fail_job(row, &format!("Its `if` does not read: {problem}")).await,
980 }
981 // Its matrix, which may come from a needed job's outputs.
982 let combinations = match &job.matrix {
983 None => vec![Map::new()],
984 Some(matrix) => {
985 let value = match expr::interpolate_value(matrix, &scope) {
986 Ok(value) => value,
987 Err(problem) => return self.fail_job(row, &format!("Its matrix does not read: {problem}")).await,
988 };
989 match matrix::expand(&value) {
990 Ok(combinations) if !combinations.is_empty() => combinations,
991 Ok(_) => return self.fail_job(row, "Its matrix makes no jobs.").await,
992 Err(problem) => return self.fail_job(row, &problem).await,
993 }
994 }
995 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r9996 // A job that calls a reusable workflow calls it once for each.
997 if let Some(uses) = &job.uses {
998 return self.call_workflow(run, job, row, uses, &contexts, &combinations, caller.as_deref()).await;
999 }
Download ZIP reads blobs eight batches at a time and deflates files together1000 let total = combinations.len();
1001 let raw = job.raw.as_object().cloned().unwrap_or_default();
1002 // Whether pull requests from forks may use self-hosted runners here,
1003 // asked once, and only for a run that is not trusted.
1004 let mut forks_allowed: Option<bool> = None;
Merge branch 'worktree-agent-a3abfcce648e87dca'1005 // Each concurrency group its jobs join, and whether it cancels.
1006 let mut groups: Vec<(String, bool)> = Vec::new();
Download ZIP reads blobs eight batches at a time and deflates files together1007 let mut statements = Vec::new();
1008 for (index, combination) in combinations.iter().enumerate() {
1009 let mut contexts = contexts.clone();
1010 contexts.insert("matrix".into(), Value::Object(combination.clone()));
1011 contexts.insert(
1012 "strategy".into(),
1013 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
1014 );
1015 let scope = Scope {
1016 contexts: &contexts,
1017 status: Status::Success,
1018 hash_files: None,
1019 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91020 let name = shown_name(job, caller.as_deref(), combination, &scope);
Download ZIP reads blobs eight batches at a time and deflates files together1021 let runs_on = expr::interpolate_value(&job.runs_on, &scope).unwrap_or(Value::Null);
1022 let labels = match &runs_on {
1023 Value::String(label) => vec![label.clone()],
1024 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
1025 Value::Object(spec) => spec.get("labels").map(|l| match l {
1026 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
1027 other => vec![expr::to_text(other)],
1028 }).unwrap_or_default(),
1029 _ => Vec::new(),
1030 };
1031 // `runs-on: self-hosted` (or a group): the workspace's own
1032 // machines, which may run any OS. Otherwise g1t's Linux sandboxes.
1033 let group = match &runs_on {
1034 Value::Object(spec) => spec.get("group").map(expr::to_text),
1035 _ => None,
1036 };
1037 let wanted = Wanted::of(&labels, group.as_deref());
1038 let mut reason = if wanted.self_hosted {
1039 None
1040 } else {
1041 labels
1042 .iter()
1043 .find(|label| {
1044 let lower = label.to_ascii_lowercase();
1045 lower.contains("windows") || lower.contains("macos")
1046 })
1047 .map(|label| {
1048 let os = if label.to_ascii_lowercase().contains("windows") { "windows" } else { "macos" };
1049 format!("`runs-on: {label}`: g1t's own runners are Linux. A self-hosted runner can run it: `runs-on: [self-hosted, {os}]`.")
1050 })
1051 };
1052 // A pull request from a fork runs code anyone could write: never
1053 // on the workspace's machines unless it said they may.
1054 if wanted.self_hosted && run.trusted == 0 {
1055 let allowed = match forks_allowed {
1056 Some(allowed) => allowed,
1057 None => {
1058 let allowed = self.effective_runner_settings(&row.namespace, Some(&run.repo_id)).await?.fork_pull_requests;
1059 forks_allowed = Some(allowed);
1060 allowed
1061 }
1062 };
1063 if !allowed {
1064 reason = Some(
1065 "Pull requests from forks do not run on self-hosted runners here. An admin can allow it under Settings, Runners.".to_owned(),
1066 );
1067 }
1068 }
1069 let max_minutes = if wanted.self_hosted { SELF_HOSTED_MAX_TIMEOUT_MINUTES } else { MAX_TIMEOUT_MINUTES };
1070 let timeout = raw
1071 .get("timeout-minutes")
1072 .and_then(|value| expr::interpolate_value(value, &scope).ok())
1073 .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.1074 .map_or(g1t_actions::workflow::DEFAULT_TIMEOUT_MINUTES.min(max_minutes), |minutes| (minutes.ceil() as u32).clamp(1, max_minutes));
Download ZIP reads blobs eight batches at a time and deflates files together1075 let continue_on_error = raw
1076 .get("continue-on-error")
1077 .and_then(|value| expr::interpolate_value(value, &scope).ok())
1078 .is_some_and(|value| expr::truthy(&value));
Merge branch 'worktree-agent-a3abfcce648e87dca'1079 let (mut status, mut conclusion, mut finished) = match &reason {
Download ZIP reads blobs eight batches at a time and deflates files together1080 Some(_) => ("completed", Some("failure"), Some(now())),
1081 None => ("queued", None, None),
1082 };
1083 // A self-hosted job waits, saying for what, until a runner takes it.
1084 let (labels_json, queued_at) = if wanted.self_hosted && reason.is_none() {
1085 reason = Some(waiting_reason(&wanted));
1086 (Some(serde_json::to_string(&wanted.stored())?), Some(now()))
1087 } else {
1088 (None, None)
1089 };
Merge branch 'worktree-agent-a3abfcce648e87dca'1090 // The environment it names, an expression read now, and what
1091 // that environment's protection rules say: it starts, waits as
1092 // `pending` (protection.rs), or may not deploy there.
1093 let environment = environment_of(&job.raw, &contexts).map(|env| env.name);
1094 if status == "queued"
1095 && let Some(name) = &environment
1096 {
1097 match self.gate(run, name).await? {
1098 Gate::Open => {}
1099 Gate::Held(why) => {
1100 status = "pending";
1101 reason = Some(why);
1102 }
1103 Gate::Refused(why) => {
1104 (status, conclusion, finished) = ("completed", Some("failure"), Some(now()));
1105 reason = Some(why);
1106 }
1107 }
1108 }
1109 // Its own concurrency group.
1110 let job_group = job
1111 .concurrency
1112 .as_ref()
1113 .and_then(|c| expr::interpolate(&c.group, &scope).ok())
1114 .map(|group| group.trim().to_owned())
1115 .filter(|group| !group.is_empty());
1116 let job_cancel = job
1117 .concurrency
1118 .as_ref()
1119 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
1120 .is_some_and(|value| expr::truthy(&value));
1121 if status != "completed"
1122 && let Some(group) = &job_group
1123 && !groups.iter().any(|(known, _)| known == group)
1124 {
1125 groups.push((group.clone(), job_cancel));
1126 }
Download ZIP reads blobs eight batches at a time and deflates files together1127 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
1128 name.into(),
1129 serde_json::to_string(combination)?.into(),
1130 status.into(),
1131 optional(conclusion),
1132 optional(reason.as_deref()),
1133 timeout.into(),
1134 u32::from(continue_on_error).into(),
1135 job.max_parallel.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1136 optional(finished.as_deref()),
1137 optional(labels_json.as_deref()),
1138 optional(queued_at.as_deref()),
Merge branch 'worktree-agent-a3abfcce648e87dca'1139 optional(environment.as_deref()),
1140 optional(job_group.as_deref()),
1141 u32::from(job_cancel).into(),
Download ZIP reads blobs eight batches at a time and deflates files together1142 ];
1143 if index == 0 {
1144 let mut bound = values;
1145 bound.push(row.id.as_str().into());
1146 statements.push(
1147 self.db
1148 .prepare(
1149 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, timeout_minutes = ?,
Merge branch 'worktree-agent-a3abfcce648e87dca'1150 continue_on_error = ?, max_parallel = ?, finished_at = ?, labels = ?, queued_at = ?, environment = ?,
1151 concurrency_group = ?, cancel_in_progress = ? WHERE id = ?",
Download ZIP reads blobs eight batches at a time and deflates files together1152 )
1153 .bind(&bound)?,
1154 );
1155 } else {
1156 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
1157 new_id("job", now_ms()).into(),
1158 row.run_id.as_str().into(),
1159 row.repo_id.as_str().into(),
1160 row.namespace.as_str().into(),
1161 row.key.as_str().into(),
1162 (index as u32).into(),
1163 row.needs.as_str().into(),
1164 optional(row.call.as_deref()),
1165 ];
1166 bound.extend(values);
1167 statements.push(
1168 self.db
1169 .prepare(
1170 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, call, name, matrix, status, conclusion, reason,
Merge branch 'worktree-agent-a3abfcce648e87dca'1171 timeout_minutes, continue_on_error, max_parallel, finished_at, labels, queued_at, environment,
1172 concurrency_group, cancel_in_progress)
1173 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Download ZIP reads blobs eight batches at a time and deflates files together1174 )
1175 .bind(&bound)?,
1176 );
1177 }
1178 }
1179 self.db.batch(statements).await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1180 self.replace_in_groups(run, &row.key, &groups).await
1181 }
1182
1183 /// A job queued in its own concurrency group: with `cancel-in-progress`
1184 /// it cancels the group's other jobs; otherwise it replaces any that
1185 /// are queued and not started, and waits for the one running
1186 /// (`start_queued` starts one of a group at a time). As on GitHub.
1187 async fn replace_in_groups(&self, run: &RunRow, key: &str, groups: &[(String, bool)]) -> Result<()> {
1188 if groups.is_empty() {
1189 return Ok(());
1190 }
1191 #[derive(Deserialize)]
1192 struct Id {
1193 id: String,
1194 }
1195 let mine: Vec<String> = self
1196 .db
1197 .prepare("SELECT id FROM jobs WHERE run_id = ? AND key = ?")
1198 .bind(&[run.id.as_str().into(), key.into()])?
1199 .all()
1200 .await?
1201 .results::<Id>()?
1202 .into_iter()
1203 .map(|row| row.id)
1204 .collect();
1205 let mut moved: Vec<String> = Vec::new();
1206 for (group, cancel) in groups {
1207 let others = self
1208 .db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91209 // A call that waits in the group, or calls, is in it too.
1210 .prepare("SELECT * FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('waiting', 'queued', 'pending', 'in_progress', 'calling')")
Merge branch 'worktree-agent-a3abfcce648e87dca'1211 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1212 .all()
1213 .await?
1214 .results::<JobRow>()?;
1215 for other in others.iter().filter(|other| !mine.contains(&other.id)) {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91216 if *cancel || matches!(other.status.as_str(), "queued" | "waiting") {
Merge branch 'worktree-agent-a3abfcce648e87dca'1217 self.stop_job(other, "A newer job in the same concurrency group replaced it.").await?;
1218 if !moved.contains(&other.run_id) {
1219 moved.push(other.run_id.clone());
1220 }
1221 }
1222 }
1223 }
1224 for other_run in moved.iter().filter(|id| **id != run.id) {
1225 Box::pin(self.advance(other_run)).await?;
1226 }
Download ZIP reads blobs eight batches at a time and deflates files together1227 Ok(())
1228 }
1229
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1230 /// A job that calls a reusable workflow, in the repository or another
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91231 /// (reach.rs): once for each combination of its matrix, as on GitHub,
1232 /// each with the inputs and secrets it passes read with that
1233 /// combination's `matrix`. Each call waits (`holds_call`) until its
1234 /// matrix's `max-parallel` and its concurrency group have room, when
1235 /// `start_call` puts the workflow's jobs in the run under it.
1236 #[allow(clippy::too_many_arguments)]
1237 async fn call_workflow(
1238 &self,
1239 run: &RunRow,
1240 job: &workflow::Job,
1241 row: &JobRow,
1242 uses: &str,
1243 contexts: &Map<String, Value>,
1244 combinations: &[Map<String, Value>],
1245 caller: Option<&str>,
1246 ) -> Result<()> {
Download ZIP reads blobs eight batches at a time and deflates files together1247 let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1;
1248 if depth > MAX_CALL_DEPTH {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91249 return self.fail_job(row, &call_allowed(depth, &[], uses).unwrap_err()).await;
Download ZIP reads blobs eight batches at a time and deflates files together1250 }
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1251 let (file, source, origin) = match self.called_workflow(run, row, uses).await? {
1252 Ok(found) => found,
1253 Err(why) => return self.fail_job(row, &why).await,
Download ZIP reads blobs eight batches at a time and deflates files together1254 };
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91255 let jobs = self.job_rows(&run.id).await?;
1256 // Each call so far, by the workflow it called.
1257 let called_paths: Vec<String> = jobs
1258 .iter()
1259 .filter(|other| other.call_prefix().is_some())
1260 .filter_map(JobRow::call)
1261 .filter_map(|call| call["path"].as_str().map(str::to_owned))
1262 .collect();
1263 let called_already: Vec<&str> = called_paths.iter().map(String::as_str).collect();
1264 if let Err(why) = call_allowed(depth, &called_already, &file) {
1265 return self.fail_job(row, &why).await;
1266 }
Download ZIP reads blobs eight batches at a time and deflates files together1267 let called = match workflow::parse(&source) {
1268 Ok(called) => called,
1269 Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await,
1270 };
1271 let Some(trigger) = called.trigger("workflow_call") else {
1272 return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await;
1273 };
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1274 // Secrets: none but the job's token unless `secrets:` passes them,
1275 // by name or with `inherit`; read when each job starts (job_spec).
1276 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-7nm8r91277 let total = combinations.len();
1278 let mut groups: Vec<(String, bool)> = Vec::new();
1279 let mut statements = Vec::new();
1280 for (index, combination) in combinations.iter().enumerate() {
1281 let mut contexts = contexts.clone();
1282 contexts.insert("matrix".into(), Value::Object(combination.clone()));
1283 contexts.insert(
1284 "strategy".into(),
1285 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
1286 );
1287 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1288 let name = shown_name(job, caller, combination, &scope);
1289 // Inputs: what `with:` passes, as the called workflow types them.
1290 let given = match job.raw.get("with") {
1291 Some(with) => match expr::interpolate_value(with, &scope) {
1292 Ok(Value::Object(given)) => Ok(given),
1293 Ok(_) => Ok(Map::new()),
1294 Err(problem) => Err(format!("Its `with` does not read: {problem}")),
1295 },
1296 None => Ok(Map::new()),
1297 };
1298 let inputs = given.and_then(|given| crate::reach::call_inputs(&file, &trigger.inputs, &given));
1299 let secrets = crate::reach::secrets_plan(&job.raw, &contexts, outer.clone());
1300 let problem = match &inputs {
1301 Err(why) => Some(why.clone()),
1302 Ok(_) => crate::reach::missing_secrets(&called.raw, &secrets)
1303 .first()
1304 .map(|name| format!("`{file}` needs the secret `{name}`: pass it under `secrets:`, or use `secrets: inherit`."))
1305 .or_else(|| {
1306 crate::reach::undeclared_secrets(&called.raw, &secrets)
1307 .first()
1308 .map(|name| format!("Invalid secret: `{name}` is not defined in `{file}`."))
1309 }),
1310 };
1311 // Its own concurrency group, which it holds while its workflow runs.
1312 let group = job
1313 .concurrency
1314 .as_ref()
1315 .and_then(|c| expr::interpolate(&c.group, &scope).ok())
1316 .map(|group| group.trim().to_owned())
1317 .filter(|group| !group.is_empty());
1318 let cancel = job
1319 .concurrency
1320 .as_ref()
1321 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
1322 .is_some_and(|value| expr::truthy(&value));
1323 if problem.is_none()
1324 && let Some(group) = &group
1325 && !groups.iter().any(|(known, _)| known == group)
1326 {
1327 groups.push((group.clone(), cancel));
1328 }
1329 let call = json!({
1330 "role": "caller", "path": file, "source": source, "origin": origin, "depth": depth,
1331 "prefix": call_prefix(&row.key, job.matrix.is_some().then_some(index)), "title": name,
1332 "inputs": inputs.unwrap_or_default(), "secrets": secrets, "fail_fast": job.fail_fast,
1333 });
1334 let (status, conclusion, finished) = match &problem {
1335 Some(_) => ("completed", Some("failure"), Some(now())),
1336 None => ("waiting", None, None),
1337 };
1338 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
1339 name.as_str().into(),
1340 serde_json::to_string(combination)?.into(),
1341 status.into(),
1342 optional(conclusion),
1343 optional(problem.as_deref()),
1344 job.max_parallel.map(|max| max.max(1)).map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1345 optional(finished.as_deref()),
1346 optional(group.as_deref()),
1347 u32::from(cancel).into(),
1348 serde_json::to_string(&call)?.into(),
1349 ];
1350 if index == 0 {
1351 let mut bound = values;
1352 bound.push(row.id.as_str().into());
1353 statements.push(
1354 self.db
1355 .prepare(
1356 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, max_parallel = ?, finished_at = ?,
1357 concurrency_group = ?, cancel_in_progress = ?, call = ? WHERE id = ?",
1358 )
1359 .bind(&bound)?,
1360 );
1361 } else {
1362 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
1363 new_id("job", now_ms()).into(),
1364 row.run_id.as_str().into(),
1365 row.repo_id.as_str().into(),
1366 row.namespace.as_str().into(),
1367 row.key.as_str().into(),
1368 (index as u32).into(),
1369 row.needs.as_str().into(),
1370 ];
1371 bound.extend(values);
1372 statements.push(
1373 self.db
1374 .prepare(
1375 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, name, matrix, status, conclusion, reason,
1376 max_parallel, finished_at, concurrency_group, cancel_in_progress, call)
1377 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
1378 )
1379 .bind(&bound)?,
1380 );
1381 }
1382 }
1383 self.db.batch(statements).await?;
1384 self.replace_in_groups(run, &row.key, &groups).await
1385 }
1386
1387 /// Whether another job of `row`'s concurrency group runs, or calls a
1388 /// workflow, so that `row` waits for it.
1389 async fn group_busy(&self, row: &JobRow) -> Result<bool> {
1390 let Some(group) = &row.concurrency_group else { return Ok(false) };
1391 let busy = self
1392 .db
1393 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?")
1394 .bind(&[row.repo_id.as_str().into(), group.as_str().into(), row.id.as_str().into()])?
1395 .first::<Count>(None)
1396 .await?
1397 .map_or(0, |count| count.n);
1398 Ok(busy > 0)
1399 }
1400
1401 /// A call that waited (`holds_call`) is made: the called workflow's
1402 /// jobs join the run under it, named under its name and keyed under its
1403 /// prefix. False when it no longer waits.
1404 async fn start_call(&self, row: &JobRow) -> Result<bool> {
1405 let Some(call) = row.call() else { return Ok(false) };
1406 let Some(prefix) = row.call_prefix() else { return Ok(false) };
1407 let Some(called) = call["source"].as_str().and_then(|source| workflow::parse(source).ok()) else { return Ok(false) };
1408 let file = call["path"].as_str().unwrap_or_default();
1409 let started = self
1410 .db
1411 .prepare("UPDATE jobs SET status = 'calling', reason = ?, started_at = ? WHERE id = ? AND status = 'waiting' RETURNING id")
1412 .bind(&[format!("Calls `{file}`.").into(), now().into(), row.id.as_str().into()])?
1413 .first::<Value>(None)
1414 .await?;
1415 if started.is_none() {
1416 return Ok(false);
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1417 }
Download ZIP reads blobs eight batches at a time and deflates files together1418 let mut statements = Vec::new();
1419 for called_job in &called.jobs {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91420 let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{prefix}/{n}")).collect();
1421 let callee = json!({
1422 "role": "callee", "parent": prefix, "title": row.name, "job": called_job.id, "path": file,
1423 "source": call["source"], "inputs": call["inputs"], "depth": call["depth"], "origin": call["origin"], "secrets": call["secrets"],
Download ZIP reads blobs eight batches at a time and deflates files together1424 });
1425 statements.push(
1426 self.db
1427 .prepare("INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status, call) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting', ?)")
1428 .bind(&[
1429 new_id("job", now_ms()).into(),
1430 row.run_id.as_str().into(),
1431 row.repo_id.as_str().into(),
1432 row.namespace.as_str().into(),
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91433 format!("{prefix}/{}", called_job.id).into(),
Download ZIP reads blobs eight batches at a time and deflates files together1434 format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(),
1435 serde_json::to_string(&needs)?.into(),
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91436 serde_json::to_string(&callee)?.into(),
Download ZIP reads blobs eight batches at a time and deflates files together1437 ])?,
1438 );
1439 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91440 if !statements.is_empty() {
1441 self.db.batch(statements).await?;
1442 }
1443 Ok(true)
1444 }
1445
1446 /// The runs with a call waiting in `group`, but `run_id`, moved along
1447 /// now that a job of the group has finished.
1448 async fn wake_group(&self, repo_id: &str, group: &str, run_id: &str) -> Result<()> {
1449 #[derive(Deserialize)]
1450 struct RunId {
1451 run_id: String,
1452 }
1453 let waiting = self
1454 .db
1455 .prepare("SELECT DISTINCT run_id FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'waiting' AND run_id != ?")
1456 .bind(&[repo_id.into(), group.into(), run_id.into()])?
1457 .all()
1458 .await?
1459 .results::<RunId>()?;
1460 for other in waiting {
1461 Box::pin(self.advance(&other.run_id)).await?;
1462 }
Download ZIP reads blobs eight batches at a time and deflates files together1463 Ok(())
1464 }
1465
1466 /// A job that called a workflow, finished with its jobs: their result,
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91467 /// and the outputs the workflow declares. A failed call of a fail-fast
1468 /// matrix stops the matrix's other calls.
1469 async fn finish_call(&self, run: &RunRow, row: &JobRow, children: &[&JobRow]) -> Result<()> {
Download ZIP reads blobs eight batches at a time and deflates files together1470 let call = row.call().unwrap_or_default();
1471 let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok());
1472 let mut jobs_context = Map::new();
1473 let mut by_key: std::collections::BTreeMap<String, Vec<&JobRow>> = std::collections::BTreeMap::new();
1474 for child in children {
1475 by_key.entry(child.key.clone()).or_default().push(child);
1476 }
1477 for (key, rows) in &by_key {
1478 let mut outputs = Map::new();
1479 for child in rows {
1480 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&child.outputs) {
1481 outputs.extend(more);
1482 }
1483 }
1484 let id = key.rsplit('/').next().unwrap_or(key);
1485 jobs_context.insert(id.to_owned(), json!({ "result": key_result(rows), "outputs": outputs }));
1486 }
1487 let inputs = children.first().and_then(|c| c.call()).map(|c| c["inputs"].clone()).unwrap_or(json!({}));
1488 let mut contexts = Map::new();
1489 contexts.insert("jobs".into(), Value::Object(jobs_context));
1490 contexts.insert("inputs".into(), inputs);
1491 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1492 let mut outputs = Map::new();
1493 if let Some(Value::Object(declared)) = called.as_ref().map(|w| {
1494 let on = w.raw.get("on").or_else(|| w.raw.get("true")).cloned().unwrap_or(Value::Null);
1495 on.get("workflow_call").and_then(|c| c.get("outputs")).cloned().unwrap_or(Value::Null)
1496 }) {
1497 for (name, spec) in declared {
1498 if let Some(value) = spec.get("value") {
1499 let value = expr::interpolate_value(value, &scope).unwrap_or(Value::Null);
1500 outputs.insert(name, Value::String(expr::to_text(&value)));
1501 }
1502 }
1503 }
1504 let conclusion = key_result(children);
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91505 let finished = self
1506 .db
1507 .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling' RETURNING id")
Download ZIP reads blobs eight batches at a time and deflates files together1508 .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-7nm8r91509 .first::<Value>(None)
Download ZIP reads blobs eight batches at a time and deflates files together1510 .await?;
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91511 if finished.is_none() {
1512 return Ok(());
1513 }
1514 if conclusion == "failure" && row.matrix.as_deref().is_some_and(|m| m != "{}") && call["fail_fast"] != json!(false) {
1515 let siblings = self
1516 .db
1517 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1518 .bind(&[row.run_id.as_str().into(), row.key.as_str().into()])?
1519 .all()
1520 .await?
1521 .results::<JobRow>()?;
1522 for sibling in siblings {
1523 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1524 }
1525 }
1526 if let Some(group) = &row.concurrency_group {
1527 self.wake_group(&run.repo_id, group, &run.id).await?;
1528 }
Download ZIP reads blobs eight batches at a time and deflates files together1529 Ok(())
1530 }
1531
1532 /// A file's text at a commit, if it is there.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1533 pub(crate) async fn read_file(&self, path: &RepoPath, ws: &g1t_contracts::User, sha: &str, file: &str) -> Result<Option<String>> {
Download ZIP reads blobs eight batches at a time and deflates files together1534 let blob: Outcome<g1t_contracts::repos::BlobView> = g1t_kit::call(
1535 &self.repos,
1536 "blob",
1537 &g1t_contracts::repos::BlobArgs {
1538 path: path.clone(),
1539 viewer: Some(ws.clone()),
1540 git_ref: sha.to_owned(),
1541 file_path: file.to_owned(),
1542 },
1543 )
1544 .await?;
1545 Ok(match blob {
1546 Outcome::Ok(view) => view.text,
1547 Outcome::Fail(_) => None,
1548 })
1549 }
1550
1551 async fn skip_job(&self, row: &JobRow, reason: Option<&str>) -> Result<()> {
1552 self.db
1553 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'skipped', reason = ?, finished_at = ? WHERE id = ?")
1554 .bind(&[optional(reason), now().into(), row.id.as_str().into()])?
1555 .run()
1556 .await?;
1557 Ok(())
1558 }
1559
1560 async fn fail_job(&self, row: &JobRow, reason: &str) -> Result<()> {
1561 self.db
1562 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'failure', reason = ?, finished_at = ? WHERE id = ? AND status != 'completed'")
1563 .bind(&[reason.into(), now().into(), row.id.as_str().into()])?
1564 .run()
1565 .await?;
1566 Ok(())
1567 }
1568
1569 /// Starts queued jobs, oldest first, while their workspace has room.
1570 pub async fn start_queued(&self) -> Result<()> {
1571 let queued = self
1572 .db
1573 // Self-hosted jobs are taken by their runners (runners.rs).
1574 .prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NULL ORDER BY rowid LIMIT 50")
1575 .all()
1576 .await?
1577 .results::<JobRow>()?;
1578 let mut running: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
1579 for job in queued {
1580 let in_workspace = match running.get(&job.namespace) {
1581 Some(n) => *n,
1582 None => {
1583 let n = self
1584 .db
1585 // Only g1t's own sandboxes count against the workspace's room.
1586 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE status = 'in_progress' AND namespace = ? AND runner_id IS NULL")
1587 .bind(&[job.namespace.as_str().into()])?
1588 .first::<Count>(None)
1589 .await?
1590 .map_or(0, |count| count.n);
1591 running.insert(job.namespace.clone(), n);
1592 n
1593 }
1594 };
1595 if in_workspace >= RUNNING_PER_WORKSPACE {
1596 continue;
1597 }
1598 if let Some(max) = job.max_parallel {
1599 let siblings = self
1600 .db
1601 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")
1602 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1603 .first::<Count>(None)
1604 .await?
1605 .map_or(0, |count| count.n);
1606 if siblings >= max {
1607 continue;
1608 }
1609 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1610 // One job of a concurrency group runs at a time.
1611 if let Some(group) = &job.concurrency_group {
1612 let running = self
1613 .db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91614 .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'1615 .bind(&[job.repo_id.as_str().into(), group.as_str().into(), job.id.as_str().into()])?
1616 .first::<Count>(None)
1617 .await?
1618 .map_or(0, |count| count.n);
1619 if running > 0 {
1620 continue;
1621 }
1622 }
Download ZIP reads blobs eight batches at a time and deflates files together1623 let token = random_hex(24);
1624 let at = now();
1625 let claimed = self
1626 .db
1627 .prepare(
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.1628 "UPDATE jobs SET status = 'in_progress', token_hash = ?, first_token_hash = NULL, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
Download ZIP reads blobs eight batches at a time and deflates files together1629 )
1630 .bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
1631 .first::<Value>(None)
1632 .await?;
1633 if claimed.is_none() {
1634 continue;
1635 }
1636 running.insert(job.namespace.clone(), in_workspace + 1);
1637 self.db
1638 .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")
1639 .bind(&[at.as_str().into(), job.run_id.as_str().into()])?
1640 .run()
1641 .await?;
1642 let run = self.run_row(&job.run_id).await?;
1643 let repo: RepoPath = run.as_ref().map(|run| repo_path(&run.repo)).unwrap_or(RepoPath {
1644 namespace: job.namespace.clone(),
1645 name: String::new(),
1646 });
1647 // Its environment and the machine it asked for, for the runner.
1648 let details = run.as_ref().map(|run| start_details(run, &job)).unwrap_or_default();
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.1649 let trusted = run.as_ref().is_some_and(|run| run.trusted != 0);
1650 // Only a trusted run's jobs get secrets, brokered ones included.
1651 let brokered = trusted && self.may_broker(&job.repo_id, &job.namespace).await?;
Download ZIP reads blobs eight batches at a time and deflates files together1652 let started: Outcome<Value> = g1t_kit::call(
1653 &self.runner,
1654 "start_actions_job",
1655 &StartJobArgs {
1656 job: job.id.clone(),
1657 token,
1658 repo,
1659 timeout_minutes: job.timeout_minutes,
1660 workflow: run.as_ref().map(|run| run.path.clone()),
1661 environment: details.environment,
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.1662 trusted,
Download ZIP reads blobs eight batches at a time and deflates files together1663 instance: details.instance,
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.1664 brokered,
Download ZIP reads blobs eight batches at a time and deflates files together1665 },
1666 )
1667 .await
1668 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
1669 if let Outcome::Fail(refused) = started {
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1670 // Billing would not pay for a scheduled run's job: the
1671 // schedule waits rather than making runs only to refuse them.
1672 if refused.code == FailureCode::PaymentRequired
1673 && let Some(run) = run.as_ref().filter(|run| run.event == "schedule")
1674 {
1675 self.db
1676 .prepare("UPDATE workflows SET schedule_refused_until = ? WHERE id = ?")
1677 .bind(&[rfc3339(now_ms() + crate::SCHEDULE_REFUSED_MS).into(), run.workflow_id.as_str().into()])?
1678 .run()
1679 .await?;
1680 }
Download ZIP reads blobs eight batches at a time and deflates files together1681 Box::pin(self.finish_job(&job.id, "failure", Some(&refused.message), None)).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971682 } else {
1683 self.job_started(&job.id).await?;
Download ZIP reads blobs eight batches at a time and deflates files together1684 }
1685 }
1686 Ok(())
1687 }
1688
1689 /// Finishes a job and moves its run along.
1690 pub async fn finish_job(&self, job_id: &str, conclusion: &str, reason: Option<&str>, outputs: Option<&Map<String, Value>>) -> Result<()> {
1691 let finished = self
1692 .db
1693 .prepare(
1694 "UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.1695 finished_at = ?, token_hash = NULL, first_token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
Download ZIP reads blobs eight batches at a time and deflates files together1696 )
1697 .bind(&[
1698 conclusion.into(),
1699 optional(reason),
1700 outputs.map(|o| serde_json::to_string(o).unwrap_or_default()).as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1701 now().into(),
1702 job_id.into(),
1703 ])?
1704 .first::<JobRow>(None)
1705 .await?;
1706 let Some(job) = finished else { return Ok(()) };
Merge branch 'worktree-agent-a5c94df323642630e'1707 self.tell_live(&job.repo_id, &job.run_id, None, None).await;
Merge branch 'worktree-agent-a3abfcce648e87dca'1708 // Its G1T_TOKEN stops working with it.
1709 self.revoke_job_tokens(&job.id).await;
Merge GitHub pull request #8 (claude/cool-newton-7nm8r9): Docker deployments on g1t.page behind a switch that stays off until Cloudflare's container APIs are checked live, workflows a workspace manages in its repositories, tarball and zipball downloads, refs/pull/* kept to g1t alone, framework detection for g1t.page builds, and a workflow's event payload matching GitHub's.1710 // A g1t.page build it opened and did not finish fails now, not at
1711 // the deployments sweep. Only a job that ran can have opened one.
1712 if job.started_at.is_some() {
1713 self.end_job_builds(&job.id, conclusion).await;
1714 }
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971715 // A job that deploys failed: so did its run's deployment, now.
1716 if conclusion == "failure" && job.continue_on_error == 0 && job.started_at.is_some()
1717 && let Some(run) = self.run_row(&job.run_id).await?
1718 && let Some(env) = deploys_to(&run, &job)
1719 {
1720 self.report_deployment(&run, &env, "failure", false).await;
1721 }
Download ZIP reads blobs eight batches at a time and deflates files together1722 // A self-hosted runner's job: the runner is free again, and its time
1723 // is recorded, at nothing.
1724 if job.runner_id.is_some() {
1725 self.released(&job).await?;
1726 }
1727 // Steps still marked as going are not going any more.
1728 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
1729 let mut touched = false;
1730 for step in steps.iter_mut() {
1731 if step["status"] != "completed" {
1732 let was_running = step["status"] == "in_progress";
1733 step["status"] = json!("completed");
1734 step["conclusion"] = json!(if was_running { conclusion } else { "skipped" });
1735 touched = true;
1736 }
1737 }
1738 if touched {
1739 self.db
1740 .prepare("UPDATE jobs SET steps = ? WHERE id = ?")
1741 .bind(&[serde_json::to_string(&steps)?.into(), job.id.as_str().into()])?
1742 .run()
1743 .await?;
1744 }
1745 // fail-fast: one failed combination stops the rest of its matrix.
1746 if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") {
1747 let run = self.run_row(&job.run_id).await?;
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91748 // As its workflow, or the workflow it is a job of, says.
1749 let fail_fast = run.as_ref().and_then(|run| job_spec(run, &job)).is_none_or(|j| j.fail_fast);
Download ZIP reads blobs eight batches at a time and deflates files together1750 if fail_fast {
1751 let siblings = self
1752 .db
1753 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1754 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1755 .all()
1756 .await?
1757 .results::<JobRow>()?;
1758 for sibling in siblings {
1759 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1760 }
1761 }
1762 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91763 // A call waiting in its concurrency group, in another run, may start.
1764 if let Some(group) = &job.concurrency_group {
1765 self.wake_group(&job.repo_id, group, &job.run_id).await?;
1766 }
Download ZIP reads blobs eight batches at a time and deflates files together1767 Box::pin(self.advance(&job.run_id)).await
1768 }
1769
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1770 /// Cancels a job. One that is running is told to stop (the answer to
1771 /// its next report), and runs its `if: always()` and `cancelled()`
1772 /// steps and its post steps before it ends `cancelled`; one that has
1773 /// not started is cancelled at once.
Download ZIP reads blobs eight batches at a time and deflates files together1774 async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91775 // A job calling a workflow stops with that workflow's jobs.
1776 if job.status == "calling"
1777 && let Some(prefix) = job.call_prefix()
1778 {
1779 let under = format!("{prefix}/");
1780 for child in self.job_rows(&job.run_id).await?.iter().filter(|child| child.key.starts_with(&under) && child.status != "completed") {
1781 Box::pin(self.stop_job(child, reason)).await?;
1782 }
1783 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1784 if job.status == "in_progress" && job.started_at.is_some() {
1785 self.db
1786 .prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'")
1787 .bind(&[now().into(), reason.into(), job.id.as_str().into()])?
1788 .run()
1789 .await?;
1790 return Ok(());
1791 }
1792 self.hard_stop(job, reason).await
1793 }
1794
1795 /// Cancels a job outright, stopping its sandbox if it has one.
1796 async fn hard_stop(&self, job: &JobRow, reason: &str) -> Result<()> {
Download ZIP reads blobs eight batches at a time and deflates files together1797 // A self-hosted runner hears it was cancelled on its next poll.
1798 if job.status == "in_progress" && job.runner_id.is_none() {
1799 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
1800 }
1801 let stopped = self
1802 .db
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.1803 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL, first_token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *")
Download ZIP reads blobs eight batches at a time and deflates files together1804 .bind(&[reason.into(), now().into(), job.id.as_str().into()])?
1805 .first::<JobRow>(None)
1806 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1807 if stopped.as_ref().is_some_and(|row| row.started_at.is_some()) {
1808 self.revoke_job_tokens(&job.id).await;
Merge GitHub pull request #8 (claude/cool-newton-7nm8r9): Docker deployments on g1t.page behind a switch that stays off until Cloudflare's container APIs are checked live, workflows a workspace manages in its repositories, tarball and zipball downloads, refs/pull/* kept to g1t alone, framework detection for g1t.page builds, and a workflow's event payload matching GitHub's.1809 self.end_job_builds(&job.id, "cancelled").await;
Merge branch 'worktree-agent-a3abfcce648e87dca'1810 }
Download ZIP reads blobs eight batches at a time and deflates files together1811 if let Some(stopped) = stopped.filter(|row| row.runner_id.is_some()) {
1812 self.released(&stopped).await?;
1813 }
1814 Ok(())
1815 }
1816
Merge branch 'worktree-agent-a3abfcce648e87dca'1817 /// Ends a job's tokens at once. A failure is logged: the token expires
1818 /// on its own soon after the job's time limit.
1819 async fn revoke_job_tokens(&self, job_id: &str) {
1820 let revoked: Result<bool> =
1821 g1t_kit::call(&self.identity, "revoke_job_tokens", &RevokeJobTokensArgs { job_id: job_id.to_owned() }).await;
1822 if let Err(error) = revoked {
1823 worker::console_error!("actions: the tokens of job {job_id} were not revoked: {error}");
1824 }
1825 }
1826
Download ZIP reads blobs eight batches at a time and deflates files together1827 /// Finishes the run when every job has.
1828 async fn finish_if_done(&self, run_id: &str) -> Result<()> {
1829 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1830 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
Download ZIP reads blobs eight batches at a time and deflates files together1831 return Ok(());
1832 }
1833 let jobs = self.job_rows(run_id).await?;
1834 if !jobs.iter().all(|job| job.status == "completed") {
1835 return Ok(());
1836 }
1837 self.finish_run(&run, None).await
1838 }
1839
1840 async fn finish_run(&self, run: &RunRow, error: Option<&str>) -> Result<()> {
1841 let jobs = self.job_rows(&run.id).await?;
1842 let rows: Vec<&JobRow> = jobs.iter().collect();
1843 let conclusion = if error.is_some() {
1844 "failure"
1845 } else if run.conclusion.as_deref() == Some("cancelled") {
1846 "cancelled"
1847 } else if rows.is_empty() {
1848 "skipped"
1849 } else {
1850 key_result(&rows)
1851 };
1852 let done = self
1853 .db
1854 .prepare("UPDATE runs SET status = 'completed', conclusion = ?, error = COALESCE(?, error), finished_at = ? WHERE id = ? AND status != 'completed' RETURNING id")
1855 .bind(&[conclusion.into(), optional(error), now().into(), run.id.as_str().into()])?
1856 .first::<Value>(None)
1857 .await?;
1858 if done.is_none() {
1859 return Ok(());
1860 }
1861 self.report_status(run, conclusion).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971862 self.settle_deployments(run, &jobs, conclusion == "cancelled").await;
Download ZIP reads blobs eight batches at a time and deflates files together1863 let published: Result<()> = g1t_kit::call(
1864 &self.events,
1865 "publish",
1866 &g1t_contracts::events::Publish {
1867 events: vec![g1t_contracts::events::NewEvent {
1868 kind: "workflow.completed",
1869 source: "actions",
1870 repo_id: Some(run.repo_id.clone()),
1871 actor: run.actor_id.clone(),
1872 data: g1t_contracts::events::WorkflowEvent {
1873 run_id: run.id.clone(),
1874 repo_id: run.repo_id.clone(),
1875 workflow: run.name.clone(),
1876 path: run.path.clone(),
1877 number: run.number,
1878 event: run.event.clone(),
1879 conclusion: conclusion.to_owned(),
1880 git_ref: run.git_ref.clone(),
1881 sha: run.sha.clone(),
1882 pull: run.pull,
1883 },
1884 }],
1885 },
1886 )
1887 .await;
1888 if let Err(error) = published {
1889 worker::console_error!("actions: could not publish workflow.completed: {error}");
1890 }
1891 // The next run waiting in its concurrency group.
1892 if let Some(group) = &run.concurrency_group {
1893 let next = self
1894 .db
1895 .prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND status = 'pending' ORDER BY id LIMIT 1")
1896 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1897 .first::<RunRow>(None)
1898 .await?;
1899 if let Some(next) = next {
1900 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[next.id.as_str().into()])?.run().await?;
1901 Box::pin(self.advance(&next.id)).await?;
1902 }
1903 }
1904 Ok(())
1905 }
1906
Merge GitHub pull request #8 (claude/cool-newton-7nm8r9): Docker deployments on g1t.page behind a switch that stays off until Cloudflare's container APIs are checked live, workflows a workspace manages in its repositories, tarball and zipball downloads, refs/pull/* kept to g1t alone, framework detection for g1t.page builds, and a workflow's event payload matching GitHub's.1907 /// Tells the deployments service that a job ended (`workflow_job_ended`),
1908 /// so a build it opened with `start_from_workflow` and left open fails
1909 /// at once. Never fails the job: an answer that does not come is
1910 /// logged, and the sweep fails the build later.
1911 async fn end_job_builds(&self, job_id: &str, conclusion: &str) {
1912 let ended: Result<Value> =
1913 g1t_kit::call(&self.deployments, "workflow_job_ended", &json!({ "jobId": job_id, "conclusion": conclusion })).await;
1914 if let Err(error) = ended {
1915 worker::console_log!("actions: could not tell deployments that job {job_id} ended: {error}");
1916 }
1917 }
1918
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971919 /// Tells the deployments service how a run's deployment to `env` stands:
1920 /// made when its first job naming the environment starts, failed when one
1921 /// of them fails, and settled (`last`) when the run finishes. Never fails
1922 /// the run: a deployment that could not be recorded is logged.
1923 async fn report_deployment(&self, run: &RunRow, env: &JobEnvironment, state: &str, last: bool) {
1924 let path = repo_path(&run.repo);
1925 let reported: Result<Value> = g1t_kit::call(
1926 &self.deployments,
1927 "actions_deployment",
1928 &json!({
1929 "repoId": run.repo_id,
1930 "repo": { "namespace": path.namespace, "name": path.name },
1931 "runId": run.id,
1932 "attempt": run.attempt,
1933 "runUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
1934 "environment": env.name,
1935 "url": env.url,
1936 "ref": run.git_ref,
1937 "sha": run.sha,
1938 "state": state,
1939 "final": last,
1940 "creator": run.actor,
1941 "workflow": run.name,
1942 }),
1943 )
1944 .await;
1945 if let Err(error) = reported {
1946 worker::console_error!("actions: deployment to {} not recorded for run {}: {error}", env.name, run.id);
1947 }
1948 }
1949
1950 /// A job that deploys has started: its run's deployment to the
1951 /// environment is under way.
1952 pub(crate) async fn job_started(&self, job_id: &str) -> Result<()> {
1953 let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[job_id.into()])?.first::<JobRow>(None).await? else {
1954 return Ok(());
1955 };
1956 let Some(run) = self.run_row(&job.run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a5c94df323642630e'1957 self.tell_live(&run.repo_id, &run.id, Some(&run.sha), run.pull).await;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971958 if let Some(env) = deploys_to(&run, &job) {
1959 self.report_deployment(&run, &env, "in_progress", false).await;
1960 }
1961 Ok(())
1962 }
1963
Merge branch 'worktree-agent-a5c94df323642630e'1964 /// Pages open on the repository hear that a run's job started or
1965 /// finished, which is no event (`g1t_contracts::live`). Logged and left
1966 /// when notify cannot be reached: the pages catch up by themselves.
1967 pub(crate) async fn tell_live(&self, repo_id: &str, run_id: &str, sha: Option<&str>, pull: Option<u32>) {
1968 let Some(notify) = &self.notify else { return };
1969 let change = g1t_contracts::live::RepoChange {
1970 repo_id: repo_id.to_owned(),
1971 kind: g1t_contracts::live::RUN_UPDATED.to_owned(),
1972 run: Some(run_id.to_owned()),
1973 sha: sha.map(str::to_owned),
1974 number: pull,
1975 at: now(),
1976 ..Default::default()
1977 };
1978 let args = g1t_contracts::live::RepoChangedArgs { changes: vec![change] };
1979 let told: Result<Value> = g1t_kit::call(notify, g1t_contracts::live::REPO_CHANGED, &args).await;
1980 if let Err(error) = told {
1981 worker::console_error!("actions: live pages not told of run {run_id}: {error}");
1982 }
1983 }
1984
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971985 /// The run is over: each environment its jobs deployed to takes the
1986 /// outcome of those jobs (`deployment_outcome`).
1987 async fn settle_deployments(&self, run: &RunRow, jobs: &[JobRow], cancelled: bool) {
1988 let mut seen: Vec<(JobEnvironment, Vec<Option<String>>)> = Vec::new();
1989 for job in jobs {
1990 let Some(env) = deploys_to(run, job) else { continue };
1991 let conclusion = if cancelled && job.conclusion.as_deref() != Some("skipped") && job.started_at.is_some() {
1992 Some("cancelled".to_owned())
1993 } else if job.started_at.is_none() {
1994 Some("skipped".to_owned())
1995 } else {
1996 job.conclusion.clone()
1997 };
1998 match seen.iter_mut().find(|(known, _)| known.name.eq_ignore_ascii_case(&env.name)) {
1999 Some((known, conclusions)) => {
2000 if known.url.is_none() {
2001 known.url = env.url.clone();
2002 }
2003 conclusions.push(conclusion);
2004 }
2005 None => seen.push((env, vec![conclusion])),
2006 }
2007 }
2008 for (env, conclusions) in seen {
2009 if let Some(state) = deployment_outcome(&conclusions) {
2010 self.report_deployment(run, &env, state, true).await;
2011 }
2012 }
2013 }
2014
Download ZIP reads blobs eight batches at a time and deflates files together2015 /// Tells the pull request (or commit) how the run went, as a status.
2016 async fn report_status(&self, run: &RunRow, conclusion: &str) -> Result<()> {
2017 let state = match conclusion {
2018 "success" | "skipped" => "success",
2019 "cancelled" => "error",
2020 _ => "failure",
2021 };
2022 let _: Result<Value> = g1t_kit::call(
2023 &self.work,
2024 "set_commit_status",
2025 &json!({
2026 "repoId": run.repo_id,
2027 "sha": run.sha,
2028 "context": format!("{} / {}", run.name, run.event),
2029 "state": state,
2030 "description": format!("{} {}", run.name, match conclusion {
2031 "success" => "passed",
2032 "skipped" => "was skipped",
2033 "cancelled" => "was cancelled",
2034 _ => "failed",
2035 }),
2036 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge2037 "source": "actions",
Download ZIP reads blobs eight batches at a time and deflates files together2038 }),
2039 )
2040 .await;
2041 Ok(())
2042 }
2043
2044 /// Tells the pull request a run has started on its head.
2045 pub async fn report_pending(&self, run: &RunRow) -> Result<()> {
2046 let _: Result<Value> = g1t_kit::call(
2047 &self.work,
2048 "set_commit_status",
2049 &json!({
2050 "repoId": run.repo_id,
2051 "sha": run.sha,
2052 "context": format!("{} / {}", run.name, run.event),
2053 "state": "pending",
Merge branch 'worktree-agent-a3abfcce648e87dca'2054 "description": if run.status == "action_required" {
2055 format!("{} is waiting for approval", run.name)
2056 } else {
2057 format!("{} is running", run.name)
2058 },
Download ZIP reads blobs eight batches at a time and deflates files together2059 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge2060 "source": "actions",
Download ZIP reads blobs eight batches at a time and deflates files together2061 }),
2062 )
2063 .await;
2064 Ok(())
2065 }
2066
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2067 /// Cancels a run: its waiting and queued jobs at once, and its running
2068 /// ones gracefully (`stop_job`), or outright when `force`.
2069 pub async fn cancel_run(&self, run: &RunRow, reason: &str, force: bool) -> Result<()> {
Download ZIP reads blobs eight batches at a time and deflates files together2070 self.db
2071 .prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")
2072 .bind(&[run.id.as_str().into()])?
2073 .run()
2074 .await?;
2075 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)2076 if force {
2077 self.hard_stop(job, reason).await?;
2078 } else {
2079 self.stop_job(job, reason).await?;
2080 }
Download ZIP reads blobs eight batches at a time and deflates files together2081 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2082 if run.status == "pending" || run.status == "action_required" {
Download ZIP reads blobs eight batches at a time and deflates files together2083 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;
2084 }
2085 self.advance(&run.id).await
2086 }
2087
2088 /// Cancels every run of the repository that has not finished, for
2089 /// `repo.deleted` and `repo.archived`.
2090 pub async fn stop_runs(&self, repo_id: &str) -> Result<()> {
2091 let runs = self
2092 .db
2093 .prepare("SELECT * FROM runs WHERE repo_id = ? AND status != 'completed'")
2094 .bind(&[repo_id.into()])?
2095 .all()
2096 .await?
2097 .results::<RunRow>()?;
2098 for run in runs {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2099 self.cancel_run(&run, "The repository was archived or deleted.", true).await?;
Download ZIP reads blobs eight batches at a time and deflates files together2100 }
2101 Ok(())
2102 }
2103
2104 pub async fn cancel(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
2105 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
2106 return Ok(Outcome::Fail(refused));
2107 }
2108 let run = check!(self.run_in(&a.repo, &a.id).await?);
2109 if run.status == "completed" {
2110 return Ok(fail(FailureCode::Conflict, "The run has already finished."));
2111 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2112 // Cancelling a run that is already cancelling stops its jobs
2113 // outright, without waiting for their cleanup steps.
2114 let force = a.force || run.conclusion.as_deref() == Some("cancelled");
2115 let reason = if force {
2116 format!("{} stopped the run without waiting for its cleanup steps.", a.actor.username)
2117 } else {
2118 format!("{} cancelled the run.", a.actor.username)
2119 };
2120 self.cancel_run(&run, &reason, force).await?;
Download ZIP reads blobs eight batches at a time and deflates files together2121 self.run_summary(&run.id).await
2122 }
2123
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2124 /// Runs again, as a new attempt: every job, with `failed_only` those
2125 /// that did not succeed, or with `job` that one; each with the jobs that
2126 /// need them. The attempt that ends is kept, its jobs and their logs and
2127 /// summaries with it (`job_attempts`, `run_attempts`).
Download ZIP reads blobs eight batches at a time and deflates files together2128 pub async fn rerun(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
2129 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
2130 return Ok(Outcome::Fail(refused));
2131 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2132 // A job alone (`POST …/jobs/{job}/rerun`) names its run.
2133 let run_id = match (&a.job, a.id.is_empty()) {
2134 (Some(job), true) => {
2135 #[derive(Deserialize)]
2136 struct Of {
2137 run_id: String,
2138 }
2139 let of = self.db.prepare("SELECT run_id FROM jobs WHERE id = ?").bind(&[job.as_str().into()])?.first::<Of>(None).await?;
2140 match of {
2141 Some(of) => of.run_id,
2142 None => return Ok(fail(FailureCode::NotFound, "No such job.")),
2143 }
2144 }
2145 _ => a.id.clone(),
2146 };
2147 let run = check!(self.run_in(&a.repo, &run_id).await?);
Download ZIP reads blobs eight batches at a time and deflates files together2148 if run.status != "completed" {
2149 return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));
2150 }
2151 // Nothing starts again on an archived repository.
2152 match self.visible_repo(&a.repo, &Some(a.actor.clone())).await? {
2153 Some(repo) if repo.archived() => {
2154 return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));
2155 }
2156 Some(_) => {}
2157 None => return Ok(fail(FailureCode::NotFound, "There is no such repository.")),
2158 }
2159 if run.error.is_some() {
2160 return Ok(fail(FailureCode::Conflict, "This run never started: fix the workflow file and push again."));
2161 }
2162 let jobs = self.job_rows(&run.id).await?;
2163 let workflow = workflow::parse(&run.source).ok();
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2164 let target = match &a.job {
2165 Some(id) => match jobs.iter().find(|job| &job.id == id) {
2166 Some(job) => Some(top_key(&job.key).to_owned()),
2167 None => return Ok(fail(FailureCode::NotFound, "That job is not in the run's latest attempt.")),
2168 },
2169 None => None,
2170 };
2171 let which = match (&target, a.failed_only) {
2172 (Some(key), _) => Rerun::Job(key),
2173 (None, true) => Rerun::Failed,
2174 (None, false) => Rerun::All,
2175 };
2176 let order = workflow.as_ref().map(|w| w.job_order()).unwrap_or_default();
2177 let again = rerun_keys(
2178 &order,
2179 |key| jobs.iter().find(|j| j.key == key).map(JobRow::needs).unwrap_or_default(),
2180 |key| jobs.iter().filter(|j| j.key == key).all(|j| j.conclusion.as_deref() == Some("success")),
2181 which,
2182 );
Download ZIP reads blobs eight batches at a time and deflates files together2183 if again.is_empty() {
2184 return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));
2185 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2186 // The jobs that run again, a called workflow's with the job calling it.
2187 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();
2188 let ids = serde_json::to_string(&rerun_ids)?;
2189 let ended = run.attempt.to_string();
2190 let ended = ended.as_str();
2191 let mut statements = vec![
2192 // The attempt that ends, as it ended.
2193 self.db
2194 .prepare(
2195 "INSERT OR REPLACE INTO run_attempts (run_id, attempt, repo_id, conclusion, actor, debug, started_at, finished_at)
2196 SELECT id, attempt, repo_id, conclusion, COALESCE(triggering_actor, actor), debug, started_at, finished_at FROM runs WHERE id = ?",
2197 )
2198 .bind(&[run.id.as_str().into()])?,
2199 // Earlier attempts that showed a job's logs from where it ran
2200 // then now find them where they move to.
2201 self.db
2202 .prepare("UPDATE job_attempts SET log_id = log_id || '.' || ?1 WHERE run_id = ?2 AND log_id IN (SELECT value FROM json_each(?3))")
2203 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
2204 // Each of its jobs: one that runs again keeps its logs under
2205 // `{id}.{attempt}`; one left alone is still the live job's.
2206 self.db
2207 .prepare(
2208 "INSERT OR REPLACE INTO job_attempts (id, run_id, repo_id, attempt, job_id, log_id, key, ordinal, name, needs, status, conclusion,
2209 steps, annotations, reason, environment, labels, runner_name, started_at, finished_at)
2210 SELECT id || '.' || ?1, run_id, repo_id, ?1, id,
2211 CASE WHEN id IN (SELECT value FROM json_each(?3)) THEN id || '.' || ?1 ELSE id END,
2212 key, ordinal, name, needs, status, conclusion, steps, annotations, reason, environment, labels, runner_name, started_at, finished_at
2213 FROM jobs WHERE run_id = ?2 ORDER BY rowid",
2214 )
2215 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
2216 self.db
2217 .prepare("UPDATE logs SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
2218 .bind(&[ended.into(), ids.as_str().into()])?,
2219 self.db
2220 .prepare("UPDATE job_summaries SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
2221 .bind(&[ended.into(), ids.as_str().into()])?,
2222 ];
Download ZIP reads blobs eight batches at a time and deflates files together2223 for key in &again {
2224 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-7nm8r92225 // The jobs of a workflow it called (each matrix combination's
2226 // under `key[i]/`) are made again when it calls it again.
Download ZIP reads blobs eight batches at a time and deflates files together2227 statements.push(
2228 self.db
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92229 .prepare("DELETE FROM jobs WHERE run_id = ? AND (key LIKE ? OR key LIKE ?)")
2230 .bind(&[run.id.as_str().into(), format!("{key}/%").into(), format!("{key}[%").into()])?,
Download ZIP reads blobs eight batches at a time and deflates files together2231 );
2232 statements.push(
2233 self.db
2234 .prepare(
2235 "UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2236 matrix = NULL, call = NULL, token_hash = NULL, first_token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
Merge branch 'worktree-agent-a3abfcce648e87dca'2237 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)2238 concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",
Download ZIP reads blobs eight batches at a time and deflates files together2239 )
2240 .bind(&[run.id.as_str().into(), key.as_str().into()])?,
2241 );
2242 }
2243 statements.push(
2244 self.db
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2245 .prepare(
2246 "UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL,
2247 triggering_actor = ?, debug = ? WHERE id = ?",
2248 )
2249 .bind(&[a.actor.username.as_str().into(), u32::from(a.debug).into(), run.id.as_str().into()])?,
Download ZIP reads blobs eight batches at a time and deflates files together2250 );
2251 self.db.batch(statements).await?;
2252 if let Some(run) = self.run_row(&run.id).await? {
2253 self.report_pending(&run).await?;
2254 }
2255 self.advance(&run.id).await?;
Merge the run page fixes: Cancel run shows on every view of a run that is still going, including an earlier attempt's and job links, and pages refresh when a run is re-run from anywhere; a job's log follows to its true end, above anything covering the bottom of the window, and Jump to latest lands there.2256 // A page left open on the attempt that ended, which asks for nothing
2257 // while its run is over, hears that it is going again: re-run from
2258 // the API, the MCP tool, a pull request or another tab, as well as
2259 // from that page, it offers Cancel run and not the re-runs.
2260 self.tell_live(&run.repo_id, &run.id, Some(&run.sha), run.pull).await;
Download ZIP reads blobs eight batches at a time and deflates files together2261 self.run_summary(&run.id).await
2262 }
2263
2264 pub async fn run_in(&self, repo: &RepoPath, id: &str) -> Result<Outcome<RunRow>> {
2265 let row = self
2266 .db
2267 .prepare("SELECT * FROM runs WHERE id = ? AND lower(repo) = lower(?)")
2268 .bind(&[id.into(), format!("{}/{}", repo.namespace, repo.name).into()])?
2269 .first::<RunRow>(None)
2270 .await?;
2271 Ok(row.map_or_else(|| fail(FailureCode::NotFound, "No such run."), Outcome::Ok))
2272 }
2273
2274 // --- The sandbox's side -----------------------------------------------------
2275
2276 pub(crate) async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2277 self.job_for(a, false).await
2278 }
2279
2280 /// As [`Self::job_for_token`], and also with the token the job started
2281 /// with after its harness swapped it (`job_spec` with `rotate`): for
2282 /// `brokered_secrets` alone, which the runner Worker asks with the
2283 /// token it was given at the start, and which no API route passes on.
2284 pub(crate) async fn job_for_started_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
2285 self.job_for(a, true).await
2286 }
2287
2288 async fn job_for(&self, a: &JobCallArgs, started: bool) -> Result<Outcome<JobRow>> {
Download ZIP reads blobs eight batches at a time and deflates files together2289 let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;
2290 Ok(match job {
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2291 Some(job) if job.status == "in_progress" && token_opens(job.token_hash.as_deref(), job.first_token_hash.as_deref(), &a.token, started) => Outcome::Ok(job),
Download ZIP reads blobs eight batches at a time and deflates files together2292 _ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),
2293 })
2294 }
2295
2296 /// `job_auth`: which run and repository a running job's token is for,
2297 /// so the API can keep its artifacts and cache.
2298 pub async fn job_auth(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2299 let job = check!(self.job_for_token(&a).await?);
2300 Ok(Outcome::Ok(json!({ "run": job.run_id, "repoId": job.repo_id })))
2301 }
2302
2303 /// `job_spec`: everything the sandbox needs to run the job.
2304 pub async fn job_spec(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2305 let job = check!(self.job_for_token(&a).await?);
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2306 // Asked to, the token the harness asked with is spent here: it gets
2307 // a new one with the spec, for everything after, so a copy of the
2308 // first left on the runner's computer (a Windows job's script, a
2309 // container's environment) cannot fetch the secrets again. The new
2310 // token is made now, as the runtime token is signed with its hash,
2311 // and swapped in last, so a spec that could not be made leaves the
2312 // old one working.
2313 let next = a.rotate.then(|| random_hex(24));
2314 let token_hash = match &next {
2315 Some(next) => sha256_hex(next),
2316 None => job.token_hash.clone().unwrap_or_default(),
2317 };
Download ZIP reads blobs eight batches at a time and deflates files together2318 let Some(run) = self.run_row(&job.run_id).await? else {
2319 return Ok(fail(FailureCode::NotFound, "No such run."));
2320 };
2321 let Ok(caller) = workflow::parse(&run.source) else {
2322 return Ok(fail(FailureCode::Invalid, "The workflow no longer reads."));
2323 };
2324 // A called workflow's job runs as that workflow defines it.
2325 let callee = job.callee();
2326 let (workflow, spec, call_inputs) = match callee {
2327 Some((called, spec, call)) => (called, spec, Some(call["inputs"].clone())),
2328 None => match caller.jobs.iter().find(|j| j.id == job.key) {
2329 Some(spec) => (caller.clone(), spec.clone(), None),
2330 None => return Ok(fail(FailureCode::NotFound, "The job is not in the workflow.")),
2331 },
2332 };
2333 let spec = &spec;
2334 let repo = repo_path(&run.repo);
2335 let trusted = run.trusted != 0;
Merge branch 'worktree-agent-a3abfcce648e87dca'2336 // The job's `environment:`, by name, as read when its needs were done
2337 // (an expression included), once the environment's protection rules
2338 // let it start: entries with a value for it give that value instead
2339 // of their default, as GitHub's environment secrets do.
2340 let environment: Option<String> = job.environment.clone();
2341 // What its token may do: its `permissions:` (a called workflow's
2342 // jobs no more than the job that calls it), else the repository's
2343 // default; read-only for a pull request from outside.
2344 // The repository's default, its workspace's taken in, and whether
2345 // its jobs may open and approve pull requests.
2346 let (default, pull_requests) = self.token_policy(&run.repo_id).await?;
2347 let mut permissions = spec.permissions(&workflow, default);
2348 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-7nm8r92349 let top = top_key(parent);
Merge branch 'worktree-agent-a3abfcce648e87dca'2350 if let Some(caller_job) = caller.jobs.iter().find(|j| j.id == top) {
2351 permissions = permissions.capped_by(&caller_job.permissions(&caller, default));
Download ZIP reads blobs eight batches at a time and deflates files together2352 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2353 }
2354 if !trusted {
2355 permissions = permissions.read_only();
2356 }
2357 // G1T_TOKEN, and GITHUB_TOKEN as its alias: a token of the
2358 // workspace's that reaches this repository only, with the scopes
2359 // its permissions give, until the job ends.
2360 let token = match self.workspace_actor(&repo.namespace).await? {
2361 Some(workspace) => {
2362 let created: CreatedAccessToken = g1t_kit::call(
2363 &self.identity,
2364 "create_job_token",
2365 &CreateJobTokenArgs {
2366 workspace,
2367 repo: repo.clone(),
2368 run_id: run.id.clone(),
2369 job_id: job.id.clone(),
2370 name: format!("G1T_TOKEN for {} run {}", run.repo, run.number),
2371 ttl_seconds: u64::from(job.timeout_minutes) * 60 + 600,
2372 scopes: permissions.scopes().into_iter().map(str::to_owned).collect(),
2373 pull_requests: pull_requests && trusted,
2374 },
2375 )
2376 .await?;
2377 created.token
2378 }
2379 None => String::new(),
Download ZIP reads blobs eight batches at a time and deflates files together2380 };
2381 // A run that is not trusted (a pull request from outside the
2382 // workspace) gets no secrets and an empty token.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts2383 let passed = job.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.2384 // Brokered secrets among them (broker.rs), whatever keys they reach
2385 // the job under.
2386 let mut brokers: Vec<Broker> = Vec::new();
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts2387 let mut secrets = match (trusted, passed) {
2388 (false, _) => Map::new(),
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.2389 (true, None) => {
2390 let resolved = self.secrets_for(&run.repo_id, &run.repo, environment.as_deref(), true).await?;
2391 brokers.extend(resolved.brokers());
2392 resolved.values
2393 }
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts2394 // A called workflow's job: what its callers passed it (reach.rs),
2395 // and its own environment's secrets over them.
2396 (true, Some(plan)) => {
2397 let base = self.secrets_for(&run.repo_id, &run.repo, None, true).await?;
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.2398 brokers.extend(base.brokers());
2399 let base = base.values;
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts2400 let vars = self.variables_for(&run.repo_id, &run.repo, None, true).await?;
2401 let github = run.info().context(&job.key, "", run.action.as_deref());
2402 let mut passed = crate::reach::resolve_secrets(&plan, &base, &github, &vars);
2403 if let Some(name) = environment.as_deref() {
2404 let own = self.secrets_for(&run.repo_id, &run.repo, Some(name), true).await?;
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.2405 brokers.extend(own.brokers());
2406 for (key, value) in own.values {
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts2407 if base.get(&key) != Some(&value) {
2408 passed.insert(key, value);
2409 }
2410 }
2411 }
2412 passed
2413 }
Download ZIP reads blobs eight batches at a time and deflates files together2414 };
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.2415 // A brokered secret's value never reaches the sandbox: the job gets
2416 // a stand-in, which the runner Worker fills in on its requests to
2417 // the secret's hosts. A self-hosted runner's requests never pass
2418 // the Worker, so there it stays a stand-in, and the log says so.
2419 let stand_ins = broker::substitute(&mut secrets, &brokers, broker::stand_in);
2420 if !stand_ins.is_empty() {
2421 self.keep_stand_ins(&job.id, job.timeout_minutes, &stand_ins).await?;
2422 if job.runner_id.is_some() {
2423 let mut names: Vec<&str> = stand_ins.iter().map(|made| made.broker.name.as_str()).collect();
2424 names.sort_unstable();
2425 let notes: String = names.iter().map(|name| broker::self_hosted_note(name) + "\n").collect();
2426 self.note_log(&job.id, 0, &notes).await?;
2427 }
2428 }
Download ZIP reads blobs eight batches at a time and deflates files together2429 secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));
2430 secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));
Merge branch 'worktree-agent-a3abfcce648e87dca'2431 // Each secret as it is, a line at a time, base64 and JSON-escaped.
2432 let mut masks: Vec<String> = g1t_actions::mask::all_variants(secrets.values().filter_map(|v| v.as_str()));
Download ZIP reads blobs eight batches at a time and deflates files together2433 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 R22434 // The toolkit's runtime token (runtime.rs), for as long as the job
2435 // may run; the API puts it and the toolkit's addresses in the
2436 // job's variables.
2437 let runtime_token = crate::runtime::runtime_token(
2438 &job.id,
2439 &job.run_id,
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2440 &token_hash,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22441 now_ms() / 1000,
2442 u64::from(job.timeout_minutes) * 60 + 600,
2443 );
2444 masks.push(runtime_token.clone());
2445 let retention_days = self.retention_setting(&run.repo_id).await?;
Download ZIP reads blobs eight batches at a time and deflates files together2446
2447 let jobs = self.job_rows(&run.id).await?;
2448 let mut needs = Map::new();
2449 // In a called workflow, its jobs' keys sit under the job that called it.
2450 let parent = job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));
2451 for need in &spec.needs {
2452 let key = match &parent {
2453 Some(parent) => format!("{parent}/{need}"),
2454 None => need.clone(),
2455 };
2456 let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();
2457 let mut outputs = Map::new();
2458 for row in &rows {
2459 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
2460 outputs.extend(more);
2461 }
2462 }
2463 needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));
2464 }
2465 let siblings = jobs.iter().filter(|row| row.key == job.key).count();
2466 let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));
2467 let info = run.info();
2468 let mut github = info.context(&job.key, &token, run.action.as_deref());
2469 github["token"] = json!(token);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22470 github["retention_days"] = json!(retention_days);
Download ZIP reads blobs eight batches at a time and deflates files together2471 // On a self-hosted runner, `runner` and `RUNNER_*` describe that
2472 // machine rather than g1t's sandbox.
2473 let mut variables = info.variables(&job.key);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22474 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)2475 let mut runner = match &job.runner_id {
Download ZIP reads blobs eight batches at a time and deflates files together2476 Some(id) => self.runner_context_for(id, &mut variables).await?,
2477 None => runner_context(),
2478 };
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2479 // A re-run with debug logging: what GitHub sets for one.
2480 if run.debug != 0 {
2481 debug_logging(&mut variables, &mut runner);
2482 }
Download ZIP reads blobs eight batches at a time and deflates files together2483
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2484 // Where to check out: the repository, which holds a pull request's
2485 // test merge commit; a pull request's fork when the run builds its
2486 // head instead (`run.sha` is the head, `info.sha` what is built).
Download ZIP reads blobs eight batches at a time and deflates files together2487 let clone_url = match run.pull {
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2488 Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" && info.sha == run.sha => {
Download ZIP reads blobs eight batches at a time and deflates files together2489 let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(
2490 &self.work,
2491 "get_pull",
2492 &g1t_contracts::work::ViewArgs {
2493 repo: repo.clone(),
2494 number,
2495 viewer: self.workspace_actor(&repo.namespace).await?,
2496 after_seq: 0,
2497 },
2498 )
2499 .await?;
2500 match located {
2501 Outcome::Ok(detail) => match detail.pull.fork {
2502 Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),
2503 None => format!("{SITE}/{}.git", run.repo),
2504 },
2505 Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),
2506 }
2507 }
2508 _ => format!("{SITE}/{}.git", run.repo),
2509 };
2510
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2511 // Only one fetch with a token gets its spec: of two at once, the
2512 // second finds the hash already changed and is refused.
2513 if next.is_some() {
2514 let swapped = self
2515 .db
2516 .prepare(TOKEN_ROTATE)
2517 .bind(&[token_hash.as_str().into(), job.id.as_str().into(), job.token_hash.as_deref().unwrap_or_default().into()])?
2518 .first::<Value>(None)
2519 .await?;
2520 if swapped.is_none() {
2521 return Ok(fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."));
2522 }
2523 }
2524
2525 let mut spec = json!({
Download ZIP reads blobs eight batches at a time and deflates files together2526 "job": job.id,
2527 "run": run.id,
2528 "key": job.key,
2529 "name": job.name,
2530 "spec": spec.raw,
2531 "workflow": {
2532 "env": workflow.env,
2533 "defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),
2534 },
2535 "github": github,
2536 "variables": variables,
2537 "event": info.event,
2538 "contexts": {
2539 "vars": vars,
2540 "secrets": secrets,
2541 "inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),
2542 "matrix": matrix,
2543 "needs": needs,
2544 "strategy": {
2545 "fail-fast": spec.fail_fast,
2546 "job-index": job.ordinal,
2547 "job-total": siblings,
2548 "max-parallel": spec.max_parallel.unwrap_or(siblings as u32),
2549 },
2550 "runner": runner,
2551 },
2552 "checkout": {
2553 "repository": run.repo,
2554 "url": clone_url,
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2555 "sha": info.sha,
Download ZIP reads blobs eight batches at a time and deflates files together2556 "ref": run.git_ref,
2557 "token": token,
2558 },
2559 "timeoutMinutes": job.timeout_minutes,
2560 "masks": masks,
Merge branch 'worktree-agent-a3abfcce648e87dca'2561 // As the job's log lists them at its start.
2562 "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 R22563 // Whether the job may ask for an OIDC token decides whether it
2564 // is told where to.
2565 "runtime": {
2566 "token": runtime_token,
2567 "idToken": self.oidc_allowed(&run, &job),
2568 },
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2569 });
2570 if let Some(next) = next {
2571 spec["nextToken"] = json!(next);
2572 }
2573 Ok(Outcome::Ok(spec))
Download ZIP reads blobs eight batches at a time and deflates files together2574 }
2575
2576 /// `job_report`: the sandbox telling how the job is going.
2577 pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2578 let job = check!(self.job_for_token(&a).await?);
2579 let report = &a.report;
2580 let at = now();
2581 match report["kind"].as_str().unwrap_or_default() {
2582 "steps" => {
2583 // The list can grow as the job goes (post steps), so steps
2584 // already reported keep where they stand.
2585 let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2586 let steps: Vec<Value> = report["steps"]
2587 .as_array()
2588 .map(|names| {
2589 names
2590 .iter()
2591 .enumerate()
2592 .map(|(i, name)| match known.get(i) {
2593 Some(step) if step["status"] != "queued" => step.clone(),
2594 _ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),
2595 })
2596 .collect()
2597 })
2598 .unwrap_or_default();
2599 self.db
2600 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2601 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2602 .run()
2603 .await?;
2604 }
2605 "step" => {
2606 let number = report["number"].as_u64().unwrap_or(0) as usize;
2607 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2608 if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {
2609 let status = report["status"].as_str().unwrap_or("in_progress");
2610 step["status"] = json!(status);
2611 if status == "in_progress" {
2612 step["startedAt"] = json!(at);
2613 }
2614 if status == "completed" {
2615 step["finishedAt"] = json!(at);
2616 step["conclusion"] = report["conclusion"].clone();
2617 }
2618 if let Some(name) = report["name"].as_str() {
2619 step["name"] = json!(name);
2620 }
2621 }
2622 self.db
2623 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2624 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2625 .run()
2626 .await?;
2627 }
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.2628 // A line from g1t itself, not the job: the runner Worker's note
2629 // that a brokered secret went to a host it is not for. Put
2630 // under the step that is running.
2631 "note" => {
2632 let text = report["text"].as_str().unwrap_or_default().chars().take(1000).collect::<String>();
2633 let steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2634 let step = steps.iter().rev().find(|step| step["status"] == "in_progress").and_then(|step| step["number"].as_u64()).unwrap_or(0);
2635 if !text.trim().is_empty() {
2636 self.note_log(&job.id, step as u32, &text).await?;
2637 }
2638 }
Download ZIP reads blobs eight batches at a time and deflates files together2639 "log" => {
2640 let mut text = report["text"].as_str().unwrap_or_default().to_owned();
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.2641 // A brokered secret's value never enters the sandbox, but an
2642 // answer that echoed it whole could carry it in; it is
2643 // masked here, where the value is known.
2644 if job.brokered != 0 {
2645 let values: Vec<String> = self.stand_ins_of(&job.id).await?.into_iter().map(|secret| secret.value).collect();
2646 if !values.is_empty() {
2647 text = g1t_actions::mask::apply(&text, &g1t_actions::mask::all_variants(values.iter().map(String::as_str)));
2648 }
2649 }
Download ZIP reads blobs eight batches at a time and deflates files together2650 if text.len() > MAX_CHUNK_BYTES {
2651 let mut cut = MAX_CHUNK_BYTES;
2652 while !text.is_char_boundary(cut) {
2653 cut -= 1;
2654 }
2655 text.truncate(cut);
2656 }
2657 #[derive(Deserialize)]
2658 struct Size {
2659 n: Option<f64>,
2660 }
2661 let size = self
2662 .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.2663 .prepare("SELECT SUM(LENGTH(text)) AS n FROM logs WHERE job_id = ?")
Download ZIP reads blobs eight batches at a time and deflates files together2664 .bind(&[job.id.as_str().into()])?
2665 .first::<Size>(None)
2666 .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.2667 let used = size.and_then(|s| s.n).unwrap_or(0.0) as usize;
Download ZIP reads blobs eight batches at a time and deflates files together2668 if used < MAX_LOG_BYTES {
2669 if used + text.len() >= MAX_LOG_BYTES {
2670 text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");
2671 }
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.2672 // The next `seq` is taken in the insert itself, so two
2673 // batches at once (a report sent again while the first
2674 // was still being written) never reach for the same one
2675 // and lose a chunk to the primary key.
Download ZIP reads blobs eight batches at a time and deflates files together2676 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.2677 .prepare(LOG_INSERT)
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2678 .bind(&[
2679 job.id.as_str().into(),
2680 (report["step"].as_u64().unwrap_or(0) as u32).into(),
2681 text.into(),
2682 report["chunk"].as_u64().map_or(worker::wasm_bindgen::JsValue::NULL, |n| (n as f64).into()),
2683 ])?
Download ZIP reads blobs eight batches at a time and deflates files together2684 .run()
2685 .await?;
2686 }
2687 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2688 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2689 "summary" => {
2690 // `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the
2691 // step's summary, up to 1 MiB a step and 20 steps a job.
2692 let step = report["step"].as_u64().unwrap_or(0) as u32;
2693 let markdown = report["markdown"].as_str().unwrap_or_default();
2694 #[derive(Deserialize)]
2695 struct Held {
2696 steps: u32,
2697 mine: Option<f64>,
2698 }
2699 let held = self
2700 .db
2701 .prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")
2702 .bind(&[step.into(), job.id.as_str().into()])?
2703 .first::<Held>(None)
2704 .await?;
2705 let (steps, mine) = held.map_or((0, None), |held| (held.steps, held.mine.map(|n| n as usize)));
2706 if summary_fits(steps, mine, markdown.len()) {
2707 self.db
2708 .prepare(
2709 "INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)
2710 ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown || excluded.markdown",
2711 )
2712 .bind(&[job.id.as_str().into(), step.into(), markdown.into()])?
2713 .run()
2714 .await?;
2715 }
2716 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2717 }
Download ZIP reads blobs eight batches at a time and deflates files together2718 "annotation" => {
2719 let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
2720 if annotations.len() < MAX_ANNOTATIONS {
2721 annotations.push(json!({
2722 "level": report["level"].as_str().unwrap_or("notice"),
2723 "message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),
2724 "title": report["title"],
2725 "file": report["file"],
2726 "line": report["line"],
2727 }));
2728 self.db
2729 .prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")
2730 .bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?
2731 .run()
2732 .await?;
2733 }
2734 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2735 // Nothing to tell; the answer says whether to stop.
2736 "ping" => {
2737 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2738 }
Download ZIP reads blobs eight batches at a time and deflates files together2739 "done" => {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2740 // A job told to stop ends cancelled, however its cleanup went.
2741 let conclusion = if job.cancel_requested_at.is_some() {
2742 "cancelled"
2743 } else {
2744 report["conclusion"]
2745 .as_str()
2746 .filter(|c| matches!(*c, "success" | "failure" | "cancelled"))
2747 .unwrap_or("failure")
2748 };
Download ZIP reads blobs eight batches at a time and deflates files together2749 let outputs = report["outputs"].as_object().cloned();
2750 Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
2751 }
2752 other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
2753 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2754 // `cancelled`: the run was cancelled, and the runner should stop
2755 // the step it is on and run only its cleanup steps.
2756 Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))
Download ZIP reads blobs eight batches at a time and deflates files together2757 }
2758
2759 // --- Every minute ---------------------------------------------------------------
2760
2761 pub async fn on_minute(&self, now_ms: u64) -> Result<()> {
2762 let minute = now_ms / 60_000 * 60_000;
Merge GitHub pull request #8 (claude/cool-newton-7nm8r9): Docker deployments on g1t.page behind a switch that stays off until Cloudflare's container APIs are checked live, workflows a workspace manages in its repositories, tarball and zipball downloads, refs/pull/* kept to g1t alone, framework detection for g1t.page builds, and a workflow's event payload matching GitHub's.2763 // Repositories a workspace's managed workflows changed in, beyond
2764 // those synced at once (managed.rs): their schedules wait for it.
2765 if let Err(error) = self.catch_up_syncs().await {
2766 worker::console_error!("actions: catching up syncs failed: {error}");
2767 }
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)2768 // Staff (or billing's usage watcher) paused scheduled runs across
2769 // g1t: this minute's schedules are skipped, not queued for later.
2770 // Kept 30 seconds in the isolate (g1t_kit::pause).
2771 if g1t_kit::pause::paused(&self.billing, g1t_contracts::billing::PauseLevel::Schedules).await {
2772 worker::console_log!("actions: schedules are paused across g1t; skipped this minute's");
2773 } else if let Err(error) = self.run_schedules(minute).await {
Download ZIP reads blobs eight batches at a time and deflates files together2774 worker::console_error!("actions: schedules failed: {error}");
2775 }
2776 // Jobs whose sandbox went quiet or ran past their time.
2777 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
2778 for job in running {
2779 // Times in g1t's format compare as text.
2780 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
2781 let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());
2782 let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
2783 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)2784 let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(|at| at < before(CANCEL_GRACE_MS).as_str());
2785 if cancel_overdue {
2786 // Cancelled, and still going after its grace period.
2787 self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;
2788 self.advance(&job.run_id).await?;
2789 } else if over {
Download ZIP reads blobs eight batches at a time and deflates files together2790 let reason = format!("It ran longer than its time limit of {} minutes.", job.timeout_minutes);
2791 // A self-hosted runner is told to stop on its next poll.
2792 if job.runner_id.is_none() {
2793 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2794 }
2795 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2796 } else if silent {
2797 let reason = match &job.runner_name {
2798 Some(name) => format!("The self-hosted runner {name} stopped answering."),
2799 None => "The runner stopped answering.".to_owned(),
2800 };
2801 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2802 }
2803 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92804 // Calls of reusable workflows waiting in a concurrency group, in
2805 // case the job they waited for was cancelled in another run.
2806 #[derive(Deserialize)]
2807 struct RunId {
2808 run_id: String,
2809 }
2810 let held = self
2811 .db
2812 .prepare("SELECT DISTINCT run_id FROM jobs WHERE status = 'waiting' AND concurrency_group IS NOT NULL LIMIT 20")
2813 .all()
2814 .await?
2815 .results::<RunId>()?;
2816 for run in held {
2817 self.advance(&run.run_id).await?;
2818 }
Download ZIP reads blobs eight batches at a time and deflates files together2819 if let Err(error) = self.sweep_runners(now_ms).await {
2820 worker::console_error!("actions: the runners' sweep failed: {error}");
2821 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2822 // Jobs held at an environment whose wait timer has run out.
2823 if let Err(error) = self.release_gates().await {
2824 worker::console_error!("actions: environments' gates failed: {error}");
2825 }
Download ZIP reads blobs eight batches at a time and deflates files together2826 // Once an hour: the cache's expired entries, and its storage.
2827 if (now_ms / 60_000) % 60 == 7
2828 && let Err(error) = self.sweep_cache(now_ms).await
2829 {
2830 worker::console_error!("actions: the cache's sweep failed: {error}");
2831 }
Merge remote-tracking branch 'origin/claude/peaceful-franklin-7qtkwx'2832 // And the archives of GitHub's actions no job uses any more.
2833 if (now_ms / 60_000) % 60 == 47
2834 && let Err(error) = self.sweep_action_archives(now_ms).await
2835 {
2836 worker::console_error!("actions: the action archives' sweep failed: {error}");
2837 }
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22838 // And artifacts past their time, and the toolkit's abandoned parts.
2839 if (now_ms / 60_000) % 60 == 37 {
2840 if let Err(error) = self.sweep_artifacts(now_ms).await {
2841 worker::console_error!("actions: the artifacts' sweep failed: {error}");
2842 }
2843 if let Err(error) = self.sweep_blob_parts(now_ms).await {
2844 worker::console_error!("actions: the blob parts' sweep failed: {error}");
2845 }
2846 }
Merge branch 'open-signups'2847 // And logs of jobs that finished longer ago than LOG_RETENTION_DAYS.
2848 if (now_ms / 60_000) % 60 == 17
2849 && let Err(error) = self.sweep_logs(now_ms).await
2850 {
2851 worker::console_error!("actions: the logs' sweep failed: {error}");
2852 }
Download ZIP reads blobs eight batches at a time and deflates files together2853 self.start_queued().await
2854 }
2855}
2856
2857
2858#[cfg(test)]
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2859mod log_chunks {
2860 use rusqlite::{Connection, params};
2861
2862 use super::LOG_INSERT;
2863
2864 /// The logs table as the migrations leave it.
2865 fn database() -> Connection {
2866 let db = Connection::open_in_memory().unwrap();
2867 for name in ["0001_init.sql", "0016_log_chunks.sql"] {
2868 let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));
2869 db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
2870 }
2871 db
2872 }
2873
2874 fn log(db: &Connection, job: &str) -> Vec<(i64, i64, String)> {
2875 let mut rows = db.prepare("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq").unwrap();
2876 rows.query_map([job], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).unwrap().map(Result::unwrap).collect()
2877 }
2878
2879 #[test]
2880 fn a_batch_sent_again_is_kept_once() {
2881 let db = database();
2882 let insert = |job: &str, step: i64, text: &str, chunk: Option<i64>| db.execute(LOG_INSERT, params![job, step, text, chunk]).unwrap();
2883 assert_eq!(insert("job_1", 0, "one\n", Some(1)), 1);
2884 // The runner did not hear the answer and sent the batch again.
2885 assert_eq!(insert("job_1", 0, "one\n", Some(1)), 0);
2886 assert_eq!(insert("job_1", 1, "two\n", Some(2)), 1);
2887 // Another job numbers its batches from 1 too.
2888 assert_eq!(insert("job_2", 0, "other\n", Some(1)), 1);
2889 // A runner that numbers nothing has every batch kept, as before.
2890 assert_eq!(insert("job_3", 0, "a\n", None), 1);
2891 assert_eq!(insert("job_3", 0, "a\n", None), 1);
2892 assert_eq!(log(&db, "job_1"), [(1, 0, "one\n".into()), (2, 1, "two\n".into())]);
2893 assert_eq!(log(&db, "job_2"), [(1, 0, "other\n".into())]);
2894 assert_eq!(log(&db, "job_3").len(), 2);
2895 // A job run again moves its log aside (`{id}.{attempt}`), and the
2896 // new attempt's runner numbers from 1 again.
2897 db.execute("UPDATE logs SET job_id = job_id || '.1' WHERE job_id = 'job_1'", []).unwrap();
2898 assert_eq!(insert("job_1", 0, "again\n", Some(1)), 1);
2899 assert_eq!(log(&db, "job_1"), [(1, 0, "again\n".into())]);
2900 assert_eq!(log(&db, "job_1.1").len(), 2);
2901 }
2902}
2903
2904#[cfg(test)]
Merge the self-hosted runner's host-side hardening: a job's token is spent once its spec is fetched (rotated, with brokered secrets still found by the token the job started with), job folders a crashed runner left are removed, the runner's folders are owner-only on Windows and Unix, the sandbox warns of a host-mode runner under the same account, its WSL default user is a locked account rather than root, a VM's seed no longer keeps the registration token, and the self-hosted runners guide says what your computer can see.2905mod token_rotation {
2906 use g1t_secrets::sha256_hex;
2907 use rusqlite::{Connection, OptionalExtension, params};
2908
2909 use super::{TOKEN_ROTATE, token_opens};
2910
2911 fn database() -> Connection {
2912 let db = Connection::open_in_memory().unwrap();
2913 for name in ["0001_init.sql", "0020_job_token_rotation.sql"] {
2914 let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));
2915 db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
2916 }
2917 db.execute(
2918 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, status, token_hash) VALUES ('job_1', 'run_1', 'repo_1', 'acme', 'build', 'build', 'in_progress', ?1)",
2919 [sha256_hex("first")],
2920 )
2921 .unwrap();
2922 db
2923 }
2924
2925 fn rotate(db: &Connection, next: &str, shown: &str) -> bool {
2926 db.query_row(TOKEN_ROTATE, params![sha256_hex(next), "job_1", sha256_hex(shown)], |row| row.get::<_, String>(0))
2927 .optional()
2928 .unwrap()
2929 .is_some()
2930 }
2931
2932 /// The job's current token's hash, and the one it started with.
2933 fn hashes(db: &Connection) -> (Option<String>, Option<String>) {
2934 db.query_row("SELECT token_hash, first_token_hash FROM jobs WHERE id = 'job_1'", [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()
2935 }
2936
2937 #[test]
2938 fn a_job_token_fetches_its_spec_once() {
2939 let db = database();
2940 assert!(rotate(&db, "second", "first"));
2941 assert_eq!(hashes(&db), (Some(sha256_hex("second")), Some(sha256_hex("first"))));
2942 // The first token, shown again (a copy left on the computer, or a
2943 // second fetch racing the first), swaps nothing and is refused.
2944 assert!(!rotate(&db, "third", "first"));
2945 assert_eq!(hashes(&db).0, Some(sha256_hex("second")));
2946 // A job that ended has no token to swap.
2947 db.execute("UPDATE jobs SET status = 'completed', token_hash = NULL, first_token_hash = NULL", []).unwrap();
2948 assert!(!rotate(&db, "fourth", "second"));
2949 assert_eq!(hashes(&db), (None, None));
2950 }
2951
2952 #[test]
2953 fn the_worker_finds_brokered_secrets_with_the_token_the_job_started_with() {
2954 let db = database();
2955 let opens = |db: &Connection, token: &str, started: bool| {
2956 let (current, first) = hashes(db);
2957 token_opens(current.as_deref(), first.as_deref(), token, started)
2958 };
2959 // Before any swap, the job's one token opens everything.
2960 assert!(opens(&db, "first", false) && opens(&db, "first", true));
2961 assert!(rotate(&db, "second", "first"));
2962 // The harness's new token opens everything.
2963 assert!(opens(&db, "second", false) && opens(&db, "second", true));
2964 // The token it started with, which the runner Worker holds, still
2965 // finds the job's brokered secrets, and opens nothing else.
2966 assert!(opens(&db, "first", true));
2967 assert!(!opens(&db, "first", false));
2968 assert!(!opens(&db, "other", true));
2969 // Another swap keeps the token it started with, not the one between.
2970 assert!(rotate(&db, "third", "second"));
2971 assert!(opens(&db, "first", true));
2972 assert!(!opens(&db, "second", true));
2973 }
2974
2975 #[test]
2976 fn the_runtime_token_is_signed_for_the_token_that_replaces_the_first() {
2977 let now = 1_800_000_000;
2978 let token = crate::runtime::runtime_token("job_1", "run_1", &sha256_hex("second"), now, 600);
2979 assert!(crate::runtime::runtime_valid(&token, "job_1", &sha256_hex("second"), now));
2980 assert!(!crate::runtime::runtime_valid(&token, "job_1", &sha256_hex("first"), now));
2981 }
2982}
2983
2984#[cfg(test)]
Download ZIP reads blobs eight batches at a time and deflates files together2985mod stopping {
2986 use super::stops_runs;
2987 use g1t_contracts::events::Event;
2988 use serde_json::{Value, json};
2989
2990 fn event(kind: &str, data: Value) -> Event {
2991 Event {
2992 id: "evt_1".into(),
2993 kind: kind.into(),
2994 source: "repos".into(),
2995 time: "2026-10-05T00:00:00Z".into(),
2996 repo_id: Some("rep_1".into()),
2997 actor: None,
2998 data,
2999 }
3000 }
3001
3002 #[test]
3003 fn deleting_or_archiving_stops_runs() {
3004 assert_eq!(stops_runs(&event("repo.deleted", json!({ "repoId": "rep_1" }))).as_deref(), Some("rep_1"));
3005 assert_eq!(stops_runs(&event("repo.archived", json!({ "archived": true }))).as_deref(), Some("rep_1"));
3006 assert_eq!(stops_runs(&event("repo.unarchived", json!({ "archived": false }))), None);
3007 assert_eq!(stops_runs(&event("repo.restored", json!({}))), None);
3008 assert_eq!(stops_runs(&event("git.push", json!({}))), None);
3009 }
3010}
3011
3012#[cfg(test)]
3013mod status_of_needs {
3014 use std::collections::HashMap;
3015
3016 use super::ancestor_failed;
3017
3018 /// check -> plan -> (migrate) -> core -> edge, as deploy.yml has them,
3019 /// and a job that needs only the last.
3020 fn graph() -> HashMap<&'static str, Vec<&'static str>> {
3021 HashMap::from([
3022 ("check", vec![]),
3023 ("plan", vec!["check"]),
3024 ("migrate", vec!["plan"]),
3025 ("core", vec!["plan", "migrate"]),
3026 ("edge", vec!["plan", "migrate", "core"]),
3027 ("notify", vec!["edge"]),
3028 ])
3029 }
3030
3031 #[test]
3032 fn a_failure_is_seen_however_far_back() {
3033 let needs = graph();
3034 let failed = |which: &'static str| move |key: &str| key == which;
3035 // check failed; plan, and everything after, was skipped for it.
3036 assert!(ancestor_failed(&needs, "notify", failed("check")));
3037 assert!(ancestor_failed(&needs, "core", failed("check")));
3038 assert!(ancestor_failed(&needs, "edge", failed("core")));
3039 // Nothing before a job failed: a skipped migrate is not a failure.
3040 assert!(!ancestor_failed(&needs, "edge", |_| false));
3041 assert!(!ancestor_failed(&needs, "core", failed("edge")));
3042 assert!(!ancestor_failed(&needs, "check", failed("check")));
3043 }
3044
3045 #[test]
3046 fn cycles_and_unknown_keys_end() {
3047 let needs = HashMap::from([("a", vec!["b"]), ("b", vec!["a"])]);
3048 assert!(!ancestor_failed(&needs, "a", |_| false));
3049 assert!(!ancestor_failed(&needs, "missing", |_| true));
3050 }
3051}
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb973052
3053#[cfg(test)]
3054mod deployments {
3055 use serde_json::{Map, Value, json};
3056
3057 use super::{JobEnvironment, deployment_outcome, environment_of};
3058
3059 #[test]
3060 fn a_jobs_environment_is_read_for_deployments() {
3061 let contexts: Map<String, Value> = serde_json::from_value(json!({
3062 "github": { "ref_name": "main", "repository": "acme/web" },
3063 "inputs": { "target": "staging" },
3064 "matrix": {},
3065 }))
3066 .unwrap();
3067 let read = |raw: Value| environment_of(&raw, &contexts);
3068 assert_eq!(read(json!({})), None);
3069 assert_eq!(
3070 read(json!({ "environment": "production" })),
3071 Some(JobEnvironment { name: "production".into(), url: None, deploys: true })
3072 );
3073 assert_eq!(
3074 read(json!({ "environment": { "name": "production", "url": "https://g1t.sh" } })),
3075 Some(JobEnvironment { name: "production".into(), url: Some("https://g1t.sh".into()), deploys: true })
3076 );
3077 // Expressions are filled in from the run.
3078 assert_eq!(
3079 read(json!({ "environment": { "name": "${{ inputs.target }}", "url": "https://${{ github.ref_name }}.example.com" } })),
3080 Some(JobEnvironment { name: "staging".into(), url: Some("https://main.example.com".into()), deploys: true })
3081 );
3082 // Secrets only: no deployment.
3083 assert!(!read(json!({ "environment": { "name": "production", "deployment": false } })).unwrap().deploys);
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r93084 // The same, said in the job's g1t settings.
3085 assert!(!read(json!({ "environment": "production", "g1t": { "deployment": false } })).unwrap().deploys);
3086 assert!(!read(json!({ "environment": { "name": "production" }, "g1t": { "deployment": false } })).unwrap().deploys);
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb973087 // Only http(s) addresses.
3088 assert_eq!(read(json!({ "environment": { "name": "production", "url": "javascript:alert(1)" } })).unwrap().url, None);
3089 }
3090
3091 #[test]
3092 fn a_runs_outcome_for_an_environment() {
3093 let of = |list: &[&str]| deployment_outcome(&list.iter().map(|c| Some((*c).to_owned())).collect::<Vec<_>>());
3094 assert_eq!(of(&["success", "skipped"]), Some("success"));
3095 assert_eq!(of(&["success", "failure"]), Some("failure"));
3096 assert_eq!(of(&["success", "cancelled"]), Some("error"));
3097 assert_eq!(of(&["skipped"]), None);
3098 assert_eq!(deployment_outcome(&[None]), None);
3099 }
3100}
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)3101
3102#[cfg(test)]
3103mod reruns {
3104 use serde_json::{Map, Value, json};
3105
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r93106 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)3107
3108 /// build ← test ← deploy, and lint on its own.
3109 fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {
3110 let order = ["build", "lint", "test", "deploy"];
3111 let needs = |key: &str| -> Vec<String> {
3112 match key {
3113 "test" => vec!["build".into()],
3114 "deploy" => vec!["test".into()],
3115 _ => Vec::new(),
3116 }
3117 };
3118 rerun_keys(&order, needs, |key| !failed.contains(&key), which)
3119 }
3120
3121 #[test]
3122 fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {
3123 assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);
3124 assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);
3125 assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);
3126 assert!(keys(Rerun::Failed, &[]).is_empty());
3127 // One job, whatever it came to, and what depends on it.
3128 assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);
3129 assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);
3130 assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);
3131 assert!(keys(Rerun::Job("missing"), &[]).is_empty());
3132 }
3133
3134 #[test]
3135 fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {
3136 assert_eq!(top_key("build/test"), "build");
3137 assert_eq!(top_key("build/inner/test"), "build");
3138 assert_eq!(top_key("lint"), "lint");
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r93139 // A matrix's calls each have their own jobs, under `key[i]/`.
3140 assert_eq!(top_key("build[1]/test"), "build");
3141 assert_eq!(top_key("build/inner[0]/test"), "build");
3142 assert_eq!(call_prefix("build", Some(1)), "build[1]");
3143 assert_eq!(call_prefix("build/inner", None), "build/inner");
3144 }
3145
3146 #[test]
3147 fn a_run_connects_ten_levels_of_workflows_and_fifty_reusable_ones() {
3148 assert!(call_allowed(1, &[], "a.yml").is_ok());
3149 assert!(call_allowed(9, &[], "a.yml").is_ok());
3150 let deep = call_allowed(10, &[], "a.yml").unwrap_err();
3151 assert!(deep.contains("at most ten") && deep.contains("nine levels"), "{deep}");
3152 let names: Vec<String> = (0..50).map(|n| format!("w{n}.yml")).collect();
3153 let fifty: Vec<&str> = names.iter().map(String::as_str).collect();
3154 // A workflow called again, as by each combination of a matrix, is one.
3155 assert!(call_allowed(1, &fifty, "w3.yml").is_ok());
3156 assert!(call_allowed(1, &fifty, "another.yml").unwrap_err().contains("at most 50 different"));
3157 }
3158
3159 #[test]
3160 fn a_matrixs_calls_are_named_by_the_caller_and_its_values() {
3161 let workflow = super::workflow::parse(
3162 "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' }]",
3163 )
3164 .unwrap();
3165 let job = |id: &str| workflow.jobs.iter().find(|job| job.id == id).unwrap();
3166 let combination: Map<String, Value> = serde_json::from_value(json!({ "os": "linux", "node": 20 })).unwrap();
3167 let mut contexts = Map::new();
3168 contexts.insert("matrix".into(), Value::Object(combination.clone()));
3169 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
3170 assert_eq!(shown_name(job("build"), None, &combination, &scope), "Build (linux, 20)");
3171 assert_eq!(shown_name(job("plain"), None, &Map::new(), &scope), "plain");
3172 // A called workflow's job, under the call that made it.
3173 assert_eq!(shown_name(job("plain"), Some("Build (linux, 20)"), &Map::new(), &scope), "Build (linux, 20) / plain");
3174 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)3175 }
3176
3177 #[test]
3178 fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {
3179 assert!(summary_fits(0, None, 10));
3180 assert!(!summary_fits(0, None, 0), "nothing to keep");
3181 assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");
3182 assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");
3183 assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));
3184 assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));
3185 assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));
3186 }
3187
3188 #[test]
3189 fn a_debug_rerun_sets_what_github_sets() {
3190 let mut variables = Map::new();
3191 let mut runner = json!({ "name": "g1t", "debug": "" });
3192 debug_logging(&mut variables, &mut runner);
3193 assert_eq!(variables["RUNNER_DEBUG"], "1");
3194 assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");
3195 assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");
3196 assert_eq!(runner["debug"], Value::String("1".into()));
3197 assert_eq!(runner["name"], "g1t");
3198 }
3199}

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