OXYGENHUB

Flow on your own runners 1870e4fa9808

A repository's maintainers may turn OxygenFlow on for it themselves once
it, or its owner, has a runner of its own: the new setting 'own' (on,
own runners only). Its jobs then go only to runners scoped to the
repository or its owner; the site's runners skip them when claiming.
Letting the site's runners take a repository's jobs ('on', or the site
default) stays a site administrator's choice. The settings page offers
what the viewer may choose, and tells a maintainer without a runner how
to register one.

Closes #225
Author
Andy McCall @andymccall <[email protected]>
via
agent Claude Code <[email protected]> · delegate for changes, until 2026-11-13 19:00Z · key ea5f78160088
When
· 2026-10-15 21:57 UTC
Signature
✔ signed by Claude Code
Hash
1870e4fa98086468364a61d2fa556f7fe5913bccd31404c75d8836cb12097d6c
Depends on
155cab142b63 19bb31ff1f5c 42fa95540468 4a4400c5812d 74be97962ed8 7974adb0be66 7ebb7052bc7f 9008bf6c2b8c a7e6e8e6094c b5668108487f b64e2537d702
Depended on by
c7067c7bda56 b9eb70fb9509 a9f1c0bf403a c8684d44d549 5d90e0e045e2 cd437c4f5f8d
In bookmarks
api-unknown-fields cancel-landed-runs diffusion-pin-bookmark draft-ready main oh-comment-body-file proposal-list-fast
In tags
v0.6.0 v0.6.1 v0.7.0 v0.8.0 v0.9.0 v0.9.1 v0.9.11 v0.9.15 v0.9.16 v0.9.19 v0.9.2 v0.9.20 v0.9.21 v0.9.22 v0.9.23 v0.9.24 v0.9.3 v0.9.4 v0.9.5 v0.9.6 v0.9.7 v0.9.8 v0.9.9
Edits
24 atoms, 0 tree operations, 7 files

Browse the tree at this change

modified CHANGELOG.md +2 −0
…
3 unchanged lines
4 4
5 5
## Unreleased
6 6
7 +
- **Flow on your own runners (#225):** a repository's maintainers can turn OxygenFlow on for it themselves once it (or its owner) has a runner of its own: Settings › OxygenFlow › "on, own runners only". Its jobs then go only to those runners, never to the site's; letting the site's runners take them stays a site administrator's choice.
8 +
7 9
## 0.5.2 (2026-10-15)
8 10
9 11
- **Your year on your profile (#217):** above the repositories, a waveform of the changes you pushed each day over the last 3, 6, 9 or 12 months (as long as you have been pushing), coloured by the repository the work went to, with the share pushed through agent tokens as a violet trace and release tags on the peaks. It draws itself in, a heartbeat runs along it while idle, and the pointer lights the days around it with a card for each one; the legend shows one repository's year. Pushes now record the agent token they came through.
…
199 unchanged lines
modified crates/hub/src/db/flow.rs +4 −0
…
494 unchanged lines
495 495
        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \
496 496
         WHERE j.status = 'queued' AND r.status IN ('queued', 'running') \
497 497
         AND (? = '' OR rep.owner = ?) AND (? IS NULL OR r.repo_id = ?) \
498 +
         AND NOT (? = '' AND ? IS NULL AND rep.flow = 'own') \
498 499
         ORDER BY j.run_id, j.depth, j.id LIMIT 200",
499 500
    )
500 501
    .bind(&scope.owner)
501 502
    .bind(&scope.owner)
502 503
    .bind(scope.repo_id)
503 504
    .bind(scope.repo_id)
505 +
    // A repository on its own runners only (#225): the site's runners skip it.
506 +
    .bind(&scope.owner)
507 +
    .bind(scope.repo_id)
504 508
    .fetch_all(&mut *tx)
505 509
    .await?;
506 510
    for j in queued {
…
547 unchanged lines
modified crates/hub/src/flow.rs +2 −1
…
22 unchanged lines
23 23
/// site administrator made one, else hub.toml's `[flow] repos`.
24 24
pub fn enabled(app: &Shared, row: &db::repos::RepoRow) -> bool {
25 25
    match row.flow.as_str() {
26 -
        "on" => true,
26 +
        // `own`: on, run only by the repository's or its owner's runners (#225).
27 +
        "on" | "own" => true,
27 28
        "off" => false,
28 29
        _ => app.config.flow.allows(&row.owner, &row.name),
29 30
    }
…
2056 unchanged lines
modified crates/hub/src/web/repo_settings.rs +43 −7
…
174 unchanged lines
175 175
    diffusion_site: String,
176 176
    flow_enabled: bool,
177 177
    site_admin: bool,
178 +
    /// Whether the repository or its owner has a runner of its own (#225).
179 +
    own_runner: bool,
178 180
    notice: Option<String>,
179 181
    error: Option<String>,
180 182
    /// Runners that may take this repository's jobs: (runner, what it
…
35 unchanged lines
216 218
    let diffusion_site = app.config.diffusion.follow.clone();
217 219
    let flow_enabled = crate::flow::enabled(app, &row);
218 220
    let site_admin = shell.is_admin;
221 +
    let own_runner = has_own_runner(app, &row).await.map_err(|e| Error::from_db(shell.clone(), e))?;
219 222
    let runners = db::runners::serving(&app.db, &row.owner, row.id)
220 223
        .await
221 224
        .map_err(|e| Error::from_db(shell.clone(), e))?
…
36 unchanged lines
258 261
        diffusion_site,
259 262
        flow_enabled,
260 263
        site_admin,
264 +
        own_runner,
261 265
        notice,
262 266
        error,
263 267
    }))
