Runners: registration, Admin -> Runners, history, drain/disable/remove b6b7e4cf6b24

-- Runners (hub spec §13.2): registered from the UI with a one-time token,

-- or configured with a hub.toml token (tracked here from their first

-- claim). A runner signs receipts with one key: bound at registration, or

-- on the first signed receipt of a configured runner.

CREATE TABLE flow_runners (

    id INTEGER PRIMARY KEY,

    name TEXT NOT NULL,

    token_hash TEXT NOT NULL UNIQUE,

    -- 'registered' or 'config'

    source TEXT NOT NULL,

    key TEXT NOT NULL DEFAULT '',

    key_name TEXT NOT NULL DEFAULT '',

    arch TEXT NOT NULL DEFAULT '',

    os TEXT NOT NULL DEFAULT '',

    version TEXT NOT NULL DEFAULT '',

    -- comma-separated

    labels TEXT NOT NULL DEFAULT '',

    max_jobs INTEGER NOT NULL DEFAULT 1,

    memory_mib INTEGER,

    cpus INTEGER,

    -- 'active', 'draining' (finishes its jobs, claims no more) or 'disabled'

    state TEXT NOT NULL DEFAULT 'active',

    created_ms INTEGER NOT NULL,

    created_by TEXT NOT NULL DEFAULT '',

    last_seen_ms INTEGER

);

-- Keys already bound to configured runners move over.

INSERT INTO flow_runners (name, token_hash, source, key, key_name, created_ms)

    SELECT name, token_hash, 'config', key, name, first_ms FROM flow_runner_keys;

CREATE TABLE flow_runner_registrations (

    token_hash TEXT PRIMARY KEY,

    created_by TEXT NOT NULL,

    created_ms INTEGER NOT NULL,

    expires_ms INTEGER NOT NULL,

    used_ms INTEGER,

    runner_id INTEGER

);

ALTER TABLE flow_jobs ADD COLUMN runner_id INTEGER;

CREATE INDEX flow_jobs_runner ON flow_jobs(runner_id);

//! OxygenFlow runners (hub spec §13.2): who may claim jobs, what they are,

//! whether they are taking work, and what they have run.

use super::{now_ms, Pool};

use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]

pub struct Runner {

    pub id: i64,

    pub name: String,

    pub token_hash: String,

    pub source: String,

    pub key: String,

    pub key_name: String,

    pub arch: String,

    pub os: String,

    pub version: String,

    pub labels: String,

    pub max_jobs: i64,

    pub memory_mib: Option<i64>,

    pub cpus: Option<i64>,

    pub state: String,

    pub created_ms: i64,

    pub created_by: String,

    pub last_seen_ms: Option<i64>,

}

/// A runner that has not polled for this long is shown offline.

pub const ONLINE_MS: i64 = 90_000;

impl Runner {

    pub fn online(&self) -> bool {

        self.last_seen_ms.is_some_and(|t| now_ms() - t < ONLINE_MS)

    }

    /// `online`, `offline`, `draining` or `disabled`.

    pub fn status(&self) -> &'static str {

        match self.state.as_str() {

            "disabled" => "disabled",

            "draining" => "draining",

            _ if self.online() => "online",

            _ => "offline",

        }

    }

    pub fn label_list(&self) -> Vec<&str> {

        self.labels.split(',').filter(|l| !l.is_empty()).collect()

    }

}

pub async fn by_token(pool: &Pool, token_hash: &str) -> sqlx::Result<Option<Runner>> {

    sqlx::query_as("SELECT * FROM flow_runners WHERE token_hash = ?").bind(token_hash).fetch_optional(pool).await

}

pub async fn get(pool: &Pool, id: i64) -> sqlx::Result<Option<Runner>> {

    sqlx::query_as("SELECT * FROM flow_runners WHERE id = ?").bind(id).fetch_optional(pool).await

}

pub async fn list(pool: &Pool) -> sqlx::Result<Vec<Runner>> {

    sqlx::query_as("SELECT * FROM flow_runners ORDER BY name, id").fetch_all(pool).await

}

/// The row of a runner configured with a hub.toml token, made on first sight.

pub async fn configured(pool: &Pool, token_hash: &str, name: &str) -> sqlx::Result<Runner> {

    sqlx::query("INSERT OR IGNORE INTO flow_runners (name, token_hash, source, created_ms) VALUES (?, ?, 'config', ?)")

        .bind(name)

        .bind(token_hash)

        .bind(now_ms())

        .execute(pool)

        .await?;

    Ok(by_token(pool, token_hash).await?.expect("just made"))

}

