Skip to content
Sign up

flagon-io/g1t

public
0
1,945 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.

Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1//! Self-hosted runners: registering them, handing them work, hearing back,
2//! and the people's side (listing, removing, groups, settings). See
3//! `g1t_contracts::runners` for the model and the wire.
4//!
5//! Work reaches a runner only when it asks for it (`runner_poll`), so a
6//! runner never needs an open port. A workflow job is taken by flipping its
7//! row from `queued` to `in_progress` in one statement, exactly as g1t's
8//! own sandboxes take one, and the harness on the machine then fetches it
9//! and reports with the job's own token through the same API. Agent work
10//! from the runner service is kept as a task with its environment sealed,
11//! and handed over once.
12
13use g1t_contracts::access::Capability;
14use g1t_contracts::audit::{AuditActor, AuditOutcome, AuditTarget, NewAuditEntry, RecordAuditArgs, Surface};
15use g1t_contracts::billing::{ComputeKind, RecordSandboxArgs};
16use g1t_contracts::repos::{Repo, RepoPath};
17use g1t_contracts::runners::{
Merge the desktop app's sandbox: its default runner now starts g1t-runner's sandbox on Windows, macOS and Linux, and says per kind of job (Linux, Windows) what it is doing or why it cannot and how to fix it.18 self as model, ACTIVITY_LIMIT, CloudJob, DesktopCapability, DesktopDiagnosis, DesktopRunner, DesktopRunnersArgs, ReportDesktopArgs, HandedOverTask, RunnerActivity, RunnerActivityArgs, StuckJob, StuckJobsArgs, Assignment, CancelTaskArgs, CreateRegistrationTokenArgs, DeleteRunnerGroupArgs, EnqueueTaskArgs, FinishedArgs,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays19 ListRunnersArgs, Poll, PollArgs, RegisterArgs, Registered, RegistrationToken, RemoveRunnerArgs, RemoveSelfArgs, RouteArgs, Runner,
20 RunnerAuth, RunnerGroup, RunnerGroupsArgs, RunnerSettings, RunnerSettingsArgs, RunnerWork, RunnersOwner, SetRunnerGroupArgs,
21 SetRunnerSettingsArgs, Wanted,
22};
23use g1t_contracts::time::rfc3339;
24use g1t_contracts::{FailureCode, Outcome, PrincipalKind, Role, User, new_id};
25use g1t_kit::now_ms;
26use g1t_secrets::{random_hex, sha256_hex};
27use serde::Deserialize;
28use serde_json::{Map, Value, json};
29use worker::Result;
30use worker::wasm_bindgen::JsValue;
31
32use crate::plan::JobRow;
33use crate::{Actions, Count, SILENT_MS, check, fail, optional};
34
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.35/// What one poll has learned while it looks for work, so its looks again
36/// do not ask again: each repository's visibility, and the jobs it already
37/// said wait for a sandboxed runner.
38#[derive(Default)]
39struct Looked {
40 visibilities: std::collections::HashMap<String, Option<String>>,
41 noted: std::collections::HashSet<String>,
42}
43
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays44/// How often a poll looks again for work while it waits.
45const POLL_EVERY_MS: u64 = 2_500;
46/// A runner that has been offline this long is removed, as on GitHub.
47const FORGET_OFFLINE_MS: u64 = 14 * 24 * 60 * 60 * 1000;
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails48/// How often a poll writes down that a runner, or the work it holds, is
49/// still there. Polls come every twenty seconds or so; a write each time
50/// would be most of the service's writes for nothing. Well inside
51/// [`model::ONLINE_WITHIN_MS`] and [`crate::SILENT_MS`].
52const SEEN_RESOLUTION_MS: u64 = 60 * 1000;
53
54/// Whether `seen` is old enough, at `at`, to be written down again.
55fn seen_is_stale(seen: Option<&str>, at: u64) -> bool {
56 seen.is_none_or(|seen| seen < rfc3339(at.saturating_sub(SEEN_RESOLUTION_MS)).as_str())
57}
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays58
59#[derive(Clone, Debug, Deserialize)]
60pub struct RunnerRow {
61 pub id: String,
62 pub workspace: String,
63 pub repo_id: Option<String>,
64 pub repo: Option<String>,
65 pub group_id: Option<String>,
66 pub name: String,
67 pub labels: String,
68 pub os: String,
69 pub arch: String,
70 pub version: String,
71 pub ephemeral: u32,
72 pub credential_hash: String,
73 pub previous_hash: Option<String>,
74 pub rotated_at: String,
75 pub work_id: Option<String>,
76 pub work_kind: Option<String>,
77 pub spent: u32,
78 pub last_seen_at: Option<String>,
79 pub created_at: String,
80 pub created_by: Option<String>,
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.81 /// `nested-container`, `container` or `host` (migration 0015); none
82 /// for a runner that never said.
83 #[serde(default)]
84 pub isolation: Option<String>,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays85}
86
87impl RunnerRow {
88 fn labels(&self) -> Vec<String> {
89 serde_json::from_str(&self.labels).unwrap_or_default()
90 }
91
92 fn online(&self, now: u64) -> bool {
93 self.last_seen_at.as_deref().is_some_and(|seen| seen >= rfc3339(now.saturating_sub(model::ONLINE_WITHIN_MS)).as_str())
94 }
95}
96
97#[derive(Clone, Debug, Deserialize)]
98struct GroupRow {
99 id: String,
100 name: String,
101 is_default: u32,
102 repositories: String,
103 updated_at: String,
104}
105
106impl GroupRow {
107 fn repositories(&self) -> Vec<String> {
108 serde_json::from_str(&self.repositories).unwrap_or_default()
109 }
110
111 /// Whether `repo` (`owner/name`, or a bare name) may use the group.
112 fn allows(&self, repo: &str) -> bool {
113 let name = repo.rsplit('/').next().unwrap_or(repo);
114 let listed = self.repositories();
115 listed.is_empty() || listed.iter().any(|r| r.eq_ignore_ascii_case(name))
116 }
117}
118
119#[derive(Clone, Debug, Deserialize)]
120struct RegistrationRow {
121 workspace: String,
122 repo_id: Option<String>,
123 repo: Option<String>,
124 group_id: Option<String>,
125 expires_at: String,
Merge the desktop app's runner: it starts with the app for the signed-in person, sandboxed by default and running nothing where no sandbox is available yet, registers through the site, keeps itself up to date, and says everywhere what it is doing, why it is not, and how to fix it.126 /// Who made the token: the runners it registers are theirs.
127 #[serde(default)]
128 created_by: Option<String>,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays129}
130
131#[derive(Clone, Debug, Deserialize)]
132struct SettingsRow {
133 agents: u32,
134 agent_labels: String,
135 fork_pulls: u32,
136}
137
138#[derive(Clone, Debug, Deserialize)]
139pub struct TaskRow {
140 pub id: String,
141 pub sandbox: String,
142 pub repo_id: Option<String>,
143 pub repo: String,
144 pub kind: String,
145 pub title: String,
146 pub labels: String,
147 pub env: Option<String>,
148 pub timeout_minutes: u32,
149 pub status: String,
150 pub runner_id: Option<String>,
151 pub runner_name: Option<String>,
152 pub started_at: Option<String>,
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails153 #[serde(default)]
154 pub seen_at: Option<String>,
The Runners page shows the machines a workspace's agents and workflow jobs run on, in Workspace under Compute: g1t's cloud beside your own runners, what each is running now with a link to it, what's waiting, this month's machine time and its cost (your own runners' free), where agent work and workflow jobs run, adding a runner, and groups; runner_activity in Actions says what runs where, work handed to your runners keeps its agent run, and the self-hosted runners guide says how.155 /// The agent run it belongs to (migration 0012); none for older tasks.
156 #[serde(default)]
157 pub run_id: Option<String>,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays158}
159
160/// Who may do what with a set of runners: the workspace, and the
161/// repository when they are a repository's own.
162struct Place {
163 workspace: String,
164 repo: Option<Repo>,
165}
166
167fn now() -> String {
168 rfc3339(now_ms())
169}
170
171fn valid_name(name: &str) -> std::result::Result<String, String> {
172 let name = name.trim();
173 let ok = !name.is_empty()
174 && name.len() <= 64
175 && name.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'));
176 if ok {
177 Ok(name.to_owned())
178 } else {
179 Err("A runner's name is 1 to 64 letters, digits, `-`, `_` and `.`.".to_owned())
180 }
181}
182
183fn valid_group_name(name: &str) -> std::result::Result<String, String> {
184 let name = name.trim();
185 if name.is_empty() || name.len() > 64 || name.chars().any(|c| c.is_control()) {
186 return Err("A group's name is 1 to 64 characters.".to_owned());
187 }
188 Ok(name.to_owned())
189}
190
191/// What a runner's credential is checked against, and what a registration
192/// token is found by.
193fn hash(secret: &str) -> String {
194 sha256_hex(secret)
195}
196
197/// The actor of a runner's own entries in the audit log.
198fn runner_actor(runner: &RunnerRow) -> AuditActor {
199 AuditActor {
200 actor_kind: Some(g1t_contracts::audit::ActorKind::Runner),
201 actor: runner.name.clone(),
202 actor_id: runner.id.clone(),
203 ..AuditActor::default()
204 }
205}
206
207fn entry(actor: AuditActor, action: &str, workspace: &str, repo: Option<String>, path: Option<String>, message: String) -> NewAuditEntry {
208 NewAuditEntry {
209 actor,
210 action: action.to_owned(),
211 surface: Surface::Rest,
212 target: AuditTarget {
213 workspace: workspace.to_owned(),
214 repo,
215 path,
216 ..AuditTarget::default()
217 },
218 outcome: AuditOutcome::Allowed,
219 rule: "runner".to_owned(),
220 result: Some("ok".to_owned()),
221 message: Some(message),
222 request_id: new_id("req", now_ms()),
223 }
224}
225
226impl Actions {
227 // --- Shared ------------------------------------------------------------
228
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.229 pub(crate) async fn audit(&self, entries: Vec<NewAuditEntry>) {
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays230 let recorded: Result<u32> = g1t_kit::call(&self.events, "audit_record", &RecordAuditArgs { entries }).await;
231 if let Err(error) = recorded {
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.232 worker::console_error!("actions: audit entries not recorded: {error}");
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays233 }
234 }
235
236 /// Where `owner` points, if `actor` may see it (`manage` false) or
237 /// change it (`manage` true): a workspace's runners are seen by its
238 /// members and changed by its owners; a repository's own, by its
239 /// admins. Agents and workspace tokens never change them, so a
240 /// workflow's `G1T_TOKEN` cannot add a machine to run its own jobs.
241 async fn runner_place(&self, actor: &User, owner: &RunnersOwner, manage: bool) -> Result<Outcome<Place>> {
242 if actor.kind == PrincipalKind::Agent {
243 return Ok(fail(FailureCode::Forbidden, "Agents cannot see or change self-hosted runners."));
244 }
245 if manage && actor.kind == PrincipalKind::Workspace {
246 return Ok(fail(
247 FailureCode::Forbidden,
248 "A workspace's tokens, G1T_TOKEN included, cannot change self-hosted runners. Use a person's token with the runners:admin scope.",
249 ));
250 }
251 match (&owner.repo, &owner.workspace) {
252 (Some(path), _) => {
253 let repo = check!(self.may(actor, path, Capability::ManageIntegrations).await?);
254 Ok(Outcome::Ok(Place { workspace: repo.namespace.to_lowercase(), repo: Some(repo) }))
255 }
256 (None, Some(slug)) => {
257 let slug = slug.to_lowercase();
258 let role = actor.workspaces.iter().find(|m| m.slug.eq_ignore_ascii_case(&slug)).map(|m| m.role);
259 match role {
260 None => Ok(fail(FailureCode::NotFound, "There is no such workspace, or you are not a member of it.")),
A workspace's runners are its owners' to see, through the API too; docs for the new sidebar and Settings261 // A workspace's machines and their tokens are its owners' alone, to see as well as change.
262 Some(Role::Member) => Ok(fail(
263 FailureCode::Forbidden,
264 format!("Only owners of {slug} can {} its self-hosted runners.", if manage { "change" } else { "see" }),
265 )),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays266 Some(_) => Ok(Outcome::Ok(Place { workspace: slug, repo: None })),
267 }
268 }
269 (None, None) => Ok(fail(FailureCode::Invalid, "Give `repo` or `workspace`.")),
270 }
271 }
272
273 async fn group_rows(&self, workspace: &str) -> Result<Vec<GroupRow>> {
274 self.db
275 .prepare("SELECT * FROM runner_groups WHERE workspace = ? ORDER BY is_default DESC, name")
276 .bind(&[workspace.into()])?
277 .all()
278 .await?
279 .results::<GroupRow>()
280 }
281
282 /// The workspace's default group, made the first time it is needed.
283 async fn default_group(&self, workspace: &str) -> Result<GroupRow> {
284 let at = now();
285 self.db
286 .prepare(
287 "INSERT OR IGNORE INTO runner_groups (id, workspace, name, is_default, repositories, created_at, updated_at)
288 VALUES (?, ?, 'Default', 1, '[]', ?, ?)",
289 )
290 .bind(&[new_id("rng", now_ms()).into(), workspace.into(), at.as_str().into(), at.as_str().into()])?
291 .run()
292 .await?;
293 let row = self
294 .db
295 .prepare("SELECT * FROM runner_groups WHERE workspace = ? AND is_default = 1 LIMIT 1")
296 .bind(&[workspace.into()])?
297 .first::<GroupRow>(None)
298 .await?;
299 row.ok_or_else(|| worker::Error::RustError("the default runner group could not be made".into()))
300 }
301
302 /// A group of the workspace by id or name.
303 async fn find_group(&self, workspace: &str, wanted: &str) -> Result<Option<GroupRow>> {
304 self.default_group(workspace).await?;
305 Ok(self
306 .group_rows(workspace)
307 .await?
308 .into_iter()
309 .find(|g| g.id == wanted || g.name.eq_ignore_ascii_case(wanted.trim())))
310 }
311
312 async fn runner_rows(&self, workspace: &str) -> Result<Vec<RunnerRow>> {
313 self.db
314 .prepare("SELECT * FROM runners WHERE workspace = ? ORDER BY name")
315 .bind(&[workspace.into()])?
316 .all()
317 .await?
318 .results::<RunnerRow>()
319 }
320
321 async fn runner_by_id(&self, id: &str) -> Result<Option<RunnerRow>> {
322 self.db.prepare("SELECT * FROM runners WHERE id = ?").bind(&[id.into()])?.first::<RunnerRow>(None).await
323 }
324
325 /// What a runner is doing, for people.
326 async fn work_of(&self, row: &RunnerRow) -> Result<Option<RunnerWork>> {
327 let Some(id) = &row.work_id else { return Ok(None) };
328 if row.work_kind.as_deref() == Some("agent") {
329 let task = self.db.prepare("SELECT * FROM runner_tasks WHERE id = ?").bind(&[id.as_str().into()])?.first::<TaskRow>(None).await?;
330 return Ok(task.filter(|t| t.status == "in_progress").map(|t| RunnerWork {
331 kind: "agent".into(),
332 id: t.id,
333 name: t.title,
334 repo: Some(t.repo),
The Runners page shows the machines a workspace's agents and workflow jobs run on, in Workspace under Compute: g1t's cloud beside your own runners, what each is running now with a link to it, what's waiting, this month's machine time and its cost (your own runners' free), where agent work and workflow jobs run, adding a runner, and groups; runner_activity in Actions says what runs where, work handed to your runners keeps its agent run, and the self-hosted runners guide says how.335 run_id: t.run_id,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays336 started_at: t.started_at,
337 }));
338 }
339 #[derive(Deserialize)]
340 struct Found {
341 id: String,
342 name: String,
343 run_id: String,
344 status: String,
345 started_at: Option<String>,
346 repo: String,
347 }
348 let job = self
349 .db
350 .prepare("SELECT jobs.id, jobs.name, jobs.run_id, jobs.status, jobs.started_at, runs.repo FROM jobs JOIN runs ON runs.id = jobs.run_id WHERE jobs.id = ?")
351 .bind(&[id.as_str().into()])?
352 .first::<Found>(None)
353 .await?;
354 Ok(job.filter(|j| j.status == "in_progress").map(|j| RunnerWork {
355 kind: "workflow".into(),
356 id: j.id,
357 name: j.name,
358 repo: Some(j.repo),
359 run_id: Some(j.run_id),
360 started_at: j.started_at,
361 }))
362 }
363
364 async fn runner_view(&self, row: &RunnerRow, groups: &[GroupRow], at: u64) -> Result<Runner> {
365 let online = row.online(at);
366 let work = if online { self.work_of(row).await? } else { None };
367 Ok(Runner {
368 id: row.id.clone(),
369 name: row.name.clone(),
370 workspace: row.workspace.clone(),
371 repo: row.repo.clone(),
372 group: row
373 .group_id
374 .as_ref()
375 .and_then(|id| groups.iter().find(|g| &g.id == id))
376 .map(|g| g.name.clone()),
377 labels: row.labels(),
378 os: row.os.clone(),
379 arch: row.arch.clone(),
380 version: row.version.clone(),
381 ephemeral: row.ephemeral != 0,
382 status: if !online {
383 "offline"
384 } else if work.is_some() {
385 "busy"
386 } else {
387 "online"
388 }
389 .to_owned(),
390 work,
391 last_seen_at: row.last_seen_at.clone(),
392 created_at: row.created_at.clone(),
393 created_by: row.created_by.clone(),
Merge the runner sandbox: g1t-runner sandbox up makes a person's own computer a self-hosted runner that nothing a job runs can get out of. Linux jobs run in a Linux VM of the runner's own (a WSL 2 distribution with no host mounts, interop or Docker socket on Windows; vfkit on macOS; Sysbox, or QEMU and KVM, on Linux), each in a fresh container under Sysbox inside it; on Windows 11 Pro and up, Windows jobs each run in a Windows Sandbox discarded after the job. Runners report their isolation, and the Runners page shows it.394 isolation: row.isolation.clone(),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays395 })
396 }
397
398 // --- People's side -----------------------------------------------------
399
400 /// `runners`: a workspace's runners, or a repository's: its own and
401 /// the workspace's that its group lets it use.
402 pub async fn runners(&self, a: ListRunnersArgs) -> Result<Outcome<Vec<Runner>>> {
403 let place = check!(self.runner_place(&a.actor, &a.owner, false).await?);
404 self.default_group(&place.workspace).await?;
405 let groups = self.group_rows(&place.workspace).await?;
406 let at = now_ms();
407 let mut out = Vec::new();
408 for row in self.runner_rows(&place.workspace).await? {
409 let shown = match &place.repo {
410 None => row.repo_id.is_none(),
411 Some(repo) => match &row.repo_id {
412 Some(id) => id == &repo.id,
413 None => groups.iter().find(|g| Some(&g.id) == row.group_id.as_ref()).is_none_or(|g| g.allows(&repo.name)),
414 },
415 };
416 if shown {
417 out.push(self.runner_view(&row, &groups, at).await?);
418 }
419 }
420 Ok(Outcome::Ok(out))
421 }
422
423 /// `create_registration_token`.
424 pub async fn create_registration_token(&self, a: CreateRegistrationTokenArgs) -> Result<Outcome<RegistrationToken>> {
425 let place = check!(self.runner_place(&a.actor, &a.owner, true).await?);
426 let made = self
427 .db
428 .prepare("SELECT COUNT(*) AS n FROM runner_registrations WHERE workspace = ? AND created_at > ?")
429 .bind(&[place.workspace.as_str().into(), rfc3339(now_ms().saturating_sub(60 * 60 * 1000)).into()])?
430 .first::<Count>(None)
431 .await?
432 .map_or(0, |c| c.n);
433 if made >= model::MAX_TOKENS_PER_HOUR {
434 return Ok(fail(FailureCode::Conflict, "Too many registration tokens in the last hour. Each lasts an hour and registers any number of runners."));
435 }
436 let group = match (&place.repo, a.group.as_deref().map(str::trim).filter(|g| !g.is_empty())) {
437 (Some(_), Some(_)) => return Ok(fail(FailureCode::Invalid, "A repository's own runners are in no group.")),
438 (Some(_), None) => None,
439 (None, Some(wanted)) => match self.find_group(&place.workspace, wanted).await? {
440 Some(group) => Some(group),
441 None => return Ok(fail(FailureCode::NotFound, format!("There is no runner group called {wanted}."))),
442 },
443 (None, None) => Some(self.default_group(&place.workspace).await?),
444 };
445 let token = format!("{}{}", model::REGISTRATION_PREFIX, random_hex(24));
446 let at = now_ms();
447 let expires_at = rfc3339(at + model::REGISTRATION_TTL_SECONDS * 1000);
448 let repo_name = place.repo.as_ref().map(|r| format!("{}/{}", r.namespace, r.name));
449 self.db
450 .prepare(
451 "INSERT INTO runner_registrations (token_hash, workspace, repo_id, repo, group_id, created_by, created_at, expires_at)
452 VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
453 )
454 .bind(&[
455 hash(&token).into(),
456 place.workspace.as_str().into(),
457 optional(place.repo.as_ref().map(|r| r.id.as_str())),
458 optional(repo_name.as_deref()),
459 optional(group.as_ref().map(|g| g.id.as_str())),
460 a.actor.username.as_str().into(),
461 rfc3339(at).into(),
462 expires_at.as_str().into(),
463 ])?
464 .run()
465 .await?;
466 self.audit(vec![entry(
467 AuditActor::of(&a.actor),
468 "create_runner_registration_token",
469 &place.workspace,
470 repo_name.clone(),
471 None,
472 format!("Made a runner registration token{}.", group.as_ref().map(|g| format!(" for group {}", g.name)).unwrap_or_default()),
473 )])
474 .await;
475 Ok(Outcome::Ok(RegistrationToken {
476 token,
477 expires_at,
478 workspace: place.workspace,
479 repo: repo_name,
480 group: group.map(|g| g.name),
481 url: crate::SITE.to_owned(),
482 }))
483 }
484
485 /// `remove_runner`: a person removing one.
486 pub async fn remove_runner(&self, a: RemoveRunnerArgs) -> Result<Outcome<bool>> {
487 let place = check!(self.runner_place(&a.actor, &a.owner, true).await?);
488 let Some(row) = self.runner_by_id(&a.id).await? else {
489 return Ok(fail(FailureCode::NotFound, "There is no such runner."));
490 };
491 let ours = row.workspace == place.workspace
492 && match &place.repo {
493 Some(repo) => row.repo_id.as_deref() == Some(repo.id.as_str()),
494 None => row.repo_id.is_none(),
495 };
496 if !ours {
497 return Ok(fail(FailureCode::NotFound, "There is no such runner."));
498 }
499 self.forget_runner(&row, &format!("{} removed the runner {}.", a.actor.username, row.name)).await?;
500 self.audit(vec![entry(
501 AuditActor::of(&a.actor),
502 "remove_runner",
503 &row.workspace,
504 row.repo.clone(),
505 Some(row.name.clone()),
506 format!("Removed the self-hosted runner {}.", row.name),
507 )])
508 .await;
509 Ok(Outcome::Ok(true))
510 }
511
512 /// Removes a runner: what it was running fails, and its credential
513 /// stops working at once.
514 async fn forget_runner(&self, row: &RunnerRow, why: &str) -> Result<()> {
515 self.db.prepare("DELETE FROM runners WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
516 self.abandon_work(&row.id, why).await
517 }
518
519 /// Fails whatever a runner was running.
520 async fn abandon_work(&self, runner_id: &str, why: &str) -> Result<()> {
521 let jobs = self
522 .db
523 .prepare("SELECT id FROM jobs WHERE runner_id = ? AND status = 'in_progress'")
524 .bind(&[runner_id.into()])?
525 .all()
526 .await?
527 .results::<IdRow>()?;
528 for job in jobs {
529 self.finish_job(&job.id, "failure", Some(why), None).await?;
530 }
531 let tasks = self
532 .db
533 .prepare("SELECT * FROM runner_tasks WHERE runner_id = ? AND status = 'in_progress'")
534 .bind(&[runner_id.into()])?
535 .all()
536 .await?
537 .results::<TaskRow>()?;
538 for task in tasks {
539 self.end_task(&task, 1, why).await?;
540 }
541 Ok(())
542 }
543
544 /// `runner_groups`.
545 pub async fn runner_groups(&self, a: RunnerGroupsArgs) -> Result<Outcome<Vec<RunnerGroup>>> {
546 let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) };
547 let place = check!(self.runner_place(&a.actor, &owner, false).await?);
548 self.default_group(&place.workspace).await?;
549 let groups = self.group_rows(&place.workspace).await?;
550 let runners = self.runner_rows(&place.workspace).await?;
551 Ok(Outcome::Ok(groups.iter().map(|g| group_view(g, &runners)).collect()))
552 }
553
554 /// `set_runner_group`.
555 pub async fn set_runner_group(&self, a: SetRunnerGroupArgs) -> Result<Outcome<RunnerGroup>> {
556 let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) };
557 let place = check!(self.runner_place(&a.actor, &owner, true).await?);
558 self.default_group(&place.workspace).await?;
559 let groups = self.group_rows(&place.workspace).await?;
560 let name = match a.name.as_deref().map(valid_group_name).transpose() {
561 Ok(name) => name,
562 Err(problem) => return Ok(fail(FailureCode::Invalid, problem)),
563 };
564 let repositories: Option<Vec<String>> = a.repositories.map(|list| {
565 let mut out: Vec<String> = Vec::new();
566 for repo in list {
567 let name = repo.trim().rsplit('/').next().unwrap_or_default().to_owned();
568 if !name.is_empty() && !out.iter().any(|r: &String| r.eq_ignore_ascii_case(&name)) {
569 out.push(name);
570 }
571 }
572 out
573 });
574 if let Some(name) = &name
575 && groups.iter().any(|g| g.name.eq_ignore_ascii_case(name) && Some(&g.id) != a.id.as_ref())
576 {
577 return Ok(fail(FailureCode::Conflict, format!("There is already a group called {name}.")));
578 }
579 let at = now();
580 let id = match &a.id {
581 Some(id) => {
582 let Some(group) = groups.iter().find(|g| &g.id == id) else {
583 return Ok(fail(FailureCode::NotFound, "There is no such runner group."));
584 };
585 self.db
586 .prepare("UPDATE runner_groups SET name = ?, repositories = ?, updated_at = ? WHERE id = ?")
587 .bind(&[
588 name.clone().unwrap_or(group.name.clone()).into(),
589 serde_json::to_string(&repositories.clone().unwrap_or(group.repositories()))?.into(),
590 at.as_str().into(),
591 id.as_str().into(),
592 ])?
593 .run()
594 .await?;
595 id.clone()
596 }
597 None => {
598 let Some(name) = name.clone() else {
599 return Ok(fail(FailureCode::Invalid, "A new group needs a name."));
600 };
601 if groups.len() >= 50 {
602 return Ok(fail(FailureCode::Conflict, "A workspace has at most 50 runner groups."));
603 }
604 let id = new_id("rng", now_ms());
605 self.db
606 .prepare(
607 "INSERT INTO runner_groups (id, workspace, name, is_default, repositories, created_at, updated_at) VALUES (?, ?, ?, 0, ?, ?, ?)",
608 )
609 .bind(&[
610 id.as_str().into(),
611 place.workspace.as_str().into(),
612 name.into(),
613 serde_json::to_string(&repositories.clone().unwrap_or_default())?.into(),
614 at.as_str().into(),
615 at.as_str().into(),
616 ])?
617 .run()
618 .await?;
619 id
620 }
621 };
622 let groups = self.group_rows(&place.workspace).await?;
623 let runners = self.runner_rows(&place.workspace).await?;
624 let Some(group) = groups.iter().find(|g| g.id == id) else {
625 return Ok(fail(FailureCode::NotFound, "There is no such runner group."));
626 };
627 self.audit(vec![entry(
628 AuditActor::of(&a.actor),
629 if a.id.is_some() { "update_runner_group" } else { "create_runner_group" },
630 &place.workspace,
631 None,
632 Some(group.name.clone()),
633 format!(
634 "Runner group {}: {}.",
635 group.name,
636 if group.repositories().is_empty() { "every repository".to_owned() } else { group.repositories().join(", ") }
637 ),
638 )])
639 .await;
640 Ok(Outcome::Ok(group_view(group, &runners)))
641 }
642
643 /// `delete_runner_group`.
644 pub async fn delete_runner_group(&self, a: DeleteRunnerGroupArgs) -> Result<Outcome<bool>> {
645 let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) };
646 let place = check!(self.runner_place(&a.actor, &owner, true).await?);
647 let default = self.default_group(&place.workspace).await?;
648 let Some(group) = self.group_rows(&place.workspace).await?.into_iter().find(|g| g.id == a.id) else {
649 return Ok(fail(FailureCode::NotFound, "There is no such runner group."));
650 };
651 if group.is_default != 0 {
652 return Ok(fail(FailureCode::Conflict, "The default group cannot be deleted."));
653 }
654 self.db
655 .batch(vec![
656 self.db.prepare("UPDATE runners SET group_id = ? WHERE group_id = ?").bind(&[default.id.as_str().into(), group.id.as_str().into()])?,
657 self.db.prepare("DELETE FROM runner_groups WHERE id = ?").bind(&[group.id.as_str().into()])?,
658 ])
659 .await?;
660 self.audit(vec![entry(
661 AuditActor::of(&a.actor),
662 "delete_runner_group",
663 &place.workspace,
664 None,
665 Some(group.name.clone()),
666 format!("Deleted the runner group {}; its runners joined {}.", group.name, default.name),
667 )])
668 .await;
669 Ok(Outcome::Ok(true))
670 }
671
672 async fn settings_row(&self, owner: &str) -> Result<Option<SettingsRow>> {
673 self.db.prepare("SELECT * FROM runner_settings WHERE owner = ?").bind(&[owner.into()])?.first::<SettingsRow>(None).await
674 }
675
676 fn settings_of(row: Option<SettingsRow>, inherited: bool) -> RunnerSettings {
677 match row {
678 Some(row) => RunnerSettings {
679 agents_on_self_hosted: row.agents != 0,
680 agent_labels: serde_json::from_str(&row.agent_labels).unwrap_or_else(|_| vec![model::SELF_HOSTED.to_owned()]),
681 fork_pull_requests: row.fork_pulls != 0,
682 inherited,
683 },
684 None => RunnerSettings {
685 agent_labels: vec![model::SELF_HOSTED.to_owned()],
686 inherited,
687 ..RunnerSettings::default()
688 },
689 }
690 }
691
692 /// The settings that apply in a workspace, or in one of its
693 /// repositories: the repository's own if it has them.
694 pub async fn effective_runner_settings(&self, workspace: &str, repo_id: Option<&str>) -> Result<RunnerSettings> {
695 if let Some(repo_id) = repo_id
696 && let Some(own) = self.settings_row(repo_id).await?
697 {
698 return Ok(Self::settings_of(Some(own), false));
699 }
700 let inherited = repo_id.is_some();
701 Ok(Self::settings_of(self.settings_row(&workspace.to_lowercase()).await?, inherited))
702 }
703
704 /// `runner_settings`.
705 pub async fn runner_settings(&self, a: RunnerSettingsArgs) -> Result<Outcome<RunnerSettings>> {
706 let place = check!(self.runner_place(&a.actor, &a.owner, false).await?);
707 Ok(Outcome::Ok(self.effective_runner_settings(&place.workspace, place.repo.as_ref().map(|r| r.id.as_str())).await?))
708 }
709
Merge the desktop app's runner: it starts with the app for the signed-in person, sandboxed by default and running nothing where no sandbox is available yet, registers through the site, keeps itself up to date, and says everywhere what it is doing, why it is not, and how to fix it.710 /// `report_desktop`: what a person's desktop app says of its runner,
711 /// in a workspace they own. Kept per computer, replaced each time.
712 pub async fn report_desktop(&self, a: ReportDesktopArgs) -> Result<Outcome<bool>> {
713 let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) };
714 let place = check!(self.runner_place(&a.actor, &owner, true).await?);
715 let machine = match valid_name(&a.machine) {
716 Ok(name) => name,
717 Err(problem) => return Ok(fail(FailureCode::Invalid, problem)),
718 };
719 if !matches!(a.backend.as_str(), "sandboxed" | "docker" | "host") {
720 return Ok(fail(FailureCode::Invalid, "A desktop runs jobs sandboxed, in Docker or on the host."));
721 }
722 let clip = |text: &str, most: usize| text.chars().take(most).collect::<String>();
723 let tidy = |d: &DesktopDiagnosis| DesktopDiagnosis {
724 code: clip(&d.code, 40),
725 tone: clip(&d.tone, 20),
726 title: clip(&d.title, 200),
727 cause: d.cause.as_deref().map(|t| clip(t, 1000)),
728 fix: d.fix.as_deref().map(|t| clip(t, 1000)),
Merge the desktop app's sandbox: its default runner now starts g1t-runner's sandbox on Windows, macOS and Linux, and says per kind of job (Linux, Windows) what it is doing or why it cannot and how to fix it.729 capabilities: d
730 .capabilities
731 .iter()
732 .take(model::MAX_DESKTOP_CAPABILITIES)
733 .map(|c| DesktopCapability {
734 id: clip(&c.id, 20),
735 label: clip(&c.label, 60),
736 tone: clip(&c.tone, 20),
737 title: clip(&c.title, 200),
738 cause: c.cause.as_deref().map(|t| clip(t, 1000)),
739 fix: c.fix.as_deref().map(|t| clip(t, 1000)),
740 })
741 .collect(),
Merge the desktop app's runner: it starts with the app for the signed-in person, sandboxed by default and running nothing where no sandbox is available yet, registers through the site, keeps itself up to date, and says everywhere what it is doing, why it is not, and how to fix it.742 };
743 let at = now_ms();
744 let known = self
745 .db
746 .prepare("SELECT COUNT(*) AS n FROM desktop_runners WHERE workspace = ? AND lower(machine) <> lower(?)")
747 .bind(&[place.workspace.as_str().into(), machine.as_str().into()])?
748 .first::<Count>(None)
749 .await?
750 .map_or(0, |c| c.n);
751 if known >= model::MAX_DESKTOPS {
752 // The oldest report makes room: a computer long gone.
753 self.db
754 .prepare("DELETE FROM desktop_runners WHERE rowid IN (SELECT rowid FROM desktop_runners WHERE workspace = ? ORDER BY reported_at LIMIT 1)")
755 .bind(&[place.workspace.as_str().into()])?
756 .run()
757 .await?;
758 }
759 self.db
760 .prepare(
761 "INSERT INTO desktop_runners (workspace, machine, person, backend, version, diagnosis, closed, reported_at)
762 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
763 ON CONFLICT (workspace, machine) DO UPDATE SET person = ?3, backend = ?4, version = ?5, diagnosis = ?6, closed = ?7, reported_at = ?8",
764 )
765 .bind(&[
766 place.workspace.as_str().into(),
767 machine.as_str().into(),
768 a.actor.username.as_str().into(),
769 a.backend.as_str().into(),
770 optional(a.version.as_deref().map(|v| clip(v, 40)).as_deref()),
771 serde_json::to_string(&tidy(&a.diagnosis))?.into(),
772 serde_json::to_string(&tidy(&a.closed))?.into(),
773 rfc3339(at).into(),
774 ])?
775 .run()
776 .await?;
777 Ok(Outcome::Ok(true))
778 }
779
780 /// `desktop_runners`: the workspace's desktops, as each app last said.
781 pub async fn desktop_runners(&self, a: DesktopRunnersArgs) -> Result<Outcome<Vec<DesktopRunner>>> {
782 let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) };
783 let place = check!(self.runner_place(&a.actor, &owner, false).await?);
784 #[derive(Deserialize)]
785 struct Row {
786 workspace: String,
787 machine: String,
788 person: String,
789 backend: String,
790 version: Option<String>,
791 diagnosis: String,
792 closed: String,
793 reported_at: String,
794 }
795 let rows = self
796 .db
797 .prepare("SELECT * FROM desktop_runners WHERE workspace = ? ORDER BY lower(machine)")
798 .bind(&[place.workspace.as_str().into()])?
799 .all()
800 .await?
801 .results::<Row>()?;
802 let open_since = rfc3339(now_ms().saturating_sub(model::DESKTOP_OPEN_WITHIN_MS));
803 Ok(Outcome::Ok(
804 rows.into_iter()
805 .map(|row| DesktopRunner {
806 open: row.reported_at.as_str() > open_since.as_str(),
807 diagnosis: serde_json::from_str(&row.diagnosis).unwrap_or_default(),
808 closed: serde_json::from_str(&row.closed).unwrap_or_default(),
809 workspace: row.workspace,
810 machine: row.machine,
811 person: row.person,
812 backend: row.backend,
813 version: row.version,
814 reported_at: row.reported_at,
815 })
816 .collect(),
817 ))
818 }
819
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays820 /// `set_runner_settings`.
821 pub async fn set_runner_settings(&self, a: SetRunnerSettingsArgs) -> Result<Outcome<RunnerSettings>> {
822 let place = check!(self.runner_place(&a.actor, &a.owner, true).await?);
823 let (owner, scope) = match &place.repo {
824 Some(repo) => (repo.id.clone(), "repository"),
825 None => (place.workspace.clone(), "workspace"),
826 };
827 if a.inherit && place.repo.is_some() {
828 self.db.prepare("DELETE FROM runner_settings WHERE owner = ?").bind(&[owner.as_str().into()])?.run().await?;
829 } else {
830 let current = self.effective_runner_settings(&place.workspace, place.repo.as_ref().map(|r| r.id.as_str())).await?;
831 let labels = match &a.agent_labels {
832 Some(given) => match model::runner_labels(given, model::SELF_HOSTED, model::SELF_HOSTED) {
833 Ok(labels) => labels,
834 Err(problem) => return Ok(fail(FailureCode::Invalid, problem)),
835 },
836 None => current.agent_labels.clone(),
837 };
838 let agents = a.agents_on_self_hosted.unwrap_or(current.agents_on_self_hosted);
839 let forks = a.fork_pull_requests.unwrap_or(current.fork_pull_requests);
840 self.db
841 .prepare(
842 "INSERT INTO runner_settings (owner, scope, agents, agent_labels, fork_pulls, updated_at, updated_by)
843 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
844 ON CONFLICT (owner) DO UPDATE SET agents = ?3, agent_labels = ?4, fork_pulls = ?5, updated_at = ?6, updated_by = ?7",
845 )
846 .bind(&[
847 owner.as_str().into(),
848 scope.into(),
849 u32::from(agents).into(),
850 serde_json::to_string(&labels)?.into(),
851 u32::from(forks).into(),
852 now().into(),
853 a.actor.username.as_str().into(),
854 ])?
855 .run()
856 .await?;
857 }
858 let settings = self.effective_runner_settings(&place.workspace, place.repo.as_ref().map(|r| r.id.as_str())).await?;
859 self.audit(vec![entry(
860 AuditActor::of(&a.actor),
861 "update_runner_settings",
862 &place.workspace,
863 place.repo.as_ref().map(|r| format!("{}/{}", r.namespace, r.name)),
864 None,
865 format!(
866 "Agents on self-hosted runners: {}; labels {}; pull requests from forks: {}.",
867 if settings.agents_on_self_hosted { "on" } else { "off" },
868 settings.agent_labels.join(", "),
869 if settings.fork_pull_requests { "allowed" } else { "not allowed" }
870 ),
871 )])
872 .await;
873 Ok(Outcome::Ok(settings))
874 }
875
876 // --- The runner's side ---------------------------------------------------
877
878 /// `runner_register`.
879 pub async fn runner_register(&self, a: RegisterArgs) -> Result<Outcome<Registered>> {
880 let refused = || fail(FailureCode::Unauthenticated, "That registration token is not valid, or it expired. Make a new one under Settings, Runners.");
881 if !a.token.starts_with(model::REGISTRATION_PREFIX) {
882 return Ok(refused());
883 }
884 let registration = self
885 .db
886 .prepare("SELECT * FROM runner_registrations WHERE token_hash = ?")
887 .bind(&[hash(&a.token).into()])?
888 .first::<RegistrationRow>(None)
889 .await?;
890 let at = now_ms();
891 let Some(registration) = registration.filter(|r| r.expires_at.as_str() > rfc3339(at).as_str()) else {
892 return Ok(refused());
893 };
894 let name = match valid_name(&a.name) {
895 Ok(name) => name,
896 Err(problem) => return Ok(fail(FailureCode::Invalid, problem)),
897 };
898 let (Some(os), Some(arch)) = (model::os_label(&a.os), model::arch_label(&a.arch)) else {
899 return Ok(fail(FailureCode::Invalid, "A runner's os is linux, macos or windows, and its arch x64 or arm64."));
900 };
901 let labels = match model::runner_labels(&a.labels, os, arch) {
902 Ok(labels) => labels,
903 Err(problem) => return Ok(fail(FailureCode::Invalid, problem)),
904 };
905 let workspace = registration.workspace.clone();
906 // Abuse: so many machines, so fast, are not a team's.
907 let recent = self
908 .db
909 .prepare("SELECT COUNT(*) AS n FROM runners WHERE workspace = ? AND created_at > ?")
910 .bind(&[workspace.as_str().into(), rfc3339(at.saturating_sub(60 * 1000)).into()])?
911 .first::<Count>(None)
912 .await?
913 .map_or(0, |c| c.n);
914 if recent >= model::MAX_REGISTRATIONS_PER_MINUTE {
915 return Ok(fail(FailureCode::Conflict, "Too many runners registered in the last minute. Wait a moment and try again."));
916 }
917 let total = self
918 .db
919 .prepare("SELECT COUNT(*) AS n FROM runners WHERE workspace = ?")
920 .bind(&[workspace.as_str().into()])?
921 .first::<Count>(None)
922 .await?
923 .map_or(0, |c| c.n);
924 // A group: the one asked for, the token's, or the default.
925 let group = if registration.repo_id.is_some() {
926 None
927 } else {
928 match a.group.as_deref().map(str::trim).filter(|g| !g.is_empty()) {
929 Some(wanted) => match self.find_group(&workspace, wanted).await? {
930 Some(group) => Some(group.id),
931 None => return Ok(fail(FailureCode::NotFound, format!("There is no runner group called {wanted}."))),
932 },
933 None => match registration.group_id.clone() {
934 Some(id) => Some(id),
935 None => Some(self.default_group(&workspace).await?.id),
936 },
937 }
938 };
939 let existing = self
940 .db
941 .prepare("SELECT * FROM runners WHERE workspace = ? AND COALESCE(repo_id, '') = ? AND lower(name) = lower(?)")
942 .bind(&[workspace.as_str().into(), registration.repo_id.clone().unwrap_or_default().into(), name.as_str().into()])?
943 .first::<RunnerRow>(None)
944 .await?;
945 if let Some(existing) = &existing {
946 if !a.replace {
947 return Ok(fail(
948 FailureCode::Conflict,
949 format!("A runner called {name} is already registered here. Register with --replace to take its place, or choose another --name."),
950 ));
951 }
952 self.forget_runner(existing, &format!("The runner {name} was replaced by a new registration.")).await?;
953 } else if total >= model::MAX_RUNNERS {
954 return Ok(fail(FailureCode::Conflict, format!("A workspace has at most {} self-hosted runners.", model::MAX_RUNNERS)));
955 }
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r9956 // How it keeps work from its machine, if it said (older runners do not).
957 let isolation = a.isolation.as_deref().filter(|level| model::ISOLATIONS.contains(level));
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays958 let id = new_id("rnr", at);
959 let credential = format!("{}{}", model::CREDENTIAL_PREFIX, random_hex(32));
960 let created = rfc3339(at);
961 self.db
962 .prepare(
963 "INSERT INTO runners (id, workspace, repo_id, repo, group_id, name, labels, os, arch, version, ephemeral, credential_hash,
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r9964 rotated_at, last_seen_at, created_at, created_by, isolation)
965 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays966 )
967 .bind(&[
968 id.as_str().into(),
969 workspace.as_str().into(),
970 optional(registration.repo_id.as_deref()),
971 optional(registration.repo.as_deref()),
972 optional(group.as_deref()),
973 name.as_str().into(),
974 serde_json::to_string(&labels)?.into(),
975 os.into(),
976 arch.into(),
977 a.version.chars().take(40).collect::<String>().into(),
978 u32::from(a.ephemeral).into(),
979 hash(&credential).into(),
980 created.as_str().into(),
981 created.as_str().into(),
982 created.as_str().into(),
Merge the desktop app's runner: it starts with the app for the signed-in person, sandboxed by default and running nothing where no sandbox is available yet, registers through the site, keeps itself up to date, and says everywhere what it is doing, why it is not, and how to fix it.983 // The person who made the token: a desktop's runner is theirs.
984 optional(registration.created_by.as_deref()),
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r9985 optional(isolation),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays986 ])?
987 .run()
988 .await?;
989 self.db
990 .prepare("UPDATE runner_registrations SET used = used + 1 WHERE token_hash = ?")
991 .bind(&[hash(&a.token).into()])?
992 .run()
993 .await?;
994 let row = self.runner_by_id(&id).await?.ok_or_else(|| worker::Error::RustError("the runner was not kept".into()))?;
995 self.audit(vec![entry(
996 runner_actor(&row),
997 "register_runner",
998 &workspace,
999 row.repo.clone(),
1000 Some(name.clone()),
1001 format!("Registered the self-hosted runner {name} ({os}, {arch}; labels {}).", labels.join(", ")),
1002 )])
1003 .await;
1004 let groups = self.group_rows(&workspace).await?;
1005 Ok(Outcome::Ok(Registered {
1006 runner: self.runner_view(&row, &groups, at).await?,
1007 credential,
1008 }))
1009 }
1010
1011 /// The runner a call is from, if its credential is that runner's (or
1012 /// the one before a rotation, until the new one is used).
1013 async fn authenticated(&self, auth: &RunnerAuth) -> Result<Option<(RunnerRow, bool)>> {
1014 if !auth.credential.starts_with(model::CREDENTIAL_PREFIX) {
1015 return Ok(None);
1016 }
1017 let Some(row) = self.runner_by_id(&auth.runner).await? else { return Ok(None) };
1018 let given = hash(&auth.credential);
1019 if g1t_secrets::same(&row.credential_hash, &given) {
1020 if row.previous_hash.is_some() {
1021 self.db.prepare("UPDATE runners SET previous_hash = NULL WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
1022 }
1023 return Ok(Some((row, true)));
1024 }
1025 if row.previous_hash.as_deref().is_some_and(|previous| g1t_secrets::same(previous, &given)) {
1026 return Ok(Some((row, false)));
1027 }
1028 Ok(None)
1029 }
1030
1031 fn unknown_runner<T>() -> Outcome<T> {
1032 fail(FailureCode::Unauthenticated, "This runner is not registered, or its credential is not valid. Register it again.")
1033 }
1034
1035 /// `runner_poll`.
1036 pub async fn runner_poll(&self, a: PollArgs) -> Result<Outcome<Poll>> {
1037 let Some((row, current)) = self.authenticated(&a.auth).await? else {
1038 return Ok(Self::unknown_runner());
1039 };
1040 let at = now_ms();
1041 let seen = rfc3339(at);
1042 // A new credential every day, handed over on a poll that used the
1043 // current one.
1044 let mut credential = None;
1045 let rotate_after = rfc3339(at.saturating_sub(model::ROTATE_AFTER_MS));
1046 if current && row.rotated_at.as_str() < rotate_after.as_str() {
1047 let fresh = format!("{}{}", model::CREDENTIAL_PREFIX, random_hex(32));
1048 self.db
1049 .prepare("UPDATE runners SET previous_hash = credential_hash, credential_hash = ?, rotated_at = ? WHERE id = ?")
1050 .bind(&[hash(&fresh).into(), seen.as_str().into(), row.id.as_str().into()])?
1051 .run()
1052 .await?;
1053 credential = Some(fresh);
1054 }
1055 // What it holds that it should stop, and the heartbeat for the rest.
1056 let mut cancel = Vec::new();
1057 let mut active: Option<(String, &'static str)> = None;
1058 for id in a.running.iter().take(10) {
1059 if let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[id.as_str().into()])?.first::<JobRow>(None).await? {
1060 if job.status == "in_progress" && job.runner_id.as_deref() == Some(row.id.as_str()) {
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1061 if seen_is_stale(job.seen_at.as_deref(), at) {
1062 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?;
1063 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1064 active = Some((id.clone(), "workflow"));
1065 } else {
1066 cancel.push(id.clone());
1067 }
1068 continue;
1069 }
1070 match self.db.prepare("SELECT * FROM runner_tasks WHERE id = ?").bind(&[id.as_str().into()])?.first::<TaskRow>(None).await? {
1071 Some(task) if task.status == "in_progress" && task.runner_id.as_deref() == Some(row.id.as_str()) => {
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1072 if seen_is_stale(task.seen_at.as_deref(), at) {
1073 self.db.prepare("UPDATE runner_tasks SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?;
1074 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1075 active = Some((id.clone(), "agent"));
1076 }
1077 _ => cancel.push(id.clone()),
1078 }
1079 }
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1080 // Written when something changed, or once a minute to keep it online.
1081 let version: String = if a.version.is_empty() { row.version.clone() } else { a.version.chars().take(40).collect() };
1082 let work_id = active.as_ref().map(|(id, _)| id.as_str());
1083 let work_kind = active.as_ref().map(|(_, kind)| *kind);
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.1084 // How it is isolated, as it says now: a runner re-registered or
1085 // restarted with or without Docker says so on its next poll.
1086 let isolation = a
1087 .isolation
1088 .as_deref()
1089 .filter(|level| model::ISOLATIONS.contains(level))
1090 .map(str::to_owned)
1091 .or_else(|| row.isolation.clone());
1092 let changed = version != row.version
1093 || work_id != row.work_id.as_deref()
1094 || work_kind != row.work_kind.as_deref()
1095 || isolation != row.isolation;
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1096 if changed || seen_is_stale(row.last_seen_at.as_deref(), at) {
1097 self.db
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.1098 .prepare("UPDATE runners SET last_seen_at = ?, version = ?, work_id = ?, work_kind = ?, isolation = ? WHERE id = ?")
1099 .bind(&[
1100 seen.as_str().into(),
1101 version.into(),
1102 optional(work_id),
1103 optional(work_kind),
1104 optional(isolation.as_deref()),
1105 row.id.as_str().into(),
1106 ])?
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1107 .run()
1108 .await?;
1109 }
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.1110 let row = RunnerRow { isolation, ..row };
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1111 let mut poll = Poll { assignment: None, cancel, credential, removed: false };
1112 // Busy, or an ephemeral runner that has had its one job.
1113 if active.is_some() || !a.running.is_empty() {
1114 return Ok(Outcome::Ok(poll));
1115 }
1116 if row.ephemeral != 0 && row.spent != 0 {
1117 self.forget_runner(&row, "The ephemeral runner finished its job.").await?;
1118 poll.removed = true;
1119 return Ok(Outcome::Ok(poll));
1120 }
1121 let groups = self.group_rows(&row.workspace).await?;
1122 let group = row.group_id.as_ref().and_then(|id| groups.iter().find(|g| &g.id == id)).cloned();
1123 let deadline = at + a.wait_ms.min(model::MAX_POLL_WAIT_MS);
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.1124 let mut looked = Looked::default();
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1125 loop {
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.1126 if let Some(assignment) = self.claim(&row, group.as_ref(), &mut looked).await? {
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1127 poll.assignment = Some(assignment);
1128 return Ok(Outcome::Ok(poll));
1129 }
1130 if now_ms() + POLL_EVERY_MS > deadline {
1131 return Ok(Outcome::Ok(poll));
1132 }
1133 worker::Delay::from(std::time::Duration::from_millis(POLL_EVERY_MS)).await;
1134 }
1135 }
1136
1137 /// Takes the oldest work this runner can do, if there is any.
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.1138 async fn claim(&self, runner: &RunnerRow, group: Option<&GroupRow>, looked: &mut Looked) -> Result<Option<Assignment>> {
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1139 let labels = runner.labels();
1140 let group_name = group.map(|g| g.name.clone());
1141 // Workflow jobs.
1142 let mut sql = "SELECT jobs.*, runs.repo AS run_repo FROM jobs JOIN runs ON runs.id = jobs.run_id
1143 WHERE jobs.status = 'queued' AND jobs.labels IS NOT NULL AND lower(jobs.namespace) = ?"
1144 .to_owned();
1145 let mut binds: Vec<JsValue> = vec![runner.workspace.as_str().into()];
1146 if let Some(repo_id) = &runner.repo_id {
1147 sql.push_str(" AND jobs.repo_id = ?");
1148 binds.push(repo_id.as_str().into());
1149 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1150 // One job of a concurrency group runs at a time.
1151 sql.push_str(
1152 " AND (jobs.concurrency_group IS NULL OR NOT EXISTS (SELECT 1 FROM jobs o WHERE o.repo_id = jobs.repo_id
Merge branch 'worktree-agent-af75396ad4c6ee7a8' into claude/cool-newton-7nm8r91153 AND o.concurrency_group = jobs.concurrency_group AND o.status IN ('in_progress', 'calling')))",
Merge branch 'worktree-agent-a3abfcce648e87dca'1154 );
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1155 sql.push_str(" ORDER BY jobs.rowid LIMIT 50");
1156 #[derive(Deserialize)]
1157 struct Queued {
1158 #[serde(flatten)]
1159 job: JobRow,
1160 run_repo: String,
1161 }
1162 let queued = self.db.prepare(sql).bind(&binds)?.all().await?.results::<Queued>()?;
1163 for Queued { job, run_repo } in queued {
1164 let stored: Vec<String> = job.labels.as_deref().and_then(|l| serde_json::from_str(l).ok()).unwrap_or_default();
1165 let wanted = Wanted::from_stored(&stored);
1166 if !wanted.matches(&labels, group_name.as_deref()) {
1167 continue;
1168 }
1169 if runner.repo_id.is_none() && !group.is_none_or(|g| g.allows(&run_repo)) {
1170 continue;
1171 }
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.1172 // A host-mode runner takes no public repository's job: it waits
1173 // for a sandboxed runner, and says so when none could take it.
1174 if runner.isolation.as_deref() == Some("host") {
1175 let visibility = self.visibility_once(looked, &job.repo_id).await;
1176 if !model::isolation_allows(runner.isolation.as_deref(), visibility.as_deref()) {
1177 if looked.noted.insert(job.id.clone()) {
1178 self.wait_for_sandbox(&job, &run_repo, &wanted).await?;
1179 }
1180 continue;
1181 }
1182 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1183 if let Some(max) = job.max_parallel {
1184 let siblings = self
1185 .db
1186 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")
1187 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1188 .first::<Count>(None)
1189 .await?
1190 .map_or(0, |c| c.n);
1191 if siblings >= max {
1192 continue;
1193 }
1194 }
1195 let token = random_hex(24);
1196 let at = now();
1197 let claimed = self
1198 .db
1199 .prepare(
1200 "UPDATE jobs SET status = 'in_progress', token_hash = ?, runner_id = ?, runner_name = ?, reason = NULL, started_at = ?, seen_at = ?
1201 WHERE id = ? AND status = 'queued' RETURNING id",
1202 )
1203 .bind(&[
1204 sha256_hex(&token).into(),
1205 runner.id.as_str().into(),
1206 runner.name.as_str().into(),
1207 at.as_str().into(),
1208 at.as_str().into(),
1209 job.id.as_str().into(),
1210 ])?
1211 .first::<Value>(None)
1212 .await?;
1213 if claimed.is_none() {
1214 continue;
1215 }
1216 self.db
1217 .batch(vec![
1218 self.db
1219 .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")
1220 .bind(&[at.as_str().into(), job.run_id.as_str().into()])?,
1221 self.db
1222 .prepare("UPDATE runners SET work_id = ?, work_kind = 'workflow', spent = ephemeral WHERE id = ?")
1223 .bind(&[job.id.as_str().into(), runner.id.as_str().into()])?,
1224 ])
1225 .await?;
1226 self.audit(vec![entry(
1227 runner_actor(runner),
1228 "runner.take_job",
1229 &runner.workspace,
1230 Some(run_repo.clone()),
1231 Some(job.name.clone()),
1232 format!("The self-hosted runner {} took the job {}.", runner.name, job.name),
1233 )])
1234 .await;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971235 // A job that deploys: its run's deployment is under way.
1236 self.job_started(&job.id).await?;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1237 let image = self.job_image(&job).await?;
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r91238 let visibility = self.visibility(&job.repo_id).await;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1239 return Ok(Some(Assignment {
1240 kind: "workflow".into(),
1241 id: job.id.clone(),
1242 name: job.name.clone(),
1243 repo: run_repo,
1244 timeout_minutes: job.timeout_minutes,
1245 image,
1246 token: Some(token),
1247 env: None,
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r91248 visibility,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1249 }));
1250 }
1251
1252 // Agent work.
1253 let tasks = self
1254 .db
1255 .prepare("SELECT * FROM runner_tasks WHERE status = 'queued' AND workspace = ? ORDER BY created_at LIMIT 20")
1256 .bind(&[runner.workspace.as_str().into()])?
1257 .all()
1258 .await?
1259 .results::<TaskRow>()?;
1260 for task in tasks {
1261 let wanted = Wanted::of(&serde_json::from_str::<Vec<String>>(&task.labels).unwrap_or_default(), None);
1262 if !wanted.matches(&labels, None) {
1263 continue;
1264 }
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.1265 // Nor a public repository's agent work.
1266 if runner.isolation.as_deref() == Some("host")
1267 && let Some(repo_id) = &task.repo_id
1268 {
1269 let visibility = self.visibility_once(looked, repo_id).await;
1270 if !model::isolation_allows(runner.isolation.as_deref(), visibility.as_deref()) {
1271 continue;
1272 }
1273 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1274 if let Some(repo_id) = &runner.repo_id
1275 && task.repo_id.as_deref() != Some(repo_id.as_str())
1276 {
1277 continue;
1278 }
1279 if runner.repo_id.is_none() && !group.is_none_or(|g| g.allows(&task.repo)) {
1280 continue;
1281 }
1282 let Some(sealed) = task.env.clone() else { continue };
1283 let at = now();
1284 let claimed = self
1285 .db
1286 .prepare(
1287 "UPDATE runner_tasks SET status = 'in_progress', runner_id = ?, runner_name = ?, env = NULL, started_at = ?, seen_at = ?
1288 WHERE id = ? AND status = 'queued' RETURNING id",
1289 )
1290 .bind(&[runner.id.as_str().into(), runner.name.as_str().into(), at.as_str().into(), at.as_str().into(), task.id.as_str().into()])?
1291 .first::<Value>(None)
1292 .await?;
1293 if claimed.is_none() {
1294 continue;
1295 }
1296 self.db
1297 .prepare("UPDATE runners SET work_id = ?, work_kind = 'agent', spent = ephemeral WHERE id = ?")
1298 .bind(&[task.id.as_str().into(), runner.id.as_str().into()])?
1299 .run()
1300 .await?;
1301 let env: Map<String, Value> = self
1302 .sealer
1303 .as_ref()
1304 .and_then(|sealer| sealer.open(&sealed, &task.id))
1305 .and_then(|text| serde_json::from_str(&text).ok())
1306 .unwrap_or_default();
1307 if env.is_empty() {
1308 self.end_task(&TaskRow { status: "in_progress".into(), ..task.clone() }, 1, "The work's environment could not be opened.").await?;
1309 continue;
1310 }
1311 self.audit(vec![entry(
1312 runner_actor(runner),
1313 "runner.take_task",
1314 &runner.workspace,
1315 Some(task.repo.clone()),
1316 Some(task.kind.clone()),
1317 format!("The self-hosted runner {} took agent work: {}.", runner.name, task.title),
1318 )])
1319 .await;
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r91320 let visibility = match &task.repo_id {
1321 Some(repo_id) => self.visibility(repo_id).await,
1322 None => None,
1323 };
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1324 return Ok(Some(Assignment {
1325 kind: "agent".into(),
1326 id: task.id.clone(),
1327 name: task.title.clone(),
1328 repo: task.repo.clone(),
1329 timeout_minutes: task.timeout_minutes,
1330 image: None,
1331 token: None,
1332 env: Some(env),
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r91333 visibility,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1334 }));
1335 }
1336 Ok(None)
1337 }
1338
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.1339 /// [`Self::visibility`], asked once per repository in a poll.
1340 async fn visibility_once(&self, looked: &mut Looked, repo_id: &str) -> Option<String> {
1341 if let Some(known) = looked.visibilities.get(repo_id) {
1342 return known.clone();
1343 }
1344 let found = self.visibility(repo_id).await;
1345 looked.visibilities.insert(repo_id.to_owned(), found.clone());
1346 found
1347 }
1348
1349 /// A public repository's job that a host-mode runner passed over: when
1350 /// no online runner that may take it could (its labels, group,
1351 /// repository, and isolation), its page says it waits for a sandboxed
1352 /// one. Written only when that changes what it says.
1353 async fn wait_for_sandbox(&self, job: &JobRow, run_repo: &str, wanted: &Wanted) -> Result<()> {
1354 let at = now_ms();
1355 let groups = self.group_rows(&job.namespace.to_lowercase()).await?;
1356 let runners = self.runner_rows(&job.namespace.to_lowercase()).await?;
1357 let sandboxed = runners.iter().any(|other| takes_public_job(other, &groups, &job.repo_id, run_repo, wanted, at));
1358 if sandboxed {
1359 return Ok(());
1360 }
1361 self.db
1362 .prepare("UPDATE jobs SET reason = ? WHERE id = ? AND status = 'queued' AND reason IS NOT ?")
1363 .bind(&[model::WAITING_FOR_SANDBOX.into(), job.id.as_str().into(), model::WAITING_FOR_SANDBOX.into()])?
1364 .run()
1365 .await?;
1366 Ok(())
1367 }
1368
Merge branch 'worktree-agent-ac79a505eca749d4d' into claude/cool-newton-7nm8r91369 /// A repository's visibility for a runner, `public` or `private`; none
1370 /// when it cannot be read, and the runner decides without it.
1371 async fn visibility(&self, repo_id: &str) -> Option<String> {
1372 match self.repo_by_id(repo_id).await {
1373 Ok(Some((repo, _))) => Some(if repo.is_private { "private" } else { "public" }.to_owned()),
1374 _ => None,
1375 }
1376 }
1377
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1378 /// The image a job's `container:` names, when it names one plainly.
1379 async fn job_image(&self, job: &JobRow) -> Result<Option<String>> {
1380 let Some(run) = self.run_row(&job.run_id).await? else { return Ok(None) };
1381 let raw = match job.callee() {
1382 Some((_, spec, _)) => spec.raw,
1383 None => match g1t_actions::workflow::parse(&run.source) {
1384 Ok(workflow) => match workflow.jobs.into_iter().find(|j| j.id == job.key) {
1385 Some(spec) => spec.raw,
1386 None => return Ok(None),
1387 },
1388 Err(_) => return Ok(None),
1389 },
1390 };
1391 let image = match raw.get("container") {
1392 Some(Value::String(image)) => Some(image.clone()),
1393 Some(Value::Object(container)) => container.get("image").and_then(Value::as_str).map(str::to_owned),
1394 _ => None,
1395 };
1396 Ok(image.filter(|image| !image.contains("${{") && !image.trim().is_empty()))
1397 }
1398
1399 /// `runner_finished`.
1400 pub async fn runner_finished(&self, a: FinishedArgs) -> Result<Outcome<bool>> {
1401 let Some((row, _)) = self.authenticated(&a.auth).await? else {
1402 return Ok(Self::unknown_runner());
1403 };
1404 let reason = a.reason.clone().filter(|r| !r.trim().is_empty()).map(|r| r.chars().take(2000).collect::<String>());
1405 let mut changed = false;
1406 if let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.id.as_str().into()])?.first::<JobRow>(None).await? {
1407 if job.runner_id.as_deref() == Some(row.id.as_str()) && job.status == "in_progress" {
1408 let why = reason.unwrap_or_else(|| format!("The job's process on {} exited with code {} before it reported how the job went.", row.name, a.exit_code));
1409 self.finish_job(&job.id, "failure", Some(&why), None).await?;
1410 changed = true;
1411 }
1412 } else if let Some(task) = self.db.prepare("SELECT * FROM runner_tasks WHERE id = ?").bind(&[a.id.as_str().into()])?.first::<TaskRow>(None).await?
1413 && task.runner_id.as_deref() == Some(row.id.as_str())
1414 && task.status == "in_progress"
1415 {
1416 let why = reason.unwrap_or_else(|| if a.exit_code == 0 { String::new() } else { format!("The work exited with code {} on {}.", a.exit_code, row.name) });
1417 self.end_task(&task, a.exit_code, &why).await?;
1418 changed = true;
1419 }
1420 self.db
1421 .prepare("UPDATE runners SET work_id = NULL, work_kind = NULL, last_seen_at = ? WHERE id = ? AND work_id = ?")
1422 .bind(&[now().into(), row.id.as_str().into(), a.id.as_str().into()])?
1423 .run()
1424 .await?;
1425 Ok(Outcome::Ok(changed))
1426 }
1427
1428 /// `runner_remove_self`.
1429 pub async fn runner_remove_self(&self, a: RemoveSelfArgs) -> Result<Outcome<bool>> {
1430 let Some((row, _)) = self.authenticated(&a.auth).await? else {
1431 return Ok(Self::unknown_runner());
1432 };
1433 self.forget_runner(&row, &format!("The runner {} was removed from its machine.", row.name)).await?;
1434 self.audit(vec![entry(
1435 runner_actor(&row),
1436 "remove_runner",
1437 &row.workspace,
1438 row.repo.clone(),
1439 Some(row.name.clone()),
1440 format!("The self-hosted runner {} removed itself.", row.name),
1441 )])
1442 .await;
1443 Ok(Outcome::Ok(true))
1444 }
1445
1446 /// A self-hosted job finished, however it did: its runner is free, and
1447 /// its time goes on the workspace's usage as self-hosted, at $0.
1448 pub async fn released(&self, job: &JobRow) -> Result<()> {
1449 let Some(runner_id) = &job.runner_id else { return Ok(()) };
1450 self.db
1451 .prepare("UPDATE runners SET work_id = NULL, work_kind = NULL WHERE id = ? AND work_id = ?")
1452 .bind(&[runner_id.as_str().into(), job.id.as_str().into()])?
1453 .run()
1454 .await?;
1455 let (Some(started), Some(finished)) = (job.started_at.as_deref(), job.finished_at.as_deref().map(str::to_owned).or_else(|| Some(now()))) else {
1456 return Ok(());
1457 };
1458 let seconds = match (g1t_contracts::time::parse_rfc3339(started), g1t_contracts::time::parse_rfc3339(&finished)) {
1459 (Some(a), Some(b)) if b > a => ((b - a) / 1000).max(1),
1460 _ => 1,
1461 };
1462 let run = self.run_row(&job.run_id).await?;
1463 let repo = run.as_ref().map(|r| r.repo.clone());
1464 let recorded: Result<Outcome<bool>> = g1t_kit::call(
1465 &self.billing,
1466 "record_sandbox",
1467 &RecordSandboxArgs {
1468 workspace: job.namespace.to_lowercase(),
1469 seconds: u32::try_from(seconds).unwrap_or(u32::MAX),
1470 description: format!(
1471 "{} in {} on the self-hosted runner {}",
1472 job.name,
1473 repo.clone().unwrap_or_default(),
1474 job.runner_name.clone().unwrap_or_default()
1475 ),
1476 repo,
1477 reference: format!("selfhosted/{}/{}", job.id, started),
1478 kind: Some(ComputeKind::Workflow),
1479 cpu_seconds: None,
A workflow job on a self-hosted runner records its sandbox time with no agent and nobody who asked, so the actions service builds again now that every sandbox line can name them.1480 // A workflow job on a self-hosted runner: no agent ran it and nobody asked an agent for it.
1481 agent: None,
1482 asked_by: None,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1483 reservation_id: None,
1484 self_hosted: true,
1485 instance: None,
1486 },
1487 )
1488 .await;
1489 if let Err(error) = recorded {
1490 worker::console_error!("actions: self-hosted time not recorded for {}: {error}", job.id);
1491 }
1492 Ok(())
1493 }
1494
1495 /// `runner` and `RUNNER_*` for a job on a self-hosted runner.
1496 pub async fn runner_context_for(&self, runner_id: &str, variables: &mut Map<String, Value>) -> Result<Value> {
1497 let mut context = g1t_actions::events::runner_context();
1498 let Some(runner) = self.runner_by_id(runner_id).await? else { return Ok(context) };
1499 let os = match runner.os.as_str() {
1500 "macos" => "macOS",
1501 "windows" => "Windows",
1502 _ => "Linux",
1503 };
1504 let arch = if runner.arch == "arm64" { "ARM64" } else { "X64" };
1505 context["name"] = json!(runner.name);
1506 context["os"] = json!(os);
1507 context["arch"] = json!(arch);
1508 context["environment"] = json!("self-hosted");
1509 {
1510 let vars = variables;
1511 vars.insert("RUNNER_NAME".into(), json!(runner.name));
1512 vars.insert("RUNNER_OS".into(), json!(os));
1513 vars.insert("RUNNER_ARCH".into(), json!(arch));
1514 vars.insert("RUNNER_ENVIRONMENT".into(), json!("self-hosted"));
1515 }
1516 Ok(context)
1517 }
1518
1519 // --- Agent work, for the runner service ----------------------------------
1520
1521 /// `runner_route`: the labels g1t's own work in `repo` runs on, when
1522 /// its workspace (or the repository) sends it to self-hosted runners.
1523 pub async fn runner_route(&self, a: RouteArgs) -> Result<Option<Vec<String>>> {
1524 let repo_id = match self.repo_by_path(&a.repo).await? {
1525 Some(repo) => Some(repo.id),
1526 None => None,
1527 };
1528 let settings = self.effective_runner_settings(&a.workspace, repo_id.as_deref()).await?;
1529 Ok(settings.agents_on_self_hosted.then_some(settings.agent_labels))
1530 }
1531
1532 async fn repo_by_path(&self, path: &RepoPath) -> Result<Option<Repo>> {
1533 let Some(actor) = self.workspace_actor(&path.namespace).await? else { return Ok(None) };
1534 self.visible_repo(path, &Some(actor)).await
1535 }
1536
1537 /// `enqueue_task`.
1538 pub async fn enqueue_task(&self, a: EnqueueTaskArgs) -> Result<Outcome<String>> {
1539 let Some(sealer) = &self.sealer else {
1540 return Ok(fail(FailureCode::Conflict, "Agent work cannot be handed to self-hosted runners yet: g1t's key for it is not set."));
1541 };
1542 let id = new_id("rtk", now_ms());
1543 let sealed = sealer.seal(&serde_json::to_string(&a.env)?, &id);
1544 let repo_id = self.repo_by_path(&a.repo).await?.map(|r| r.id);
1545 let labels = match model::runner_labels(&a.labels, model::SELF_HOSTED, model::SELF_HOSTED) {
1546 Ok(labels) => labels,
1547 Err(problem) => return Ok(fail(FailureCode::Invalid, problem)),
1548 };
1549 let at = now();
1550 let inserted = self
1551 .db
1552 .prepare(
The Runners page shows the machines a workspace's agents and workflow jobs run on, in Workspace under Compute: g1t's cloud beside your own runners, what each is running now with a link to it, what's waiting, this month's machine time and its cost (your own runners' free), where agent work and workflow jobs run, adding a runner, and groups; runner_activity in Actions says what runs where, work handed to your runners keeps its agent run, and the self-hosted runners guide says how.1553 "INSERT OR IGNORE INTO runner_tasks (id, sandbox, workspace, repo_id, repo, kind, title, labels, env, timeout_minutes, status, created_at, run_id)
1554 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?) RETURNING id",
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1555 )
1556 .bind(&[
1557 id.as_str().into(),
1558 a.sandbox.as_str().into(),
1559 a.workspace.to_lowercase().into(),
1560 optional(repo_id.as_deref()),
1561 format!("{}/{}", a.repo.namespace, a.repo.name).into(),
1562 a.kind.as_str().into(),
1563 a.title.chars().take(300).collect::<String>().into(),
1564 serde_json::to_string(&labels)?.into(),
1565 sealed.into(),
1566 a.timeout_minutes.max(1).into(),
1567 at.into(),
The Runners page shows the machines a workspace's agents and workflow jobs run on, in Workspace under Compute: g1t's cloud beside your own runners, what each is running now with a link to it, what's waiting, this month's machine time and its cost (your own runners' free), where agent work and workflow jobs run, adding a runner, and groups; runner_activity in Actions says what runs where, work handed to your runners keeps its agent run, and the self-hosted runners guide says how.1568 optional(a.run_id.as_deref().filter(|id| !id.is_empty())),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1569 ])?
1570 .first::<Value>(None)
1571 .await?;
1572 if inserted.is_none() {
1573 return Ok(fail(FailureCode::Conflict, "That sandbox already handed work to a self-hosted runner."));
1574 }
1575 Ok(Outcome::Ok(id))
1576 }
1577
1578 /// `cancel_task`: the sandbox stopped it. Its runner hears on its next
1579 /// poll; the sandbox does its own ending.
1580 pub async fn cancel_task(&self, a: CancelTaskArgs) -> Result<Outcome<bool>> {
1581 let done = self
1582 .db
1583 .prepare(
1584 "UPDATE runner_tasks SET status = 'completed', exit_code = -1, reason = ?, env = NULL, finished_at = ?
1585 WHERE sandbox = ? AND status != 'completed' RETURNING id",
1586 )
1587 .bind(&[optional(a.reason.as_deref()), now().into(), a.sandbox.as_str().into()])?
1588 .first::<Value>(None)
1589 .await?;
1590 Ok(Outcome::Ok(done.is_some()))
1591 }
1592
1593 /// Ends a task and tells the sandbox waiting on it, once.
1594 async fn end_task(&self, task: &TaskRow, exit_code: i32, reason: &str) -> Result<()> {
1595 let ended = self
1596 .db
1597 .prepare(
1598 "UPDATE runner_tasks SET status = 'completed', exit_code = ?, reason = ?, env = NULL, finished_at = ?
1599 WHERE id = ? AND status != 'completed' RETURNING id",
1600 )
1601 .bind(&[exit_code.into(), optional(Some(reason).filter(|r| !r.is_empty())), now().into(), task.id.as_str().into()])?
1602 .first::<Value>(None)
1603 .await?;
1604 if ended.is_none() {
1605 return Ok(());
1606 }
1607 if let Some(runner_id) = &task.runner_id {
1608 self.db
1609 .prepare("UPDATE runners SET work_id = NULL, work_kind = NULL WHERE id = ? AND work_id = ?")
1610 .bind(&[runner_id.as_str().into(), task.id.as_str().into()])?
1611 .run()
1612 .await?;
1613 }
1614 let told: Result<Value> = g1t_kit::call(
1615 &self.runner,
1616 "task_ended",
1617 &json!({
1618 "sandbox": task.sandbox,
1619 "exitCode": exit_code,
1620 "reason": if reason.is_empty() { Value::Null } else { json!(reason) },
1621 "runner": task.runner_name,
1622 }),
1623 )
1624 .await;
1625 if let Err(error) = told {
1626 worker::console_error!("actions: the sandbox waiting on task {} was not told: {error}", task.id);
1627 }
1628 Ok(())
1629 }
1630
The Runners page shows the machines a workspace's agents and workflow jobs run on, in Workspace under Compute: g1t's cloud beside your own runners, what each is running now with a link to it, what's waiting, this month's machine time and its cost (your own runners' free), where agent work and workflow jobs run, adding a runner, and groups; runner_activity in Actions says what runs where, work handed to your runners keeps its agent run, and the self-hosted runners guide says how.1631 /// `runner_activity`: what runs for the workspace now, in g1t's
1632 /// sandboxes and handed to its own runners. Its owners only, as its
1633 /// runners are.
1634 pub async fn runner_activity(&self, a: RunnerActivityArgs) -> Result<Outcome<RunnerActivity>> {
1635 let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) };
1636 let place = check!(self.runner_place(&a.actor, &owner, false).await?);
1637 let workspace = place.workspace;
1638 let cloud_jobs = self
1639 .db
1640 .prepare(
1641 "SELECT jobs.id, jobs.name, jobs.run_id, runs.repo, jobs.started_at FROM jobs JOIN runs ON runs.id = jobs.run_id
1642 WHERE lower(jobs.namespace) = ? AND jobs.status = 'in_progress' AND jobs.labels IS NULL
1643 ORDER BY jobs.started_at DESC LIMIT ?",
1644 )
1645 .bind(&[workspace.as_str().into(), ACTIVITY_LIMIT.into()])?
1646 .all()
1647 .await?
1648 .results::<CloudJob>()?;
1649 let cloud_jobs_queued = self.queued_jobs(&workspace, false).await?;
1650 let self_hosted_jobs_queued = self.queued_jobs(&workspace, true).await?;
1651 let handed_over = self
1652 .db
1653 .prepare(
1654 "SELECT id, run_id, kind, title, repo, status, runner_name, created_at, started_at FROM runner_tasks
1655 WHERE status IN ('queued', 'in_progress') AND workspace = ?
1656 ORDER BY created_at DESC LIMIT ?",
1657 )
1658 .bind(&[workspace.as_str().into(), ACTIVITY_LIMIT.into()])?
1659 .all()
1660 .await?
1661 .results::<HandedOverTask>()?;
1662 Ok(Outcome::Ok(RunnerActivity { cloud_jobs, cloud_jobs_queued, handed_over, self_hosted_jobs_queued }))
1663 }
1664
1665 /// The workspace's workflow jobs queued for its own runners
1666 /// (`labelled`), or for g1t's sandboxes.
1667 async fn queued_jobs(&self, workspace: &str, labelled: bool) -> Result<u32> {
1668 let sql = if labelled {
1669 "SELECT COUNT(*) AS n FROM jobs WHERE lower(namespace) = ? AND status = 'queued' AND labels IS NOT NULL"
1670 } else {
1671 "SELECT COUNT(*) AS n FROM jobs WHERE lower(namespace) = ? AND status = 'queued' AND labels IS NULL"
1672 };
1673 Ok(self.db.prepare(sql).bind(&[workspace.into()])?.first::<Count>(None).await?.map_or(0, |c| c.n))
1674 }
1675
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1676 /// `stuck_jobs`: jobs of the viewer's workspaces that have waited ten
1677 /// minutes or more for a self-hosted runner while none that could take
1678 /// them is online, for Mission control's Needs you.
1679 pub async fn stuck_jobs(&self, a: StuckJobsArgs) -> Result<Vec<StuckJob>> {
1680 let Some(viewer) = a.viewer else { return Ok(Vec::new()) };
1681 let at = now_ms();
1682 let mut out = Vec::new();
1683 for membership in viewer.workspaces.iter().take(20) {
1684 let workspace = membership.slug.to_lowercase();
1685 #[derive(Deserialize)]
1686 struct Waiting {
1687 id: String,
1688 name: String,
1689 run_id: String,
1690 labels: String,
1691 queued_at: Option<String>,
1692 repo: String,
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.1693 repo_id: String,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1694 }
1695 let waiting = self
1696 .db
1697 .prepare(
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.1698 "SELECT jobs.id, jobs.name, jobs.run_id, jobs.labels, jobs.queued_at, runs.repo, jobs.repo_id FROM jobs JOIN runs ON runs.id = jobs.run_id
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1699 WHERE jobs.status = 'queued' AND jobs.labels IS NOT NULL AND lower(jobs.namespace) = ? AND jobs.queued_at < ?
1700 ORDER BY jobs.rowid LIMIT 20",
1701 )
1702 .bind(&[workspace.as_str().into(), rfc3339(at.saturating_sub(10 * 60 * 1000)).into()])?
1703 .all()
1704 .await?
1705 .results::<Waiting>()?;
1706 if waiting.is_empty() {
1707 continue;
1708 }
1709 let groups = self.group_rows(&workspace).await?;
1710 let online: Vec<RunnerRow> = self.runner_rows(&workspace).await?.into_iter().filter(|r| r.online(at)).collect();
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.1711 // A repository's visibility is asked only when a host-mode runner
1712 // is online, since only those pass over a public repository's jobs.
1713 let any_host = online.iter().any(|runner| runner.isolation.as_deref() == Some("host"));
1714 let mut looked = Looked::default();
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1715 for job in waiting {
1716 let wanted = Wanted::from_stored(&serde_json::from_str::<Vec<String>>(&job.labels).unwrap_or_default());
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.1717 let visibility = if any_host { self.visibility_once(&mut looked, &job.repo_id).await } else { None };
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1718 let served = online.iter().any(|runner| {
1719 let group = runner.group_id.as_ref().and_then(|id| groups.iter().find(|g| &g.id == id));
1720 wanted.matches(&runner.labels(), group.map(|g| g.name.as_str()))
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.1721 && model::isolation_allows(runner.isolation.as_deref(), visibility.as_deref())
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1722 });
1723 if !served {
1724 out.push(StuckJob {
1725 id: job.id,
1726 name: job.name,
1727 run_id: job.run_id,
1728 repo: job.repo,
1729 labels: wanted.describe(),
1730 queued_at: job.queued_at.unwrap_or_default(),
1731 });
1732 }
1733 }
1734 }
1735 Ok(out)
1736 }
1737
1738 // --- Every minute ----------------------------------------------------------
1739
1740 /// Jobs and tasks that waited too long, tasks whose runner went quiet,
1741 /// runners offline for weeks, and registration tokens long expired.
1742 pub async fn sweep_runners(&self, at: u64) -> Result<()> {
1743 let before = |ms: u64| rfc3339(at.saturating_sub(ms));
1744 let waited = model::MAX_WAIT_HOURS * 60 * 60 * 1000;
1745 let stale = self
1746 .db
1747 .prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NOT NULL AND queued_at < ? LIMIT 50")
1748 .bind(&[before(waited).into()])?
1749 .all()
1750 .await?
1751 .results::<JobRow>()?;
1752 for job in stale {
1753 let stored: Vec<String> = job.labels.as_deref().and_then(|l| serde_json::from_str(l).ok()).unwrap_or_default();
1754 let why = format!(
1755 "No self-hosted runner with labels {} took it within {} hours.",
1756 Wanted::from_stored(&stored).describe(),
1757 model::MAX_WAIT_HOURS
1758 );
1759 self.finish_job(&job.id, "failure", Some(&why), None).await?;
1760 }
1761 let tasks = self
1762 .db
1763 .prepare(
1764 "SELECT * FROM runner_tasks WHERE (status = 'queued' AND created_at < ?) OR (status = 'in_progress' AND seen_at < ?) LIMIT 50",
1765 )
1766 .bind(&[before(waited).into(), before(SILENT_MS).into()])?
1767 .all()
1768 .await?
1769 .results::<TaskRow>()?;
1770 for task in tasks {
1771 let why = if task.status == "queued" {
1772 format!("No self-hosted runner with labels {} took it within {} hours.", Wanted::of(&serde_json::from_str::<Vec<String>>(&task.labels).unwrap_or_default(), None).describe(), model::MAX_WAIT_HOURS)
1773 } else {
1774 format!("The self-hosted runner {} stopped answering.", task.runner_name.clone().unwrap_or_default())
1775 };
1776 self.end_task(&task, 1, &why).await?;
1777 }
1778 let forgotten = self
1779 .db
1780 .prepare("SELECT * FROM runners WHERE COALESCE(last_seen_at, created_at) < ? LIMIT 50")
1781 .bind(&[before(FORGET_OFFLINE_MS).into()])?
1782 .all()
1783 .await?
1784 .results::<RunnerRow>()?;
1785 for row in forgotten {
1786 self.forget_runner(&row, "The runner was offline for 14 days.").await?;
1787 }
1788 // An ephemeral runner that took its job and then went quiet.
1789 let spent = self
1790 .db
1791 .prepare("SELECT * FROM runners WHERE ephemeral = 1 AND spent = 1 AND work_id IS NULL AND COALESCE(last_seen_at, created_at) < ? LIMIT 50")
1792 .bind(&[before(60 * 60 * 1000).into()])?
1793 .all()
1794 .await?
1795 .results::<RunnerRow>()?;
1796 for row in spent {
1797 self.forget_runner(&row, "The ephemeral runner finished its job.").await?;
1798 }
1799 self.db
1800 .prepare("DELETE FROM runner_registrations WHERE expires_at < ?")
1801 .bind(&[before(24 * 60 * 60 * 1000).into()])?
1802 .run()
1803 .await?;
1804 Ok(())
1805 }
1806}
1807
1808#[derive(Deserialize)]
1809struct IdRow {
1810 id: String,
1811}
1812
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.1813/// Whether `runner` is online and could take a public repository's job:
1814/// not host-mode, and its repository, group and labels let it.
1815fn takes_public_job(runner: &RunnerRow, groups: &[GroupRow], repo_id: &str, run_repo: &str, wanted: &Wanted, at: u64) -> bool {
1816 let group = runner.group_id.as_ref().and_then(|id| groups.iter().find(|g| &g.id == id));
1817 runner.online(at)
1818 && model::isolation_allows(runner.isolation.as_deref(), Some("public"))
1819 && runner.repo_id.as_deref().is_none_or(|repo| repo == repo_id)
1820 && (runner.repo_id.is_some() || group.is_none_or(|g| g.allows(run_repo)))
1821 && wanted.matches(&runner.labels(), group.map(|g| g.name.as_str()))
1822}
1823
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1824fn group_view(group: &GroupRow, runners: &[RunnerRow]) -> RunnerGroup {
1825 RunnerGroup {
1826 id: group.id.clone(),
1827 name: group.name.clone(),
1828 default: group.is_default != 0,
1829 repositories: group.repositories(),
1830 runners: runners.iter().filter(|r| r.group_id.as_deref() == Some(group.id.as_str())).count() as u32,
1831 updated_at: group.updated_at.clone(),
1832 }
1833}
1834
1835#[cfg(test)]
1836mod tests {
1837 use super::*;
1838
1839 fn group(repositories: &[&str]) -> GroupRow {
1840 GroupRow {
1841 id: "rng_1".into(),
1842 name: "Default".into(),
1843 is_default: 1,
1844 repositories: serde_json::to_string(&repositories).unwrap(),
1845 updated_at: "2026-10-06T00:00:00Z".into(),
1846 }
1847 }
1848
1849 #[test]
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1850 fn being_seen_is_written_once_a_minute() {
1851 let at = 2_000_000_000_000;
1852 assert!(seen_is_stale(None, at));
1853 assert!(!seen_is_stale(Some(&rfc3339(at - 59_000)), at));
1854 assert!(seen_is_stale(Some(&rfc3339(at - 61_000)), at));
1855 // A runner written down at the last moment is still online until the
1856 // next write, a poll later.
1857 assert!(SEEN_RESOLUTION_MS + model::MAX_POLL_WAIT_MS < model::ONLINE_WITHIN_MS);
1858 assert!(SEEN_RESOLUTION_MS < crate::SILENT_MS);
1859 }
1860
1861 #[test]
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1862 fn a_group_lets_its_repositories_use_it() {
1863 assert!(group(&[]).allows("acme/web"));
1864 assert!(group(&["web", "api"]).allows("acme/WEB"));
1865 assert!(group(&["web"]).allows("web"));
1866 assert!(!group(&["web"]).allows("acme/docs"));
1867 }
1868
1869 #[test]
1870 fn names_are_checked() {
1871 assert_eq!(valid_name(" build-01 ").unwrap(), "build-01");
1872 assert!(valid_name("").is_err());
1873 assert!(valid_name("has space").is_err());
1874 assert!(valid_name(&"x".repeat(65)).is_err());
1875 assert!(valid_group_name("GPU machines").is_ok());
1876 assert!(valid_group_name("").is_err());
1877 }
1878
1879 #[test]
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.1880 fn only_a_sandboxed_runner_that_could_take_it_serves_a_public_job() {
1881 let at = 1_000_000_000;
1882 let runner = |isolation: Option<&str>| RunnerRow {
1883 id: "rnr_1".into(),
1884 workspace: "acme".into(),
1885 repo_id: None,
1886 repo: None,
1887 group_id: None,
1888 name: "a".into(),
1889 labels: r#"["self-hosted","linux","x64"]"#.into(),
1890 os: "linux".into(),
1891 arch: "x64".into(),
1892 version: String::new(),
1893 ephemeral: 0,
1894 credential_hash: String::new(),
1895 previous_hash: None,
1896 rotated_at: String::new(),
1897 work_id: None,
1898 work_kind: None,
1899 spent: 0,
1900 last_seen_at: Some(rfc3339(at)),
1901 created_at: String::new(),
1902 created_by: None,
1903 isolation: isolation.map(str::to_owned),
1904 };
1905 let wanted = Wanted::of(&["self-hosted".to_owned(), "linux".to_owned()], None);
1906 let takes = |row: &RunnerRow| takes_public_job(row, &[], "rep_1", "acme/web", &wanted, at);
1907 assert!(!takes(&runner(Some("host"))));
1908 assert!(takes(&runner(Some("container"))));
1909 assert!(takes(&runner(Some("nested-container"))));
1910 assert!(takes(&runner(None)), "an older runner, which refuses for itself");
1911 assert!(!takes(&RunnerRow { repo_id: Some("rep_2".into()), ..runner(None) }), "another repository's");
1912 assert!(!takes(&RunnerRow { labels: r#"["self-hosted","macos"]"#.into(), ..runner(None) }));
1913 assert!(!takes(&RunnerRow { last_seen_at: None, ..runner(None) }), "offline");
1914 }
1915
1916 #[test]
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1917 fn online_means_seen_in_the_last_ninety_seconds() {
1918 let row = RunnerRow {
1919 id: "rnr_1".into(),
1920 workspace: "acme".into(),
1921 repo_id: None,
1922 repo: None,
1923 group_id: None,
1924 name: "a".into(),
1925 labels: "[]".into(),
1926 os: "linux".into(),
1927 arch: "x64".into(),
1928 version: String::new(),
1929 ephemeral: 0,
1930 credential_hash: String::new(),
1931 previous_hash: None,
1932 rotated_at: String::new(),
1933 work_id: None,
1934 work_kind: None,
1935 spent: 0,
1936 last_seen_at: Some(rfc3339(1_000_000_000)),
1937 created_at: String::new(),
1938 created_by: None,
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.1939 isolation: None,
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1940 };
1941 assert!(row.online(1_000_000_000 + 60_000));
1942 assert!(!row.online(1_000_000_000 + 120_000));
1943 assert_eq!(row.workspace, "acme");
1944 }
1945}

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