…
541 unchanged lines
805 809
    flow: String,
806 810
}
807 811
808 -
/// Turn OxygenFlow on or off for one repository. Runs use the site's
809 -
/// runners, so only site administrators decide.
812 +
/// Turn OxygenFlow on or off for one repository. Site administrators choose
813 +
/// anything; the repository's maintainers may turn it on on their own
814 +
/// runners (`own`), when the repository or its owner has one, or off (#225).
815 +
/// Only site administrators let the site's runners take its jobs.
810 816
pub async fn flow(
811 817
    State(app): State<Shared>,
812 818
    viewer: Viewer,
…
3 unchanged lines
816 822
    let shell = Shell::new(&app, &format!("/{owner}/{name}/settings"), &viewer);
817 823
    let (_, row, role) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Maintain).await?;
818 824
    csrf(&viewer, &f.csrf_token, &shell)?;
819 -
    if !viewer.is_admin() {
820 -
        return settings_page(&app, shell, row, role, None, Some("Only site administrators turn OxygenFlow on or off.".into())).await;
821 -
    }
822 825
    let setting = match f.flow.as_str() {
823 826
        "on" => "on",
827 +
        "own" => "own",
824 828
        "off" => "off",
825 829
        "default" => "",
826 -
        _ => return settings_page(&app, shell, row, role, None, Some("Pick on, off or the site default.".into())).await,
830 +
        _ => return settings_page(&app, shell, row, role, None, Some("Pick on, own runners, off or the site default.".into())).await,
827 831
    };
832 +
    if !viewer.is_admin() {
833 +
        if matches!(setting, "on" | "") {
834 +
            return settings_page(
835 +
                &app,
836 +
                shell,
837 +
                row,
838 +
                role,
839 +
                None,
840 +
                Some("Only site administrators let the site's runners take this repository's jobs.".into()),
841 +
            )
842 +
            .await;
843 +
        }
844 +
        if setting == "own" && !has_own_runner(&app, &row).await.map_err(|e| Error::from_db(shell.clone(), e))? {
845 +
            return settings_page(
846 +
                &app,
847 +
                shell,
848 +
                row,
849 +
                role,
850 +
                None,
851 +
                Some("Register a runner for this repository (Runners, below) or its owner first.".into()),
852 +
            )
853 +
            .await;
854 +
        }
855 +
    }
828 856
    db::repos::set_flow(&app.db, row.id, setting).await.map_err(|e| Error::from_db(shell.clone(), e))?;
829 857
    let label = if setting.is_empty() { "default" } else { setting };
830 858
    accounts::audit(&app.db, viewer.account(), "repo.flow", &format!("{owner}/{name}"), label).await.ok();
