Skip to content
Sign up

flagon-io/g1t

public
0

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

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

Merge branch 'worktree-agent-a6a121745e81f639f'1//! 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};
Merge branch 'worktree-agent-a6a121745e81f639f'13use 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;
Merge branch 'worktree-agent-a6a121745e81f639f'24use 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 main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.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}
Merge branch 'worktree-agent-a6a121745e81f639f'71
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>,
Merge branch 'worktree-agent-a6a121745e81f639f'107}
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,
Merge branch 'worktree-agent-a6a121745e81f639f'150}
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
Merge branch 'worktree-agent-a6a121745e81f639f'157 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 main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.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");
Merge branch 'worktree-agent-a6a121745e81f639f'358}
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
Merge branch 'worktree-agent-a6a121745e81f639f'383 /// 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;
Merge branch 'worktree-agent-a6a121745e81f639f'402
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
Merge branch 'worktree-agent-a6a121745e81f639f'508/// 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();
Merge branch 'worktree-agent-a6a121745e81f639f'527 // `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);
Merge branch 'worktree-agent-a6a121745e81f639f'530 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
Merge branch 'worktree-agent-a6a121745e81f639f'569impl 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
Merge branch 'worktree-agent-a6a121745e81f639f'632 /// 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'654 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 });
Merge branch 'worktree-agent-a6a121745e81f639f'688 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",
Merge branch 'worktree-agent-a6a121745e81f639f'695 )
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(),
Merge branch 'worktree-agent-a6a121745e81f639f'709 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(),
Merge branch 'worktree-agent-a6a121745e81f639f'711 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(),
Merge branch 'worktree-agent-a6a121745e81f639f'722 ])?
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;
Merge branch 'worktree-agent-a6a121745e81f639f'771 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()])?
Merge branch 'worktree-agent-a6a121745e81f639f'777 .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?;
Merge branch 'worktree-agent-a6a121745e81f639f'784 }
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()])?
Merge branch 'worktree-agent-a6a121745e81f639f'790 .run()
791 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'792 return Ok(());
Merge branch 'worktree-agent-a6a121745e81f639f'793 }
794 }
Merge branch 'worktree-agent-a3abfcce648e87dca'795 self.advance(&run.id).await
Merge branch 'worktree-agent-a6a121745e81f639f'796 }
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") {
Merge branch 'worktree-agent-a6a121745e81f639f'855 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()) {
Merge branch 'worktree-agent-a6a121745e81f639f'891 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());
Merge branch 'worktree-agent-a6a121745e81f639f'905 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())))
Merge branch 'worktree-agent-a6a121745e81f639f'908 .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?;
Merge branch 'worktree-agent-a6a121745e81f639f'911 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'930 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 });
Merge branch 'worktree-agent-a6a121745e81f639f'950 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1000 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();
Merge branch 'worktree-agent-a6a121745e81f639f'1007 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);
Merge branch 'worktree-agent-a6a121745e81f639f'1021 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));
Merge branch 'worktree-agent-a6a121745e81f639f'1075 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 {
Merge branch 'worktree-agent-a6a121745e81f639f'1080 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1127 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(),
Merge branch 'worktree-agent-a6a121745e81f639f'1142 ];
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 = ?",
Merge branch 'worktree-agent-a6a121745e81f639f'1152 )
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Merge branch 'worktree-agent-a6a121745e81f639f'1174 )
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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1227 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<()> {
Merge branch 'worktree-agent-a6a121745e81f639f'1247 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;
Merge branch 'worktree-agent-a6a121745e81f639f'1250 }
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,
Merge branch 'worktree-agent-a6a121745e81f639f'1254 };
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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1267 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1418 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"],
Merge branch 'worktree-agent-a6a121745e81f639f'1424 });
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(),
Merge branch 'worktree-agent-a6a121745e81f639f'1434 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(),
Merge branch 'worktree-agent-a6a121745e81f639f'1437 ])?,
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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1463 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<()> {
Merge branch 'worktree-agent-a6a121745e81f639f'1470 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")
Merge branch 'worktree-agent-a6a121745e81f639f'1508 .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)
Merge branch 'worktree-agent-a6a121745e81f639f'1510 .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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1529 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>> {
Merge branch 'worktree-agent-a6a121745e81f639f'1534 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1623 let token = random_hex(24);
1624 let at = now();
1625 let claimed = self
1626 .db
1627 .prepare(
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.1628 "UPDATE jobs SET status = 'in_progress', token_hash = ?, first_token_hash = NULL, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
Merge branch 'worktree-agent-a6a121745e81f639f'1629 )
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?;
Merge branch 'worktree-agent-a6a121745e81f639f'1652 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,
Merge branch 'worktree-agent-a6a121745e81f639f'1663 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,
Merge branch 'worktree-agent-a6a121745e81f639f'1665 },
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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1681 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?;
Merge branch 'worktree-agent-a6a121745e81f639f'1684 }
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 main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.1695 finished_at = ?, token_hash = NULL, first_token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
Merge branch 'worktree-agent-a6a121745e81f639f'1696 )
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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1722 // 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);
Merge branch 'worktree-agent-a6a121745e81f639f'1750 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1767 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.
Merge branch 'worktree-agent-a6a121745e81f639f'1774 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<()> {
Merge branch 'worktree-agent-a6a121745e81f639f'1797 // 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 main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.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 *")
Merge branch 'worktree-agent-a6a121745e81f639f'1804 .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 }
Merge branch 'worktree-agent-a6a121745e81f639f'1811 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
Merge branch 'worktree-agent-a6a121745e81f639f'1827 /// 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") {
Merge branch 'worktree-agent-a6a121745e81f639f'1831 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;
Merge branch 'worktree-agent-a6a121745e81f639f'1863 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
Merge branch 'worktree-agent-a6a121745e81f639f'2015 /// 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",
Merge branch 'worktree-agent-a6a121745e81f639f'2038 }),
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 },
Merge branch 'worktree-agent-a6a121745e81f639f'2059 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge2060 "source": "actions",
Merge branch 'worktree-agent-a6a121745e81f639f'2061 }),
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<()> {
Merge branch 'worktree-agent-a6a121745e81f639f'2070 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 }
Merge branch 'worktree-agent-a6a121745e81f639f'2081 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2082 if run.status == "pending" || run.status == "action_required" {
Merge branch 'worktree-agent-a6a121745e81f639f'2083 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?;
Merge branch 'worktree-agent-a6a121745e81f639f'2100 }
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?;
Merge branch 'worktree-agent-a6a121745e81f639f'2121 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`).
Merge branch 'worktree-agent-a6a121745e81f639f'2128 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?);
Merge branch 'worktree-agent-a6a121745e81f639f'2148 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 );
Merge branch 'worktree-agent-a6a121745e81f639f'2183 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 ];
Merge branch 'worktree-agent-a6a121745e81f639f'2223 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.
Merge branch 'worktree-agent-a6a121745e81f639f'2227 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()])?,
Merge branch 'worktree-agent-a6a121745e81f639f'2231 );
2232 statements.push(
2233 self.db
2234 .prepare(
2235 "UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.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 = ?",
Merge branch 'worktree-agent-a6a121745e81f639f'2239 )
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()])?,
Merge branch 'worktree-agent-a6a121745e81f639f'2250 );
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;
Merge branch 'worktree-agent-a6a121745e81f639f'2261 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 main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.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>> {
Merge branch 'worktree-agent-a6a121745e81f639f'2289 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 main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.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),
Merge branch 'worktree-agent-a6a121745e81f639f'2292 _ => 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 main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.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 };
Merge branch 'worktree-agent-a6a121745e81f639f'2318 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));
Merge branch 'worktree-agent-a6a121745e81f639f'2352 }
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(),
Merge branch 'worktree-agent-a6a121745e81f639f'2380 };
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 }
Merge branch 'worktree-agent-a6a121745e81f639f'2414 };
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
Merge branch 'main' into rm-agents2418 // the Worker: a harness there that brokers is given each stand-in's
2419 // hosts, and passes its job's requests to them through g1t
2420 // (broker.rs); any other keeps a stand-in that does nothing, and
2421 // the log says so.
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.2422 let stand_ins = broker::substitute(&mut secrets, &brokers, broker::stand_in);
Merge branch 'main' into rm-agents2423 let relayed = job.runner_id.is_some() && a.broker && !stand_ins.is_empty();
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.2424 if !stand_ins.is_empty() {
2425 self.keep_stand_ins(&job.id, job.timeout_minutes, &stand_ins).await?;
2426 if job.runner_id.is_some() {
Merge branch 'main' into rm-agents2427 let mut made: Vec<&broker::StandIn> = stand_ins.iter().collect();
2428 made.sort_unstable_by(|a, b| a.broker.name.cmp(&b.broker.name));
2429 let notes: String = made
2430 .iter()
2431 .map(|made| {
2432 let note = if relayed { broker::brokered_note(&made.broker.name, &made.broker.hosts) } else { broker::self_hosted_note(&made.broker.name) };
2433 note + "\n"
2434 })
2435 .collect();
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.2436 self.note_log(&job.id, 0, &notes).await?;
2437 }
2438 }
Merge branch 'worktree-agent-a6a121745e81f639f'2439 secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));
2440 secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));
Merge branch 'worktree-agent-a3abfcce648e87dca'2441 // Each secret as it is, a line at a time, base64 and JSON-escaped.
2442 let mut masks: Vec<String> = g1t_actions::mask::all_variants(secrets.values().filter_map(|v| v.as_str()));
Merge branch 'worktree-agent-a6a121745e81f639f'2443 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 R22444 // The toolkit's runtime token (runtime.rs), for as long as the job
2445 // may run; the API puts it and the toolkit's addresses in the
2446 // job's variables.
2447 let runtime_token = crate::runtime::runtime_token(
2448 &job.id,
2449 &job.run_id,
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.2450 &token_hash,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22451 now_ms() / 1000,
2452 u64::from(job.timeout_minutes) * 60 + 600,
2453 );
2454 masks.push(runtime_token.clone());
2455 let retention_days = self.retention_setting(&run.repo_id).await?;
Merge branch 'worktree-agent-a6a121745e81f639f'2456
2457 let jobs = self.job_rows(&run.id).await?;
2458 let mut needs = Map::new();
2459 // In a called workflow, its jobs' keys sit under the job that called it.
2460 let parent = job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));
2461 for need in &spec.needs {
2462 let key = match &parent {
2463 Some(parent) => format!("{parent}/{need}"),
2464 None => need.clone(),
2465 };
2466 let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();
2467 let mut outputs = Map::new();
2468 for row in &rows {
2469 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
2470 outputs.extend(more);
2471 }
2472 }
2473 needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));
2474 }
2475 let siblings = jobs.iter().filter(|row| row.key == job.key).count();
2476 let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));
2477 let info = run.info();
2478 let mut github = info.context(&job.key, &token, run.action.as_deref());
2479 github["token"] = json!(token);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22480 github["retention_days"] = json!(retention_days);
Merge branch 'worktree-agent-a6a121745e81f639f'2481 // On a self-hosted runner, `runner` and `RUNNER_*` describe that
2482 // machine rather than g1t's sandbox.
2483 let mut variables = info.variables(&job.key);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22484 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)2485 let mut runner = match &job.runner_id {
Merge branch 'worktree-agent-a6a121745e81f639f'2486 Some(id) => self.runner_context_for(id, &mut variables).await?,
2487 None => runner_context(),
2488 };
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2489 // A re-run with debug logging: what GitHub sets for one.
2490 if run.debug != 0 {
2491 debug_logging(&mut variables, &mut runner);
2492 }
Merge branch 'worktree-agent-a6a121745e81f639f'2493
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2494 // Where to check out: the repository, which holds a pull request's
2495 // test merge commit; a pull request's fork when the run builds its
2496 // head instead (`run.sha` is the head, `info.sha` what is built).
Merge branch 'worktree-agent-a6a121745e81f639f'2497 let clone_url = match run.pull {
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2498 Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" && info.sha == run.sha => {
Merge branch 'worktree-agent-a6a121745e81f639f'2499 let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(
2500 &self.work,
2501 "get_pull",
2502 &g1t_contracts::work::ViewArgs {
2503 repo: repo.clone(),
2504 number,
2505 viewer: self.workspace_actor(&repo.namespace).await?,
2506 after_seq: 0,
2507 },
2508 )
2509 .await?;
2510 match located {
2511 Outcome::Ok(detail) => match detail.pull.fork {
2512 Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),
2513 None => format!("{SITE}/{}.git", run.repo),
2514 },
2515 Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),
2516 }
2517 }
2518 _ => format!("{SITE}/{}.git", run.repo),
2519 };
2520
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.2521 // Only one fetch with a token gets its spec: of two at once, the
2522 // second finds the hash already changed and is refused.
2523 if next.is_some() {
2524 let swapped = self
2525 .db
2526 .prepare(TOKEN_ROTATE)
2527 .bind(&[token_hash.as_str().into(), job.id.as_str().into(), job.token_hash.as_deref().unwrap_or_default().into()])?
2528 .first::<Value>(None)
2529 .await?;
2530 if swapped.is_none() {
2531 return Ok(fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."));
2532 }
2533 }
2534
2535 let mut spec = json!({
Merge branch 'worktree-agent-a6a121745e81f639f'2536 "job": job.id,
2537 "run": run.id,
2538 "key": job.key,
2539 "name": job.name,
2540 "spec": spec.raw,
2541 "workflow": {
2542 "env": workflow.env,
2543 "defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),
2544 },
2545 "github": github,
2546 "variables": variables,
2547 "event": info.event,
2548 "contexts": {
2549 "vars": vars,
2550 "secrets": secrets,
2551 "inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),
2552 "matrix": matrix,
2553 "needs": needs,
2554 "strategy": {
2555 "fail-fast": spec.fail_fast,
2556 "job-index": job.ordinal,
2557 "job-total": siblings,
2558 "max-parallel": spec.max_parallel.unwrap_or(siblings as u32),
2559 },
2560 "runner": runner,
2561 },
2562 "checkout": {
2563 "repository": run.repo,
2564 "url": clone_url,
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2565 "sha": info.sha,
Merge branch 'worktree-agent-a6a121745e81f639f'2566 "ref": run.git_ref,
2567 "token": token,
2568 },
2569 "timeoutMinutes": job.timeout_minutes,
2570 "masks": masks,
Merge branch 'worktree-agent-a3abfcce648e87dca'2571 // As the job's log lists them at its start.
2572 "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 R22573 // Whether the job may ask for an OIDC token decides whether it
2574 // is told where to.
2575 "runtime": {
2576 "token": runtime_token,
2577 "idToken": self.oidc_allowed(&run, &job),
2578 },
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.2579 });
2580 if let Some(next) = next {
2581 spec["nextToken"] = json!(next);
2582 }
Merge branch 'main' into rm-agents2583 if relayed {
2584 spec["brokered"] = broker::spec_entries(&stand_ins);
2585 }
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.2586 Ok(Outcome::Ok(spec))
Merge branch 'worktree-agent-a6a121745e81f639f'2587 }
2588
2589 /// `job_report`: the sandbox telling how the job is going.
2590 pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2591 let job = check!(self.job_for_token(&a).await?);
2592 let report = &a.report;
2593 let at = now();
2594 match report["kind"].as_str().unwrap_or_default() {
2595 "steps" => {
2596 // The list can grow as the job goes (post steps), so steps
2597 // already reported keep where they stand.
2598 let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2599 let steps: Vec<Value> = report["steps"]
2600 .as_array()
2601 .map(|names| {
2602 names
2603 .iter()
2604 .enumerate()
2605 .map(|(i, name)| match known.get(i) {
2606 Some(step) if step["status"] != "queued" => step.clone(),
2607 _ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),
2608 })
2609 .collect()
2610 })
2611 .unwrap_or_default();
2612 self.db
2613 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2614 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2615 .run()
2616 .await?;
2617 }
2618 "step" => {
2619 let number = report["number"].as_u64().unwrap_or(0) as usize;
2620 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2621 if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {
2622 let status = report["status"].as_str().unwrap_or("in_progress");
2623 step["status"] = json!(status);
2624 if status == "in_progress" {
2625 step["startedAt"] = json!(at);
2626 }
2627 if status == "completed" {
2628 step["finishedAt"] = json!(at);
2629 step["conclusion"] = report["conclusion"].clone();
2630 }
2631 if let Some(name) = report["name"].as_str() {
2632 step["name"] = json!(name);
2633 }
2634 }
2635 self.db
2636 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2637 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2638 .run()
2639 .await?;
2640 }
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 line from g1t itself, not the job: the runner Worker's note
2642 // that a brokered secret went to a host it is not for. Put
2643 // under the step that is running.
2644 "note" => {
2645 let text = report["text"].as_str().unwrap_or_default().chars().take(1000).collect::<String>();
2646 let steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2647 let step = steps.iter().rev().find(|step| step["status"] == "in_progress").and_then(|step| step["number"].as_u64()).unwrap_or(0);
2648 if !text.trim().is_empty() {
2649 self.note_log(&job.id, step as u32, &text).await?;
2650 }
2651 }
Merge branch 'worktree-agent-a6a121745e81f639f'2652 "log" => {
2653 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.2654 // A brokered secret's value never enters the sandbox, but an
2655 // answer that echoed it whole could carry it in; it is
2656 // masked here, where the value is known.
2657 if job.brokered != 0 {
2658 let values: Vec<String> = self.stand_ins_of(&job.id).await?.into_iter().map(|secret| secret.value).collect();
2659 if !values.is_empty() {
2660 text = g1t_actions::mask::apply(&text, &g1t_actions::mask::all_variants(values.iter().map(String::as_str)));
2661 }
2662 }
Merge branch 'worktree-agent-a6a121745e81f639f'2663 if text.len() > MAX_CHUNK_BYTES {
2664 let mut cut = MAX_CHUNK_BYTES;
2665 while !text.is_char_boundary(cut) {
2666 cut -= 1;
2667 }
2668 text.truncate(cut);
2669 }
2670 #[derive(Deserialize)]
2671 struct Size {
2672 n: Option<f64>,
2673 }
2674 let size = self
2675 .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.2676 .prepare("SELECT SUM(LENGTH(text)) AS n FROM logs WHERE job_id = ?")
Merge branch 'worktree-agent-a6a121745e81f639f'2677 .bind(&[job.id.as_str().into()])?
2678 .first::<Size>(None)
2679 .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.2680 let used = size.and_then(|s| s.n).unwrap_or(0.0) as usize;
Merge branch 'worktree-agent-a6a121745e81f639f'2681 if used < MAX_LOG_BYTES {
2682 if used + text.len() >= MAX_LOG_BYTES {
2683 text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");
2684 }
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.2685 // The next `seq` is taken in the insert itself, so two
2686 // batches at once (a report sent again while the first
2687 // was still being written) never reach for the same one
2688 // and lose a chunk to the primary key.
Merge branch 'worktree-agent-a6a121745e81f639f'2689 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.2690 .prepare(LOG_INSERT)
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2691 .bind(&[
2692 job.id.as_str().into(),
2693 (report["step"].as_u64().unwrap_or(0) as u32).into(),
2694 text.into(),
2695 report["chunk"].as_u64().map_or(worker::wasm_bindgen::JsValue::NULL, |n| (n as f64).into()),
2696 ])?
Merge branch 'worktree-agent-a6a121745e81f639f'2697 .run()
2698 .await?;
2699 }
2700 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2701 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2702 "summary" => {
2703 // `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the
2704 // step's summary, up to 1 MiB a step and 20 steps a job.
2705 let step = report["step"].as_u64().unwrap_or(0) as u32;
2706 let markdown = report["markdown"].as_str().unwrap_or_default();
2707 #[derive(Deserialize)]
2708 struct Held {
2709 steps: u32,
2710 mine: Option<f64>,
2711 }
2712 let held = self
2713 .db
2714 .prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")
2715 .bind(&[step.into(), job.id.as_str().into()])?
2716 .first::<Held>(None)
2717 .await?;
2718 let (steps, mine) = held.map_or((0, None), |held| (held.steps, held.mine.map(|n| n as usize)));
2719 if summary_fits(steps, mine, markdown.len()) {
2720 self.db
2721 .prepare(
2722 "INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)
2723 ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown || excluded.markdown",
2724 )
2725 .bind(&[job.id.as_str().into(), step.into(), markdown.into()])?
2726 .run()
2727 .await?;
2728 }
2729 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2730 }
Merge branch 'worktree-agent-a6a121745e81f639f'2731 "annotation" => {
2732 let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
2733 if annotations.len() < MAX_ANNOTATIONS {
2734 annotations.push(json!({
2735 "level": report["level"].as_str().unwrap_or("notice"),
2736 "message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),
2737 "title": report["title"],
2738 "file": report["file"],
2739 "line": report["line"],
2740 }));
2741 self.db
2742 .prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")
2743 .bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?
2744 .run()
2745 .await?;
2746 }
2747 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2748 // Nothing to tell; the answer says whether to stop.
2749 "ping" => {
2750 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2751 }
Merge branch 'worktree-agent-a6a121745e81f639f'2752 "done" => {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2753 // A job told to stop ends cancelled, however its cleanup went.
2754 let conclusion = if job.cancel_requested_at.is_some() {
2755 "cancelled"
2756 } else {
2757 report["conclusion"]
2758 .as_str()
2759 .filter(|c| matches!(*c, "success" | "failure" | "cancelled"))
2760 .unwrap_or("failure")
2761 };
Merge branch 'worktree-agent-a6a121745e81f639f'2762 let outputs = report["outputs"].as_object().cloned();
2763 Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
2764 }
2765 other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
2766 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2767 // `cancelled`: the run was cancelled, and the runner should stop
2768 // the step it is on and run only its cleanup steps.
2769 Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))
Merge branch 'worktree-agent-a6a121745e81f639f'2770 }
2771
2772 // --- Every minute ---------------------------------------------------------------
2773
2774 pub async fn on_minute(&self, now_ms: u64) -> Result<()> {
2775 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.2776 // Repositories a workspace's managed workflows changed in, beyond
2777 // those synced at once (managed.rs): their schedules wait for it.
2778 if let Err(error) = self.catch_up_syncs().await {
2779 worker::console_error!("actions: catching up syncs failed: {error}");
2780 }
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)2781 // Staff (or billing's usage watcher) paused scheduled runs across
2782 // g1t: this minute's schedules are skipped, not queued for later.
2783 // Kept 30 seconds in the isolate (g1t_kit::pause).
2784 if g1t_kit::pause::paused(&self.billing, g1t_contracts::billing::PauseLevel::Schedules).await {
2785 worker::console_log!("actions: schedules are paused across g1t; skipped this minute's");
2786 } else if let Err(error) = self.run_schedules(minute).await {
Merge branch 'worktree-agent-a6a121745e81f639f'2787 worker::console_error!("actions: schedules failed: {error}");
2788 }
2789 // Jobs whose sandbox went quiet or ran past their time.
2790 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
2791 for job in running {
2792 // Times in g1t's format compare as text.
2793 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
2794 let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());
2795 let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
2796 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)2797 let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(|at| at < before(CANCEL_GRACE_MS).as_str());
2798 if cancel_overdue {
2799 // Cancelled, and still going after its grace period.
2800 self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;
2801 self.advance(&job.run_id).await?;
2802 } else if over {
Merge branch 'worktree-agent-a6a121745e81f639f'2803 let reason = format!("It ran longer than its time limit of {} minutes.", job.timeout_minutes);
2804 // A self-hosted runner is told to stop on its next poll.
2805 if job.runner_id.is_none() {
2806 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2807 }
2808 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2809 } else if silent {
2810 let reason = match &job.runner_name {
2811 Some(name) => format!("The self-hosted runner {name} stopped answering."),
2812 None => "The runner stopped answering.".to_owned(),
2813 };
2814 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2815 }
2816 }
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r92817 // Calls of reusable workflows waiting in a concurrency group, in
2818 // case the job they waited for was cancelled in another run.
2819 #[derive(Deserialize)]
2820 struct RunId {
2821 run_id: String,
2822 }
2823 let held = self
2824 .db
2825 .prepare("SELECT DISTINCT run_id FROM jobs WHERE status = 'waiting' AND concurrency_group IS NOT NULL LIMIT 20")
2826 .all()
2827 .await?
2828 .results::<RunId>()?;
2829 for run in held {
2830 self.advance(&run.run_id).await?;
2831 }
Merge branch 'worktree-agent-a6a121745e81f639f'2832 if let Err(error) = self.sweep_runners(now_ms).await {
2833 worker::console_error!("actions: the runners' sweep failed: {error}");
2834 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2835 // Jobs held at an environment whose wait timer has run out.
2836 if let Err(error) = self.release_gates().await {
2837 worker::console_error!("actions: environments' gates failed: {error}");
2838 }
Merge branch 'worktree-agent-a6a121745e81f639f'2839 // Once an hour: the cache's expired entries, and its storage.
2840 if (now_ms / 60_000) % 60 == 7
2841 && let Err(error) = self.sweep_cache(now_ms).await
2842 {
2843 worker::console_error!("actions: the cache's sweep failed: {error}");
2844 }
Merge remote-tracking branch 'origin/claude/peaceful-franklin-7qtkwx'2845 // And the archives of GitHub's actions no job uses any more.
2846 if (now_ms / 60_000) % 60 == 47
2847 && let Err(error) = self.sweep_action_archives(now_ms).await
2848 {
2849 worker::console_error!("actions: the action archives' sweep failed: {error}");
2850 }
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22851 // And artifacts past their time, and the toolkit's abandoned parts.
2852 if (now_ms / 60_000) % 60 == 37 {
2853 if let Err(error) = self.sweep_artifacts(now_ms).await {
2854 worker::console_error!("actions: the artifacts' sweep failed: {error}");
2855 }
2856 if let Err(error) = self.sweep_blob_parts(now_ms).await {
2857 worker::console_error!("actions: the blob parts' sweep failed: {error}");
2858 }
2859 }
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.2860 // And logs of jobs that finished longer ago than LOG_RETENTION_DAYS.
2861 if (now_ms / 60_000) % 60 == 17
2862 && let Err(error) = self.sweep_logs(now_ms).await
2863 {
2864 worker::console_error!("actions: the logs' sweep failed: {error}");
2865 }
Merge branch 'worktree-agent-a6a121745e81f639f'2866 self.start_queued().await
2867 }
2868}
2869
2870
2871#[cfg(test)]
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9'2872mod log_chunks {
2873 use rusqlite::{Connection, params};
2874
2875 use super::LOG_INSERT;
2876
2877 /// The logs table as the migrations leave it.
2878 fn database() -> Connection {
2879 let db = Connection::open_in_memory().unwrap();
2880 for name in ["0001_init.sql", "0016_log_chunks.sql"] {
2881 let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));
2882 db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
2883 }
2884 db
2885 }
2886
2887 fn log(db: &Connection, job: &str) -> Vec<(i64, i64, String)> {
2888 let mut rows = db.prepare("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq").unwrap();
2889 rows.query_map([job], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).unwrap().map(Result::unwrap).collect()
2890 }
2891
2892 #[test]
2893 fn a_batch_sent_again_is_kept_once() {
2894 let db = database();
2895 let insert = |job: &str, step: i64, text: &str, chunk: Option<i64>| db.execute(LOG_INSERT, params![job, step, text, chunk]).unwrap();
2896 assert_eq!(insert("job_1", 0, "one\n", Some(1)), 1);
2897 // The runner did not hear the answer and sent the batch again.
2898 assert_eq!(insert("job_1", 0, "one\n", Some(1)), 0);
2899 assert_eq!(insert("job_1", 1, "two\n", Some(2)), 1);
2900 // Another job numbers its batches from 1 too.
2901 assert_eq!(insert("job_2", 0, "other\n", Some(1)), 1);
2902 // A runner that numbers nothing has every batch kept, as before.
2903 assert_eq!(insert("job_3", 0, "a\n", None), 1);
2904 assert_eq!(insert("job_3", 0, "a\n", None), 1);
2905 assert_eq!(log(&db, "job_1"), [(1, 0, "one\n".into()), (2, 1, "two\n".into())]);
2906 assert_eq!(log(&db, "job_2"), [(1, 0, "other\n".into())]);
2907 assert_eq!(log(&db, "job_3").len(), 2);
2908 // A job run again moves its log aside (`{id}.{attempt}`), and the
2909 // new attempt's runner numbers from 1 again.
2910 db.execute("UPDATE logs SET job_id = job_id || '.1' WHERE job_id = 'job_1'", []).unwrap();
2911 assert_eq!(insert("job_1", 0, "again\n", Some(1)), 1);
2912 assert_eq!(log(&db, "job_1"), [(1, 0, "again\n".into())]);
2913 assert_eq!(log(&db, "job_1.1").len(), 2);
2914 }
2915}
2916
2917#[cfg(test)]
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.2918mod token_rotation {
2919 use g1t_secrets::sha256_hex;
2920 use rusqlite::{Connection, OptionalExtension, params};
2921
2922 use super::{TOKEN_ROTATE, token_opens};
2923
2924 fn database() -> Connection {
2925 let db = Connection::open_in_memory().unwrap();
2926 for name in ["0001_init.sql", "0020_job_token_rotation.sql"] {
2927 let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));
2928 db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();
2929 }
2930 db.execute(
2931 "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)",
2932 [sha256_hex("first")],
2933 )
2934 .unwrap();
2935 db
2936 }
2937
2938 fn rotate(db: &Connection, next: &str, shown: &str) -> bool {
2939 db.query_row(TOKEN_ROTATE, params![sha256_hex(next), "job_1", sha256_hex(shown)], |row| row.get::<_, String>(0))
2940 .optional()
2941 .unwrap()
2942 .is_some()
2943 }
2944
2945 /// The job's current token's hash, and the one it started with.
2946 fn hashes(db: &Connection) -> (Option<String>, Option<String>) {
2947 db.query_row("SELECT token_hash, first_token_hash FROM jobs WHERE id = 'job_1'", [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()
2948 }
2949
2950 #[test]
2951 fn a_job_token_fetches_its_spec_once() {
2952 let db = database();
2953 assert!(rotate(&db, "second", "first"));
2954 assert_eq!(hashes(&db), (Some(sha256_hex("second")), Some(sha256_hex("first"))));
2955 // The first token, shown again (a copy left on the computer, or a
2956 // second fetch racing the first), swaps nothing and is refused.
2957 assert!(!rotate(&db, "third", "first"));
2958 assert_eq!(hashes(&db).0, Some(sha256_hex("second")));
2959 // A job that ended has no token to swap.
2960 db.execute("UPDATE jobs SET status = 'completed', token_hash = NULL, first_token_hash = NULL", []).unwrap();
2961 assert!(!rotate(&db, "fourth", "second"));
2962 assert_eq!(hashes(&db), (None, None));
2963 }
2964
2965 #[test]
2966 fn the_worker_finds_brokered_secrets_with_the_token_the_job_started_with() {
2967 let db = database();
2968 let opens = |db: &Connection, token: &str, started: bool| {
2969 let (current, first) = hashes(db);
2970 token_opens(current.as_deref(), first.as_deref(), token, started)
2971 };
2972 // Before any swap, the job's one token opens everything.
2973 assert!(opens(&db, "first", false) && opens(&db, "first", true));
2974 assert!(rotate(&db, "second", "first"));
2975 // The harness's new token opens everything.
2976 assert!(opens(&db, "second", false) && opens(&db, "second", true));
2977 // The token it started with, which the runner Worker holds, still
2978 // finds the job's brokered secrets, and opens nothing else.
2979 assert!(opens(&db, "first", true));
2980 assert!(!opens(&db, "first", false));
2981 assert!(!opens(&db, "other", true));
2982 // Another swap keeps the token it started with, not the one between.
2983 assert!(rotate(&db, "third", "second"));
2984 assert!(opens(&db, "first", true));
2985 assert!(!opens(&db, "second", true));
2986 }
2987
2988 #[test]
2989 fn the_runtime_token_is_signed_for_the_token_that_replaces_the_first() {
2990 let now = 1_800_000_000;
2991 let token = crate::runtime::runtime_token("job_1", "run_1", &sha256_hex("second"), now, 600);
2992 assert!(crate::runtime::runtime_valid(&token, "job_1", &sha256_hex("second"), now));
2993 assert!(!crate::runtime::runtime_valid(&token, "job_1", &sha256_hex("first"), now));
2994 }
2995}
2996
2997#[cfg(test)]
Merge branch 'worktree-agent-a6a121745e81f639f'2998mod stopping {
2999 use super::stops_runs;
3000 use g1t_contracts::events::Event;
3001 use serde_json::{Value, json};
3002
3003 fn event(kind: &str, data: Value) -> Event {
3004 Event {
3005 id: "evt_1".into(),
3006 kind: kind.into(),
3007 source: "repos".into(),
3008 time: "2026-10-05T00:00:00Z".into(),
3009 repo_id: Some("rep_1".into()),
3010 actor: None,
3011 data,
3012 }
3013 }
3014
3015 #[test]
3016 fn deleting_or_archiving_stops_runs() {
3017 assert_eq!(stops_runs(&event("repo.deleted", json!({ "repoId": "rep_1" }))).as_deref(), Some("rep_1"));
3018 assert_eq!(stops_runs(&event("repo.archived", json!({ "archived": true }))).as_deref(), Some("rep_1"));
3019 assert_eq!(stops_runs(&event("repo.unarchived", json!({ "archived": false }))), None);
3020 assert_eq!(stops_runs(&event("repo.restored", json!({}))), None);
3021 assert_eq!(stops_runs(&event("git.push", json!({}))), None);
3022 }
3023}
3024
3025#[cfg(test)]
3026mod status_of_needs {
3027 use std::collections::HashMap;
3028
3029 use super::ancestor_failed;
3030
3031 /// check -> plan -> (migrate) -> core -> edge, as deploy.yml has them,
3032 /// and a job that needs only the last.
3033 fn graph() -> HashMap<&'static str, Vec<&'static str>> {
3034 HashMap::from([
3035 ("check", vec![]),
3036 ("plan", vec!["check"]),
3037 ("migrate", vec!["plan"]),
3038 ("core", vec!["plan", "migrate"]),
3039 ("edge", vec!["plan", "migrate", "core"]),
3040 ("notify", vec!["edge"]),
3041 ])
3042 }
3043
3044 #[test]
3045 fn a_failure_is_seen_however_far_back() {
3046 let needs = graph();
3047 let failed = |which: &'static str| move |key: &str| key == which;
3048 // check failed; plan, and everything after, was skipped for it.
3049 assert!(ancestor_failed(&needs, "notify", failed("check")));
3050 assert!(ancestor_failed(&needs, "core", failed("check")));
3051 assert!(ancestor_failed(&needs, "edge", failed("core")));
3052 // Nothing before a job failed: a skipped migrate is not a failure.
3053 assert!(!ancestor_failed(&needs, "edge", |_| false));
3054 assert!(!ancestor_failed(&needs, "core", failed("edge")));
3055 assert!(!ancestor_failed(&needs, "check", failed("check")));
3056 }
3057
3058 #[test]
3059 fn cycles_and_unknown_keys_end() {
3060 let needs = HashMap::from([("a", vec!["b"]), ("b", vec!["a"])]);
3061 assert!(!ancestor_failed(&needs, "a", |_| false));
3062 assert!(!ancestor_failed(&needs, "missing", |_| true));
3063 }
3064}
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb973065
3066#[cfg(test)]
3067mod deployments {
3068 use serde_json::{Map, Value, json};
3069
3070 use super::{JobEnvironment, deployment_outcome, environment_of};
3071
3072 #[test]
3073 fn a_jobs_environment_is_read_for_deployments() {
3074 let contexts: Map<String, Value> = serde_json::from_value(json!({
3075 "github": { "ref_name": "main", "repository": "acme/web" },
3076 "inputs": { "target": "staging" },
3077 "matrix": {},
3078 }))
3079 .unwrap();
3080 let read = |raw: Value| environment_of(&raw, &contexts);
3081 assert_eq!(read(json!({})), None);
3082 assert_eq!(
3083 read(json!({ "environment": "production" })),
3084 Some(JobEnvironment { name: "production".into(), url: None, deploys: true })
3085 );
3086 assert_eq!(
3087 read(json!({ "environment": { "name": "production", "url": "https://g1t.sh" } })),
3088 Some(JobEnvironment { name: "production".into(), url: Some("https://g1t.sh".into()), deploys: true })
3089 );
3090 // Expressions are filled in from the run.
3091 assert_eq!(
3092 read(json!({ "environment": { "name": "${{ inputs.target }}", "url": "https://${{ github.ref_name }}.example.com" } })),
3093 Some(JobEnvironment { name: "staging".into(), url: Some("https://main.example.com".into()), deploys: true })
3094 );
3095 // Secrets only: no deployment.
3096 assert!(!read(json!({ "environment": { "name": "production", "deployment": false } })).unwrap().deploys);
Merge branch 'worktree-agent-a641fa0e2e4fffccb' into claude/cool-newton-7nm8r93097 // The same, said in the job's g1t settings.
3098 assert!(!read(json!({ "environment": "production", "g1t": { "deployment": false } })).unwrap().deploys);
3099 assert!(!read(json!({ "environment": { "name": "production" }, "g1t": { "deployment": false } })).unwrap().deploys);
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb973100 // Only http(s) addresses.
3101 assert_eq!(read(json!({ "environment": { "name": "production", "url": "javascript:alert(1)" } })).unwrap().url, None);
3102 }
3103
3104 #[test]
3105 fn a_runs_outcome_for_an_environment() {
3106 let of = |list: &[&str]| deployment_outcome(&list.iter().map(|c| Some((*c).to_owned())).collect::<Vec<_>>());
3107 assert_eq!(of(&["success", "skipped"]), Some("success"));
3108 assert_eq!(of(&["success", "failure"]), Some("failure"));
3109 assert_eq!(of(&["success", "cancelled"]), Some("error"));
3110 assert_eq!(of(&["skipped"]), None);
3111 assert_eq!(deployment_outcome(&[None]), None);
3112 }
3113}
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)3114
3115#[cfg(test)]
3116mod reruns {
3117 use serde_json::{Map, Value, json};
3118
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r93119 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)3120
3121 /// build ← test ← deploy, and lint on its own.
3122 fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {
3123 let order = ["build", "lint", "test", "deploy"];
3124 let needs = |key: &str| -> Vec<String> {
3125 match key {
3126 "test" => vec!["build".into()],
3127 "deploy" => vec!["test".into()],
3128 _ => Vec::new(),
3129 }
3130 };
3131 rerun_keys(&order, needs, |key| !failed.contains(&key), which)
3132 }
3133
3134 #[test]
3135 fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {
3136 assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);
3137 assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);
3138 assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);
3139 assert!(keys(Rerun::Failed, &[]).is_empty());
3140 // One job, whatever it came to, and what depends on it.
3141 assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);
3142 assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);
3143 assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);
3144 assert!(keys(Rerun::Job("missing"), &[]).is_empty());
3145 }
3146
3147 #[test]
3148 fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {
3149 assert_eq!(top_key("build/test"), "build");
3150 assert_eq!(top_key("build/inner/test"), "build");
3151 assert_eq!(top_key("lint"), "lint");
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r93152 // A matrix's calls each have their own jobs, under `key[i]/`.
3153 assert_eq!(top_key("build[1]/test"), "build");
3154 assert_eq!(top_key("build/inner[0]/test"), "build");
3155 assert_eq!(call_prefix("build", Some(1)), "build[1]");
3156 assert_eq!(call_prefix("build/inner", None), "build/inner");
3157 }
3158
3159 #[test]
3160 fn a_run_connects_ten_levels_of_workflows_and_fifty_reusable_ones() {
3161 assert!(call_allowed(1, &[], "a.yml").is_ok());
3162 assert!(call_allowed(9, &[], "a.yml").is_ok());
3163 let deep = call_allowed(10, &[], "a.yml").unwrap_err();
3164 assert!(deep.contains("at most ten") && deep.contains("nine levels"), "{deep}");
3165 let names: Vec<String> = (0..50).map(|n| format!("w{n}.yml")).collect();
3166 let fifty: Vec<&str> = names.iter().map(String::as_str).collect();
3167 // A workflow called again, as by each combination of a matrix, is one.
3168 assert!(call_allowed(1, &fifty, "w3.yml").is_ok());
3169 assert!(call_allowed(1, &fifty, "another.yml").unwrap_err().contains("at most 50 different"));
3170 }
3171
3172 #[test]
3173 fn a_matrixs_calls_are_named_by_the_caller_and_its_values() {
3174 let workflow = super::workflow::parse(
3175 "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' }]",
3176 )
3177 .unwrap();
3178 let job = |id: &str| workflow.jobs.iter().find(|job| job.id == id).unwrap();
3179 let combination: Map<String, Value> = serde_json::from_value(json!({ "os": "linux", "node": 20 })).unwrap();
3180 let mut contexts = Map::new();
3181 contexts.insert("matrix".into(), Value::Object(combination.clone()));
3182 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
3183 assert_eq!(shown_name(job("build"), None, &combination, &scope), "Build (linux, 20)");
3184 assert_eq!(shown_name(job("plain"), None, &Map::new(), &scope), "plain");
3185 // A called workflow's job, under the call that made it.
3186 assert_eq!(shown_name(job("plain"), Some("Build (linux, 20)"), &Map::new(), &scope), "Build (linux, 20) / plain");
3187 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)3188 }
3189
3190 #[test]
3191 fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {
3192 assert!(summary_fits(0, None, 10));
3193 assert!(!summary_fits(0, None, 0), "nothing to keep");
3194 assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");
3195 assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");
3196 assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));
3197 assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));
3198 assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));
3199 }
3200
3201 #[test]
3202 fn a_debug_rerun_sets_what_github_sets() {
3203 let mut variables = Map::new();
3204 let mut runner = json!({ "name": "g1t", "debug": "" });
3205 debug_logging(&mut variables, &mut runner);
3206 assert_eq!(variables["RUNNER_DEBUG"], "1");
3207 assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");
3208 assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");
3209 assert_eq!(runner["debug"], Value::String("1".into()));
3210 assert_eq!(runner["name"], "g1t");
3211 }
3212}

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