/// What a claim says about the runner, and that it was seen now. A

/// configured runner takes the name it claims with.

pub async fn seen(pool: &Pool, id: i64, name: Option<&str>, info: Option<&flow_core::api::RunnerInfo>) -> sqlx::Result<()> {

    let now = now_ms();

    match info {

        Some(i) => {

            sqlx::query(

                "UPDATE flow_runners SET arch = ?, os = ?, version = ?, labels = ?, max_jobs = ?, memory_mib = ?, cpus = ?, last_seen_ms = ? WHERE id = ?",

            )

            .bind(&i.arch)

            .bind(&i.os)

            .bind(&i.version)

            .bind(i.labels.join(","))

            .bind(i.max_jobs as i64)

            .bind(i.memory_mib.map(i64::from))

            .bind(i.cpus.map(i64::from))

            .bind(now)

            .bind(id)

            .execute(pool)

            .await?;

        }

        None => {

            sqlx::query("UPDATE flow_runners SET last_seen_ms = ? WHERE id = ?").bind(now).bind(id).execute(pool).await?;

        }

    }

    if let Some(n) = name {

        sqlx::query("UPDATE flow_runners SET name = ? WHERE id = ? AND source = 'config'").bind(n).bind(id).execute(pool).await?;

    }

    Ok(())

}

pub async fn set_state(pool: &Pool, id: i64, state: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET state = ? WHERE id = ?").bind(state).bind(id).execute(pool).await?;

    Ok(())

}

/// Forget a runner: its credential stops working. Its jobs keep its name.

pub async fn remove(pool: &Pool, id: i64) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM flow_runners WHERE id = ?").bind(id).execute(pool).await?;

    Ok(())

}

/// The key a runner signs receipts with: the one bound, or `key` (named

/// `name`) when none is yet.

pub async fn bind_key(pool: &Pool, id: i64, key: &str, name: &str) -> sqlx::Result<String> {

    sqlx::query("UPDATE flow_runners SET key = ?, key_name = ? WHERE id = ? AND key = ''")

        .bind(key)

        .bind(name)

        .bind(id)

        .execute(pool)

        .await?;

    sqlx::query_scalar("SELECT key FROM flow_runners WHERE id = ?").bind(id).fetch_one(pool).await

}

/// Keys that runners of this hub sign with.

pub async fn keys(pool: &Pool) -> sqlx::Result<Vec<String>> {

    sqlx::query_scalar("SELECT key FROM flow_runners WHERE key != ''").fetch_all(pool).await

}

/// A one-time registration token, valid for a day.

pub async fn add_registration(pool: &Pool, token_hash: &str, by: &str) -> sqlx::Result<()> {

    let now = now_ms();

    sqlx::query("INSERT INTO flow_runner_registrations (token_hash, created_by, created_ms, expires_ms) VALUES (?, ?, ?, ?)")

        .bind(token_hash)

        .bind(by)

        .bind(now)

        .bind(now + 24 * 3_600_000)

        .execute(pool)

        .await?;

    Ok(())

}

/// Registration tokens not used yet and not expired: (created by, created, expires).

pub async fn pending_registrations(pool: &Pool) -> sqlx::Result<Vec<(String, i64, i64)>> {

    sqlx::query_as("SELECT created_by, created_ms, expires_ms FROM flow_runner_registrations WHERE used_ms IS NULL AND expires_ms > ? ORDER BY created_ms DESC")

        .bind(now_ms())

        .fetch_all(pool)

        .await

}

/// Use registration token `token_hash` to register a runner: `None` when

/// the token is unknown, used or expired.

#[allow(clippy::too_many_arguments)]

