johnandersen777.bsky.social/atlibs-agents
agents and atirc modules for atlibs: launch agents from a composition, chat on atirc; pages and the agents service
0
0
// agents: launch an atirc agent on a socialweb.computer VM and shut it down by its launch rkey.
// shell.socialweb.computer does this with a ComputeContract spec.agent, a vault
// secrets bundle and an agentStart ContractEvent; here it is one module composed with
// the requester (composition.json):
//
// launch createRecord com.fedfork.agents.launch (its rkey is the launch id) + describeRepo
// -> chat.atirc.mint <name>-<rkey>.<handle> via the PDS (atproto-proxy, service auth)
// -> com.fedfork.market.contract (routed to requester `contract`, release: false) whose
// ssh command writes /etc/atirc/atirc.env + the agent unit and starts it
// -> putRecord status running
// status getRecord -> com.fedfork.market.watch (one poll: guest reports, terminated)
// shutdown getRecord -> chat.atirc.revoke + com.fedfork.market.release -> watch until terminated
// -> deleteRecord the round's records -> delete (or mark stopped) the launch record
//
// A launch that fails after its record exists tears down the same way (status failed).
use abi::repo::{create_record, delete_record, get_record, put_record};
use abi::{AtUri, Call, Event, Handle, Run, cap, entries};
use serde_json::{Map, Value, json};
const LAUNCH: &str = "com.fedfork.agents.launch";
const ATIRC: &str = "did:web:irc.fedfork.com#atirc";
const CONTRACT: &str = "com.fedfork.market.contract";
const RELEASE: &str = "com.fedfork.market.release";
const WATCH: &str = "com.fedfork.market.watch";
const DEFAULT_AGENT: &str = "atirc agent run";
const STARTED: &str = "atlibs-agents: started";
// ---- the guest side: what the ssh command installs ------------------------------------------
// The same shape as socialweb-computer-kcp internal/atproto/cloudinit/agent.go (atirc-agent.service,
// atirc-agent-run.sh, /etc/atirc/atirc.env), delivered over ssh instead of a secrets bundle, so the
// token never lands in a record (compute.vm user_data is public).
const RUN_SCRIPT: &str = r#"#!/usr/bin/env bash
set -euo pipefail
nick="$(atirc nick)"
if ! [[ "${nick}" =~ ^[a-z_][a-z0-9_-]{0,31}$ ]]; then echo "atirc-agent: nick ${nick@Q} is not a user name" >&2; exit 1; fi
getent group wheel >/dev/null || groupadd wheel
printf '%%wheel ALL=(ALL) NOPASSWD: ALL\n' > /etc/sudoers.d/wheel && chmod 440 /etc/sudoers.d/wheel
id "${nick}" >/dev/null 2>&1 || useradd -m -G wheel -s /bin/bash "${nick}"
home="$(getent passwd "${nick}" | cut -d: -f6)"
cd "${home}"
# runuser without --login keeps the environment: ATIRC_TOKEN comes from the unit's EnvironmentFile
# (0600), never from argv, so `ps` on the guest does not show it.
exec runuser -u "${nick}" -- env HOME="${home}" PATH=/usr/local/bin:/usr/bin:/bin bash -c "$(cat /etc/atirc/agent.cmd)"
"#;
const UNIT: &str = "[Unit]
Description=atirc agent (atlibs agents)
After=network-online.target
Wants=network-online.target
[Service]
EnvironmentFile=/etc/atirc/atirc.env
ExecStart=/usr/local/bin/atirc-agent-run.sh
Restart=on-failure
RestartSec=30
";
fn b64(data: &[u8]) -> String {
const T: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
let mut out = String::with_capacity(data.len().div_ceil(3) * 4);
for c in data.chunks(3) {
let n = (c[0] as u32) << 16 | (*c.get(1).unwrap_or(&0) as u32) << 8 | *c.get(2).unwrap_or(&0) as u32;
for i in 0..4 {
out.push(if i <= c.len() { T[(n >> (18 - 6 * i)) as usize & 63] as char } else { '=' });
}
}
out
}
// The ssh command (root on the guest) and its stdin: the env file (0600), the agent command, the run
// script and unit, atirc if the image has none, then start the unit: what agentStart does in kcp.
// The token is the stdin, read by the shell's builtin `read` and written by builtin `printf`: it is
// in no argv (ssh's on the host, the login shell's on the guest), so `ps` shows it nowhere.
pub fn provision(token: &str, agent: &str) -> Result<(String, String), String> {
if token.is_empty() || !token.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-~+/=".contains(&b)) {
return Err("atirc answered a token with unexpected characters".into());
}
let command = format!(
"set -e; umask 077; install -d -m 0700 /etc/atirc; IFS= read -r t; printf 'ATIRC_TOKEN=%s\\n' \"$t\" > /etc/atirc/atirc.env; unset t; \
echo {cmd} | base64 -d > /etc/atirc/agent.cmd; echo {run} | base64 -d > /usr/local/bin/atirc-agent-run.sh; \
chmod 0700 /usr/local/bin/atirc-agent-run.sh; umask 022; echo {unit} | base64 -d > /etc/systemd/system/atirc-agent.service; \
command -v atirc >/dev/null || {{ curl -fsSLo /usr/local/bin/atirc https://irc.fedfork.com/dl/$(uname -m) && chmod 0755 /usr/local/bin/atirc; }}; \
systemctl daemon-reload; systemctl restart atirc-agent.service; echo '{STARTED}'",
cmd = b64(agent.as_bytes()),
run = b64(RUN_SCRIPT.as_bytes()),
unit = b64(UNIT.as_bytes()),
);
Ok((command, format!("{token}\n")))
}
// ---- helpers --------------------------------------------------------------------------------
fn str_arg<'a>(a: &'a Value, key: &str) -> Option<&'a str> {
a[key].as_str().map(str::trim).filter(|s| !s.is_empty())
}
fn did_of(a: &Value) -> Result<String, String> {
str_arg(a, "did").filter(|d| d.starts_with("did:")).map(String::from).ok_or_else(|| "did is required (the session's DID)".into())
}
// Form fields arrive as strings; tests and hosts may pass numbers.
fn number(value: &Value, default: u64, max: u64) -> Result<u64, String> {
let n = match value {
Value::Null => return Ok(default),
Value::Number(n) => n.as_u64(),
Value::String(s) => s.trim().parse().ok(),
_ => None,
};
n.filter(|n| (60..=max).contains(n)).ok_or_else(|| format!("exp_s must be 60 to {max}"))
}
// Unix seconds of the host's `now` (ISO 8601 UTC, `2026-11-09T01:02:03.456Z`), for the token's exp.
fn unix(now: &str) -> Option<u64> {
let n = |r: std::ops::Range<usize>| now.get(r)?.parse::<i64>().ok();
let (y, m, d) = (n(0..4)?, n(5..7)?, n(8..10)?);
let (y, mp) = if m <= 2 { (y - 1, m + 9) } else { (y, m - 3) };
let days = 365 * y + y / 4 - y / 100 + y / 400 + (153 * mp + 2) / 5 + d - 1 - 719_468;
u64::try_from(days * 86400 + n(11..13)? * 3600 + n(14..16)? * 60 + n(17..19)?).ok()
}
// A launch rkey: a TID or any record key the PDS gave (letters, digits, `.-_:~`, at most 512).
fn rkey_of(a: &Value) -> Result<String, String> {
str_arg(a, "rkey")
.filter(|r| r.len() <= 512 && *r != "." && *r != ".." && r.bytes().all(|b| b.is_ascii_alphanumeric() || b".-_:~".contains(&b)))
.map(String::from)
.ok_or_else(|| "rkey is required (the launch id)".into())
}
fn bidders_of(a: &Value) -> Result<Vec<String>, String> {
let list: Vec<String> = match &a["bidders"] {
Value::Array(l) => l.iter().map(|b| b.as_str().unwrap_or_default().trim().to_string()).collect(),
Value::String(s) => s.split(|c: char| c == ',' || c.is_whitespace()).filter(|s| !s.is_empty()).map(String::from).collect(),
_ => vec![],
};
if list.is_empty() || list.len() > 8 || list.iter().any(|b| !b.starts_with("did:")) {
return Err("bidders must be 1 to 8 bidder DIDs".into());
}
Ok(list)
}
// The atirc label: <name>-<rkey>.<handle> must be a valid subhandle (kcp keeps labels to 14).
fn name_of(a: &Value) -> Result<String, String> {
let name = str_arg(a, "name").unwrap_or("agent");
let ok = name.len() <= 14
&& name.starts_with(|c: char| c.is_ascii_lowercase())
&& !name.ends_with('-')
&& name.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-');
ok.then(|| name.to_string()).ok_or_else(|| "name must be 1 to 14 of a-z 0-9 -, starting with a letter".into())
}
fn caller_of(a: &Value) -> Result<Option<String>, String> {
match &a["caller"] {
Value::Null => Ok(None),
Value::String(c) if c.starts_with("did:") => Ok(Some(c.clone())),
_ => Err("caller must be a DID".into()),
}
}
// A launch record of another caller: a host serving many accounts names the caller on every call.
fn owned(a: &Value, record: &Value, rkey: &Value) -> Result<(), String> {
match caller_of(a)? {
Some(caller) if record["caller"].as_str() != Some(caller.as_str()) => {
Err(format!("403 Forbidden: launch {} belongs to another caller", rkey.as_str().unwrap_or("?")))
}
_ => Ok(()),
}
}
fn at_uri(uri: &Value) -> Option<AtUri> {
uri.as_str()?.parse().ok()
}
fn launch_uri(s: &Value) -> AtUri {
AtUri::new(s["did"].as_str().unwrap_or_default(), LAUNCH, s["rkey"].as_str().unwrap_or_default())
}
// Ids still pending in this phase; true when `id` was the last one.
fn settle(s: &mut Value, id: &str) -> bool {
if let Some(wait) = s["wait"].as_array_mut() {
wait.retain(|w| w != id);
return wait.is_empty();
}
true
}
fn waiting(s: &mut Value, id: &str) {
match s["wait"].as_array_mut() {
Some(wait) => wait.push(json!(id)),
None => s["wait"] = json!([id]),
}
}
fn note(s: &mut Value, error: String) {
match s["errors"].as_array_mut() {
Some(errors) => errors.push(json!(error)),
None => s["errors"] = json!([error]),
}
}
// What the launch record and the results show (never the token).
fn summary(s: &Value) -> Value {
json!({
"rkey": s["rkey"], "uri": s["uri"], "status": s["record"]["status"], "subhandle": s["record"]["subhandle"],
"nick": s["record"]["nick"], "receipt": s["record"]["receipt"], "errors": s["errors"],
})
}
// ---- launch ---------------------------------------------------------------------------------
fn launch_handler(run: Run<Value>) -> Result<Handle, String> {
let a = &run.args;
let mut s = run.state;
let now = a["now"].as_str().unwrap_or_default().to_string();
match run.event {
Event::Start => {
let did = did_of(a)?;
let name = name_of(a)?;
let bidders = bidders_of(a)?;
str_arg(a, "user_data").ok_or("user_data is required (the guest needs an iroh or webrtc ssh path)")?;
let via = str_arg(a, "via").unwrap_or("iroh");
if via != "iroh" && via != "webrtc" {
return Err("via must be iroh or webrtc".into());
}
let agent = str_arg(a, "agent").unwrap_or(DEFAULT_AGENT);
if agent.len() > 1024 {
return Err("agent must be a command of at most 1024 bytes".into());
}
// Every agent token expires (kcp: "pass --exp on every leased mint"); shutdown revokes it sooner.
let exp = unix(&now).ok_or("now must be an ISO 8601 UTC time")? + number(&a["exp_s"], 7200, 86400)?;
let mut record = json!({ "$type": LAUNCH, "name": name, "status": "launching", "bidders": bidders, "via": via,
"agent": agent, "createdAt": now });
// caller: who launched it, when a host serves other accounts (the XRPC service sets it to
// the verified service-auth issuer); status and shutdown then answer only that caller.
if let Some(caller) = caller_of(a)? {
record["caller"] = json!(caller);
}
s = json!({ "did": did, "record": record, "exp": exp, "poll_ms": a["poll_ms"], "wait": ["record", "handle"], "phase": "launch", "errors": [] });
Ok(Handle::new()
.call(create_record(&did, LAUNCH, None, s["record"].clone()).id("record"))
.call(Call::new("handle", "com.atproto.repo.describeRepo").params(json!({ "repo": did })))
.state(s))
}
Event::Result { id, result, .. } => {
if s["phase"] == "teardown" {
return teardown_result(s, &id, result.output().cloned(), &now);
}
match (id.as_str(), result.output()) {
("record", Ok(out)) => {
let uri = at_uri(&out["uri"]).ok_or("createRecord returned no uri")?;
s["rkey"] = json!(uri.rkey);
s["uri"] = json!(uri.to_string());
if settle(&mut s, "record") {
return mint(s);
}
// The launch id is known: hosts (the XRPC service) answer with it from here on.
let early = summary(&s);
Ok(Handle::new().result(early).state(s))
}
("record", Err(e)) => Err(format!("creating the launch record: {e}")),
("handle", Ok(out)) => {
let handle = out["handle"].as_str().filter(|h| h.contains('.') && *h != "handle.invalid");
match handle {
Some(h) => s["handle"] = json!(h.to_ascii_lowercase()),
None => note(&mut s, "describeRepo returned no valid handle".into()),
}
if settle(&mut s, "handle") { mint(s) } else { Ok(Handle::new().state(s)) }
}
("handle", Err(e)) => {
note(&mut s, format!("describeRepo: {e}"));
if settle(&mut s, "handle") { mint(s) } else { Ok(Handle::new().state(s)) }
}
("mint", Ok(out)) => {
let token = out["token"].as_str().unwrap_or_default();
let agent = s["record"]["agent"].as_str().unwrap_or(DEFAULT_AGENT).to_string();
let (command, stdin) = match provision(token, &agent) {
Ok(c) => c,
Err(e) => return fail(s, e, &now),
};
s["record"]["nick"] = out["nick"].clone();
// progress: the requester calls it with the round's refs once the receipt lands.
let mut input = json!({ "did": s["did"], "bidders": a["bidders"], "user_data": a["user_data"],
"via": s["record"]["via"], "command": command, "stdin": stdin, "release": false, "progress": cap("progress") });
for key in ["tag", "role", "cpus", "mem", "disk", "network", "policies", "poll_ms", "max_polls", "accept_paid"] {
if !a[key].is_null() {
input[key] = a[key].clone();
}
}
s["phase"] = json!("contract");
let early = summary(&s);
// The subhandle is in the record before any VM exists, so a run that dies from here
// on still leaves what shutdown needs to revoke the token.
Ok(Handle::new().call(put_launch(&s, "minted", &now)).call(Call::new("contract", CONTRACT).input(input)).result(early).state(s))
}
("mint", Err(e)) => fail(s, format!("chat.atirc.mint: {e}"), &now),
("contract", Ok(out)) => contract_done(s, out.clone(), &now),
("contract", Err(e)) => fail(s, format!("contract: {e}"), &now),
("minted", Err(e)) => {
note(&mut s, format!("putRecord: {e}"));
Ok(Handle::new().state(s))
}
("receipt", out) => {
if let Err(e) = out {
note(&mut s, format!("putRecord: {e}"));
}
// Answer the requester's progress call only once the receipt is recorded.
let reply = s["reply"].as_str().unwrap_or_default().to_string();
Ok(Handle::new().call(Call::new("progress-ack", reply).input(Value::Null)).state(s))
}
("running", Ok(_)) => {
let out = summary(&s);
Ok(Handle::new().result(out).state(s).done())
}
("running", Err(e)) => {
note(&mut s, format!("putRecord: {e}"));
let out = summary(&s);
Ok(Handle::new().result(out).state(s).done())
}
_ => Ok(Handle::new()),
}
}
// The requester's progress call: the receipt (and the round's records) before the guest is up,
// recorded so a run that dies mid-contract leaves a VM that shutdown can release.
Event::Callback { cap, input, reply, .. } if cap == "progress" && s["phase"] == "contract" => {
record_round(&mut s, &input);
s["reply"] = json!(reply);
let early = summary(&s);
Ok(Handle::new().call(put_launch(&s, "receipt", &now)).result(early).state(s))
}
Event::Callback { reply, .. } => Ok(Handle::new().call(Call::new("callback-ack", reply).input(Value::Null))),
Event::Timer { .. } | Event::Closed { .. } | Event::Message { .. } | Event::Input { .. } | Event::Unknown => Ok(Handle::new()),
}
}
fn put_launch(s: &Value, id: &str, now: &str) -> Call {
let mut record = s["record"].clone();
record["updatedAt"] = json!(now);
put_record(&launch_uri(s), record).id(id)
}
// The round's refs from a contract result or progress call: what teardown releases and deletes.
fn record_round(s: &mut Value, out: &Value) {
if out["receipt"]["uri"].is_string() {
s["record"]["receipt"] = json!({ "uri": out["receipt"]["uri"], "cid": out["receipt"]["cid"], "submitEvent": out["receipt"]["submitEvent"] });
s["record"]["bidder"] = json!(at_uri(&out["receipt"]["uri"]).map(|u| u.repo));
}
let records: Map<String, Value> = ["vm", "rfp", "accept"].into_iter().filter(|k| out[*k]["uri"].is_string()).map(|k| (k.to_string(), out[k]["uri"].clone())).collect();
if !records.is_empty() {
s["record"]["records"] = json!(records);
}
}
fn mint(mut s: Value) -> Result<Handle, String> {
let now = s["record"]["createdAt"].as_str().unwrap_or_default().to_string();
let Some(handle) = s["handle"].as_str().map(String::from) else {
return fail(s, "no account handle to mint the agent's atirc subhandle under".into(), &now);
};
let name = s["record"]["name"].as_str().unwrap_or("agent").to_string();
let rkey = s["rkey"].as_str().unwrap_or_default().to_ascii_lowercase();
let subhandle = format!("{name}-{rkey}.{handle}");
s["record"]["subhandle"] = json!(subhandle);
s["phase"] = json!("mint");
let input = json!({ "subhandle": subhandle, "exp": s["exp"] });
let early = summary(&s);
Ok(Handle::new().call(Call::new("mint", "chat.atirc.mint").proxy(ATIRC).input(input)).result(early).state(s))
}
// contract finished: a receipt and the started marker in the ssh output mean the agent runs.
fn contract_done(mut s: Value, out: Value, now: &str) -> Result<Handle, String> {
for e in out["errors"].as_array().into_iter().flatten() {
note(&mut s, format!("contract: {}", e.as_str().unwrap_or("error")));
}
// First, so teardown deletes whatever records the contract left, receipt or not.
record_round(&mut s, &out);
if !out["receipt"]["uri"].is_string() {
return fail(s, "no VM: the contract ended without a receipt".into(), now);
}
let shell = &out["shell"];
let started = shell["output"].as_str().is_some_and(|o| o.contains(STARTED)) && shell["exitCode"].as_i64().unwrap_or(0) == 0;
if !started {
let why = shell["error"].as_str().map(String::from).unwrap_or_else(|| format!("exit {} {}", shell["exitCode"], shell["output"].as_str().unwrap_or("")));
return fail(s, format!("starting the agent on the guest: {why}"), now);
}
s["record"]["status"] = json!("running");
s["record"]["updatedAt"] = json!(now);
s["phase"] = json!("running");
let put = put_record(&launch_uri(&s), s["record"].clone()).id("running");
let early = summary(&s);
Ok(Handle::new().call(put).result(early).state(s))
}
fn fail(mut s: Value, error: String, now: &str) -> Result<Handle, String> {
note(&mut s, error);
s["final"] = json!("failed");
teardown(s, now)
}
// ---- status ---------------------------------------------------------------------------------
fn status_handler(run: Run<Value>) -> Result<Handle, String> {
let a = &run.args;
let mut s = run.state;
match run.event {
Event::Start => {
let uri = AtUri::new(did_of(a)?, LAUNCH, rkey_of(a)?);
s = json!({ "did": uri.repo, "rkey": uri.rkey, "uri": uri.to_string() });
Ok(Handle::new().call(get_record(&uri).id("get")).state(s))
}
Event::Result { id, result, .. } if id == "get" => {
let out = result.output().map_err(|e| format!("reading the launch record: {e}"))?;
s["record"] = out["value"].clone();
owned(a, &s["record"], &s["rkey"])?;
let receipt = &s["record"]["receipt"];
if s["record"]["status"] != "running" || !receipt["uri"].is_string() {
return Ok(Handle::new().result(json!({ "rkey": s["rkey"], "uri": s["uri"], "record": s["record"] })).done());
}
let input = json!({ "did": s["did"], "bidder": s["record"]["bidder"], "receipt": receipt["uri"], "until": "terminated", "max_polls": 1 });
Ok(Handle::new().call(Call::new("watch", WATCH).input(input)).state(s))
}
Event::Result { id, result, .. } if id == "watch" => {
let guest = match result.output() {
Ok(o) => json!({ "iroh": o["iroh"], "webrtc": o["webrtc"], "hostKeys": o["hostKeys"], "terminated": o["terminated"] }),
Err(e) => json!({ "error": e }),
};
Ok(Handle::new().result(json!({ "rkey": s["rkey"], "uri": s["uri"], "record": s["record"], "guest": guest })).done())
}
_ => Ok(Handle::new()),
}
}
// ---- shutdown -------------------------------------------------------------------------------
fn shutdown_handler(run: Run<Value>) -> Result<Handle, String> {
let a = &run.args;
let mut s = run.state;
let now = a["now"].as_str().unwrap_or_default().to_string();
match run.event {
Event::Start => {
let uri = AtUri::new(did_of(a)?, LAUNCH, rkey_of(a)?);
s = json!({ "did": uri.repo, "rkey": uri.rkey, "uri": uri.to_string(), "delete": a["delete"] == true,
"poll_ms": a["poll_ms"], "final": "stopped", "phase": "get", "errors": [] });
Ok(Handle::new().call(get_record(&uri).id("get")).state(s))
}
Event::Result { id, result, .. } => {
if s["phase"] == "get" && id == "get" {
let out = result.output().map_err(|e| format!("reading the launch record: {e}"))?;
s["record"] = out["value"].clone();
owned(a, &s["record"], &s["rkey"])?;
return teardown(s, &now);
}
teardown_result(s, &id, result.output().cloned(), &now)
}
_ => Ok(Handle::new()),
}
}
// ---- teardown: revoke + release, watch terminated, delete the round, finish the record -------
fn teardown(mut s: Value, now: &str) -> Result<Handle, String> {
s["phase"] = json!("teardown");
s["wait"] = json!([]);
let mut h = Handle::new();
if let Some(subhandle) = s["record"]["subhandle"].as_str() {
h = h.call(Call::new("revoke", "chat.atirc.revoke").proxy(ATIRC).input(json!({ "subhandle": subhandle })));
waiting(&mut s, "revoke");
}
let receipt = s["record"]["receipt"].clone();
if receipt["uri"].is_string() {
// the requester's release: vm.delete + market.event, wait for vm.terminated, delete the round's records
let records: Vec<Value> = s["record"]["records"].as_object().map(|r| r.values().cloned().collect()).unwrap_or_default();
let mut input = json!({ "did": s["did"], "receipt": { "uri": receipt["uri"], "cid": receipt["cid"], "submitEvent": receipt["submitEvent"] },
"records": records, "reason": "agent_shutdown" });
if !s["poll_ms"].is_null() {
input["poll_ms"] = s["poll_ms"].clone();
}
h = h.call(Call::new("release", RELEASE).input(input));
waiting(&mut s, "release");
}
if s["wait"].as_array().is_some_and(Vec::is_empty) {
return cleanup(s, now);
}
Ok(h.state(s))
}
fn teardown_result(mut s: Value, id: &str, output: Result<Value, String>, now: &str) -> Result<Handle, String> {
let h = Handle::new();
match (id, output) {
("revoke", Ok(_)) => s["revoked"] = json!(true),
// A token that already died (its exp, an earlier revoke) is as good as revoked.
("revoke", Err(e)) => note(&mut s, format!("chat.atirc.revoke: {e}")),
// {release: {terminated, payload, event}, deleted, errors, ...}: what it deleted is not deleted again
("release", Ok(out)) => {
s["release"] = json!({ "payload": out["release"]["payload"], "event": out["release"]["event"] });
s["terminated"] = out["release"]["terminated"].clone();
s["released"] = out["deleted"].clone();
for e in out["errors"].as_array().into_iter().flatten() {
note(&mut s, format!("release: {}", e.as_str().map_or_else(|| e.to_string(), str::to_string)));
}
if out["release"]["terminated"] != true {
note(&mut s, "the bidder has not reported vm.terminated yet".into());
}
}
("release", Err(e)) => note(&mut s, format!("release: {e}")),
(rm, out) if rm.starts_with("rm:") => {
let uri = s["rmUris"][rm[3..].parse::<usize>().unwrap_or(usize::MAX)].clone();
match out {
Ok(_) => {
if let Some(d) = s["deleted"].as_array_mut() {
d.push(uri);
}
}
// Refused for a record this host no longer manages, or gone: the requester's own
// cleanup deleted it already.
Err(e) if e.contains("was not created by this module") || e.contains("RecordNotFound") => {}
Err(e) => note(&mut s, format!("deleteRecord {}: {e}", uri.as_str().unwrap_or("?"))),
}
if settle(&mut s, rm) {
return finish_record(s, now);
}
return Ok(h.state(s));
}
("final", out) => {
match out {
Ok(_) if s["delete"] == true => {
let uri = s["uri"].clone();
if let Some(d) = s["deleted"].as_array_mut() {
d.push(uri);
}
}
Ok(_) => {}
Err(e) => note(&mut s, format!("the launch record: {e}")),
}
let result = json!({ "rkey": s["rkey"], "uri": s["uri"], "status": s["final"], "subhandle": s["record"]["subhandle"],
"revoked": s["revoked"] == true, "release": s["release"], "terminated": s["terminated"], "deleted": s["deleted"],
"errors": s["errors"] });
return Ok(Handle::new().result(result).state(s).done());
}
_ => return Ok(h),
}
if settle(&mut s, id) {
return cleanup(s, now);
}
Ok(h.state(s))
}
// The round's records (vm, rfp, accept, the release's vm.delete and event) in the session's repo that the
// release did not delete (it had none to make: no receipt).
fn cleanup(mut s: Value, now: &str) -> Result<Handle, String> {
s["deleted"] = if s["released"].is_array() { s["released"].clone() } else { json!([]) };
let did = s["did"].as_str().unwrap_or_default().to_string();
let r = &s["record"]["records"];
let released = &s["deleted"];
let uris: Vec<AtUri> = [&r["vm"], &r["rfp"], &r["accept"], &s["release"]["payload"]["uri"], &s["release"]["event"]["uri"]]
.into_iter()
.filter(|u| !released.as_array().is_some_and(|d| d.contains(u)))
.filter_map(at_uri)
.filter(|u| u.repo == did)
.collect();
if uris.is_empty() {
return finish_record(s, now);
}
let mut h = Handle::new();
s["wait"] = json!([]);
for (k, uri) in uris.iter().enumerate().rev() {
h = h.call(delete_record(uri).id(format!("rm:{k}")));
waiting(&mut s, &format!("rm:{k}"));
}
s["rmUris"] = json!(uris);
Ok(h.state(s))
}
fn finish_record(mut s: Value, now: &str) -> Result<Handle, String> {
let call = if s["delete"] == true {
delete_record(&launch_uri(&s)).id("final")
} else {
s["record"]["status"] = s["final"].clone();
s["record"]["updatedAt"] = json!(now);
s["record"]["error"] = json!(s["errors"].as_array().into_iter().flatten().filter_map(Value::as_str).collect::<Vec<_>>().join("; "));
if s["record"]["error"] == "" {
s["record"].as_object_mut().map(|o| o.remove("error"));
}
put_record(&launch_uri(&s), s["record"].clone()).id("final")
};
s["wait"] = json!(["final"]);
Ok(Handle::new().call(call).state(s))
}
// ---- allow_bidders: an edge policy -----------------------------------------------------------
// input { args: { bidders: [did] }, call: { input: { bidders } } }: every bidder the contract would
// ask must be in args.bidders, so whoever calls launch (the XRPC service's clients) cannot spend
// the account on other bidders. Verdict { allow, violations: [{ msg, policyId }] }.
fn allow_bidders_handler(run: Run<Value>) -> Result<Handle, String> {
let allowed: Vec<&str> = run.args["args"]["bidders"].as_array().into_iter().flatten().filter_map(Value::as_str).collect();
let asked = bidders_of(&run.args["call"]["input"]).unwrap_or_default();
let denied: Vec<&String> = asked.iter().filter(|b| !allowed.contains(&b.as_str())).collect();
let verdict = if asked.is_empty() {
json!({ "allow": false, "violations": [{ "msg": "the call names no bidders", "policyId": "allow-bidders" }] })
} else if denied.is_empty() {
json!({ "allow": true, "violations": [] })
} else {
json!({ "allow": false, "violations": [{ "msg": format!("bidders not allowed: {}", denied.iter().map(|b| b.as_str()).collect::<Vec<_>>().join(", ")), "policyId": "allow-bidders" }] })
};
Ok(Handle::new().result(verdict).done())
}
entries! { launch => launch_handler, status => status_handler, shutdown => shutdown_handler, allow_bidders => allow_bidders_handler }