flagon-io/g1t

public

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

g1t.sh

git agents cloudflare rust forge mcp

2,035 linesCode BlameRaw

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