831 859
    let row = db::repos::get_by_id(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?.unwrap_or(row);
832 -
    let notice = if crate::flow::enabled(&app, &row) {
860 +
    let notice = if row.flow == "own" {
861 +
        "OxygenFlow is on, on this repository's own runners: pushes with a .flow/pipeline.kdl start runs that only they take."
862 +
    } else if crate::flow::enabled(&app, &row) {
833 863
        "OxygenFlow is on: pushes with a .flow/pipeline.kdl start runs."
834 864
    } else {
835 865
        "OxygenFlow is off for this repository."
…
135 unchanged lines
971 1001
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Admin).await?;
972 1002
    csrf(&viewer, &f.csrf_token, &shell)?;
973 1003
    Ok(axum::Json(super::assistants::models(&app, &repo_assistants(&row), f).await.map_err(|e| Error::from_db(shell.clone(), e))?))
1004 +
}
1005 +
1006 +
/// Whether a runner serves this repository that is not the site's: one
1007 +
/// scoped to it or to its owner (#225).
1008 +
async fn has_own_runner(app: &Shared, row: &db::repos::RepoRow) -> sqlx::Result<bool> {
1009 +
    Ok(db::runners::serving(&app.db, &row.owner, row.id).await?.iter().any(|r| !r.scope().is_site()))
974 1010
}
modified crates/hub/tests/api.rs +1 −1
…
931 unchanged lines
932 932
    assert!(!oxygenhub::flow::enabled(app, &row().await));
933 933
    set("default").await;
934 934
    assert_eq!(row().await.flow, "");
935 -
    assert!(set("sometimes").await.body.contains("Pick on, off"));
935 +
    assert!(set("sometimes").await.body.contains("Pick on, own runners, off"));
936 936
}
937 937
938 938
/// Deleting bookmarks (hub spec §6.4): pushed deletions and the delete
…
535 unchanged lines
modified crates/hub/tests/flow.rs +80 −0
…
4070 unchanged lines
4071 4071
    db::runners::set_state(&h.app.db, id, "active").await.unwrap();
4072 4072
    assert_eq!(db::runners::erroring(&h.app.db, id, 5).await.unwrap(), None);
4073 4073
}
4074 +
4075 +
/// #225: an ordinary user turns Flow on for their repository on their own
4076 +
/// runner, and the site's runners never take its jobs.
4077 +
#[tokio::test(flavor = "multi_thread")]
4078 +
async fn owners_run_flow_on_their_own_runners_only() {
4079 +
    use oxy_core::identity::SecretKey;
4080 +
    let h = hub().await;
4081 +
    let register = |name: &'static str| {
4082 +
        let app = h.app.clone();
4083 +
        async move {
4084 +
            let body = format!("username={name}&email={name}%40example.com&password=correct-horse-battery");
4085 +
            let res = oxygenhub::router(app)
4086 +
                .oneshot(
4087 +
                    Request::post("/register")
4088 +
                        .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
4089 +
                        .body(Body::from(body))
4090 +
                        .unwrap(),
4091 +
                )
4092 +
                .await
4093 +
                .unwrap();
4094 +
            res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string()
4095 +
        }
4096 +
    };
4097 +
    register("andy").await; // the first account: the site's administrator
4098 +
    let bob = register("bob").await;
4099 +
    create(&h.app, "bob", "game").await;
4100 +
    let web = |method: &'static str, path: &str, body: Option<String>| {
4101 +
        let (app, cookie, path) = (h.app.clone(), bob.clone(), path.to_string());
4102 +
        async move {
4103 +
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
4104 +
            if body.is_some() {
4105 +
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
4106 +
            }
4107 +
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
4108 +
            String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned()
4109 +
        }
4110 +
    };
4111 +
    let csrf_of = |page: &str| {
4112 +
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
4113 +
        page[i..].split('"').next().unwrap().to_string()
4114 +
    };
4115 +
    // No runner of his own yet: he is told how, and cannot switch it on.
4116 +
    let page = web("GET", "/bob/game/settings", None).await;
4117 +
    assert!(page.contains("register a runner for this repository") && !page.contains("value=\"own\""), "{page}");
4118 +
    let page = web("POST", "/bob/game/settings/flow", Some(format!("_csrf={}&flow=own", csrf_of(&page)))).await;
4119 +
    assert!(page.contains("Register a runner for this repository"), "{page}");
4120 +
    // He registers one.
4121 +
    let page = web("GET", "/bob/game/settings", None).await;
4122 +
    let page = web("POST", "/bob/game/settings/runners/token", Some(format!("_csrf={}", csrf_of(&page)))).await;
4123 +
    let i = page.find("--token ").unwrap() + "--token ".len();
4124 +
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
4125 +
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
4126 +
    let identity = SecretKey::generate().identity("bobs", "", 1);
4127 +
    let reg = serde_json::json!({ "token": token, "name": "bobs", "identity": hex::encode(identity.encode()), "info": info });
4128 +
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
4129 +
    assert_eq!(st, StatusCode::OK, "{body}");
4130 +
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
4131 +
    // The site's runners stay a site administrator's to grant; his own he may choose.
