johnandersen777.bsky.social/atlibs-socialweb-computer-requester
socialweb.computer market requester and policy modules for atlibs, with its web pages and lexicons
0
0
// vm-mcp: VMs for a model as four small tools, composed over the requester and the guest's process
// contract (hooks; a composition routes them, else the defaults: requester, process-mcp). The model
// sees a short id per VM; receipts, records and transport refs stay in one record of the user's,
// computer.socialweb.vm.kept at rkey `pool` (getRecord at that key: never listRecords).
// vm_create {cpus?, mem?, disk?, bidders?, hold_minutes?} -> provision (release: false) -> the VM kept
// vm_list {} -> the VMs kept
// vm_exec {vm, command, stdin?, timeout_s?} -> process.exec on its shell -> {exit, stdout, stderr}
// vm_release {vm} -> release (vm.delete, vm.terminated, records deleted) -> the VM dropped
// vm_forget {vm} -> a bounded release attempt, then the VM dropped either way (its bidder is gone)
// vm_create sets compute.vm expiresAt (now + hold_minutes): the bidder ends the VM then.
// Host-set input (not tools' fields): bidders (default: BIDDERS; a call's own win), via (iroh|webrtc), pool, and the limits
// max_vms, max_cpus, max_mem, max_disk, max_hold_minutes, wait_minutes.
use abi::repo::{PUT_RECORD, get_record};
use abi::{AtUri, Call, Event, Handle, Run, entries};
use serde_json::{Map, Value, json};
const KEPT: &str = "computer.socialweb.vm.kept";
const PROVISION: &str = "computer.socialweb.vm.provision";
const RELEASE: &str = "computer.socialweb.vm.release";
const EXEC: &str = "computer.socialweb.cicd.process.exec";
const POLL_MS: u64 = 5000;
// The bidders asked when neither the call nor the host names any: hat-7, hat-9, hat-8 (the requester
// reads each one's offering at self and skips one without compute).
const BIDDERS: [&str; 3] = ["did:plc:7q5lnitxv4meizdon5fmhzrq", "did:plc:4vcig7s43z3nyqrnepmcfzqt", "did:plc:2l2wdry77c7jtucbslstagkj"];
// putRecord compares and swaps on the record's CID; a lost race re-reads, this many times
const SWAPS: u64 = 3;
type Args = Map<String, Value>;
/// A host limit: a positive integer, else the default.
fn limit(args: &Args, key: &str, default: u64) -> u64 {
args.get(key).and_then(Value::as_u64).filter(|n| *n > 0).unwrap_or(default)
}
/// A tool's integer field: 1..=max, default when absent.
fn int(args: &Args, key: &str, default: u64, max: u64) -> Result<u64, String> {
match args.get(key) {
None | Some(Value::Null) => Ok(default.min(max)),
Some(v) => v.as_u64().filter(|n| (1..=max).contains(n)).ok_or(format!("{key} must be an integer from 1 to {max}")),
}
}
fn text<'a>(args: &'a Args, key: &str) -> Result<&'a str, String> {
args.get(key).and_then(Value::as_str).filter(|s| !s.is_empty()).ok_or(format!("{key} is required"))
}
fn pool(args: &Args) -> Result<AtUri, String> {
let did = args.get("did").and_then(Value::as_str).ok_or("sign in first: the VMs are kept in your repo")?;
let rkey = args.get("pool").and_then(Value::as_str).unwrap_or("self");
if rkey.is_empty() || rkey.len() > 512 || !rkey.bytes().all(|b| b.is_ascii_alphanumeric() || b".-_:~".contains(&b)) {
return Err("pool must be a record key".into());
}
Ok(AtUri::new(did, KEPT, rkey))
}
fn get(args: &Args, id: &str) -> Result<Handle, String> {
Ok(Handle::new().call(get_record(&pool(args)?).id(id)))
}
/// The kept list and its CID from a getRecord; a missing record is an empty list.
fn kept(result: Result<Value, String>) -> Result<(Vec<Value>, Value), String> {
match result {
Ok(out) => Ok((out["value"]["vms"].as_array().cloned().unwrap_or_default(), out["cid"].clone())),
Err(e) if e.contains("RecordNotFound") || e.to_lowercase().contains("not found") => Ok((vec![], Value::Null)),
Err(e) => Err(format!("reading your VMs: {e}")),
}
}
fn put(args: &Args, vms: Vec<Value>, cid: &Value) -> Result<Call, String> {
let uri = pool(args)?;
let record = json!({ "$type": KEPT, "vms": vms, "updatedAt": args.get("now") });
let mut input = json!({ "repo": uri.repo, "collection": KEPT, "rkey": uri.rkey, "record": record });
if cid.is_string() {
input["swapRecord"] = cid.clone();
}
Ok(Call::new("put", PUT_RECORD).input(input))
}
fn find<'a>(vms: &'a [Value], vm: &str) -> Result<&'a Value, String> {
vms.iter().find(|v| v["vm"] == vm).ok_or_else(|| {
let ids: Vec<&str> = vms.iter().filter_map(|v| v["vm"].as_str()).collect();
format!("no VM {vm} here (yours: {})", if ids.is_empty() { "none".into() } else { ids.join(", ") })
})
}
// ---- time: the host's `now` (RFC 3339, UTC) <-> seconds since the epoch ----------------------------
fn secs(iso: &str) -> Option<i64> {
let n = |a: usize, b: usize| -> Option<i64> { iso.get(a..b)?.parse().ok() };
let (y, m, d) = (n(0, 4)?, n(5, 7)?, n(8, 10)?);
let y = if m <= 2 { y - 1 } else { y };
let era = y.div_euclid(400);
let yoe = y - era * 400;
let doy = (153 * (if m > 2 { m - 3 } else { m + 9 }) + 2) / 5 + d - 1;
let days = era * 146097 + yoe * 365 + yoe / 4 - yoe / 100 + doy - 719468;
Some(days * 86400 + n(11, 13)? * 3600 + n(14, 16)? * 60 + n(17, 19)?)
}
fn iso(t: i64) -> String {
let (days, s) = (t.div_euclid(86400) + 719468, t.rem_euclid(86400));
let era = days.div_euclid(146097);
let doe = days - era * 146097;
let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = yoe + era * 400 + i64::from(m <= 2);
format!("{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z", s / 3600, s / 60 % 60, s % 60)
}
/// What the model sees of a kept VM.
fn view(v: &Value, now: Option<i64>) -> Value {
let expired = now.zip(v["expires"].as_str().and_then(secs)).is_some_and(|(n, x)| n >= x);
json!({ "vm": v["vm"], "bidder": v["bidder"], "cpus": v["cpus"], "mem": v["mem"], "disk": v["disk"], "created": v["created"], "expires": v["expires"], "expired": expired, "shell": v["shell"] })
}
fn now(args: &Args) -> Option<i64> {
args.get("now").and_then(Value::as_str).and_then(secs)
}
// ---- vm_create ------------------------------------------------------------------------------------
/// The provision input for this call, checked against the host's limits.
fn rent(args: &Args) -> Result<Value, String> {
let cpus = int(args, "cpus", 2, limit(args, "max_cpus", 8))?;
let mem = int(args, "mem", 4, limit(args, "max_mem", 16))?;
let bidders = match args.get("bidders") {
None | Some(Value::Null) => json!(BIDDERS),
Some(b) => b.as_array().filter(|b| !b.is_empty() && b.len() <= 10 && b.iter().all(|d| d.as_str().is_some_and(|d| d.starts_with("did:")))).map(|_| b.clone()).ok_or("bidders must be 1 to 10 bidder DIDs (did:plc:...), or left out for the default bidders")?,
};
let polls = (limit(args, "wait_minutes", 5) * 60_000 / POLL_MS).clamp(1, 200);
let mut input = json!({ "did": args.get("did"), "bidders": bidders, "cpus": cpus, "mem": mem, "release": false, "poll_ms": POLL_MS, "max_polls": polls });
if args.get("disk").is_some_and(|d| !d.is_null()) {
input["disk"] = json!(format!("{}G", int(args, "disk", 20, limit(args, "max_disk", 100))?));
}
match args.get("via").and_then(Value::as_str) {
Some("webrtc") => (input["via"], input["tag"]) = (json!("webrtc"), json!("shell")),
Some("iroh") => input["via"] = json!("iroh"),
None => {}
Some(_) => return Err("via must be iroh or webrtc".into()),
}
Ok(input)
}
/// A kept VM from provision's result, or why there is none (the requester gave back a VM that never came up).
fn entry(input: &Value, created: &Value, out: &Value) -> Result<Value, String> {
let mcp = out["mcp"].as_array().filter(|m| !m.is_empty() && out["release"].is_null());
let Some(mcp) = mcp else {
let errors: Vec<&str> = out["errors"].as_array().into_iter().flatten().filter_map(Value::as_str).collect();
let why = if errors.is_empty() { "the market returned none".into() } else { errors.join("; ") };
return Err(format!("no VM ({why}); nothing is kept. Try vm_create again in a few minutes, with fewer cpus or mem, or with other bidders"));
};
let uri = out["receipt"]["uri"].as_str().ok_or("provision returned no receipt")?;
let rkey = uri.rsplit('/').next().unwrap_or_default();
let vm = format!("vm-{}", rkey.strip_prefix("receipt-").unwrap_or(rkey).chars().take(8).collect::<String>());
let urls: Vec<&str> = mcp.iter().filter_map(|m| m["server"]["url"].as_str()).collect();
let scheme = input["via"].as_str().map(|v| format!("{v}:"));
let shell = urls.iter().find(|u| scheme.as_ref().is_some_and(|s| u.starts_with(s.as_str()))).or(urls.first()).ok_or("provision named no shell")?;
let bidder = out["bid"]["bidder"].as_str().or(uri.split('/').nth(2));
Ok(json!({
"vm": vm, "bidder": bidder, "cpus": input["cpus"], "mem": input["mem"], "disk": input["disk"],
"created": created, "expires": input["expires_at"], "shell": shell,
"receipt": out["receipt"], "records": out["records"],
}))
}
fn create_handler(run: Run<Args>) -> Result<Handle, String> {
let args = &run.args;
let input = rent(args)?;
let mut state = if run.state.is_null() { json!({ "tries": 0 }) } else { run.state };
Ok(match run.event {
Event::Start => {
int(args, "hold_minutes", 60, limit(args, "max_hold_minutes", 240))?;
get(args, "count")?
}
Event::Result { id, result, .. } => match id.as_str() {
"count" => {
let (vms, cid) = kept(result.into_result())?;
let max = limit(args, "max_vms", 2);
if vms.len() as u64 >= max {
return Err(format!("you keep {} VMs, at most {max}: vm_release one first", vms.len()));
}
// the list written back as it is, before renting: a session without the kept scope fails here, not after a VM is up
Handle::new().call(put(args, vms, &cid)?.id("probe"))
}
"probe" => {
// a lost swap still shows the write is allowed
if let Err(e) = result.into_result().map(|_| ()).or_else(|e| if e.contains("InvalidSwap") { Ok(()) } else { Err(e) }) {
return Err(format!("cannot keep your VM list ({e}); nothing was rented. Sign in again and allow keeping the list of VMs"));
}
// the bidder ends the VM at compute.vm expiresAt: now + hold_minutes
let mut input = input;
let (t, hold) = (now(args).ok_or("the host gave no time")?, int(args, "hold_minutes", 60, limit(args, "max_hold_minutes", 240))?);
input["expires_at"] = json!(iso(t + hold as i64 * 60));
state["created"] = json!(iso(t));
state["input"] = input.clone();
Handle::new().call(Call::new("rent", PROVISION).input(input)).state(state)
}
"rent" => {
state["entry"] = entry(&state["input"], &state["created"], &result.into_result()?)?;
get(args, "reget")?.state(state)
}
"reget" => {
let (mut vms, cid) = kept(result.into_result())?;
vms.retain(|v| v["vm"] != state["entry"]["vm"]);
vms.push(state["entry"].clone());
Handle::new().call(put(args, vms, &cid)?).state(state)
}
"put" => match result.into_result() {
Ok(_) => Handle::new().result(view(&state["entry"], now(args))).done(),
Err(e) if e.contains("InvalidSwap") && state["tries"].as_u64() < Some(SWAPS) => {
state["tries"] = json!(state["tries"].as_u64().unwrap_or(0) + 1);
get(args, "reget")?.state(state)
}
// rented but not recorded: give it back rather than leave a VM nobody can name
Err(e) => {
state["error"] = json!(e);
let e = &state["entry"];
Handle::new().call(Call::new("undo", RELEASE).input(json!({ "did": args.get("did"), "receipt": e["receipt"], "records": e["records"], "poll_ms": POLL_MS }))).state(state)
}
},
"undo" => {
let undone = match result.into_result() {
Ok(_) => "it was released".to_string(),
Err(e) => format!("releasing it failed too ({e})"),
};
return Err(format!("the VM came up but could not be recorded ({}); {undone}", state["error"].as_str().unwrap_or_default()));
}
_ => Handle::new(),
},
_ => Handle::new(),
})
}
// ---- vm_list, vm_exec, vm_release ---------------------------------------------------------------
fn list_handler(run: Run<Args>) -> Result<Handle, String> {
Ok(match run.event {
Event::Start => get(&run.args, "list")?,
Event::Result { result, .. } => {
let (vms, _) = kept(result.into_result())?;
let t = now(&run.args);
Handle::new().result(json!({ "vms": vms.iter().map(|v| view(v, t)).collect::<Vec<_>>() })).done()
}
_ => Handle::new(),
})
}
fn exec_handler(run: Run<Args>) -> Result<Handle, String> {
let args = &run.args;
let (vm, command) = (text(args, "vm")?, text(args, "command")?);
let timeout = int(args, "timeout_s", 60, 600)?;
Ok(match run.event {
Event::Start => get(args, "find")?,
Event::Result { id, result, .. } if id == "find" => {
let (vms, _) = kept(result.into_result())?;
let mut input = json!({ "url": find(&vms, vm)?["shell"], "command": command, "timeoutMs": timeout * 1000 });
if let Some(stdin) = args.get("stdin").and_then(Value::as_str) {
input["stdin"] = json!(stdin);
}
Handle::new().call(Call::new("exec", EXEC).input(input))
}
Event::Result { result, .. } => {
let out = result.into_result().map_err(|e| format!("{vm}: {e}"))?;
let mut said = json!({ "exit": out["exitCode"], "stdout": out["stdout"].as_str().unwrap_or_default(), "stderr": out["stderr"].as_str().unwrap_or_default() });
for (from, to) in [("truncated", "truncated"), ("timedOut", "timed_out")] {
if out[from] == true {
said[to] = json!(true);
}
}
Handle::new().result(said).done()
}
_ => Handle::new(),
})
}
fn release_handler(run: Run<Args>) -> Result<Handle, String> {
drop_vm(run, false)
}
fn forget_handler(run: Run<Args>) -> Result<Handle, String> {
drop_vm(run, true)
}
// ponytail: forget waits FORGET_MS for the release, then drops the entry whatever came of it
const FORGET_MS: u64 = 60_000;
/// vm_release: the release, then the entry dropped; one that fails keeps it. vm_forget: a bounded release
/// attempt, then the entry dropped either way (a bidder that never answers).
fn drop_vm(run: Run<Args>, forget: bool) -> Result<Handle, String> {
let args = &run.args;
let vm = text(args, "vm")?;
let mut state = if run.state.is_null() { json!({ "tries": 0 }) } else { run.state };
Ok(match run.event {
Event::Start => get(args, "find")?,
Event::Result { id, result, .. } => match id.as_str() {
"find" => {
let (vms, _) = kept(result.into_result())?;
let e = find(&vms, vm)?;
let polls = if forget { 2 } else { (limit(args, "wait_minutes", 5) * 60_000 / POLL_MS).clamp(1, 200) };
let call = Call::new("release", RELEASE).input(json!({ "did": args.get("did"), "receipt": e["receipt"], "records": e["records"], "poll_ms": POLL_MS, "max_polls": polls }));
Handle::new().call(if forget { call.timeout_ms(FORGET_MS) } else { call })
}
"release" => {
match result.into_result() {
Ok(out) => (state["released"], state["terminated"]) = (json!(true), json!(out["release"]["terminated"] == true)),
Err(e) if forget => (state["released"], state["terminated"], state["note"]) = (json!(false), json!(false), json!(format!("the release did not go through ({e}); the bidder may run it until its expiry, and the records renting it stay in your repo"))),
Err(e) => return Err(format!("{vm} is still kept: {e}. If its bidder is gone for good, vm_forget drops it from your list")),
}
get(args, "reget")?.state(state)
}
"reget" => {
let (mut vms, cid) = kept(result.into_result())?;
vms.retain(|v| v["vm"] != vm);
Handle::new().call(put(args, vms, &cid)?).state(state)
}
"put" => match result.into_result() {
Ok(_) => {
let mut said = json!({ "vm": vm, "released": state["released"], "terminated": state["terminated"] });
if forget {
said["forgotten"] = json!(true);
if state["note"].is_string() {
said["note"] = state["note"].clone();
}
}
Handle::new().result(said).done()
}
Err(e) if e.contains("InvalidSwap") && state["tries"].as_u64() < Some(SWAPS) => {
state["tries"] = json!(state["tries"].as_u64().unwrap_or(0) + 1);
get(args, "reget")?.state(state)
}
Err(e) => return Err(format!("{vm} was released, but your VM list was not updated: {e}")),
},
_ => Handle::new(),
},
_ => Handle::new(),
})
}
entries!(vm_create => create_handler, vm_list => list_handler, vm_exec => exec_handler, vm_release => release_handler, vm_forget => forget_handler);
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn time_round_trips() {
for t in ["1970-01-01T00:00:00Z", "2026-11-06T12:34:56Z", "2028-02-29T23:59:59Z", "2100-03-01T00:00:00Z"] {
assert_eq!(iso(secs(t).unwrap()), t);
}
assert_eq!(secs("2026-11-06T12:34:56.789Z"), Some(1_793_968_496));
assert_eq!(iso(secs("2026-12-31T23:30:00Z").unwrap() + 3600), "2027-01-01T00:30:00Z");
}
}