pub async fn register(

    pool: &Pool,

    registration: &str,

    name: &str,

    token_hash: &str,

    key: &str,

    key_name: &str,

    by_default: &str,

) -> sqlx::Result<Option<Runner>> {

    let mut tx = super::begin_write(pool).await?;

    let now = now_ms();

    let who: Option<String> =

        sqlx::query_scalar("SELECT created_by FROM flow_runner_registrations WHERE token_hash = ? AND used_ms IS NULL AND expires_ms > ?")

            .bind(registration)

            .bind(now)

            .fetch_optional(&mut *tx)

            .await?;

    let Some(who) = who else { return Ok(None) };

    let id: i64 = sqlx::query_scalar(

        "INSERT INTO flow_runners (name, token_hash, source, key, key_name, created_ms, created_by) VALUES (?, ?, 'registered', ?, ?, ?, ?) RETURNING id",

    )

    .bind(name)

    .bind(token_hash)

    .bind(key)

    .bind(key_name)

    .bind(now)

    .bind(if who.is_empty() { by_default } else { &who })

    .fetch_one(&mut *tx)

    .await?;

    sqlx::query("UPDATE flow_runner_registrations SET used_ms = ?, runner_id = ? WHERE token_hash = ?")

        .bind(now)

        .bind(id)

        .bind(registration)

        .execute(&mut *tx)

        .await?;

    tx.commit().await?;

    get(pool, id).await

}

/// A job a runner ran, for its history.

#[derive(Clone, Debug, FromRow)]

pub struct RunnerJob {

    pub owner: String,

    pub repo: String,

    pub run_number: i64,

    pub job: String,

    pub status: String,

    pub started_ms: Option<i64>,

    pub finished_ms: Option<i64>,

    pub receipt_attest: String,

}

/// Jobs runner `id` ran or is running, newest first.

pub async fn history(pool: &Pool, id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<RunnerJob>> {

    sqlx::query_as(

        "SELECT p.owner AS owner, p.name AS repo, r.number AS run_number, j.name AS job, j.status AS status, j.started_ms AS started_ms, \

         j.finished_ms AS finished_ms, j.receipt_attest AS receipt_attest \

         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \

         WHERE j.runner_id = ? AND (? IS NULL OR j.started_ms < ?) ORDER BY j.started_ms DESC LIMIT ?",

    )

    .bind(id)

    .bind(before)

    .bind(before)

    .bind(limit)

    .fetch_all(pool)

    .await

}

/// (jobs run, passed, busy milliseconds) of runner `id`.

pub async fn totals(pool: &Pool, id: i64) -> sqlx::Result<(i64, i64, i64)> {

    sqlx::query_as(

        "SELECT COUNT(*), COALESCE(SUM(status = 'passed'), 0), COALESCE(SUM(finished_ms - started_ms), 0) \

         FROM flow_jobs WHERE runner_id = ? AND finished_ms IS NOT NULL",

    )

    .bind(id)

    .fetch_one(pool)

    .await

}

/// Jobs running on each runner now, by runner id.

pub async fn running(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {

    sqlx::query_as("SELECT runner_id, COUNT(*) FROM flow_jobs WHERE status = 'running' AND runner_id IS NOT NULL GROUP BY runner_id")

        .fetch_all(pool)

        .await

}

pub async fn set_job_runner(pool: &Pool, job_id: i64, runner_id: i64) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET runner_id = ? WHERE id = ?").bind(runner_id).bind(job_id).execute(pool).await?;

    Ok(())

}

//! Admin → Runners (hub spec §13.2): the OxygenFlow runners of this hub,

//! what they are and are running, their history, registration tokens, and

//! draining, disabling and removing them.

use super::settings::csrf;

use super::{filters, Error, Page, Shell};

use crate::auth::Viewer;

use crate::db::{self, accounts, runners};

use crate::Shared;

use askama::Template;

use axum::extract::{Form, Path, Query, State};

use axum::response::{IntoResponse, Redirect, Response};

use serde::Deserialize;

use std::collections::HashMap;

fn require_admin(viewer: &Viewer, shell: &Shell) -> Result<accounts::Account, Error> {

    match viewer.account() {

        Some(a) if a.is_admin => Ok(a.clone()),

        _ => Err(Error::not_found(shell.clone())),

    }

}

pub struct RunnerRow {

    pub r: runners::Runner,

    /// Jobs it is running now.

    pub running: i64,

}

#[derive(Template)]

#[template(path = "admin_runners.html")]

pub struct RunnersPage {

    shell: Shell,

    runners: Vec<RunnerRow>,

    /// Registration tokens waiting to be used: (created by, created, expires).

    pending: Vec<(String, i64, i64)>,

    /// A token just made, shown once, with the command that uses it.

    new_token: Option<String>,

    hub: String,

    notice: Option<String>,

    error: Option<String>,

}

async fn page(

    app: &Shared,

    shell: Shell,

    new_token: Option<String>,

    notice: Option<String>,

    error: Option<String>,

) -> Result<Page<RunnersPage>, Error> {

    let e = |e| Error::from_db(shell.clone(), e);

    let list = runners::list(&app.db).await.map_err(e)?;

    let running: HashMap<i64, i64> = runners::running(&app.db).await.map_err(e)?.into_iter().collect();

    let pending = runners::pending_registrations(&app.db).await.map_err(e)?;

    let rows = list.into_iter().map(|r| RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), r }).collect();

    Ok(Page(RunnersPage { hub: app.config.base_url().to_string(), shell, runners: rows, pending, new_token, notice, error }))

}

