flagon-io/g1t

public

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