4132 +
    let page = web("GET", "/bob/game/settings", None).await;
4133 +
    assert!(page.contains("value=\"own\"") && !page.contains("value=\"on\""), "{page}");
4134 +
    let page = web("POST", "/bob/game/settings/flow", Some(format!("_csrf={}&flow=on", csrf_of(&page)))).await;
4135 +
    assert!(page.contains("Only site administrators let the site's runners"), "{page}");
4136 +
    let page = web("POST", "/bob/game/settings/flow", Some(format!("_csrf={}&flow=own", csrf_of(&page)))).await;
4137 +
    assert!(page.contains("value=\"own\" checked") && page.contains("start runs"), "{page}");
4138 +
    assert_eq!(db::repos::get(&h.app.db, "bob", "game").await.unwrap().unwrap().flow, "own");
4139 +
    // A push starts a run; the site's runner is not offered it, his is.
4140 +
    let work = tempfile::tempdir().unwrap();
4141 +
    let c = client(&format!("{}/bob/game", h.base), work.path());
4142 +
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
4143 +
    write(work.path(), "test.sh", "echo ok\n");
4144 +
    c.snapshot("t").unwrap();
4145 +
    c.commit("A pipeline", &[]).unwrap();
4146 +
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
4147 +
    assert_eq!(runs_of(&h.app, "bob", "game").await.len(), 1, "a run started");
4148 +
    assert!(claim(&h.app).await.is_none(), "the site's runner is not offered it");
4149 +
    let body = serde_json::json!({ "runner": "ignored", "runtimes": ["microvm"], "info": info }).to_string();
4150 +
    let (st, job) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(body)).await;
4151 +
    assert_eq!(st, StatusCode::OK, "{job}");
4152 +
    assert_eq!(serde_json::from_str::<serde_json::Value>(&job).unwrap()["repo"], "bob/game");
4153 +
}
modified templates/repo_settings.html +6 −5
…
70 unchanged lines
71 71
  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">
72 72
  <h2>OxygenFlow</h2>
73 73
  <p>Pushes to this repository {% if flow_enabled %}<strong>start runs</strong>{% else %}<strong>do not start runs</strong>{% endif %} of its <code>.flow/pipeline.kdl</code>.</p>
74 -
  {% if site_admin %}
74 +
  {% if site_admin || own_runner || row.flow == "own" %}
75 75
  <div class="choice">
76 -
    <label><input type="radio" name="flow" value="on"{% if row.flow == "on" %} checked{% endif %}> on</label>
76 +
    {% if site_admin %}<label><input type="radio" name="flow" value="on"{% if row.flow == "on" %} checked{% endif %}> on, any runner (the site's too)</label>{% endif %}
77 +
    {% if site_admin || own_runner %}<label><input type="radio" name="flow" value="own"{% if row.flow == "own" %} checked{% endif %}> on, own runners only</label>{% endif %}
77 78
    <label><input type="radio" name="flow" value="off"{% if row.flow == "off" %} checked{% endif %}> off</label>
78 -
    <label><input type="radio" name="flow" value="default"{% if row.flow.is_empty() %} checked{% endif %}> site default ({% if flow_default %}on{% else %}off{% endif %}, from <code>[flow] repos</code> in <code>hub.toml</code>)</label>
79 +
    {% if site_admin %}<label><input type="radio" name="flow" value="default"{% if row.flow.is_empty() %} checked{% endif %}> site default ({% if flow_default %}on{% else %}off{% endif %}, from <code>[flow] repos</code> in <code>hub.toml</code>)</label>{% endif %}
79 80
  </div>
80 -
  <p class="small mute">Runs use the site's runners, so only site administrators change this. It takes effect at the next push.</p>
81 +
  <p class="small mute"><strong>Own runners only</strong>: jobs go only to runners registered for this repository or its owner (Runners, below), never to the site's. Only site administrators let the site's runners take them. It takes effect at the next push.</p>
81 82
  <button type="submit">SAVE</button>
82 83
  {% else %}
83 -
  <p class="small mute">A site administrator turns OxygenFlow on or off here.</p>
84 +
  <p class="small mute">To run pipelines on your own machines, register a runner for this repository under <a href="#runners">Runners</a>, then choose <strong>on, own runners only</strong> here. A site administrator can also let the site's runners take its jobs.</p>
84 85
  {% endif %}
85 86
</form>
86 87
<form class="card wide" id="diffusion" method="post" action="{{ repo.url }}/settings/diffusion">
…
144 unchanged lines