/// `GET /admin/runners`.

pub async fn list(State(app): State<Shared>, viewer: Viewer) -> Result<Page<RunnersPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners", &viewer);

    require_admin(&viewer, &shell)?;

    page(&app, shell, None, None, None).await

}

#[derive(Deserialize)]

pub struct CsrfOnly {

    #[serde(rename = "_csrf")]

    csrf_token: String,

}

/// `POST /admin/runners/token`: a one-time registration token, shown once.

pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<CsrfOnly>) -> Result<Page<RunnersPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let token = format!("flowreg_{}", crate::auth::random_hex(24));

    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", "registration token created").await.ok();

    page(&app, shell, Some(token), None, None).await

}

#[derive(Deserialize)]

pub struct StateForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// `active`, `draining` or `disabled`.

    state: String,

}

/// `POST /admin/runners/{id}/state`: drain, disable or resume a runner.

pub async fn set_state(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<StateForm>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, "/admin/runners", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    if !matches!(f.state.as_str(), "active" | "draining" | "disabled") {

        return Err(Error::bad(shell, "a runner is active, draining or disabled"));

    }

    let r =

        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;

    runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok();

    Ok(Redirect::to("/admin/runners").into_response())

}

/// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing.

pub async fn remove(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<CsrfOnly>,

) -> Result<Page<RunnersPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;

    let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0);

    if busy {

        return page(

            &app,

            shell,

            None,

            None,

            Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)),

        )

        .await;

    }

    runners::remove(&app.db, id).await.map_err(e)?;

    accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok();

    let also = if r.source == "config" {

        " It is configured in hub.toml; remove its token there too, or it comes back on its next claim."

    } else {

        ""

    };

    page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await

}

#[derive(Template)]

#[template(path = "admin_runner.html")]

pub struct RunnerPage {

    shell: Shell,

    r: runners::Runner,

    running: i64,

    history: Vec<runners::RunnerJob>,

    older: Option<i64>,

    jobs: i64,

    passed: i64,

    busy_ms: i64,

}

impl RunnerPage {

    /// Time spent running jobs, as pages show durations.

    pub fn busy(&self) -> String {

        crate::flow::human_ms(self.busy_ms)

    }

}

#[derive(Deserialize, Default)]

pub struct HistoryQuery {

    before: Option<i64>,

}

const PER_PAGE: i64 = 50;

/// `GET /admin/runners/{id}`: a runner and the jobs it ran.

pub async fn show(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Query(q): Query<HistoryQuery>,

) -> Result<Page<RunnerPage>, Error> {

    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);

    require_admin(&viewer, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;

    let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?;

    let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten();

    history.truncate(PER_PAGE as usize);

    let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?;

    let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0);

    Ok(Page(RunnerPage { shell, r, running, history, older, jobs, passed, busy_ms }))

}

pub fn duration(j: &runners::RunnerJob) -> String {

    match (j.started_ms, j.finished_ms) {

        (Some(a), Some(b)) => crate::flow::human_ms(b - a),

        (Some(a), None) => crate::flow::human_ms(db::now_ms() - a),

        _ => "–".into(),

    }

}

{% extends "base.html" %}

{% block title %}{{ r.name }} · Runners{% endblock %}

{% block mainclass %} class="wide"{% endblock %}

{% block content %}

<div class="head"><h1>{{ r.name }} <small><span class="runner-status st-{{ r.status() }}">{{ r.status() }}</span> · <a href="/admin/runners">Runners</a></small></h1></div>

