johnandersen777.bsky.social/atlibs-socialweb-computer-requester

Switch repository

Loading…

// cicd-workload: computer.socialweb.cicd.workload.run, a CI job's steps on a machine, through the
// computer.socialweb.cicd.process.* hooks (a composition routes them: process-mcp over the guest agent's
// MCP shell). The checkout is fetched by the machine itself (git over https: knotless serves a repository
// from its owner's PDS); each step is `sh -s` with its variables on stdin, in /work. Output is the run's
// result as it comes: {step, output}; the last result: {steps: [{name, exitCode}], ok}.
//   run  {url, job: {checkout?: {git, commit, merge?: {git, commit}}, env?, steps?: [{name, run, env?}]}}
// (a runner checks out once, then runs one step a call: the machine keeps /work between them)
use abi::{Call, Event, Handle, Run, entries};
use serde_json::{Map, Value, json};

const START: &str = "computer.socialweb.cicd.process.start";
const WRITE: &str = "computer.socialweb.cicd.process.write";
const READ: &str = "computer.socialweb.cicd.process.read";
const WAIT_MS: u64 = 25000;
const MAX_BYTES: u64 = 262144;

fn quoted(s: &str) -> String {
    format!("'{}'", s.replace('\'', "'\\''"))
}

fn text<'a>(v: &'a Value, key: &str, at: &str) -> Result<&'a str, String> {
    v.get(key).and_then(Value::as_str).filter(|s| !s.is_empty()).ok_or(format!("{at}.{key} is required"))
}

fn exports(env: Option<&Value>) -> Result<String, String> {
    let mut out = String::new();
    for (k, v) in env.and_then(Value::as_object).into_iter().flatten() {
        let ok = !k.is_empty() && k.bytes().enumerate().all(|(i, b)| b == b'_' || b.is_ascii_alphabetic() || (i > 0 && b.is_ascii_digit()));
        let Some(v) = v.as_str().filter(|_| ok) else { return Err(format!("env {k}: a NAME and a string value")) };
        out.push_str(&format!("export {k}={}\n", quoted(v)));
    }
    Ok(out)
}

/// The checkout, then the steps: (name, the stdin of `sh -s`).
fn scripts(job: &Value) -> Result<Vec<(String, String)>, String> {
    let mut out = vec![];
    if let Some(checkout) = job.get("checkout") {
        out.push(("checkout".to_string(), checkout_script(checkout)?));
    }
    let shared = exports(job.get("env"))?;
    for (i, step) in job.get("steps").and_then(Value::as_array).into_iter().flatten().enumerate() {
        let at = format!("job.steps[{i}]");
        let name = step.get("name").and_then(Value::as_str).map(String::from).unwrap_or_else(|| format!("step {}", i + 1));
        let script = format!("cd /work\nexport KNOTLESS_RUN_OUTPUTS=/outputs\n{shared}{}exec sh -ec {}\n", exports(step.get("env"))?, quoted(text(step, "run", &at)?));
        out.push((name, script));
    }
    Ok(out)
}

/// Clones the commit into /work (and merges `merge` into it), making /outputs.
fn checkout_script(checkout: &Value) -> Result<String, String> {
    let (git, commit) = (text(checkout, "git", "job.checkout")?, text(checkout, "commit", "job.checkout")?);
    let mut fetch = format!(
        "set -e\ncommand -v git >/dev/null || {{ apt-get update -qq && DEBIAN_FRONTEND=noninteractive apt-get install -y -qq git >/dev/null; }}\nrm -rf /work /outputs\nmkdir -p /outputs\ngit clone -q {} /work\ncd /work\ngit checkout -q {}\n",
        quoted(git),
        quoted(commit)
    );
    if let Some(merge) = checkout.get("merge") {
        let (git, commit) = (text(merge, "git", "job.checkout.merge")?, text(merge, "commit", "job.checkout.merge")?);
        fetch.push_str(&format!("git fetch -q {} {}\ngit -c user.name=knotless -c user.email=knotless@localhost merge -q --no-edit FETCH_HEAD\n", quoted(git), quoted(commit)));
    }
    Ok(fetch)
}

fn run_handler(run: Run<Map<String, Value>>) -> Result<Handle, String> {
    let url = run.args.get("url").and_then(Value::as_str).ok_or("url is required: the machine's shell")?;
    let job = run.args.get("job").ok_or("job is required")?;
    let steps = scripts(job)?;
    let mut state = if run.state.is_null() { json!({ "step": 0, "codes": [] }) } else { run.state };
    let step = state["step"].as_u64().unwrap_or(0) as usize;
    let start = |state: &Value, i: usize| {
        Handle::new().call(Call::new("start", START).input(json!({ "url": url, "command": "sh -s" }))).state(json!({ "step": i, "codes": state["codes"] }))
    };
    let read = |state: Value| Handle::new().call(Call::new("read", READ).input(json!({ "url": url, "id": state["id"], "cursor": state["cursor"], "waitMs": WAIT_MS, "maxBytes": MAX_BYTES }))).state(state);
    Ok(match run.event {
        Event::Start if steps.is_empty() => Handle::new().result(json!({ "steps": [], "ok": true })).done(),
        Event::Start => start(&state, 0),
        Event::Result { id, result, .. } => {
            let out = result.output().map_err(|e| format!("{}: {e}", steps[step].0))?;
            match id.as_str() {
                "start" => {
                    state["id"] = out["id"].clone();
                    state["cursor"] = json!(0);
                    // the script in the write that closes stdin: a start's own stdin is written as it starts,
                    // so a close right after could come first
                    Handle::new().call(Call::new("write", WRITE).input(json!({ "url": url, "id": out["id"], "data": steps[step].1, "close": true }))).state(state)
                }
                "write" => read(state),
                "read" => {
                    state["cursor"] = out["cursor"].clone();
                    // this read's output, the run's result for now (the host shows each as it comes)
                    let mut said = Map::new();
                    if let Some(chunk) = out["output"].as_str().filter(|s| !s.is_empty()) {
                        said.insert("step".into(), json!(steps[step].0));
                        said.insert("output".into(), json!(chunk));
                    }
                    let next = if out["done"] != true {
                        read(state)
                    } else {
                        let code = out["exitCode"].as_i64().unwrap_or(-1);
                        state["codes"].as_array_mut().expect("a list").push(json!({ "name": steps[step].0, "exitCode": code }));
                        if code != 0 || step + 1 == steps.len() {
                            said.insert("steps".into(), state["codes"].clone());
                            said.insert("ok".into(), json!(code == 0));
                            return Ok(Handle::new().result(Value::Object(said)).done());
                        }
                        start(&state, step + 1)
                    };
                    if said.is_empty() { next } else { next.result(Value::Object(said)) }
                }
                _ => Handle::new(),
            }
        }
        _ => Handle::new(),
    })
}

entries!(run => run_handler);