<section class="panel">

  <dl class="facts">

    <dt>Signing key</dt><dd>{% if r.key.is_empty() %}<span class="mute">bound on its first signed receipt</span>{% else %}<code class="full">{{ r.key }}</code> {{ r.key_name }}{% endif %}</dd>

    <dt>Machine</dt><dd>{% if r.arch.is_empty() %}<span class="mute">not reported yet</span>{% else %}{{ r.arch }} · {{ r.os }} · flow {{ r.version }}{% endif %}</dd>

    <dt>Labels</dt><dd>{% for l in r.label_list() %}<span class="lbl">{{ l }}</span> {% endfor %}{% if r.label_list().is_empty() %}<span class="mute">none</span>{% endif %}</dd>

    <dt>Capacity</dt><dd>{{ r.max_jobs|plural("job", "jobs") }} at once{% if let Some(m) = r.memory_mib %}, {{ m }} MiB{% endif %}{% if let Some(c) = r.cpus %}, {{ c }} vCPU{% endif %}</dd>

    <dt>Running now</dt><dd>{{ running|plural("job", "jobs") }}</dd>

    <dt>Ran</dt><dd>{{ jobs|plural("job", "jobs") }}, {{ passed }} passed{% if jobs > 0 %} ({{ passed * 100 / jobs }}%){% endif %}, busy {{ self.busy() }}</dd>

    <dt>Registered</dt><dd>{% if r.source == "config" %}configured in hub.toml{% else %}by {{ r.created_by }}{% endif %}, <time title="{{ r.created_ms|when_i }}">{{ r.created_ms|ago_i }}</time></dd>

    <dt>Last seen</dt><dd>{% if let Some(t) = r.last_seen_ms %}<time title="{{ t|when_i }}">{{ t|ago_i }}</time>{% else %}never{% endif %}</dd>

  </dl>

</section>

<section class="panel">

  <h2>Jobs</h2>

  {% if history.is_empty() %}<p class="mute">No jobs yet.</p>{% else %}

  <div class="table"><table>

  <tr><th>Repository</th><th>Run</th><th>Job</th><th>Status</th><th>Duration</th><th>Started</th><th>Receipt</th></tr>

  {% for j in history %}<tr>

    <td><a href="/{{ j.owner }}/{{ j.repo }}">{{ j.owner }}/{{ j.repo }}</a></td>

    <td><a href="/{{ j.owner }}/{{ j.repo }}/flow/{{ j.run_number }}">#{{ j.run_number }}</a></td>

    <td>{{ j.job }}</td>

    <td><span class="flow-glyph st-{{ j.status }}">{{ crate::web::flow::glyph(j.status) }}</span> {{ j.status }}</td>

    <td>{{ crate::web::runners::duration(j) }}</td>

    <td class="small">{% if let Some(t) = j.started_ms %}<time title="{{ t|when_i }}">{{ t|ago_i }}</time>{% endif %}</td>

    <td class="small">{% if !j.receipt_attest.is_empty() %}<a href="/{{ j.owner }}/{{ j.repo }}/receipts/{{ j.receipt_attest }}">signed</a>{% endif %}</td>

  </tr>{% endfor %}

  </table></div>

  {% if let Some(b) = older %}<p><a href="?before={{ b }}">Older jobs</a></p>{% endif %}

  {% endif %}

</section>

{% endblock %}

{% extends "base.html" %}

{% block title %}Runners · Administration{% endblock %}

{% block mainclass %} class="wide"{% endblock %}

{% block content %}

<div class="head"><h1>Runners <small>{{ runners.len()|plural("runner", "runners") }} · <a href="/admin">Administration</a></small></h1></div>

{% if let Some(n) = notice %}<p class="note">{{ n }}</p>{% endif %}

{% if let Some(e) = error %}<p class="error">{{ e }}</p>{% endif %}

{% if let Some(t) = new_token %}

<section class="panel strip ok">

  <h2>Registration token</h2>

  <p>Use it once, within a day, on the machine that will run jobs. It is shown only now.</p>

  <pre class="snippet">flow runner register {{ hub }} --token {{ t }} --name NAME [--ca hub-ca.pem] [--label LABEL]

flow runner</pre>

  <p class="small mute">The runner sends its signing identity when it registers; this hub accepts receipts from it signed by that key only.</p>

</section>

{% endif %}

<section class="panel">

  {% if runners.is_empty() %}<p class="mute">No runners yet. Create a registration token below and run <code>flow runner register</code> on the runner.</p>{% else %}

  <div class="table"><table>

  <tr><th>Runner</th><th>Status</th><th>Signing key</th><th>Machine</th><th>Labels</th><th>Capacity</th><th>Running</th><th>Last seen</th><th></th></tr>

  {% for row in runners %}{% let r = row.r %}

  <tr>

    <td><a href="/admin/runners/{{ r.id }}"><strong>{{ r.name }}</strong></a>{% if r.source == "config" %} <span class="tag" title="Configured with a hub.toml token">hub.toml</span>{% endif %}</td>

    <td><span class="runner-status st-{{ r.status() }}">{{ r.status() }}</span></td>

    <td>{% if r.key.is_empty() %}<span class="mute" title="Bound on its first signed receipt">not yet</span>{% else %}<code title="{{ r.key }} · {{ r.key_name }}">{{ r.key|short }}</code>{% endif %}</td>

    <td class="small">{% if r.arch.is_empty() %}<span class="mute">unknown</span>{% else %}{{ r.arch }} · {{ r.os }} · flow {{ r.version }}{% endif %}</td>

    <td class="small">{% for l in r.label_list() %}<span class="lbl">{{ l }}</span> {% endfor %}</td>

    <td class="small">{{ r.max_jobs|plural("job", "jobs") }}{% if let Some(m) = r.memory_mib %} · {{ m }} MiB{% endif %}{% if let Some(c) = r.cpus %} · {{ c }} vCPU{% endif %}</td>

    <td>{{ row.running }}</td>

    <td class="small">{% if let Some(t) = r.last_seen_ms %}<time title="{{ t|when_i }}">{{ t|ago_i }}</time>{% else %}<span class="mute">never</span>{% endif %}</td>

    <td><form class="tiny" method="post" action="/admin/runners/{{ r.id }}/state"><input type="hidden" name="_csrf" value="{{ shell.csrf }}">

      {% if r.state == "active" %}<button name="state" value="draining" title="Finish current jobs, take no new ones">drain</button> <button name="state" value="disabled" title="Refuse its requests at once">disable</button>{% else %}<button name="state" value="active">resume</button>{% endif %}</form>

      <form class="tiny" method="post" action="/admin/runners/{{ r.id }}/remove" onsubmit="return confirm('Remove {{ r.name }}? Its credential stops working; its jobs keep its name.')"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button class="danger-btn">remove</button></form></td>

  </tr>

  {% endfor %}

  </table></div>

  {% endif %}

</section>

<section class="panel">

  <h2>Add a runner</h2>

  <p class="small">A runner registers with a one-time token: it sends its signing identity, architecture and labels, and gets its own credential, kept in <code>runner.toml</code> beside its <code>flow.toml</code>.</p>

  <form method="post" action="/admin/runners/token"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" class="primary-btn">Create a registration token</button></form>

  {% if !pending.is_empty() %}<p class="small mute">Unused tokens: {% for (by, made, expires) in pending %}by {{ by }} {{ made|ago_i }}, expires {{ expires|when_i }}{% if !loop.last %}; {% endif %}{% endfor %}</p>{% endif %}

</section>

{% endblock %}

        .route("/flow/runners/register", post(crate::flow::register))

pub mod runners;

/// The runner a bearer token belongs to: a registered runner, or one

/// configured with a hub.toml token (tracked from its first request). A

/// disabled runner is refused.

async fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<db::runners::Runner, ApiError> {

    };

    let hash = blake3::hash(t.as_bytes()).to_hex().to_string();

    let runner = match db::runners::by_token(&app.db, &hash).await? {

        Some(r) => r,

        None if app.config.flow.runner_tokens.iter().any(|x| crate::auth::mac_equal(x, t)) => {

            db::runners::configured(&app.db, &hash, "runner").await?

        }

        None => return Err(ApiError(StatusCode::UNAUTHORIZED, "not a runner token".into())),

    if runner.state == "disabled" {

        return Err(ApiError(StatusCode::FORBIDDEN, format!("runner {} is disabled on this hub", runner.name)));

    Ok(runner)

/// `POST /api/v1/flow/runners/register`: a one-time registration token

/// from Admin → Runners buys a runner credential, binding the runner's

/// signing identity (hub spec §13.2).

pub async fn register(

    State(app): State<Shared>,

    Json(req): Json<flow_core::api::Register>,

) -> Result<Json<flow_core::api::Registered>, ApiError> {

    let bytes = hex::decode(req.identity.trim()).map_err(|_| ApiError(StatusCode::BAD_REQUEST, "identity: not hex".into()))?;

    let identity = oxy_core::identity::Identity::decode(&bytes).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("identity: {e}")))?;

    let name: String = req.name.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).take(64).collect();

    if name.is_empty() {

        return Err(ApiError(StatusCode::BAD_REQUEST, "give the runner a name of letters, digits, - _ or .".into()));

    }

    let credential = format!("flowrunner_{}", crate::auth::random_hex(32));

    let hash = blake3::hash(credential.as_bytes()).to_hex().to_string();

    let reg = blake3::hash(req.token.trim().as_bytes()).to_hex().to_string();

    let runner =

        db::runners::register(&app.db, &reg, &name, &hash, &identity.hash().to_hex(), &identity.name, "").await?.ok_or_else(|| {

            ApiError(StatusCode::FORBIDDEN, "that registration token is unknown, used or expired; create one on Admin → Runners".into())

        })?;

    db::runners::seen(&app.db, runner.id, None, Some(&req.info)).await?;

    tracing::info!("flow: runner {} registered (key {})", runner.name, &runner.key[..12]);

    Ok(Json(flow_core::api::Registered { id: runner.id, name: runner.name, credential }))

}

    let me = runner_auth(&app, &headers).await?;

    let claimed: String = req.runner.chars().filter(|c| !c.is_control()).take(64).collect();

    db::runners::seen(&app.db, me.id, Some(&claimed), req.info.as_ref()).await?;

    // A draining runner finishes what it has and takes nothing new.

    if me.state == "draining" {

        return Ok(StatusCode::NO_CONTENT.into_response());

    }

    let runner = if me.source == "registered" { me.name.clone() } else { claimed };

    db::runners::set_job_runner(&app.db, job.id, me.id).await?;

    runner_auth(&app, &headers).await?;

    runner_auth(&app, &headers).await?;

    let me = runner_auth(&app, &headers).await?;

                let (attest, signer, note) = match keep_receipt(&app, &me, &row, &run, r, signed).await {

    runner: &db::runners::Runner,

    let bound = db::runners::bind_key(&app.db, runner.id, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;

pub mod runners;

        .route("/admin/runners", get(runners::list))

        .route("/admin/runners/token", post(runners::new_token))

        .route("/admin/runners/{id}", get(runners::show))

        .route("/admin/runners/{id}/state", post(runners::set_state))

        .route("/admin/runners/{id}/remove", post(runners::remove))

    let known_runner_keys: Vec<String> = db::runners::keys(&app.db).await.unwrap_or_default();

/// Runners (hub spec §13.2): an administrator makes a registration token,

/// a runner registers with it (binding its key), claims, shows on Admin →

/// Runners with its history, and is drained, disabled and removed.

#[tokio::test(flavor = "multi_thread")]

async fn runners_register_show_their_history_and_are_drained_and_removed() {

    use oxy_core::identity::SecretKey;

    let h = hub().await;

    create(&h.app, "andy", "demo").await;

    // A signed-in site administrator.

    let res = oxygenhub::router(h.app.clone())

        .oneshot(

            Request::post("/register")

                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery"))

                .unwrap(),

        )

        .await

        .unwrap();

    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();

    let admin = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();

    db::accounts::set_admin(&h.app.db, admin.id, true).await.unwrap();

    let web = |method: &'static str, path: String, body: Option<String>| {

        let (app, cookie) = (h.app.clone(), cookie.clone());

        async move {

            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);

            if body.is_some() {

                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");

            }

            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();

            let st = res.status();

            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

        }

    };

    let csrf_of = |page: &str| {

        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();

        page[i..].split('"').next().unwrap().to_string()

    };

    let (st, page) = web("GET", "/admin/runners".into(), None).await;

    assert_eq!(st, StatusCode::OK);

    assert!(page.contains("No runners yet"));

    assert_eq!(call(&h.app, "GET", "/admin/runners", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    // A one-time registration token, shown once with the command.

    let (_, page) = web("POST", "/admin/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;

    let i = page.find("--token ").unwrap() + "--token ".len();

    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();

    assert!(token.starts_with("flowreg_"), "{token}");

    // The runner registers, binding its key; the token then is spent.

    let sk = SecretKey::generate();

    let identity = sk.identity("plex (OxygenFlow runner)", "", 1);

    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": ["big-disk"], "max_jobs": 2, "memory_mib": 16384, "cpus": 4 });

    let reg = serde_json::json!({ "token": token, "name": "plex", "identity": hex::encode(identity.encode()), "info": info });

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();

    assert!(credential.starts_with("flowrunner_"));

    assert_eq!(

        call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0,

        StatusCode::FORBIDDEN,

        "used once"

    );

    // It claims with its credential; the hub.toml runner shows too.

    let claim_as = |who: String| {

        let app = h.app.clone();

        let info = info.clone();

        async move {

            let body = serde_json::json!({ "runner": "ignored", "runtimes": ["microvm"], "info": info }).to_string();

            call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await

        }

    };

    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "nothing queued");

    assert_eq!(claim(&h.app).await, None);

    let (_, page) = web("GET", "/admin/runners".into(), None).await;

    assert!(

        page.contains(">plex</strong>") && page.contains("st-online") && page.contains("x86_64 · linux") && page.contains("big-disk"),

        "{page}"

    );

    assert!(page.contains(">test-1</strong>") && page.contains("hub.toml"), "the configured runner, by its claimed name");

    let plex = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex").unwrap();

    assert_eq!(plex.key, identity.hash().to_hex(), "bound at registration");

    // A run: plex takes a job, and its history shows it.

    let url = format!("{}/andy/demo", h.base);

    let work = tempfile::tempdir().unwrap();

    let c = client(&url, work.path());

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    c.snapshot("test").unwrap();

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    runs_of(&h.app, "andy", "demo").await;

    let (st, body) = claim_as(credential.clone()).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let job: serde_json::Value = serde_json::from_str(&body).unwrap();

    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;

    assert!(

        page.contains("andy/demo") && page.contains(job["job"]["name"].as_str().unwrap()) && page.contains("Running now</dt><dd>1 job"),

        "{page}"

    );

    // Draining: no new work; removing while busy is refused.

    let (_, page) = web("GET", "/admin/runners".into(), None).await;

    let csrf = csrf_of(&page);

    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=draining"))).await;

    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "a draining runner takes nothing");

    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;

    assert!(page.contains("is running jobs: drain it first"), "{page}");

    // Its job finishes; disabled, it is refused; removed, its credential is dead.

    let id = job["job_id"].as_i64().unwrap();

    let done = call(

        &h.app,

        "POST",

        &format!("/api/v1/flow/jobs/{id}/finish"),

        Some(&credential),

        Some(r#"{"exit_code":0,"duration_ms":5}"#.into()),

    )

    .await;

    assert_eq!(done.0, StatusCode::NO_CONTENT);

    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=disabled"))).await;

    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::FORBIDDEN);

    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;

    assert!(page.contains("Removed plex."), "{page}");

    assert_eq!(claim_as(credential).await.0, StatusCode::UNAUTHORIZED);

}

- **Runners** (Admin → Runners, site administrators).

  - **Registering:**

    - An administrator creates a one-time registration token, valid for a day and shown once with the command that uses it.

    - On the runner, `flow runner register https://hub --token …` sends the runner's signing identity, architecture, OS, version and labels (`POST /api/v1/flow/runners/register`) and gets its own credential (`flowrunner_…`, stored hashed).

    - The runner's key is bound at registration: receipts from that runner signed by another key are not kept.

    - Runners configured with a `hub.toml` token still work. They are tracked from their first request under the name they claim with, and their key is bound on their first signed receipt.

  - **Each claim** carries the runner's description and marks it seen.

  - **The Runners page** lists each runner: online (seen in the last 90 s), offline, draining or disabled; signing key; machine; labels; capacity; jobs running now; and when it was last seen.

  - **Actions:**

    - **Drain:** the runner finishes its jobs and claims return nothing.

    - **Disable:** its requests are refused.

    - **Resume.**

    - **Remove:** refused while the runner has running jobs. Its credential stops working, and its jobs keep its name.

  - **A runner's page** shows its totals (jobs, pass rate, time busy) and its job history, newest first, with links to runs and signed receipts. Jobs record the runner that claimed them (`flow_jobs.runner_id`).

<p><a class="go small" href="/admin/storage">Storage and maintenance</a> <a class="go small" href="/admin/runners">Runners</a></p>

/* Runners (hub spec §13.2) */

.runner-status { font-size: 11px; font-weight: 600; text-transform: uppercase; letter-spacing: .06em; padding: 1px 6px; border: 1px solid var(--line); }

.runner-status.st-online { color: var(--ok); border-color: var(--ok); }

.runner-status.st-offline { color: var(--mute); }

.runner-status.st-draining { color: var(--warn); border-color: var(--warn); }

.runner-status.st-disabled { color: var(--bad); border-color: var(--bad); }